"""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