Moved exception handlers to sub apps.

This commit is contained in:
Leo Vasanko 2025-09-02 14:57:06 -06:00
parent 8c07945661
commit 9feac6e9a8
3 changed files with 32 additions and 38 deletions

View File

@ -10,6 +10,7 @@ the /auth/admin path prefix. The routes defined here therefore omit the
from __future__ import annotations
import contextlib
import logging
from pathlib import Path
from uuid import UUID, uuid4
@ -32,6 +33,12 @@ async def value_error_handler(_request, exc: ValueError): # pragma: no cover -
return JSONResponse(status_code=400, content={"detail": str(exc)})
@app.exception_handler(Exception)
async def general_exception_handler(_request, exc: Exception):
logging.exception("Unhandled exception in admin app")
return JSONResponse(status_code=500, content={"detail": "Internal server error"})
async def _get_ctx_and_admin_flags(auth_cookie: str):
"""Helper to get session context and admin flags from cookie."""
if not auth_cookie:
@ -47,13 +54,8 @@ async def _get_ctx_and_admin_flags(auth_cookie: str):
@app.get("/")
@app.get("")
async def serve_admin_root(auth=Cookie(None)):
"""Serve the admin SPA root if an authenticated session exists.
Mirrors previous behavior from mainapp. If no valid session, serve the
main index.html with 401 so frontend can trigger login flow.
"""
async def admin_frontend(auth=Cookie(None)):
"""Serve the admin SPA root if an authorized session exists."""
if auth:
with contextlib.suppress(ValueError):
s = await get_session(auth)

View File

@ -1,3 +1,4 @@
import logging
from contextlib import suppress
from uuid import UUID
@ -11,6 +12,7 @@ from fastapi import (
Request,
Response,
)
from fastapi.responses import JSONResponse
from fastapi.security import HTTPBearer
from .. import aaguid
@ -26,6 +28,19 @@ bearer_auth = HTTPBearer(auto_error=True)
app = FastAPI()
@app.exception_handler(ValueError)
async def value_error_handler(_request: Request, exc: ValueError): # pragma: no cover
return JSONResponse(status_code=400, content={"detail": str(exc)})
@app.exception_handler(Exception)
async def general_exception_handler(
_request: Request, exc: Exception
): # pragma: no cover
logging.exception("Unhandled exception in API app")
return JSONResponse(status_code=500, content={"detail": "Internal server error"})
@app.post("/validate")
async def validate_token(perm=Query(None), auth=Cookie(None)):
s = await authz.verify(auth, perm)

View File

@ -4,12 +4,11 @@ from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import Cookie, FastAPI, HTTPException, Query, Request, Response
from fastapi.responses import FileResponse, JSONResponse, RedirectResponse
from fastapi.responses import FileResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from passkey.util import passphrase
from ..globals import passkey as global_passkey
from . import admin, api, authz, ws
STATIC_DIR = Path(__file__).parent.parent / "frontend-build"
@ -50,39 +49,20 @@ async def lifespan(app: FastAPI): # pragma: no cover - startup path
app = FastAPI(lifespan=lifespan)
app.mount("/auth/ws", ws.app)
app.mount("/auth/admin", admin.app)
app.mount("/auth/api", api.app)
# Global exception handlers
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
"""Handle ValueError exceptions globally with 400 status code."""
return JSONResponse(status_code=400, content={"detail": str(exc)})
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
"""Handle all other exceptions globally with 500 status code."""
logging.exception("Internal Server Error")
return JSONResponse(status_code=500, content={"detail": "Internal server error"})
# Serve static files
app.mount(
"/auth/assets", StaticFiles(directory=STATIC_DIR / "assets"), name="static assets"
)
app.mount("/auth/ws", ws.app)
app.mount("/auth/assets", StaticFiles(directory=STATIC_DIR / "assets"), name="assets")
@app.get("/auth/")
async def redirect_to_index():
async def frontend():
"""Serve the main authentication app."""
return FileResponse(STATIC_DIR / "index.html")
@app.get("/auth/{reset_token}")
async def reset_authentication(request: Request, reset_token: str):
@app.get("/auth/{reset}")
async def reset_authentication(request: Request, reset: str):
"""Validate reset token and redirect with it as query parameter (no cookies).
After validation we 303 redirect to /auth/?reset=<token>. The frontend will:
@ -90,12 +70,9 @@ async def reset_authentication(request: Request, reset_token: str):
- Use it via Authorization header or websocket query param
- history.replaceState to remove it from the address bar/history
"""
if not passphrase.is_well_formed(reset_token):
if not passphrase.is_well_formed(reset):
raise HTTPException(status_code=404)
origin = global_passkey.instance.origin
# Do not verify existence/expiry here; frontend + user-info endpoint will handle invalid tokens.
url = f"{origin}/auth/?reset={reset_token}"
return RedirectResponse(url=url, status_code=303)
return RedirectResponse(request.url_for("frontend", reset=reset), status_code=303)
@app.get("/auth/forward-auth")