From 26d0e52172d559cd6ad9ae198c8519fdf4245eec Mon Sep 17 00:00:00 2001 From: tegwick Date: Wed, 29 Jul 2026 23:29:36 +0200 Subject: [PATCH] Add identity drift reconciliation --- identity-provisioner/provisioner.py | 152 +++++++++++++++++- identity-provisioner/server.py | 2 + .../tests/test_provisioner.py | 17 +- 3 files changed, 164 insertions(+), 7 deletions(-) diff --git a/identity-provisioner/provisioner.py b/identity-provisioner/provisioner.py index ab6e12e..2db2524 100644 --- a/identity-provisioner/provisioner.py +++ b/identity-provisioner/provisioner.py @@ -18,6 +18,15 @@ class Result: resumed: bool +@dataclass(frozen=True) +class DriftResult: + provider: str + external_subject: str + status: str + drift: tuple[str, ...] + changed: tuple[str, ...] = () + + class LLDAPProvisioner: def __init__(self, *, base_url: str, admin_password: str, opener: Callable = urlopen) -> None: self.base_url = base_url.rstrip("/") @@ -32,7 +41,8 @@ class LLDAPProvisioner: users, groups = self._directory(token) existing = next((user for user in users if user.get("id") == username), None) resumed = existing is not None - if existing is None: + created = existing is None + if created: self._gql(token, """ mutation CreateUser($id: String!, $email: String!, $display: String!) { createUser(user: {id: $id, email: $email, displayName: $display}) { id } @@ -47,9 +57,17 @@ mutation CreateUser($id: String!, $email: String!, $display: String!) { group_names = [f"{payload['tenant']}:users"] if "tenant-admin" in roles: group_names.append(f"{payload['tenant']}:admins") - for name in group_names: - group_id = self._ensure_group(token, groups, name) - self._add_group(token, username, group_id) + try: + for name in group_names: + group_id = self._ensure_group(token, groups, name) + self._add_group(token, username, group_id) + except Exception: + if created: + try: + self._delete(token, username) + except Exception: + pass + raise return Result("netkingdom-lldap", username, "password_setup_required", resumed) def suspend(self, subject: str) -> Result: @@ -72,9 +90,54 @@ mutation Remove($userId: String!, $groupId: Int!) { def deprovision(self, subject: str) -> Result: token = self._login() - self._gql(token, "mutation Delete($id: String!) { deleteUser(userId: $id) { ok } }", {"id": subject}) + if self._user(token, subject) is None: + return Result("netkingdom-lldap", subject, "deprovisioned", True) + self._delete(token, subject) return Result("netkingdom-lldap", subject, "deprovisioned", False) + def drift(self, payload: dict[str, Any]) -> DriftResult: + subject, email, desired_groups, desired_status = _desired(payload) + token = self._login() + user = self._user(token, subject) + drift = self._drift(user, email, desired_groups, desired_status, str(payload["tenant"])) + status = "in_sync" if not drift else "drifted" + return DriftResult("netkingdom-lldap", subject, status, tuple(drift)) + + def reconcile(self, payload: dict[str, Any]) -> DriftResult: + subject, email, desired_groups, desired_status = _desired(payload) + token = self._login() + user = self._user(token, subject) + changed: list[str] = [] + if user is None: + result = self.provision(payload) + changed.append("user:created") + subject = result.external_subject + token = self._login() + user = self._user(token, subject) + if user is None: + raise RuntimeError("directory reconciliation did not create the identity") + if str(user.get("email", "")).lower() != email: + raise ValueError("directory email drift requires explicit identity repair") + + groups = list(self._directory(token)[1]) + current = {str(item["displayName"]): int(item["id"]) for item in user.get("groups", ())} + tenant = str(payload["tenant"]) + managed = { + name for name in current + if name in {f"{tenant}:users", f"{tenant}:admins", "netkingdom-suspended"} + } + for name in sorted(desired_groups - managed): + self._add_group(token, subject, self._ensure_group(token, groups, name)) + changed.append(f"group:added:{name}") + for name in sorted(managed - desired_groups): + self._remove_group(token, subject, current[name]) + changed.append(f"group:removed:{name}") + + user = self._user(token, subject) + remaining = self._drift(user, email, desired_groups, desired_status, tenant) + status = "reconciled" if not remaining else "drifted" + return DriftResult("netkingdom-lldap", subject, status, tuple(remaining), tuple(changed)) + def _login(self) -> str: request = Request( self.base_url + "/auth/simple/login", @@ -89,6 +152,14 @@ mutation Remove($userId: String!, $groupId: Int!) { value = self._gql(token, "query { users { id email displayName } groups { id displayName } }", {}) return list(value["users"]), list(value["groups"]) + def _user(self, token: str, subject: str) -> dict[str, Any] | None: + value = self._gql(token, """ +query User($id: String!) { + user(userId: $id) { id email displayName groups { id displayName } } +}""", {"id": subject}) + user = value.get("user") + return dict(user) if user else None + def _ensure_group(self, token: str, groups: list[dict], name: str) -> int: existing = next((group for group in groups if group.get("displayName") == name), None) if existing: @@ -111,6 +182,44 @@ mutation Add($userId: String!, $groupId: Int!) { if "already" not in str(exc).lower() and "unique" not in str(exc).lower(): raise + def _remove_group(self, token: str, username: str, group_id: int) -> None: + self._gql(token, """ +mutation Remove($userId: String!, $groupId: Int!) { + removeUserFromGroup(userId: $userId, groupId: $groupId) { ok } +}""", {"userId": username, "groupId": group_id}) + + def _delete(self, token: str, subject: str) -> None: + self._gql( + token, + "mutation Delete($id: String!) { deleteUser(userId: $id) { ok } }", + {"id": subject}, + ) + + @staticmethod + def _drift( + user: dict[str, Any] | None, + email: str, + desired_groups: set[str], + desired_status: str, + tenant: str, + ) -> list[str]: + if user is None: + return ["user:missing"] + drift: list[str] = [] + if str(user.get("email", "")).lower() != email: + drift.append("email:mismatch") + current = {str(item["displayName"]) for item in user.get("groups", ())} + managed = { + name for name in current + if name in {f"{tenant}:users", f"{tenant}:admins", "netkingdom-suspended"} + } + drift.extend(f"group:missing:{name}" for name in sorted(desired_groups - managed)) + drift.extend(f"group:unexpected:{name}" for name in sorted(managed - desired_groups)) + suspended = "netkingdom-suspended" in current + if suspended != (desired_status == "suspended"): + drift.append("status:mismatch") + return drift + def _gql(self, token: str, query: str, variables: dict[str, Any]) -> dict: request = Request( self.base_url + "/api/graphql", @@ -125,9 +234,15 @@ mutation Add($userId: String!, $groupId: Int!) { return dict(payload.get("data") or {}) -def dispatch(provisioner: LLDAPProvisioner, path: str, payload: dict[str, Any]) -> Result: +def dispatch( + provisioner: LLDAPProvisioner, path: str, payload: dict[str, Any] +) -> Result | DriftResult: if path == "/v1/identities/provision": return provisioner.provision(payload) + if path == "/v1/identities/drift": + return provisioner.drift(payload) + if path == "/v1/identities/reconcile": + return provisioner.reconcile(payload) _required(payload, "external_subject", "idempotency_key", "correlation_id") subject = str(payload["external_subject"]) if path == "/v1/identities/suspend": @@ -153,3 +268,28 @@ def _required(payload: dict[str, Any], *fields: str) -> None: raise ValueError("missing required fields: " + ", ".join(missing)) if len(str(payload.get("idempotency_key", ""))) < 16: raise ValueError("idempotency_key must contain at least 16 characters") + + +def _desired(payload: dict[str, Any]) -> tuple[str, str, set[str], str]: + _required( + payload, + "external_subject", + "tenant", + "primary_email", + "idempotency_key", + "correlation_id", + ) + subject = str(payload["external_subject"]) + email = str(payload["primary_email"]).strip().lower() + if _username(email) != subject: + raise ValueError("external_subject does not match canonical email username") + desired_status = str(payload.get("desired_status", "active")) + if desired_status not in {"active", "suspended"}: + raise ValueError("desired_status must be active or suspended") + tenant = str(payload["tenant"]) + groups = {f"{tenant}:users"} + if "tenant-admin" in {str(role) for role in payload.get("roles", ())}: + groups.add(f"{tenant}:admins") + if desired_status == "suspended": + groups.add("netkingdom-suspended") + return subject, email, groups, desired_status diff --git a/identity-provisioner/server.py b/identity-provisioner/server.py index 426e791..6eb04bc 100644 --- a/identity-provisioner/server.py +++ b/identity-provisioner/server.py @@ -41,6 +41,8 @@ class Handler(BaseHTTPRequestHandler): return self._send(404, {"error": "not_found"}) except (ValueError, json.JSONDecodeError) as exc: return self._send(400, {"error": "invalid_request", "message": str(exc)}) + except RuntimeError: + return self._send(503, {"error": "dependency_unavailable"}) response = asdict(result) if path == "/v1/identities/provision" and result.status == "password_setup_required": response["password_setup_url"] = self.password_setups.issue( diff --git a/identity-provisioner/tests/test_provisioner.py b/identity-provisioner/tests/test_provisioner.py index 91b1909..e84051d 100644 --- a/identity-provisioner/tests/test_provisioner.py +++ b/identity-provisioner/tests/test_provisioner.py @@ -3,7 +3,7 @@ import sys import unittest sys.path.insert(0, str(pathlib.Path(__file__).parents[1])) -from provisioner import _username, dispatch, Result +from provisioner import _username, dispatch, DriftResult, Result class Fake: @@ -11,6 +11,8 @@ class Fake: def suspend(self, subject): return Result("p", subject, "suspended", False) def reactivate(self, subject): return Result("p", subject, "active", False) def deprovision(self, subject): return Result("p", subject, "deprovisioned", False) + def drift(self, payload): return DriftResult("p", payload["external_subject"], "drifted", ("group:missing:t:users",)) + def reconcile(self, payload): return DriftResult("p", payload["external_subject"], "reconciled", (), ("group:added:t:users",)) class ProvisionerTests(unittest.TestCase): @@ -30,3 +32,16 @@ class ProvisionerTests(unittest.TestCase): "correlation_id": "c", }) self.assertEqual("suspended", result.status) + + def test_drift_and_reconcile_dispatch(self): + payload = { + "external_subject": "u", + "tenant": "t", + "primary_email": "u@example.com", + "idempotency_key": "1234567890123456", + "correlation_id": "c", + } + drift = dispatch(Fake(), "/v1/identities/drift", payload) + reconciled = dispatch(Fake(), "/v1/identities/reconcile", payload) + self.assertEqual("drifted", drift.status) + self.assertEqual("reconciled", reconciled.status)