Loading src/portal_service/adapters/keycloak/admin_api.py 0 → 100644 +53 −0 Changes for src/portal_service/adapters/keycloak/admin_api.py: 53 added lines, 0 removed lines. Original line number Diff line number Diff line """Keycloak's Admin API as the provisioner: the base URL, the bearer token and the errors.""" from collections.abc import Awaitable, Callable from typing import Any import httpx from portal_service.adapters.keycloak import _http _WHAT = "Keycloak admin API" class AdminApi: def __init__( self, *, admin_url: str, provisioner_token: Callable[[], Awaitable[str]], timeout: float = 10.0, transport: httpx.AsyncBaseTransport | None = None, ) -> None: self._admin_url = admin_url self._provisioner_token = provisioner_token self._timeout = timeout self._transport = transport async def call( self, method: str, path: str, json: list[dict[str, Any]] | None = None ) -> httpx.Response: response = await self._send(method, path, json) _http.ensure_ok(response, what=_WHAT) return response async def find(self, path: str) -> httpx.Response | None: """A GET where "not found" is an answer: None.""" response = await self._send("GET", path, None) if response.status_code == 404: return None _http.ensure_ok(response, what=_WHAT) return response async def _send( self, method: str, path: str, json: list[dict[str, Any]] | None ) -> httpx.Response: return 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()}"}, ) src/portal_service/adapters/keycloak/admin_directory.py 0 → 100644 +40 −0 Changes for src/portal_service/adapters/keycloak/admin_directory.py: 40 added lines, 0 removed lines. Original line number Diff line number Diff line """Users as Keycloak's Admin API shows them to the provisioner (view-users).""" from pydantic import BaseModel, RootModel from portal_service.adapters.keycloak import _http from portal_service.adapters.keycloak.admin_api import AdminApi from portal_service.domain.ports.identity_directory import DirectoryUser, IdentityDirectory _WHAT = "Keycloak admin API" class _User(BaseModel): enabled: bool attributes: dict[str, list[str]] = {} class _Role(BaseModel): name: str class KeycloakAdminDirectory(IdentityDirectory): def __init__(self, *, admin: AdminApi, oeg_client_uuid: str) -> None: self._admin = admin self._oeg_client_uuid = oeg_client_uuid async def get_user(self, subject: str) -> DirectoryUser | None: found = await self._admin.find(f"/users/{subject}") if found is None: return None user = _http.parse(_User, found, what=_WHAT) # The effective roles: a plain role list misses the ones a group gives. response = await self._admin.call( "GET", f"/users/{subject}/role-mappings/clients/{self._oeg_client_uuid}/composite" ) roles = _http.parse(RootModel[list[_Role]], response, what=_WHAT).root return DirectoryUser( enabled=user.enabled, organization=user.attributes.get("organization", [None])[0], products=frozenset(role.name for role in roles), ) src/portal_service/adapters/keycloak/product_roles.py +6 −35 Changes for src/portal_service/adapters/keycloak/product_roles.py: 6 added lines, 35 removed lines. Original line number Diff line number Diff line Loading @@ -5,13 +5,10 @@ The provisioner may map only the product roles, and read only the `oeg` client ( 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.adapters.keycloak.admin_api import AdminApi from portal_service.domain.errors import ServiceAccountNotFoundError from portal_service.domain.ports.product_roles import ProductRoleAssigner Loading @@ -28,20 +25,9 @@ class _Role(BaseModel): 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 def __init__(self, *, admin: AdminApi, oeg_client_uuid: str) -> None: self._admin = admin 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) Loading @@ -53,35 +39,20 @@ class KeycloakProductRoles(ProductRoleAssigner): 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()]) await self._admin.call(method, mappings, [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") response = await self._admin.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") response = await self._admin.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/ports/identity_directory.py 0 → 100644 +18 −0 Changes for src/portal_service/domain/ports/identity_directory.py: 18 added lines, 0 removed lines. Original line number Diff line number Diff line """Identity directory port: what Keycloak knows about a user, read-only.""" from abc import ABC, abstractmethod from pydantic import BaseModel class DirectoryUser(BaseModel): enabled: bool organization: str | None # Names of the product roles the user holds, including roles given through groups. products: frozenset[str] class IdentityDirectory(ABC): @abstractmethod async def get_user(self, subject: str) -> DirectoryUser | None: """None if there is no such user.""" tests/unit/test_admin_directory.py 0 → 100644 +66 −0 Changes for tests/unit/test_admin_directory.py: 66 added lines, 0 removed lines. Original line number Diff line number Diff line """Reading users through the Admin API, against a fake Keycloak.""" import httpx import pytest from portal_service.adapters.keycloak.admin_api import AdminApi from portal_service.adapters.keycloak.admin_directory import KeycloakAdminDirectory from portal_service.domain.errors import IdentityProviderUnavailableError ADMIN = "http://keycloak.test/admin/realms/oop" OEG = "oeg-uuid" def _directory(handler: httpx.MockTransport) -> KeycloakAdminDirectory: async def token() -> str: return "provisioner-token" admin = AdminApi(admin_url=ADMIN, provisioner_token=token, transport=handler) return KeycloakAdminDirectory(admin=admin, oeg_client_uuid=OEG) def _keycloak(user: dict[str, object] | None, roles: list[str]) -> httpx.MockTransport: def handler(request: httpx.Request) -> httpx.Response: path = request.url.path.removeprefix("/admin/realms/oop") if request.headers["Authorization"] != "Bearer provisioner-token": return httpx.Response(401) if path == "/users/maria-sub": return httpx.Response(200, json=user) if user is not None else httpx.Response(404) if path == f"/users/maria-sub/role-mappings/clients/{OEG}/composite": return httpx.Response(200, json=[{"id": name, "name": name} for name in roles]) return httpx.Response(404) return httpx.MockTransport(handler) async def test_a_user_with_an_organization_and_products() -> None: directory = _directory( _keycloak( {"enabled": True, "attributes": {"organization": ["acme"]}}, ["quality-on-demand", "edge-application-management"], ) ) user = await directory.get_user("maria-sub") assert user is not None assert (user.enabled, user.organization) == (True, "acme") assert user.products == {"quality-on-demand", "edge-application-management"} async def test_a_user_without_an_organization_or_products() -> None: user = await _directory(_keycloak({"enabled": False}, [])).get_user("maria-sub") assert user is not None assert (user.enabled, user.organization, user.products) == (False, None, frozenset()) async def test_an_unknown_user_is_none() -> None: assert await _directory(_keycloak(None, [])).get_user("maria-sub") is None async def test_a_refused_request_is_unavailable() -> None: directory = _directory(httpx.MockTransport(lambda request: httpx.Response(403))) with pytest.raises(IdentityProviderUnavailableError): await directory.get_user("maria-sub") Loading
src/portal_service/adapters/keycloak/admin_api.py 0 → 100644 +53 −0 Changes for src/portal_service/adapters/keycloak/admin_api.py: 53 added lines, 0 removed lines. Original line number Diff line number Diff line """Keycloak's Admin API as the provisioner: the base URL, the bearer token and the errors.""" from collections.abc import Awaitable, Callable from typing import Any import httpx from portal_service.adapters.keycloak import _http _WHAT = "Keycloak admin API" class AdminApi: def __init__( self, *, admin_url: str, provisioner_token: Callable[[], Awaitable[str]], timeout: float = 10.0, transport: httpx.AsyncBaseTransport | None = None, ) -> None: self._admin_url = admin_url self._provisioner_token = provisioner_token self._timeout = timeout self._transport = transport async def call( self, method: str, path: str, json: list[dict[str, Any]] | None = None ) -> httpx.Response: response = await self._send(method, path, json) _http.ensure_ok(response, what=_WHAT) return response async def find(self, path: str) -> httpx.Response | None: """A GET where "not found" is an answer: None.""" response = await self._send("GET", path, None) if response.status_code == 404: return None _http.ensure_ok(response, what=_WHAT) return response async def _send( self, method: str, path: str, json: list[dict[str, Any]] | None ) -> httpx.Response: return 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()}"}, )
src/portal_service/adapters/keycloak/admin_directory.py 0 → 100644 +40 −0 Changes for src/portal_service/adapters/keycloak/admin_directory.py: 40 added lines, 0 removed lines. Original line number Diff line number Diff line """Users as Keycloak's Admin API shows them to the provisioner (view-users).""" from pydantic import BaseModel, RootModel from portal_service.adapters.keycloak import _http from portal_service.adapters.keycloak.admin_api import AdminApi from portal_service.domain.ports.identity_directory import DirectoryUser, IdentityDirectory _WHAT = "Keycloak admin API" class _User(BaseModel): enabled: bool attributes: dict[str, list[str]] = {} class _Role(BaseModel): name: str class KeycloakAdminDirectory(IdentityDirectory): def __init__(self, *, admin: AdminApi, oeg_client_uuid: str) -> None: self._admin = admin self._oeg_client_uuid = oeg_client_uuid async def get_user(self, subject: str) -> DirectoryUser | None: found = await self._admin.find(f"/users/{subject}") if found is None: return None user = _http.parse(_User, found, what=_WHAT) # The effective roles: a plain role list misses the ones a group gives. response = await self._admin.call( "GET", f"/users/{subject}/role-mappings/clients/{self._oeg_client_uuid}/composite" ) roles = _http.parse(RootModel[list[_Role]], response, what=_WHAT).root return DirectoryUser( enabled=user.enabled, organization=user.attributes.get("organization", [None])[0], products=frozenset(role.name for role in roles), )
src/portal_service/adapters/keycloak/product_roles.py +6 −35 Changes for src/portal_service/adapters/keycloak/product_roles.py: 6 added lines, 35 removed lines. Original line number Diff line number Diff line Loading @@ -5,13 +5,10 @@ The provisioner may map only the product roles, and read only the `oeg` client ( 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.adapters.keycloak.admin_api import AdminApi from portal_service.domain.errors import ServiceAccountNotFoundError from portal_service.domain.ports.product_roles import ProductRoleAssigner Loading @@ -28,20 +25,9 @@ class _Role(BaseModel): 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 def __init__(self, *, admin: AdminApi, oeg_client_uuid: str) -> None: self._admin = admin 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) Loading @@ -53,35 +39,20 @@ class KeycloakProductRoles(ProductRoleAssigner): 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()]) await self._admin.call(method, mappings, [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") response = await self._admin.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") response = await self._admin.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/ports/identity_directory.py 0 → 100644 +18 −0 Changes for src/portal_service/domain/ports/identity_directory.py: 18 added lines, 0 removed lines. Original line number Diff line number Diff line """Identity directory port: what Keycloak knows about a user, read-only.""" from abc import ABC, abstractmethod from pydantic import BaseModel class DirectoryUser(BaseModel): enabled: bool organization: str | None # Names of the product roles the user holds, including roles given through groups. products: frozenset[str] class IdentityDirectory(ABC): @abstractmethod async def get_user(self, subject: str) -> DirectoryUser | None: """None if there is no such user."""
tests/unit/test_admin_directory.py 0 → 100644 +66 −0 Changes for tests/unit/test_admin_directory.py: 66 added lines, 0 removed lines. Original line number Diff line number Diff line """Reading users through the Admin API, against a fake Keycloak.""" import httpx import pytest from portal_service.adapters.keycloak.admin_api import AdminApi from portal_service.adapters.keycloak.admin_directory import KeycloakAdminDirectory from portal_service.domain.errors import IdentityProviderUnavailableError ADMIN = "http://keycloak.test/admin/realms/oop" OEG = "oeg-uuid" def _directory(handler: httpx.MockTransport) -> KeycloakAdminDirectory: async def token() -> str: return "provisioner-token" admin = AdminApi(admin_url=ADMIN, provisioner_token=token, transport=handler) return KeycloakAdminDirectory(admin=admin, oeg_client_uuid=OEG) def _keycloak(user: dict[str, object] | None, roles: list[str]) -> httpx.MockTransport: def handler(request: httpx.Request) -> httpx.Response: path = request.url.path.removeprefix("/admin/realms/oop") if request.headers["Authorization"] != "Bearer provisioner-token": return httpx.Response(401) if path == "/users/maria-sub": return httpx.Response(200, json=user) if user is not None else httpx.Response(404) if path == f"/users/maria-sub/role-mappings/clients/{OEG}/composite": return httpx.Response(200, json=[{"id": name, "name": name} for name in roles]) return httpx.Response(404) return httpx.MockTransport(handler) async def test_a_user_with_an_organization_and_products() -> None: directory = _directory( _keycloak( {"enabled": True, "attributes": {"organization": ["acme"]}}, ["quality-on-demand", "edge-application-management"], ) ) user = await directory.get_user("maria-sub") assert user is not None assert (user.enabled, user.organization) == (True, "acme") assert user.products == {"quality-on-demand", "edge-application-management"} async def test_a_user_without_an_organization_or_products() -> None: user = await _directory(_keycloak({"enabled": False}, [])).get_user("maria-sub") assert user is not None assert (user.enabled, user.organization, user.products) == (False, None, frozenset()) async def test_an_unknown_user_is_none() -> None: assert await _directory(_keycloak(None, [])).get_user("maria-sub") is None async def test_a_refused_request_is_unavailable() -> None: directory = _directory(httpx.MockTransport(lambda request: httpx.Response(403))) with pytest.raises(IdentityProviderUnavailableError): await directory.get_user("maria-sub")
