feat: implement Keycloak client registration and provisioner token management with error handling (38e70fde) · Commits · OOP / code / portal

src/portal_service/adapters/keycloak/_http.py

0 → 100644

+57 −0

Changes for src/portal_service/adapters/keycloak/_http.py: 57 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""What every Keycloak adapter does the same way: send, check the status, read the body."""

		from typing import Any

		import httpx
		import structlog
		from pydantic import BaseModel, ValidationError

		from portal_service.domain.errors import IdentityProviderUnavailableError

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


		class TokenResponse(BaseModel):
		"""The fields of Keycloak's token response that we use."""

		access_token: str
		expires_in: int
		refresh_token: str \| None = None
		refresh_expires_in: int \| None = None
		id_token: str \| None = None


		async def send(
		method: str,
		url: str,
		*,
		what: str,
		timeout: float,
		transport: httpx.AsyncBaseTransport \| None,
		data: dict[str, str] \| None = None,
		json: dict[str, Any] \| None = None,
		headers: dict[str, str] \| None = None,
		) -> httpx.Response:
		"""Any status comes back as a response; only a network failure raises."""
		try:
		async with httpx.AsyncClient(timeout=timeout, transport=transport) as client:
		return await client.request(method, url, data=data, json=json, headers=headers)
		except httpx.HTTPError as exc:
		logger.error(f"{what} unreachable", error=str(exc))
		raise IdentityProviderUnavailableError(f"{what} unreachable") from exc


		def ensure_ok(response: httpx.Response, *, what: str) -> None:
		if response.status_code >= 400:
		logger.error(f"{what} rejected", status=response.status_code)
		raise IdentityProviderUnavailableError(f"{what} returned {response.status_code}")


		def parse[M: BaseModel](model: type[M], response: httpx.Response, *, what: str) -> M:
		try:
		return model.model_validate_json(response.content)
		except ValidationError as exc:
		# Field names only: the payload may hold tokens and must not reach the logs.
		fields = ", ".join(str(err["loc"][0]) for err in exc.errors() if err["loc"])
		logger.error(f"{what} response not usable", fields=fields)
		raise IdentityProviderUnavailableError(f"{what} response invalid: {fields}") from exc

src/portal_service/adapters/keycloak/client_registration.py

0 → 100644

+174 −0

