from __future__ import annotations import os import secrets from urllib.parse import urlparse from fastapi import Depends, HTTPException, Request, status from fastapi.security import HTTPBasic, HTTPBasicCredentials, HTTPBearer, HTTPAuthorizationCredentials from app.config import Settings, get_settings basic = HTTPBasic(auto_error=False) bearer = HTTPBearer(auto_error=False) def require_dashboard_auth( credentials: HTTPBasicCredentials | None = Depends(basic), settings: Settings = Depends(get_settings), ) -> None: if not settings.security.dashboard_auth_enabled: return username = os.getenv(settings.security.dashboard_username_env) password = os.getenv(settings.security.dashboard_password_env) if not username or not password: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Dashboard authentication is enabled but credentials are not configured.", ) valid = credentials and secrets.compare_digest(credentials.username, username) and secrets.compare_digest(credentials.password, password) if not valid: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required", headers={"WWW-Authenticate": "Basic"}, ) def require_homepage_token( credentials: HTTPAuthorizationCredentials | None = Depends(bearer), settings: Settings = Depends(get_settings), ) -> None: if not settings.security.api_token_required: return expected = os.getenv(settings.security.homepage_token_env, "") if not credentials or not expected or not secrets.compare_digest(credentials.credentials, expected): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid bearer token") def _same_origin(candidate: str, allowed_hosts: set[str]) -> bool: parsed = urlparse(candidate) return bool(parsed.scheme in {"http", "https"} and parsed.netloc in allowed_hosts) def require_admin_csrf(request: Request, settings: Settings = Depends(get_settings)) -> None: if not settings.security.dashboard_auth_enabled or request.method in {"GET", "HEAD", "OPTIONS", "TRACE"}: return allowed_hosts = {host for host in {request.headers.get("host"), urlparse(settings.app.base_url).netloc} if host} origin = request.headers.get("origin") if origin: if _same_origin(origin, allowed_hosts): return raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cross-site admin POST rejected.") referer = request.headers.get("referer") if referer: if _same_origin(referer, allowed_hosts): return raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cross-site admin POST rejected.") if request.headers.get("x-requested-with") == "XMLHttpRequest": return raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin POST requires same-origin headers.")