ImpTune/imptune/services/session.py

"""Cookie-scoped Owner session — opaque bearer key, no accounts."""
from __future__ import annotations

import secrets

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

import imptune.config as cfg
from imptune.db.models import Owner

COOKIE_NAME = "imptune_owner_key"
COOKIE_MAX_AGE = 10 * 365 * 24 * 60 * 60  # 10 years

# Paths that never need an Owner. Minting one for them means a DB write per
# request for any client that doesn't carry the cookie — asset fetches racing
# the first page load, health probes, crawlers, favicon hunts — and every row
# is dead weight, since only a browser holding the cookie can ever use it.
_UNSCOPED_PATHS = frozenset({"/health", "/favicon.ico", "/robots.txt"})
_UNSCOPED_PREFIXES = ("/static/",)


def generate_key() -> str:
    return secrets.token_urlsafe(32)


def cookie_kwargs() -> dict:
    """`set_cookie` attributes for the owner key — persistent only when Secure.

    With `COOKIE_SECURE=false` the key travels over plain HTTP, so persisting it
    for ten years would leave a long-lived bearer credential on disk and in
    cleartext traffic. Instead `max_age` is dropped: the browser holds the cookie
    in memory only. The app stays fully usable and keeps remembering printers,
    configs and clients for as long as the window is open — the session just
    ends when the browser does, and the UI says so (see `ephemeral_session`).
    The backup key remains the way to carry a session across browsers.
    """
    kwargs: dict = {
        "httponly": True,
        "samesite": "lax",
        "secure": cfg.COOKIE_SECURE,
    }
    if cfg.COOKIE_SECURE:
        kwargs["max_age"] = COOKIE_MAX_AGE
    return kwargs


def single_user_owner() -> Owner:
    """The one Owner every request shares under `COOKIE_SECURE=single_user`.

    No cookie is read, so the owner has to be findable from the DB alone: the
    oldest row wins. That way a deployment that used to run cookie-scoped keeps
    the printers it already had instead of waking up empty, and repeat requests
    never mint a second row. It is marked permanent so nothing treats it as a
    throwaway first-visit session.
    """
    owner = Owner.select().order_by(Owner.id).first()
    if owner is None:
        owner = Owner.create(key=generate_key(), is_permanent=True)
    return owner


def _request_hosts(request: Request) -> set[str]:
    """Every host spelling that legitimately identifies this deployment."""
    hosts = {request.url.netloc}
    for header in ("host", "x-forwarded-host"):
        value = request.headers.get(header)
        if value:
            # X-Forwarded-Host may be a proxy chain: the client-facing host is first.
            hosts.add(value.split(",")[0].strip())
    return hosts


def is_same_origin(request: Request) -> bool:
    """Origin/Referer check — the CSRF guard for /session/restore.

    Restoring a key re-points the cookie at a *different* Owner, so unlike
    the rest of the app's unprotected POSTs (which only ever mutate the
    caller's own data), a forged cross-site POST here is a login-CSRF /
    session-fixation vector: an attacker who knows their own key can force
    a victim's browser onto the attacker's Owner. Browsers always send
    Origin (and usually Referer) on form POSTs, same-site or not, so
    requiring a match — and rejecting when both are absent — blocks a plain
    auto-submitting HTML form without needing a token.

    Only the *host* is compared, not the scheme: behind a TLS-terminating
    proxy the browser sends `Origin: https://host` while uvicorn sees
    `http` (it only trusts X-Forwarded-Proto from `forwarded_allow_ips`,
    which excludes a proxy in a sibling container), so a scheme comparison
    rejected every legitimate restore in production. A same-host attacker
    origin is not a capability the scheme check was buying.
    """
    from urllib.parse import urlparse

    hosts = _request_hosts(request)

    origin = request.headers.get("origin")
    if origin is not None:
        return urlparse(origin).netloc in hosts

    referer = request.headers.get("referer")
    if referer:
        return urlparse(referer).netloc in hosts

    return False


class OwnerSessionMiddleware(BaseHTTPMiddleware):
    """Resolves request.state.owner from a cookie, creating one on first visit."""

    async def dispatch(self, request: Request, call_next):
        path = request.url.path
        if path in _UNSCOPED_PATHS or path.startswith(_UNSCOPED_PREFIXES):
            return await call_next(request)

        if cfg.SINGLE_USER:
            # No cookie is read and none is set: identity comes from the DB, so
            # onboarding ("keep your printers?") and the ephemeral-cookie
            # warning are both meaningless here.
            request.state.owner = single_user_owner()
            request.state.is_new_owner = False
            request.state.ephemeral_session = False
            request.state.single_user = True
            return await call_next(request)

        key = request.cookies.get(COOKIE_NAME)
        owner = Owner.get_or_none(Owner.key == key) if key else None
        is_new = owner is None
        if owner is None:
            owner = Owner.create(key=generate_key(), is_permanent=False)

        request.state.owner = owner
        request.state.is_new_owner = is_new
        request.state.single_user = False
        # Templates warn about it; see cookie_kwargs().
        request.state.ephemeral_session = not cfg.COOKIE_SECURE

        response = await call_next(request)

        # Routes like /session/restore intentionally set this cookie themselves
        # (to a *different* owner than the one this middleware just minted) —
        # don't clobber that with the auto-provisioned one.
        route_already_set_cookie = any(
            header.lower() == b"set-cookie" and value.startswith(f"{COOKIE_NAME}=".encode())
            for header, value in response.raw_headers
        )

        if is_new and not route_already_set_cookie:
            response.set_cookie(COOKIE_NAME, owner.key, **cookie_kwargs())

        return response