authorization_api.py

"""HTTP surface for the authorization admission store.

Three endpoints, and deliberately no judgement of its own:

* ``POST /api/v0/authorizations/admit`` — the gate turns an approved decision
  and a sealed envelope into one spendable record.
* ``GET  /api/v0/authorizations/{id}`` — status, for anyone who may read.
* ``POST /api/v0/authorizations/{id}/consume`` — the addressed agent spends it,
  exactly once, for one stated action.

WHY THIS MODULE DECIDES ALMOST NOTHING

``authorization_store`` already answers who may admit, who may spend, and what
a refusal is called. The temptation in an adapter is to answer those a second
time -- a scope check at the door, an expiry check before the call -- and a
second answer is a second authority that can drift from the first. So the two
mutating endpoints authenticate the caller and hand the principal to the store
unexamined. A 403 from ``/admit`` is the store's ``not_gate_principal``, not a
decision made here.

That is not fastidiousness. ``main.has_scope`` expands wildcards, so a key
holding ``gate:*`` satisfies it; ``_check_gate_authority`` requires
``gate:admit_authorization`` by exact membership, specifically so that a
wildcard on an agent key cannot become the power to authorize that agent's own
work. Pre-checking with the looser rule and then calling the store would leave
the endpoint's *answer* correct while making the adapter say yes to something
the store exists to refuse. The next person to add a fast path would inherit
that as precedent.

THE ONE CHECK THIS MODULE DOES OWN

``read_authorization`` is a bare SELECT with no authority check at all, by
design -- it grants nothing and spends nothing. That makes this module the only
thing standing between an authorization id and the sealed envelope, which
carries allowed paths, budgets, and the nonce. So the read endpoint owns its
scope check, and returns an allowlisted projection rather than the row.

THE CLOCK IS THE SERVER'S

``now`` is taken from the injected clock and never from the request. An expiry
the caller can move is not an expiry. Callers cannot smuggle one either: the
envelope contract refuses unknown fields, so a payload carrying ``now`` is
malformed rather than helpful.
"""

from __future__ import annotations

import time
import uuid
from datetime import datetime, timezone
from typing import Any, Callable, Dict, List, Optional

from fastapi import APIRouter, Body, Depends, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel

from authorization_envelope import AuthorizationError, Expectation
from authorization_store import (
    AuditRecorder,
    Principal,
    admit_authorization,
    consume_authorization,
    read_authorization,
)
from temporal_contract import system_temporal_profile
from temporal_ledger import record_temporal_mutation


# Refusal code -> HTTP status. Everything absent is 422: the request was
# understood and refused on its merits, which is the common case here.
#
# The 503s are the exception worth naming. `audit_unavailable` and
# `audit_not_independent` are not the caller's fault -- they mean this server
# cannot record attempts durably, so it refuses to act at all. Reporting those
# as 4xx would tell a caller to fix a request that was never the problem.
REFUSAL_STATUS: Dict[str, int] = {
    "not_gate_principal": 403,
    "recipient_mismatch": 403,
    "unknown_gate_decision": 404,
    "unknown_authorization": 404,
    "already_admitted": 409,
    "authorization_id_conflict": 409,
    "nonce_reused": 409,
    "replayed_authorization": 409,
    "audit_unavailable": 503,
    "audit_not_independent": 503,
}
DEFAULT_REFUSAL_STATUS = 422

READ_SCOPE = "reader:authorizations"

# An allowlist, not a denylist. A column added to authorization_admissions
# later must not start appearing here because nobody remembered to exclude it;
# `envelope`, `nonce` and `consumed_expectation` are exactly the fields that
# would be worst to leak by omission.
READABLE_FIELDS = (
    "authorization_id",
    "task_id",
    "spec_sha256",
    "envelope_sha256",
    "gate_decision_id",
    "addressed_agent",
    "lane",
    "branch",
    "base_sha",
    "issued_at",
    "expires_at",
    "admitted_by",
    "admitted_at",
    "consumed_at",
    "consumed_by",
)


# `consume_authorization` returns the verified Envelope object, because holding
# it is the permission. Over HTTP that has to become a chosen projection: the
# consumer already has the envelope it just spent, so what it needs back is
# confirmation plus the operative grant it is now acting under -- paths,
# budgets, and the window. The nonce is never echoed, spent or not; it is
# single-use material, not a receipt.
CONSUMED_FIELDS = ("authorization_id", "status", "consumed_by", "consumed_at")
GRANT_FIELDS = (
    "requested_action",
    "lane",
    "branch",
    "base_sha",
    "allowed_paths",
    "risk_tier",
    "max_tokens",
    "max_seconds",
    "tests",
    "stop_conditions",
    "depends_on",
)


