CentralizedStreamServer

def __init__(
    self,
    settings: Any,
    services: Optional[Dict[str, BaseStreamingService]] = None,
) -> None:
    self.settings = settings
    self.services = services or {}
    self.current_mode: Optional[str] = None
    self.lock = asyncio.Lock()
    self.active_task: Optional[asyncio.Task] = None

    self.app: Optional[web.Application] = None
    self.runner: Optional[web.AppRunner] = None
    self.sites: List[web.BaseSite] = []
    self.cert_watcher: Optional[asyncio.Task] = None
    self.ssl_context: Optional[ssl.SSLContext] = None
    self.static_fs_path: str = ""
    limit_mbps = float(self.settings.file_transfer_limit_mbps or 0.0)
    if not math.isfinite(limit_mbps) or limit_mbps < 0:
        logger.warning(
            f"Ignoring file_transfer_limit_mbps={limit_mbps!r}: not a usable rate."
        )
        limit_mbps = 0.0
    self.transfer_cap = TransferPacer(static_bps=int(limit_mbps * 125000))
    self.download_pacer = TransferPacer(adaptive=True)
    self.upload_pacer = TransferPacer(adaptive=True)
    self.upload_dir = pathlib.Path(
        os.path.expanduser(self.settings.file_manager_path)
    ).resolve()
    self.print_spool = pathlib.Path(
        os.path.expanduser(self.settings.print_spool_path)
    ).resolve()
    self.print_watcher: Optional[printing.SpoolWatcher] = None
    self.print_queue: Optional[printing.PrintQueue] = None
    self._recording_audio: Optional[Any] = None
    self._chunked_uploads: Dict[str, Dict[str, Any]] = {}
    self.web_files_ctx: Optional[tempfile.TemporaryDirectory] = None

    self._clients_present: bool = False
    self._client_hook_tasks: Set[asyncio.Task] = set()

    if bool(self.settings.wayland[0]):
        try:
            from pixelflux import ensure_wayland_display
            socket_name = ensure_wayland_display(
                width=int(self.settings.manual_width or 0),
                height=int(self.settings.manual_height or 0),
                render_node=self.settings.render_dri or "",
                auto_gpu=str(self.settings.auto_gpu or ""),
                cursor_size=int(self.settings.cursor_size),
            )
            if socket_name:
                host_display = str(
                    getattr(self.settings, "wayland_host_display", "") or "")
                os.environ["WAYLAND_DISPLAY"] = host_display or socket_name
                logger.info(f"Wayland compositor socket: {socket_name}")
            else:
                logger.warning(
                    "Wayland compositor socket did not come up within its "
                    "startup window; WAYLAND_DISPLAY left unchanged."
                )
        except ImportError:
            logger.warning("pixelflux unavailable; Wayland display not initialized.")

    self.STREAMING_MODE_WEBRTC = "webrtc"
    self.STREAMING_MODE_WEBSOCKETS = "websockets"
    self.STATIC_CONTENT_PATH = "selkies.selkies_web"
    self.MIME_TYPES = {
        "html": "text/html",
        "js": "text/javascript",
        "css": "text/css",
        "json": "application/json",
        "png": "image/png",
        "jpg": "image/jpeg",
        "ico": "image/x-icon",
        "svg": "image/svg+xml",
    }

def set_clients_present(self, present: bool) -> None:
    """Record client presence and fire the configured presence hook.

    Both streaming modes report here; the ``run_after_connect`` /
    ``run_after_disconnect`` hook command runs when the first client
    connects or the last one disconnects.

    Args:
        present: Whether at least one client is currently connected.
    """
    if present == self._clients_present:
        return
    self._clients_present = present
    cmd = (self.settings.run_after_connect if present
           else self.settings.run_after_disconnect)
    if cmd:
        # Hold a strong reference: the loop only keeps weak refs to tasks.
        task = asyncio.create_task(self._run_client_hook(cmd, present))
        self._client_hook_tasks.add(task)
        task.add_done_callback(self._client_hook_tasks.discard)

async def _run_client_hook(self, cmd: str, present: bool) -> None:
    """Run a presence hook shell command, killing it if it wedges."""
    name = "run_after_connect" if present else "run_after_disconnect"
    try:
        proc = await asyncio.create_subprocess_shell(cmd)
        try:
            returncode = await asyncio.wait_for(proc.wait(), timeout=300)
        except asyncio.TimeoutError:
            logger.warning(f"{name} command timed out after 300s; killing: {cmd}")
            try:
                proc.kill()
            except ProcessLookupError:
                pass
            returncode = await proc.wait()
        if returncode != 0:
            logger.warning(f"{name} command exited with status {returncode}: {cmd}")
    except OSError as e:
        logger.error(f"Failed to run {name} command {cmd!r}: {e}")

def _b64_decode(self, data: str) -> str:
    return base64.b64decode(data).decode("utf-8")

def _get_https_certs(self) -> Tuple[Optional[str], Optional[str]]:
    """Return absolute paths of the configured cert and key files, each None
    when unset or missing on disk."""
    https_cert = getattr(self.settings, "https_cert", None)
    https_key = getattr(self.settings, "https_key", None)
    cert_pem = (
        os.path.abspath(https_cert)
        if https_cert and os.path.isfile(https_cert)
        else None
    )
    key_pem = (
        os.path.abspath(https_key)
        if https_key and os.path.isfile(https_key)
        else None
    )
    return cert_pem, key_pem

def _self_signed_candidates(self) -> List[Tuple[str, str]]:
    """Where a generated pair may live, most preferred first.

    The configured path leads so an operator who pointed at one gets the
    pair there; the state directory catches a user install, which for the
    default `ssl-cert-snakeoil` path means anyone not running as root.

    Returns:
        `(certificate, key)` absolute path pairs; a pair with no key path
        is dropped, since a generated key is always written separately.
    """
    candidates = []
    configured = getattr(self.settings, "https_cert", None)
    if configured:
        candidates.append((os.path.abspath(configured),
                           os.path.abspath(getattr(self.settings, "https_key", "") or "")))
    state = os.path.join(
        os.environ.get("XDG_STATE_HOME") or os.path.expanduser("~/.local/state"),
        "selkies")
    candidates.append((os.path.join(state, "selkies.pem"),
                       os.path.join(state, "selkies.key")))
    return [(cert, key) for cert, key in candidates if cert and key]

