src/endopath/auth.py

"""Per-user password authentication for the write endpoints (#128).

Replaces the single shared bearer token with real accounts. A reviewer logs in with a username and a
password; the server verifies it against a scrypt hash in the `users` table and issues a signed session
token. Every write then carries that token, and the server reads the confirmer and their role from the
authenticated account rather than from a self-declared field in the request body, so PRD section 4.4's
"who confirmed each value, and in what role" is non-repudiable rather than an honest guess.

Standard library only for the crypto (`hashlib.scrypt` for hashing, `hmac` for the token signature): the
serving image carries no auth dependency, and password hashing plus stateless token signing this
self-contained need none. `Identity` is the account an authenticated request resolves to.

Auth is enabled exactly when a session secret is configured (`ENDOPATH_SESSION_SECRET`), mirroring how the
old shared token gated the same endpoints. With no secret the gate is open, which keeps local dev and the
test suite free of ceremony; a public Space that sets no secret refuses to boot (see api.lifespan), so the
open default is never exposed to the internet.
"""

from __future__ import annotations

import base64
import hashlib
import hmac
import json
import os
import time
from typing import Optional

from pydantic import BaseModel, ConfigDict

SESSION_SECRET_ENV = "ENDOPATH_SESSION_SECRET"
# A JSON list of {username, password, role, holds_licence, display_name}, seeded into the users table at
# startup so a fresh deployment has accounts to log in with. Set as a Space Secret by deploy_space.py.
SEED_USERS_ENV = "ENDOPATH_SEED_USERS"

# A session lasts a working day, then the reviewer logs in again. Short enough that a leaked token expires,
# long enough not to interrupt a review sitting.
SESSION_TTL_SECONDS = 12 * 60 * 60

# scrypt work factors. n=2**15 keeps a single verify in the tens of milliseconds; maxmem is set explicitly
# (128 * n * r * p bytes, with headroom) so the call never trips OpenSSL's default memory cap.
_SCRYPT_N = 2**15
_SCRYPT_R = 8
_SCRYPT_P = 1
_SCRYPT_MAXMEM = 128 * _SCRYPT_N * _SCRYPT_R * _SCRYPT_P * 2
_SCRYPT_DKLEN = 32


class Identity(BaseModel):
    """The account an authenticated request resolves to. `role` and `holds_licence` are attributes of the
    account, set when it is provisioned, not values the client sends: that is what makes a confirmation's
    recorded role trustworthy (PRD section 4.4)."""

    model_config = ConfigDict(frozen=True)

    username: str
    role: str
    holds_licence: bool = False
    display_name: str = ""

    @property
    def confirmer(self) -> str:
        """The name recorded on a confirmation: the human display name where set, else the username."""
        return self.display_name.strip() or self.username


def session_secret() -> str:
    """The HMAC key the session tokens are signed with, read at call time so a Space Secret injected after
    import still takes effect (and a test can set or clear it). Empty when auth is disabled."""
    return os.environ.get(SESSION_SECRET_ENV, "").strip()


def auth_enabled() -> bool:
    """Whether write endpoints demand a logged-in account. True exactly when a session secret is set."""
    return bool(session_secret())


# --- password hashing ------------------------------------------------------


def hash_password(password: str) -> str:
    """A self-describing scrypt hash, `scrypt$n$r$p$salt_hex$hash_hex`, so `verify_password` can read back
    the parameters a hash was produced under and a later work-factor change does not strand old hashes."""
    salt = os.urandom(16)
    derived = hashlib.scrypt(
        password.encode("utf-8"),
        salt=salt,
        n=_SCRYPT_N,
        r=_SCRYPT_R,
        p=_SCRYPT_P,
        maxmem=_SCRYPT_MAXMEM,
        dklen=_SCRYPT_DKLEN,
    )
    return f"scrypt${_SCRYPT_N}${_SCRYPT_R}${_SCRYPT_P}${salt.hex()}${derived.hex()}"


def verify_password(password: str, stored: str) -> bool:
    """Constant-time check of a password against a stored `hash_password` string. False on any malformed
    stored value rather than raising, so a corrupt row denies access instead of crashing the login."""
    parts = stored.split("$")
    if len(parts) != 6 or parts[0] != "scrypt":
        return False
    _, n_s, r_s, p_s, salt_hex, hash_hex = parts
    try:
        expected = bytes.fromhex(hash_hex)
        derived = hashlib.scrypt(
            password.encode("utf-8"),
            salt=bytes.fromhex(salt_hex),
            n=int(n_s),
            r=int(r_s),
            p=int(p_s),
            maxmem=128 * int(n_s) * int(r_s) * int(p_s) * 2,
            dklen=len(expected),
        )
    except (ValueError, MemoryError):
        return False
    return hmac.compare_digest(derived, expected)


# --- stateless signed session tokens ---------------------------------------


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


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


def _sign(body: str, secret: str) -> str:
    return _b64e(hmac.new(secret.encode("utf-8"), body.encode("ascii"), hashlib.sha256).digest())


def issue_token(
    identity: Identity, secret: str, *, now: Optional[int] = None, ttl: int = SESSION_TTL_SECONDS
) -> str:
    """A `<payload>.<signature>` token carrying the account and an expiry, signed with the session secret.
    Stateless: no server-side session store, so any process holding the secret can verify it."""
    issued = int(now if now is not None else time.time())
    payload = {
        "u": identity.username,
        "r": identity.role,
        "l": bool(identity.holds_licence),
        "n": identity.display_name,
        "exp": issued + ttl,
    }
    body = _b64e(json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8"))
    return f"{body}.{_sign(body, secret)}"


def decode_token(token: str, secret: str, *, now: Optional[int] = None) -> Optional[Identity]:
    """The `Identity` a token resolves to, or None when it is missing, malformed, wrongly signed, or
    expired. The signature is checked before the payload is trusted, in constant time."""
    if not token or "." not in token:
        return None
    body, _, signature = token.partition(".")
    if not hmac.compare_digest(signature, _sign(body, secret)):
        return None
    try:
        payload = json.loads(_b64d(body))
    except (ValueError, json.JSONDecodeError):
        return None
    current = int(now if now is not None else time.time())
    if not isinstance(payload, dict) or current >= int(payload.get("exp", 0)):
        return None
    try:
        return Identity(
            username=payload["u"],
            role=payload["r"],
            holds_licence=bool(payload["l"]),
            display_name=payload.get("n", ""),
        )
    except (KeyError, TypeError):
        return None


# --- account provisioning input --------------------------------------------


class SeedUser(BaseModel):
    """One account to provision from the seed environment: the credentials and the account-bound role and
    licence. `password` is the plaintext to hash once, never stored."""

    username: str
    password: str
    role: str
    holds_licence: bool = False
    display_name: str = ""


def parse_seed_users(raw: str) -> list[SeedUser]:
    """The accounts named in `ENDOPATH_SEED_USERS`, a JSON list. Empty when unset. Raises ValueError on a
    payload that is present but not a JSON list of the expected shape, so a misconfigured secret fails
    loudly at startup rather than silently seeding nothing."""
    if not raw or not raw.strip():
        return []
    data = json.loads(raw)
    if not isinstance(data, list):
        raise ValueError(f"{SEED_USERS_ENV} must be a JSON list, got {type(data).__name__}")
    return [SeedUser(**item) for item in data]