Files
BanGUI/backend/app/routers/auth.py
Lukas 7ec80fdeec refactor(logging): replace structlog with stdlib logging compat layer
- Remove structlog dependency from backend/pyproject.toml
- Add app.utils.logging_compat shim for keyword-arg logging API
- Add app.utils.json_formatter for JSON log output with extra fields
- Update all backend modules to use logging_compat.get_logger()
- Update docstrings in log_sanitizer.py and json_formatter.py
- Update test comment in test_async_utils.py
- Record 406 failing tests in Docs/Tasks.md for tracking
2026-05-10 13:37:54 +02:00

209 lines
7.2 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 set as an ``HttpOnly`` ``SameSite=Lax`` cookie for
browser-based SPAs. The cookie is automatically included in all requests
and is inaccessible to JavaScript, protecting it from XSS attacks and
malicious scripts.
For programmatic API clients (non-browser), use ``POST /api/auth/token``
which returns a token in the response body for use in the ``Authorization``
header. This endpoint does not set a cookie.
"""
from __future__ import annotations
from app.utils.logging_compat import get_logger
from fastapi import APIRouter, Request, Response
from app.dependencies import (
AuthDep,
SessionCacheDep,
SessionServiceContextDep,
SettingsDep,
)
from app.exceptions import AuthenticationError
from app.models.auth import LoginRequest, LoginResponse, LogoutResponse, SessionValidResponse
from app.services import auth_service
from app.utils.client_ip import get_client_ip
from app.utils.constants import SESSION_COOKIE_NAME
log = get_logger(__name__)
router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
@router.post(
"/login",
response_model=LoginResponse,
summary="Authenticate with the master password",
responses={
200: {"description": "Login successful", "model": LoginResponse},
401: {"description": "Invalid password"},
422: {"description": "Validation error — invalid request body"},
503: {"description": "Setup not complete"},
},
)
async def login(
body: LoginRequest,
response: Response,
request: Request,
session_ctx: SessionServiceContextDep,
settings: SettingsDep,
session_cache: SessionCacheDep,
) -> 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.
Cache invalidation: On successful login, any existing cached sessions for
the same user are invalidated so that stale tokens (e.g., from a stolen
device) cannot be reused beyond the cache TTL window.
Args:
body: Login request validated by Pydantic.
response: FastAPI response object used to set the cookie.
request: The incoming HTTP request (used to extract client IP).
session_ctx: Session service context containing db and repository.
settings: Application settings (used for session duration and trusted proxies).
session_cache: Session cache for invalidating old sessions on login.
Returns:
:class:`~app.models.auth.LoginResponse` containing the token.
Raises:
AuthenticationError: if the password is incorrect.
"""
client_ip = get_client_ip(request, trusted_proxies=settings.trusted_proxies)
try:
signed_token, expires_at, session = await auth_service.login(
session_ctx.db,
password=body.password,
session_duration_minutes=settings.session_duration_minutes,
session_secret=settings.session_secret,
session_repo=session_ctx.session_repo,
)
except ValueError as exc:
log.warning("login_failed", client_ip=client_ip, error=str(exc))
raise AuthenticationError(str(exc)) from exc
# Invalidate any cached sessions for the same user to prevent reuse of
# stale tokens (e.g., from a stolen device) beyond the cache TTL window.
session_cache.invalidate_by_user(session.id)
response.set_cookie(
key=SESSION_COOKIE_NAME,
value=signed_token,
httponly=settings.session_cookie_httponly,
samesite=settings.session_cookie_samesite,
secure=settings.session_cookie_secure,
max_age=settings.session_duration_minutes * 60,
)
log.info("login_success", client_ip=client_ip)
return LoginResponse(expires_at=expires_at)
@router.get(
"/session",
response_model=SessionValidResponse,
summary="Validate the current session",
responses={
200: {"description": "Session valid", "model": SessionValidResponse},
401: {"description": "Session missing, expired, or invalid"},
},
)
async def validate_session(
_: AuthDep,
) -> SessionValidResponse:
"""Validate the current session.
This endpoint requires a valid session and returns 200 if the session is
valid and still active. If the session is invalid, expired, or missing,
FastAPI's ``require_auth`` dependency returns 401 automatically.
The frontend calls this on mount to bootstrap its authentication state
from the backend rather than relying solely on cached ``sessionStorage``.
Args:
_: The injected session object (unused, but its presence triggers validation).
Returns:
:class:`~app.models.auth.SessionValidResponse` confirming the session state.
"""
return SessionValidResponse(valid=True)
@router.post(
"/logout",
response_model=LogoutResponse,
summary="Revoke the current session",
responses={
200: {"description": "Logout successful", "model": LogoutResponse},
401: {"description": "Session missing or invalid (silently successful)"},
},
)
async def logout(
request: Request,
response: Response,
session_ctx: SessionServiceContextDep,
settings: SettingsDep,
session_cache: SessionCacheDep,
) -> 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).
session_ctx: Session service context containing db and repository.
settings: Application settings (used to unwrap signed tokens).
session_cache: Session cache for invalidation.
Returns:
:class:`~app.models.auth.LogoutResponse`.
"""
token = _extract_token(request)
if token:
raw_token = await auth_service.logout(
session_ctx.db,
token,
settings.session_secret,
settings.session_secret_previous,
session_repo=session_ctx.session_repo,
)
if raw_token:
session_cache.invalidate(raw_token)
session_cache.invalidate(token)
response.delete_cookie(key=SESSION_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(SESSION_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