def _self_signed_names(self) -> Tuple[list, list]:
    """The names a generated certificate is issued for.

    Loopback under every spelling, since that is what a published container
    port is reached by, plus the host's own name. Its routable address is
    deliberately absent: the pair outlives the address a container is given
    on each run, and one naming last week's is worse than one claiming none.

    Returns:
        `(dns names, ip addresses)`, both ready for `SubjectAlternativeName`.
    """
    from cryptography import x509

    dns = ["localhost"]
    hostname = socket.gethostname()
    if hostname and hostname != "localhost":
        dns.append(hostname)
    addresses = [ipaddress.ip_address("127.0.0.1"), ipaddress.ip_address("::1")]
    return ([x509.DNSName(name) for name in dns],
            [x509.IPAddress(address) for address in addresses])

def _self_signed_pair_usable(self, cert_path: str, key_path: str) -> bool:
    """Whether an existing generated pair can be served again as it is.

    Reused only if it loads, has not expired, and names loopback by address:
    one issued before those SANs fails hostname verification at 127.0.0.1,
    and keeping it would carry that forward for the life of the install.
    Anything unreadable counts as absent, so a truncated file is replaced.

    Args:
        cert_path: Certificate to examine.
        key_path: Its private key.

    Returns:
        True when the pair should be reused instead of regenerated.
    """
    from cryptography import x509

    if not (os.path.isfile(cert_path) and os.path.isfile(key_path)):
        return False
    try:
        with open(cert_path, "rb") as handle:
            cert = x509.load_pem_x509_certificate(handle.read())
        # not_valid_after_utc where cryptography has it, else the naive
        # attribute it replaced, read as UTC.
        expiry = getattr(cert, "not_valid_after_utc", None)
        now = datetime.now(tz=timezone.utc)
        if expiry is None:
            expiry = cert.not_valid_after.replace(tzinfo=timezone.utc)
        if expiry <= now:
            return False
        addresses = cert.extensions.get_extension_for_class(
            x509.SubjectAlternativeName).value.get_values_for_type(x509.IPAddress)
        if ipaddress.ip_address("127.0.0.1") not in addresses:
            return False
        context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)
        context.load_cert_chain(cert_path, keyfile=key_path)
    except Exception:
        return False
    return True

def _make_self_signed_cert(self) -> Tuple[str, str]:
    """Return a self-signed certificate and key, writing one where none is usable.

    Turning HTTPS on is otherwise a two-step job — make a certificate, then
    point at it — and browsers gate the clipboard, gamepads, pointer lock
    and the camera on a secure context, so the step is in everyone's way.
    The configured paths are used when their directory is writable, which
    for the default `ssl-cert-snakeoil` pair means running as root; a user
    install falls back to its own state directory. That directory is not a
    path `_get_https_certs` consults, so the pair already there is found
    here instead — and reused, since minting one per start would invalidate
    the exception the browser was told to make on every restart.

    Returns:
        The certificate and key paths, both absolute.

    Raises:
        OSError: No candidate directory could be written to.
    """
    from cryptography import x509
    from cryptography.hazmat.primitives import hashes, serialization
    from cryptography.hazmat.primitives.asymmetric import ec

    candidates = self._self_signed_candidates()
    for cert_path, key_path in candidates:
        if self._self_signed_pair_usable(cert_path, key_path):
            logger.info(
                "HTTPS is running on the self-signed certificate already at %s.",
                cert_path)
            return cert_path, key_path

    key = ec.generate_private_key(ec.SECP256R1())
    name = x509.Name([x509.NameAttribute(x509.NameOID.COMMON_NAME, "localhost")])
    dns_names, addresses = self._self_signed_names()
    now = datetime.now(tz=timezone.utc)
    cert = (
        x509.CertificateBuilder()
        .subject_name(name)
        .issuer_name(name)
        .public_key(key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now - timedelta(days=1))
        .not_valid_after(now + timedelta(days=3650))
        .add_extension(x509.SubjectAlternativeName(dns_names + addresses), critical=False)
        # Not a CA: a key sitting in the session container is a far worse
        # thing to put in a trust store than a leaf is to except.
        .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
        .sign(key, hashes.SHA256())
    )
    errors = []
    for cert_path, key_path in candidates:
        try:
            os.makedirs(os.path.dirname(cert_path), exist_ok=True)
            os.makedirs(os.path.dirname(key_path), exist_ok=True)
            with open(cert_path, "wb") as handle:
                handle.write(cert.public_bytes(serialization.Encoding.PEM))
            fd = os.open(key_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
            with os.fdopen(fd, "wb") as handle:
                handle.write(key.private_bytes(
                    encoding=serialization.Encoding.PEM,
                    format=serialization.PrivateFormat.PKCS8,
                    encryption_algorithm=serialization.NoEncryption()))
            # Set after the write: O_CREAT leaves an existing file's mode alone.
            os.chmod(key_path, 0o600)
        except OSError as error:
            errors.append(f"{cert_path}: {error}")
            continue
        logger.warning(
            "No certificate at the configured path, so HTTPS runs on a self-signed one "
            "written to %s, issued for %s and no other address. Browsers will warn once "
            "until it is trusted; a certificate from an authority, or a reverse proxy "
            "terminating TLS, avoids that.",
            cert_path,
            ", ".join([n.value for n in dns_names] + [str(a.value) for a in addresses]))
        return cert_path, key_path
    raise OSError("could not write a self-signed certificate: " + "; ".join(errors))

def _create_ssl_context(self) -> Optional[ssl.SSLContext]:
    """Build the server TLS context from the configured cert/key.

    Returns:
        The loaded context, or None when HTTPS is disabled.

    Raises:
        FileNotFoundError: HTTPS is enabled but the certificate is missing.
    """
    enable_https = getattr(self.settings, "enable_https", None)
    if not enable_https or not enable_https[0]:
        return None

    cert_pem, key_pem = self._get_https_certs()
    if not cert_pem:
        cert_pem, key_pem = self._make_self_signed_cert()

    logger.debug(
        "Creating TLS context with certificate=%s key=%s", cert_pem, key_pem
    )
    sslctx = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)
    sslctx.check_hostname = False
    sslctx.verify_mode = ssl.CERT_NONE
    try:
        sslctx.load_cert_chain(cert_pem, keyfile=key_pem if key_pem else None)
    except Exception:
        logger.error(
            "Certificate or private key file not found or incorrect. "
            'To use a self-signed certificate, install the package "ssl-cert" '
            'and add the group "ssl-cert" to your user in Debian-based '
            "distributions, or generate a new certificate with root using: "
            "openssl req -x509 -newkey rsa:4096 "
            "-keyout /etc/ssl/private/ssl-cert-snakeoil.key "
            "-out /etc/ssl/certs/ssl-cert-snakeoil.pem -days 3650 -nodes "
            '-subj "/CN=localhost"'
        )
        raise
    return sslctx

