Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 28 additions & 0 deletions backend/api/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -232,6 +232,13 @@ def _oidc_unverified_header(token: str) -> dict[str, Any]:
_reject_unsupported_critical_headers(header)
if header.get("alg") != "RS256":
raise _authentication_error()
# RFC 9068 access tokens declare "at+jwt"; only ID-token material ("JWT",
# or an absent typ from IdPs that omit it) may become an API session.
token_type = header.get("typ")
if token_type is not None and (
not isinstance(token_type, str) or token_type.strip().upper() != "JWT"
):
raise _authentication_error()
key_id = header.get("kid")
if not isinstance(key_id, str) or not key_id.strip():
raise _authentication_error()
Expand Down Expand Up @@ -263,10 +270,31 @@ def _decode_cached_oidc_session_payload(token: str) -> dict[str, Any]:
raise _authentication_error()
if not isinstance(payload, dict):
raise _authentication_error()
_reject_non_id_token_payload(payload)
return payload
raise _authentication_error()


def _reject_non_id_token_payload(payload: dict[str, Any]) -> None:
"""Reject same-issuer OIDC material that is not an ID token for this client.

The enterprise IdP mints naruon session claims directly into the ID token,
so the ID token is the intended API bearer credential. Access tokens and
tokens minted for other clients can share this issuer and carry this API's
client_id in aud; their claim shapes distinguish them: access tokens carry
token_use/scope/scp claims, and tokens for other clients name that client
in azp (OIDC Core 3.1.3.7).
"""
usage_claim = payload.get("token_use")
if usage_claim is not None and usage_claim != "id":
raise _authentication_error()
if "scope" in payload or "scp" in payload:
raise _authentication_error()
authorized_party = payload.get("azp")
if authorized_party is not None and authorized_party != settings.OIDC_CLIENT_ID:
raise _authentication_error()


def _reject_unsupported_critical_headers(header: dict[str, Any]) -> None:
if "crit" in header:
raise _authentication_error()
Expand Down
128 changes: 128 additions & 0 deletions backend/tests/test_auth_real.py
Original file line number Diff line number Diff line change
Expand Up @@ -1051,6 +1051,134 @@ def mock_jwt_decode(*args, **kwargs):
assert context.user_id == "alice"


def _oidc_test_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 _snapshot_oidc_settings():
return (
settings.OIDC_ISSUER_URL,
settings.OIDC_CLIENT_ID,
settings.AUTH_SESSION_HMAC_SECRET,
)


def _oidc_id_token_payload(**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


class _OidcMockKey:
key_id = "test-key"
key = "public_key"


@pytest.mark.asyncio
@pytest.mark.parametrize("token_type", ["at+jwt", 123])
async def test_oidc_rejects_access_token_typ_header(monkeypatch, token_type):
import jwt

previous = _snapshot_oidc_settings()
_oidc_test_settings()
monkeypatch.setattr("api.auth.jwks_client", object())
monkeypatch.setattr("api.auth._cached_oidc_signing_keys", (_OidcMockKey(),))
monkeypatch.setattr(jwt, "decode", lambda *a, **k: _oidc_id_token_payload())

token = _signed_session_token(
_valid_session_payload(),
header={"alg": "RS256", "typ": token_type, "kid": "test-key"},
)
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(
"payload_overrides",
[
{"token_use": "access"},
{"scope": "openid api.read"},
{"scp": ["api.read"]},
{"azp": "other-client"},
],
)
async def test_oidc_rejects_non_id_token_claim_shapes(monkeypatch, payload_overrides):
import jwt

previous = _snapshot_oidc_settings()
_oidc_test_settings()
monkeypatch.setattr("api.auth.jwks_client", object())
monkeypatch.setattr("api.auth._cached_oidc_signing_keys", (_OidcMockKey(),))
monkeypatch.setattr(
jwt, "decode", lambda *a, **k: _oidc_id_token_payload(**payload_overrides)
)

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:
_restore_oidc_settings(previous)

assert exc.value.status_code == 401


@pytest.mark.asyncio
async def test_oidc_accepts_id_token_with_matching_azp_and_token_use(monkeypatch):
import jwt

previous = _snapshot_oidc_settings()
_oidc_test_settings()
monkeypatch.setattr("api.auth.jwks_client", object())
monkeypatch.setattr("api.auth._cached_oidc_signing_keys", (_OidcMockKey(),))
monkeypatch.setattr(
jwt,
"decode",
lambda *a, **k: _oidc_id_token_payload(token_use="id", azp="naruon-api"),
)

# No typ header: IdPs that omit typ still mint valid ID tokens.
token = _signed_session_token(
_valid_session_payload(),
header={"alg": "RS256", "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
Expand Down
Loading