"""Admin authentication: cookie/JWT session + role gate. The panel authenticates operators by email+password (the existing web-auth flow) and stores the resulting access JWT in an HttpOnly cookie so the browser can drive HTMX navigation. The same JWT is also accepted via the ``Authorization: Bearer`` header, so the panel can be scripted if needed. Access requires ``users.role = 'admin'`` and ``is_active = true``. When the guard cannot establish an admin session it raises :class:`AdminAuthError`, which app.py maps to a redirect to ``/admin/login`` (bypassing the generic 500 handler, which would otherwise swallow it). """ from __future__ import annotations from dataclasses import dataclass from uuid import UUID from fastapi import Depends, HTTPException, Request, status from sqlalchemy.ext.asyncio import AsyncSession from src.contract_check.api.deps import AsyncSessionDep from src.contract_check.core.auth import AuthError, verify_access_token from src.contract_check.core.config import get_settings from src.contract_check.core.db.repositories import UserRepository # HttpOnly cookie carrying the admin access JWT. ADMIN_COOKIE = "cc_admin_token" # One day — matches the default JWT_ACCESS_TTL_MINUTES (1440). ADMIN_COOKIE_MAX_AGE = 24 * 3600 class AdminAuthError(Exception): """Raised to short-circuit a request into a redirect to ``/admin/login``.""" def __init__(self, reason: str = "auth") -> None: self.reason = reason super().__init__(reason) @dataclass(slots=True) class AdminUser: """The authenticated operator driving the admin panel.""" user_id: UUID telegram_id: int email: str | None role: str def _token_from_request(request: Request) -> str | None: cookie = request.cookies.get(ADMIN_COOKIE) if cookie: return cookie header = request.headers.get("authorization") if header and header.lower().startswith("bearer "): return header[7:].strip() return None async def resolve_admin(request: Request, session: AsyncSession) -> AdminUser | None: """Return the admin identity if the request carries a valid admin session. Returns ``None`` for missing/expired tokens, unknown users, inactive accounts, or non-admin roles — the caller decides how to react. """ if not get_settings().web_admin_enabled: return None token = _token_from_request(request) if not token: return None try: claims = verify_access_token(token) except AuthError: return None user = await UserRepository(session).get_by_id(claims.sub) if user is None: return None if not user.is_active or user.role != get_settings().admin_required_role: return None return AdminUser( user_id=user.id, telegram_id=int(user.telegram_id or 0), email=user.email, role=user.role, ) async def require_admin(request: Request, session: AsyncSessionDep) -> AdminUser: """Dependency: ensure an admin session exists, else redirect to login.""" admin = await resolve_admin(request, session) if admin is None: raise AdminAuthError() return admin def require_htmx(request: Request) -> None: """Dependency: only accept mutating requests coming from HTMX. HTMX sends ``HX-Request: true`` on every request; a cross-site HTML form cannot set a custom header without triggering a CORS preflight, so this is an effective CSRF defence for cookie-authed, htmx-driven forms. """ if request.headers.get("hx-request") != "true": raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="htmx request required") HtmxGuard = Depends(require_htmx)