"""Per-request auth context — propagates "via=jwt" / "via=pat" to audit logs without every domain service having to thread the request object. ``get_current_user`` sets ``current_auth_method`` and (for PATs) ``current_pat_token_id`` at the start of every authenticated request. ``AuditService.log`` reads them and merges them into the ``detail`` JSON column. The contextvars reset automatically when the request task ends (FastAPI runs each request in its own asyncio task with its own contextvar copy), so there is no leakage across requests. """ from __future__ import annotations from contextvars import ContextVar from uuid import UUID current_auth_method: ContextVar[str | None] = ContextVar( "helix_current_auth_method", default=None, ) current_pat_token_id: ContextVar[UUID | None] = ContextVar( "helix_current_pat_token_id", default=None, ) # Pre-launch P1 fix: per-request actor + trace contextvars so the # leaver-blocked SQLAlchemy ``before_flush`` listener can stamp # ``attempted_by_user_id`` and a coherent ``trace_id`` on its trap # audit rows even for unguarded writes (the canonical guard already # sets its own contextvars; this pair fills in the unguarded case # where attribution matters most). Set by ``get_current_user`` and # the trace middleware respectively. current_actor_user_id: ContextVar[UUID | None] = ContextVar( "helix_current_actor_user_id", default=None, ) current_request_trace_id: ContextVar[str | None] = ContextVar( "helix_current_request_trace_id", default=None, ) def auth_metadata() -> dict[str, str]: """Return a dict suitable for merging into ``audit_logs.detail``. Empty when no auth context is set (e.g. background jobs running without a request). """ method = current_auth_method.get() if method is None: return {} out: dict[str, str] = {"via": method} token_id = current_pat_token_id.get() if token_id is not None: out["pat_token_id"] = str(token_id) return out __all__ = [ "auth_metadata", "current_actor_user_id", "current_auth_method", "current_pat_token_id", "current_request_trace_id", ]