Centralise all cookie handling to session.py.

passkey/fastapi/admin.py

@@ -2,7 +2,7 @@ import logging
 from datetime import timezone
 from uuid import UUID, uuid4
 
-from fastapi import Body, Cookie, FastAPI, HTTPException, Request
+from fastapi import Body, FastAPI, HTTPException, Request
 from fastapi.responses import FileResponse, JSONResponse
 
 from ..authsession import reset_expires
@@ -18,6 +18,7 @@ from ..util import (
 )
 from ..util.tokens import encode_session_key, session_key
 from . import authz
+from .session import AUTH_COOKIE
 
 app = FastAPI()
 
@@ -34,7 +35,7 @@ async def general_exception_handler(_request, exc: Exception):
 
 
 @app.get("/")
-async def adminapp(request: Request, auth=Cookie(None, alias="__Host-auth")):
+async def adminapp(request: Request, auth=AUTH_COOKIE):
     """Serve admin SPA only for authenticated users with admin/org permissions.
 
     On missing/invalid session or insufficient permissions, serve restricted SPA.
@@ -57,7 +58,7 @@ async def adminapp(request: Request, auth=Cookie(None, alias="__Host-auth")):
 
 
 @app.get("/orgs")
-async def admin_list_orgs(request: Request, auth=Cookie(None, alias="__Host-auth")):
+async def admin_list_orgs(request: Request, auth=AUTH_COOKIE):
     ctx = await authz.verify(
         auth,
         ["auth:admin", "auth:org:*"],
@@ -100,7 +101,7 @@ async def admin_list_orgs(request: Request, auth=Cookie(None, alias="__Host-auth
 
 @app.post("/orgs")
 async def admin_create_org(
-    request: Request, payload: dict = Body(...), auth=Cookie(None, alias="__Host-auth")
+    request: Request, payload: dict = Body(...), auth=AUTH_COOKIE
 ):
     await authz.verify(
         auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all
@@ -132,7 +133,7 @@ async def admin_update_org(
     org_uuid: UUID,
     request: Request,
     payload: dict = Body(...),
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     ctx = await authz.verify(
         auth,
@@ -165,9 +166,7 @@ async def admin_update_org(
 
 
 @app.delete("/orgs/{org_uuid}")
-async def admin_delete_org(
+async def admin_delete_org(org_uuid: UUID, request: Request, auth=AUTH_COOKIE):
-    org_uuid: UUID, request: Request, auth=Cookie(None, alias="__Host-auth")
-):
     ctx = await authz.verify(
         auth,
         ["auth:admin", f"auth:org:{org_uuid}"],
@@ -200,7 +199,7 @@ async def admin_add_org_permission(
     org_uuid: UUID,
     permission_id: str,
     request: Request,
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     await authz.verify(
         auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all
@@ -214,7 +213,7 @@ async def admin_remove_org_permission(
     org_uuid: UUID,
     permission_id: str,
     request: Request,
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     await authz.verify(
         auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all
@@ -231,7 +230,7 @@ async def admin_create_role(
     org_uuid: UUID,
     request: Request,
     payload: dict = Body(...),
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     await authz.verify(
         auth,
@@ -266,7 +265,7 @@ async def admin_update_role(
     role_uuid: UUID,
     request: Request,
     payload: dict = Body(...),
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     # Verify caller is global admin or admin of provided org
     ctx = await authz.verify(
@@ -315,7 +314,7 @@ async def admin_delete_role(
     org_uuid: UUID,
     role_uuid: UUID,
     request: Request,
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     ctx = await authz.verify(
         auth,
@@ -343,7 +342,7 @@ async def admin_create_user(
     org_uuid: UUID,
     request: Request,
     payload: dict = Body(...),
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     await authz.verify(
         auth,
@@ -379,7 +378,7 @@ async def admin_update_user_role(
     user_uuid: UUID,
     request: Request,
     payload: dict = Body(...),
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     ctx = await authz.verify(
         auth,
@@ -422,7 +421,7 @@ async def admin_create_user_registration_link(
     org_uuid: UUID,
     user_uuid: UUID,
     request: Request,
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     try:
         user_org, _role_name = await db.instance.get_user_organization(user_uuid)
@@ -472,7 +471,7 @@ async def admin_get_user_detail(
     org_uuid: UUID,
     user_uuid: UUID,
     request: Request,
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     try:
         user_org, role_name = await db.instance.get_user_organization(user_uuid)
@@ -619,7 +618,7 @@ async def admin_update_user_display_name(
     user_uuid: UUID,
     request: Request,
     payload: dict = Body(...),
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     try:
         user_org, _role_name = await db.instance.get_user_organization(user_uuid)
@@ -653,7 +652,7 @@ async def admin_delete_user_credential(
     user_uuid: UUID,
     credential_uuid: UUID,
     request: Request,
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     try:
         user_org, _role_name = await db.instance.get_user_organization(user_uuid)
@@ -680,9 +679,7 @@ async def admin_delete_user_credential(
 
 
 @app.get("/permissions")
-async def admin_list_permissions(
+async def admin_list_permissions(request: Request, auth=AUTH_COOKIE):
-    request: Request, auth=Cookie(None, alias="__Host-auth")
-):
     ctx = await authz.verify(
         auth,
         ["auth:admin", "auth:org:*"],
@@ -705,7 +702,7 @@ async def admin_list_permissions(
 async def admin_create_permission(
     request: Request,
     payload: dict = Body(...),
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     await authz.verify(
         auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all
@@ -726,7 +723,7 @@ async def admin_update_permission(
     permission_id: str,
     display_name: str,
     request: Request,
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     await authz.verify(
         auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all
@@ -746,7 +743,7 @@ async def admin_update_permission(
 async def admin_rename_permission(
     request: Request,
     payload: dict = Body(...),
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     await authz.verify(
         auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all
@@ -777,7 +774,7 @@ async def admin_rename_permission(
 async def admin_delete_permission(
     permission_id: str,
     request: Request,
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     await authz.verify(
         auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all

passkey/fastapi/api.py

@@ -3,7 +3,6 @@ from contextlib import suppress
 from datetime import datetime, timedelta, timezone
 
 from fastapi import (
-    Cookie,
     Depends,
     FastAPI,
     HTTPException,
@@ -29,6 +28,7 @@ from ..globals import passkey as global_passkey
 from ..util import hostutil, passphrase, permutil
 from ..util.tokens import encode_session_key, session_key
 from . import authz, session, user
+from .session import AUTH_COOKIE
 
 bearer_auth = HTTPBearer(auto_error=True)
 
@@ -69,7 +69,7 @@ async def validate_token(
     request: Request,
     response: Response,
     perm: list[str] = Query([]),
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     """Validate the current session and extend its expiry.
 