def _get_cert_mtime(self) -> float:
    """Return the most recent modification time of the cert and key files.

    Falls back to the generated pair actually being served, so the reload
    watcher is not blind to it when the configured path holds nothing.
    """
    cert_pem, key_pem = self._get_https_certs()
    if not cert_pem:
        for candidate_cert, candidate_key in self._self_signed_candidates():
            if os.path.isfile(candidate_cert):
                cert_pem, key_pem = candidate_cert, candidate_key
                break
    if not cert_pem:
        return 0.0
    try:
        cert_mtime = os.stat(cert_pem).st_mtime
        key_mtime = os.stat(key_pem).st_mtime if key_pem else 0.0
        return max(cert_mtime, key_mtime)
    except OSError:
        return 0.0

async def _watch_and_reload_certs(self) -> None:
    """Poll the TLS cert/key mtimes and swap in a new listening site on change.

    The new SSL context is built before the old site is stopped so a bad
    certificate never takes the server offline, and the recorded mtime only
    advances once the new site is up, so a failed reload is retried on the
    next poll. Both the stat and the context build run off the loop: they
    read files that may sit on a network mount, and this runs while the
    session is streaming.
    """
    reload_interval = getattr(self.settings, "cert_reload_interval", 30)
    if reload_interval <= 0:
        logger.debug("Automatic certificate reloading is disabled (interval=0)")
        return

    current_sites = self.sites
    last_mtime = await asyncio.to_thread(self._get_cert_mtime)
    logger.debug(
        "Certificate reload watcher started (interval=%ds, initial mtime=%.0f)",
        reload_interval,
        last_mtime,
    )

    while True:
        await asyncio.sleep(reload_interval)
        try:
            new_mtime = await asyncio.to_thread(self._get_cert_mtime)
        except Exception as exc:
            logger.warning("Could not stat cert/key files: %s", exc)
            continue

        if new_mtime <= last_mtime:
            continue

        logger.info(
            "Certificate change detected (mtime %.0f -> %.0f), reloading…",
            last_mtime,
            new_mtime,
        )
        try:
            new_ssl_context = await asyncio.to_thread(self._create_ssl_context)
        except Exception as exc:
            logger.error(
                "Failed to create new SSL context, keeping old certificate: %s",
                exc,
            )
            continue

        if new_ssl_context is None:
            logger.error(
                "New SSL context is None (HTTPS disabled?), keeping old site."
            )
            continue

        for site in current_sites:
            try:
                await site.stop()
            except Exception as exc:
                logger.warning("Error stopping old %s: %s", self._site_kind(), exc)
        current_sites = self.sites = []
        logger.info("Old %s stopped.", self._site_kind())

        try:
            current_sites = await self._start_sites(new_ssl_context)
            last_mtime = new_mtime
            logger.info(
                "New %s started with reloaded certificates on %s",
                self._site_kind(),
                self._site_endpoint(),
            )
        except Exception as exc:
            logger.critical(
                "Failed to start new %s: %s. HTTPS server may be down; "
                "will retry on the next certificate poll.",
                self._site_kind(),
                exc,
            )

@staticmethod
def _check_master_token(auth_header: Optional[str], master_token: Any) -> bool:
    """Timing-safe check of a ``Bearer <master_token>`` header, compared as
    UTF-8 bytes so non-ASCII tokens are safe."""
    if not auth_header or not auth_header.startswith("Bearer "):
        return False
    parts = auth_header.split()
    if len(parts) < 2:
        return False
    return hmac.compare_digest(
        parts[1].encode("utf-8"), str(master_token).encode("utf-8")
    )

@staticmethod
def _basic_auth_challenge(text: str = "Invalid Credentials") -> web.Response:
    """A 401 that re-opens the browser's login prompt.

    Wrong credentials carry the challenge too: without one the browser renders this
    body as a page instead of asking again, so a mistyped password ends the session.
    """
    return web.Response(
        status=401,
        headers={
            "WWW-Authenticate": f'Basic realm="{AUTH_REALM}", charset="UTF-8"'
        },
        text=text,
    )

@staticmethod
def _bearer_challenge(text: str = "Unauthorized") -> web.Response:
    """A 401 for a route that wants a token.

    Names the Bearer scheme so the web client's 401 guard leaves it alone: a
    reload cannot change the token a page holds, whereas a Basic challenge
    is exactly what a reload re-presents. Browsers show no prompt for it.
    """
    return web.Response(
        status=401,
        headers={"WWW-Authenticate": f'Bearer realm="{AUTH_REALM}"'},
        text=text,
    )

@staticmethod
def _session_token_carriers(request: web.Request) -> List[Tuple[str, str]]:
    """The session-token carriers a request presents, most explicit first.

    The Bearer header is what scripts send; the ``?token=`` query is what
    URLs the client navigates to rather than fetches carry (the page itself,
    the file-manager listing it opens); the cookie is the mirror the client
    keeps for requests it can put neither on. The cookie value is tried as
    sent and URL-decoded, since the client stores it encoded.

    Returns:
        ``(source, token)`` pairs, source being "header", "query" or "cookie".
    """
    carriers: List[Tuple[str, str]] = []
    auth_header = request.headers.get("Authorization", "")
    if auth_header.startswith("Bearer "):
        parts = auth_header.split()
        if len(parts) >= 2:
            carriers.append(("header", parts[1]))
    query_token = request.query.get("token")
    if query_token:
        carriers.append(("query", query_token))
    cookie = request.cookies.get(SESSION_TOKEN_COOKIE)
    if cookie:
        carriers.append(("cookie", cookie))
        decoded = urllib.parse.unquote(cookie)
        if decoded != cookie:
            carriers.append(("cookie", decoded))
    return carriers

def _session_token_verdict(
    self, request: web.Request, settings: Any
) -> Optional[Tuple[str, str]]:
    """Authenticate an API request by token in secure mode.

    The master token (Bearer) counts as a controller, as does a
    controller-role session token; a viewer-role token is tagged so the
    handlers that refuse view-only credentials refuse it too. Session
    tokens are looked up in the same constant-time table lookup the
    WebSocket handshakes use.

    Returns:
        ``(role_ceiling, source)`` for the first carrier holding a valid
        token, or None when none does.
    """
    # Resolved per call: selkies imports this module, so the token table
    # it owns cannot be imported at module load.
    from .sessions import _lookup_session_token

    if self._check_master_token(request.headers.get("Authorization"), settings.master_token):
        return "controller", "header"
    for source, token in self._session_token_carriers(request):
        perms = _lookup_session_token(token)
        if perms is not None:
            role = "controller" if perms.get("role") == "controller" else "viewer"
            return role, source
    return None

