NK-WP-0054-T05: rehearsal findings and remaining phases recorded; lint fixes
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

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
This commit is contained in:
tegwick 2026-10-28 19:01:40 +01:00
parent 84cc7bf13c
commit 3a83fed2c6
2 changed files with 236 additions and 58 deletions

View file

@ -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

View file

@ -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