api/aios_session.py

"""aios_session.py — X3: the signed, STATELESS session cookie (EXIT-3a, 2026-07-30).

    aios_session = base64url(json {t,u,e,i,x}) . base64url(HMAC-SHA256)

`t` tenant slug · `u` username · `e` the user's revocation epoch · `i` idle deadline ·
`x` absolute expiry. Signed with `AIOS_SESSION_SECRET` using stdlib `hmac` — no new dependency,
and nothing to store server-side, which is the whole point: EXIT-4 requires a process that holds
no per-tenant state, and a session TABLE is per-tenant state. D2's Postgres sessions-MIRROR (an
audit view, not the source of truth) arrives with C-2 and is deliberately deferred.

WHY STATELESS STILL REVOKES. A cookie nobody can delete sounds like a cookie nobody can revoke.
The answer is the `epoch` integer on the user record (`core.users.epoch` / `bump_epoch`): the
cookie carries the epoch it was minted under, every verification re-reads the account's current
epoch, and a password change or a deactivation bumps it — so every outstanding cookie for that
user dies at once, without a session table.

⚠ NO FALLBACK SECRET, EVER. With `AIOS_SESSION_SECRET` unset this module mints a RANDOM
per-process key: sessions then die on restart, which is an inconvenience. A hardcoded default
would instead let anyone who has read this repo forge a cookie for any user of any tenant — an
inconvenience is the correct trade against a forgery key. `EPHEMERAL_SECRET` records which
happened so the app can say so at startup and a gate can assert it.
"""
import base64
import hmac
import json
import os
import secrets
import time
from hashlib import sha256

COOKIE_NAME = "aios_session"
IDLE_SECONDS = 8 * 60 * 60           # D2: reissued on every authenticated request
ABSOLUTE_SECONDS = 30 * 24 * 60 * 60  # D2: a hard ceiling no amount of activity extends

_env_secret = os.environ.get("AIOS_SESSION_SECRET") or ""
EPHEMERAL_SECRET = not _env_secret
_SECRET = _env_secret.encode("utf-8") if _env_secret else secrets.token_bytes(32)


def _b64(raw: bytes) -> str:
    return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")


def _unb64(txt: str) -> bytes:
    return base64.urlsafe_b64decode(txt + "=" * (-len(txt) % 4))


def _sign(payload: bytes) -> str:
    return _b64(hmac.new(_SECRET, payload, sha256).digest())


def mint(tenant, username, epoch, now=None, absolute_expiry=None):
    """A fresh cookie value. `absolute_expiry` carries over on reissue so refreshing the idle
    window can never extend the 30-day ceiling — that would make the ceiling unreachable."""
    now = int(now if now is not None else time.time())
    claims = {"t": str(tenant), "u": str(username), "e": int(epoch),
              "i": now + IDLE_SECONDS,
              "x": int(absolute_expiry if absolute_expiry is not None
                       else now + ABSOLUTE_SECONDS)}
    payload = json.dumps(claims, separators=(",", ":"), sort_keys=True).encode("utf-8")
    return f"{_b64(payload)}.{_sign(payload)}", claims


def read(value, now=None):
    """Parse + verify a cookie value → claims dict, or None.

    None for EVERY failure mode — bad shape, bad signature, expired, unparseable. The caller
    answers 401 and must not learn which: an error that distinguishes "signature wrong" from
    "expired" tells a forger whether their key is right.
    """
    if not value or not isinstance(value, str) or value.count(".") != 1:
        return None
    body, sig = value.split(".", 1)
    try:
        payload = _unb64(body)
    except Exception:
        return None
    # compare_digest on the SIGNATURE, always — a byte-wise early exit is a timing oracle.
    if not hmac.compare_digest(sig, _sign(payload)):
        return None
    try:
        claims = json.loads(payload.decode("utf-8"))
    except Exception:
        return None
    if not isinstance(claims, dict):
        return None
    if not isinstance(claims.get("u"), str) or not isinstance(claims.get("t"), str):
        return None
    if not claims["u"] or not claims["t"]:
        return None
    try:
        idle, absolute, epoch = int(claims["i"]), int(claims["x"]), int(claims["e"])
    except (KeyError, TypeError, ValueError):
        return None
    now = int(now if now is not None else time.time())
    if now >= idle or now >= absolute:
        return None
    claims["e"], claims["i"], claims["x"] = epoch, idle, absolute
    return claims


def is_secure_request(request):
    """Secure flag ON unless this is plain-HTTP localhost. Set unconditionally in production;
    unset on a local http:// dev origin, where a Secure cookie is simply never sent back and
    login would appear to succeed and then not work."""
    host = (request.headers.get("host") or "").split(":")[0].lower()
    proto = (request.headers.get("x-forwarded-proto")
             or request.url.scheme or "http").split(",")[0].strip().lower()
    if proto == "https":
        return True
    return host not in ("localhost", "127.0.0.1", "[::1]", "::1")


def set_cookie(response, request, value):
    response.set_cookie(
        COOKIE_NAME, value, httponly=True, samesite="lax",
        secure=is_secure_request(request), max_age=ABSOLUTE_SECONDS, path="/")


def clear_cookie(response, request):
    # Same attributes as when it was set: a browser matches on path/secure/samesite when
    # deleting, and a mismatched delete silently leaves the cookie in place.
    response.delete_cookie(COOKIE_NAME, path="/", httponly=True, samesite="lax",
                           secure=is_secure_request(request))