tegwick
8e41f85121
reuse_policy leaves external-secrets-audit-core intact so database leases keep working. AppRole delivered to Kubernetes; interim static ESO token retired.
229 lines
8.4 KiB
Python
229 lines
8.4 KiB
Python
import json
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from ops_mason.executor import (
|
|
AppRoleKVSpec,
|
|
BuildError,
|
|
BuildRefused,
|
|
KubernetesKVSpec,
|
|
_policy_hcl,
|
|
build_approle_kv_lane,
|
|
build_kubernetes_kv_lane,
|
|
)
|
|
|
|
|
|
def test_policy_hcl_uses_kv_v2_data_and_metadata_paths() -> None:
|
|
hcl = _policy_hcl("reins/rein-openweights/openrouter", ("read",))
|
|
assert 'path "reins/data/rein-openweights/openrouter" {' in hcl
|
|
assert 'path "reins/metadata/rein-openweights/openrouter" {' in hcl
|
|
assert 'path "reins/rein-openweights/openrouter" {' not in hcl
|
|
assert 'capabilities = ["read"]' in hcl
|
|
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",
|
|
token_num_uses=8,
|
|
delivery_dir=tmp_path / "delivery",
|
|
audit_log_path=tmp_path / "audit.jsonl",
|
|
)
|
|
|
|
|
|
def test_token_num_uses_has_no_silently_unbounded_default() -> None:
|
|
import inspect
|
|
|
|
sig = inspect.signature(AppRoleKVSpec)
|
|
assert sig.parameters["token_num_uses"].default is inspect.Parameter.empty
|
|
|
|
|
|
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
|
|
# -- 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 "role-id-value" not in audit_text
|
|
assert "secret-id-value" not in audit_text
|
|
|
|
|
|
def test_reuse_policy_does_not_rewrite_existing_policy(tmp_path) -> None:
|
|
plan = _plan(tmp_path)
|
|
spec = _spec(tmp_path)
|
|
spec.reuse_policy = True
|
|
|
|
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:
|
|
build_approle_kv_lane(plan, spec)
|
|
|
|
bao_cmds = [c.args[0] for c in run.call_args_list]
|
|
assert not any(cmd[:2] == ["bao", "policy"] for cmd in bao_cmds)
|
|
assert any(cmd[1:3] == ["write", "auth/approle/role/test-lane"] for cmd in bao_cmds)
|
|
|
|
|
|
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)
|
|
|
|
|
|
def _kubernetes_spec(tmp_path) -> KubernetesKVSpec:
|
|
return KubernetesKVSpec(
|
|
policy_name="workload-kv-read-binky-qonto-api",
|
|
role_name="external-secrets-rapp-qonto",
|
|
service_account_names=("external-secrets",),
|
|
service_account_namespaces=("external-secrets",),
|
|
audit_log_path=tmp_path / "audit.jsonl",
|
|
)
|
|
|
|
|
|
def test_kubernetes_lane_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_kubernetes_kv_lane(plan, _kubernetes_spec(tmp_path))
|
|
run.assert_not_called()
|
|
|
|
|
|
def test_kubernetes_lane_builds_exact_service_account_binding(tmp_path) -> None:
|
|
plan = _plan(tmp_path)
|
|
spec = _kubernetes_spec(tmp_path)
|
|
result = MagicMock(returncode=0, stderr="", stdout="")
|
|
with (
|
|
patch("ops_mason.executor.subprocess.run", return_value=result) as run,
|
|
patch("ops_mason.executor.record_build") as audit,
|
|
):
|
|
objects = build_kubernetes_kv_lane(plan, spec)
|
|
|
|
command = run.call_args.args[0]
|
|
assert command == [
|
|
"bao",
|
|
"write",
|
|
"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",
|
|
]
|
|
assert objects["kubernetes_role_name"] == "external-secrets-rapp-qonto"
|
|
audit.assert_called_once()
|
|
|
|
|
|
def test_kubernetes_lane_requires_nonempty_bindings(tmp_path) -> None:
|
|
plan = _plan(tmp_path)
|
|
spec = KubernetesKVSpec(
|
|
policy_name="policy",
|
|
role_name="role",
|
|
service_account_names=(),
|
|
service_account_namespaces=("external-secrets",),
|
|
)
|
|
with patch("ops_mason.executor.subprocess.run") as run:
|
|
with pytest.raises(BuildRefused, match="explicit service account"):
|
|
build_kubernetes_kv_lane(plan, spec)
|
|
run.assert_not_called()
|