"""Shared auth helpers used by all route files.""" import os from typing import Optional from fastapi import Request, HTTPException from src.owner_identity import auth_disabled, effective_storage_owner def get_current_user(request: Request) -> Optional[str]: """Get current username from request state (set by auth middleware).""" return getattr(request.state, 'current_user', None) def effective_user(request: Request) -> Optional[str]: """The real human behind the request, for ownership/attribution. Cookie sessions resolve to the logged-in username. Bearer ``ody_`` callers come through as the sandboxed pseudo-user "api" so they can't wander into cookie/user routes by default, but their token was minted by, and belongs to, a real owner stamped on ``request.state.api_token_owner``. Routes that should attribute a token's actions to that owner (sessions, chat history) call this instead of :func:`get_current_user`, so a paired client sees and creates the SAME data as the owner's desktop UI rather than a separate "api"-owned silo. For cookie sessions this is identical to :func:`get_current_user`, so swapping a route over is a no-op for browser users. A bearer token with no owner falls back to :func:`get_current_user` (the "api" pseudo-user), so it never escalates. """ if getattr(request.state, "api_token", False): owner = getattr(request.state, "api_token_owner", None) if owner: return owner return get_current_user(request) def _is_api_token_request(request: Request) -> bool: """Return True when middleware authenticated a bearer API token.""" return bool(getattr(request.state, "api_token", False)) def is_delegated_credential(request: Request) -> bool: """Whether this request arrived on a credential acting FOR a human. A bearer API token is minted by a person and then handed to something else: an integration, a script, a third party. :func:`effective_user` resolves it back to that person for ownership and attribution, which is correct for data but wrong for authority. Only admins can mint tokens, so every token resolves to an admin, and any gate that asks "is the owner an admin?" answers yes for a credential the owner has given away. Security decisions about what the AGENT may do should ask this instead, so a token cannot inherit the shell merely because its owner could use one. """ return _is_api_token_request(request) def require_api_token_scope(request: Request, scope: str) -> Optional[str]: """Require ``scope`` when the request is authenticated by an API token. Browser sessions are unaffected. Scoped bearer routes use this before touching owner data so resolving the token back to its owner never also grants the owner's interactive-session authority. """ if not _is_api_token_request(request): return get_current_user(request) scopes = set(getattr(request.state, "api_token_scopes", []) or []) if scope not in scopes: raise HTTPException(403, f"API token missing required scope: {scope}") owner = getattr(request.state, "api_token_owner", None) if not owner: raise HTTPException(403, "API token has no owner") return owner def require_chat_api_token_scope(request: Request) -> Optional[str]: """FastAPI dependency for chat/session/history bearer surfaces.""" return require_api_token_scope(request, "chat") def require_authenticated_request(request: Request) -> str: """Allow either a browser session or a valid bearer API token. This is intentionally narrower than :func:`require_user`: use it only for routes that need authentication but do not read or mutate owner-scoped user data. Owner-scoped routes should use ``require_user`` for browser sessions or their own API-token scope/owner gate. """ if _is_api_token_request(request): return effective_user(request) or "" return require_user(request) def _auth_disabled() -> bool: """True when the operator has explicitly turned off auth via .env. Mirrors the AUTH_ENABLED parse in app.py / core/middleware.py so the three call sites agree on what "off" means.""" return auth_disabled() def storage_owner_for_request(request: Request) -> Optional[str]: """Resolve the storage owner for code paths that need an owner bucket. This does not replace route authentication. It only gives auth-disabled no-login mode a stable storage identity instead of writing new data as legacy NULL/ownerless state. """ return effective_storage_owner(effective_user(request)) def require_user(request: Request) -> str: """FastAPI dependency: reject unauthenticated callers when the upstream auth middleware was bypassed unexpectedly (e.g. SSRF from a sibling service). Returns the resolved username, or "" in single-user / anonymous modes where no username is available. The three "" cases are: 1. AUTH_ENABLED=false — the operator explicitly turned auth off. The full /login flow is skipped (issue #622), so route-level require_user must let the request through too instead of 401-ing and forcing the browser to /login. 2. Unconfigured first-run + loopback caller — pre-setup access from localhost so the operator can hit the SPA before creating the first admin. 3. LOCALHOST_BYPASS=true + loopback caller — documented dev bypass. Use this on routes that touch user data so middleware misconfig can't open them up. """ if _is_api_token_request(request): raise HTTPException(403, "API tokens must use a scope-aware API route") u = get_current_user(request) if u: return u # Operator-disabled auth: honor it at the route layer too. Without this, # routes that depend on require_user 401, the front-end fetch wrapper # redirects to /login, and the user sees a login page despite # AUTH_ENABLED=false (issue #622). Docker / reverse-proxy deployments # hit this because requests arrive from a non-loopback client.host, so # the loopback fall-through below never fires. if _auth_disabled(): return "" auth_mgr = getattr(request.app.state, "auth_manager", None) client = getattr(request, "client", None) host = (client.host if client else "") or "" is_loopback = host in ("127.0.0.1", "::1", "localhost") # LOCALHOST_BYPASS=true is the dev-only "I'm on loopback, skip auth" # switch. Mirror the middleware so routes don't 401 the same caller # the middleware just let through. if is_loopback and os.getenv("LOCALHOST_BYPASS", "false").lower() == "true": return "" if auth_mgr is not None and getattr(auth_mgr, "is_configured", False): raise HTTPException(401, "Not authenticated") # Unconfigured / first-run mode: only allow loopback callers. if is_loopback: return "" raise HTTPException(401, "Not authenticated") def require_privilege(request: Request, key: str) -> str: """Reject callers whose `auth.json` privilege flag for `key` is False. Returns the username so the route handler can keep using it. Admins always have every privilege via `auth_manager.get_privileges` (which returns ADMIN_PRIVILEGES wholesale), so this is a no-op for them. In unauthenticated single-user mode (`require_user` returns ""), privileges aren't enforced. """ user = require_user(request) if not user: return user auth_mgr = getattr(request.app.state, "auth_manager", None) if auth_mgr is None: return user try: privs = auth_mgr.get_privileges(user) or {} except Exception: return user if not isinstance(privs, dict): privs = {} # True = permitted; missing key defaults to permitted (unknown privileges # fail open — the UI gates display-side). if not privs.get(key, True): raise HTTPException(403, f"Your account is not allowed to {key.replace('_', ' ')}.") return user def owner_filter(query, model_cls, user: str, *, include_shared: bool = True): """Filter `query` so only rows owned by `user` (and optionally null-owner 'shared' rows) come through. No-op when `user` is empty (single-user mode). Returns the modified query.""" if not user: return query if include_shared: return query.filter((model_cls.owner == user) | (model_cls.owner == None)) # noqa: E711 return query.filter(model_cls.owner == user)