generated from coulomb/repo-seed
Configure OpenBao auth for both netkingdom and keycape mounts with browser redirect URIs; update verify scripts and runtime architecture notes.
143 lines
4.6 KiB
Python
143 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Patch or verify non-secret KeyCape live config requirements.
|
|
|
|
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",
|
|
"redirectUris": [
|
|
"http://localhost:8250/oidc/callback",
|
|
"http://127.0.0.1:8250/oidc/callback",
|
|
"https://bao.coulomb.social/ui/vault/auth/netkingdom/oidc/callback",
|
|
"https://bao.coulomb.social/ui/vault/auth/keycape/oidc/callback",
|
|
],
|
|
"allowedScopes": ["openid", "profile", "email", "groups"],
|
|
"grantTypes": ["authorization_code"],
|
|
"clientType": "public",
|
|
}
|
|
|
|
LLDAP_REQUIRED = {
|
|
"userOU": "ou=people",
|
|
"groupOU": "ou=groups",
|
|
}
|
|
|
|
|
|
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 lldap_errors(config: dict[str, Any]) -> list[str]:
|
|
lldap = config.get("lldap")
|
|
if not isinstance(lldap, dict):
|
|
return ["lldap must be a mapping"]
|
|
return [
|
|
f"lldap.{key} should be {expected!r}"
|
|
for key, expected in LLDAP_REQUIRED.items()
|
|
if lldap.get(key) != expected
|
|
]
|
|
|
|
|
|
def enforce_lldap_defaults(config: dict[str, Any]) -> dict[str, Any]:
|
|
lldap = config.get("lldap")
|
|
if not isinstance(lldap, dict):
|
|
lldap = {}
|
|
config["lldap"] = lldap
|
|
lldap.update(LLDAP_REQUIRED)
|
|
return config
|
|
|
|
|
|
def render_patch(config: dict[str, Any]) -> None:
|
|
updated = enforce_lldap_defaults(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) + lldap_errors(config)
|
|
if errors:
|
|
for error in errors:
|
|
print(f"[FAIL] {error}")
|
|
raise SystemExit(1)
|
|
print("[PASS] openbao-admin client and LLDAP OU lookup settings are present")
|
|
|
|
|
|
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()
|