coulomb/ops-mason
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

MASON-WP-0015-T04: structure-only AppRole build, post-build policy check

Browse source 
build_approle_kv_lane no longer reads role-id or mints secret-id;
deliver_approle_credentials is the separate founder/applier step. New
policies are read back and compared before the role is written. The
Kubernetes build uses token_policies/token_ttl (legacy policies is denied
by ops-mason-construct).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 278674@bnt-lap001
Assistant-Session: 64827407-e196-4ce3-b70c-76e7b537493f
This commit is contained in:
tegwick 2026-10-18 12:52:57 +02:00
parent d19e6b1633
commit 91c4e7b527
6 changed files with 144 additions and 62 deletions
Download patch file Download diff file Expand all files Collapse all files

86
tests/test_executor.py
View file

@ -18,6 +18,7 @@ from ops_mason.executor import (
build_spec_document,
build_spec_digest,
build_approle_kv_lane,
deliver_approle_credentials,
build_kubernetes_kv_lane,
)
@ -83,14 +84,17 @@ def test_refusal_never_calls_bao(tmp_path) -> None:
run.assert_not_called()
def test_build_writes_policy_approle_and_delivers_credentials(tmp_path) -> None:
plan = _plan(tmp_path)
spec = _spec(tmp_path)
def _fake_bao(stored_policy=None):
"""Fake bao: `policy read` returns what `policy write` stored (or stored_policy)."""
written = {}
def fake_run(cmd, input=None, capture_output=True, text=True, timeout=30):
result = MagicMock(returncode=0, stderr="")
if cmd[1:3] == ["policy", "read"]:
result.stdout = POLICY
if cmd[1:3] == ["policy", "write"]:
written[cmd[3]] = input
result.stdout = ""
elif cmd[1:3] == ["policy", "read"]:
result.stdout = stored_policy if stored_policy is not None else written.get(cmd[3], POLICY)
elif cmd[1:3] == ["read", "-field=role_id"]:
result.stdout = "role-id-value\n"
elif "-field=secret_id" in cmd:
@ -99,13 +103,57 @@ def test_build_writes_policy_approle_and_delivers_credentials(tmp_path) -> None:
result.stdout = ""
return result
with patch("ops_mason.executor.subprocess.run", side_effect=fake_run) as run:
return fake_run
def test_build_writes_policy_and_approle_structure_only(tmp_path) -> None:
plan = _plan(tmp_path)
spec = _spec(tmp_path)
with patch("ops_mason.executor.subprocess.run", side_effect=_fake_bao()) as run:
objects = build_approle_kv_lane(plan, spec)
assert objects["policy_name"] == "workload-kv-read-test-lane"
assert objects["approle_name"] == "test-lane"
assert objects["kv_path"] == "reins/test/openrouter"
assert objects["credential_delivery"].startswith("founder:")
# MASON-IN-0005: the build never becomes the role it built.
bao_cmds = [c.args[0] for c in run.call_args_list]
assert not any("role-id" in " ".join(cmd) or "secret-id" in " ".join(cmd) for cmd in bao_cmds)
assert not spec.delivery_dir.exists()
# policy write call carried the HCL on stdin, not argv -- never in a log line
# -- KV v2 shape: data/ and metadata/ sub-paths, not the bare path
policy_call = next(c for c in run.call_args_list if c.args[0][:3] == ["bao", "policy", "write"])
assert 'path "reins/data/test/openrouter"' in policy_call.kwargs["input"]
assert 'path "reins/metadata/test/openrouter"' in policy_call.kwargs["input"]
assert 'path "reins/test/openrouter"' not in policy_call.kwargs["input"]
# post-build check read the policy back before the role was written
assert ["bao", "policy", "read", "workload-kv-read-test-lane"] in bao_cmds
def test_build_refuses_when_written_policy_differs(tmp_path) -> None:
plan = _plan(tmp_path)
spec = _spec(tmp_path)
with patch("ops_mason.executor.subprocess.run", side_effect=_fake_bao(stored_policy="path \"platform/data/*\" {}\n")) as run:
with pytest.raises(BuildError, match="does not match the approved text"):
build_approle_kv_lane(plan, spec)
bao_cmds = [c.args[0] for c in run.call_args_list]
assert not any(cmd[1:3] == ["write", "auth/approle/role/test-lane"] for cmd in bao_cmds)
assert not spec.audit_log_path.exists()
def test_delivery_writes_private_credentials(tmp_path) -> None:
plan = _plan(tmp_path)
spec = _spec(tmp_path)
with patch("ops_mason.executor.subprocess.run", side_effect=_fake_bao()):
objects = deliver_approle_credentials(plan, spec)
assert objects["delivered_to"] == str(spec.delivery_dir)
role_id_file = spec.delivery_dir / "role_id"
secret_id_file = spec.delivery_dir / "secret_id"
assert role_id_file.read_text() == "role-id-value\n"
@ -113,16 +161,9 @@ def test_build_writes_policy_approle_and_delivers_credentials(tmp_path) -> None:
assert oct(role_id_file.stat().st_mode)[-3:] == "600"
assert oct(secret_id_file.stat().st_mode)[-3:] == "600"
# policy write call carried the HCL on stdin, not argv -- never in a log line
# -- KV v2 shape: data/ and metadata/ sub-paths, not the bare path
policy_call = next(c for c in run.call_args_list if c.args[0][:2] == ["bao", "policy"])
assert 'path "reins/data/test/openrouter"' in policy_call.kwargs["input"]
assert 'path "reins/metadata/test/openrouter"' in policy_call.kwargs["input"]
assert 'path "reins/test/openrouter"' not in policy_call.kwargs["input"]
# audit record landed at the explicit path, metadata only, no role_id/secret_id values
audit_text = spec.audit_log_path.read_text()
assert "policy_name" in audit_text
assert "delivered_to" in audit_text
assert "role-id-value" not in audit_text
assert "secret-id-value" not in audit_text
@ -158,13 +199,8 @@ def test_build_appends_audit_record(tmp_path) -> None:
plan = _plan(tmp_path)
spec = _spec(tmp_path)
def fake_run(cmd, input=None, capture_output=True, text=True, timeout=30):
result = MagicMock(returncode=0, stderr="")
result.stdout = "value\n"
return result
with (
patch("ops_mason.executor.subprocess.run", side_effect=fake_run),
patch("ops_mason.executor.subprocess.run", side_effect=_fake_bao()),
patch("ops_mason.executor.record_build") as record_build_mock,
):
build_approle_kv_lane(plan, spec)
@ -224,8 +260,8 @@ def test_kubernetes_lane_builds_exact_service_account_binding(tmp_path) -> None:
"auth/kubernetes/role/external-secrets-rapp-qonto",
"bound_service_account_names=external-secrets",
"bound_service_account_namespaces=external-secrets",
"policies=workload-kv-read-binky-qonto-api",
"ttl=15m",
"token_policies=workload-kv-read-binky-qonto-api",
"token_ttl=15m",
]
assert objects["kubernetes_role_name"] == "external-secrets-rapp-qonto"
audit.assert_called_once()
@ -365,7 +401,7 @@ def test_credentials_are_private_before_first_bao_call_even_with_open_umask(tmp_
previous = os.umask(0)
try:
with patch("ops_mason.executor.subprocess.run", side_effect=fake_run):
build_approle_kv_lane(plan, spec)
deliver_approle_credentials(plan, spec)
finally:
os.umask(previous)
assert "synthetic-credential" not in spec.audit_log_path.read_text()
@ -397,7 +433,7 @@ def test_unsafe_delivery_refuses_without_mutation_or_overwrite(tmp_path, case):
plan = _plan(tmp_path, spec=spec)
with patch("ops_mason.executor.subprocess.run") as run:
with pytest.raises(BuildRefused):
build_approle_kv_lane(plan, spec)
deliver_approle_credentials(plan, spec)
run.assert_not_called()
assert victim.read_text() == "leave untouched"
if case == "existing":
@ -413,7 +449,7 @@ def test_command_errors_never_echo_credentials(tmp_path, failure):
{"side_effect": subprocess.TimeoutExpired(["bao"], 30, output=marker, stderr=marker)})
with patch("ops_mason.executor.subprocess.run", **kwargs):
with pytest.raises(BuildError) as error:
build_approle_kv_lane(plan, spec)
deliver_approle_credentials(plan, spec)
assert marker not in str(error.value)
assert not spec.audit_log_path.exists()
assert stat.S_IMODE((spec.delivery_dir / "secret_id").stat().st_mode) == 0o600