fix(auth): require preferred username claim

AUTHENTICATION_AND_USER_MANAGEMENT.md

@@ -16,7 +16,7 @@ trust frontend-supplied user IDs.
 ## Login Snapshot Refresh
 
 Every authenticated metadata-user resolution validates the Keycloak access token
-and reads `sub`, `preferred_username` or `username`, and `email` claims. The
+and reads `sub`, `preferred_username`, and `email` claims. The
 backend finds `users` by `keycloak_id = sub`, rejects inactive or missing users,
 then refreshes `username`, `email`, and `last_login_at`.
 

app/auth/keycloak_dependencies.py

@@ -73,14 +73,18 @@ async def get_current_keycloak_sub(
         ) from exc
 
 
-async def get_current_keycloak_username(
+def get_keycloak_preferred_username(payload: dict) -> str:
-    payload: dict = Depends(get_current_keycloak_payload),
+    username = payload.get("preferred_username")
-) -> str:
-    username = payload.get("preferred_username") or payload.get("username")
     if not username:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Missing username claim",
+            detail="Missing preferred_username claim",
             headers={"WWW-Authenticate": "Bearer"},
         )
     return str(username)
+
+
+async def get_current_keycloak_username(
+    payload: dict = Depends(get_current_keycloak_payload),
+) -> str:
+    return get_keycloak_preferred_username(payload)

app/auth/metadata_dependencies.py

@@ -6,7 +6,10 @@ from fastapi import Depends, HTTPException, status
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.auth.keycloak_dependencies import get_current_keycloak_payload
+from app.auth.keycloak_dependencies import (
+    get_current_keycloak_payload,
+    get_keycloak_preferred_username,
+)
 from app.infra.db.metadb.database import get_metadata_session
 from app.infra.db.metadb.repositories.metadata_repository import MetadataRepository
 
@@ -38,11 +41,6 @@ def _keycloak_sub_from_payload(payload: dict) -> UUID:
         ) from exc
 
 
-def _username_from_payload(payload: dict) -> str | None:
-    username = payload.get("preferred_username") or payload.get("username")
-    return str(username) if username else None
-
-
 def _email_from_payload(payload: dict) -> str | None:
     email = payload.get("email")
     return str(email) if email else None
@@ -53,6 +51,7 @@ async def get_current_metadata_user(
     metadata_repo: MetadataRepository = Depends(get_metadata_repository),
 ):
     keycloak_sub = _keycloak_sub_from_payload(keycloak_payload)
+    username = get_keycloak_preferred_username(keycloak_payload)
     try:
         user = await metadata_repo.get_user_by_keycloak_id(keycloak_sub)
     except SQLAlchemyError as exc:
@@ -71,7 +70,7 @@ async def get_current_metadata_user(
     try:
         user = await metadata_repo.refresh_user_keycloak_snapshot(
             user,
-            username=_username_from_payload(keycloak_payload),
+            username=username,
             email=_email_from_payload(keycloak_payload),
         )
     except SQLAlchemyError as exc:

tests/auth/test_keycloak_dependencies.py

@@ -0,0 +1,30 @@
+import pytest
+from fastapi import HTTPException
+
+from app.auth.keycloak_dependencies import get_current_keycloak_username
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+@pytest.mark.anyio
+async def test_current_username_uses_preferred_username_only():
+    username = await get_current_keycloak_username(
+        {
+            "preferred_username": "tjwater",
+            "username": "legacy-name",
+        }
+    )
+
+    assert username == "tjwater"
+
+
+@pytest.mark.anyio
+async def test_current_username_rejects_username_fallback():
+    with pytest.raises(HTTPException) as exc:
+        await get_current_keycloak_username({"username": "legacy-name"})
+
+    assert exc.value.status_code == 401
+    assert exc.value.detail == "Missing preferred_username claim"

tests/auth/test_metadata_dependencies.py

@@ -81,3 +81,26 @@ async def test_current_metadata_user_rejects_invalid_keycloak_sub():
     assert exc.value.status_code == 401
     repo.get_user_by_keycloak_id.assert_not_called()
     repo.refresh_user_keycloak_snapshot.assert_not_called()
+
+
+@pytest.mark.anyio
+async def test_current_metadata_user_rejects_username_claim_fallback():
+    keycloak_id = uuid4()
+    repo = SimpleNamespace(
+        get_user_by_keycloak_id=AsyncMock(),
+        refresh_user_keycloak_snapshot=AsyncMock(),
+    )
+
+    with pytest.raises(HTTPException) as exc:
+        await metadata_dependencies.get_current_metadata_user(
+            {
+                "sub": str(keycloak_id),
+                "username": "legacy-name",
+            },
+            metadata_repo=repo,
+        )
+
+    assert exc.value.status_code == 401
+    assert exc.value.detail == "Missing preferred_username claim"
+    repo.get_user_by_keycloak_id.assert_not_called()
+    repo.refresh_user_keycloak_snapshot.assert_not_called()