@staticmethod
def _is_origin_allowed(request: web.Request, settings: Any) -> bool:
    """Return whether a browser request's Origin is permitted.

    Applied to WebSocket upgrades and to the mode-switch POST. Empty
    ``allowed_origins`` means same-origin only (plus non-browser clients
    that send no Origin); ``*`` allows any; otherwise the Origin must be
    listed or match the Host header. A forwarded Host without a port
    (nginx's ``proxy_set_header Host $host``, as the bundled config does)
    is matched by hostname: a browser Origin carries any non-default port,
    so a strict netloc comparison would reject every same-origin
    connection reached via an explicit port such as the :6080 mapping.
    """
    origin = request.headers.get("Origin")
    if not origin:
        return True
    allowed = {
        o.strip()
        for o in (getattr(settings, "allowed_origins", "") or "").split(",")
        if o.strip()
    }
    if "*" in allowed or origin in allowed:
        return True
    host = request.headers.get("Host")
    if host:
        try:
            origin_parts = urllib.parse.urlsplit(origin)
            if origin_parts.netloc == host:
                return True
            host_parts = urllib.parse.urlsplit("//" + host)
            if (
                host_parts.port is None
                and origin_parts.hostname
                and origin_parts.hostname == host_parts.hostname
            ):
                return True
        except ValueError:
            pass
    return False

@web.middleware
async def _auth_middleware(
    self,
    request: web.Request,
    handler: Callable[[web.Request], Awaitable[web.StreamResponse]],
) -> web.StreamResponse:
    """Global auth guard for every route on the server: see ``_authorize``.

    A refusal is deliberately taken before the request's body is read -- an
    unauthenticated upload must not be paid for -- so the connection is closed
    with it. Leaving those bytes unread on a keep-alive connection has the
    server parse them as the next request's method, which fails that request
    and every later one on the same connection.
    """
    response = await self._authorize(request, handler)
    if response.status >= 400 and request.body_exists and request.can_read_body:
        response.force_close()
    return response

async def _authorize(
    self,
    request: web.Request,
    handler: Callable[[web.Request], Awaitable[web.StreamResponse]],
) -> web.StreamResponse:
    """Authorize one request and hand it to `handler`, or refuse it.

    Layered gates, in order: cross-site WebSocket upgrades are rejected by
    Origin; health/liveness endpoints pass without credentials; the token
    endpoint and the control endpoints (the mode switch, a session
    disconnect, a recording start or stop) accept the Bearer master token,
    trying `Authorization` first and the `MASTER_TOKEN_HEADER` fallback
    second for callers whose Authorization header a Basic login in front
    already owns (a
    mode switch not so authenticated is held to the same Origin rule as
    the upgrades, since a browser attaches cached Basic credentials to a
    cross-site POST); in secure mode every other API route accepts a
    session token (Bearer header, ``?token=`` query, or the client's
    cookie), which is the only credential when Basic auth is off; the mode
    switch takes the master token ahead of the Origin rule and otherwise
    authenticates like any other API route, its handler refusing every
    role but controller; and everything else falls through to Basic Auth
    when enabled. A cookie-
    carried token on a state-changing request is held to the Origin rule
    too, since the browser attaches it; and with a master token set, the
    WebSocket handshakes skip Basic (a browser cannot attach fresh Basic
    credentials to a handshake, so it would add only an undebuggable
    401). A request that authenticates with the view-only password or a
    viewer-role token is tagged with ``auth_role_ceiling = "viewer"`` for
    downstream handlers to enforce.
    """
    settings = request.app["settings"]
    auth_header = request.headers.get("Authorization")
    path = request.path
    is_ws_upgrade = request.headers.get("Upgrade", "").lower() == "websocket"
    # Match the exact route, not a suffix, so /foo/tokens isn't treated as control-plane.
    api_prefix = settings.subfolder
    is_ws_handshake = is_ws_upgrade and path.rstrip("/") in {
        f"{api_prefix}{route}" for route in WEBSOCKET_ROUTES}
    if is_ws_upgrade:
        if not self._is_origin_allowed(request, settings):
            logger.warning(
                "Rejected WebSocket upgrade from disallowed Origin: %r",
                request.headers.get("Origin", ""),
            )
            return web.Response(status=403, text="Forbidden origin")
    if path in (f"{api_prefix}/api/status", f"{api_prefix}/api/health"):
        return await handler(request)
    token_path = path == f"{api_prefix}/api/tokens"
    if settings.master_token and token_path:
        if not any(
            self._check_master_token(header, settings.master_token)
            for header in (auth_header, request.headers.get(MASTER_TOKEN_HEADER))
        ):
            return self._bearer_challenge()
        return await handler(request)

    # The operator's own paths, ahead of the Origin rule a browser is held
    # to; a session token reaches them through the verdict below.
    is_control_path = path == f"{api_prefix}/api/switch" or (
        request.method in ("POST", "DELETE")
        and (path == f"{api_prefix}/api/recording"
             or path.startswith(f"{api_prefix}/api/sessions/")))
    if settings.master_token and is_control_path:
        if any(
            self._check_master_token(header, settings.master_token)
            for header in (auth_header, request.headers.get(MASTER_TOKEN_HEADER))
        ):
            return await handler(request)
    if is_control_path and not self._is_origin_allowed(request, settings):
        logger.warning(
            "Rejected %s %s from disallowed Origin: %r",
            request.method, path, request.headers.get("Origin", ""),
        )
        return web.Response(status=403, text="Forbidden origin")

    if (settings.master_token and not is_ws_handshake and not token_path
            and path.startswith(f"{api_prefix}/api/")):
        verdict = self._session_token_verdict(request, settings)
        if verdict is not None:
            ceiling, source = verdict
            if (source == "cookie" and request.method not in ("GET", "HEAD")
                    and not self._is_origin_allowed(request, settings)):
                logger.warning(
                    "Rejected cookie-authenticated %s %s from disallowed Origin: %r",
                    request.method, path, request.headers.get("Origin", ""),
                )
                return web.Response(status=403, text="Forbidden origin")
            request["auth_role_ceiling"] = ceiling
            return await handler(request)
        if not settings.enable_basic_auth[0]:
            return self._bearer_challenge()

    if not settings.enable_basic_auth[0]:
        logger.debug("Basic auth not enabled, forwarding to routers")
        return await handler(request)
    if is_ws_handshake and settings.master_token:
        return await handler(request)
    if not auth_header or not auth_header.startswith("Basic "):
        if is_ws_upgrade:
            # A WS handshake gets no challenge/retry: without this log the
            # rejection is invisible on both ends.
            logger.warning(
                "Rejected WebSocket upgrade from %s: basic auth is enabled and the "
                "handshake carried no Authorization header (browsers only attach "
                "cached credentials; set a master token or disable basic auth for "
                "browser clients behind proxies).",
                request.remote,
            )
        return self._basic_auth_challenge("Authorization Required")
    try:
        auth_decoded = self._b64_decode(auth_header[6:])
        if ":" not in auth_decoded:
            return self._basic_auth_challenge()
        username, password = auth_decoded.split(":", 1)
        # Compare as UTF-8 bytes; hmac.compare_digest rejects non-ASCII str.
        user_ok = hmac.compare_digest(
            username.encode("utf-8"), str(settings.basic_auth_user).encode("utf-8")
        )
        pw = password.encode("utf-8")
        main_ok = hmac.compare_digest(
            pw, str(settings.basic_auth_password).encode("utf-8")
        )
        # Both comparisons run so reply timing never reveals which password was sent.
        viewonly_secret = str(getattr(settings, "basic_auth_viewonly_password", "") or "")
        viewonly_ok = bool(viewonly_secret) and hmac.compare_digest(
            pw, viewonly_secret.encode("utf-8")
        )
        if not (user_ok and (main_ok or viewonly_ok)):
            logger.warning(
                f"Invalid credentials provided for user: {settings.basic_auth_user}"
            )
            return self._basic_auth_challenge()
        request["auth_role_ceiling"] = (
            "viewer" if (viewonly_ok and not main_ok) else "controller"
        )
    except Exception:
        return self._basic_auth_challenge()
    return await handler(request)

