Spaces:
Sleeping
Sleeping
| """Provide FastAPI dependencies for protected routes. | |
| Session validity is checked against expiry and durable session version. | |
| """ | |
| import time | |
| from typing import Any | |
| from fastapi import HTTPException, Request, status | |
| from app.models import StoredConfig | |
| def api_http_error(status_code: int, code: str, message: str) -> HTTPException: | |
| """Create an HTTP exception understood by the envelope handler.""" | |
| return HTTPException( | |
| status_code=status_code, | |
| detail={"code": code, "message": message}, | |
| ) | |
| def require_login(request: Request) -> StoredConfig: | |
| """Require a current signed session and return durable config.""" | |
| config: StoredConfig | None = request.app.state.config_store.load() | |
| if config is None: | |
| raise api_http_error( | |
| status.HTTP_503_SERVICE_UNAVAILABLE, | |
| "setup_required", | |
| "Application setup is required", | |
| ) | |
| session: dict[str, Any] = request.session | |
| authenticated = session.get("auth") is True | |
| version_matches = session.get("sv") == config.session_version | |
| expiry = session.get("exp") | |
| unexpired = isinstance(expiry, (int, float)) and expiry >= time.time() | |
| if not (authenticated and version_matches and unexpired): | |
| request.session.clear() | |
| raise api_http_error( | |
| status.HTTP_401_UNAUTHORIZED, | |
| "unauthorized", | |
| "Login required", | |
| ) | |
| return config | |