"""FastAPI dependencies — auth gates for routers. Two dependencies live here (Stage 2): - :func:`current_user` — extracts the session cookie, resolves it to a :class:`User` via :func:`src.lib.auth.repo.from_session_token`, and raises :class:`Unauthenticated` (401, code ``UNAUTHENTICATED``) on miss. The token lookup bumps ``last_seen`` as a side effect, so an active user's session row stays warm (see :mod:`src.api.auth.sessions`). - :func:`require_admin` — depends on :func:`current_user` and raises :class:`Forbidden` (403, code ``FORBIDDEN``) for non-admin roles. Other deps (db handle, storage backend, qdrant client) will land in later stages as the read-only routers need them. Keeping this module thin per BACKEND_BUILD.md §6.3 — most state still comes from process-globals via ``src.config.load_config``. """ from __future__ import annotations from fastapi import Cookie, Depends, Path from src.api.errors import BookNotFound, Forbidden, Unauthenticated from src.config import load_config from src.lib.auth.models import User from src.lib.auth.repo import from_session_token def current_user(session: str | None = Cookie(default=None)) -> User: """Return the logged-in :class:`User` or raise 401. The cookie name is ``session`` (CONTRACT.md §3). FastAPI's ``Cookie`` dep handles the parsing; we treat both "no cookie sent" and "cookie present but unknown/expired" as the same 401 — leaking "this token used to exist" is not useful and risks a session-probing oracle. """ if not session: raise Unauthenticated("Sign in to continue.") user = from_session_token(session) if user is None: raise Unauthenticated("Your session has expired. Please sign in again.") return user def require_admin(user: User = Depends(current_user)) -> User: """Gate an endpoint to admin role only. Used as a FastAPI dependency: ``def handler(_: User = Depends(require_admin))``. The dependency chains through :func:`current_user`, so a missing cookie surfaces as 401 (``UNAUTHENTICATED``) before this function ever sees a viewer attempting an admin action. """ if not user.is_admin: raise Forbidden("This action requires an admin role.") return user def require_query_access(user: User = Depends(current_user)) -> User: """Gate ``POST /query`` (and other paid query paths) by capability. The documented role intent (``src/lib/auth/repo.py``) is that the low-privilege ``viewer`` role is read-only and **cannot run queries** — queries fan out to paid LLM providers. That intent was historically enforced only in the Streamlit UI; this dependency enforces it server-side so the FastAPI backend matches the contract. Admins always pass. The ``reader`` role is NEVER permitted (readers are Reading-Room-only and the ``viewer_can_query`` config toggle must never open a paid path to them). Whether ``viewer`` may query is config-gated via ``api.viewer_can_query`` (default ``False``) so the operator can open it up without a code change. """ if user.is_admin: return user # Readers are hard-denied regardless of any config toggle — a paid LLM # fan-out must never be reachable by a Reading-Room-only account. if user.role == "reader": raise Forbidden("Running queries requires an account with query access.") try: allowed = bool(load_config().section("api").get("viewer_can_query", False)) except Exception: # noqa: BLE001 — fail closed if config is unreadable allowed = False if not allowed: raise Forbidden("Running queries requires an account with query access.") return user def forbid_readers(user: User = Depends(current_user)) -> User: """Gate library/research endpoints OFF for the ``reader`` role. Readers are Reading-Room-only: they may read PUBLIC book content (see :func:`require_book_read_access`) but must never touch the library / research surface — full book metadata, query history, eval runs, tools, indexes, costs, system topology, etc. Admins and viewers pass through unchanged (this preserves the existing two-role behavior exactly). Raises 403 ``FORBIDDEN`` for readers — unlike book content (which 404s to avoid a private-id oracle), these routes leak no per-book secret by admitting they exist, so a plain "not allowed" is correct and clearer. """ if user.role == "reader": raise Forbidden("This area isn't available for your account.") return user def require_book_read_access( book_id: str = Path(..., alias="bookId"), user: User = Depends(current_user), ) -> User: """Gate per-book reader content (PDF / page image / OCR / clean / audio). Access rules: - ``admin`` and ``viewer``: full access — preserves current behavior exactly (they never hit the visibility check). - ``reader``: allowed when the book's ``is_public == 1`` OR when an explicit per-account grant exists for ``(reader, book_id)`` (D2). For a private AND ungranted book — OR a book that doesn't exist — the reader gets a 404 ``BOOK_NOT_FOUND`` (NOT a 403): a 403 would confirm the private book id exists, handing a reader an enumeration oracle over the private library. Grants only ADD access; they never restrict a public book. The ``bookId`` path param is read straight off the route (every reader content endpoint declares ``{bookId}``), so this dep is drop-in: ``user: User = Depends(require_book_read_access)``. """ if user.role != "reader": return user # Reader path: load just the visibility flag. None => book doesn't exist. # Imported locally to keep the books repo (and its inventory-DB import) # out of this thin module's import-time cost. from src.lib.books.repo import get_book_is_public if get_book_is_public(book_id) is True: return user # Not public — maybe it was explicitly granted to this reader. The grant # check is also imported locally for the same import-cost reason. A grant # only ADDS access; it never restricts a public book (already handled # above). Private AND ungranted → identical 404, no existence oracle. from src.lib.auth.book_access_repo import is_granted if is_granted(user.username, book_id): return user raise BookNotFound(book_id=book_id)