def _consumed_body(result: Dict[str, Any]) -> Dict[str, Any]:
    body = {name: result[name] for name in CONSUMED_FIELDS if name in result}
    envelope = result.get("envelope")
    if envelope is None:
        return body
    body["envelope_sha256"] = getattr(envelope, "envelope_sha256", None)
    body["expires_at"] = getattr(envelope, "expires_at", None)
    grant: Dict[str, Any] = {}
    for name in GRANT_FIELDS:
        value = getattr(envelope, name, None)
        if value is None:
            continue
        grant[name] = list(value) if isinstance(value, tuple) else value
    body["grant"] = grant
    return body


class ExpectationIn(BaseModel):
    agent: str
    lane: str
    action: str
    branch: str
    base_sha: str
    intended_paths: List[str] = []


class ConsumeIn(BaseModel):
    expected_envelope_sha256: str
    expected_task_id: str
    expected_spec_sha256: str
    expectation: ExpectationIn


def _refusal(error: AuthorizationError) -> JSONResponse:
    """One shape for every refusal, with the store's code preserved verbatim.

    The code is the stable part of this API. Callers branch on
    ``refusal_code``; the message is for humans and may be reworded.
    """
    body: Dict[str, Any] = {
        "refusal_code": error.code,
        "detail": str(error),
    }
    field = getattr(error, "field", None)
    if field:
        body["field"] = field
    return JSONResponse(
        status_code=REFUSAL_STATUS.get(error.code, DEFAULT_REFUSAL_STATUS),
        content=body,
    )


def _store_principal(caller: Any) -> Principal:
    """The caller as the store expects it. No scopes are added or expanded."""
    return Principal(agent_id=caller.agent_id, scopes=tuple(caller.scopes or ()))


def _may_read(caller: Any) -> bool:
    """Deliberately narrower than ``main.has_scope``.

    Only the exact read scope or a full reader wildcard. `writer:*` grants
    nothing here: being able to write threads is not a reason to be handed the
    shape of somebody's authorization.
    """
    scopes = set(caller.scopes or ())
    return READ_SCOPE in scopes or "reader:*" in scopes


def _recorded_at(now: Any) -> datetime:
    if isinstance(now, (int, float)) and not isinstance(now, bool):
        try:
            return datetime.fromtimestamp(float(now), timezone.utc)
        except (OverflowError, OSError, ValueError):
            pass
    return datetime.now(timezone.utc)


class _TemporalAuditRecorder(AuditRecorder):
    """Commit each authorization attempt and its temporal row together."""

    def __init__(
        self,
        cursor: Any,
        commit: Callable,
        rollback: Callable,
        temporal_recorder: Callable,
    ):
        super().__init__(cursor, lambda: None)
        self._atomic_commit = commit
        self._atomic_rollback = rollback
        self._temporal_recorder = temporal_recorder

    def record(
        self,
        kind: str,
        authorization_id: Optional[str],
        principal: Principal,
        outcome: str,
        refusal_code: Optional[str],
        detail: Optional[str],
        now: Any,
    ) -> None:
        operation_id = f"auth_attempt_{uuid.uuid4().hex}"
        occurred_at = _recorded_at(now)
        super().record(
            kind, authorization_id, principal, outcome, refusal_code, detail, now,
        )
        try:
            self._temporal_recorder(
                self.cursor,
                profile=system_temporal_profile(
                    now=occurred_at,
                    action_basis=f"authorization_attempt:{kind}:{outcome}",
                    discovery_basis="authorization_request_received_at",
                ),
                actor_id=principal.agent_id,
                operation_id=operation_id,
                entity_type="authorization_attempt",
                entity_id=operation_id,
                mutation_kind="create",
                source="authorization_api.audit",
                recorded_at=occurred_at,
            )
        except Exception:
            self._atomic_rollback()
            raise
        self._atomic_commit()


