Files
BanGUI/backend/app/routers/auth.py
Lukas d931e8c6a3 Reduce per-request DB overhead (Task 4)
- Cache setup_completed flag in app.state._setup_complete_cached after
  first successful is_setup_complete() call; all subsequent API requests
  skip the DB query entirely (one-way transition, cleared on restart).
- Add in-memory session token TTL cache (10 s) in require_auth; the second
  request with the same token within the window skips session_repo.get_session.
- Call invalidate_session_cache() on logout so revoked tokens are evicted
  immediately rather than waiting for TTL expiry.
- Add clear_session_cache() for test isolation.
- 5 new tests covering the cached fast-path for both optimisations.
- 460 tests pass, 83% coverage, zero ruff/mypy warnings.
2026-03-10 19:16:00 +01:00

130 lines
3.8 KiB
Python

"""Authentication router.
``POST /api/auth/login`` — verify master password and issue a session.
``POST /api/auth/logout`` — revoke the current session.
The session token is returned both in the JSON body (for API-first
consumers) and as an ``HttpOnly`` cookie (for the browser SPA).
"""
from __future__ import annotations
import structlog
from fastapi import APIRouter, HTTPException, Request, Response, status
from app.dependencies import DbDep, SettingsDep, invalidate_session_cache
from app.models.auth import LoginRequest, LoginResponse, LogoutResponse
from app.services import auth_service
log: structlog.stdlib.BoundLogger = structlog.get_logger()
router = APIRouter(prefix="/api/auth", tags=["auth"])
_COOKIE_NAME = "bangui_session"
@router.post(
"/login",
response_model=LoginResponse,
summary="Authenticate with the master password",
)
async def login(
body: LoginRequest,
response: Response,
db: DbDep,
settings: SettingsDep,
) -> LoginResponse:
"""Verify the master password and return a session token.
On success the token is also set as an ``HttpOnly`` ``SameSite=Lax``
cookie so the browser SPA benefits from automatic credential handling.
Args:
body: Login request validated by Pydantic.
response: FastAPI response object used to set the cookie.
db: Injected aiosqlite connection.
settings: Application settings (used for session duration).
Returns:
:class:`~app.models.auth.LoginResponse` containing the token.
Raises:
HTTPException: 401 if the password is incorrect.
"""
try:
session = await auth_service.login(
db,
password=body.password,
session_duration_minutes=settings.session_duration_minutes,
)
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=str(exc),
) from exc
response.set_cookie(
key=_COOKIE_NAME,
value=session.token,
httponly=True,
samesite="lax",
secure=False, # Set to True in production behind HTTPS
max_age=settings.session_duration_minutes * 60,
)
return LoginResponse(token=session.token, expires_at=session.expires_at)
@router.post(
"/logout",
response_model=LogoutResponse,
summary="Revoke the current session",
)
async def logout(
request: Request,
response: Response,
db: DbDep,
) -> LogoutResponse:
"""Invalidate the active session.
The session token is read from the ``bangui_session`` cookie or the
``Authorization: Bearer`` header. If no token is present the request
is silently treated as a successful logout (idempotent).
Args:
request: FastAPI request (used to extract the token).
response: FastAPI response (used to clear the cookie).
db: Injected aiosqlite connection.
Returns:
:class:`~app.models.auth.LogoutResponse`.
"""
token = _extract_token(request)
if token:
await auth_service.logout(db, token)
invalidate_session_cache(token)
response.delete_cookie(key=_COOKIE_NAME)
return LogoutResponse()
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _extract_token(request: Request) -> str | None:
"""Extract the session token from cookie or Authorization header.
Args:
request: The incoming FastAPI request.
Returns:
The token string, or ``None`` if absent.
"""
token: str | None = request.cookies.get(_COOKIE_NAME)
if token:
return token
auth_header: str = request.headers.get("Authorization", "")
if auth_header.startswith("Bearer "):
return auth_header[len("Bearer "):]
return None