def _require_configured_credentials(self) -> None:
    """Refuse to serve a login that nobody chose a password for.

    The built-in password is a placeholder, not a credential: reaching this point
    still carrying it means no password was set anywhere, and the server would
    otherwise put an unconfigured login on the network. What counts is that a value
    was supplied, on the command line or in the environment — not what the value is.
    An image that ships its own default is choosing it deliberately, the way most
    container images do, and rejecting known-weak values here would break every one
    of them while stopping nobody who meant it.

    Raises:
        SystemExit: With `EXIT_CONFIG_ERROR`, which says the settings are the
            problem and that starting again will not change the answer — a
            supervisor respawning this forever would otherwise leave a
            container that looks healthy and never answers.
    """
    if not self.settings.enable_basic_auth[0]:
        return
    if self.settings.was_provided("basic_auth_password"):
        return
    logger.error(
        "Basic authentication is enabled but no password was set. Set one with "
        "--basic-auth-password, or the SELKIES_BASIC_AUTH_PASSWORD, PASSWORD or "
        "PASSWD environment variable; or serve without a login by passing "
        "--enable-basic-auth=false."
    )
    raise SystemExit(EXIT_CONFIG_ERROR)

def _require_one_listen_setting(self) -> None:
    """Refuse `--public` beside an explicit `--addr`: both choose the TCP
    listen addresses, and letting one win in silence would leave the
    operator who typed the other believing it took effect.

    Raises:
        SystemExit: With `EXIT_CONFIG_ERROR`, as for missing credentials.
    """
    if self.settings.public[0] and self.settings.was_provided("addr"):
        logger.error(
            "--public and --addr (SELKIES_PUBLIC, SELKIES_ADDR) were both given: "
            "--public listens on every interface, --addr on the addresses it "
            "names. Pass one of them."
        )
        raise SystemExit(EXIT_CONFIG_ERROR)

async def switch_to_mode(self, mode_name: str) -> None:
    """Stop the active streaming service and start ``mode_name`` in its place.

    Serialized under the supervisor lock so two switches can never overlap;
    switching to the already-active mode is a no-op. The service reads
    the settings at start, so the encoder knob is brought in line with the
    transport first (a websockets-only encoder such as jpeg or striped
    h264enc cannot ride the WebRTC pipeline, and a switch back restores
    the operator's menu and value) and only then does an unpinned
    rate-control mode resolve, since its websockets default depends on
    the resolved encoder, the same order as `_post_process_settings`.

    Args:
        mode_name: Registered service name ("websockets" or "webrtc").

    Raises:
        ValueError: ``mode_name`` is not a registered service.
    """
    if mode_name not in self.services:
        raise ValueError(f"Service {mode_name} not found")

    async with self.lock:
        if self.current_mode == mode_name:
            logger.info(f"Mode {mode_name} is already active.")
            return

        await self._stop_service()
        logger.info(f"Starting service: {mode_name}")
        self.settings.mode = mode_name
        self.settings.apply_webrtc_encoder_filter()
        self.settings.resolve_rate_control_default()
        service = self.services[mode_name]
        task = asyncio.create_task(service.start())
        self.active_task = task
        self.current_mode = mode_name

        def _on_service_done(finished: asyncio.Task, mode: str = mode_name) -> None:
            """Clear the stale mode when the service dies unexpectedly; the
            exception is retrieved so asyncio never logs it as unretrieved."""
            if finished.cancelled():
                return
            exc = finished.exception()
            if exc is not None:
                logger.error(f"Service '{mode}' terminated unexpectedly: {exc!r}")
                if self.active_task is finished:
                    self.current_mode = None
                    self.active_task = None

        task.add_done_callback(_on_service_done)

async def _stop_service(self) -> None:
    """Stop the active service, escalating to a forced cancel on timeout.

    The grace period lets a teardown that includes ~2 s gamepad-close
    waits finish before a forced cancel that would leak resources.
    """
    if not self.current_mode:
        return
    logger.info(f"Stopping service: {self.current_mode}")

    await self.services[self.current_mode].stop()
    if self.active_task:
        try:
            await asyncio.wait_for(self.active_task, timeout=15)
        except asyncio.TimeoutError:
            logger.warning(
                f"Timeout while stopping '{self.current_mode}'. Canceling task."
            )
            self.active_task.cancel()
            try:
                await self.active_task
            except asyncio.CancelledError:
                logger.info(
                    f"Task canceled after timeout for '{self.current_mode}'."
                )
            except Exception as e:
                logger.warning(f"Service task raised during forced stop: {e!r}")
    self.current_mode = None
    self.active_task = None

def _get_status(self) -> Dict[str, Any]:
    """The /api/status body.

    `enable_dual_mode` is surfaced here so the dashboard can render the
    WebSocket/WebRTC toggle from this early, transport-independent probe:
    serverSettings only arrives once a stream connects, so a WebRTC
    session that never comes up would otherwise strand the user with no
    way back to WebSockets.
    """
    return {
        "current_mode": self.current_mode,
        "available_modes": list(self.services.keys()),
        "enable_dual_mode": bool(
            getattr(self.settings, "enable_dual_mode", (False,))[0]
        ),
    }

@staticmethod
def _viewer_ceiling(request: web.Request) -> bool:
    """Return whether the credential that authenticated this request caps it
    at the viewer role (the view-only basic-auth password, or a viewer-role
    session token in secure mode).

    The control-plane endpoints that change host or session state — the
    streaming-mode switch and file uploads — refuse those requests the way
    the streaming plane refuses input from a viewer. Read-only endpoints
    stay available: a viewer is already watching the session.
    """
    return request.get("auth_role_ceiling") == "viewer"

