diff --git a/src/user_engine/adapters/provisioning.py b/src/user_engine/adapters/provisioning.py index c5bacfe..f85d878 100644 --- a/src/user_engine/adapters/provisioning.py +++ b/src/user_engine/adapters/provisioning.py @@ -3,20 +3,29 @@ from __future__ import annotations import json +import time from typing import Any -from urllib.error import HTTPError +from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen -from user_engine.ports import ProvisioningRequest, ProvisioningResult +from user_engine.ports import IdentityDriftResult, ProvisioningRequest, ProvisioningResult class HTTPIdentityProvisioningAdapter: - def __init__(self, *, base_url: str, bearer_token: str, timeout: float = 10) -> None: + def __init__( + self, + *, + base_url: str, + bearer_token: str, + timeout: float = 10, + retry_delay: float = 0.1, + ) -> None: self.base_url = base_url.rstrip("/") self.bearer_token = bearer_token.strip() if not self.bearer_token: raise ValueError("bearer token must not be empty") self.timeout = timeout + self.retry_delay = retry_delay def provision(self, request: ProvisioningRequest) -> ProvisioningResult: return self._post("/v1/identities/provision", { @@ -38,6 +47,24 @@ class HTTPIdentityProvisioningAdapter: def deprovision(self, *, external_subject: str, idempotency_key: str, correlation_id: str) -> ProvisioningResult: return self._lifecycle("deprovision", external_subject, idempotency_key, correlation_id) + def drift( + self, + request: ProvisioningRequest, + *, + external_subject: str, + desired_status: str = "active", + ) -> IdentityDriftResult: + return self._desired("drift", request, external_subject, desired_status) + + def reconcile( + self, + request: ProvisioningRequest, + *, + external_subject: str, + desired_status: str = "active", + ) -> IdentityDriftResult: + return self._desired("reconcile", request, external_subject, desired_status) + def _lifecycle(self, action: str, subject: str, key: str, correlation_id: str) -> ProvisioningResult: return self._post(f"/v1/identities/{action}", { "external_subject": subject, @@ -45,22 +72,34 @@ class HTTPIdentityProvisioningAdapter: "correlation_id": correlation_id, }) - def _post(self, path: str, payload: dict[str, Any]) -> ProvisioningResult: - request = Request( - self.base_url + path, - data=json.dumps(payload).encode(), - headers={ - "Authorization": f"Bearer {self.bearer_token}", - "Content-Type": "application/json", - }, - method="POST", + def _desired( + self, + action: str, + request: ProvisioningRequest, + external_subject: str, + desired_status: str, + ) -> IdentityDriftResult: + result = self._request(f"/v1/identities/{action}", { + "external_subject": external_subject, + "user_id": request.user_id, + "tenant": request.tenant, + "primary_email": request.primary_email, + "display_name": request.display_name, + "idempotency_key": request.idempotency_key, + "correlation_id": request.correlation_id, + "roles": request.roles, + "desired_status": desired_status, + }) + return IdentityDriftResult( + provider=str(result["provider"]), + external_subject=str(result["external_subject"]), + status=str(result["status"]), + drift=tuple(str(item) for item in result.get("drift", ())), + changed=tuple(str(item) for item in result.get("changed", ())), ) - try: - with urlopen(request, timeout=self.timeout) as response: - result = json.loads(response.read()) - except HTTPError as exc: - message = exc.read(4096).decode("utf-8", "replace") - raise RuntimeError(f"identity provisioning failed ({exc.code}): {message}") from exc + + def _post(self, path: str, payload: dict[str, Any]) -> ProvisioningResult: + result = self._request(path, payload) return ProvisioningResult( provider=str(result["provider"]), external_subject=str(result["external_subject"]), @@ -72,3 +111,30 @@ class HTTPIdentityProvisioningAdapter: else None ), ) + + def _request(self, path: str, payload: dict[str, Any]) -> dict[str, Any]: + request = Request( + self.base_url + path, + data=json.dumps(payload).encode(), + headers={ + "Authorization": f"Bearer {self.bearer_token}", + "Content-Type": "application/json", + }, + method="POST", + ) + for attempt in range(2): + try: + with urlopen(request, timeout=self.timeout) as response: + result = json.loads(response.read()) + break + except HTTPError as exc: + if exc.code not in {502, 503, 504} or attempt: + message = exc.read(4096).decode("utf-8", "replace") + raise RuntimeError( + f"identity provisioning failed ({exc.code}): {message}" + ) from exc + except URLError as exc: + if attempt: + raise RuntimeError("identity provisioning dependency unavailable") from exc + time.sleep(self.retry_delay) + return dict(result) diff --git a/src/user_engine/ports.py b/src/user_engine/ports.py index 2493033..5e4b18b 100644 --- a/src/user_engine/ports.py +++ b/src/user_engine/ports.py @@ -66,6 +66,15 @@ class ProvisioningResult: password_setup_url: str | None = None +@dataclass(frozen=True) +class IdentityDriftResult: + provider: str + external_subject: str + status: str + drift: tuple[str, ...] = () + changed: tuple[str, ...] = () + + class IdentityProvisioningPort(Protocol): """Lifecycle seam owned by NetKingdom adapters, not the user domain.""" @@ -87,6 +96,24 @@ class IdentityProvisioningPort(Protocol): ) -> ProvisioningResult: """Remove or tombstone an identity according to provider policy.""" + def drift( + self, + request: ProvisioningRequest, + *, + external_subject: str, + desired_status: str = "active", + ) -> IdentityDriftResult: + """Inspect provider state without changing it or exposing credentials.""" + + def reconcile( + self, + request: ProvisioningRequest, + *, + external_subject: str, + desired_status: str = "active", + ) -> IdentityDriftResult: + """Converge managed provider state toward the requested lifecycle.""" + class UserEngineStore(Protocol): """Durable persistence boundary for user-engine service behavior. diff --git a/tests/test_provisioning_adapter.py b/tests/test_provisioning_adapter.py index 5f4a162..98a1740 100644 --- a/tests/test_provisioning_adapter.py +++ b/tests/test_provisioning_adapter.py @@ -2,6 +2,7 @@ import io import json import unittest from unittest.mock import patch +from urllib.error import URLError from user_engine.adapters.provisioning import HTTPIdentityProvisioningAdapter from user_engine.ports import ProvisioningRequest @@ -41,3 +42,66 @@ class ProvisioningAdapterTests(unittest.TestCase): request = opener.call_args.args[0] self.assertEqual("Bearer secret", request.headers["Authorization"]) self.assertIn(b"idem-1234567890123456", request.data) + + @patch("user_engine.adapters.provisioning.urlopen") + def test_drift_and_reconcile_use_provider_neutral_desired_state(self, opener): + opener.side_effect = [ + Response(json.dumps({ + "provider": "netkingdom-lldap", + "external_subject": "person", + "status": "drifted", + "drift": ["status:mismatch"], + "changed": [], + }).encode()), + Response(json.dumps({ + "provider": "netkingdom-lldap", + "external_subject": "person", + "status": "reconciled", + "drift": [], + "changed": ["group:removed:netkingdom-suspended"], + }).encode()), + ] + adapter = HTTPIdentityProvisioningAdapter( + base_url="http://provisioner", bearer_token="secret" + ) + request = ProvisioningRequest( + user_id="person", + tenant="tenant:friendly:binky", + primary_email="person@example.test", + display_name="Person", + idempotency_key="idem-1234567890123456", + correlation_id="corr-1", + roles=("user",), + ) + self.assertEqual( + ("status:mismatch",), + adapter.drift(request, external_subject="person").drift, + ) + reconciled = adapter.reconcile(request, external_subject="person") + self.assertEqual("reconciled", reconciled.status) + self.assertEqual( + ("group:removed:netkingdom-suspended",), reconciled.changed + ) + + @patch("user_engine.adapters.provisioning.urlopen") + def test_idempotent_request_retries_one_transient_provider_failure(self, opener): + opener.side_effect = [ + URLError("temporary"), + Response(json.dumps({ + "provider": "netkingdom-lldap", + "external_subject": "person", + "status": "active", + }).encode()), + ] + adapter = HTTPIdentityProvisioningAdapter( + base_url="http://provisioner", + bearer_token="secret", + retry_delay=0, + ) + result = adapter.reactivate( + external_subject="person", + idempotency_key="idem-1234567890123456", + correlation_id="corr-1", + ) + self.assertEqual("active", result.status) + self.assertEqual(2, opener.call_count) diff --git a/workplans/USER-WP-0020-self-service-and-user-administration-portal.md b/workplans/USER-WP-0020-self-service-and-user-administration-portal.md index a25f705..03c0706 100644 --- a/workplans/USER-WP-0020-self-service-and-user-administration-portal.md +++ b/workplans/USER-WP-0020-self-service-and-user-administration-portal.md @@ -73,6 +73,14 @@ invokes a provider-neutral provisioning port with roles, correlation, and a required idempotency key. Remaining API breadth and OpenAPI/outbox work keep this task in progress. +2026-07-29 increment: `IdentityProvisioningPort` now includes provider-neutral +read-only drift inspection and active reconciliation. The HTTP adapter carries +the same desired identity envelope used for provisioning, returns only +machine-readable differences and changes, and retries one transient dependency +failure under the required idempotency key. NetKingdom's live LLDAP adapter +proved drift detection, convergence, and replay-safe cleanup. Remaining API +breadth and OpenAPI/outbox delivery keep this task in progress. + ## T03 - Implement self-service login and registration UX ```task @@ -161,6 +169,12 @@ accessibility, and restore matrix remains. The admin increment adds explicit regression coverage for missing/wrong CSRF, required API idempotency, provider-link persistence, and suspension calls. +2026-07-29 evidence: the provider-neutral adapter suite now covers drift +inspection, convergence results, transient retry, and replay-safe identity +cleanup; the full user-engine suite passes 107 tests with three provider +integration skips. A live disposable identity was suspended, reported as +drifted, reconciled to `in_sync`, and removed without leaving directory state. + ## T07 - Binky production acceptance ```task