fix(auth): require an access-token type on the OIDC bearer path (supersedes #1077)

backend/api/auth.py

@@ -148,6 +148,18 @@ def _build_oidc_jwks_client() -> PyJWKClient | None:
 SESSION_ISSUER = "naruon-control-plane"
 SESSION_AUDIENCE = "naruon-api"
 JWT_DECODE_REQUIRED_CLAIMS = ("exp", "iss", "aud")
+# An OIDC ID token carries aud == client_id, so verifying signature/issuer/aud
+# alone lets a frontend ID token be replayed as an API access token (RFC 8725
+# §3.11). naruon's API credential is the OIDC access token (the frontend sends
+# token_response.access_token). Accept only material the IdP marks as an access
+# token: Keycloak sets the body claim typ="Bearer"; RFC 9068 sets the header
+# typ "at+jwt". ID tokens (Keycloak typ="ID") and unmarked material are rejected.
+# ponytail: extend these sets if a non-Keycloak/non-RFC9068 IdP is onboarded.
+# Bandit B105 is inapplicable: these are RFC 9068's public typ identifiers.
+OIDC_ACCESS_TOKEN_HEADER_TYPES = frozenset(  # nosec B105
+    {"at+jwt", "application/at+jwt"}
+)
+OIDC_ACCESS_TOKEN_BODY_TYPES = frozenset({"bearer"})
 MIN_SESSION_SECRET_BYTES = 32
 MAX_SIGNED_SESSION_EXPIRATION_SECONDS = 12 * 60 * 60
 MAX_SIGNED_SESSION_CLOCK_SKEW_SECONDS = 60
@@ -263,10 +275,48 @@ def _decode_cached_oidc_session_payload(token: str) -> dict[str, Any]:
             raise _authentication_error()
         if not isinstance(payload, dict):
             raise _authentication_error()
+        _require_oidc_access_token(header, payload)
         return payload
     raise _authentication_error()
 
 
+def _require_oidc_access_token(header: dict[str, Any], payload: dict[str, Any]) -> None:
+    """Reject OIDC ID tokens replayed as API access tokens (RFC 8725 §3.11).
+
+    The frontend transmits the OIDC access token as the API bearer, so only
+    material the IdP marks as an access token may build a session: an RFC 9068
+    header typ of "at+jwt", or a Keycloak body typ of "Bearer". ID tokens
+    (Keycloak typ="ID") and tokens with no access-token marker are rejected even
+    when signature, issuer, and audience verify.
+    """
+    raw_header_type = header.get("typ")
+    raw_body_type = payload.get("typ")
+    if raw_header_type is not None and not isinstance(raw_header_type, str):
+        raise _authentication_error()
+    if raw_body_type is not None and not isinstance(raw_body_type, str):
+        raise _authentication_error()
+
+    header_type = (
+        raw_header_type.strip().lower() if isinstance(raw_header_type, str) else None
+    )
+    body_type = (
+        raw_body_type.strip().lower() if isinstance(raw_body_type, str) else None
+    )
+    header_marks_access = header_type in OIDC_ACCESS_TOKEN_HEADER_TYPES
+    body_marks_access = body_type in OIDC_ACCESS_TOKEN_BODY_TYPES
+    header_is_compatible = (
+        header_type is None or header_type == "jwt" or header_marks_access
+    )
+    body_is_compatible = body_type is None or body_marks_access
+    if (
+        (header_marks_access or body_marks_access)
+        and header_is_compatible
+        and body_is_compatible
+    ):
+        return
+    raise _authentication_error()
+
+
 def _reject_unsupported_critical_headers(header: dict[str, Any]) -> None:
     if "crit" in header:
         raise _authentication_error()

backend/tests/test_auth_real.py

@@ -972,6 +972,7 @@ def mock_jwt_decode(*args, **kwargs):
         return {
             "iss": "https://login.example.test/realms/naruon",
             "aud": "naruon-api",
+            "typ": "Bearer",
             "sub": "alice",
             "role": "member",
             "org": "org-acme",
@@ -1026,6 +1027,7 @@ def mock_jwt_decode(*args, **kwargs):
         return {
             "iss": "https://login.example.test/realms/naruon",
             "aud": ("naruon-api", "naruon-admin"),
+            "typ": "Bearer",
             "sub": "alice",
             "role": "member",
             "org": "org-acme",
@@ -1051,6 +1053,112 @@ def mock_jwt_decode(*args, **kwargs):
     assert context.user_id == "alice"
 
 
+def _oidc_settings_snapshot():
+    return (
+        settings.OIDC_ISSUER_URL,
+        settings.OIDC_CLIENT_ID,
+        settings.AUTH_SESSION_HMAC_SECRET,
+    )
+
+
+def _apply_oidc_settings():
+    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)
+
+
+def _restore_oidc_settings(previous):
+    (
+        settings.OIDC_ISSUER_URL,
+        settings.OIDC_CLIENT_ID,
+        settings.AUTH_SESSION_HMAC_SECRET,
+    ) = previous
+
+
+def _install_oidc_decode(monkeypatch, payload):
+    import jwt
+
+    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(),))
+    monkeypatch.setattr(jwt, "decode", lambda *a, **k: payload)
+
+
+def _oidc_claims(**overrides):
+    payload = {
+        "iss": "https://login.example.test/realms/naruon",
+        "aud": "naruon-api",
+        "sub": "alice",
+        "role": "member",
+        "org": "org-acme",
+        "groups": ["group-1", "group-2"],
+        "workspace": "workspace-org-acme",
+        "exp": int(time.time()) + 300,
+    }
+    payload.update(overrides)
+    return payload
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "header,claims",
+    [
+        # Keycloak ID token replayed as an API bearer (the Strix finding).
+        ({"alg": "RS256", "typ": "JWT", "kid": "test-key"}, {"typ": "ID"}),
+        # Forged/opaque material with no access-token marker (the PoC shape).
+        ({"alg": "RS256", "typ": "JWT", "kid": "test-key"}, {}),
+        # An ID-token typ must not sneak through via the body either.
+        ({"alg": "RS256", "typ": "JWT", "kid": "test-key"}, {"typ": "id"}),
+        # Contradictory signed markers must fail closed instead of trusting one side.
+        ({"alg": "RS256", "typ": "ID", "kid": "test-key"}, {"typ": "Bearer"}),
+        ({"alg": "RS256", "typ": "at+jwt", "kid": "test-key"}, {"typ": "ID"}),
+        # Explicit malformed marker values are not equivalent to absent markers.
+        ({"alg": "RS256", "typ": "", "kid": "test-key"}, {"typ": "Bearer"}),
+        ({"alg": "RS256", "typ": "at+jwt", "kid": "test-key"}, {"typ": ""}),
+        ({"alg": "RS256", "typ": 7, "kid": "test-key"}, {"typ": "Bearer"}),
+    ],
+)
+async def test_oidc_rejects_id_token_replayed_as_api_bearer(
+    monkeypatch, header, claims
+):
+    previous = _oidc_settings_snapshot()
+    _apply_oidc_settings()
+    _install_oidc_decode(monkeypatch, _oidc_claims(**claims))
+    token = _signed_session_token(_valid_session_payload(), header=header)
+
+    try:
+        with pytest.raises(HTTPException) as exc:
+            await get_auth_context(authorization=f"Bearer {token}")
+    finally:
+        _restore_oidc_settings(previous)
+
+    assert exc.value.status_code == 401
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("header_type", ["at+jwt", "application/at+jwt"])
+async def test_oidc_accepts_rfc9068_access_token_header_typ(monkeypatch, header_type):
+    previous = _oidc_settings_snapshot()
+    _apply_oidc_settings()
+    # RFC 9068 access token: marker in the header typ, no body typ claim.
+    _install_oidc_decode(monkeypatch, _oidc_claims())
+    token = _signed_session_token(
+        _valid_session_payload(),
+        header={"alg": "RS256", "typ": header_type, "kid": "test-key"},
+    )
+
+    try:
+        context = await get_auth_context(authorization=f"Bearer {token}")
+    finally:
+        _restore_oidc_settings(previous)
+
+    assert context.session_verifier == "oidc"
+    assert context.user_id == "alice"
+
+
 @pytest.mark.asyncio
 async def test_oidc_session_rejects_missing_client_id_after_decode(monkeypatch):
     import jwt
@@ -1072,6 +1180,7 @@ class MockKey:
     def mock_jwt_decode(*args, **kwargs):
         return {
             "iss": "https://login.example.test/realms/naruon",
+            "typ": "Bearer",
             "sub": "alice",
             "role": "member",
             "org": "org-acme",

backend/tests/test_repo_hygiene.py

@@ -131,6 +131,10 @@ def test_infra_compose_services_use_read_only_hardening_anchor():
         "keycloak",
     ):
         assert f"  {service}:\n    <<: *service-hardening" in compose
+        service_block = compose.split(f"  {service}:", 1)[1].split("\n\n  ", 1)[0]
+        assert "security_opt:" in service_block
+        assert "- no-new-privileges:true" in service_block
+        assert "read_only: true" in service_block
 
     assert "GF_SECURITY_ADMIN_PASSWORD=admin" not in compose
     assert (