diff --git a/tools/service-identity/keycape_alias_disable.py b/tools/service-identity/keycape_alias_disable.py index 83f26c1..6031efc 100755 --- a/tools/service-identity/keycape_alias_disable.py +++ b/tools/service-identity/keycape_alias_disable.py @@ -17,6 +17,7 @@ import argparse import json import os import pathlib +import re import subprocess import sys from datetime import datetime, timezone @@ -26,15 +27,31 @@ KEEP = "netkingdom/" SCHEMA = "net-kingdom.openbao-keycape-alias-disable.v1" +TOKENISH = re.compile(r"\b[a-z]{1,4}\.[A-Za-z0-9_-]{16,}|[A-Za-z0-9_-]{24,}") +STEP = {"name": "start"} + + class Refused(Exception): pass def bao_json(*args: str) -> dict: + STEP["name"] = " ".join(args[:2]) out = subprocess.run(["bao", args[0], "-format=json", *args[1:]], capture_output=True, check=True, timeout=30) return json.loads(out.stdout) +def error_summary(error: BaseException) -> dict: + """Error class, failing step and bao's status lines only; token-shaped strings redacted.""" + summary = {"error": type(error).__name__, "step": STEP["name"]} + if isinstance(error, subprocess.CalledProcessError): + text = (error.stderr or b"").decode("utf-8", "replace") + lines = [ln.strip() for ln in text.splitlines() if ln.strip().startswith(("Code:", "* ", "URL:", "Error"))] + summary["exit"] = error.returncode + summary["detail"] = [TOKENISH.sub("[redacted]", ln)[:200] for ln in lines[:4]] + return summary + + def require_attended() -> None: if pathlib.Path.home().parent.name != ".warden-attended-login" or os.getenv("BAO_TOKEN") or os.getenv("VAULT_TOKEN"): raise Refused("attended_envelope_required") @@ -81,6 +98,7 @@ def run(apply: bool) -> dict: result["alias"] = alias_view() if not apply: return {**result, "status": "dry_run_ok"} + STEP["name"] = "auth disable" subprocess.run(["bao", "auth", "disable", MOUNT.rstrip("/")], capture_output=True, check=True, timeout=60) after = mounts() still_me = session() # our own token, from auth/netkingdom, must survive @@ -108,8 +126,9 @@ def main(argv: list[str] | None = None) -> int: body = run(args.apply) except Refused as error: body = {"status": str(error), "applied": False} - except subprocess.CalledProcessError as error: - body = {"status": "bao_failed", "step": " ".join(error.cmd[:3]), "applied": None} + except Exception as error: # noqa: BLE001 - every failure must still leave a receipt + body = {"status": "bao_failed" if isinstance(error, subprocess.CalledProcessError) else "helper_failed", + "applied": None if STEP["name"] == "auth disable" else False, **error_summary(error)} write_receipt(args.receipt, body) return 0 if body["status"] in ("dry_run_ok", "disabled_verified", "already_disabled") else 1 diff --git a/tools/service-identity/tests/test_keycape_alias_disable.py b/tools/service-identity/tests/test_keycape_alias_disable.py index fd2dbf4..873a3c9 100644 --- a/tools/service-identity/tests/test_keycape_alias_disable.py +++ b/tools/service-identity/tests/test_keycape_alias_disable.py @@ -98,3 +98,22 @@ def test_refuses_existing_receipt(tmp_path): receipt.write_text("{}") done = subprocess.run([str(SCRIPT), "--receipt", str(receipt)], env=env) assert done.returncode != 0 and receipt.read_text() == "{}" + + +def test_unexpected_bao_output_still_leaves_a_receipt(tmp_path): + env, state = prepare(tmp_path) + bao = pathlib.Path(env["PATH"].split(":")[0]) / "bao" + bao.write_text("#!/bin/sh\necho 'not json'\n") + code, body = run(tmp_path, env) + assert code == 1 and body["status"] == "helper_failed" and body["step"] == "read auth/token/lookup-self" + assert body["applied"] is False + + +def test_bao_error_is_summarised_without_tokens(tmp_path): + env, state = prepare(tmp_path) + bao = pathlib.Path(env["PATH"].split(":")[0]) / "bao" + bao.write_text("#!/bin/sh\necho 'Error reading: Code: 403. Errors:' >&2\n" + "echo '* permission denied hvs.ABCDEFGHIJKLMNOPQRSTUVWX' >&2\nexit 2\n") + code, body = run(tmp_path, env) + assert body["status"] == "bao_failed" and body["exit"] == 2 + assert "hvs.ABCDEFGHIJKLMNOPQRSTUVWX" not in json.dumps(body) and body["detail"]