tegwick
0d62ac501d
docs/review-optimize-checklist.md: six checks (naming, TTL/scoping,
redundancy, compaction, ease of use, posture), applied for real to the
rein-openweights plan's section 4 -- including a genuinely useful
finding (credentials.py already expects this exact path/delivery shape,
zero code changes needed to consume it).
docs/executive-summary-format.md: six fixed fields, no bao syntax, no
restating earlier sections, explicit approve/reject/revise decision.
Rendered for real into the plan's section 5 -- ready for an actual
decision.
src/ops_mason/{plan,executor,audit}.py: the phase-4 build executor for
credential_type openbao-approle-kv. Refuses to run against anything but
an approved plan -- verified the refusal never even calls subprocess.run.
role_id/secret_id (the AppRole's own access credential, not the
downstream secret) land as 0600 files, never logged; the HCL policy
goes over stdin, never argv; the audit trail is metadata-only. 12 tests,
all mocked at the bao boundary (no live OpenBao access from this
session).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
123 lines
4.6 KiB
Python
123 lines
4.6 KiB
Python
import json
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from ops_mason.executor import AppRoleKVSpec, BuildError, BuildRefused, build_approle_kv_lane
|
|
from ops_mason.plan import ConstructionPlan
|
|
|
|
|
|
def _plan(tmp_path, *, status="approved", approved_by="bernd", approved_at="2026-07-27"):
|
|
text = (
|
|
"---\n"
|
|
"id: test-lane\n"
|
|
f"status: {status}\n"
|
|
+ (f'approved_by: "{approved_by}"\n' if approved_by else "approved_by: null\n")
|
|
+ (f'approved_at: "{approved_at}"\n' if approved_at else "approved_at: null\n")
|
|
+ "---\n# Plan\n"
|
|
)
|
|
p = tmp_path / "plan.md"
|
|
p.write_text(text)
|
|
return ConstructionPlan.load(str(p))
|
|
|
|
|
|
def _spec(tmp_path) -> AppRoleKVSpec:
|
|
return AppRoleKVSpec(
|
|
policy_name="workload-kv-read-test-lane",
|
|
kv_path="reins/test/openrouter",
|
|
approle_name="test-lane",
|
|
delivery_dir=tmp_path / "delivery",
|
|
audit_log_path=tmp_path / "audit.jsonl",
|
|
)
|
|
|
|
|
|
def test_refuses_when_not_approved(tmp_path) -> None:
|
|
plan = _plan(tmp_path, status="draft", approved_by=None, approved_at=None)
|
|
with pytest.raises(BuildRefused, match="not approved"):
|
|
build_approle_kv_lane(plan, _spec(tmp_path))
|
|
|
|
|
|
def test_refuses_when_approved_status_but_missing_approver(tmp_path) -> None:
|
|
plan = _plan(tmp_path, status="approved", approved_by=None, approved_at="2026-07-27")
|
|
with pytest.raises(BuildRefused, match="not approved"):
|
|
build_approle_kv_lane(plan, _spec(tmp_path))
|
|
|
|
|
|
def test_refusal_never_calls_bao(tmp_path) -> None:
|
|
plan = _plan(tmp_path, status="reviewed", approved_by=None, approved_at=None)
|
|
with patch("ops_mason.executor.subprocess.run") as run:
|
|
with pytest.raises(BuildRefused):
|
|
build_approle_kv_lane(plan, _spec(tmp_path))
|
|
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_run(cmd, input=None, capture_output=True, text=True, timeout=30):
|
|
result = MagicMock(returncode=0, stderr="")
|
|
if cmd[1:3] == ["read", "-field=role_id"]:
|
|
result.stdout = "role-id-value\n"
|
|
elif "-field=secret_id" in cmd:
|
|
result.stdout = "secret-id-value\n"
|
|
else:
|
|
result.stdout = ""
|
|
return result
|
|
|
|
with patch("ops_mason.executor.subprocess.run", side_effect=fake_run) 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"
|
|
|
|
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"
|
|
assert secret_id_file.read_text() == "secret-id-value\n"
|
|
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
|
|
policy_call = next(c for c in run.call_args_list if c.args[0][:2] == ["bao", "policy"])
|
|
assert "reins/test/openrouter" 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 "role-id-value" not in audit_text
|
|
assert "secret-id-value" not in audit_text
|
|
|
|
|
|
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.record_build") as record_build_mock,
|
|
):
|
|
build_approle_kv_lane(plan, spec)
|
|
|
|
record_build_mock.assert_called_once()
|
|
kwargs = record_build_mock.call_args.kwargs
|
|
assert kwargs["plan_id"] == "test-lane"
|
|
assert kwargs["approved_by"] == "bernd"
|
|
assert "policy_name" in kwargs["objects"]
|
|
assert kwargs["log_path"] == spec.audit_log_path
|
|
|
|
|
|
def test_bao_failure_raises_build_error(tmp_path) -> None:
|
|
plan = _plan(tmp_path)
|
|
spec = _spec(tmp_path)
|
|
|
|
fake_result = MagicMock(returncode=1, stderr="permission denied", stdout="")
|
|
with patch("ops_mason.executor.subprocess.run", return_value=fake_result):
|
|
with pytest.raises(BuildError, match="permission denied"):
|
|
build_approle_kv_lane(plan, spec)
|