#!/usr/bin/env python3 from __future__ import annotations import argparse import getpass import os import shlex import subprocess import sys from pathlib import Path from typing import Any REPO_DIR = Path(__file__).resolve().parents[1] APPLIERS: dict[str, dict[str, Any]] = { "nonprod": { "title": "Credential change non-production metadata applier", "policy_name": "credential-change-nonprod-applier", "policy_file": "openbao/policies/credential-change-nonprod-applier.hcl", "token_role": "credential-change-nonprod-applier", "max_ttl": "1h", }, "prod": { "title": "Credential change production metadata applier", "policy_name": "credential-change-prod-applier", "policy_file": "openbao/policies/credential-change-prod-applier.hcl", "token_role": "credential-change-prod-applier", "max_ttl": "30m", }, } DISALLOWED_POLICIES = ("root", "platform-admin") class BaoRunner: def __init__( self, *, kubectl: str, namespace: str, release: str, dry_run: bool, use_token_helper: bool, token: str | None, ) -> None: self.kubectl_parts = shlex.split(kubectl) self.namespace = namespace self.pod = f"{release}-0" self.dry_run = dry_run self.use_token_helper = use_token_helper self.token = token def run( self, args: list[str], input_text: str | None = None ) -> subprocess.CompletedProcess[str]: rendered = "bao " + shlex.join(args) if self.dry_run: print(f"DRY-RUN: {rendered}") return subprocess.CompletedProcess(args, 0, "", "") if self.use_token_helper: cmd = ( self.kubectl_parts + ["exec", "-i", "-n", self.namespace, self.pod, "--", "bao"] + args ) proc_input = input_text else: if not self.token: raise RuntimeError( "OpenBao token is required unless --use-token-helper is set" ) cmd = ( self.kubectl_parts + [ "exec", "-i", "-n", self.namespace, self.pod, "--", "sh", "-c", 'read -r BAO_TOKEN; export BAO_TOKEN; exec bao "$@"', "sh", ] + args ) proc_input = self.token + "\n" + (input_text or "") result = subprocess.run(cmd, input=proc_input, capture_output=True, text=True) if result.stdout: print(result.stdout, end="") if result.returncode != 0: if result.stderr: print(result.stderr, file=sys.stderr, end="") raise SystemExit(result.returncode) if result.stderr: print(result.stderr, file=sys.stderr, end="") return result def read_token( token_file: str | None, dry_run: bool, use_token_helper: bool ) -> str | None: if dry_run or use_token_helper: return None if token_file: path = Path(token_file) if not path.exists(): raise SystemExit(f"ERROR: OPENBAO_TOKEN_FILE does not exist: {path}") lines = path.read_text(encoding="utf-8").splitlines() token = lines[0].strip() if lines else "" if not token: raise SystemExit(f"ERROR: OPENBAO_TOKEN_FILE is empty: {path}") return token token = getpass.getpass("OpenBao token: ") if not token: raise SystemExit("ERROR: empty OpenBao token") return token def selected_appliers(selector: str) -> list[dict[str, Any]]: if selector == "all": return [APPLIERS["nonprod"], APPLIERS["prod"]] try: return [APPLIERS[selector]] except KeyError: raise SystemExit(f"ERROR: applier must be one of {sorted(APPLIERS) + ['all']}") def role_args(applier: dict[str, Any]) -> list[str]: return [ "write", f"auth/token/roles/{applier['token_role']}", f"allowed_policies={applier['policy_name']}", f"disallowed_policies={','.join(DISALLOWED_POLICIES)}", "orphan=true", "renewable=false", f"token_explicit_max_ttl={applier['max_ttl']}", "token_no_default_policy=true", "token_type=service", ] def write_policy(runner: BaoRunner, applier: dict[str, Any], policy_dir: Path) -> None: policy_file = policy_dir / Path(applier["policy_file"]).name if not policy_file.exists(): raise SystemExit(f"ERROR: missing policy file: {policy_file}") if runner.dry_run: print(f"DRY-RUN: bao policy write {applier['policy_name']} {policy_file}") return runner.run( ["policy", "write", applier["policy_name"], "-"], input_text=policy_file.read_text(encoding="utf-8"), ) print(f"OK: policy {applier['policy_name']} applied") def apply_applier(runner: BaoRunner, applier: dict[str, Any], policy_dir: Path) -> None: write_policy(runner, applier, policy_dir) runner.run(role_args(applier)) runner.run(["read", f"auth/token/roles/{applier['token_role']}"]) print( "OK: applier role " f"{applier['token_role']} configured for policy {applier['policy_name']}" ) def main() -> int: parser = argparse.ArgumentParser( description="Apply OpenBao credential-change delegated applier policies and token roles." ) parser.add_argument( "--applier", choices=["nonprod", "prod", "all"], default="all" ) parser.add_argument("--policy-dir", default="openbao/policies") parser.add_argument("--dry-run", action="store_true") parser.add_argument( "--use-token-helper", action="store_true", help="Use the OpenBao CLI token helper inside the pod", ) parser.add_argument("--namespace", default=None) parser.add_argument("--release", default=None) parser.add_argument("--kubectl", default=None) args = parser.parse_args() namespace = args.namespace or os.environ.get("OPENBAO_NAMESPACE", "openbao") release = args.release or os.environ.get("OPENBAO_RELEASE", "openbao") kubectl = args.kubectl or os.environ.get("KUBECTL", "kubectl") token_file = os.environ.get("OPENBAO_TOKEN_FILE") token = read_token(token_file, args.dry_run, args.use_token_helper) runner = BaoRunner( kubectl=kubectl, namespace=namespace, release=release, dry_run=args.dry_run, use_token_helper=args.use_token_helper, token=token, ) if not args.dry_run: runner.run(["status"]) policy_dir = REPO_DIR / args.policy_dir for applier in selected_appliers(args.applier): apply_applier(runner, applier, policy_dir) print("NEXT: issue short-lived child tokens through an approved custody path only.") print("NEXT: run scripts/credential-change.py applier-apply with that ambient delegated authority.") return 0 if __name__ == "__main__": raise SystemExit(main())