fix(auth): select OIDC signing key by kid

backend/api/auth.py

@@ -243,28 +243,31 @@ def _decode_cached_oidc_session_payload(token: str) -> dict[str, Any]:
         raise _authentication_error()
     header = _oidc_unverified_header(token)
     key_id = header["kid"].strip()
+    matching_keys = [
+        signing_key
+        for signing_key in _cached_oidc_signing_keys
+        if getattr(signing_key, "key_id", None) == key_id
+    ]
+    if len(matching_keys) != 1:
+        raise _authentication_error()
 
-    for signing_key in _cached_oidc_signing_keys:
-        try:
-            payload = jwt.decode(
-                token,
-                signing_key.key,
-                algorithms=["RS256"],
-                audience=settings.OIDC_CLIENT_ID,
-                issuer=settings.OIDC_ISSUER_URL,
-                options={
-                    "require": JWT_DECODE_REQUIRED_CLAIMS,
-                    "verify_signature": True,
-                },
-            )
-        except jwt.PyJWTError:
-            continue
-        if getattr(signing_key, "key_id", None) != key_id:
-            raise _authentication_error()
-        if not isinstance(payload, dict):
-            raise _authentication_error()
-        return payload
-    raise _authentication_error()
+    try:
+        payload = jwt.decode(
+            token,
+            matching_keys[0].key,
+            algorithms=["RS256"],
+            audience=settings.OIDC_CLIENT_ID,
+            issuer=settings.OIDC_ISSUER_URL,
+            options={
+                "require": JWT_DECODE_REQUIRED_CLAIMS,
+                "verify_signature": True,
+            },
+        )
+    except jwt.PyJWTError:
+        raise _authentication_error() from None
+    if not isinstance(payload, dict):
+        raise _authentication_error()
+    return payload
 
 
 def _reject_unsupported_critical_headers(header: dict[str, Any]) -> None:
@@ -413,7 +416,10 @@ def _reject_signed_session_admin_payload(payload: dict[str, Any]) -> None:
         raise _authentication_error()
     # Admin roles require explicit server-side assignment, not externally
     # supplied HMAC or enterprise OIDC session claims.
-    if role_claim in ADMIN_ROLES:
+    normalized_role = role_claim.strip()
+    # Reject surrounding whitespace before the later claim normalization can
+    # turn a non-admin-looking value into an administrative role.
+    if normalized_role != role_claim or normalized_role in ADMIN_ROLES:
         raise _authentication_error()
 
 

backend/tests/test_auth_real.py

@@ -665,6 +665,31 @@ async def test_hmac_session_rejects_admin_role_claim(admin_role: str):
     assert exc.value.status_code == 401
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "padded_admin_role",
+    (
+        " system_admin",
+        "platform_admin ",
+        "\ttenant_admin",
+        "organization_admin\n",
+    ),
+)
+async def test_hmac_session_rejects_whitespace_padded_admin_role_claim(
+    padded_admin_role: str,
+):
+    """A signed role must not gain admin meaning after whitespace is stripped."""
+    settings.AUTH_SESSION_HMAC_SECRET = SecretStr(TEST_SESSION_HMAC_SECRET)
+    token = _signed_session_token(
+        _valid_session_payload(role=padded_admin_role, org="org-acme")
+    )
+
+    with pytest.raises(HTTPException) as exc:
+        await get_auth_context(authorization=f"Bearer {token}")
+
+    assert exc.value.status_code == 401
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("role_claim", (["system_admin"], 123, True, None))
 async def test_hmac_session_rejects_non_string_role_claim(role_claim: object):
@@ -1187,10 +1212,17 @@ class MockKey:
         key = "trusted_public_key"
 
     monkeypatch.setattr("api.auth.jwks_client", object())
-    monkeypatch.setattr("api.auth._cached_oidc_signing_keys", (MockKey(),))
+    class DecoyKey:
+        key_id = "decoy-key"
+        key = "decoy_public_key"
+
+    monkeypatch.setattr("api.auth._cached_oidc_signing_keys", (MockKey(), DecoyKey()))
+
+    decode_called = False
 
     def mock_jwt_decode(token, key, **kwargs):
-        assert key == "trusted_public_key"
+        nonlocal decode_called
+        decode_called = True
         return {
             "iss": "https://login.example.test/realms/naruon",
             "aud": "naruon-api",
@@ -1217,6 +1249,7 @@ def mock_jwt_decode(token, key, **kwargs):
         settings.AUTH_SESSION_HMAC_SECRET = previous_secret
 
     assert exc.value.status_code == 401
+    assert decode_called is False
 
 
 @pytest.mark.asyncio
@@ -1319,6 +1352,60 @@ def mock_jwt_decode(*args, **kwargs):
     assert exc.value.status_code == 401
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "padded_admin_role",
+    (" system_admin", "platform_admin ", "\ttenant_admin", "organization_admin\n"),
+)
+async def test_oidc_session_rejects_whitespace_padded_admin_role_claim(
+    monkeypatch, padded_admin_role: str
+):
+    """OIDC sessions use the same strict role boundary as HMAC sessions."""
+    import jwt
+
+    previous_issuer_url = settings.OIDC_ISSUER_URL
+    previous_client_id = settings.OIDC_CLIENT_ID
+    previous_secret = settings.AUTH_SESSION_HMAC_SECRET
+    settings.OIDC_ISSUER_URL = "https://login.example.test/realms/naruon"
+    settings.OIDC_CLIENT_ID = "naruon-api"
+    settings.AUTH_SESSION_HMAC_SECRET = SecretStr(TEST_SESSION_HMAC_SECRET)
+
+    class MockKey:
+        key_id = "test-key"
+        key = "public_key"
+
+    monkeypatch.setattr("api.auth.jwks_client", object())
+    monkeypatch.setattr("api.auth._cached_oidc_signing_keys", (MockKey(),))
+
+    def mock_jwt_decode(*args, **kwargs):
+        return {
+            "iss": "https://login.example.test/realms/naruon",
+            "aud": "naruon-api",
+            "sub": "operator",
+            "role": padded_admin_role,
+            "org": None,
+            "groups": [],
+            "workspace": "workspace-root",
+            "exp": int(time.time()) + 300,
+        }
+
+    monkeypatch.setattr(jwt, "decode", mock_jwt_decode)
+    token = _signed_session_token(
+        _valid_session_payload(),
+        header={"alg": "RS256", "typ": "JWT", "kid": "test-key"},
+    )
+
+    try:
+        with pytest.raises(HTTPException) as exc:
+            await get_auth_context(authorization=f"Bearer {token}")
+    finally:
+        settings.OIDC_ISSUER_URL = previous_issuer_url
+        settings.OIDC_CLIENT_ID = previous_client_id
+        settings.AUTH_SESSION_HMAC_SECRET = previous_secret
+
+    assert exc.value.status_code == 401
+
+
 @pytest.mark.asyncio
 async def test_oidc_validation_failure_does_not_fallback_to_signed_session(monkeypatch):
     import jwt