NK-WP-0054-T05: synthetic bind-split rehearsal on pinned LLDAP (phase 1 of 4)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 308701@bnt-lap001
Assistant-Session: 5548e447-444c-4dcc-bed1-05c16eebfe56

docs/evidence/2026-10-28-nk54-bind-split-rehearsal/synthetic-rehearsal.json

@@ -0,0 +1,62 @@
+{
+  "data": "synthetic",
+  "findings": {
+    "admin_after_switch": {
+      "can_create_user": true,
+      "groups": [
+        "lldap_admin"
+      ],
+      "login_with_unchanged_password": true
+    },
+    "cleanup": true,
+    "ld_admin_introduced": {
+      "can_create_group": true,
+      "groups": [
+        "lldap_admin"
+      ],
+      "login": true
+    },
+    "provisioner_password_manager": {
+      "can_add_group_member": false,
+      "can_create_user": false,
+      "can_set_user_password": true,
+      "login": true,
+      "refused_admin_password": true,
+      "users_group_id_known": true
+    },
+    "read_only_binds": {
+      "svc-authelia": {
+        "bind": true,
+        "groups_visible": 6,
+        "people_visible": 8,
+        "sees_alice": true,
+        "write_refused": true
+      },
+      "svc-keycape": {
+        "bind": true,
+        "groups_visible": 6,
+        "people_visible": 8,
+        "sees_alice": true,
+        "write_refused": true
+      },
+      "svc-privacyidea": {
+        "bind": true,
+        "groups_visible": 6,
+        "people_visible": 8,
+        "sees_alice": true,
+        "write_refused": true
+      }
+    },
+    "si3_interactive_login": {
+      "svc-authelia": true,
+      "svc-keycape": true,
+      "svc-privacyidea": true
+    }
+  },
+  "image": "docker.io/lldap/lldap@sha256:9e605a66c02514bfcffd1b67cafb1e98d50992216bb2871d7ae44622047dd09d",
+  "observed_at": "2026-10-28T17:59:22.643767+00:00",
+  "production_contacted": false,
+  "schema": "netkingdom.lldap-bind-split-rehearsal.v1",
+  "task": "NK-WP-0054-T05",
+  "values_recorded": false
+}

sso-mfa/k8s/lldap/bind-split-rehearsal.py