Changes for src/portal_service/adapters/keycloak/client_registration.py: 174 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""
		Keycloak's client registration service, over httpx.

		Registering a client needs the provisioner's token. Everything after that uses the client's
		own registration token, so the provisioner never has rights over other clients. An update
		reads the client first and changes only its own field: sending a fresh representation would
		silently reset the rest.
		"""

		from collections.abc import Awaitable, Callable
		from typing import Any

		import httpx
		from pydantic import BaseModel, Field

		from portal_service.adapters.keycloak import _http
		from portal_service.domain.errors import (
		IdentityProviderUnavailableError,
		RegistrationTokenRejectedError,
		)
		from portal_service.domain.ports.client_registry import ClientRegistry, ClientSpec, RegisteredClient

		_WHAT = "Client registration"
		_OEG_AUDIENCE = "oeg"
		# `basic` and `roles` put `sub` and the realm roles in the token; Keycloak adds `service_account`.
		_BASE_SCOPES = frozenset({"basic", "roles"})
		_INFRASTRUCTURE_SCOPES = _BASE_SCOPES \| {"service_account"}
		_MIN_SECRET_LENGTH = 32


		class _Registration(BaseModel):
		secret: str
		registration_token: str = Field(alias="registrationAccessToken")


		class KeycloakClientRegistry(ClientRegistry):
		def __init__(
		self,
		*,
		registration_url: str,
		provisioner_token: Callable[[], Awaitable[str]],
		timeout: float = 10.0,
		transport: httpx.AsyncBaseTransport \| None = None,
		) -> None:
		self._url = registration_url
		self._provisioner_token = provisioner_token
		self._timeout = timeout
		self._transport = transport

		async def register(self, spec: ClientSpec) -> RegisteredClient:
		body = {
		"clientId": spec.keycloak_client_id,
		"name": spec.name,
		"description": spec.description,
		"publicClient": False,
		"serviceAccountsEnabled": True,
		"standardFlowEnabled": False,
		"implicitFlowEnabled": False,
		"directAccessGrantsEnabled": False,
		"redirectUris": [],
		"defaultClientScopes": sorted(_BASE_SCOPES \| spec.scopes),
		"optionalClientScopes": [],
		"protocolMappers": [
		_audience_mapper(),
		_claim_mapper("organization", "organization", spec.organization),
		_claim_mapper("app-provider-id", "app_provider_id", spec.app_provider_id),
		],
		}
		response = await self._request("POST", self._url, await self._provisioner_token(), body)
		_http.ensure_ok(response, what=_WHAT)
		registered = _registration(response)
		return RegisteredClient(
		secret=registered.secret, registration_token=registered.registration_token
		)

		async def read_scopes(self, keycloak_client_id: str, registration_token: str) -> set[str]:
		current = await self._read(keycloak_client_id, registration_token)
		return set(current["defaultClientScopes"]) - _INFRASTRUCTURE_SCOPES

		async def set_scopes(
		self, keycloak_client_id: str, registration_token: str, scopes: frozenset[str]
		) -> str:
		current = await self._read(keycloak_client_id, registration_token)
		current["defaultClientScopes"] = sorted(_BASE_SCOPES \| scopes)
		return await self._update(keycloak_client_id, registration_token, current)

		async def set_secret(
		self, keycloak_client_id: str, registration_token: str, secret: str
		) -> str:
		# Keycloak accepts an empty secret, and then lets the client in without one.
		if len(secret) < _MIN_SECRET_LENGTH:
		raise ValueError(f"client secrets must be at least {_MIN_SECRET_LENGTH} characters")
		current = await self._read(keycloak_client_id, registration_token)
		current["secret"] = secret
		return await self._update(keycloak_client_id, registration_token, current)

		async def delete(self, keycloak_client_id: str, registration_token: str) -> None:
		await self._own("DELETE", keycloak_client_id, registration_token)

		async def _read(self, keycloak_client_id: str, registration_token: str) -> dict[str, Any]:
		return _json(await self._own("GET", keycloak_client_id, registration_token))

		async def _update(
		self, keycloak_client_id: str, registration_token: str, current: dict[str, Any]
		) -> str:
		current.pop("registrationAccessToken", None)
		response = await self._own("PUT", keycloak_client_id, registration_token, current)
		return _registration(response).registration_token

		async def _own(
		self,
		method: str,
		keycloak_client_id: str,
		registration_token: str,
		json: dict[str, Any] \| None = None,
		) -> httpx.Response:
		"""A call made with the client's own token, which Keycloak rejects with 401."""
		response = await self._request(
		method, f"{self._url}/{keycloak_client_id}", registration_token, json
		)
		if response.status_code == 401:
		raise RegistrationTokenRejectedError("Keycloak rejected the registration token")
		_http.ensure_ok(response, what=_WHAT)
		return response

		async def _request(
		self, method: str, url: str, token: str, json: dict[str, Any] \| None
		) -> httpx.Response:
		return await _http.send(
		method,
		url,
		what=_WHAT,
		timeout=self._timeout,
		transport=self._transport,
		json=json,
		headers={"Authorization": f"Bearer {token}"},
		)


		def _audience_mapper() -> dict[str, Any]:
		return {
		"name": "audience-oeg",
		"protocol": "openid-connect",
		"protocolMapper": "oidc-audience-mapper",
		"config": {"included.custom.audience": _OEG_AUDIENCE, "access.token.claim": "true"},
		}


		def _claim_mapper(name: str, claim: str, value: str) -> dict[str, Any]:
		return {
		"name": name,
		"protocol": "openid-connect",
		"protocolMapper": "oidc-hardcoded-claim-mapper",
		"config": {
		"claim.name": claim,
		"claim.value": value,
		"jsonType.label": "String",
		"access.token.claim": "true",
		},
		}


		def _registration(response: httpx.Response) -> _Registration:
		return _http.parse(_Registration, response, what=_WHAT)


		def _json(response: httpx.Response) -> dict[str, Any]:
		try:
		parsed = response.json()
		except ValueError as exc:
		raise IdentityProviderUnavailableError(f"{_WHAT} response invalid") from exc
		if not isinstance(parsed, dict):
		raise IdentityProviderUnavailableError(f"{_WHAT} response invalid")
		return parsed

src/portal_service/adapters/keycloak/oidc_client.py

+7 −26

