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.tokens_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(
        printing.spool_path(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.minimum_version = ssl.TLSVersion.TLSv1_2
    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(
    request: web.Request, 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.
    Only a page load gets the Basic scheme: a browser holds a background request
    (any `Sec-Fetch-Mode` but a navigation's) open behind its login prompt, which a
    fullscreen session hides, so the page stalled. Answered at once, the web
    client's 401 guard reloads the page and the prompt comes with the page itself.
    A client that sends no `Sec-Fetch-Mode` gets the Basic scheme.
    """
    mode = request.headers.get("Sec-Fetch-Mode")
    scheme = "Basic" if mode in (None, "navigate", "nested-navigate") else "PageLogin"
    return web.Response(
        status=401,
        headers={
            "WWW-Authenticate": f'{scheme} 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 a
    page opened with one carries on the URLs it navigates to rather than
    fetches (a file the manager downloads), which a page holding its token
    in the fragment leaves to the cookie; 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

@classmethod
def _fetched_cross_site(cls, request: web.Request, settings: Any) -> bool:
    """Whether a browser sent this request for another site's page.

    A GET from another site's image, link, or form carries no Origin but
    does carry `Sec-Fetch-Site`, and the browser attaches a cached Basic
    login to it, so a route whose GET changes something holds it to this:
    an Origin the operator allows (`_is_origin_allowed`) still passes, and
    clients that are not browsers send neither header.
    """
    if request.headers.get("Sec-Fetch-Site") in ("cross-site", "same-site"):
        return not (request.headers.get("Origin") and cls._is_origin_allowed(request, settings))
    return not cls._is_origin_allowed(request, settings)

@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

@staticmethod
def _socket_cookie_value(role: str, expires: int) -> str:
    """The socket cookie for `role` ("controller" or "viewer") until `expires` (epoch seconds)."""
    body = f"{role}.{expires}"
    return f"{body}.{hmac.new(_SOCKET_COOKIE_KEY, body.encode(), hashlib.sha256).hexdigest()}"

@classmethod
def _socket_cookie_role(cls, request: web.Request) -> Optional[Tuple[str, int]]:
    """The role a request's socket cookie grants and when it ends, or None for none, one past its time, or
    one this process did not sign."""
    value = request.cookies.get(SOCKET_COOKIE, "")
    parts = value.split(".")
    if len(parts) != 3 or parts[0] not in ("controller", "viewer") or not parts[1].isdigit():
        return None
    expires = int(parts[1])
    if expires <= time.time() or not hmac.compare_digest(cls._socket_cookie_value(parts[0], expires), value):
        return None
    return parts[0], expires

@classmethod
def _renew_socket_cookie(cls, request: web.Request, response: web.StreamResponse, role: str) -> None:
    """Answer a Basic-authenticated request with the socket cookie for its `role` where it carries none for
    that role with more than half its time left: HttpOnly, SameSite=Strict, under the route prefix, Secure
    over https. A page's load sets it for its first handshake, and the reconnect's probe for the next."""
    held = cls._socket_cookie_role(request)
    if held is not None and held[0] == role and held[1] - time.time() > SOCKET_COOKIE_S / 2:
        return
    expires = int(time.time()) + SOCKET_COOKIE_S
    secure = request.secure or (
        request.headers.get("X-Forwarded-Proto", "").split(",")[0].strip().lower() == "https")
    response.set_cookie(
        SOCKET_COOKIE, cls._socket_cookie_value(role, expires), max_age=SOCKET_COOKIE_S,
        path=(request.app["settings"].subfolder or "") + "/", httponly=True, samesite="Strict", secure=secure)

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, or in secure mode reaches an API route with a Basic
    login and no 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")

    # In secure mode the token decides an API request's role: a Basic login
    # without one counts as a viewer, as a proxy in front may send the session's
    # own login for every participant (SealSkin's rooms).
    token_bound = bool(settings.master_token and not is_ws_handshake and not token_path
                       and path.startswith(f"{api_prefix}/api/"))
    if token_bound:
        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 "):
        held = self._socket_cookie_role(request) if is_ws_handshake else None
        if held is not None:
            # WebKit attaches no Basic credentials to a handshake: the cookie the page's
            # Basic-authenticated requests were answered with stands for them, at its role.
            request["auth_role_ceiling"] = held[0]
            return await handler(request)
        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(request, "Authorization Required")
    try:
        auth_decoded = self._b64_decode(auth_header[6:])
        if ":" not in auth_decoded:
            return self._basic_auth_challenge(request)
        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)
        request["auth_role_ceiling"] = (
            "viewer" if (viewonly_ok and not main_ok) or token_bound else "controller"
        )
    except Exception:
        return self._basic_auth_challenge(request)
    response = await handler(request)
    if not is_ws_upgrade and request.method in ("GET", "HEAD") and not response.prepared:
        self._renew_socket_cookie(request, response, request["auth_role_ceiling"])
    return response

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 before it starts (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 unset
    paint-over resolve, since its default depends on the resolved encoder
    (JPEG keeps it under Turbo), the order `_post_process_settings` keeps.

    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_paint_over_default()
        self.settings.resolve_cpu_encode_threading_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. `ui_layout` and the layouts the operator
    allows are surfaced for the same reason: the dashboard paints its
    layout before the stream connects, and a page that painted one and
    switched to the operator's on connect would flash, or offer a layout
    toggle that a locked layout hides.
    """
    return {
        "current_mode": self.current_mode,
        "available_modes": list(self.services.keys()),
        "enable_dual_mode": bool(
            getattr(self.settings, "enable_dual_mode", (False,))[0]
        ),
        "ui_layout": getattr(self.settings, "ui_layout", "default"),
        "ui_layout_allowed": enum_allowed("ui_layout"),
    }

@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).

    Every endpoint that changes host or session state refuses those
    requests, the way the streaming plane refuses input from a viewer, and
    so does every one that hands out more than a viewer watches: the files
    under the file-manager directory and printed documents. What the
    session shows anyway (its pages, a screenshot, whether a recording
    runs) stays available.
    """
    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, leaf: str, dir_fd: int, append: bool) -> int:
    """Stream a request body into the file `leaf` names in the open
    directory `dir_fd`, opening, writing, and closing it on the executor.

    Creates the file when ``append`` is False, refusing one that already
    exists, and appends when True. `O_NOFOLLOW` blocks a symlink planted
    over the leaf's name, and no component above it is re-walked by name:
    the `dir_fd` is the descriptor the path confinement opened, so a
    symlink swapped in over an intermediate component cannot redirect the
    write outside the root. 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.

    Args:
        request: The upload request whose body is streamed.
        leaf: The staging file's name in the directory `dir_fd` opens.
        dir_fd: The held descriptor of the destination's directory.
        append: Append to the file instead of creating it.

    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_EXCL)
    fh = await loop.run_in_executor(
        None, lambda: os.fdopen(os.open(leaf, flags, 0o644, dir_fd=dir_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:
            await loop.run_in_executor(None, 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.

    The entry and the file it names go together with nothing awaited
    between, so a restart on the same path cannot slip its own entry in
    between and lose it; the removal blocks the loop the once a discard
    costs, which a transfer's slices do not.
    """
    self._chunked_uploads.pop(dest, None)
    _remove_quietly(part_path)

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"]]:
        self._discard_chunked_upload(key, self._chunked_uploads[key]["part"])
        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
    to disk on the executor, and so do the directory creation, the staging
    open, and the plain path's rename onto the destination, since a network
    filesystem takes tens of milliseconds over each and a directory of
    small files pays it once per file, which the event loop would otherwise
    take from the stream it serves. A chunked transfer is tracked by its
    destination, so its finalize and its discards stay synchronous to keep
    the rename and the entry's removal atomic against a restart on the same
    path; each is a once-per-transfer cost, not a per-slice one. Path safety
    mirrors the data-channel path: no traversal outside the root, and the
    write lands in a directory descriptor the confinement walked and
    verified, with `O_NOFOLLOW` on the staging leaf itself.

    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 a staging sibling that offset 0 creates
      (_upload_staging_path) and the transfer's tracking entry names; offset
      0 also removes the one an abandoned transfer to the same path left,
      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 a file of this user's they replace onto
    the replacement (`_carry_destination_mode`) 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.normpath(os.path.expanduser(root))
    parts = self._file_parts(rel)
    if not parts:
        return failed("invalid upload path")
    dest = os.path.normpath(os.path.join(root, *parts))
    if not dest.startswith(os.path.join(root, "")):
        return failed("invalid upload path")
    # The destination's name in its directory, which the rename below acts
    # on through the directory's held descriptor.
    leaf = os.path.basename(dest)
    name = "/".join(parts)

    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")

    total = -1
    if upload_id is not None:
        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")

    # The write lands in a directory held open by the confinement
    # (`_open_dir_fd`): every component down to it was opened relative to
    # the one before and verified from the descriptor it opened to, so a
    # symlink planted over an intermediate name after the request's path
    # was validated cannot redirect the body or the final rename outside
    # the root, and `O_NOFOLLOW` on the staging leaf does the same for the
    # file itself.
    try:
        parent_fd = await asyncio.to_thread(self._open_dir_fd, parts[:-1], True)
    except FileRootEscape:
        return failed("path escape rejected")
    except OSError as e:
        return failed(f"mkdir failed: {e.strerror}", 500)
    try:
        if upload_id is None:
            staging = _staging_leaf_name()
            try:
                written = await self._stream_upload_body(
                    request, staging, parent_fd, append=False)
            except Exception as e:
                await asyncio.to_thread(_remove_quietly, staging, parent_fd)
                return failed(str(e))
            try:
                await asyncio.to_thread(_finalize_upload, staging, leaf, parent_fd)
            except OSError as e:
                await asyncio.to_thread(_remove_quietly, staging, parent_fd)
                return failed(f"finalize failed: {e.strerror}", 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})

        self._expire_stale_chunked_uploads()
        final = request.headers.get("X-Upload-Final") == "1"
        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,
                )
            if state is not None:
                self._discard_chunked_upload(dest, state["part"])
            state = {"id": upload_id, "offset": 0, "ts": time.monotonic(),
                     "part": _upload_staging_path(dest), "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(state["part"]) if state is not None else -1
            except OSError:
                part_size = -1
            if (state is None or state["id"] != upload_id
                    or state["offset"] != offset or part_size != offset):
                if state is not None:
                    self._discard_chunked_upload(dest, state["part"])
                return failed(f"chunk sequence mismatch at offset {offset}; transfer discarded", 409)
        part_path = state["part"]
        part_leaf = os.path.basename(part_path)

        # Claimed before the body's awaits so a slice for this path that arrives
        # meanwhile is refused, not interleaved onto the shared .part.
        state["busy"] = True
        try:
            written = await self._stream_upload_body(
                request, part_leaf, parent_fd, 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}")
        # Synchronous, like the removal: a transfer is tracked by its destination,
        # so a restart on the same path cannot run between the rename and the entry's removal.
        try:
            _finalize_upload(part_leaf, leaf, parent_fd)
        except OSError as e:
            self._discard_chunked_upload(dest, part_path)
            return failed(f"finalize failed: {e.strerror}", 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})
    finally:
        try:
            os.close(parent_fd)
        except OSError:
            pass

@staticmethod
def _file_parts(rel: str) -> Optional[List[str]]:
    """Split a root-relative path into its name components, or None when
    it escapes the root. Empty and ``.`` components are dropped, and a
    ``..`` surviving the normalization is the escape; a path that
    normalizes to nothing is no components, which is the root itself. A
    NUL byte names nothing on the filesystem and is refused with the
    escapes, rather than raising out of the calls that would see it."""
    if "\x00" in rel:
        return None
    rel = rel.strip("/\\")
    parts = [c for c in os.path.normpath(rel).split(os.sep) if c and c != "."]
    if rel and ".." in parts:
        return None
    return parts

def _resolve_file_path(self, rel: str) -> Optional[str]:
    """Resolve a root-relative path to a filesystem path at or under the
    file-manager directory. Returns None when the path escapes the root. An
    empty path resolves to the root itself, which browse may name and mkdir
    refuses through `_at_file_root`.

    The confinement is the `realpath`-then-prefix comparison the static
    path checker recognizes, the prefix suffixed with the separator and
    the root itself allowed, so a sibling whose name extends the root's is
    not read as inside it. The endpoints that act rather than list do not
    act on the returned string: they open the directory through
    `_open_dir_fd` and `_open_resolved_fd`, whose descriptor holds the
    object this check validated."""
    parts = self._file_parts(rel)
    if parts is None:
        return None
    base = str(self.upload_dir)
    if not parts:
        return base
    parent = os.path.realpath(os.path.join(base, *parts[:-1]))
    full = os.path.join(parent, parts[-1])
    # After following in-directory symlinks the entry must still sit at or
    # under the root; a leaf symlink out of the tree is refused too.
    root = os.path.join(base, "")
    for resolved in (parent, os.path.realpath(full)):
        if resolved != base and not resolved.startswith(root):
            return None
    return full

def _open_dir_fd(self, parts: List[str], create: bool = False) -> int:
    """Open the directory `parts` names under the file-manager root as a
    file descriptor, descending one component per `os.open`, each named
    relative to the descriptor before it and verified from the descriptor
    it opened to rather than by the name it went in under.

    A symlinked component is followed the way the resolved string path
    followed it, and what makes that safe where the string was not is the
    verification: `/proc/self/fd` names where each open really landed, the
    kernel's own record of the object, which a name swapped after the open
    cannot change. A component that resolves outside the root — planted
    before the walk or in the instant of it — ends the walk with
    `FileRootEscape` instead of handing the caller an object to act
    outside with. With `create`, a missing component is made in place
    first, the way `os.makedirs` made the parents of an upload or a new
    folder; what the walk itself created is inside the descriptor it came
    from, so it needs no verification. POSIX only: `dir_fd` does not exist
    on Windows, and neither does this server.

    Args:
        parts: Name components under the root; empty names the root.
        create: Make a missing intermediate directory.

    Returns:
        A descriptor for the directory, the caller's to close.

    Raises:
        FileRootEscape: A component resolved outside the root.
        OSError: A component is missing or is not a directory.
    """
    base = str(self.upload_dir)
    root = os.path.join(base, "")
    fd = os.open(base, os.O_RDONLY | os.O_DIRECTORY)
    try:
        for part in parts:
            if part in ("", ".", "..") or os.sep in part or (
                    os.altsep and os.altsep in part):
                raise ValueError(f"not a name component: {part!r}")
            try:
                opened = os.open(part, os.O_RDONLY | os.O_DIRECTORY, dir_fd=fd)
            except FileNotFoundError:
                if not create:
                    raise
                # A concurrent request may make it first; either way it is there now.
                with contextlib.suppress(FileExistsError):
                    os.mkdir(part, dir_fd=fd)
                opened = os.open(part, os.O_RDONLY | os.O_DIRECTORY, dir_fd=fd)
            try:
                resolved = os.readlink(f"/proc/self/fd/{opened}")
            except OSError:
                resolved = ""
            if resolved != base and not resolved.startswith(root):
                os.close(opened)
                raise FileRootEscape(part)
            os.close(fd)
            fd = opened
        return fd
    except BaseException:
        os.close(fd)
        raise