generated from coulomb/repo-seed
234 lines
6.9 KiB
Bash
234 lines
6.9 KiB
Bash
#!/usr/bin/env bash
|
|
# refresh-pi-token-live.sh — refresh KeyCape's privacyIDEA admin token.
|
|
#
|
|
# This is an attended repair for "mfa check error" when the privacyIDEA admin
|
|
# JWT embedded in keycape-config has expired. It prompts for the pi-admin
|
|
# password, fetches a fresh token inside the privacyIDEA pod, updates the
|
|
# keycape-pi-token Secret, patches keycape-config without printing secrets, and
|
|
# restarts KeyCape.
|
|
#
|
|
# Usage:
|
|
# bash refresh-pi-token-live.sh [username]
|
|
#
|
|
# Optional environment:
|
|
# KUBECTL=/path/to/kubectl
|
|
# KEYCAPE_PI_REALM=coulomb|netkingdom
|
|
|
|
set -euo pipefail
|
|
|
|
USERNAME="${1:-platform-root}"
|
|
KUBECTL="${KUBECTL:-kubectl}"
|
|
SSO_NAMESPACE="${SSO_NAMESPACE:-sso}"
|
|
MFA_NAMESPACE="${MFA_NAMESPACE:-mfa}"
|
|
KEYCAPE_DEPLOYMENT="${KEYCAPE_DEPLOYMENT:-keycape}"
|
|
KEYCAPE_SECRET="${KEYCAPE_SECRET:-keycape-config}"
|
|
KEYCAPE_TOKEN_SECRET="${KEYCAPE_TOKEN_SECRET:-keycape-pi-token}"
|
|
|
|
if [[ -r /dev/tty ]]; then
|
|
printf "privacyIDEA pi-admin password: " > /dev/tty
|
|
IFS= read -r -s PI_ADMIN_PASSWORD < /dev/tty
|
|
printf "\n" > /dev/tty
|
|
else
|
|
printf "privacyIDEA pi-admin password: " >&2
|
|
IFS= read -r -s PI_ADMIN_PASSWORD
|
|
printf "\n" >&2
|
|
fi
|
|
if [[ -z "$PI_ADMIN_PASSWORD" ]]; then
|
|
echo "[FAIL] Empty pi-admin password." >&2
|
|
exit 1
|
|
fi
|
|
|
|
PI_POD="$(
|
|
"$KUBECTL" get pod -n "$MFA_NAMESPACE" \
|
|
-l app.kubernetes.io/name=privacyidea \
|
|
--field-selector=status.phase=Running \
|
|
-o jsonpath='{.items[0].metadata.name}' 2>/dev/null || true
|
|
)"
|
|
if [[ -z "$PI_POD" ]]; then
|
|
echo "[FAIL] No running privacyIDEA pod in namespace $MFA_NAMESPACE." >&2
|
|
exit 1
|
|
fi
|
|
|
|
echo "Fetching fresh privacyIDEA admin token inside pod $PI_POD ..."
|
|
PI_TOKEN="$(
|
|
"$KUBECTL" exec -n "$MFA_NAMESPACE" "$PI_POD" -- \
|
|
env PI_ADMIN_PASSWORD="$PI_ADMIN_PASSWORD" \
|
|
python3 -c '
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
payload = urllib.parse.urlencode({
|
|
"username": "pi-admin",
|
|
"password": os.environ["PI_ADMIN_PASSWORD"],
|
|
}).encode()
|
|
req = urllib.request.Request("http://localhost:8080/auth", data=payload)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as response:
|
|
body = json.load(response)
|
|
print(body["result"]["value"]["token"])
|
|
except Exception as exc:
|
|
print(str(exc), file=sys.stderr)
|
|
sys.exit(1)
|
|
' 2>/dev/null
|
|
)"
|
|
unset PI_ADMIN_PASSWORD
|
|
|
|
if [[ -z "$PI_TOKEN" ]]; then
|
|
echo "[FAIL] Could not fetch a privacyIDEA admin token." >&2
|
|
exit 1
|
|
fi
|
|
|
|
tmpdir="$(mktemp -d)"
|
|
cleanup() {
|
|
rm -rf "$tmpdir"
|
|
}
|
|
trap cleanup EXIT
|
|
|
|
api_get() {
|
|
local path="$1"
|
|
"$KUBECTL" exec -n "$MFA_NAMESPACE" "$PI_POD" -- \
|
|
env PI_TOKEN="$PI_TOKEN" PI_PATH="$path" \
|
|
python3 -c '
|
|
import os
|
|
import sys
|
|
import urllib.request
|
|
|
|
path = os.environ["PI_PATH"]
|
|
token = os.environ["PI_TOKEN"]
|
|
req = urllib.request.Request(
|
|
"http://localhost:8080" + path,
|
|
headers={"Authorization": token},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as response:
|
|
sys.stdout.write(response.read().decode())
|
|
except Exception as exc:
|
|
print(str(exc), file=sys.stderr)
|
|
sys.exit(1)
|
|
'
|
|
}
|
|
|
|
echo "Inspecting non-secret MFA state for $USERNAME ..."
|
|
realm_summary="$(
|
|
for realm in coulomb netkingdom; do
|
|
users_json="$(api_get "/user/?realm=$realm&username=$USERNAME" || true)"
|
|
tokens_json="$(api_get "/token/?realm=$realm&user=$USERNAME" || true)"
|
|
REALM="$realm" USERS_JSON="$users_json" TOKENS_JSON="$tokens_json" python3 -c '
|
|
import json
|
|
import os
|
|
|
|
realm = os.environ["REALM"]
|
|
users = []
|
|
tokens = []
|
|
try:
|
|
users = json.loads(os.environ["USERS_JSON"]).get("result", {}).get("value", {}).get("users", [])
|
|
except Exception:
|
|
pass
|
|
try:
|
|
value = json.loads(os.environ["TOKENS_JSON"]).get("result", {}).get("value", {})
|
|
tokens = value.get("tokens", []) if isinstance(value, dict) else []
|
|
except Exception:
|
|
pass
|
|
active = sum(1 for token in tokens if token.get("active", True))
|
|
print(f"{realm} users={len(users)} tokens={len(tokens)} active={active}")
|
|
'
|
|
done
|
|
)"
|
|
printf '%s\n' "$realm_summary"
|
|
|
|
current_config="$(
|
|
"$KUBECTL" get secret "$KEYCAPE_SECRET" -n "$SSO_NAMESPACE" \
|
|
-o jsonpath='{.data.config\.yaml}' | base64 -d
|
|
)"
|
|
current_realm="$(
|
|
CONFIG_YAML="$current_config" python3 -c '
|
|
import os
|
|
import re
|
|
|
|
match = re.search(r"(?m)^ realm:\s*[\"'\"']?([^\"'\"'\n]+)", os.environ["CONFIG_YAML"])
|
|
print(match.group(1).strip() if match else "")
|
|
'
|
|
)"
|
|
|
|
selected_realm="${KEYCAPE_PI_REALM:-}"
|
|
if [[ -z "$selected_realm" ]]; then
|
|
selected_realm="$(
|
|
REALM_SUMMARY="$realm_summary" CURRENT_REALM="$current_realm" python3 -c '
|
|
import os
|
|
|
|
lines = os.environ["REALM_SUMMARY"].splitlines()
|
|
counts = {}
|
|
for line in lines:
|
|
parts = dict(item.split("=", 1) for item in line.split()[1:])
|
|
counts[line.split()[0]] = {
|
|
"users": int(parts.get("users", "0")),
|
|
"tokens": int(parts.get("tokens", "0")),
|
|
"active": int(parts.get("active", "0")),
|
|
}
|
|
for realm in ("coulomb", "netkingdom"):
|
|
if counts.get(realm, {}).get("active", 0) > 0:
|
|
print(realm)
|
|
raise SystemExit
|
|
current = os.environ.get("CURRENT_REALM", "")
|
|
if current:
|
|
print(current)
|
|
else:
|
|
print("coulomb")
|
|
'
|
|
)"
|
|
fi
|
|
|
|
if [[ "$selected_realm" != "coulomb" && "$selected_realm" != "netkingdom" ]]; then
|
|
echo "[FAIL] Refusing unsupported privacyIDEA realm: $selected_realm" >&2
|
|
exit 1
|
|
fi
|
|
|
|
echo "Selected privacyIDEA realm for KeyCape: $selected_realm"
|
|
|
|
"$KUBECTL" get secret "$KEYCAPE_SECRET" -n "$SSO_NAMESPACE" \
|
|
-o jsonpath='{.data.key\.pem}' | base64 -d > "$tmpdir/key.pem"
|
|
chmod 600 "$tmpdir/key.pem"
|
|
|
|
CONFIG_YAML="$current_config" PI_TOKEN="$PI_TOKEN" PI_REALM="$selected_realm" \
|
|
python3 -c '
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
config = os.environ["CONFIG_YAML"]
|
|
token = json.dumps(os.environ["PI_TOKEN"])
|
|
realm = json.dumps(os.environ["PI_REALM"])
|
|
|
|
config, token_count = re.subn(r"(?m)^ adminToken:.*$", " adminToken: " + token, config)
|
|
config, realm_count = re.subn(r"(?m)^ realm:.*$", " realm: " + realm, config)
|
|
if token_count != 1 or realm_count != 1:
|
|
print("Could not patch exactly one adminToken and one realm field.", file=sys.stderr)
|
|
sys.exit(1)
|
|
sys.stdout.write(config)
|
|
' > "$tmpdir/config.yaml"
|
|
|
|
echo "Applying refreshed KeyCape config Secret ..."
|
|
"$KUBECTL" create secret generic "$KEYCAPE_TOKEN_SECRET" \
|
|
--namespace="$SSO_NAMESPACE" \
|
|
--from-literal=token="$PI_TOKEN" \
|
|
--dry-run=client -o yaml | "$KUBECTL" apply -f -
|
|
|
|
"$KUBECTL" create secret generic "$KEYCAPE_SECRET" \
|
|
--namespace="$SSO_NAMESPACE" \
|
|
--from-file=config.yaml="$tmpdir/config.yaml" \
|
|
--from-file=key.pem="$tmpdir/key.pem" \
|
|
--dry-run=client -o yaml | "$KUBECTL" apply -f -
|
|
|
|
unset PI_TOKEN
|
|
|
|
echo "Restarting KeyCape ..."
|
|
"$KUBECTL" rollout restart "deployment/$KEYCAPE_DEPLOYMENT" -n "$SSO_NAMESPACE"
|
|
"$KUBECTL" rollout status "deployment/$KEYCAPE_DEPLOYMENT" -n "$SSO_NAMESPACE" --timeout=90s
|
|
|
|
echo ""
|
|
echo "[OK] KeyCape privacyIDEA token refreshed. Retry the OIDC login with a fresh URL."
|