From 028a7a97e7d74282ca4a71c5e8658705aa8cb397 Mon Sep 17 00:00:00 2001 From: tegwick Date: Thu, 29 Oct 2026 22:16:59 +0100 Subject: [PATCH] NK-WP-0054-T05 phase 4: KeyCape directory-bind rollout (switch to svc-keycape, revert) Edits only lldap.bindDN/bindPW in sso/keycape-config, modelled on openbao-signin-diagnosis-rollout.py: inspect/dry-run/apply, uid and resourceVersion test patch, re-parse check, contained errors. 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 --- sso-mfa/k8s/keycape/directory-bind-rollout.py | 228 ++++++++++++++++++ tests/test_keycape_directory_bind_rollout.py | 166 +++++++++++++ 2 files changed, 394 insertions(+) create mode 100644 sso-mfa/k8s/keycape/directory-bind-rollout.py create mode 100644 tests/test_keycape_directory_bind_rollout.py diff --git a/sso-mfa/k8s/keycape/directory-bind-rollout.py b/sso-mfa/k8s/keycape/directory-bind-rollout.py new file mode 100644 index 0000000..1baad84 --- /dev/null +++ b/sso-mfa/k8s/keycape/directory-bind-rollout.py @@ -0,0 +1,228 @@ +#!/usr/bin/env python3 +"""Switch KeyCape's LLDAP bind from the human admin to svc-keycape (NK-WP-0054-T05). + +Edits exactly two scalars in `lldap` of `config.yaml` in sso/keycape-config: `bindDN` and `bindPW`. +`--operation switch` sets `uid=svc-keycape,...` and the password from the private custody copy +(`lldap/secrets.env`, key LLDAP_SVC_KEYCAPE_PASS); `--operation revert` sets `uid=admin,...` back and +takes the admin password at a hidden prompt. The signing key, every client and every other config byte +are preserved and checked by re-parsing. + +Modes default to non-mutating, as in openbao-signin-diagnosis-rollout.py: `inspect` reads and plans, +`dry-run` sends a server-side dry-run patch, `apply` writes and needs `--attended` plus the observed +uid/resourceVersion. KeyCape reads the change after `kubectl -n sso rollout restart deploy/keycape`. +No Secret value, parser message or Kubernetes error text leaves the process. +""" + +from __future__ import annotations + +import argparse +import base64 +import copy +import getpass +import importlib.util +import json +from pathlib import Path + +import yaml + +ROOT = Path(__file__).resolve().parent + + +def module(name, path): + spec = importlib.util.spec_from_file_location(name, path) + result = importlib.util.module_from_spec(spec) + spec.loader.exec_module(result) + return result + + +# Reuse the owner lane's contained kubectl, refusal type and cluster pin. +lane = module("portal_rollout", ROOT / "portal-client-rollout.py") +pin = lane.pin +Refused = lane.Refused +require = lane.require +custody = module( + "keycape_bind_custody", ROOT.parents[1] / "bootstrap/apply-service-secrets.py" +) + +SERVICE_DN = "uid=svc-keycape,ou=people,dc=netkingdom,dc=local" +HUMAN_DN = "uid=admin,ou=people,dc=netkingdom,dc=local" + + +def service_password(root): + with custody.custody(Path(root), ["lldap"]) as directories: + return custody.literal( + custody.read_env(directories["lldap"]), "LLDAP_SVC_KEYCAPE_PASS" + ) + + +def _lldap(secret): + raw = base64.b64decode(secret["data"]["config.yaml"], validate=True).decode("utf-8") + if any( + isinstance(t, (yaml.tokens.AnchorToken, yaml.tokens.AliasToken)) + for t in yaml.scan(raw) + ): + raise Refused("yaml_alias_or_anchor") + config = yaml.load(raw, Loader=pin.UniqueLoader) + require( + isinstance(config, dict) and isinstance(config.get("lldap"), dict), + "lldap_mapping_required", + ) + node = next(v for k, v in yaml.compose(raw).value if k.value == "lldap") + fields = {k.value: v for k, v in node.value} + for name in ("bindDN", "bindPW"): + require( + isinstance(fields.get(name), yaml.nodes.ScalarNode), "bind_scalars_required" + ) + return raw, config, fields + + +def replace(secret, dn, password): + """Return (base64 config, changed, prior) with only bindDN and bindPW replaced.""" + raw, config, fields = _lldap(secret) + current = config["lldap"]["bindDN"] + prior = ( + "service" + if current == SERVICE_DN + else "human" + if current == HUMAN_DN + else "other" + ) + require(prior != "other", "unexpected_bind_dn_requires_reconciliation") + if current == dn and config["lldap"]["bindPW"] == password: + return secret["data"]["config.yaml"], False, prior + # Replace later spans first so earlier indexes stay valid. + spans = sorted( + ((fields["bindDN"], dn), (fields["bindPW"], password)), + key=lambda p: p[0].start_mark.index, + reverse=True, + ) + updated = raw + for scalar, value in spans: + updated = ( + updated[: scalar.start_mark.index] + + json.dumps(value) + + updated[scalar.end_mark.index :] + ) + expected = copy.deepcopy(config) + expected["lldap"].update(bindDN=dn, bindPW=password) + require( + yaml.load(updated, Loader=pin.UniqueLoader) == expected, + "unrelated_configuration_changed", + ) + return base64.b64encode(updated.encode()).decode(), True, prior + + +def run(args, kubectl=None, prompt=getpass.getpass): + kubectl = kubectl or lane.kubectl + cluster = kubectl(["get", "namespace", "kube-system", "-o", "json"]) + require(cluster["metadata"]["uid"] == lane.CLUSTER_UID, "cluster_identity_changed") + before = kubectl(["-n", "sso", "get", "secret", "keycape-config", "-o", "json"]) + require(not before["metadata"].get("ownerReferences"), "controller_owned_secret") + metadata = pin.safe_metadata(before) + if args.operation == "switch": + dn, password = SERVICE_DN, service_password(args.custody_root) + else: + dn, password = ( + HUMAN_DN, + prompt("Human admin password for the KeyCape bind rollback: "), + ) + require(bool(password) and "\n" not in password, "password_required") + encoded, changed, prior = replace(before, dn, password) + receipt = { + "schema": "netkingdom.keycape-directory-bind.v1", + "task": "NK-WP-0054-T05", + "target": "sso/keycape-config", + "before": metadata, + "operation": args.operation, + "prior_bind": prior, + "change_needed": changed, + "mode": args.mode, + "values_emitted": False, + "changed": False, + } + if args.mode == "inspect": + return receipt + if args.mode == "apply": + require(args.attended, "attended_operator_required") + require( + metadata + == { + "uid": args.expected_uid, + "resource_version": args.expected_resource_version, + }, + "observed_revision_changed", + ) + if changed: + patch = [ + {"op": "test", "path": "/metadata/uid", "value": metadata["uid"]}, + { + "op": "test", + "path": "/metadata/resourceVersion", + "value": metadata["resource_version"], + }, + {"op": "replace", "path": "/data/config.yaml", "value": encoded}, + ] + command = [ + "-n", + "sso", + "patch", + "secret", + "keycape-config", + "--type=json", + "--patch-file=/dev/stdin", + "-o", + "json", + ] + if args.mode == "dry-run": + command += ["--dry-run=server"] + result = kubectl(command, patch) + require( + result["data"] == dict(before["data"], **{"config.yaml": encoded}), + "patch_response_mismatch", + ) + if args.mode == "apply": + after = kubectl(["-n", "sso", "get", "secret", "keycape-config", "-o", "json"]) + require( + after["metadata"]["uid"] == metadata["uid"] + and after["data"] == dict(before["data"], **{"config.yaml": encoded}), + "readback_mismatch_stop_without_stale_replay", + ) + receipt.update( + changed=changed, after=pin.safe_metadata(after), restart_required=changed + ) + receipt.update( + unrelated_config_bytes_preserved=True, other_secret_data_unchanged=True + ) + return receipt + + +def main(argv=None, kubectl=None, prompt=getpass.getpass): + parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) + parser.add_argument( + "mode", nargs="?", default="inspect", choices=["inspect", "dry-run", "apply"] + ) + parser.add_argument("--operation", default="switch", choices=["switch", "revert"]) + parser.add_argument( + "--custody-root", help="private custody copy; required for switch" + ) + parser.add_argument("--expected-uid") + parser.add_argument("--expected-resource-version") + parser.add_argument("--attended", action="store_true", help="required for apply") + args = parser.parse_args(argv) + try: + if args.operation == "switch": + require(bool(args.custody_root), "custody_root_required") + print(json.dumps(run(args, kubectl, prompt), sort_keys=True)) + return 0 + except (Refused, custody.Refused) as error: + print(json.dumps({"status": "refused", "reason": str(error) or "refused"})) + except pin.IssuerPinError as error: + print(json.dumps({"status": "refused", "reason": str(error)})) + except Exception: # noqa: BLE001 - contain any error that could carry Secret material + # Parser and Kubernetes errors can contain full Secret material. + print(json.dumps({"status": "failed", "reason": "contained_operation_failed"})) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_keycape_directory_bind_rollout.py b/tests/test_keycape_directory_bind_rollout.py new file mode 100644 index 0000000..7ecdb99 --- /dev/null +++ b/tests/test_keycape_directory_bind_rollout.py @@ -0,0 +1,166 @@ +"""Synthetic tests for the KeyCape LLDAP bind switch (NK-WP-0054-T05); no cluster, no real values.""" + +import base64 +import copy +import importlib.util +import json +import types +from pathlib import Path + +import pytest +import yaml + +ROOT = Path(__file__).resolve().parents[1] +spec = importlib.util.spec_from_file_location( + "bind_rollout", ROOT / "sso-mfa/k8s/keycape/directory-bind-rollout.py" +) +m = importlib.util.module_from_spec(spec) +spec.loader.exec_module(m) + +SERVICE_PASSWORD = "S" * 20 + "0123456789abcdefghij" +HUMAN_PASSWORD = "synthetic-human-password" +CONFIG = """# synthetic KeyCape config +issuer: https://kc.example.test +signingKey: | + -----BEGIN SYNTHETIC----- + abc +lldap: + url: ldap://lldap.sso.svc.cluster.local:3890 + bindDN: uid=admin,ou=people,dc=netkingdom,dc=local + bindPW: old-shared-password + baseDN: dc=netkingdom,dc=local +authelia: + issuer: https://auth.coulomb.social + clientSecret: synthetic-client +clients: + - clientId: one +""" + + +def secret(config=CONFIG, version="41"): + return { + "metadata": { + "uid": "11111111-2222-3333-4444-555555555555", + "resourceVersion": version, + }, + "data": { + "config.yaml": base64.b64encode(config.encode()).decode(), + "other": "b3RoZXI=", + }, + } + + +class Cluster: + def __init__(self, obj): + self.obj, self.calls = obj, [] + + def __call__(self, args, payload=None): + self.calls.append((args, payload)) + if args[:2] == ["get", "namespace"]: + return {"metadata": {"uid": m.lane.CLUSTER_UID}} + if "get" in args: + return copy.deepcopy(self.obj) + assert payload[0]["value"] == self.obj["metadata"]["uid"] + assert payload[1]["value"] == self.obj["metadata"]["resourceVersion"] + result = copy.deepcopy(self.obj) + result["data"]["config.yaml"] = payload[2]["value"] + if "--dry-run=server" not in args: + self.obj = dict( + result, metadata=dict(result["metadata"], resourceVersion="42") + ) + return result + + +def args(mode="inspect", operation="switch", attended=False, version="41"): + return types.SimpleNamespace( + mode=mode, + operation=operation, + custody_root="unused", + attended=attended, + expected_uid="11111111-2222-3333-4444-555555555555", + expected_resource_version=version, + ) + + +@pytest.fixture(autouse=True) +def custody(monkeypatch): + monkeypatch.setattr(m, "service_password", lambda root: SERVICE_PASSWORD) + + +def config_of(obj): + return base64.b64decode(obj["data"]["config.yaml"]).decode() + + +def test_switch_changes_exactly_the_two_bind_scalars(): + cluster = Cluster(secret()) + receipt = m.run(args("apply", attended=True), cluster) + assert ( + receipt["changed"] + and receipt["prior_bind"] == "human" + and receipt["restart_required"] + ) + after = config_of(cluster.obj) + expected = yaml.safe_load(CONFIG) + expected["lldap"].update(bindDN=m.SERVICE_DN, bindPW=SERVICE_PASSWORD) + assert yaml.safe_load(after) == expected + assert after.replace(json.dumps(m.SERVICE_DN), "X").replace( + json.dumps(SERVICE_PASSWORD), "Y" + ) == CONFIG.replace("uid=admin,ou=people,dc=netkingdom,dc=local", "X").replace( + "old-shared-password", "Y" + ) + assert cluster.obj["data"]["other"] == "b3RoZXI=" + assert SERVICE_PASSWORD not in json.dumps( + receipt + ) and "old-shared-password" not in json.dumps(receipt) + + +def test_inspect_and_dry_run_write_nothing(): + cluster = Cluster(secret()) + assert m.run(args("inspect"), cluster)["change_needed"] + m.run(args("dry-run"), cluster) + assert config_of(cluster.obj) == CONFIG + + +def test_apply_needs_attended_and_the_observed_revision(): + with pytest.raises(m.Refused, match="attended"): + m.run(args("apply"), Cluster(secret())) + with pytest.raises(m.Refused, match="observed_revision_changed"): + m.run(args("apply", attended=True, version="40"), Cluster(secret())) + + +def test_second_switch_is_a_noop_and_revert_restores_the_human_bind(): + cluster = Cluster(secret()) + m.run(args("apply", attended=True), cluster) + again = m.run(args("apply", attended=True, version="42"), cluster) + assert not again["change_needed"] and again["prior_bind"] == "service" + reverted = m.run( + args("apply", "revert", attended=True, version="42"), + cluster, + prompt=lambda _: HUMAN_PASSWORD, + ) + assert reverted["changed"] + assert yaml.safe_load(config_of(cluster.obj))["lldap"] == dict( + yaml.safe_load(CONFIG)["lldap"], bindPW=HUMAN_PASSWORD + ) + + +@pytest.mark.parametrize( + "change", + [ + lambda c: c.replace("uid=admin,", "uid=someone,"), + lambda c: c.replace(" bindPW: old-shared-password\n", ""), + lambda c: c.replace("lldap:\n", "lldap: &a\n"), + ], +) +def test_unexpected_shapes_are_refused(change): + with pytest.raises((m.Refused, m.pin.IssuerPinError)): + m.run(args("inspect"), Cluster(secret(change(CONFIG)))) + + +def test_main_contains_failures_without_values(capsys): + def broken(args, payload=None): + raise RuntimeError("Secret data: old-shared-password") + + assert m.main(["inspect", "--custody-root", "x"], kubectl=broken) == 1 + out = capsys.readouterr().out + assert "contained_operation_failed" in out and "old-shared-password" not in out