Add production authorization and delivery adapters

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",
 ]

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}")

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)

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,
     )
 
 

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()