Loading src/portal_service/adapters/keycloak/_http.py +1 −1 Changes for src/portal_service/adapters/keycloak/_http.py: 1 added line, 1 removed line. Original line number Diff line number Diff line Loading @@ -29,7 +29,7 @@ async def send( timeout: float, transport: httpx.AsyncBaseTransport | None, data: dict[str, str] | None = None, json: dict[str, Any] | None = None, json: dict[str, Any] | list[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.""" Loading src/portal_service/adapters/keycloak/product_roles.py 0 → 100644 +87 −0 Changes for src/portal_service/adapters/keycloak/product_roles.py: 87 added lines, 0 removed lines. Original line number Diff line number Diff line """ Product roles on API clients' service accounts, through Keycloak's Admin API. The provisioner may map only the product roles, and read only the `oeg` client (to get the roles' ids), so the `oeg` client's id has to be known: it cannot be searched for. """ from collections.abc import Awaitable, Callable from typing import Any import httpx from pydantic import BaseModel, RootModel from portal_service.adapters.keycloak import _http from portal_service.domain.errors import ServiceAccountNotFoundError from portal_service.domain.ports.product_roles import ProductRoleAssigner _WHAT = "Keycloak admin API" class _User(BaseModel): id: str class _Role(BaseModel): id: str name: str class KeycloakProductRoles(ProductRoleAssigner): def __init__( self, *, admin_url: str, oeg_client_uuid: str, provisioner_token: Callable[[], Awaitable[str]], timeout: float = 10.0, transport: httpx.AsyncBaseTransport | None = None, ) -> None: self._admin_url = admin_url self._oeg_client_uuid = oeg_client_uuid self._provisioner_token = provisioner_token self._timeout = timeout self._transport = transport async def grant(self, keycloak_client_id: str, product: str) -> None: await self._map("POST", keycloak_client_id, product) async def revoke(self, keycloak_client_id: str, product: str) -> None: await self._map("DELETE", keycloak_client_id, product) async def _map(self, method: str, keycloak_client_id: str, product: str) -> None: user_id = await self._service_account_id(keycloak_client_id) role = await self._product_role(product) mappings = f"/users/{user_id}/role-mappings/clients/{self._oeg_client_uuid}" await self._call(method, mappings, json=[role.model_dump()]) async def _service_account_id(self, keycloak_client_id: str) -> str: username = f"service-account-{keycloak_client_id}" response = await self._call("GET", f"/users?username={username}&exact=true") users = _http.parse(RootModel[list[_User]], response, what=_WHAT).root if not users: raise ServiceAccountNotFoundError(f"No service account for {keycloak_client_id}") return users[0].id async def _product_role(self, product: str) -> _Role: response = await self._call("GET", f"/clients/{self._oeg_client_uuid}/roles") roles = _http.parse(RootModel[list[_Role]], response, what=_WHAT).root for role in roles: if role.name == product: return role raise ValueError(f"No product role named {product!r} on the oeg client") async def _call( self, method: str, path: str, json: list[dict[str, Any]] | None = None ) -> httpx.Response: response = await _http.send( method, f"{self._admin_url}{path}", what=_WHAT, timeout=self._timeout, transport=self._transport, json=json, headers={"Authorization": f"Bearer {await self._provisioner_token()}"}, ) _http.ensure_ok(response, what=_WHAT) return response src/portal_service/domain/errors.py +5 −0 Changes for src/portal_service/domain/errors.py: 5 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -44,3 +44,8 @@ class IdentityProviderUnavailableError(Exception): # the console, or its token was replaced. Keycloak does not say which. class RegistrationTokenRejectedError(Exception): pass # The client has no service-account user: it was never registered, or it was deleted. class ServiceAccountNotFoundError(Exception): pass src/portal_service/domain/ports/product_roles.py 0 → 100644 +18 −0 Changes for src/portal_service/domain/ports/product_roles.py: 18 added lines, 0 removed lines. Original line number Diff line number Diff line """Product role port: giving an API client's service account the roles of its API products.""" from abc import ABC, abstractmethod class ProductRoleAssigner(ABC): """ Keycloak puts a product's scopes in a token only if its holder has the product's role, so an API client needs the role as well as the scopes. """ @abstractmethod async def grant(self, keycloak_client_id: str, product: str) -> None: pass @abstractmethod async def revoke(self, keycloak_client_id: str, product: str) -> None: pass tests/unit/test_product_roles.py 0 → 100644 +116 −0 Changes for tests/unit/test_product_roles.py: 116 added lines, 0 removed lines. Original line number Diff line number Diff line """Product roles on service accounts, against a fake Keycloak admin API.""" import json from urllib.parse import parse_qs import httpx import pytest from portal_service.adapters.keycloak.product_roles import KeycloakProductRoles from portal_service.domain.errors import ( IdentityProviderUnavailableError, ServiceAccountNotFoundError, ) ADMIN = "http://keycloak.test/admin/realms/oop" OEG = "oeg-uuid" ROLES = [ {"id": "role-qod", "name": "quality-on-demand"}, {"id": "role-eam", "name": "edge-application-management"}, ] class FakeAdminApi: """One app with a service account, and the product roles mapped to it.""" def __init__(self) -> None: self.requests: list[httpx.Request] = [] self.mapped: set[str] = set() def __call__(self, request: httpx.Request) -> httpx.Response: self.requests.append(request) if request.headers["Authorization"] != "Bearer provisioner-token": return httpx.Response(401) path = request.url.path.removeprefix("/admin/realms/oop") if path == "/users": wanted = parse_qs(request.url.query.decode())["username"][0] found = ( [{"id": "sa-id", "username": wanted}] if wanted == "service-account-api-1" else [] ) return httpx.Response(200, json=found) if path == f"/clients/{OEG}/roles": return httpx.Response(200, json=ROLES) if path == f"/users/sa-id/role-mappings/clients/{OEG}": names = {role["name"] for role in json.loads(request.content)} self.mapped = self.mapped | names if request.method == "POST" else self.mapped - names return httpx.Response(204) return httpx.Response(404) async def _token() -> str: return "provisioner-token" @pytest.fixture def admin() -> FakeAdminApi: return FakeAdminApi() @pytest.fixture def roles(admin: FakeAdminApi) -> KeycloakProductRoles: return KeycloakProductRoles( admin_url=ADMIN, oeg_client_uuid=OEG, provisioner_token=_token, transport=httpx.MockTransport(admin), ) async def test_grant_maps_the_product_role_to_the_service_account( roles: KeycloakProductRoles, admin: FakeAdminApi ) -> None: await roles.grant("api-1", "quality-on-demand") assert admin.mapped == {"quality-on-demand"} sent = json.loads(admin.requests[-1].content) assert sent == [{"id": "role-qod", "name": "quality-on-demand"}] assert "username=service-account-api-1&exact=true" in str(admin.requests[0].url) async def test_revoke_removes_only_that_product_role( roles: KeycloakProductRoles, admin: FakeAdminApi ) -> None: await roles.grant("api-1", "quality-on-demand") await roles.grant("api-1", "edge-application-management") await roles.revoke("api-1", "quality-on-demand") assert admin.mapped == {"edge-application-management"} async def test_an_unknown_product_is_refused_before_anything_is_mapped( roles: KeycloakProductRoles, admin: FakeAdminApi ) -> None: with pytest.raises(ValueError, match="number-verification"): await roles.grant("api-1", "number-verification") assert admin.mapped == set() async def test_a_client_without_a_service_account_is_reported( roles: KeycloakProductRoles, ) -> None: with pytest.raises(ServiceAccountNotFoundError, match="api-gone"): await roles.grant("api-gone", "quality-on-demand") async def test_a_refused_request_is_unavailable() -> None: forbidden = KeycloakProductRoles( admin_url=ADMIN, oeg_client_uuid=OEG, provisioner_token=_token, transport=httpx.MockTransport(lambda request: httpx.Response(403)), ) with pytest.raises(IdentityProviderUnavailableError): await forbidden.grant("api-1", "quality-on-demand") Loading
src/portal_service/adapters/keycloak/_http.py +1 −1 Changes for src/portal_service/adapters/keycloak/_http.py: 1 added line, 1 removed line. Original line number Diff line number Diff line Loading @@ -29,7 +29,7 @@ async def send( timeout: float, transport: httpx.AsyncBaseTransport | None, data: dict[str, str] | None = None, json: dict[str, Any] | None = None, json: dict[str, Any] | list[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.""" Loading
src/portal_service/adapters/keycloak/product_roles.py 0 → 100644 +87 −0 Changes for src/portal_service/adapters/keycloak/product_roles.py: 87 added lines, 0 removed lines. Original line number Diff line number Diff line """ Product roles on API clients' service accounts, through Keycloak's Admin API. The provisioner may map only the product roles, and read only the `oeg` client (to get the roles' ids), so the `oeg` client's id has to be known: it cannot be searched for. """ from collections.abc import Awaitable, Callable from typing import Any import httpx from pydantic import BaseModel, RootModel from portal_service.adapters.keycloak import _http from portal_service.domain.errors import ServiceAccountNotFoundError from portal_service.domain.ports.product_roles import ProductRoleAssigner _WHAT = "Keycloak admin API" class _User(BaseModel): id: str class _Role(BaseModel): id: str name: str class KeycloakProductRoles(ProductRoleAssigner): def __init__( self, *, admin_url: str, oeg_client_uuid: str, provisioner_token: Callable[[], Awaitable[str]], timeout: float = 10.0, transport: httpx.AsyncBaseTransport | None = None, ) -> None: self._admin_url = admin_url self._oeg_client_uuid = oeg_client_uuid self._provisioner_token = provisioner_token self._timeout = timeout self._transport = transport async def grant(self, keycloak_client_id: str, product: str) -> None: await self._map("POST", keycloak_client_id, product) async def revoke(self, keycloak_client_id: str, product: str) -> None: await self._map("DELETE", keycloak_client_id, product) async def _map(self, method: str, keycloak_client_id: str, product: str) -> None: user_id = await self._service_account_id(keycloak_client_id) role = await self._product_role(product) mappings = f"/users/{user_id}/role-mappings/clients/{self._oeg_client_uuid}" await self._call(method, mappings, json=[role.model_dump()]) async def _service_account_id(self, keycloak_client_id: str) -> str: username = f"service-account-{keycloak_client_id}" response = await self._call("GET", f"/users?username={username}&exact=true") users = _http.parse(RootModel[list[_User]], response, what=_WHAT).root if not users: raise ServiceAccountNotFoundError(f"No service account for {keycloak_client_id}") return users[0].id async def _product_role(self, product: str) -> _Role: response = await self._call("GET", f"/clients/{self._oeg_client_uuid}/roles") roles = _http.parse(RootModel[list[_Role]], response, what=_WHAT).root for role in roles: if role.name == product: return role raise ValueError(f"No product role named {product!r} on the oeg client") async def _call( self, method: str, path: str, json: list[dict[str, Any]] | None = None ) -> httpx.Response: response = await _http.send( method, f"{self._admin_url}{path}", what=_WHAT, timeout=self._timeout, transport=self._transport, json=json, headers={"Authorization": f"Bearer {await self._provisioner_token()}"}, ) _http.ensure_ok(response, what=_WHAT) return response
src/portal_service/domain/errors.py +5 −0 Changes for src/portal_service/domain/errors.py: 5 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -44,3 +44,8 @@ class IdentityProviderUnavailableError(Exception): # the console, or its token was replaced. Keycloak does not say which. class RegistrationTokenRejectedError(Exception): pass # The client has no service-account user: it was never registered, or it was deleted. class ServiceAccountNotFoundError(Exception): pass
src/portal_service/domain/ports/product_roles.py 0 → 100644 +18 −0 Changes for src/portal_service/domain/ports/product_roles.py: 18 added lines, 0 removed lines. Original line number Diff line number Diff line """Product role port: giving an API client's service account the roles of its API products.""" from abc import ABC, abstractmethod class ProductRoleAssigner(ABC): """ Keycloak puts a product's scopes in a token only if its holder has the product's role, so an API client needs the role as well as the scopes. """ @abstractmethod async def grant(self, keycloak_client_id: str, product: str) -> None: pass @abstractmethod async def revoke(self, keycloak_client_id: str, product: str) -> None: pass
tests/unit/test_product_roles.py 0 → 100644 +116 −0 Changes for tests/unit/test_product_roles.py: 116 added lines, 0 removed lines. Original line number Diff line number Diff line """Product roles on service accounts, against a fake Keycloak admin API.""" import json from urllib.parse import parse_qs import httpx import pytest from portal_service.adapters.keycloak.product_roles import KeycloakProductRoles from portal_service.domain.errors import ( IdentityProviderUnavailableError, ServiceAccountNotFoundError, ) ADMIN = "http://keycloak.test/admin/realms/oop" OEG = "oeg-uuid" ROLES = [ {"id": "role-qod", "name": "quality-on-demand"}, {"id": "role-eam", "name": "edge-application-management"}, ] class FakeAdminApi: """One app with a service account, and the product roles mapped to it.""" def __init__(self) -> None: self.requests: list[httpx.Request] = [] self.mapped: set[str] = set() def __call__(self, request: httpx.Request) -> httpx.Response: self.requests.append(request) if request.headers["Authorization"] != "Bearer provisioner-token": return httpx.Response(401) path = request.url.path.removeprefix("/admin/realms/oop") if path == "/users": wanted = parse_qs(request.url.query.decode())["username"][0] found = ( [{"id": "sa-id", "username": wanted}] if wanted == "service-account-api-1" else [] ) return httpx.Response(200, json=found) if path == f"/clients/{OEG}/roles": return httpx.Response(200, json=ROLES) if path == f"/users/sa-id/role-mappings/clients/{OEG}": names = {role["name"] for role in json.loads(request.content)} self.mapped = self.mapped | names if request.method == "POST" else self.mapped - names return httpx.Response(204) return httpx.Response(404) async def _token() -> str: return "provisioner-token" @pytest.fixture def admin() -> FakeAdminApi: return FakeAdminApi() @pytest.fixture def roles(admin: FakeAdminApi) -> KeycloakProductRoles: return KeycloakProductRoles( admin_url=ADMIN, oeg_client_uuid=OEG, provisioner_token=_token, transport=httpx.MockTransport(admin), ) async def test_grant_maps_the_product_role_to_the_service_account( roles: KeycloakProductRoles, admin: FakeAdminApi ) -> None: await roles.grant("api-1", "quality-on-demand") assert admin.mapped == {"quality-on-demand"} sent = json.loads(admin.requests[-1].content) assert sent == [{"id": "role-qod", "name": "quality-on-demand"}] assert "username=service-account-api-1&exact=true" in str(admin.requests[0].url) async def test_revoke_removes_only_that_product_role( roles: KeycloakProductRoles, admin: FakeAdminApi ) -> None: await roles.grant("api-1", "quality-on-demand") await roles.grant("api-1", "edge-application-management") await roles.revoke("api-1", "quality-on-demand") assert admin.mapped == {"edge-application-management"} async def test_an_unknown_product_is_refused_before_anything_is_mapped( roles: KeycloakProductRoles, admin: FakeAdminApi ) -> None: with pytest.raises(ValueError, match="number-verification"): await roles.grant("api-1", "number-verification") assert admin.mapped == set() async def test_a_client_without_a_service_account_is_reported( roles: KeycloakProductRoles, ) -> None: with pytest.raises(ServiceAccountNotFoundError, match="api-gone"): await roles.grant("api-gone", "quality-on-demand") async def test_a_refused_request_is_unavailable() -> None: forbidden = KeycloakProductRoles( admin_url=ADMIN, oeg_client_uuid=OEG, provisioner_token=_token, transport=httpx.MockTransport(lambda request: httpx.Response(403)), ) with pytest.raises(IdentityProviderUnavailableError): await forbidden.grant("api-1", "quality-on-demand")
