From 84cc7bf13cbea7a8fdba3e7859c1cb7516ec728e Mon Sep 17 00:00:00 2001 From: tegwick Date: Wed, 28 Oct 2026 18:59:49 +0100 Subject: [PATCH] NK-WP-0054-T05: synthetic bind-split rehearsal on pinned LLDAP (phase 1 of 4) Co-Authored-By: Claude Opus 5.5 Assistant: claude-code Assistant-Model: opus Assistant-Process: 308701@bnt-lap001 Assistant-Session: 5548e447-444c-4dcc-bed1-05c16eebfe56 --- .../synthetic-rehearsal.json | 62 +++++ sso-mfa/k8s/lldap/bind-split-rehearsal.py | 242 ++++++++++++++++++ 2 files changed, 304 insertions(+) create mode 100644 docs/evidence/2026-10-28-nk54-bind-split-rehearsal/synthetic-rehearsal.json create mode 100644 sso-mfa/k8s/lldap/bind-split-rehearsal.py diff --git a/docs/evidence/2026-10-28-nk54-bind-split-rehearsal/synthetic-rehearsal.json b/docs/evidence/2026-10-28-nk54-bind-split-rehearsal/synthetic-rehearsal.json new file mode 100644 index 0000000..be68252 --- /dev/null +++ b/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 +} diff --git a/sso-mfa/k8s/lldap/bind-split-rehearsal.py b/sso-mfa/k8s/lldap/bind-split-rehearsal.py new file mode 100644 index 0000000..252c76d --- /dev/null +++ b/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())