refactor: move token verification into the Keycloak adapter (5d4c9cec) · Commits · OOP / code / portal

src/portal_service/adapters/keycloak/token_verifier.py

0 → 100644

+113 −0

Changes for src/portal_service/adapters/keycloak/token_verifier.py: 113 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""
		Checking the tokens Keycloak signs: signature against the realm's published keys,
		then issuer, audience, expiry and Keycloak's own `typ`.

		Raises domain errors only, so it knows nothing of HTTP; callers translate them.
		"""

		from hmac import compare_digest
		from typing import Any, Protocol

		import jwt
		import structlog
		from jwt import PyJWKClient
		from pydantic import ValidationError

		from portal_service.config import OIDCSettings
		from portal_service.domain.errors import (
		IdentityProviderUnavailableError,
		InvalidTokenError,
		TokenExpiredError,
		)
		from portal_service.domain.models import TokenClaims

		logger: structlog.BoundLogger = structlog.get_logger(__name__)


		class SigningKeySource(Protocol):
		"""Finds the public key a token was signed with. PyJWKClient in production."""

		def get_signing_key_from_jwt(self, token: str) -> Any: ...


		# One key-set client per realm, shared across requests: a verifier is built per
		# request, and PyJWKClient's own cache only helps if the client outlives it.
		_jwks_clients: dict[str, PyJWKClient] = {}


		def _jwks_client_for(jwks_uri: str) -> PyJWKClient:
		if jwks_uri not in _jwks_clients:
		_jwks_clients[jwks_uri] = PyJWKClient(jwks_uri, cache_jwk_set=True, lifespan=300)
		return _jwks_clients[jwks_uri]


		class KeycloakTokenVerifier:
		def __init__(self, settings: OIDCSettings, keys: SigningKeySource \| None = None) -> None:
		self._settings = settings
		# Tests pass a fake key source here; in production it is the realm's JWKS.
		self._keys = keys if keys is not None else _jwks_client_for(settings.jwks_uri)

		def verify_access_token(self, token: str) -> TokenClaims:
		return self._decode(token, expected_typ="Bearer")

		def verify_id_token(self, 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 = self._decode(token, expected_typ="ID")
		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

		def _decode(self, token: str, expected_typ: str) -> TokenClaims:
		"""
		Decode and verify a JWT, returning its claims.

		Raises domain errors (no HTTP knowledge here):
		TokenExpiredError — signature valid but token is past exp
		InvalidTokenError — anything else that makes the token unacceptable
		IdentityProviderUnavailableError — signing keys unreachable; not the caller's fault

		`expected_typ` is Keycloak's own `typ` claim: "Bearer" for access tokens, "ID"
		for ID tokens. Both share issuer, audience and signing key, so without this an
		ID token would pass as an access token.
		"""
		try:
		signing_key = self._keys.get_signing_key_from_jwt(token)
		payload: dict[str, Any] = jwt.decode(
		token,
		signing_key.key,
		algorithms=["RS256"],
		issuer=self._settings.issuer,
		# 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=self._settings.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.
		raise IdentityProviderUnavailableError("Could not fetch signing keys") from exc
		except jwt.PyJWTError as exc:
		raise InvalidTokenError(f"Token validation failed: {exc}") from exc
		if payload.get("typ") != expected_typ:
		raise InvalidTokenError(f"Expected a {expected_typ} token")
		try:
		return TokenClaims(**payload)
		except ValidationError as exc:
		# 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

src/portal_service/api/auth/init.py

+1 −16

Changes for src/portal_service/api/auth/__init__.py: 1 added line, 16 removed lines.
Original line number	Diff line number	Diff line
		"""
		Authentication for the portal's own login flow.

		Organised into focused modules:
		* verification — JWT/JWKS decode
		* routes — login, callback, /me and logout
		* schemas — what the browser is told about the user

		Public names are re-exported here so callers keep importing from
		`portal_service.api.auth`.
		Token checking is Keycloak work and lives in `adapters/keycloak/token_verifier`.
		"""

		from portal_service.api.auth.verification import (
		_decode_id_token,
		_decode_token,
		_get_jwks_client,
		)

		__all__ = [
		"_decode_id_token",
		"_decode_token",
		"_get_jwks_client",
		]

src/portal_service/api/auth/routes.py

+3 −2

Changes for src/portal_service/api/auth/routes.py: 3 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -22,10 +22,10 @@ from fastapi.responses import RedirectResponse
		from starlette.concurrency import run_in_threadpool

		from portal_service.api.auth.schemas import CurrentUser, LogoutResponse
		from portal_service.api.auth.verification import _decode_id_token
		from portal_service.api.dependencies import (
		CurrentUserDep,
		SettingsDep,
		TokenVerifierDep,
		get_identity_provider,
		get_login_service,
		get_session_service,
		@@ -101,6 +101,7 @@ async def callback(
		login_service: LoginServiceDep,
		session_service: SessionServiceDep,
		identity_provider: IdentityProviderDep,
		verifier: TokenVerifierDep,
		code: str \| None = None,
		state: str \| None = None,
		error: str \| None = None,
		@@ -130,7 +131,7 @@ async def callback(
		if issued.tokens.id_token is None:
		raise InvalidTokenError("Keycloak returned no ID token")
		claims = await run_in_threadpool(
		_decode_id_token, issued.tokens.id_token, expected_nonce=attempt.nonce
		verifier.verify_id_token, issued.tokens.id_token, expected_nonce=attempt.nonce
		)

		created = await session_service.create(subject=claims.sub, issued=issued)

src/portal_service/api/auth/verification.py

deleted100644 → 0

+0 −102

Changes for src/portal_service/api/auth/verification.py: 0 added lines, 102 removed lines.
Original line number	Diff line number	Diff line
		"""
		JWT verification: JWKS retrieval and token decoding.

		Kept free of FastAPI/HTTP concerns — it raises domain errors only, so it can be
		reused outside the request path. Callers (the auth routes) translate these into
		HTTP responses.
		"""

		from hmac import compare_digest
		from typing import Any

		import jwt
		import structlog
		from jwt import PyJWKClient
		from pydantic import ValidationError

		from portal_service.config import get_settings
		from portal_service.domain.errors import (
		IdentityProviderUnavailableError,
		InvalidTokenError,
		TokenExpiredError,
		)
		from portal_service.domain.models import TokenClaims

		logger: structlog.BoundLogger = structlog.get_logger(__name__)

		# Lazily-initialised per-issuer JWKS client.
		# Module-level is fine; PyJWKClient handles its own key-set cache internally.
		_jwks_client_cache: dict[str, PyJWKClient] = {}


		def _get_jwks_client(jwks_uri: str) -> PyJWKClient:
		if jwks_uri not in _jwks_client_cache:
		_jwks_client_cache[jwks_uri] = PyJWKClient(jwks_uri, cache_jwk_set=True, lifespan=300)
		return _jwks_client_cache[jwks_uri]


		def _decode_token(token: str, expected_typ: str = "Bearer") -> TokenClaims:
		"""
		Decode and verify a JWT, returning its claims.

		Raises domain errors (no HTTP knowledge here):
		TokenExpiredError — signature valid but token is past exp
		InvalidTokenError — anything else that makes the token unacceptable
		IdentityProviderUnavailableError — signing keys unreachable; not the caller's fault

		`expected_typ` is Keycloak's own `typ` claim: "Bearer" for access tokens, "ID"
		for ID tokens. Both share issuer, audience and signing key, so without this an
		ID token would pass as an access token.
		"""
		oidc = get_settings().oidc
		if oidc is None:
		raise InvalidTokenError("OIDC settings not configured")
		try:
		signing_key = _get_jwks_client(oidc.jwks_uri).get_signing_key_from_jwt(token)
		payload: dict[str, Any] = jwt.decode(
		token,
		signing_key.key,
		algorithms=["RS256"],
		issuer=oidc.issuer,
		# 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.
		raise IdentityProviderUnavailableError("Could not fetch signing keys") from exc
		except jwt.PyJWTError as exc:
		raise InvalidTokenError(f"Token validation failed: {exc}") from exc
		if payload.get("typ") != expected_typ:
		raise InvalidTokenError(f"Expected a {expected_typ} token")
		try:
		return TokenClaims(**payload)
		except ValidationError as exc:
		# 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, expected_typ="ID")
		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/api/dependencies.py

+12 −3

Changes for src/portal_service/api/dependencies.py: 12 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		@@ -13,7 +13,7 @@ from portal_service.adapters.database.repos import (
		SqlLoginSessionRepository,
		)
		from portal_service.adapters.keycloak.oidc_client import KeycloakOIDCClient
		from portal_service.api.auth.verification import _decode_token
		from portal_service.adapters.keycloak.token_verifier import KeycloakTokenVerifier
		from portal_service.app_state import AppState
		from portal_service.application.login_service import LoginService
		from portal_service.application.session_service import ActiveSession, SessionService
		@@ -89,6 +89,15 @@ def get_identity_provider(
		return KeycloakOIDCClient(oidc)


		def get_token_verifier(
		oidc: Annotated[OIDCSettings, Depends(get_oidc_settings)],
		) -> KeycloakTokenVerifier:
		return KeycloakTokenVerifier(oidc)


		TokenVerifierDep = Annotated[KeycloakTokenVerifier, Depends(get_token_verifier)]


		def get_login_service(
		attempts: Annotated[LoginAttemptRepository, Depends(get_login_attempt_repo)],
		identity_provider: Annotated[IdentityProvider, Depends(get_identity_provider)],
		@@ -148,14 +157,14 @@ async def get_current_session(
		CurrentSessionDep = Annotated[ActiveSession, Depends(get_current_session)]


		async def get_current_user(session: CurrentSessionDep) -> TokenClaims:
		async def get_current_user(session: CurrentSessionDep, verifier: TokenVerifierDep) -> TokenClaims:
		"""
		Who the request acts for, read from the session's access token rather than
		asking Keycloak, so it costs no network call. The token was verified when it was
		issued and is verified again here, including its expiry.
		"""
		try:
		return await run_in_threadpool(_decode_token, session.tokens.access_token)
		return await run_in_threadpool(verifier.verify_access_token, session.tokens.access_token)
		except AuthenticationError as exc:
		logger.info("session token no longer usable", error=str(exc))
		raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Not authenticated") from exc