def mount_authorization_api(
    app: Any,
    db_factory: Callable[[], Any],
    principal_dependency: Callable[..., Any],
    clock: Callable[[], float] = time.time,
    temporal_recorder: Callable[..., Dict[str, Any]] = record_temporal_mutation,
) -> APIRouter:
    """Mount the three endpoints on ``app``.

    ``db_factory`` must return a *new* connection per call. The store refuses an
    audit recorder that shares the caller's connection, so the mutating paths
    open two and would fail closed rather than silently log into the same
    transaction they might roll back.
    """

    router = APIRouter(prefix="/api/v0/authorizations", tags=["authorization"])

    def _mutate(handler: Callable[[Any, AuditRecorder, float], Dict[str, Any]],
                success_status: int,
                project: Callable[[Dict[str, Any]], Dict[str, Any]] = dict,
                *,
                actor_id: str,
                entity_type: str,
                entity_id: str,
                mutation_kind: str,
                action_basis: str,
                ) -> JSONResponse:
        """Run one store call on its own connection, audited on a second.

        The audit connection is committed row by row inside the recorder, so it
        is not rolled back here even when the operation is: that is the point of
        it. The operation connection is rolled back on refusal so a partial
        statement cannot outlive the answer.
        """
        now = clock()
        occurred_at = _recorded_at(now)
        operation_id = f"auth_mutation_{uuid.uuid4().hex}"
        with db_factory() as conn:
            with db_factory() as audit_conn:
                with conn.cursor() as cur:
                    with audit_conn.cursor() as audit_cur:
                        audit = _TemporalAuditRecorder(
                            audit_cur, audit_conn.commit, audit_conn.rollback,
                            temporal_recorder,
                        )
                        try:
                            result = handler(cur, audit, now)
                        except AuthorizationError as error:
                            conn.rollback()
                            return _refusal(error)
                        try:
                            temporal_recorder(
                                cur,
                                profile=system_temporal_profile(
                                    now=occurred_at,
                                    action_basis=action_basis,
                                    discovery_basis="authorization_request_received_at",
                                ),
                                actor_id=actor_id,
                                operation_id=operation_id,
                                entity_type=entity_type,
                                entity_id=entity_id,
                                mutation_kind=mutation_kind,
                                source="authorization_api",
                                recorded_at=occurred_at,
                            )
                        except Exception:
                            conn.rollback()
                            raise
                        conn.commit()
        return JSONResponse(status_code=success_status, content=project(result))

    @router.post("/admit")
    def admit(request: Request,
              payload: Dict[str, Any] = Body(...),
              caller: Any = Depends(principal_dependency)) -> JSONResponse:
        """Admit one envelope. Only the human gate can succeed here.

        No scope check precedes the store call, and that is not an omission --
        see the module docstring. The payload is passed through exactly as
        received so the envelope contract can refuse unknown fields; parsing it
        into a model here would silently drop them.
        """
        principal = _store_principal(caller)
        return _mutate(
            lambda cur, audit, now: admit_authorization(
                cur, payload, principal=principal, now=now, audit=audit),
            success_status=201,
            actor_id=principal.agent_id,
            entity_type="authorization_admission",
            entity_id=str(payload.get("authorization_id") or "Q?"),
            mutation_kind="admit",
            action_basis="authorization_admitted",
        )

    @router.get("/{authorization_id}")
    def read(authorization_id: str,
             caller: Any = Depends(principal_dependency)) -> JSONResponse:
        """Status only. Spends nothing, writes no ledger row, grants nothing."""
        if not _may_read(caller):
            return JSONResponse(
                status_code=403,
                content={"refusal_code": "missing_scope",
                         "detail": f"Missing scope: {READ_SCOPE}"},
            )
        with db_factory() as conn:
            with conn.cursor() as cur:
                record = read_authorization(cur, authorization_id)
        if record is None:
            return JSONResponse(
                status_code=404,
                content={"refusal_code": "unknown_authorization",
                         "detail": f"no admitted authorization "
                                   f"'{authorization_id}'"},
            )
        body = {name: record[name] for name in READABLE_FIELDS
                if name in record}
        body["status"] = ("admitted" if record.get("consumed_at") is None
                          else "consumed")
        return JSONResponse(status_code=200, content=body)

    @router.post("/{authorization_id}/consume")
    def consume(authorization_id: str,
                body: ConsumeIn,
                caller: Any = Depends(principal_dependency)) -> JSONResponse:
        """Spend one authorization for one stated action, exactly once.

        Single use is decided by the store's guarded UPDATE, not by this
        endpoint reading first: two concurrent callers both reach it and the
        database picks the winner.
        """
        principal = _store_principal(caller)
        expectation = Expectation(
            agent=body.expectation.agent,
            lane=body.expectation.lane,
            action=body.expectation.action,
            branch=body.expectation.branch,
            base_sha=body.expectation.base_sha,
            intended_paths=tuple(body.expectation.intended_paths),
        )
        return _mutate(
            lambda cur, audit, now: consume_authorization(
                cur, authorization_id,
                principal=principal,
                expectation=expectation,
                now=now,
                expected_envelope_sha256=body.expected_envelope_sha256,
                expected_task_id=body.expected_task_id,
                expected_spec_sha256=body.expected_spec_sha256,
                audit=audit),
            success_status=200,
            project=_consumed_body,
            actor_id=principal.agent_id,
            entity_type="authorization_admission",
            entity_id=authorization_id,
            mutation_kind="consume",
            action_basis="authorization_consumed",
        )

    app.include_router(router)
    return router