Changes for src/portal_service/adapters/keycloak/oidc_client.py: 7 added lines, 26 removed lines.
Original line number	Diff line number	Diff line
		@@ -11,8 +11,8 @@ from urllib.parse import urlencode

		import httpx
		import structlog
		from pydantic import BaseModel, ValidationError

		from portal_service.adapters.keycloak import _http
		from portal_service.config import OIDCSettings
		from portal_service.domain.errors import (
		IdentityProviderUnavailableError,
		@@ -26,16 +26,6 @@ logger: structlog.BoundLogger = structlog.get_logger(__name__)
		_SCOPE = "openid profile email"


		class _TokenResponse(BaseModel):
		"""The fields of Keycloak's token response that we use."""

		access_token: str
		expires_in: int
		refresh_token: str \| None = None
		refresh_expires_in: int \| None = None
		id_token: str \| None = None


		class KeycloakOIDCClient(IdentityProvider):
		def __init__(
		self,
		@@ -88,18 +78,15 @@ class KeycloakOIDCClient(IdentityProvider):
		"client_id": self._settings.client_id,
		"client_secret": self._settings.client_secret,
		}
		try:
		async with httpx.AsyncClient(
		timeout=self._settings.timeout, transport=self._transport
		) as client:
		response = await client.post(
		response = await _http.send(
		"POST",
		self._settings.token_endpoint,
		what="Token endpoint",
		timeout=self._settings.timeout,
		transport=self._transport,
		data=data,
		headers={"Accept": "application/json"},
		)
		except httpx.HTTPError as exc:
		logger.error("token endpoint unreachable", grant=form["grant_type"], error=str(exc))
		raise IdentityProviderUnavailableError("Token endpoint unreachable") from exc

		if response.status_code >= 400:
		self._raise_for_error(response, form["grant_type"])
		@@ -123,13 +110,7 @@ class KeycloakOIDCClient(IdentityProvider):
		)

		def _to_issued_tokens(self, response: httpx.Response) -> IssuedTokens:
		try:
		parsed = _TokenResponse.model_validate_json(response.content)
		except ValidationError as exc:
		# Field names only: the payload holds tokens and must not reach the logs.
		fields = ", ".join(str(err["loc"][0]) for err in exc.errors() if err["loc"])
		logger.error("token response not usable", fields=fields)
		raise IdentityProviderUnavailableError(f"Token response invalid: {fields}") from exc
		parsed = _http.parse(_http.TokenResponse, response, what="Token")
		return IssuedTokens(
		tokens=SessionTokens(
		access_token=parsed.access_token,

src/portal_service/adapters/keycloak/provisioner_token.py

0 → 100644

+52 −0

Changes for src/portal_service/adapters/keycloak/provisioner_token.py: 52 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""The provisioner's own access token: client credentials, reused until close to expiry."""

		import time
		from collections.abc import Callable

		import httpx

		from portal_service.adapters.keycloak import _http

		_RENEW_BEFORE_EXPIRY_SECONDS = 30
		_WHAT = "Provisioner token request"


		class ProvisionerToken:
		def __init__(
		self,
		*,
		token_endpoint: str,
		client_id: str,
		client_secret: str,
		timeout: float = 10.0,
		transport: httpx.AsyncBaseTransport \| None = None,
		clock: Callable[[], float] = time.monotonic,
		) -> None:
		self._token_endpoint = token_endpoint
		self._client_id = client_id
		self._client_secret = client_secret
		self._timeout = timeout
		self._transport = transport
		self._clock = clock
		self._token: str \| None = None
		self._renew_at = 0.0

		async def get(self) -> str:
		if self._token is None or self._clock() >= self._renew_at:
		response = await _http.send(
		"POST",
		self._token_endpoint,
		what=_WHAT,
		timeout=self._timeout,
		transport=self._transport,
		data={
		"grant_type": "client_credentials",
		"client_id": self._client_id,
		"client_secret": self._client_secret,
		},
		)
		_http.ensure_ok(response, what=_WHAT)
		parsed = _http.parse(_http.TokenResponse, response, what=_WHAT)
		self._token = parsed.access_token
		self._renew_at = self._clock() + parsed.expires_in - _RENEW_BEFORE_EXPIRY_SECONDS
		return self._token

src/portal_service/domain/errors.py

+6 −0

Changes for src/portal_service/domain/errors.py: 6 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -38,3 +38,9 @@ class InvalidGrantError(AuthenticationError):
		# Not an AuthenticationError: the token was never judged, so it is not the caller's fault.
		class IdentityProviderUnavailableError(Exception):
		pass


		# Keycloak refused a client's registration access token: the client was deleted, disabled in
		# the console, or its token was replaced. Keycloak does not say which.
		class RegistrationTokenRejectedError(Exception):
		pass