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

#!/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
#
# Options:
#   --note TEXT                 operator context recorded verbatim in the receipt
#   --predecessor-unavailable   the exposed predecessor cannot be produced; the
#                               denial bind is not attempted and the receipt
#                               records it as NOT-PROVEN

set -euo pipefail

MODE=""
NOTE=""
PREDECESSOR="required"
usage() {
    echo "Usage: $0 --check|--apply [--note TEXT] [--predecessor-unavailable]" >&2
    exit 2
}
while [[ $# -gt 0 ]]; do
    case "$1" in
        --check|--apply)
            if [[ -n "$MODE" ]]; then usage; fi
            MODE="$1"; shift ;;
        --note)
            # Never put a credential here: the receipt goes to a terminal and
            # is copied into the incident record.
            if [[ -z "${2:-}" ]]; then usage; fi
            NOTE="$2"; shift 2 ;;
        --predecessor-unavailable)
            # Typing a placeholder at the predecessor prompt also fails the
            # bind, and would be recorded as a passing denial proof. An absent
            # proof is honest; a fabricated one asserts a test that never ran.
            PREDECESSOR="unavailable"; shift ;;
        *)
            usage ;;
    esac
done
if [[ "$MODE" != "--apply" && "$MODE" != "--check" ]]; then
    usage
fi
# One bounded line: a receipt that breaks its own format is not evidence
# anyone can read back.
NOTE="${NOTE//$'\n'/ }"
NOTE="${NOTE//$'\r'/ }"
if (( ${#NOTE} > 200 )); then
    echo "ERROR: --note must be 200 characters or fewer." >&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}"
# A resolver write replaces the whole object, so every field the body omits is
# dropped. A resolver with these unset still resolves users, but the WebUI
# refuses to save or test it until they are filled in — so omitting them here
# silently un-repairs a resolver an operator fixed by hand.
LDAP_TIMEOUT="${LDAP_TIMEOUT:-5}"
LDAP_CACHE_TIMEOUT="${LDAP_CACHE_TIMEOUT:-120}"
LDAP_SIZELIMIT="${LDAP_SIZELIMIT:-500}"
for _v in LDAP_TIMEOUT LDAP_CACHE_TIMEOUT LDAP_SIZELIMIT; do
    if [[ ! "${!_v}" =~ ^[0-9]+$ ]]; then
        echo "ERROR: $_v must be a non-negative integer, got '${!_v}'." >&2
        exit 2
    fi
done
unset _v

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"
if [[ "$PREDECESSOR" == "required" ]]; then
    prompt_secret "exposed predecessor LLDAP password (denial proof only)" "$tmp/lldap-old"
else
    echo "  [INFO] predecessor unavailable: denial proof recorded as NOT-PROVEN." >&2
fi
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" "$PREDECESSOR" \
    "$LDAP_TIMEOUT" "$LDAP_CACHE_TIMEOUT" "$LDAP_SIZELIMIT" <<'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,
    predecessor, ldap_timeout, ldap_cache_timeout, ldap_sizelimit,
) = 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]:
    # Content-Type is only set on requests with a body — Werkzeug 3.x raises
    # BadRequest if Content-Type: application/json is sent on a bodyless GET,
    # and the rejection happens in front of privacyIDEA, so the reply is an HTML
    # error page rather than a JSON result. Same fix as bootstrap-realm.sh's
    # pi_api helper.
    headers = {}
    if payload is not 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")
    if predecessor == "unavailable":
        # Not a pass. The bind is not attempted, so this run claims nothing
        # about the predecessor's disposition.
        phase("predecessor-denial-not-proven")
    else:
        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,
        "TIMEOUT": int(ldap_timeout), "CACHE_TIMEOUT": int(ldap_cache_timeout),
        "SIZELIMIT": int(ldap_sizelimit),
    }
    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,
    )
    # privacyIDEA returns result.value as a list of user objects for /user/.
    # Accept the dict-with-"users" shape too: other endpoints use it, and a
    # lookup that guesses wrong raises AttributeError past the except clause.
    _value = (result or {}).get("result", {}).get("value", [])
    if isinstance(_value, dict):
        _value = _value.get("users", [])
    users = _value if isinstance(_value, list) else []
    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,
        AttributeError, TypeError, KeyError, ValueError) as exc:
    # A malformed reply must still produce a receipt naming the phase it died
    # in. A traceback tells the operator nothing about what was proven.
    raise SystemExit(1)
PY
then
    rc=0
else
    rc=$?
fi

phase_result="$(cat "$tmp/phase" 2>/dev/null || echo unknown)"
if [[ "$PREDECESSOR" == "required" ]]; then
    pred_text="predecessor denial"
else
    pred_text="predecessor denial=NOT-PROVEN"
fi
note_text=""
if [[ -n "$NOTE" ]]; then
    note_text="; note=$NOTE"
fi
cleanup
trap - EXIT INT TERM
if [[ "$rc" -ne 0 ]]; then
    echo "NK-WP-0033 receipt FAIL: phase=$phase_result; $pred_text; cleanup=PASS$note_text" >&2
    exit "$rc"
fi
if [[ -d "$tmp" ]]; then
    echo "NK-WP-0033 receipt FAIL: cleanup=FAIL$note_text" >&2
    exit 1
fi
if [[ "$MODE" == "--check" ]]; then
    echo "NK-WP-0033 receipt PASS: read-only resolver lookup, privacyIDEA MFA, $pred_text, readiness, health, cleanup=PASS$note_text"
else
    echo "NK-WP-0033 receipt PASS: resolver lookup, privacyIDEA MFA, $pred_text, readiness, health, cleanup=PASS$note_text"
fi