@@ -110,7 +110,7 @@ async def forward_authentication(
     request: Request,
     response: Response,
     perm: list[str] = Query([]),
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     """Forward auth validation for Caddy/Nginx.
 
@@ -175,7 +175,7 @@ async def api_user_info(
     request: Request,
     response: Response,
     reset: str | None = None,
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     authenticated = False
     session_record = None
@@ -360,9 +360,7 @@ async def api_user_info(
 
 
 @app.post("/logout")
-async def api_logout(
+async def api_logout(request: Request, response: Response, auth=AUTH_COOKIE):
-    request: Request, response: Response, auth=Cookie(None, alias="__Host-auth")
-):
     if not auth:
         return {"message": "Already logged out"}
     try:

passkey/fastapi/mainapp.py

@@ -2,13 +2,14 @@ import logging
 import os
 from contextlib import asynccontextmanager
 
-from fastapi import Cookie, FastAPI, HTTPException, Request, Response
+from fastapi import FastAPI, HTTPException, Request, Response
 from fastapi.responses import FileResponse, RedirectResponse
 from fastapi.staticfiles import StaticFiles
 
 from passkey.util import frontend, hostutil, passphrase
 
 from . import admin, api, auth_host, ws
+from .session import AUTH_COOKIE
 
 
 @asynccontextmanager
@@ -63,9 +64,7 @@ app.mount(
 
 @app.get("/")
 @app.get("/auth/")
-async def frontapp(
+async def frontapp(request: Request, response: Response, auth=AUTH_COOKIE):
-    request: Request, response: Response, auth=Cookie(None, alias="__Host-auth")
-):
     """Serve the user profile SPA only for authenticated sessions; otherwise restricted SPA.
 
     Login / authentication UX is centralized in the restricted app.
@@ -98,7 +97,7 @@ async def admin_root_redirect():
 
 
 @app.get("/admin/", include_in_schema=False)
-async def admin_root(request: Request, auth=Cookie(None, alias="__Host-auth")):
+async def admin_root(request: Request, auth=AUTH_COOKIE):
     return await admin.adminapp(request, auth)  # Delegated (enforces access control)
 
 

passkey/fastapi/session.py

@@ -8,11 +8,12 @@ This module provides FastAPI-specific session management functionality:
 Generic session management functions have been moved to authsession.py
 """
 
-from fastapi import Request, Response, WebSocket
+from fastapi import Cookie, Request, Response, WebSocket
 
 from ..authsession import EXPIRES
 
 AUTH_COOKIE_NAME = "__Host-auth"
+AUTH_COOKIE = Cookie(None, alias=AUTH_COOKIE_NAME)
 
 
 def infodict(request: Request | WebSocket, type: str) -> dict:

passkey/fastapi/user.py

@@ -3,7 +3,6 @@ from uuid import UUID
 
 from fastapi import (
     Body,
-    Cookie,
     FastAPI,
     HTTPException,
     Request,
@@ -19,6 +18,7 @@ from ..globals import db
 from ..util import hostutil, passphrase, tokens
 from ..util.tokens import decode_session_key, session_key
 from . import session
+from .session import AUTH_COOKIE
 
 app = FastAPI()
 
@@ -28,7 +28,7 @@ async def user_update_display_name(
     request: Request,
     response: Response,
     payload: dict = Body(...),
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     if not auth:
         raise HTTPException(status_code=401, detail="Authentication Required")
@@ -46,9 +46,7 @@ async def user_update_display_name(
 
 
 @app.post("/logout-all")
-async def api_logout_all(
+async def api_logout_all(request: Request, response: Response, auth=AUTH_COOKIE):
-    request: Request, response: Response, auth=Cookie(None, alias="__Host-auth")
-):
     if not auth:
         return {"message": "Already logged out"}
     try:
@@ -65,7 +63,7 @@ async def api_delete_session(
     request: Request,
     response: Response,
     session_id: str,
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     if not auth:
         raise HTTPException(status_code=401, detail="Authentication Required")
@@ -97,7 +95,7 @@ async def api_delete_credential(
     request: Request,
     response: Response,
     uuid: UUID,
-    auth: str = Cookie(None, alias="__Host-auth"),
+    auth: str = AUTH_COOKIE,
 ):
     try:
         await delete_credential(uuid, auth, host=request.headers.get("host"))
@@ -110,7 +108,7 @@ async def api_delete_credential(
 async def api_create_link(
     request: Request,
     response: Response,
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     try:
         s = await get_session(auth, host=request.headers.get("host"))

passkey/fastapi/ws.py

@@ -2,14 +2,14 @@ import logging
 from functools import wraps
 from uuid import UUID
 
-from fastapi import Cookie, FastAPI, WebSocket, WebSocketDisconnect
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 from webauthn.helpers.exceptions import InvalidAuthenticationResponse
 
 from ..authsession import create_session, get_reset, get_session
 from ..globals import db, passkey
 from ..util import passphrase
 from ..util.tokens import create_token, session_key
-from .session import infodict
+from .session import AUTH_COOKIE, infodict
 
 
 # WebSocket error handling decorator
@@ -59,7 +59,7 @@ async def websocket_register_add(
     ws: WebSocket,
     reset: str | None = None,
     name: str | None = None,
-    auth=Cookie(None, alias="__Host-auth"),
+    auth=AUTH_COOKIE,
 ):
     """Register a new credential for an existing user.