Added a protected API

Author: hakan-persson

pyapp/app/auth_dep.py

@@ -0,0 +1,103 @@
+import logging
+from typing import Any, Dict
+from urllib.parse import urljoin
+
+import httpx
+from fastapi import HTTPException, Request, status
+from joserfc import jwt
+from joserfc.jwk import KeySet
+from pydantic import BaseModel, Field
+
+logger = logging.getLogger(__name__)
+
+_jwks_keyset_cache: Dict[str, KeySet] = {}
+
+
+class AuthenticatedUser(BaseModel):
+    subject: str
+    name: str | None = None
+    preferred_username: str | None = None
+    email: str | None = None
+    roles: list[str] = Field(default_factory=list)
+    # claims: Dict[str, Any]
+
+    def __str__(self) -> str:
+        return self.preferred_username or self.subject
+
+
+async def get_jwks_keyset(request: Request) -> KeySet | None:
+    """
+    Convert the JWKS dictionary into a key set usable for verification.
+    """
+    cache_key = str(request.base_url)
+
+    if cache_key in _jwks_keyset_cache:
+        return _jwks_keyset_cache[cache_key]
+
+    # url = urljoin(str(request.base_url), "auth/keys")
+    url = "https://dev.id.scouterna.se/realms/jamboree26/protocol/openid-connect/certs"
+    try:
+        async with httpx.AsyncClient(timeout=5.0) as http_client:
+            response = await http_client.get(url)
+            response.raise_for_status()
+            jwks_dict = response.json()
+    except Exception as exc:
+        logger.warning("Failed to fetch %s: %s", url, exc)
+        return None
+
+    try:
+        keyset = KeySet.import_key_set(jwks_dict)
+        _jwks_keyset_cache[cache_key] = keyset
+        return keyset
+    except Exception as exc:
+        logger.warning("Failed to parse JWKS: %s", exc)
+        return None
+
+
+async def decode_access_token(token: str, request: Request) -> Dict[str, Any]:
+    keyset = await get_jwks_keyset(request)
+    if keyset is None:
+        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Token validation unavailable")
+
+    try:
+        token_obj = jwt.decode(token, keyset)
+        registry = jwt.JWTClaimsRegistry(leeway=30)
+        registry.validate(token_obj.claims)
+        return dict(token_obj.claims)
+    except Exception as exc:
+        logger.warning("Failed to validate JWT: %s", exc)
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized") from exc
+
+
+def _extract_roles(claims: Dict[str, Any]) -> list[str]:
+    roles = set()
+    realm_access = claims.get("realm_access") or {}
+    realm_roles = realm_access.get("roles") or []
+    roles.update(role for role in realm_roles if isinstance(role, str))
+
+    resource_access = claims.get("resource_access") or {}
+    for resource in resource_access.values():
+        resource_roles = resource.get("roles") if isinstance(resource, dict) else []
+        roles.update(role for role in (resource_roles or []) if isinstance(role, str))
+
+    return sorted(roles)
+
+
+async def require_authenticated_user(request: Request) -> AuthenticatedUser:
+    """
+    FastAPI dependency that validates the auth cookie and returns user info + roles.
+    """
+    token = request.cookies.get("j26-auth_access-token")
+    if not token:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized")
+
+    claims = await decode_access_token(token, request)
+
+    return AuthenticatedUser(
+        subject=claims.get("sub", ""),
+        name=claims.get("name"),
+        preferred_username=claims.get("preferred_username"),
+        email=claims.get("email"),
+        roles=_extract_roles(claims),
+        # claims=claims,
+    )

pyapp/app/info_api.py

@@ -3,9 +3,10 @@
 import os
 
 import kubernetes
-from fastapi import APIRouter, Request, status
+from fastapi import APIRouter, Depends, Request, status
 from kubernetes.client.models import V1Pod
 
+from .auth_dep import AuthenticatedUser, require_authenticated_user
 from .config import get_settings
 
 settings = get_settings()
@@ -78,3 +79,16 @@ async def get_cookies(request: Request):
     Return all cookies sent with the request as a dict.
     """
     return dict(request.cookies)
+
+
+@router.get(
+    "/user",
+    response_model=dict,
+    status_code=status.HTTP_200_OK,
+    response_description="Returns the user object",
+)
+async def get_user(user: AuthenticatedUser = Depends(require_authenticated_user)):
+    """
+    Returns an autenticated user object as a dict.
+    """
+    return dict(user)