Patch KeyCape OpenBao client without bootstrap secrets

sso-mfa/k8s/keycape/openbao-client-config.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+"""Patch or verify the KeyCape openbao-admin client in a live Secret.
+
+The script reads a Kubernetes Secret JSON object from stdin. It never prints the
+decoded KeyCape config or private key; stdout is either a JSON merge patch for
+kubectl, or a short non-secret verification message.
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import json
+import sys
+from typing import Any
+
+try:
+    import yaml
+except ImportError as exc:  # pragma: no cover - operator environment guard
+    raise SystemExit("PyYAML is required: install python3-yaml or run in the NetKingdom tool environment") from exc
+
+
+OPENBAO_CLIENT = {
+    "clientId": "openbao-admin",
+    "displayName": "Railiance OpenBao Admin CLI",
+    "redirectUris": [
+        "http://localhost:8250/oidc/callback",
+        "http://127.0.0.1:8250/oidc/callback",
+    ],
+    "allowedScopes": ["openid", "profile", "email", "groups"],
+    "grantTypes": ["authorization_code"],
+    "clientType": "public",
+}
+
+
+def load_config() -> dict[str, Any]:
+    secret = json.load(sys.stdin)
+    encoded_config = (secret.get("data") or {}).get("config.yaml")
+    if not encoded_config:
+        raise SystemExit("keycape-config Secret does not contain data.config.yaml")
+    try:
+        config_text = base64.b64decode(encoded_config).decode("utf-8")
+    except Exception as exc:  # noqa: BLE001 - concise operator error
+        raise SystemExit(f"could not decode data.config.yaml: {exc}") from exc
+    config = yaml.safe_load(config_text) or {}
+    if not isinstance(config, dict):
+        raise SystemExit("KeyCape config.yaml must decode to a YAML mapping")
+    return config
+
+
+def client_errors(config: dict[str, Any]) -> list[str]:
+    clients = config.get("clients")
+    if not isinstance(clients, list):
+        return ["clients must be a list"]
+    target = next(
+        (client for client in clients if isinstance(client, dict) and client.get("clientId") == OPENBAO_CLIENT["clientId"]),
+        None,
+    )
+    if target is None:
+        return ["missing openbao-admin client"]
+
+    errors: list[str] = []
+    for key in ("displayName", "clientType"):
+        if target.get(key) != OPENBAO_CLIENT[key]:
+            errors.append(f"{key} should be {OPENBAO_CLIENT[key]!r}")
+    for key in ("redirectUris", "allowedScopes", "grantTypes"):
+        missing = sorted(set(OPENBAO_CLIENT[key]) - set(target.get(key) or []))
+        if missing:
+            errors.append(f"{key} missing: {', '.join(missing)}")
+    return errors
+
+
+def upsert_client(config: dict[str, Any]) -> dict[str, Any]:
+    clients = config.get("clients")
+    if not isinstance(clients, list):
+        clients = []
+        config["clients"] = clients
+    for index, client in enumerate(clients):
+        if isinstance(client, dict) and client.get("clientId") == OPENBAO_CLIENT["clientId"]:
+            clients[index] = dict(OPENBAO_CLIENT)
+            return config
+    clients.append(dict(OPENBAO_CLIENT))
+    return config
+
+
+def render_patch(config: dict[str, Any]) -> None:
+    updated = upsert_client(config)
+    config_text = yaml.safe_dump(updated, sort_keys=False)
+    encoded = base64.b64encode(config_text.encode("utf-8")).decode("ascii")
+    json.dump({"data": {"config.yaml": encoded}}, sys.stdout, separators=(",", ":"))
+    sys.stdout.write("\n")
+
+
+def verify(config: dict[str, Any]) -> None:
+    errors = client_errors(config)
+    if errors:
+        for error in errors:
+            print(f"[FAIL] {error}")
+        raise SystemExit(1)
+    print("[PASS] openbao-admin client has required CLI redirects, scopes, grant type, and public PKCE profile")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("mode", choices=("patch", "verify"))
+    args = parser.parse_args()
+    config = load_config()
+    if args.mode == "patch":
+        render_patch(config)
+    else:
+        verify(config)
+
+
+if __name__ == "__main__":
+    main()