From 292e7e0e3e4de6a222f1e433f4a3d67b07ec716d Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 9 Aug 2026 01:53:30 +0200 Subject: [PATCH] Add production authorization and delivery adapters --- src/user_engine/adapters/__init__.py | 4 ++ src/user_engine/adapters/delivery.py | 59 ++++++++++++++++++ src/user_engine/adapters/flex_auth.py | 87 ++++++++++++++++++++++++++ src/user_engine/runtime.py | 18 ++++-- tests/test_platform_adapters.py | 88 +++++++++++++++++++++++++++ 5 files changed, 252 insertions(+), 4 deletions(-) create mode 100644 src/user_engine/adapters/delivery.py create mode 100644 src/user_engine/adapters/flex_auth.py create mode 100644 tests/test_platform_adapters.py diff --git a/src/user_engine/adapters/__init__.py b/src/user_engine/adapters/__init__.py index 9e84843..e2ed8ca 100644 --- a/src/user_engine/adapters/__init__.py +++ b/src/user_engine/adapters/__init__.py @@ -8,6 +8,8 @@ from user_engine.adapters.postgres import PostgresUserEngineStore from user_engine.adapters.claims import VerifiedIdentityClaimsAdapter from user_engine.adapters.provisioning import HTTPIdentityProvisioningAdapter from user_engine.adapters.tenant_management import HTTPTenantManagementAdapter +from user_engine.adapters.flex_auth import FlexAuthHTTPAdapter +from user_engine.adapters.delivery import HTTPOutboxDeliveryAdapter __all__ = [ "InMemoryUserEngineStore", @@ -16,4 +18,6 @@ __all__ = [ "VerifiedIdentityClaimsAdapter", "HTTPIdentityProvisioningAdapter", "HTTPTenantManagementAdapter", + "FlexAuthHTTPAdapter", + "HTTPOutboxDeliveryAdapter", ] diff --git a/src/user_engine/adapters/delivery.py b/src/user_engine/adapters/delivery.py new file mode 100644 index 0000000..5e9f2ab --- /dev/null +++ b/src/user_engine/adapters/delivery.py @@ -0,0 +1,59 @@ +"""HTTP delivery adapters for durable platform events and invitation mail.""" + +from __future__ import annotations + +import json +from urllib.request import Request, urlopen + +from user_engine.domain import OutboxEvent + +_MAIL_EVENTS = {"family_member.invited", "family_invitation.resent"} + + +class HTTPOutboxDeliveryAdapter: + """Deliver one outbox event to idempotent platform HTTP lanes.""" + + def __init__( + self, + *, + event_url: str, + bearer_token: str, + mail_url: str | None = None, + timeout_seconds: float = 5.0, + ) -> None: + self.event_url = event_url + self.mail_url = mail_url + self.bearer_token = bearer_token + self.timeout_seconds = timeout_seconds + + def __call__(self, event: OutboxEvent) -> None: + envelope = { + "id": event.event_id, + "type": event.event_type, + "source": "user-engine", + "subject": event.aggregate_id, + "tenant": event.tenant, + "correlation_id": event.correlation_id, + "occurred_at": event.occurred_at.isoformat(), + "data": dict(event.payload), + } + if self.mail_url and event.event_type in _MAIL_EVENTS: + self._post(self.mail_url, envelope) + self._post(self.event_url, envelope) + + def _post(self, url: str, envelope: dict[str, object]) -> None: + with urlopen( + Request( + url, + data=json.dumps(envelope).encode(), + headers={ + "Authorization": f"Bearer {self.bearer_token}", + "Content-Type": "application/json", + "Idempotency-Key": str(envelope["id"]), + }, + method="POST", + ), + timeout=self.timeout_seconds, + ) as response: + if response.status < 200 or response.status >= 300: + raise RuntimeError(f"delivery rejected with HTTP {response.status}") diff --git a/src/user_engine/adapters/flex_auth.py b/src/user_engine/adapters/flex_auth.py new file mode 100644 index 0000000..a6bd061 --- /dev/null +++ b/src/user_engine/adapters/flex_auth.py @@ -0,0 +1,87 @@ +"""Fail-closed flex-auth HTTP authorization adapter.""" + +from __future__ import annotations + +import json +from collections.abc import Iterable +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + +from user_engine.domain import ( + AuthorizationDecision, + AuthorizationEffect, + AuthorizationRequest, +) + + +class FlexAuthHTTPAdapter: + """Evaluate user-engine actions through flex-auth POST /v1/check.""" + + def __init__(self, *, base_url: str, timeout_seconds: float = 3.0) -> None: + self.url = f"{base_url.rstrip('/')}/v1/check" + self.timeout_seconds = timeout_seconds + + def check(self, request: AuthorizationRequest) -> AuthorizationDecision: + payload = { + "id": request.correlation_id, + "tenant": request.tenant, + "subject": { + "id": request.actor.subject, + "type": request.actor.principal_type.value, + "tenant": request.actor.tenant, + "attributes": { + "issuer": request.actor.issuer, + "roles": list(request.actor.roles), + "groups": list(request.actor.groups), + "scopes": list(request.actor.scopes), + "assurance": dict(request.actor.assurance), + }, + }, + "action": request.action, + "resource": { + "id": request.resource_id, + "type": request.resource_type, + "system": "user-engine", + "tenant": request.tenant, + "attributes": { + "application_id": request.application_id, + "target_user_id": request.target_user_id, + }, + }, + "context": dict(request.context), + } + try: + response = urlopen( + Request( + self.url, + data=json.dumps(payload).encode(), + headers={"Content-Type": "application/json"}, + method="POST", + ), + timeout=self.timeout_seconds, + ) + body = json.load(response) + effect = AuthorizationEffect(str(body["effect"])) + decision_id = str(body["id"]) + reason = str(body.get("reason") or "flex-auth") + obligations = tuple( + str(item.get("type")) + for item in body.get("obligations", ()) + if isinstance(item, dict) and item.get("type") + ) + return AuthorizationDecision( + effect=effect, + decision_id=decision_id, + reason=reason, + obligations=obligations, + ) + except (HTTPError, URLError, TimeoutError, ValueError, KeyError, TypeError): + return AuthorizationDecision( + effect=AuthorizationEffect.DENY, + reason="authorization service unavailable", + ) + + def batch_check( + self, requests: Iterable[AuthorizationRequest] + ) -> tuple[AuthorizationDecision, ...]: + return tuple(self.check(request) for request in requests) diff --git a/src/user_engine/runtime.py b/src/user_engine/runtime.py index 2f6a4c0..acade1a 100644 --- a/src/user_engine/runtime.py +++ b/src/user_engine/runtime.py @@ -6,7 +6,8 @@ import os from wsgiref.simple_server import make_server from user_engine.adapters import ( - LocalAuthorizationCheckPort, + FlexAuthHTTPAdapter, + HTTPOutboxDeliveryAdapter, PostgresUserEngineStore, VerifiedIdentityClaimsAdapter, HTTPIdentityProvisioningAdapter, @@ -20,8 +21,7 @@ from user_engine.web import PortalApplication def create_application() -> PortalApplication: """Assemble the runtime from secret-backed environment references. - The local authorization adapter is an explicit pre-production bridge. A - flex-auth HTTP adapter must replace it before the production gate. + Production authorization and delivery are fail-closed HTTP boundaries. """ try: @@ -38,7 +38,10 @@ def create_application() -> PortalApplication: expected_issuer=_required("USER_ENGINE_OIDC_ISSUER"), expected_audience=_required("USER_ENGINE_OIDC_AUDIENCE"), ), - authorization=LocalAuthorizationCheckPort(), + authorization=FlexAuthHTTPAdapter( + base_url=_required("USER_ENGINE_FLEX_AUTH_URL"), + timeout_seconds=float(os.environ.get("USER_ENGINE_FLEX_AUTH_TIMEOUT", "3")), + ), ) tenant_management = None if os.environ.get("USER_ENGINE_TENANT_MANAGEMENT_URL"): @@ -46,6 +49,12 @@ def create_application() -> PortalApplication: base_url=_required("USER_ENGINE_TENANT_MANAGEMENT_URL"), bearer_token=_required("USER_ENGINE_TENANT_MANAGEMENT_TOKEN"), ) + outbox_delivery = HTTPOutboxDeliveryAdapter( + event_url=_required("USER_ENGINE_EVENT_URL"), + mail_url=os.environ.get("USER_ENGINE_MAIL_URL"), + bearer_token=_required("USER_ENGINE_DELIVERY_TOKEN"), + timeout_seconds=float(os.environ.get("USER_ENGINE_DELIVERY_TIMEOUT", "5")), + ) return PortalApplication( service, trusted_proxy_secret=_required("USER_ENGINE_PROXY_SECRET"), @@ -64,6 +73,7 @@ def create_application() -> PortalApplication: bearer_token=_required("USER_ENGINE_PROVISIONING_TOKEN"), ), tenant_management=tenant_management, + outbox_delivery=outbox_delivery, ) diff --git a/tests/test_platform_adapters.py b/tests/test_platform_adapters.py new file mode 100644 index 0000000..1f064c0 --- /dev/null +++ b/tests/test_platform_adapters.py @@ -0,0 +1,88 @@ +import io +import json +import unittest +from datetime import UTC, datetime +from unittest.mock import patch +from urllib.error import URLError + +from user_engine.adapters.delivery import HTTPOutboxDeliveryAdapter +from user_engine.adapters.flex_auth import FlexAuthHTTPAdapter +from user_engine.domain import ( + Actor, + AuthorizationEffect, + AuthorizationRequest, + OutboxEvent, + PrincipalType, +) + + +class _Response(io.BytesIO): + status = 200 + + def __enter__(self): + return self + + def __exit__(self, *_args): + self.close() + + +class PlatformAdapterTests(unittest.TestCase): + def test_flex_auth_maps_allow_and_decision_id(self): + body = _Response(json.dumps({"id": "decision:1", "effect": "allow"}).encode()) + with patch("user_engine.adapters.flex_auth.urlopen", return_value=body) as call: + decision = FlexAuthHTTPAdapter(base_url="http://flex-auth").check(_request()) + self.assertEqual(decision.effect, AuthorizationEffect.ALLOW) + self.assertEqual(decision.decision_id, "decision:1") + request = json.loads(call.call_args.args[0].data) + self.assertEqual(request["resource"]["system"], "user-engine") + self.assertEqual(request["context"]["self"], True) + + def test_flex_auth_fails_closed_when_unavailable(self): + with patch("user_engine.adapters.flex_auth.urlopen", side_effect=URLError("down")): + decision = FlexAuthHTTPAdapter(base_url="http://flex-auth").check(_request()) + self.assertEqual(decision.effect, AuthorizationEffect.DENY) + self.assertEqual(decision.reason, "authorization service unavailable") + + def test_invitation_delivery_calls_mail_and_event_with_idempotency(self): + adapter = HTTPOutboxDeliveryAdapter( + event_url="http://events", mail_url="http://mail", bearer_token="opaque" + ) + with patch("user_engine.adapters.delivery.urlopen", return_value=_Response()) as call: + adapter(_event("family_member.invited")) + self.assertEqual([item.args[0].full_url for item in call.call_args_list], + ["http://mail", "http://events"]) + for item in call.call_args_list: + self.assertEqual(item.args[0].get_header("Idempotency-key"), "evt-1") + + def test_non_mail_event_only_calls_event_lane(self): + adapter = HTTPOutboxDeliveryAdapter( + event_url="http://events", mail_url="http://mail", bearer_token="opaque" + ) + with patch("user_engine.adapters.delivery.urlopen", return_value=_Response()) as call: + adapter(_event("membership.added")) + self.assertEqual(call.call_count, 1) + self.assertEqual(call.call_args.args[0].full_url, "http://events") + +def _request(): + actor = Actor( + issuer="https://issuer", subject="subject-1", tenant="tenant-a", + principal_type=PrincipalType.HUMAN, audience=("user-engine",), + roles=("tenant-admin",), + ) + return AuthorizationRequest( + actor=actor, resource_type="user-engine:user", resource_id="user-1", + action="user.update", tenant="tenant-a", correlation_id="corr-1", + target_user_id="user-1", context={"self": True}, + ) + + +def _event(event_type): + return OutboxEvent( + event_id="evt-1", event_type=event_type, aggregate_id="inv-1", + payload={"primary_email": "person@example.test"}, tenant="tenant-a", + correlation_id="corr-1", occurred_at=datetime(2026, 8, 8, tzinfo=UTC), + ) + + +if __name__ == "__main__": + unittest.main()