Add permission check on forward-auth and validate.

passkey/fastapi/api.py

@@ -21,7 +21,7 @@ from ..globals import db
 from ..globals import passkey as global_passkey
 from ..util import tokens
 from ..util.tokens import session_key
-from . import session
+from . import authz, session
 
 bearer_auth = HTTPBearer(auto_error=True)
 
@@ -43,13 +43,17 @@ def register_api_routes(app: FastAPI):
         return ctx, is_global_admin, is_org_admin
 
     @app.post("/auth/validate")
-    async def validate_token(response: Response, auth=Cookie(None)):
-        """Lightweight token validation endpoint."""
-        s = await get_session(auth)
-        return {
-            "valid": True,
-            "user_uuid": str(s.user_uuid),
-        }
+    async def validate_token(
+        response: Response, perm: str | None = None, auth=Cookie(None)
+    ):
+        """Lightweight token validation endpoint.
+
+        Query Params:
+        - perm: optional permission ID the caller must possess
+        """
+
+        s = await authz.verify(auth, perm)
+        return {"valid": True, "user_uuid": str(s.user_uuid)}
 
     @app.post("/auth/user-info")
     async def api_user_info(reset: str | None = None, auth=Cookie(None)):

passkey/fastapi/authz.py

@@ -0,0 +1,36 @@
+"""Authorization utilities shared across FastAPI endpoints.
+
+Provides helper(s) to validate a session token (from cookie) and optionally
+enforce that the user possesses a given permission (either via their role or
+their organization level permissions).
+"""
+
+from fastapi import HTTPException
+
+from ..authsession import get_session
+from ..globals import db
+from ..util.tokens import session_key
+
+
+async def verify(auth: str | None, perm: str | None):
+    """Validate session token and optional permission.
+
+    Returns the Session object on success. Raises HTTPException on failure.
+    401: unauthenticated / invalid session
+    403: missing required permission
+    """
+    if not auth:
+        raise HTTPException(status_code=401, detail="Authentication required")
+    session = await get_session(auth)
+    if perm:
+        ctx = await db.instance.get_session_context(session_key(auth))
+        if not ctx:
+            raise HTTPException(status_code=401, detail="Session not found")
+        role_perms = set(ctx.role.permissions or [])
+        org_perms = set(ctx.org.permissions or []) if ctx.org else set()
+        if perm not in role_perms and perm not in org_perms:
+            raise HTTPException(status_code=403, detail="Permission required")
+    return session
+
+
+__all__ = ["verify"]

passkey/fastapi/mainapp.py

@@ -4,12 +4,12 @@ import os
 from contextlib import asynccontextmanager
 from pathlib import Path
 
-from fastapi import Cookie, FastAPI, Request, Response
+from fastapi import Cookie, FastAPI, HTTPException, Request, Response
 from fastapi.responses import FileResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
 
 from ..authsession import get_session
-from . import ws
+from . import authz, ws
 from .api import register_api_routes
 from .reset import register_reset_routes
 
@@ -72,26 +72,26 @@ app.mount("/auth/ws", ws.app)
 
 
 @app.get("/auth/forward-auth")
-async def forward_authentication(request: Request, auth=Cookie(None)):
-    """A validation endpoint to use with Caddy forward_auth or Nginx auth_request."""
-    if auth:
-        with contextlib.suppress(ValueError):
-            s = await get_session(auth)
-            # If authenticated, return a success response
-            if s.info and s.info["type"] == "authenticated":
+async def forward_authentication(
+    request: Request, perm: str | None = None, auth=Cookie(None)
+):
+    """A validation endpoint to use with Caddy forward_auth or Nginx auth_request.
+
+    Query Params:
+    - perm: optional permission ID the authenticated user must possess (role or org).
+
+    Success: 204 No Content with x-auth-user-uuid header.
+    Failure (unauthenticated / unauthorized): 4xx with index.html body so the
+    client (reverse proxy or browser) can initiate auth flow.
+    """
+    try:
+        s = await authz.verify(auth, perm)
         return Response(
             status_code=204,
-                    headers={
-                        "x-auth-user-uuid": str(s.user_uuid),
-                    },
-                )
-
-    # Serve the index.html of the authentication app if not authenticated
-    return FileResponse(
-        STATIC_DIR / "index.html",
-        status_code=401,
-        headers={"www-authenticate": "PrivateToken"},
+            headers={"x-auth-user-uuid": str(s.user_uuid)},
         )
+    except HTTPException as e:
+        return FileResponse(STATIC_DIR / "index.html", e.status_code)
 
 
 # Serve static files