async def handle_switch(self, request: web.Request) -> web.Response:
    """POST /api/switch: change the active streaming mode.

    A controller may switch: it already drives the desktop, so withholding
    the transport from it protects nothing. What the credential does decide
    is who counts as one — view-only credentials and viewer-role tokens are
    refused here, and the master token remains the operator's way in for a
    deployment that hands out tokens it does not want switching for
    everyone (the switch restarts the service under every connected page).
    Refused as well when dual mode is disabled.
    """
    if self._viewer_ceiling(request):
        return web.json_response(
            {"status": "error", "message": "View-only credentials cannot switch streaming mode"},
            status=403,
        )
    dual_mode = getattr(self.settings, "enable_dual_mode", (False,))[0]
    if not dual_mode:
        return web.json_response(
            {"status": "error", "message": "Dual streaming mode disabled"},
            status=403,
        )

    try:
        data = await request.json()
        if not isinstance(data, dict):
            raise ValueError("Request body must be a JSON object")
        target_mode = data.get("mode")
        await self.switch_to_mode(target_mode)
        return web.json_response({"status": "success", "mode": target_mode})
    except Exception as e:
        return web.json_response({"status": "error", "message": str(e)}, status=400)

def _session_gauge(self, request: web.Request) -> Optional[UplinkGauge]:
    """The `UplinkGauge` for one transfer request, upload or download, or
    None when nothing can be gauged (the transfer then must not be
    throttled blindly).

    The gauge wants the requester's own session connections — another
    client's socket crosses a different path, and its jitter would
    throttle a transfer that cannot be queuing there. The active service
    lists every client session; the requester's are narrowed by session
    token when the request carries one, else by peer address, and when
    neither discriminates the whole list is gauged: with a floor per
    session, a foreign session can only ever add false congestion, which
    slows the transfer and never harms a stream.

    Gauge state (ping clock and floor history) lives in
    `_UPLINK_SESSIONS`, keyed weakly by session websocket, so the separate
    requests of a chunked transfer's slices share one floor and a
    mid-transfer slice never re-baselines against its own standing queue.
    """
    service = self.services.get(self.current_mode) if self.current_mode else None
    if service is None:
        return None
    conns = service.uplink_session_conns()
    if not conns:
        return None
    tokens = {token for _source, token in self._session_token_carriers(request)}
    if tokens:
        matched = [c for c in conns if c[1] and c[1] in tokens]
        if matched:
            conns = matched
    if len(conns) > 1 and request.remote:
        matched = [c for c in conns if c[2] == request.remote]
        if matched:
            conns = matched
    entries = []
    for ws, _token, _ip in conns:
        state = _uplink_session_state(ws)
        entries.append([ws, state, state["seq"]])
    return UplinkGauge(entries) if entries else None

async def _stream_upload_body(self, request: web.Request, path: str, append: bool) -> int:
    """Stream a request body to ``path`` with executor-thread writes.

    Creates/truncates the file when ``append`` is False, appends when True;
    O_NOFOLLOW blocks a planted symlink either way. Enforces the declared
    Content-Length.

    Reads pace against two allowances: a paused read fills aiohttp's
    flow-control buffer, the TCP window closes, and the client's uplink is
    freed for the input/feedback traffic the stream depends on. The
    operator's static cap is the shared `transfer_cap` bucket; congestion
    control is `upload_pacer` fed one `UplinkGauge` verdict per read, so
    the transfer takes whatever the uplink has spare and backs off the
    moment the session's own delay inflates. A transfer below
    `TRANSFER_MIN_GAUGED_BYTES` is over before a queue could matter and
    skips the gauge, as does one with no gauge-able session socket; a
    gauge whose sockets all vanish mid-transfer stops pacing — the
    session it protected is gone. A gauged transfer's reads are sized to
    the current rate (`_gauged_chunk_size`), since the read size is what
    the client is handed back as receive window and sends at once.

    Returns:
        The byte count written.

    Raises:
        Exception: Propagated from the read/write path after the handle is
            closed; the caller owns removal of the target file.
    """
    declared = request.content_length
    loop = asyncio.get_running_loop()
    cap = self.transfer_cap if self.transfer_cap.active else None
    gauge = (
        self._session_gauge(request)
        if (declared or 0) >= TRANSFER_MIN_GAUGED_BYTES
        else None)
    flags = os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW | (os.O_APPEND if append else os.O_TRUNC)
    fd = os.open(path, flags, 0o644)
    fh = os.fdopen(fd, "wb")
    written = 0
    try:
        while True:
            read_size = (
                _gauged_chunk_size(self.upload_pacer) if gauge is not None
                else 1 << 20)
            chunk = await request.content.read(read_size)
            if not chunk:
                break
            if declared is not None and written + len(chunk) > declared:
                raise ValueError("body exceeds declared Content-Length")
            if cap is not None:
                await cap.pace(len(chunk))
            if gauge is not None:
                verdict = await gauge.sample()
                if gauge.alive:
                    await self.upload_pacer.pace_verdict(len(chunk), verdict, gauge.inflation_us)
                else:
                    gauge = None
            await loop.run_in_executor(None, fh.write, chunk)
            written += len(chunk)
        await loop.run_in_executor(None, fh.close)
    except Exception:
        try:
            fh.close()
        except Exception:
            pass
        raise
    return written

def _discard_chunked_upload(self, dest: str, part_path: str) -> None:
    """Drop a chunked transfer's tracking entry and its on-disk .part file."""
    self._chunked_uploads.pop(dest, None)
    try:
        os.remove(part_path)
    except OSError:
        pass

def _expire_stale_chunked_uploads(self) -> None:
    """Reap transfers idle past UPLOAD_PART_TTL_SECONDS (entry + .part file)."""
    now = time.monotonic()
    for key in [k for k, s in self._chunked_uploads.items()
                if now - s["ts"] > UPLOAD_PART_TTL_SECONDS and not s["busy"]]:
        stale = self._chunked_uploads.pop(key)
        try:
            os.remove(stale["part"])
        except OSError:
            pass
        logger.debug(f"Expired stale chunked upload: {key}")

