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 <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:
parent
05a9aefaed
commit
028a7a97e7
2 changed files with 394 additions and 0 deletions
228
sso-mfa/k8s/keycape/directory-bind-rollout.py
Normal file
228
sso-mfa/k8s/keycape/directory-bind-rollout.py
Normal file
|
|
@ -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())
|
||||
166
tests/test_keycape_directory_bind_rollout.py
Normal file
166
tests/test_keycape_directory_bind_rollout.py
Normal file
|
|
@ -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
|
||||
Loading…
Add table
Add a link
Reference in a new issue