NK-WP-0033 add attended resolver reconciliation receipt

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02929-244b-7391-b933-c04010e8eedb

sso-mfa/k8s/privacyidea/reconcile-lldap-resolver-live.sh

@@ -0,0 +1,198 @@
+#!/usr/bin/env bash
+# reconcile-lldap-resolver-live.sh — one-command attended resolver cutover.
+#
+# Performs the narrow privacyIDEA lldap-coulomb update and its safe proof set:
+# replacement LLDAP authentication, resolver lookup, one privacyIDEA MFA check,
+# predecessor LLDAP denial, health/readiness, and cleanup. It never reads a
+# Kubernetes Secret and never prints a credential, token, response body, or
+# manifest.
+#
+# Usage:
+#   ./reconcile-lldap-resolver-live.sh --apply
+
+set -euo pipefail
+
+if [[ "${1:-}" != "--apply" || "${2:-}" != "" ]]; then
+    echo "Usage: $0 --apply" >&2
+    exit 2
+fi
+if [[ ! -t 0 ]]; then
+    echo "ERROR: --apply requires an interactive terminal." >&2
+    exit 2
+fi
+
+PI_URL="${PI_URL:-https://pink.coulomb.social}"
+LLDAP_AUTH_URL="${LLDAP_AUTH_URL:-https://lldap.coulomb.social/auth/simple/login}"
+LLDAP_URL="${LLDAP_URL:-ldap://lldap.sso.svc.cluster.local:3890}"
+LLDAP_BASE_DN="${LLDAP_BASE_DN:-dc=netkingdom,dc=local}"
+LLDAP_BIND_DN="${LLDAP_BIND_DN:-uid=admin,ou=people,dc=netkingdom,dc=local}"
+RESOLVER_NAME="${RESOLVER_NAME:-lldap-coulomb}"
+MFA_USER="${MFA_USER:-platform-root}"
+MFA_REALM="${MFA_REALM:-coulomb}"
+KEYCAPE_DISCOVERY_URL="${KEYCAPE_DISCOVERY_URL:-https://kc.coulomb.social/.well-known/openid-configuration}"
+
+tmp="$(mktemp -d)"
+chmod 700 "$tmp"
+cleanup() {
+    for file in pi-admin lldap-new lldap-old otp; do
+        if [[ -f "$tmp/$file" ]]; then
+            shred -u "$tmp/$file" 2>/dev/null || rm -f "$tmp/$file"
+        fi
+    done
+    rmdir "$tmp" 2>/dev/null || true
+}
+cleanup_ok=0
+trap cleanup EXIT INT TERM
+
+prompt_secret() {
+    local label="$1" target="$2"
+    printf '%s: ' "$label" >&2
+    IFS= read -r -s value
+    printf '\n' >&2
+    if [[ -z "$value" ]]; then
+        echo "ERROR: empty protected input." >&2
+        exit 2
+    fi
+    printf '%s' "$value" > "$target"
+    unset value
+    chmod 600 "$target"
+}
+
+prompt_secret "privacyIDEA pi-admin password" "$tmp/pi-admin"
+prompt_secret "replacement LLDAP bind/admin password" "$tmp/lldap-new"
+prompt_secret "exposed predecessor LLDAP password (denial proof only)" "$tmp/lldap-old"
+prompt_secret "one-time MFA code for $MFA_USER@$MFA_REALM" "$tmp/otp"
+
+if python3 - "$tmp/pi-admin" "$tmp/lldap-new" "$tmp/lldap-old" "$tmp/otp" \
+    "$PI_URL" "$LLDAP_AUTH_URL" "$LLDAP_URL" "$LLDAP_BASE_DN" "$LLDAP_BIND_DN" \
+    "$RESOLVER_NAME" "$MFA_USER" "$MFA_REALM" "$KEYCAPE_DISCOVERY_URL" <<'PY'
+import json
+import subprocess
+import sys
+import urllib.error
+import urllib.request
+from pathlib import Path
+
+(
+    pi_path, new_path, old_path, otp_path, pi_url, lldap_auth_url, lldap_url,
+    base_dn, bind_dn, resolver_name, mfa_user, mfa_realm, discovery_url,
+) = sys.argv[1:]
+
+def secret(path: str) -> str:
+    value = Path(path).read_text(encoding="utf-8")
+    if not value:
+        raise RuntimeError("empty protected input")
+    return value
+
+def request(url: str, payload: dict | None = None, token: str | None = None) -> tuple[int, dict | None]:
+    headers = {"Content-Type": "application/json"}
+    if token:
+        headers["Authorization"] = token
+    data = json.dumps(payload).encode("utf-8") if payload is not None else None
+    req = urllib.request.Request(url, data=data, headers=headers, method="POST" if payload is not None else "GET")
+    try:
+        with urllib.request.urlopen(req, timeout=20) as response:
+            status = response.status
+            body = response.read()
+        if not body:
+            return status, None
+        try:
+            return status, json.loads(body)
+        except json.JSONDecodeError:
+            return status, None
+    except urllib.error.HTTPError as exc:
+        return exc.code, None
+    except (urllib.error.URLError, TimeoutError):
+        return 0, None
+
+def check_k8s_ready() -> None:
+    workloads = (("sso", "lldap"), ("mfa", "privacyidea"), ("sso", "keycape"),
+                 ("sso", "authelia"), ("sso", "identity-provisioner"))
+    for namespace, name in workloads:
+        proc = subprocess.run(
+            ["kubectl", "get", "deployment", name, "-n", namespace,
+             "-o", "jsonpath={.status.readyReplicas}/{.status.replicas}"],
+            capture_output=True, text=True, timeout=20,
+        )
+        if proc.returncode != 0 or proc.stdout.strip() != "1/1":
+            raise RuntimeError(f"readiness failed: {namespace}/{name}")
+
+def check_health() -> None:
+    # privacyIDEA intentionally exposes no unauthenticated HTTP health route;
+    # /token/ is documented by the deployment as a safe availability probe.
+    for label, url, accepted in (
+        ("privacyidea", pi_url + "/token/", {200, 401, 403}),
+        ("keycape", discovery_url, set(range(200, 400))),
+    ):
+        status, _ = request(url)
+        if status not in accepted:
+            raise RuntimeError(f"health failed: {label}")
+
+def lldap_login(password: str) -> tuple[int, bool]:
+    status, body = request(lldap_auth_url, {"username": "admin", "password": password})
+    return status, status == 200 and isinstance(body, dict) and bool(body.get("token"))
+
+try:
+    check_k8s_ready()
+    check_health()
+
+    status, auth = request(pi_url + "/auth", {"username": "pi-admin", "password": secret(pi_path)})
+    pi_token = str((auth or {}).get("result", {}).get("value", {}).get("token", ""))
+    if status != 200 or not pi_token:
+        raise RuntimeError("privacyIDEA authentication failed")
+
+    resolver_body = {
+        "type": "ldapresolver", "LDAPURI": lldap_url, "BINDDN": bind_dn,
+        "BINDPW": secret(new_path), "LDAPBASE": base_dn,
+        "LOGINNAMEATTRIBUTE": "uid", "LDAPSEARCHFILTER": "(objectClass=inetOrgPerson)",
+        "LDAPFILTER": "(&(objectClass=inetOrgPerson)(uid=%s))",
+        "USERINFO": json.dumps({"username": "uid", "phone": "telephoneNumber", "mobile": "mobile", "email": "mail", "surname": "sn", "givenname": "givenName"}),
+        "UIDTYPE": "uid", "NOREFERRALS": True, "NOSCHEMAS": True,
+    }
+    status, result = request(pi_url + "/resolver/" + resolver_name, resolver_body, pi_token)
+    if status != 200 or (result or {}).get("result", {}).get("status") not in (True, "true", "True"):
+        raise RuntimeError("resolver update failed")
+
+    status, result = request(pi_url + f"/user/?realm={mfa_realm}&pagesize=1", token=pi_token)
+    users = (result or {}).get("result", {}).get("value", {}).get("users", [])
+    if status != 200 or not users:
+        raise RuntimeError("replacement resolver lookup failed")
+
+    status, result = request(
+        pi_url + "/validate/check",
+        {"user": mfa_user, "realm": mfa_realm, "pass": secret(otp_path)},
+        pi_token,
+    )
+    if status != 200 or (result or {}).get("result", {}).get("value") is not True:
+        raise RuntimeError("replacement MFA validation failed")
+
+    new_status, new_authenticated = lldap_login(secret(new_path))
+    if new_status != 200 or not new_authenticated:
+        raise RuntimeError("replacement LLDAP authentication failed")
+    old_status, old_authenticated = lldap_login(secret(old_path))
+    if old_status not in (401, 403) or old_authenticated:
+        raise RuntimeError("predecessor LLDAP authentication was not denied")
+
+    check_k8s_ready()
+    check_health()
+except (OSError, RuntimeError, subprocess.SubprocessError) as exc:
+    print("NK-WP-0033 receipt FAIL: proof checks did not pass", file=sys.stderr)
+    raise SystemExit(1)
+PY
+then
+    rc=0
+else
+    rc=$?
+fi
+
+cleanup
+trap - EXIT INT TERM
+if [[ "$rc" -ne 0 ]]; then
+    echo "NK-WP-0033 receipt FAIL: proof checks did not pass; cleanup=PASS" >&2
+    exit "$rc"
+fi
+if [[ -d "$tmp" ]]; then
+    echo "NK-WP-0033 receipt FAIL: cleanup=FAIL" >&2
+    exit 1
+fi
+echo "NK-WP-0033 receipt PASS: resolver lookup, privacyIDEA MFA, predecessor denial, readiness, health, cleanup=PASS"