@@ -0,0 +1,242 @@
+#!/usr/bin/env python3
+# /// script
+# dependencies = ["ldap3>=2.9,<3"]
+# ///
+"""Synthetic rehearsal of the directory bind split on the pinned LLDAP image (NK-WP-0054-T05).
+
+Agent-runnable: a throwaway LLDAP with generated passwords and a production-shaped directory,
+published on 127.0.0.1 only and removed afterwards. No backup, no production data, no Secret.
+It answers the questions the live migration depends on and writes a value-free JSON receipt:
+
+- can `ld-admin` be introduced into an existing directory whose super-admin is `admin`, and does
+  LLDAP start with `LLDAP_LDAP_USER_DN=ld-admin` without touching `admin`;
+- does `admin` keep its rights through `lldap_admin` group membership after that switch;
+- does a `lldap_strict_readonly` bind read users and groups but fail to write;
+- what is the narrowest group with which the provisioner can create users and set membership;
+- can a `svc-*` account sign in interactively to LLDAP (SI-3).
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import secrets
+import subprocess
+import sys
+import time
+import urllib.error
+import urllib.request
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+import ldap3
+
+IMAGE = "docker.io/lldap/lldap@sha256:9e605a66c02514bfcffd1b67cafb1e98d50992216bb2871d7ae44622047dd09d"
+BASE_DN = "dc=netkingdom,dc=local"
+READ_ONLY = ("svc-authelia", "svc-keycape", "svc-privacyidea")
+PRODUCTION_GROUPS = ("net-kingdom-users", "net-kingdom-admins")
+
+
+def docker(*args, env=None, check=True):
+    done = subprocess.run(["docker", *args], capture_output=True, text=True, env=env)
+    if check and done.returncode:
+        raise RuntimeError(f"docker {args[0]} failed")
+    return done
+
+
+class Directory:
+    def __init__(self, tag):
+        self.name = f"nk54-bind-split-{tag}"
+        self.volume = self.name
+        self.jwt = secrets.token_hex(32)
+        self.http = self.ldap = None
+
+    def start(self, user_dn, password):
+        docker("rm", "--force", self.name, check=False)
+        docker("run", "--detach", "--pull=never", "--name", self.name,
+               "--volume", f"{self.volume}:/data",
+               "--publish", "127.0.0.1::17170", "--publish", "127.0.0.1::3890",
+               "--env", f"LLDAP_JWT_SECRET={self.jwt}", "--env", f"LLDAP_LDAP_BASE_DN={BASE_DN}",
+               "--env", f"LLDAP_LDAP_USER_DN={user_dn}", "--env", "LLDAP_LDAP_USER_PASS",
+               IMAGE, env={"LLDAP_LDAP_USER_PASS": password, "PATH": "/usr/bin:/bin"})
+        ports = json.loads(docker("inspect", "--format", "{{json .NetworkSettings.Ports}}", self.name).stdout)
+        self.http = f"http://127.0.0.1:{ports['17170/tcp'][0]['HostPort']}"
+        self.ldap = int(ports["3890/tcp"][0]["HostPort"])
+        for _ in range(60):
+            try:
+                with urllib.request.urlopen(self.http + "/health", timeout=2):
+                    return
+            except (urllib.error.URLError, ConnectionError):
+                time.sleep(0.5)
+        raise RuntimeError("lldap did not become healthy")
+
+    def stop(self):
+        docker("rm", "--force", self.name, check=False)
+
+    def remove(self):
+        self.stop()
+        docker("volume", "rm", "--force", self.volume, check=False)
+
+    def login(self, user, password):
+        request = urllib.request.Request(self.http + "/auth/simple/login", method="POST",
+                                         data=json.dumps({"username": user, "password": password}).encode(),
+                                         headers={"Content-Type": "application/json"})
+        try:
+            with urllib.request.urlopen(request, timeout=10) as response:
+                return json.load(response)["token"]
+        except urllib.error.HTTPError:
+            return None
+
+    def graphql(self, token, query, variables=None):
+        request = urllib.request.Request(self.http + "/api/graphql", method="POST",
+                                         data=json.dumps({"query": query, "variables": variables or {}}).encode(),
+                                         headers={"Content-Type": "application/json", "Authorization": f"Bearer {token}"})
+        try:
+            with urllib.request.urlopen(request, timeout=10) as response:
+                body = json.load(response)
+        except urllib.error.HTTPError as error:
+            return None, f"http {error.code}"
+        return body.get("data"), (body.get("errors") or [{}])[0].get("message")
+
+    def ok(self, token, query, variables=None):
+        data, error = self.graphql(token, query, variables)
+        if error:
+            raise RuntimeError(f"graphql refused: {error}")
+        return data
+
+    def group_id(self, token, name):
+        groups = self.ok(token, "{ groups { id displayName } }")["groups"]
+        return next(g["id"] for g in groups if g["displayName"] == name)
+
+    def create_user(self, token, user, groups=()):
+        self.ok(token, "mutation($u: CreateUserInput!) { createUser(user: $u) { id } }",
+                {"u": {"id": user, "email": f"{user}@rehearsal.invalid", "displayName": user}})
+        for group in groups:
+            self.ok(token, "mutation($u: String!, $g: Int!) { addUserToGroup(userId: $u, groupId: $g) { ok } }",
+                    {"u": user, "g": self.group_id(token, group)})
+
+    def set_password(self, token, user, password):
+        docker("exec", "--env", "LLDAP_USER_PASSWORD", self.name, "/app/lldap_set_password",
+               "--base-url", "http://127.0.0.1:17170", "--token", token, "--username", user,
+               env={"LLDAP_USER_PASSWORD": password, "PATH": "/usr/bin:/bin"})
+
+    def groups_of(self, token, user):
+        data = self.ok(token, "query($u: String!) { user(userId: $u) { groups { displayName } } }", {"u": user})
+        return sorted(g["displayName"] for g in data["user"]["groups"])
+
+    def bind(self, user, password):
+        server = ldap3.Server("127.0.0.1", port=self.ldap, get_info=ldap3.NONE)
+        connection = ldap3.Connection(server, f"uid={user},ou=people,{BASE_DN}", password)
+        return connection if connection.bind() else None
+
+
+def rehearse(tag):
+    passwords = {name: secrets.token_urlsafe(24) for name in
+                 ("admin", "ld-admin", "alice", *READ_ONLY, "svc-provisioner")}
+    directory = Directory(tag)
+    result = {}
+    try:
+        # Production shape: `admin` is the configured super-admin, one human user, the baseline groups.
+        directory.start("admin", passwords["admin"])
+        admin = directory.login("admin", passwords["admin"])
+        for group in PRODUCTION_GROUPS:
+            directory.ok(admin, "mutation($n: String!) { createGroup(name: $n) { id } }", {"n": group})
+        directory.create_user(admin, "alice", ["net-kingdom-users"])
+        directory.set_password(admin, "alice", passwords["alice"])
+
+        # Migration step 1: introduce ld-admin and the service accounts while admin is still configured.
+        directory.create_user(admin, "ld-admin", ["lldap_admin"])
+        directory.set_password(admin, "ld-admin", passwords["ld-admin"])
+        for user in READ_ONLY:
+            directory.create_user(admin, user, ["lldap_strict_readonly"])
+            directory.set_password(admin, user, passwords[user])
+        directory.create_user(admin, "svc-provisioner", ["lldap_password_manager"])
+        directory.set_password(admin, "svc-provisioner", passwords["svc-provisioner"])
+
+        # Migration step 2: switch the configured super-admin to ld-admin and restart on the same data.
+        directory.stop()
+        directory.start("ld-admin", passwords["ld-admin"])
+        ld_admin = directory.login("ld-admin", passwords["ld-admin"])
+        admin = directory.login("admin", passwords["admin"])
+        result["ld_admin_introduced"] = {
+            "login": ld_admin is not None,
+            "groups": directory.groups_of(ld_admin, "ld-admin") if ld_admin else None,
+            "can_create_group": ld_admin is not None and directory.graphql(
+                ld_admin, 'mutation { createGroup(name: "rehearsal-probe") { id } }')[1] is None,
+        }
+        result["admin_after_switch"] = {
+            "login_with_unchanged_password": admin is not None,
+            "groups": directory.groups_of(ld_admin, "admin") if ld_admin else None,
+            "can_create_user": admin is not None and directory.graphql(
+                admin, "mutation($u: CreateUserInput!) { createUser(user: $u) { id } }",
+                {"u": {"id": "probe-by-admin", "email": "probe-by-admin@rehearsal.invalid"}})[1] is None,
+        }
+
+        # Read-only binds: read everything needed by Authelia, KeyCape and the privacyIDEA resolver; write nothing.
+        binds = {}
+        for user in READ_ONLY:
+            connection = directory.bind(user, passwords[user])
+            if connection is None:
+                binds[user] = {"bind": False}
+                continue
+            connection.search(f"ou=people,{BASE_DN}", "(objectClass=person)", attributes=["uid", "mail", "memberOf"])
+            people = sorted(str(e.uid) for e in connection.entries)
+            connection.search(f"ou=groups,{BASE_DN}", "(objectClass=groupOfUniqueNames)", attributes=["cn"])
+            groups = len(connection.entries)
+            write = connection.modify(f"uid=alice,ou=people,{BASE_DN}", {"mail": [(ldap3.MODIFY_REPLACE, ["x@rehearsal.invalid"])]})
+            connection.unbind()
+            binds[user] = {"bind": True, "people_visible": len(people), "sees_alice": "alice" in people,
+                           "groups_visible": groups, "write_refused": not write}
+        result["read_only_binds"] = binds
+
+        # Provisioner: what can lldap_password_manager do, and what does lldap_admin add?
+        provisioner = directory.login("svc-provisioner", passwords["svc-provisioner"])
+        probe = f"probe-{uuid.uuid4().hex[:8]}"
+        create_error = directory.graphql(provisioner, "mutation($u: CreateUserInput!) { createUser(user: $u) { id } }",
+                                         {"u": {"id": probe, "email": f"{probe}@rehearsal.invalid"}})[1]
+        users_group = directory.group_id(ld_admin, "net-kingdom-users")
+        member_error = directory.graphql(provisioner, "mutation($u: String!, $g: Int!) { addUserToGroup(userId: $u, groupId: $g) { ok } }",
+                                         {"u": "alice", "g": directory.group_id(ld_admin, "net-kingdom-admins")})[1]
+        password_ok = subprocess.run(
+            ["docker", "exec", "--env", "LLDAP_USER_PASSWORD", directory.name, "/app/lldap_set_password",
+             "--base-url", "http://127.0.0.1:17170", "--token", provisioner, "--username", "alice"],
+            capture_output=True, env={"LLDAP_USER_PASSWORD": secrets.token_urlsafe(24), "PATH": "/usr/bin:/bin"}).returncode == 0
+        admin_password_error = subprocess.run(
+            ["docker", "exec", "--env", "LLDAP_USER_PASSWORD", directory.name, "/app/lldap_set_password",
+             "--base-url", "http://127.0.0.1:17170", "--token", provisioner, "--username", "ld-admin"],
+            capture_output=True, env={"LLDAP_USER_PASSWORD": secrets.token_urlsafe(24), "PATH": "/usr/bin:/bin"}).returncode != 0
+        result["provisioner_password_manager"] = {
+            "login": provisioner is not None, "can_create_user": create_error is None,
+            "can_add_group_member": member_error is None, "can_set_user_password": password_ok,
+            "refused_admin_password": admin_password_error, "users_group_id_known": users_group > 0,
+        }
+
+        # SI-3: interactive sign-in of a service account to the LLDAP web login.
+        result["si3_interactive_login"] = {user: directory.login(user, passwords[user]) is not None for user in READ_ONLY}
+    finally:
+        directory.remove()
+    result["cleanup"] = docker("volume", "inspect", directory.volume, check=False).returncode != 0
+    return result
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--receipt", type=Path, required=True)
+    args = parser.parse_args(argv)
+    if args.receipt.exists():
+        print("refused: receipt exists", file=sys.stderr)
+        return 2
+    tag = uuid.uuid4().hex[:10]
+    findings = rehearse(tag)
+    receipt = {"schema": "netkingdom.lldap-bind-split-rehearsal.v1", "task": "NK-WP-0054-T05",
+               "observed_at": datetime.now(timezone.utc).isoformat(), "image": IMAGE,
+               "data": "synthetic", "production_contacted": False, "values_recorded": False,
+               "findings": findings}
+    args.receipt.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n")
+    print(json.dumps(findings, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())