diff --git a/sso-mfa/k8s/lldap/bind-split-rehearsal.py b/sso-mfa/k8s/lldap/bind-split-rehearsal.py index 252c76d..5bab891 100644 --- a/sso-mfa/k8s/lldap/bind-split-rehearsal.py +++ b/sso-mfa/k8s/lldap/bind-split-rehearsal.py @@ -27,7 +27,7 @@ import time import urllib.error import urllib.request import uuid -from datetime import datetime, timezone +from datetime import UTC, datetime from pathlib import Path import ldap3 @@ -39,7 +39,9 @@ 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) + done = subprocess.run( + ["docker", *args], check=False, capture_output=True, text=True, env=env + ) if check and done.returncode: raise RuntimeError(f"docker {args[0]} failed") return done @@ -54,13 +56,34 @@ class Directory: 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) + 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): @@ -79,9 +102,12 @@ class Directory: 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"}) + 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"] @@ -89,9 +115,15 @@ class Directory: 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}"}) + 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) @@ -110,30 +142,61 @@ class Directory: 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}}) + 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)}) + 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"}) + 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}) + 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) + 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")} + passwords = { + name: secrets.token_urlsafe(24) + for name in ("admin", "ld-admin", "alice", *READ_ONLY, "svc-provisioner") + } directory = Directory(tag) result = {} try: @@ -141,7 +204,11 @@ def rehearse(tag): 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.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"]) @@ -162,15 +229,27 @@ def rehearse(tag): 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, + "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, + "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. @@ -180,43 +259,115 @@ def rehearse(tag): if connection is None: binds[user] = {"bind": False} continue - connection.search(f"ou=people,{BASE_DN}", "(objectClass=person)", attributes=["uid", "mail", "memberOf"]) + 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"]) + 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"])]}) + 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} + 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] + 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 + 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", + ], + check=False, + 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", + ], + check=False, + 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, + "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} + 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 + result["cleanup"] = ( + docker("volume", "inspect", directory.volume, check=False).returncode != 0 + ) return result @@ -229,10 +380,16 @@ def main(argv=None): 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} + receipt = { + "schema": "netkingdom.lldap-bind-split-rehearsal.v1", + "task": "NK-WP-0054-T05", + "observed_at": datetime.now(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 diff --git a/workplans/NK-WP-0054-service-identity-policy.md b/workplans/NK-WP-0054-service-identity-policy.md index 4f89c0c..3a23768 100644 --- a/workplans/NK-WP-0054-service-identity-policy.md +++ b/workplans/NK-WP-0054-service-identity-policy.md @@ -265,6 +265,27 @@ then the founder runs the live steps as soon as the rehearsal passes. KeyCape's as the directory admin from source (key-cape `a1ab8ff`, KEY-WP-0045-T03). The rotation lane is `lldap-directory-bind` / `lldap-service-binds` in `tools/rotation-lanes/lanes.yaml`. +**2026-10-28 synthetic rehearsal (phase 1 of 4):** `sso-mfa/k8s/lldap/bind-split-rehearsal.py` (agent-run, +pinned LLDAP v0.6.2, generated passwords, 127.0.0.1 only, removed afterwards; +[receipt](../docs/evidence/2026-10-28-nk54-bind-split-rehearsal/synthetic-rehearsal.json)). Proven: +`ld-admin` can be introduced into a directory whose super-admin is `admin` and made the configured +super-admin by restart on the same data; `admin` keeps its rights through `lldap_admin` with its +password unchanged; `lldap_strict_readonly` binds read all users and groups and cannot write. +Findings: (1) `lldap_password_manager` cannot create users or add members, so the provisioner's +narrowest working group is `lldap_admin` (own account, but super-admin rights: needs an exception or +a provisioner redesign); (2) SI-3: every `svc-*` account can sign in to the LLDAP web UI; the UI is +IP-restricted, and Authelia's user filter must require `net-kingdom-users` membership; (3) production +`deployment.yaml` runs the unpinned `lldap/lldap:stable`, while the rehearsal and the restore tools pin +v0.6.2: pin production before the live run. + +Remaining phases, each its own budgeted task: **(2)** consumer switch in source (Authelia configmap user +and users filter, KeyCape `create-secrets.py` bind, privacyIDEA `realm_bootstrap.py` resolver, +identity-provisioner deployment, LLDAP image pin) with a synthetic Authelia check of the filter; +**(3)** the guarded live helper (plan/apply/verify, value-free receipts, passwords only by hidden +prompt or fd) plus the backup profile and the restore adapters `isolated-lldap.py`, +`isolated-authelia.py`, `isolated-privacyidea-mfa.py`; **(4)** the founder runbook with filled-in +commands, a founder-run rehearsal on a restored backup, then the live run. + ## Disposition of per-service admin accounts ```task