diff --git a/identity-provisioner/Containerfile b/identity-provisioner/Containerfile
new file mode 100644
index 0000000..7d02a32
--- /dev/null
+++ b/identity-provisioner/Containerfile
@@ -0,0 +1,7 @@
+FROM python:3.12-slim@sha256:d764629ce0ddd8c71fd371e9901efb324a95789d2315a47db7e4d27e78f1b0e9
+RUN useradd --create-home --uid 10001 provisioner
+WORKDIR /app
+COPY provisioner.py server.py /app/
+USER 10001:10001
+EXPOSE 8080
+CMD ["python", "server.py"]
diff --git a/identity-provisioner/copy-client-secret.py b/identity-provisioner/copy-client-secret.py
new file mode 100644
index 0000000..a2bc2ba
--- /dev/null
+++ b/identity-provisioner/copy-client-secret.py
@@ -0,0 +1,18 @@
+"""Copy only the provisioner token into a namespace-local client Secret."""
+
+import json
+import sys
+
+
+source = json.load(sys.stdin)
+target = {
+    "apiVersion": "v1",
+    "kind": "Secret",
+    "metadata": {
+        "name": "identity-provisioner-client",
+        "namespace": "user-engine",
+    },
+    "type": "Opaque",
+    "data": {"token": source["data"]["token"]},
+}
+json.dump(target, sys.stdout)
diff --git a/identity-provisioner/provisioner.py b/identity-provisioner/provisioner.py
new file mode 100644
index 0000000..ab6e12e
--- /dev/null
+++ b/identity-provisioner/provisioner.py
@@ -0,0 +1,155 @@
+"""NetKingdom's idempotent LLDAP lifecycle adapter."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import json
+import re
+import secrets
+from typing import Any, Callable
+from urllib.request import Request, urlopen
+
+
+@dataclass(frozen=True)
+class Result:
+    provider: str
+    external_subject: str
+    status: str
+    resumed: bool
+
+
+class LLDAPProvisioner:
+    def __init__(self, *, base_url: str, admin_password: str, opener: Callable = urlopen) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.admin_password = admin_password
+        self.opener = opener
+
+    def provision(self, payload: dict[str, Any]) -> Result:
+        _required(payload, "user_id", "tenant", "primary_email", "idempotency_key", "correlation_id")
+        email = str(payload["primary_email"]).strip().lower()
+        username = _username(email)
+        token = self._login()
+        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:
+            self._gql(token, """
+mutation CreateUser($id: String!, $email: String!, $display: String!) {
+  createUser(user: {id: $id, email: $email, displayName: $display}) { id }
+}""", {
+                "id": username,
+                "email": email,
+                "display": str(payload.get("display_name") or username),
+            })
+        elif str(existing.get("email", "")).lower() != email:
+            raise ValueError("directory username collision")
+        roles = {str(role) for role in payload.get("roles", ())}
+        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)
+        return Result("netkingdom-lldap", username, "password_setup_required", resumed)
+
+    def suspend(self, subject: str) -> Result:
+        token = self._login()
+        _, groups = self._directory(token)
+        group_id = self._ensure_group(token, groups, "netkingdom-suspended")
+        self._add_group(token, subject, group_id)
+        return Result("netkingdom-lldap", subject, "suspended", False)
+
+    def reactivate(self, subject: str) -> Result:
+        token = self._login()
+        _, groups = self._directory(token)
+        group = next((item for item in groups if item.get("displayName") == "netkingdom-suspended"), None)
+        if group:
+            self._gql(token, """
+mutation Remove($userId: String!, $groupId: Int!) {
+  removeUserFromGroup(userId: $userId, groupId: $groupId) { ok }
+}""", {"userId": subject, "groupId": int(group["id"])})
+        return Result("netkingdom-lldap", subject, "active", False)
+
+    def deprovision(self, subject: str) -> Result:
+        token = self._login()
+        self._gql(token, "mutation Delete($id: String!) { deleteUser(userId: $id) { ok } }", {"id": subject})
+        return Result("netkingdom-lldap", subject, "deprovisioned", False)
+
+    def _login(self) -> str:
+        request = Request(
+            self.base_url + "/auth/simple/login",
+            data=json.dumps({"username": "admin", "password": self.admin_password}).encode(),
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        with self.opener(request, timeout=10) as response:
+            return str(json.loads(response.read())["token"])
+
+    def _directory(self, token: str) -> tuple[list[dict], list[dict]]:
+        value = self._gql(token, "query { users { id email displayName } groups { id displayName } }", {})
+        return list(value["users"]), list(value["groups"])
+
+    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:
+            return int(existing["id"])
+        created = self._gql(
+            token,
+            "mutation CreateGroup($name: String!) { createGroup(name: $name) { id displayName } }",
+            {"name": name},
+        )["createGroup"]
+        groups.append(created)
+        return int(created["id"])
+
+    def _add_group(self, token: str, username: str, group_id: int) -> None:
+        try:
+            self._gql(token, """
+mutation Add($userId: String!, $groupId: Int!) {
+  addUserToGroup(userId: $userId, groupId: $groupId) { ok }
+}""", {"userId": username, "groupId": group_id})
+        except ValueError as exc:
+            if "already" not in str(exc).lower() and "unique" not in str(exc).lower():
+                raise
+
+    def _gql(self, token: str, query: str, variables: dict[str, Any]) -> dict:
+        request = Request(
+            self.base_url + "/api/graphql",
+            data=json.dumps({"query": query, "variables": variables}).encode(),
+            headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
+            method="POST",
+        )
+        with self.opener(request, timeout=15) as response:
+            payload = json.loads(response.read())
+        if payload.get("errors"):
+            raise ValueError(str(payload["errors"][0].get("message", "LLDAP GraphQL error")))
+        return dict(payload.get("data") or {})
+
+
+def dispatch(provisioner: LLDAPProvisioner, path: str, payload: dict[str, Any]) -> Result:
+    if path == "/v1/identities/provision":
+        return provisioner.provision(payload)
+    _required(payload, "external_subject", "idempotency_key", "correlation_id")
+    subject = str(payload["external_subject"])
+    if path == "/v1/identities/suspend":
+        return provisioner.suspend(subject)
+    if path == "/v1/identities/reactivate":
+        return provisioner.reactivate(subject)
+    if path == "/v1/identities/deprovision":
+        return provisioner.deprovision(subject)
+    raise KeyError(path)
+
+
+def _username(email: str) -> str:
+    local = email.partition("@")[0].lower()
+    value = re.sub(r"[^a-z0-9._-]+", "-", local).strip("-")
+    if not value or "@" not in email:
+        raise ValueError("valid primary_email is required")
+    return value[:64]
+
+
+def _required(payload: dict[str, Any], *fields: str) -> None:
+    missing = [field for field in fields if not payload.get(field)]
+    if missing:
+        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")
diff --git a/identity-provisioner/server.py b/identity-provisioner/server.py
new file mode 100644
index 0000000..ade59b0
--- /dev/null
+++ b/identity-provisioner/server.py
@@ -0,0 +1,56 @@
+from dataclasses import asdict
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import os
+import secrets
+
+from provisioner import LLDAPProvisioner, dispatch
+
+
+class Handler(BaseHTTPRequestHandler):
+    provisioner: LLDAPProvisioner
+    service_token: str
+
+    def do_GET(self):
+        if self.path == "/healthz":
+            return self._send(200, {"status": "ok"})
+        self._send(404, {"error": "not_found"})
+
+    def do_POST(self):
+        supplied = self.headers.get("Authorization", "").removeprefix("Bearer ")
+        if not secrets.compare_digest(supplied, self.service_token):
+            return self._send(403, {"error": "access_denied"})
+        try:
+            length = min(int(self.headers.get("Content-Length", "0")), 65536)
+            payload = json.loads(self.rfile.read(length))
+            result = dispatch(self.provisioner, self.path, payload)
+        except KeyError:
+            return self._send(404, {"error": "not_found"})
+        except (ValueError, json.JSONDecodeError) as exc:
+            return self._send(400, {"error": "invalid_request", "message": str(exc)})
+        self._send(200, asdict(result))
+
+    def _send(self, status: int, payload: dict):
+        body = json.dumps(payload, separators=(",", ":")).encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Cache-Control", "no-store")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, format, *args):
+        return
+
+
+def main():
+    Handler.provisioner = LLDAPProvisioner(
+        base_url=os.environ["LLDAP_URL"],
+        admin_password=os.environ["LLDAP_ADMIN_PASSWORD"],
+    )
+    Handler.service_token = os.environ["PROVISIONER_SERVICE_TOKEN"]
+    ThreadingHTTPServer(("0.0.0.0", 8080), Handler).serve_forever()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/identity-provisioner/tests/test_provisioner.py b/identity-provisioner/tests/test_provisioner.py
new file mode 100644
index 0000000..91b1909
--- /dev/null
+++ b/identity-provisioner/tests/test_provisioner.py
@@ -0,0 +1,32 @@
+import pathlib
+import sys
+import unittest
+
+sys.path.insert(0, str(pathlib.Path(__file__).parents[1]))
+from provisioner import _username, dispatch, Result
+
+
+class Fake:
+    def provision(self, payload): return Result("p", "u", "password_setup_required", False)
+    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)
+
+
+class ProvisionerTests(unittest.TestCase):
+    def test_username_is_stable_and_sanitized(self):
+        self.assertEqual("bernd.worsch", _username("Bernd.Worsch@binky-hedgehog.com"))
+
+    def test_dispatch_requires_idempotency(self):
+        with self.assertRaisesRegex(ValueError, "idempotency_key"):
+            dispatch(Fake(), "/v1/identities/suspend", {
+                "external_subject": "u", "correlation_id": "c"
+            })
+
+    def test_lifecycle_dispatch(self):
+        result = dispatch(Fake(), "/v1/identities/suspend", {
+            "external_subject": "u",
+            "idempotency_key": "1234567890123456",
+            "correlation_id": "c",
+        })
+        self.assertEqual("suspended", result.status)
diff --git a/sso-mfa/k8s/identity-provisioner/deployment.yaml b/sso-mfa/k8s/identity-provisioner/deployment.yaml
new file mode 100644
index 0000000..d2b722b
--- /dev/null
+++ b/sso-mfa/k8s/identity-provisioner/deployment.yaml
@@ -0,0 +1,99 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: identity-provisioner
+  namespace: sso
+  labels: &labels
+    app.kubernetes.io/name: identity-provisioner
+    app.kubernetes.io/component: directory-lifecycle
+    app.kubernetes.io/part-of: net-kingdom-sso-mfa
+spec:
+  replicas: 1
+  selector:
+    matchLabels: {app.kubernetes.io/name: identity-provisioner}
+  template:
+    metadata:
+      labels: *labels
+    spec:
+      automountServiceAccountToken: false
+      securityContext:
+        runAsNonRoot: true
+        runAsUser: 10001
+        runAsGroup: 10001
+        seccompProfile: {type: RuntimeDefault}
+      containers:
+        - name: provisioner
+          image: identity-provisioner:20260728-1
+          imagePullPolicy: Never
+          ports: [{name: http, containerPort: 8080}]
+          env:
+            - {name: LLDAP_URL, value: "http://lldap.sso.svc.cluster.local:17170"}
+            - name: LLDAP_ADMIN_PASSWORD
+              valueFrom:
+                secretKeyRef: {name: lldap-secrets, key: LLDAP_LDAP_USER_PASS}
+            - name: PROVISIONER_SERVICE_TOKEN
+              valueFrom:
+                secretKeyRef: {name: identity-provisioner-token, key: token}
+          securityContext:
+            allowPrivilegeEscalation: false
+            capabilities: {drop: ["ALL"]}
+            readOnlyRootFilesystem: true
+          resources:
+            requests: {cpu: 25m, memory: 32Mi}
+            limits: {cpu: 250m, memory: 128Mi}
+          readinessProbe:
+            httpGet: {path: /healthz, port: http}
+            periodSeconds: 10
+          livenessProbe:
+            httpGet: {path: /healthz, port: http}
+            periodSeconds: 20
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: identity-provisioner
+  namespace: sso
+spec:
+  selector: {app.kubernetes.io/name: identity-provisioner}
+  ports: [{name: http, port: 8080, targetPort: http}]
+---
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: identity-provisioner
+  namespace: sso
+spec:
+  podSelector:
+    matchLabels: {app.kubernetes.io/name: identity-provisioner}
+  policyTypes: [Ingress, Egress]
+  ingress:
+    - from:
+        - namespaceSelector:
+            matchLabels: {kubernetes.io/metadata.name: user-engine}
+          podSelector:
+            matchLabels: {app.kubernetes.io/name: user-engine}
+      ports: [{protocol: TCP, port: 8080}]
+  egress:
+    - to:
+        - podSelector:
+            matchLabels: {app.kubernetes.io/name: lldap}
+      ports: [{protocol: TCP, port: 17170}]
+    - to:
+        - namespaceSelector:
+            matchLabels: {kubernetes.io/metadata.name: kube-system}
+      ports: [{protocol: UDP, port: 53}, {protocol: TCP, port: 53}]
+---
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: allow-identity-provisioner-to-lldap
+  namespace: sso
+spec:
+  podSelector:
+    matchLabels: {app.kubernetes.io/name: lldap}
+  policyTypes: [Ingress]
+  ingress:
+    - from:
+        - podSelector:
+            matchLabels: {app.kubernetes.io/name: identity-provisioner}
+      ports: [{protocol: TCP, port: 17170}]
diff --git a/sso-mfa/k8s/user-engine/runtime.yaml b/sso-mfa/k8s/user-engine/runtime.yaml
index ab3edc4..2095a0c 100644
--- a/sso-mfa/k8s/user-engine/runtime.yaml
+++ b/sso-mfa/k8s/user-engine/runtime.yaml
@@ -62,6 +62,10 @@ spec:
             - {name: USER_ENGINE_OIDC_REDIRECT_URI, value: "https://users.92-205-62-239.nip.io/oidc/callback"}
             - {name: USER_ENGINE_OIDC_BACKEND_URL, value: "http://keycape.sso.svc.cluster.local:8080"}
             - {name: USER_ENGINE_PUBLIC_REGISTRATION, value: "false"}
+            - {name: USER_ENGINE_PROVISIONING_URL, value: "http://identity-provisioner.sso.svc.cluster.local:8080"}
+            - name: USER_ENGINE_PROVISIONING_TOKEN
+              valueFrom:
+                secretKeyRef: {name: identity-provisioner-client, key: token}
           securityContext:
             allowPrivilegeEscalation: false
             capabilities: {drop: ["ALL"]}
@@ -117,6 +121,12 @@ spec:
           podSelector:
             matchLabels: {app.kubernetes.io/name: keycape}
       ports: [{protocol: TCP, port: 8080}]
+    - to:
+        - namespaceSelector:
+            matchLabels: {kubernetes.io/metadata.name: sso}
+          podSelector:
+            matchLabels: {app.kubernetes.io/name: identity-provisioner}
+      ports: [{protocol: TCP, port: 8080}]
     - to:
         - namespaceSelector:
             matchLabels: {kubernetes.io/metadata.name: kube-system}