async def handle_upload(self, request: web.Request) -> web.Response:
    """Stream a client file upload to the file-manager directory over HTTP.

    Available in every streaming mode and not bounded by the data-channel /
    WebSocket per-message size, so it saturates the link where the per-chunk
    SCTP path cannot. The destination path (relative to the file-manager
    root) arrives URL-encoded in the X-Upload-Path header; the body streams
    straight to disk on the executor, so the event loop keeps serving the
    stream during the transfer. Path safety mirrors the data-channel path:
    no traversal outside the root, and O_NOFOLLOW blocks a planted symlink.

    Two request shapes share the endpoint:

    - Plain: one POST carrying the whole file, no chunk headers — staged in a
      hidden sibling of the destination and renamed onto it when the body is
      complete.
    - Chunked (the client slices files above its 64 MiB threshold so no
      single request body exceeds a fronting proxy's per-request cap, e.g.
      Cloudflare's 100 MB): sequential POSTs for the same X-Upload-Path,
      each also carrying
        X-Upload-Id:     opaque client-chosen transfer id
        X-Upload-Offset: absolute byte offset of this slice
        X-Upload-Total:  final file size in bytes
        X-Upload-Final:  "1" on the last slice
      Slices accumulate in the staging sibling this destination derives
      (_upload_staging_path). Offset 0 (re)creates it — which is also how a
      stale one from an abandoned transfer for the same path gets replaced —
      and non-zero offsets must exactly continue the tracked transfer (same
      id, offset equal to the bytes already banked, matching staged size) or
      the transfer is discarded with 409. The final slice validates the
      accumulated size against X-Upload-Total and renames the staged file
      onto the destination atomically. Transfers idle past
      UPLOAD_PART_TTL_SECONDS are expired on the next chunked request.

    Both shapes carry the mode of the file they replace onto the replacement
    and are refused for view-only credentials (the view-only password, a
    viewer-role session token).
    """
    settings = request.app["settings"]
    rel = urllib.parse.unquote(request.headers.get("X-Upload-Path", "") or "")

    def failed(message: str, status: int = 400) -> web.Response:
        """Refuse the transfer, recording the attempt and why it was refused."""
        audit.emit("file.upload.error", filename=rel, error=message)
        return web.json_response({"status": "error", "message": message}, status=status)

    if self._viewer_ceiling(request):
        return failed("View-only credentials cannot upload files", 403)
    if "upload" not in settings.file_transfers:
        return failed("uploads disabled", 403)
    root = getattr(settings, "file_manager_path", "") or ""
    if not root:
        return failed("uploads disabled", 403)
    root = os.path.expanduser(root)
    sane = os.path.normpath(rel.strip("/\\"))
    parts = [c for c in sane.split(os.sep) if c and c != "."]
    if not parts or ".." in parts:
        return failed("invalid upload path")
    dest = os.path.join(root, *parts)
    name = "/".join(parts)
    real_root = os.path.realpath(root)
    parent = os.path.realpath(os.path.dirname(dest))
    try:
        within = os.path.commonpath([real_root, parent]) == real_root
    except ValueError:
        within = False
    if not within:
        return failed("path escape rejected")
    try:
        os.makedirs(parent, exist_ok=True)
    except OSError as e:
        return failed(f"mkdir failed: {e}", 500)

    upload_id = request.headers.get("X-Upload-Id")
    offset_header = request.headers.get("X-Upload-Offset")
    if (upload_id is None) != (offset_header is None):
        return failed("X-Upload-Id and X-Upload-Offset must be sent together")

    if upload_id is None:
        staging = _upload_staging_path(dest, os.urandom(8).hex())
        try:
            written = await self._stream_upload_body(request, staging, append=False)
        except Exception as e:
            try:
                os.remove(staging)
            except OSError:
                pass
            return failed(str(e))
        _carry_destination_mode(staging, dest)
        try:
            os.replace(staging, dest)
        except OSError as e:
            try:
                os.remove(staging)
            except OSError:
                pass
            return failed(f"finalize failed: {e}", 500)
        logger.info(f"HTTP upload finished: {dest} ({written} bytes)")
        audit.emit("file.upload.end", filename=name, size_bytes=written)
        return web.json_response({"status": "success", "bytes": written})

    try:
        offset = int(offset_header or "")
        total = int(request.headers["X-Upload-Total"]) if "X-Upload-Total" in request.headers else -1
    except ValueError:
        return failed("malformed chunk headers")
    if offset < 0 or ("X-Upload-Total" in request.headers and total < 0):
        return failed("malformed chunk headers")
    final = request.headers.get("X-Upload-Final") == "1"
    part_path = _upload_staging_path(dest, _upload_staging_token(dest))

    self._expire_stale_chunked_uploads()

    state = self._chunked_uploads.get(dest)
    if offset == 0:
        if state is not None and state["busy"]:
            return web.json_response(
                {"status": "error", "message": "another chunk for this path is in flight"},
                status=409,
            )
        state = {"id": upload_id, "offset": 0, "ts": time.monotonic(),
                 "part": part_path, "busy": False}
        self._chunked_uploads[dest] = state
    else:
        if state is not None and state["busy"]:
            # Never discard here: the in-flight writer owns the .part.
            return web.json_response(
                {"status": "error", "message": "another chunk for this path is in flight"},
                status=409,
            )
        try:
            part_size = os.path.getsize(part_path)
        except OSError:
            part_size = -1
        if (state is None or state["id"] != upload_id
                or state["offset"] != offset or part_size != offset):
            self._discard_chunked_upload(dest, part_path)
            return failed(f"chunk sequence mismatch at offset {offset}; transfer discarded", 409)

    state["busy"] = True
    try:
        written = await self._stream_upload_body(request, part_path, append=offset > 0)
    except Exception as e:
        self._discard_chunked_upload(dest, part_path)
        return failed(str(e))
    state["busy"] = False
    state["offset"] = offset + written
    state["ts"] = time.monotonic()

    if not final:
        return web.json_response({"status": "success", "bytes": state["offset"], "complete": False})

    received = state["offset"]
    if total >= 0 and received != total:
        self._discard_chunked_upload(dest, part_path)
        return failed(f"size mismatch: received {received}, expected {total}")
    _carry_destination_mode(part_path, dest)
    try:
        os.replace(part_path, dest)
    except OSError as e:
        self._discard_chunked_upload(dest, part_path)
        return failed(f"finalize failed: {e}", 500)
    self._chunked_uploads.pop(dest, None)
    logger.info(f"HTTP chunked upload finished: {dest} ({received} bytes)")
    audit.emit("file.upload.end", filename=name, size_bytes=received)
    return web.json_response({"status": "success", "bytes": received, "complete": True})

async def handle_print_document(self, request: web.Request) -> web.StreamResponse:
    """GET /api/print/<name>: hand a printed document to the page that
    prints it, and take it out of the spool once it went out whole.
    Refused to view-only credentials: printing is the session owner's act."""
    if not self.settings.printing_enabled[0]:
        return web.Response(status=403, text="Forbidden: printing disabled")
    if self._viewer_ceiling(request):
        return web.Response(status=403, text="View-only credentials cannot take printed documents")
    name = printing.document_name(request.match_info.get("name", ""))
    base = str(self.print_spool)
    full = os.path.normpath(os.path.join(base, name)) if name else ""
    if not full.startswith(base + os.sep) or not os.path.isfile(full):
        return web.Response(status=404, text="No such document")
    return _AuditedFileResponse(pathlib.Path(full), name, event="print.document", remove=True,
                                headers={"Content-Disposition": "inline"})

