api/routes_auth.py

"""routes_auth.py — X2's auth leg: login / logout / me (EXIT-3a).

Password auth against the SAME `core/users` accounts the Streamlit `gate()` uses, so both
front-ends share one identity today. The amended D-3 (our own OIDC client against Authentik, with
a branded login page) lands when D-1/D-2 unblock (owner blocker B-4); this is the interim leg it
replaces, and it is deliberately the same account store so the migration is a swap of the
CREDENTIAL check, not of the user model.
"""
import base64 as _b64
import hmac
import os
import re as _re
import time
from collections import OrderedDict

from fastapi import APIRouter, Body, Depends, Request, Response

import aios_session
from deps import Session, err, require_session, users, perms

router = APIRouter(prefix="/api/v1/auth")

#: THE SERVER OWNS BACKOFF (X5): the login page shows an inline error and never paces itself, so
#: the pacing has to be here or it does not exist. Bounded so it cannot grow into a memory leak.
#: ⚠ HONEST LIMIT: per-PROCESS. Behind several workers an attacker gets this budget per worker,
#: and it resets on deploy. It raises the cost of online guessing against PBKDF2-200k; it is not
#: a substitute for the real thing, which arrives with Authentik (D-1) owning rate limits centrally.
#: ⚠ KEYED BY (TENANT, USERNAME), not by username. Two tenants can each have an 'admin' or a
#: 'boss', and a shared bucket would mean failed logins against tenant B lock the same name out of
#: tenant A — a cross-tenant coupling in the one wave whose point is that tenants cannot reach
#: each other. It is also a (weak) existence oracle: a 429 for a name you never tried here would
#: tell you somebody else did.
_FAILS = OrderedDict()
_MAX_TRACKED = 2048
_LOCKOUT_AFTER = 8
_LOCKOUT_SECONDS = 60


def _throttled(key, now=None):
    now = now or time.time()
    hit = _FAILS.get(key)
    if not hit:
        return 0
    count, last = hit
    if count < _LOCKOUT_AFTER:
        return 0
    remaining = int(_LOCKOUT_SECONDS - (now - last))
    return remaining if remaining > 0 else 0


def _note_failure(key, now=None):
    now = now or time.time()
    count, last = _FAILS.get(key, (0, 0.0))
    # A window that has fully elapsed starts the count again — a user who mistypes twice today
    # and twice next week is not an attacker.
    if now - last > _LOCKOUT_SECONDS:
        count = 0
    _FAILS[key] = (count + 1, now)
    _FAILS.move_to_end(key)
    while len(_FAILS) > _MAX_TRACKED:
        _FAILS.popitem(last=False)


def _clear_failures(key):
    _FAILS.pop(key, None)


def _public_user(user):
    """What the client is allowed to know about itself. Never a hash, a salt, or the epoch —
    the epoch is a server-side revocation handle and a client has no use for it."""
    return {"username": user.get("username", ""), "name": user.get("name", ""),
            "role": user.get("role", "user"),
            # Wave 18 (C1-TENANT, R1): the company this session is bound to — the client shows
            # it and tests assert the binding; the SERVER-side truth stays in the signed cookie.
            "tenant": str(user.get("tenant") or "royal-imports").strip().lower(),
            "bus": perms.allowed_bu_labels(user),
            "team_id": perms.scope_team_id(user),
            "agent": perms.scope_agent(user),
            "modules": sorted(perms.allowed_modules(user) or []) or "all",
            "landing": perms.landing_page(user),
            # Wave 14 C-AVATAR — the user's OWN photo (Settings preview + the shell chip).
            # Everyone else's rides workspace.userAvatars, keyed by display name.
            "avatar": user.get("avatar") or None}


#: Wave 14 C-AVATAR — the write-side wall. PNG/JPEG data URLs only; the client downscales to
#: <=128px before posting, and the server re-validates because a stored data URL is served to
#: every grid session verbatim.
_AVATAR_RE = _re.compile(r"^data:image/(png|jpeg);base64,([A-Za-z0-9+/=]+)$")
_AVATAR_MAX_BYTES = 64 * 1024


@router.post("/me/avatar")
def set_avatar(body: dict = Body(default=None),
               session: Session = Depends(require_session)):
    import core.store as _store
    raw = str((body or {}).get("dataUrl") or "")
    m = _AVATAR_RE.match(raw)
    if not m:
        raise err(400, "bad_avatar",
                  "expected a data:image/png;base64,... or image/jpeg data URL")
    try:
        blob = _b64.b64decode(m.group(2), validate=True)
    except Exception:
        raise err(400, "bad_avatar", "that data URL is not valid base64")
    if len(blob) > _AVATAR_MAX_BYTES:
        raise err(400, "bad_avatar",
                  f"avatar too large - {_AVATAR_MAX_BYTES // 1024}KB decoded max "
                  f"(downscale to 128px before posting)")
    if not _store.available():
        raise err(503, "store_unavailable", "the tenant store is unavailable")
    # session.uname is the canonical account handle; the session's user RECORD does not
    # carry a "username" key (it is the registry's dict key), so keying the write off the
    # record would silently no-op — caught by X7's fresh /me read.
    users.set_avatar(session.uname, raw)
    u2 = dict(session.user)
    u2["avatar"] = raw
    return {"user": _public_user(u2)}


