security(annotations): require scoped capabilities

Author: raylim

app/auth.py

@@ -4,19 +4,13 @@
 import hashlib
 import hmac
 import json
-import logging
 import time
-from typing import Any
 
-import httpx
 from fastapi import Depends, HTTPException, Request, status
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
-from jose import JWTError, jwt
 
 from .config import settings
 
-logger = logging.getLogger(__name__)
-
 
 class InvalidWsiToken(ValueError):
     """Raised when a WSI capability cannot be trusted."""
@@ -82,11 +76,18 @@ def validate_wsi_token(
     if "wsi:read" in required:
         if payload.get("wsi_auth_version") != 2:
             raise InvalidWsiToken("unsupported WSI authorization contract")
-        if not isinstance(payload.get("image_id"), str) or not payload["image_id"].strip():
+        if (
+            not isinstance(payload.get("image_id"), str)
+            or not payload["image_id"].strip()
+        ):
             raise InvalidWsiToken("invalid token image")
         for claim in ("tile_source_sha256", "thumbnail_source_sha256"):
             value = payload.get(claim)
-            if not isinstance(value, str) or len(value) != 64 or any(char not in "0123456789abcdef" for char in value):
+            if (
+                not isinstance(value, str)
+                or len(value) != 64
+                or any(char not in "0123456789abcdef" for char in value)
+            ):
                 raise InvalidWsiToken("invalid token source binding")
         for claim in ("thumbnail_width", "thumbnail_height"):
             value = payload.get(claim)
@@ -102,40 +103,13 @@ def validate_wsi_token(
 
 
 _bearer = HTTPBearer(auto_error=False)
-_jwks_cache: dict[str, Any] = {}
-_jwks_fetched_at: float = 0.0
-_JWKS_TTL = 3600
-
-
-async def _get_jwks() -> dict[str, Any]:
-    global _jwks_cache, _jwks_fetched_at
-    if time.monotonic() - _jwks_fetched_at < _JWKS_TTL and _jwks_cache:
-        return _jwks_cache
-    if not settings.keycloak_jwks_url:
-        raise HTTPException(
-            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-            detail="KEYCLOAK_JWKS_URL is not configured",
-        )
-    try:
-        async with httpx.AsyncClient(timeout=10) as client:
-            resp = await client.get(settings.keycloak_jwks_url)
-            resp.raise_for_status()
-            _jwks_cache = resp.json()
-            _jwks_fetched_at = time.monotonic()
-            return _jwks_cache
-    except Exception as exc:
-        logger.error("Failed to fetch JWKS from %s: %s", settings.keycloak_jwks_url, exc)
-        raise HTTPException(
-            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-            detail="Unable to reach Keycloak JWKS endpoint",
-        )
 
 
 async def require_user(
     request: Request,
     creds: HTTPAuthorizationCredentials | None = Depends(_bearer),
-) -> dict[str, Any]:
-    """Return the authenticated Keycloak subject and groups for annotations."""
+) -> dict:
+    """Return the subject and study scope from a cBioPortal capability."""
     if not settings.annotation_auth_enabled:
         return {"sub": "dev-user", "groups": []}
     if creds is None:
@@ -159,22 +133,9 @@ async def require_user(
             "groups": [],
             "study_id": capability["study_id"],
         }
-    except InvalidWsiToken:
-        pass
-    try:
-        payload = jwt.decode(
-            creds.credentials,
-            await _get_jwks(),
-            algorithms=["RS256"],
-            options={"verify_aud": False},
-        )
-    except JWTError as exc:
+    except InvalidWsiToken as exc:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
-            detail=f"Invalid token: {exc}",
+            detail="Invalid annotation capability",
             headers={"WWW-Authenticate": "Bearer"},
-        )
-    sub: str = payload.get("sub", "")
-    if not sub:
-        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token missing 'sub' claim")
-    return {"sub": sub, "groups": payload.get("groups", []), "study_id": None}
+        ) from exc

tests/test_auth.py

@@ -5,9 +5,12 @@
 import time
 
 import pytest
