Loading src/portal_service/api/auth/__init__.py +6 −1 Changes for src/portal_service/api/auth/__init__.py: 6 added lines, 1 removed line. Original line number Diff line number Diff line Loading @@ -19,7 +19,11 @@ from portal_service.api.auth.guards import ( require_any_role, require_scopes, ) from portal_service.api.auth.verification import _decode_token, _get_jwks_client from portal_service.api.auth.verification import ( _decode_id_token, _decode_token, _get_jwks_client, ) __all__ = [ "BearerDep", Loading @@ -27,6 +31,7 @@ __all__ = [ "get_current_user", "require_any_role", "require_scopes", "_decode_id_token", "_decode_token", "_get_jwks_client", ] src/portal_service/api/auth/verification.py +24 −5 Changes for src/portal_service/api/auth/verification.py: 24 added lines, 5 removed lines. Original line number Diff line number Diff line Loading @@ -6,6 +6,7 @@ reused outside the request path. Callers (the dependencies module) translate these into HTTP responses. """ from hmac import compare_digest from typing import Any import jwt Loading Loading @@ -53,16 +54,19 @@ def _decode_token(token: str) -> TokenClaims: signing_key.key, algorithms=["RS256"], issuer=oidc.issuer, # TODO: audience is not verified, so a token issued to *any* client in the realm # is accepted here, realm roles included. Keycloak leaves our client_id out of # `aud` by default. Fix when the realm is set up: add an Audience mapper to the # portal client, then replace this option with `audience=oidc.client_id`. options={"verify_aud": False}, # The realm's audience mapper puts our client_id in `aud`, so a token minted # for another client in the same realm is refused here. audience=oidc.client_id, ) except jwt.ExpiredSignatureError as exc: raise TokenExpiredError("Token has expired") from exc except jwt.InvalidIssuerError as exc: raise InvalidTokenError("Invalid token issuer") from exc except jwt.InvalidAudienceError as exc: raise InvalidTokenError("Token was not issued for this client") from exc except jwt.MissingRequiredClaimError as exc: # No `aud` at all: the realm's audience mapper is missing or the token is not ours. raise InvalidTokenError(f"Token is missing a required claim: {exc.claim}") from exc except jwt.PyJWKClientConnectionError as exc: # Must precede PyJWTError (its ancestor). Only the connection failure maps # here — an unknown `kid` raises the parent PyJWKClientError and stays 401. Loading @@ -75,3 +79,18 @@ def _decode_token(token: str) -> TokenClaims: # Field names only: str(exc) embeds the full payload (PII) and gets logged. fields = ", ".join(str(err["loc"][0]) for err in exc.errors() if err["loc"]) raise InvalidTokenError(f"Token claims invalid: {fields}") from exc def _decode_id_token(token: str, expected_nonce: str) -> TokenClaims: """ Decode the ID token from a login and check it belongs to this login. The nonce was generated for one login redirect and stored server-side. Keycloak copies it into the ID token, so a token replayed from another login carries a different one and is refused. """ claims = _decode_token(token) if not claims.nonce or not compare_digest(claims.nonce, expected_nonce): logger.warning("id token nonce mismatch", subject=claims.sub) raise InvalidTokenError("ID token does not belong to this login") return claims src/portal_service/domain/models.py +1 −0 Changes for src/portal_service/domain/models.py: 1 added line, 0 removed lines. Original line number Diff line number Diff line Loading @@ -27,6 +27,7 @@ class TokenClaims(BaseModel): preferred_username: str | None = None email: str | None = None scope: str = "" nonce: str | None = None realm_access: dict[str, Any] | None = None resource_access: dict[str, Any] | None = None Loading tests/conftest.py +3 −0 Changes for tests/conftest.py: 3 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -73,6 +73,7 @@ def make_token(rsa_key: rsa.RSAPrivateKey) -> Callable[..., str]: email: str | None = "user@example.com", expires_in: int = 300, issuer: str = ISSUER, audience: str | list[str] | None = CLIENT_ID, preferred_username: str | None = "user", **extra: Any, ) -> str: Loading @@ -86,6 +87,8 @@ def make_token(rsa_key: rsa.RSAPrivateKey) -> Callable[..., str]: "preferred_username": preferred_username, **extra, } if audience is not None: payload["aud"] = audience if email is not None: payload["email"] = email if roles is not None: Loading tests/test_id_token.py 0 → 100644 +74 −0 Changes for tests/test_id_token.py: 74 added lines, 0 removed lines. Original line number Diff line number Diff line """ Audience and nonce checks. Both answer "is this token meant for us, right now": the audience rejects a token minted for another client in the same realm, the nonce rejects one replayed from a different login. Tokens here are really signed and really verified. """ from collections.abc import Callable import pytest from fastapi import FastAPI from portal_service.api.auth.verification import _decode_id_token, _decode_token from portal_service.domain.errors import InvalidTokenError CLIENT_ID = "oop-portal" # ------------------------------------------------------------------------ audience def test_token_for_this_client_is_accepted( app_jwt: FastAPI, patch_jwks: None, make_token: Callable[..., str] ) -> None: assert _decode_token(make_token()).aud == CLIENT_ID def test_token_for_another_client_is_refused( app_jwt: FastAPI, patch_jwks: None, make_token: Callable[..., str] ) -> None: # Same realm, same signing key, different client: roles must not carry over. with pytest.raises(InvalidTokenError, match="not issued for this client"): _decode_token(make_token(audience="oeg")) def test_token_without_an_audience_is_refused( app_jwt: FastAPI, patch_jwks: None, make_token: Callable[..., str] ) -> None: # Would happen if the realm's audience mapper were removed. with pytest.raises(InvalidTokenError, match="missing a required claim"): _decode_token(make_token(audience=None)) def test_audience_may_list_several_clients( app_jwt: FastAPI, patch_jwks: None, make_token: Callable[..., str] ) -> None: # The realm adds both `oop-portal` and `oeg`, which is what real tokens look like. claims = _decode_token(make_token(audience=[CLIENT_ID, "oeg"])) assert claims.aud == [CLIENT_ID, "oeg"] # --------------------------------------------------------------------------- nonce def test_id_token_with_the_expected_nonce_is_accepted( app_jwt: FastAPI, patch_jwks: None, make_token: Callable[..., str] ) -> None: claims = _decode_id_token(make_token(nonce="the-nonce"), expected_nonce="the-nonce") assert claims.nonce == "the-nonce" def test_id_token_from_another_login_is_refused( app_jwt: FastAPI, patch_jwks: None, make_token: Callable[..., str] ) -> None: with pytest.raises(InvalidTokenError, match="does not belong to this login"): _decode_id_token(make_token(nonce="another-login"), expected_nonce="the-nonce") def test_id_token_without_a_nonce_is_refused( app_jwt: FastAPI, patch_jwks: None, make_token: Callable[..., str] ) -> None: with pytest.raises(InvalidTokenError, match="does not belong to this login"): _decode_id_token(make_token(), expected_nonce="the-nonce") Loading
src/portal_service/api/auth/__init__.py +6 −1 Changes for src/portal_service/api/auth/__init__.py: 6 added lines, 1 removed line. Original line number Diff line number Diff line Loading @@ -19,7 +19,11 @@ from portal_service.api.auth.guards import ( require_any_role, require_scopes, ) from portal_service.api.auth.verification import _decode_token, _get_jwks_client from portal_service.api.auth.verification import ( _decode_id_token, _decode_token, _get_jwks_client, ) __all__ = [ "BearerDep", Loading @@ -27,6 +31,7 @@ __all__ = [ "get_current_user", "require_any_role", "require_scopes", "_decode_id_token", "_decode_token", "_get_jwks_client", ]
src/portal_service/api/auth/verification.py +24 −5 Changes for src/portal_service/api/auth/verification.py: 24 added lines, 5 removed lines. Original line number Diff line number Diff line Loading @@ -6,6 +6,7 @@ reused outside the request path. Callers (the dependencies module) translate these into HTTP responses. """ from hmac import compare_digest from typing import Any import jwt Loading Loading @@ -53,16 +54,19 @@ def _decode_token(token: str) -> TokenClaims: signing_key.key, algorithms=["RS256"], issuer=oidc.issuer, # TODO: audience is not verified, so a token issued to *any* client in the realm # is accepted here, realm roles included. Keycloak leaves our client_id out of # `aud` by default. Fix when the realm is set up: add an Audience mapper to the # portal client, then replace this option with `audience=oidc.client_id`. options={"verify_aud": False}, # The realm's audience mapper puts our client_id in `aud`, so a token minted # for another client in the same realm is refused here. audience=oidc.client_id, ) except jwt.ExpiredSignatureError as exc: raise TokenExpiredError("Token has expired") from exc except jwt.InvalidIssuerError as exc: raise InvalidTokenError("Invalid token issuer") from exc except jwt.InvalidAudienceError as exc: raise InvalidTokenError("Token was not issued for this client") from exc except jwt.MissingRequiredClaimError as exc: # No `aud` at all: the realm's audience mapper is missing or the token is not ours. raise InvalidTokenError(f"Token is missing a required claim: {exc.claim}") from exc except jwt.PyJWKClientConnectionError as exc: # Must precede PyJWTError (its ancestor). Only the connection failure maps # here — an unknown `kid` raises the parent PyJWKClientError and stays 401. Loading @@ -75,3 +79,18 @@ def _decode_token(token: str) -> TokenClaims: # Field names only: str(exc) embeds the full payload (PII) and gets logged. fields = ", ".join(str(err["loc"][0]) for err in exc.errors() if err["loc"]) raise InvalidTokenError(f"Token claims invalid: {fields}") from exc def _decode_id_token(token: str, expected_nonce: str) -> TokenClaims: """ Decode the ID token from a login and check it belongs to this login. The nonce was generated for one login redirect and stored server-side. Keycloak copies it into the ID token, so a token replayed from another login carries a different one and is refused. """ claims = _decode_token(token) if not claims.nonce or not compare_digest(claims.nonce, expected_nonce): logger.warning("id token nonce mismatch", subject=claims.sub) raise InvalidTokenError("ID token does not belong to this login") return claims
src/portal_service/domain/models.py +1 −0 Changes for src/portal_service/domain/models.py: 1 added line, 0 removed lines. Original line number Diff line number Diff line Loading @@ -27,6 +27,7 @@ class TokenClaims(BaseModel): preferred_username: str | None = None email: str | None = None scope: str = "" nonce: str | None = None realm_access: dict[str, Any] | None = None resource_access: dict[str, Any] | None = None Loading
tests/conftest.py +3 −0 Changes for tests/conftest.py: 3 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -73,6 +73,7 @@ def make_token(rsa_key: rsa.RSAPrivateKey) -> Callable[..., str]: email: str | None = "user@example.com", expires_in: int = 300, issuer: str = ISSUER, audience: str | list[str] | None = CLIENT_ID, preferred_username: str | None = "user", **extra: Any, ) -> str: Loading @@ -86,6 +87,8 @@ def make_token(rsa_key: rsa.RSAPrivateKey) -> Callable[..., str]: "preferred_username": preferred_username, **extra, } if audience is not None: payload["aud"] = audience if email is not None: payload["email"] = email if roles is not None: Loading
tests/test_id_token.py 0 → 100644 +74 −0 Changes for tests/test_id_token.py: 74 added lines, 0 removed lines. Original line number Diff line number Diff line """ Audience and nonce checks. Both answer "is this token meant for us, right now": the audience rejects a token minted for another client in the same realm, the nonce rejects one replayed from a different login. Tokens here are really signed and really verified. """ from collections.abc import Callable import pytest from fastapi import FastAPI from portal_service.api.auth.verification import _decode_id_token, _decode_token from portal_service.domain.errors import InvalidTokenError CLIENT_ID = "oop-portal" # ------------------------------------------------------------------------ audience def test_token_for_this_client_is_accepted( app_jwt: FastAPI, patch_jwks: None, make_token: Callable[..., str] ) -> None: assert _decode_token(make_token()).aud == CLIENT_ID def test_token_for_another_client_is_refused( app_jwt: FastAPI, patch_jwks: None, make_token: Callable[..., str] ) -> None: # Same realm, same signing key, different client: roles must not carry over. with pytest.raises(InvalidTokenError, match="not issued for this client"): _decode_token(make_token(audience="oeg")) def test_token_without_an_audience_is_refused( app_jwt: FastAPI, patch_jwks: None, make_token: Callable[..., str] ) -> None: # Would happen if the realm's audience mapper were removed. with pytest.raises(InvalidTokenError, match="missing a required claim"): _decode_token(make_token(audience=None)) def test_audience_may_list_several_clients( app_jwt: FastAPI, patch_jwks: None, make_token: Callable[..., str] ) -> None: # The realm adds both `oop-portal` and `oeg`, which is what real tokens look like. claims = _decode_token(make_token(audience=[CLIENT_ID, "oeg"])) assert claims.aud == [CLIENT_ID, "oeg"] # --------------------------------------------------------------------------- nonce def test_id_token_with_the_expected_nonce_is_accepted( app_jwt: FastAPI, patch_jwks: None, make_token: Callable[..., str] ) -> None: claims = _decode_id_token(make_token(nonce="the-nonce"), expected_nonce="the-nonce") assert claims.nonce == "the-nonce" def test_id_token_from_another_login_is_refused( app_jwt: FastAPI, patch_jwks: None, make_token: Callable[..., str] ) -> None: with pytest.raises(InvalidTokenError, match="does not belong to this login"): _decode_id_token(make_token(nonce="another-login"), expected_nonce="the-nonce") def test_id_token_without_a_nonce_is_refused( app_jwt: FastAPI, patch_jwks: None, make_token: Callable[..., str] ) -> None: with pytest.raises(InvalidTokenError, match="does not belong to this login"): _decode_id_token(make_token(), expected_nonce="the-nonce")