codex
80c3a76431
Refs CUST-WP-0068 Assistant: claude-code Assistant-Model: opus Assistant-Process: 2583210@bnt-lap001 Assistant-Session: f2bff2d5-e9b2-4338-92ca-10282a927006
228 lines
8.6 KiB
Bash
Executable file
228 lines
8.6 KiB
Bash
Executable file
#!/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 --check # read-only proof
|
|
# ./reconcile-lldap-resolver-live.sh --apply # resolver update + proof
|
|
|
|
set -euo pipefail
|
|
|
|
MODE="${1:-}"
|
|
if [[ "$MODE" != "--apply" && "$MODE" != "--check" || "${2:-}" != "" ]]; then
|
|
echo "Usage: $0 --check|--apply" >&2
|
|
exit 2
|
|
fi
|
|
if [[ ! -t 0 ]]; then
|
|
echo "ERROR: --check/--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 phase; 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" "$MODE" "$tmp/phase" <<'PY'
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
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, mode, phase_path,
|
|
) = sys.argv[1:]
|
|
|
|
def phase(name: str) -> None:
|
|
Path(phase_path).write_text(name, encoding="ascii")
|
|
|
|
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:
|
|
phase("preflight")
|
|
check_k8s_ready()
|
|
check_health()
|
|
|
|
phase("privacyidea-auth")
|
|
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")
|
|
|
|
# Prove the provider-approved replacement and predecessor disposition
|
|
# before any resolver mutation is attempted.
|
|
phase("replacement-lldap-auth")
|
|
new_status, new_authenticated = lldap_login(secret(new_path))
|
|
if new_status != 200 or not new_authenticated:
|
|
raise RuntimeError("replacement LLDAP authentication failed")
|
|
phase("predecessor-denial")
|
|
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")
|
|
|
|
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,
|
|
}
|
|
if mode == "--apply":
|
|
phase("resolver-update")
|
|
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")
|
|
else:
|
|
phase("resolver-read-only-check")
|
|
|
|
phase("resolver-lookup")
|
|
status, result = request(
|
|
pi_url + f"/user/?realm={mfa_realm}&username={urllib.parse.quote(mfa_user)}",
|
|
token=pi_token,
|
|
)
|
|
users = (result or {}).get("result", {}).get("value", {}).get("users", [])
|
|
if status != 200:
|
|
phase(f"resolver-lookup-http-{status}")
|
|
raise RuntimeError("replacement resolver lookup failed")
|
|
if not users:
|
|
phase("resolver-lookup-empty")
|
|
raise RuntimeError("replacement resolver lookup failed")
|
|
|
|
phase("privacyidea-mfa")
|
|
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")
|
|
|
|
phase("postflight")
|
|
check_k8s_ready()
|
|
check_health()
|
|
except (OSError, RuntimeError, subprocess.SubprocessError) as exc:
|
|
raise SystemExit(1)
|
|
PY
|
|
then
|
|
rc=0
|
|
else
|
|
rc=$?
|
|
fi
|
|
|
|
phase_result="$(cat "$tmp/phase" 2>/dev/null || echo unknown)"
|
|
cleanup
|
|
trap - EXIT INT TERM
|
|
if [[ "$rc" -ne 0 ]]; then
|
|
echo "NK-WP-0033 receipt FAIL: phase=$phase_result; cleanup=PASS" >&2
|
|
exit "$rc"
|
|
fi
|
|
if [[ -d "$tmp" ]]; then
|
|
echo "NK-WP-0033 receipt FAIL: cleanup=FAIL" >&2
|
|
exit 1
|
|
fi
|
|
if [[ "$MODE" == "--check" ]]; then
|
|
echo "NK-WP-0033 receipt PASS: read-only resolver lookup, privacyIDEA MFA, predecessor denial, readiness, health, cleanup=PASS"
|
|
else
|
|
echo "NK-WP-0033 receipt PASS: resolver lookup, privacyIDEA MFA, predecessor denial, readiness, health, cleanup=PASS"
|
|
fi
|