@router.delete("/me/avatar")
def clear_avatar(session: Session = Depends(require_session)):
    import core.store as _store
    if not _store.available():
        raise err(503, "store_unavailable", "the tenant store is unavailable")
    users.set_avatar(session.uname, None)
    u2 = dict(session.user)
    u2.pop("avatar", None)
    return {"user": _public_user(u2)}


@router.post("/login")
def login(request: Request, response: Response, body: dict = Body(default=None)):
    """Wave 18 (C1-TENANT, R1): ONE login box — the ACCOUNT decides the tenant.

    The pre-wave flow validated a caller-posted tenant slug and minted the session for it, so
    the same credential could sign in to any registered tenant. Now the credential resolves
    FIRST (username or email — `users.verify` takes both) and the session binds to the tenant
    ON THE RECORD; the posted `tenant` field is accepted for wire-compat and ignored. An
    account whose tenant no longer resolves (deleted / suspended record) gets the same 401 as
    a bad password — "which tenants exist" is not the login form's question to answer.
    """
    body = body or {}
    uname = str(body.get("username") or "").strip().lower()
    pw = str(body.get("password") or "")

    # Throttle on the IDENTIFIER alone: the tenant is not known until the credential resolves,
    # and a per-(tenant, uname) key would let an attacker reset their budget by rotating slugs.
    throttle_key = ("*", uname)
    wait = _throttled(throttle_key)
    if wait:
        raise err(429, "too_many_attempts",
                  f"too many failed attempts — try again in {wait} seconds")

    user = users.verify(uname, pw) if (uname and pw) else None
    if not user:
        # CONSTANT-TIME-ISH 401. `users.verify` returns immediately for an unknown username (no
        # record, no PBKDF2) and burns ~200k iterations for a known one, so the response time
        # answers "does this account exist?" to anyone with a stopwatch. Burning one equivalent
        # hash on the failure path removes that oracle. It costs nothing on the happy path.
        try:
            users._hash(pw or "x", "00" * 16)
        except Exception:
            pass
        _note_failure(throttle_key)
        raise err(401, "invalid_credentials", "that username and password do not match")

    tenant = str(user.get("tenant") or "royal-imports").strip().lower()
    from harness import runtime
    try:
        runtime.get_runtime(tenant)
    except KeyError:
        _note_failure(throttle_key)
        raise err(401, "invalid_credentials", "that username and password do not match")

    _clear_failures(throttle_key)
    value, _claims = aios_session.mint(tenant, user["username"], int(user.get("epoch") or 0))
    aios_session.set_cookie(response, request, value)
    # Wave 19 (R4): the login stamp. THE RESOLVED username, never the posted identifier — the
    # person may have typed an email address and `users.verify` resolved it to the registry key;
    # stamping what was typed would write onto a key that does not exist. Fail-silent inside
    # `touch_login` and a no-op for the emergency-master identity (which has no record to stamp),
    # so the one thing this cannot do is turn a good credential into a failed sign-in.
    users.touch_login(user["username"])
    # `touch_login` also sets `last_active`, so tell the session resolver it has been seen —
    # otherwise the very next request stamps it again and the hourly throttle is off by one write
    # per sign-in.
    import deps as _deps
    _deps.note_active(user["username"])
    return {"user": _public_user(user)}


@router.post("/logout", status_code=204)
def logout(request: Request):
    """204 and the cookie is cleared. Deliberately NOT session-gated: logging out must work from
    an already-invalid session, or a user holding a broken cookie has no way to get rid of it.

    ⚠ THE COOKIE IS CLEARED ON THE RETURNED RESPONSE, not on an injected `response` param. When a
    handler RETURNS a Response object, FastAPI ships that object — anything written to the
    injected `response` is silently dropped. The first version of this route took `response:
    Response`, called `clear_cookie` on it, then returned a fresh `Response(204)`: the 204 was
    correct, no `Set-Cookie` was ever sent, and the session stayed alive after a "successful"
    logout. Caught by asserting `/auth/me` is 401 AFTER the logout rather than trusting the 204.

    ⚠ This clears the BROWSER's copy only — the signed value stays cryptographically valid until
    it expires, which is the honest cost of a stateless session. "Sign me out everywhere" is
    `core.users.bump_epoch` (a password change already does it); D2's Postgres session mirror
    makes per-device revocation possible and arrives with C-2.
    """
    out = Response(status_code=204)
    aios_session.clear_cookie(out, request)
    return out


@router.get("/me")
def me(session: Session = Depends(require_session)):
    return {"user": _public_user(session.user), "tenant": session.tenant}