Skip to content
Merged
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
3 changes: 3 additions & 0 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,9 @@
public identity headers such as `X-User-Id`, `X-Organization-Id`,
`X-Group-Id`, `X-Group-Ids`, `X-User-Role`, or `X-Dev-Auth-Token`;
tests/mocks must exercise the signed-session path.
- JWT/session verification must reject unsupported critical headers (`crit`)
before trusting payload claims; do not rely only on library defaults for this
boundary.
- Private backend `/api/*` routers must be registered with the default
`get_auth_context` signed-session dependency; only explicitly documented
public endpoints such as `/` may omit it. Keep runtime feature/configuration
Expand Down
7 changes: 7 additions & 0 deletions backend/api/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,7 @@ def _cached_oidc_signing_key_from_jwt(token: str) -> Any:
header = jwt.get_unverified_header(token)
except Exception:
raise _authentication_error() from None
_reject_unsupported_critical_headers(header)
key_id = header.get("kid")
if not isinstance(key_id, str) or not key_id.strip():
raise _authentication_error()
Expand All @@ -136,6 +137,11 @@ def _cached_oidc_signing_key_from_jwt(token: str) -> Any:
raise _authentication_error()


def _reject_unsupported_critical_headers(header: dict[str, Any]) -> None:
if "crit" in header:
raise _authentication_error()


def _session_secret_bytes() -> bytes:
configured = settings.AUTH_SESSION_HMAC_SECRET
if configured is None:
Expand Down Expand Up @@ -216,6 +222,7 @@ def _verify_signed_session_payload(authorization: str | None) -> dict[str, Any]:
header = _json_object_from_base64url_segment(header_segment)
if header.get("alg") != SESSION_SIGNING_ALGORITHM:
raise _authentication_error()
_reject_unsupported_critical_headers(header)

secret = _session_secret_bytes()
signing_input = f"{header_segment}.{payload_segment}"
Expand Down
2 changes: 1 addition & 1 deletion backend/requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -25,4 +25,4 @@ opentelemetry-instrumentation-fastapi==0.46b0
opentelemetry-exporter-otlp==1.25.0
setuptools==78.1.1
websockets==14.1
PyJWT==2.8.0
PyJWT==2.13.0
68 changes: 68 additions & 0 deletions backend/tests/test_auth_real.py
Original file line number Diff line number Diff line change
Expand Up @@ -337,6 +337,25 @@ async def test_signed_bearer_session_rejects_rs256_algorithm():
assert exc.value.status_code == 401


@pytest.mark.asyncio
async def test_signed_bearer_session_rejects_unknown_critical_header():
settings.AUTH_SESSION_HMAC_SECRET = SecretStr(TEST_SESSION_HMAC_SECRET)
token = _signed_session_token(
_valid_session_payload(),
header={
"alg": "HS256",
"typ": "JWT",
"crit": ["x-custom-policy"],
"x-custom-policy": "require-mfa",
},
)

with pytest.raises(HTTPException) as exc:
await get_auth_context(authorization=f"Bearer {token}")

assert exc.value.status_code == 401


@pytest.mark.asyncio
async def test_signed_bearer_session_rejects_wrong_secret():
settings.AUTH_SESSION_HMAC_SECRET = SecretStr(TEST_SESSION_HMAC_SECRET)
Expand Down Expand Up @@ -718,6 +737,55 @@ def mock_jwt_decode(*args, **kwargs):
assert context.organization_id == "org-acme"


@pytest.mark.asyncio
async def test_oidc_rejects_unknown_critical_header_before_decode(monkeypatch):
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 = "http://localhost:8081/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(),))

decode_called = False

def mock_jwt_decode(*args, **kwargs):
nonlocal decode_called
decode_called = True
return {}

monkeypatch.setattr(jwt, "decode", mock_jwt_decode)
token = _signed_session_token(
_valid_session_payload(),
header={
"alg": "RS256",
"typ": "JWT",
"kid": "test-key",
"crit": ["x-custom-policy"],
"x-custom-policy": "require-mfa",
},
)

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
assert decode_called is False


@pytest.mark.asyncio
async def test_oidc_validation_failure_does_not_fallback_to_signed_session(monkeypatch):
import jwt
Expand Down
Loading