brain-of-reese/app/core/tokens.py

"""API-token service (phase 79, task 02).

Generates, stores, looks up and revokes the admin-issued access tokens
(``api_tokens`` rows — phase 79, task 01). The trust model (owner-locked
A4): the plaintext token (``bor_`` + 32 hex chars) exists only in the 201
response of the create call, returned **exactly once**; the row carries
the SHA-256 hex digest of the **full** token string and nothing else.

Lookup security — by hash, not by compare:
:func:`find_active_by_token` hashes the presented token and performs one
``token_hash ==`` lookup — a unique-index hit (``ix_api_tokens_token_hash``).
SHA-256's pre-image resistance means there is **no token-enumeration or
timing surface beyond the DB lookup itself**: an attacker holding the table
cannot turn a stored hash back into a working token, and a wrong candidate
simply misses the index. This is the deliberate contrast with
:func:`app.core.auth.check_password`'s constant-time compare — there IS
nothing to compare in constant time here, only to *look up*; replicating a
compare would be theatre, so the contrast is documented, not replicated.

House commit convention (the ``app.rag.sources_meta`` pattern): the service
functions flush but never commit — the calling endpoint owns the commit, so
a failed request can never leave a half-applied token mutation.
"""
from __future__ import annotations

import hashlib
import secrets
import uuid
from datetime import UTC, datetime

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models import ApiToken

#: The plaintext token's shape: fixed prefix + 32 hex chars
#: (``secrets.token_hex(16)`` — 128 bits of entropy).
TOKEN_PREFIX = "bor_"


def generate_token() -> str:
    """One fresh plaintext token: ``bor_`` + 32 hex chars (128-bit).

    The prefix is a human/parse marker only — the HASH covers the full
    string, so the prefix is never the secret.
    """
    return TOKEN_PREFIX + secrets.token_hex(16)


def hash_token(token: str) -> str:
    """The stored credential: the SHA-256 hex digest of the **full** token.

    Hashing the full string (not the suffix) means a stripped prefix can
    never collide with another token's hash. Deterministic — the same
    token always yields the same 64-hex digest (the unique-index key).
    """
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def create_token(db: Session, label: str) -> tuple[ApiToken, str]:
    """Create one named token; return the row + the plaintext ONCE.

    ``label`` is stripped before storing; non-emptiness is the API
    layer's job (the 422 boundary — the service trusts it, A4). The row
    only ever carries the hash (``token_hash``); the returned ``str`` is
    the one and only moment the plaintext exists outside this function
    (the endpoint ships it in the 201 body).

    Flushes, does not commit — the caller commits (so the row + its
    server-default ``created_at`` are durable only when the endpoint's
    response can be built).
    """
    plaintext = generate_token()
    row = ApiToken(label=label.strip(), token_hash=hash_token(plaintext))
    db.add(row)
    db.flush()
    return row, plaintext


def find_active_by_token(db: Session, token: str) -> ApiToken | None:
    """Resolve a presented plaintext token to its ACTIVE row, or ``None``.

    Hash → ``token_hash ==`` lookup → ``revoked_at IS NULL``. ANY other
    shape is a miss — there is no "almost" path: the hash of a malformed
    string (wrong prefix, truncated, empty, …) simply matches no row,
    so every failure mode returns the same ``None`` (the caller maps
    that to one generic 401 — no enumeration, A4's auth-error contract).
    """
    row = (
        db.execute(select(ApiToken).where(ApiToken.token_hash == hash_token(token)))
        .scalars()
        .first()
    )
    if row is None or row.revoked_at is not None:
        return None
    return row


def mark_used(tok: ApiToken) -> None:
    """Bump ``last_used_at`` to now (UTC) — the caller commits."""
    tok.last_used_at = datetime.now(UTC)


def revoke(db: Session, token_id: uuid.UUID) -> bool:
    """Stamp ``revoked_at`` (UTC now); return False when the row is missing.

    Idempotent: an already-revoked row keeps its ORIGINAL stamp (the
    revocation time is the first one, never re-stamped on a second
    call) and the call still returns True — the row exists and is dead.
    Returns False only for a missing id (the endpoint maps that to
    404 ``token not found``). Flushes, does not commit.
    """
    row = db.get(ApiToken, token_id)
    if row is None:
        return False
    if row.revoked_at is None:
        row.revoked_at = datetime.now(UTC)
    return True