tegwick
9db7f5fa1c
The founder's first dry run exited non-zero without a receipt: an unexpected error escaped the handler. Read-only; no live change resulted. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 149903@bnt-lap001 Assistant-Session: 4c20a536-1452-4a85-bc09-e1987d9817d6
119 lines
5 KiB
Python
119 lines
5 KiB
Python
"""keycape alias disable helper against a fake bao (NK-WP-0054-T04). Synthetic only."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import subprocess
|
|
import sys
|
|
|
|
HERE = pathlib.Path(__file__).resolve().parent
|
|
SCRIPT = HERE.parent / "keycape-alias-disable.sh"
|
|
|
|
FAKE_BAO = r'''#!/usr/bin/env python3
|
|
import json, os, pathlib, sys
|
|
state_file = pathlib.Path(os.environ["FAKE_BAO_STATE"])
|
|
state = json.loads(state_file.read_text())
|
|
args = [a for a in sys.argv[1:] if a != "-format=json"]
|
|
state.setdefault("calls", []).append(args)
|
|
def out(data): print(json.dumps({"data": data}))
|
|
if args == ["read", "sys/auth"]:
|
|
out({m: {"type": t, "accessor": "auth_x"} for m, t in state["mounts"].items()})
|
|
elif args == ["read", "auth/token/lookup-self"]:
|
|
if state["session"] not in [m.rstrip("/") for m in state["mounts"]]:
|
|
sys.exit(2)
|
|
out({"path": "auth/%s/oidc/callback" % state["session"], "policies": ["default", "platform-admin"]})
|
|
elif args == ["list", "auth/keycape/role"]:
|
|
out({"keys": ["platform-admin"]})
|
|
elif args == ["read", "auth/keycape/config"]:
|
|
out({"oidc_client_id": "openbao-admin", "default_role": "platform-admin", "oidc_client_secret": "never"})
|
|
elif args[:2] == ["auth", "disable"]:
|
|
state["mounts"].pop(args[2] + "/")
|
|
else:
|
|
sys.exit(3)
|
|
state_file.write_text(json.dumps(state))
|
|
'''
|
|
|
|
|
|
def prepare(tmp_path, session="netkingdom", mounts=None, envelope=True):
|
|
bindir = tmp_path / "bin"
|
|
bindir.mkdir()
|
|
(bindir / "bao").write_text(FAKE_BAO)
|
|
(bindir / "bao").chmod(0o755)
|
|
state = tmp_path / "state.json"
|
|
state.write_text(json.dumps({"session": session, "mounts": mounts or
|
|
{"keycape/": "oidc", "netkingdom/": "oidc", "token/": "token"}}))
|
|
home = tmp_path / (".warden-attended-login" if envelope else "plain") / "h"
|
|
home.mkdir(parents=True)
|
|
env = {k: v for k, v in os.environ.items() if k not in ("BAO_TOKEN", "VAULT_TOKEN")}
|
|
env.update(PATH=f"{bindir}:{env['PATH']}", HOME=str(home), FAKE_BAO_STATE=str(state))
|
|
return env, state
|
|
|
|
|
|
def run(tmp_path, env, *extra):
|
|
receipt = tmp_path / f"receipt{len(list(tmp_path.glob('receipt*')))}.json"
|
|
done = subprocess.run([str(SCRIPT), "--receipt", str(receipt), *extra], env=env, capture_output=True, text=True)
|
|
assert done.stdout == "" and done.stderr == "" # silent
|
|
body = json.loads(receipt.read_text())
|
|
assert oct(receipt.stat().st_mode & 0o777) == "0o600"
|
|
assert "never" not in receipt.read_text()
|
|
return done.returncode, body
|
|
|
|
|
|
def test_dry_run_reads_only(tmp_path):
|
|
env, state = prepare(tmp_path)
|
|
code, body = run(tmp_path, env)
|
|
assert code == 0 and body["status"] == "dry_run_ok" and body["applied"] is False
|
|
assert body["alias"] == {"roles": ["platform-admin"], "oidc_client_id": "openbao-admin",
|
|
"default_role": "platform-admin"}
|
|
assert not any(c[:2] == ["auth", "disable"] for c in json.loads(state.read_text())["calls"])
|
|
|
|
|
|
def test_apply_disables_and_verifies(tmp_path):
|
|
env, state = prepare(tmp_path)
|
|
code, body = run(tmp_path, env, "--apply")
|
|
assert code == 0 and body["status"] == "disabled_verified" and body["session_after_ok"]
|
|
assert "keycape/" not in body["mounts_after"] and "netkingdom/" in body["mounts_after"]
|
|
code, body = run(tmp_path, env, "--apply")
|
|
assert code == 0 and body["status"] == "already_disabled"
|
|
|
|
|
|
def test_refuses_session_from_the_alias(tmp_path):
|
|
env, state = prepare(tmp_path, session="keycape")
|
|
code, body = run(tmp_path, env, "--apply")
|
|
assert code == 1 and body["status"] == "session_not_from_auth_netkingdom"
|
|
assert "keycape/" in json.loads(state.read_text())["mounts"]
|
|
|
|
|
|
def test_refuses_outside_the_attended_envelope(tmp_path):
|
|
env, state = prepare(tmp_path, envelope=False)
|
|
code, body = run(tmp_path, env, "--apply")
|
|
assert code == 1 and body["status"] == "attended_envelope_required"
|
|
|
|
|
|
def test_refuses_existing_receipt(tmp_path):
|
|
env, _ = prepare(tmp_path)
|
|
receipt = tmp_path / "exists.json"
|
|
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"]
|