def _active_service(self) -> Optional[BaseStreamingService]:
    return self.services.get(self.current_mode) if self.current_mode else None

async def handle_sessions(self, request: web.Request) -> web.Response:
    """GET /api/sessions: the pages connected to the active transport."""
    service = self._active_service()
    return web.json_response({"sessions": await service.sessions() if service else []})

async def handle_session_delete(self, request: web.Request) -> web.Response:
    """DELETE /api/sessions/<id>: close that page's connection. Refused to
    view-only credentials, like every change to the session."""
    if self._viewer_ceiling(request):
        return web.Response(status=403, text="View-only credentials cannot disconnect a session")
    service = self._active_service()
    if service is None or not await service.disconnect_session(request.match_info["id"]):
        return web.Response(status=404, text="No such session")
    return web.Response(status=204)

async def handle_recording(self, request: web.Request) -> web.Response:
    """GET, POST and DELETE /api/recording: the MP4 recording pixelflux
    makes of the session, H.264 with the session's audio as an Opus track
    when audio is on and pcmflux is installed. POST starts one into the
    file-manager directory unless the body names a path, DELETE stops it
    and GET reports on the current or last one. One recording at a time;
    starting a second or stopping none is a conflict."""
    try:
        import pixelflux
    except ImportError:
        return web.Response(status=501, text="pixelflux is not installed")
    if request.method == "GET":
        return web.json_response(pixelflux.recording_status() or {"active": False})
    if self._viewer_ceiling(request):
        return web.Response(status=403, text="View-only credentials cannot record")
    try:
        if request.method == "POST":
            body = await request.json() if request.can_read_body else {}
            path = str((body or {}).get("path") or "")
            if not path:
                path = "recording-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + ".mp4"
            target = self._recording_target(path)
            if target is None:
                logger.warning(f"Refused a recording outside the file-manager directory: {path!r}")
                return web.Response(status=400, text="Recording path leaves the file-manager directory")
            path = target
            if (pixelflux.recording_status() or {}).get("active"):
                raise RuntimeError("a recording is already active")
            audio_socket = await self._start_recording_audio(pixelflux)
            args = (path, None, audio_socket) if audio_socket else (path,)
            try:
                status = await asyncio.to_thread(pixelflux.start_recording, *args)
            except RuntimeError:
                await self._stop_recording_audio()
                raise
            audit.emit("recording.start", filename=status.get("path", path))
        else:
            status = await asyncio.to_thread(pixelflux.stop_recording)
            await self._stop_recording_audio()
            audit.emit("recording.stop", filename=status.get("path", ""),
                       size_bytes=status.get("bytes", 0), duration_s=status.get("duration_s", 0.0),
                       frames=status.get("frames", 0))
    except json.JSONDecodeError:
        return web.Response(status=400, text="Body is not JSON")
    except RuntimeError as exc:
        logger.warning(f"Recording request refused: {exc}")
        return web.Response(status=409, text="Recording request refused; the server log names the reason")
    return web.json_response(status)

def _recording_target(self, name: str) -> Optional[str]:
    """Where a requested recording name writes, or `None` when it leaves the
    file-manager directory. The name is a path relative to that directory:
    an absolute path, a traversal segment and a symlinked parent pointing
    outside are all refused, since the session token that reaches this
    endpoint carries no authority over the rest of the filesystem."""
    base = os.path.realpath(self.upload_dir)
    if os.path.isabs(name):
        return None
    target = os.path.realpath(os.path.join(base, name))
    if target == base or os.path.commonpath([base, target]) != base:
        return None
    return target

async def _start_recording_audio(self, pixelflux: Any) -> str:
    """The Ogg Opus socket a recording's audio track is read from, served
    by a pcmflux capture of the session's sink that runs for the recorder
    alone with no Python callback, so no frame passes through Python.
    Empty for a video-only recording: audio off, a pixelflux or pcmflux
    without the socket, or a capture that does not start."""
    try:
        if not self.settings.audio_enabled[0] or \
                "audio_socket" not in inspect.signature(pixelflux.start_recording).parameters:
            return ""
        from pcmflux import AudioCapture
        from .audio_control import ensure_capture_sink, opus_capture_settings
        capture_settings = opus_capture_settings(self.settings.audio_device_name, self.settings.audio_channels,
                                                 int(self.settings.audio_bitrate), 20.0)
        runtime = os.environ.get("XDG_RUNTIME_DIR") or tempfile.gettempdir()
        capture_settings.output_socket = os.path.join(runtime, f"selkies-record-audio-{os.getpid()}.sock")
    except (ImportError, AttributeError, ValueError):
        return ""
    capture = AudioCapture()
    try:
        await ensure_capture_sink(self.settings.audio_device_name)
        await asyncio.to_thread(capture.start_capture, capture_settings)
        if capture.state == "failed" or capture.last_error:
            raise RuntimeError(f"capture {capture.state}: {capture.last_error}")
    except (RuntimeError, OSError) as exc:
        logger.warning(f"Recording without audio: {exc}")
        await asyncio.to_thread(capture.stop_capture)
        return ""
    self._recording_audio = capture
    return capture_settings.output_socket

async def _stop_recording_audio(self) -> None:
    capture, self._recording_audio = self._recording_audio, None
    if capture is not None:
        await asyncio.to_thread(capture.stop_capture)

async def handle_screenshot(self, request: web.Request) -> web.Response:
    """GET /api/screenshot?display=<name>: a PNG of that display with the
    cursor drawn in, by pixelflux. On X11 the root, which holds every
    display."""
    try:
        import pixelflux
        screenshot = pixelflux.screenshot_png
    except (ImportError, AttributeError):
        return web.Response(status=501, text="pixelflux without screenshots")
    display = request.query.get("display", "primary")
    output = 0
    if self.settings.wayland[0] and display != "primary":
        from .display_utils import wayland_output_id
        output = wayland_output_id(display)
    try:
        png = await asyncio.to_thread(screenshot, output)
    except RuntimeError as exc:
        logger.warning(f"Screenshot of display {display!r} refused: {exc}")
        return web.Response(status=404, text="No such display")
    return web.Response(body=png, content_type="image/png")

def pending_print_documents(self) -> List[Tuple[str, int]]:
    """The documents no page has taken yet, for a page that connects now."""
    return printing.pending(str(self.print_spool)) if self.print_watcher else []

async def _announce_print_document(self, name: str, size: int) -> None:
    service = self.services.get(self.current_mode) if self.current_mode else None
    if service is not None:
        await service.announce_print_document(name, size)