+from fastapi import HTTPException, Request
+from fastapi.security import HTTPAuthorizationCredentials
 
 from app.auth import (
     InvalidWsiToken,
+    require_user,
     source_digest,
     validate_wsi_auth_configuration,
     validate_wsi_token,
@@ -16,27 +19,39 @@
 
 def make_token(secret: str, **claims) -> str:
     def encode(value):
-        return base64.urlsafe_b64encode(json.dumps(value).encode()).rstrip(b"=").decode()
+        return (
+            base64.urlsafe_b64encode(json.dumps(value).encode()).rstrip(b"=").decode()
+        )
 
     header = encode({"alg": "HS256", "typ": "JWT"})
     payload = encode(claims)
     signing_input = f"{header}.{payload}".encode()
-    signature = base64.urlsafe_b64encode(
-        hmac.new(secret.encode(), signing_input, hashlib.sha256).digest()
-    ).rstrip(b"=").decode()
+    signature = (
+        base64.urlsafe_b64encode(
+            hmac.new(secret.encode(), signing_input, hashlib.sha256).digest()
+        )
+        .rstrip(b"=")
+        .decode()
+    )
     return f"{header}.{payload}.{signature}"
 
 
 def make_raw_token(secret: str, header, payload) -> str:
     def encode(value):
-        return base64.urlsafe_b64encode(json.dumps(value).encode()).rstrip(b"=").decode()
+        return (
+            base64.urlsafe_b64encode(json.dumps(value).encode()).rstrip(b"=").decode()
+        )
 
     encoded_header = encode(header)
     encoded_payload = encode(payload)
     signing_input = f"{encoded_header}.{encoded_payload}".encode()
-    signature = base64.urlsafe_b64encode(
-        hmac.new(secret.encode(), signing_input, hashlib.sha256).digest()
-    ).rstrip(b"=").decode()
+    signature = (
+        base64.urlsafe_b64encode(
+            hmac.new(secret.encode(), signing_input, hashlib.sha256).digest()
+        )
+        .rstrip(b"=")
+        .decode()
+    )
     return f"{encoded_header}.{encoded_payload}.{signature}"
 
 
@@ -62,9 +77,12 @@ def valid_claims(**overrides):
 
 def test_valid_wsi_token():
     secret = "s" * 32
-    assert validate_wsi_token(
-        make_token(secret, **valid_claims()), secret, "cbioportal-wsi"
-    )["sub"] == "user@example.org"
+    assert (
+        validate_wsi_token(
+            make_token(secret, **valid_claims()), secret, "cbioportal-wsi"
+        )["sub"]
+        == "user@example.org"
+    )
 
 
 @pytest.mark.parametrize(
@@ -82,11 +100,14 @@ def test_invalid_wsi_auth_configuration_is_rejected(secret, audience, max_ttl):
         validate_wsi_auth_configuration(secret, audience, max_ttl)
 
 
-@pytest.mark.parametrize("change", [
-    {"scope": "wsi:write"},
-    {"aud": "other-service"},
-    {"exp": int(time.time()) - 1},
-])
+@pytest.mark.parametrize(
+    "change",
+    [
+        {"scope": "wsi:write"},
+        {"aud": "other-service"},
+        {"exp": int(time.time()) - 1},
+    ],
+)
 def test_invalid_claims_are_rejected(change):
     secret = "s" * 32
     claims = valid_claims(**change)
@@ -114,13 +135,16 @@ def test_non_object_header_and_payload_are_rejected():
         )
 
 
-@pytest.mark.parametrize("change", [
-    {"study_id": ""},
-    {"image_id": ""},
-    {"tile_source_sha256": ""},
-    {"thumbnail_width": 0},
-    {"exp": int(time.time()) + 1000},
-])
+@pytest.mark.parametrize(
+    "change",
+    [
+        {"study_id": ""},
+        {"image_id": ""},
+        {"tile_source_sha256": ""},
+        {"thumbnail_width": 0},
+        {"exp": int(time.time()) + 1000},
+    ],
+)
 def test_source_bound_claims_and_max_ttl_are_required(change):
     secret = "s" * 32
     with pytest.raises(InvalidWsiToken):
@@ -169,3 +193,23 @@ def test_annotation_capability_requires_both_annotation_scopes():
             "cbioportal-wsi",
             required_scopes={"annotations:read", "annotations:write"},
         )
+
+
+@pytest.mark.asyncio
+async def test_annotation_auth_rejects_non_capability_tokens():
+    request = Request(
+        {
+            "type": "http",
+            "method": "GET",
+            "path": "/annotations",
+            "headers": [],
+        }
+    )
+    credentials = HTTPAuthorizationCredentials(
+        scheme="Bearer", credentials="generic-keycloak-token"
+    )
+
+    with pytest.raises(HTTPException) as error:
+        await require_user(request, credentials)
+
+    assert error.value.status_code == 401