tegwick
3a1bd4f1c8
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a0217e-8c4c-7383-be6b-f50a6e485306
247 lines
7.9 KiB
Python
247 lines
7.9 KiB
Python
import copy
|
|
import os
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from secrets_engine.apply import apply_plan
|
|
from secrets_engine.catalog import load_catalog, validate_entry
|
|
from secrets_engine.config import repo_root
|
|
from secrets_engine.errors import PolicyGuardError, ProvisioningError
|
|
from secrets_engine.handoff import write_approle_handoff
|
|
from secrets_engine.plan import build_plan
|
|
from secrets_engine.provision import provision_from_file
|
|
from secrets_engine.routing import route_lane
|
|
from secrets_engine.verify import run_verification
|
|
|
|
AUTH = {
|
|
"id": "warden-sign",
|
|
"kind": "auth-capability",
|
|
"org": "netkingdom",
|
|
"repo": "ops-warden",
|
|
"stage": "prod",
|
|
"mount": "ssh",
|
|
"path": "sign",
|
|
"consumers": [
|
|
{"name": "ops-warden-policy-smoke", "auth": "approle", "claim": "agent:agt"}
|
|
],
|
|
"delivery_modes": ["approle-login"],
|
|
"approval": {"model": "decision", "decision_ref": "SECRETS-WP-0004"},
|
|
"verification": {"positive": "allowlisted sign", "negative": "deny probes"},
|
|
"auth_capability": {
|
|
"policy_name": "warden-sign",
|
|
"role_name": "warden-sign",
|
|
"token_ttl": "15m",
|
|
"token_max_ttl": "15m",
|
|
"secret_id_ttl": "30m",
|
|
"secret_id_num_uses": 1,
|
|
"token_num_uses": 0,
|
|
"allowed_paths": [
|
|
{"path": "ssh/sign/agt-role", "capabilities": ["update"]},
|
|
{"path": "ssh/sign/adm-role", "capabilities": ["update"]},
|
|
{"path": "ssh/sign/atm-role", "capabilities": ["update"]},
|
|
],
|
|
"denied_probe_paths": ["ssh/sign/unlisted-role", "ssh/roles/agt-role"],
|
|
},
|
|
"rotation": {"expectation": "single-use handoff", "ttl": "15m"},
|
|
"deactivation": {"expectation": "delete approle and policy"},
|
|
"audit": {"evidence": "non-secret pointers only"},
|
|
}
|
|
|
|
|
|
def _auth_entry(**over):
|
|
data = copy.deepcopy(AUTH)
|
|
for key, value in over.items():
|
|
if key == "auth_capability":
|
|
data[key].update(value)
|
|
else:
|
|
data[key] = value
|
|
return validate_entry(data)
|
|
|
|
|
|
def test_auth_capability_entry_parses_without_fields():
|
|
entry = _auth_entry()
|
|
assert entry.kind == "auth-capability"
|
|
assert entry.fields == []
|
|
assert entry.policy_name == "warden-sign"
|
|
assert entry.role_name == "warden-sign"
|
|
assert sorted(entry.auth_allowed_paths) == [
|
|
"ssh/sign/adm-role",
|
|
"ssh/sign/agt-role",
|
|
"ssh/sign/atm-role",
|
|
]
|
|
|
|
|
|
def test_repo_catalog_loads_warden_sign_lane():
|
|
entries = load_catalog(repo_root() / "catalog")
|
|
entry = entries["warden-sign"]
|
|
assert entry.kind == "auth-capability"
|
|
assert entry.mount == "ssh"
|
|
assert entry.token_ttl == "15m"
|
|
assert entry.secret_id_num_uses == 1
|
|
|
|
|
|
def test_auth_capability_rejects_kv_fields():
|
|
data = copy.deepcopy(AUTH)
|
|
data["fields"] = ["token"]
|
|
with pytest.raises(Exception):
|
|
validate_entry(data)
|
|
|
|
|
|
def test_auth_capability_plan_has_no_kv_mount():
|
|
entry = _auth_entry()
|
|
plan = build_plan(entry, "prod", decision_id="d1")
|
|
assert not any(action.kind == "kv-mount" for action in plan.actions)
|
|
assert any(action.kind == "policy" and action.target == "warden-sign" for action in plan.actions)
|
|
assert "ssh/sign/agt-role" in plan.policy_hcl
|
|
assert "ssh/sign/adm-role" in plan.policy_hcl
|
|
assert "ssh/sign/atm-role" in plan.policy_hcl
|
|
assert "secret/data" not in plan.policy_hcl
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"path",
|
|
[
|
|
"ssh/*",
|
|
"ssh/sign/*",
|
|
"ssh/roles/agt-role",
|
|
"sys/policies/acl/warden-sign",
|
|
"auth/token/create",
|
|
"identity/entity/id",
|
|
],
|
|
)
|
|
def test_auth_capability_broad_or_admin_paths_refused(path):
|
|
entry = _auth_entry(
|
|
auth_capability={
|
|
"allowed_paths": [{"path": path, "capabilities": ["update"]}],
|
|
}
|
|
)
|
|
with pytest.raises(PolicyGuardError):
|
|
build_plan(entry, "prod")
|
|
|
|
|
|
def test_auth_capability_root_like_policy_name_refused():
|
|
entry = _auth_entry(auth_capability={"policy_name": "root-warden-sign"})
|
|
with pytest.raises(PolicyGuardError):
|
|
build_plan(entry, "prod")
|
|
|
|
|
|
def test_auth_capability_provision_rejected(tmp_path):
|
|
value_file = tmp_path / "value"
|
|
value_file.write_text("not-real-secret")
|
|
os.chmod(value_file, 0o600)
|
|
with pytest.raises(ProvisioningError):
|
|
provision_from_file(object(), _auth_entry(), "", value_file)
|
|
|
|
|
|
class FakeApplyClient:
|
|
def __init__(self):
|
|
self.policy = ""
|
|
self.approle = None
|
|
|
|
def kv_mount_exists(self, mount):
|
|
raise AssertionError("auth-capability apply must not inspect KV mounts")
|
|
|
|
def ensure_kv_mount(self, mount):
|
|
raise AssertionError("auth-capability apply must not create KV mounts")
|
|
|
|
def read_policy(self, name):
|
|
return None
|
|
|
|
def write_policy(self, name, hcl):
|
|
self.policy = hcl
|
|
|
|
def ensure_approle_enabled(self):
|
|
pass
|
|
|
|
def write_approle(self, role_name, policies, ttl="30m", **kwargs):
|
|
self.approle = (role_name, policies, ttl, kwargs)
|
|
|
|
|
|
def test_apply_auth_capability_bypasses_kv_and_writes_ttl_options():
|
|
entry = _auth_entry()
|
|
plan = build_plan(entry, "prod", decision_id="d1")
|
|
client = FakeApplyClient()
|
|
result = apply_plan(client, entry, plan)
|
|
assert any("not applicable" in item for item in result.skipped)
|
|
assert "ssh/sign/agt-role" in client.policy
|
|
assert client.approle[0] == "warden-sign"
|
|
assert client.approle[2] == "15m"
|
|
assert client.approle[3]["secret_id_num_uses"] == 1
|
|
assert client.approle[3]["token_num_uses"] == 0
|
|
|
|
|
|
class FakeVerifyClient:
|
|
def __init__(self):
|
|
self.token = "test-token"
|
|
self.sessions_closed = 0
|
|
|
|
@contextmanager
|
|
def approle_session(self, role_name):
|
|
assert role_name == "warden-sign"
|
|
try:
|
|
yield SimpleNamespace(client=self)
|
|
finally:
|
|
self.sessions_closed += 1
|
|
|
|
def token_capabilities(self, path, *, token):
|
|
assert token == "test-token"
|
|
if path.startswith("ssh/sign/") and path != "ssh/sign/unlisted-role":
|
|
return ["update"]
|
|
return ["deny"]
|
|
|
|
|
|
def test_auth_capability_verification_uses_capability_probes():
|
|
client = FakeVerifyClient()
|
|
results = run_verification(
|
|
client, _auth_entry(), "", positive=True, negative=True
|
|
)
|
|
assert [result.passed for result in results] == [True, True]
|
|
assert client.sessions_closed == 2
|
|
|
|
|
|
class FakeHandoffClient:
|
|
def read_approle_role_id(self, role_name):
|
|
assert role_name == "warden-sign"
|
|
return "role-id-test"
|
|
|
|
def create_approle_secret_id(self, role_name):
|
|
assert role_name == "warden-sign"
|
|
return "secret-id-test"
|
|
|
|
|
|
def test_handoff_rejects_repo_paths():
|
|
with pytest.raises(ProvisioningError):
|
|
write_approle_handoff(
|
|
FakeHandoffClient(),
|
|
_auth_entry(),
|
|
role_id_file=Path("role-id.out"),
|
|
secret_id_file=Path("secret-id.out"),
|
|
)
|
|
|
|
|
|
def test_handoff_writes_mode_0600_files(tmp_path):
|
|
role_file = tmp_path / "role-id"
|
|
secret_file = tmp_path / "secret-id"
|
|
result = write_approle_handoff(
|
|
FakeHandoffClient(),
|
|
_auth_entry(),
|
|
role_id_file=role_file,
|
|
secret_id_file=secret_file,
|
|
)
|
|
assert result.role_name == "warden-sign"
|
|
assert role_file.read_text().strip() == "role-id-test"
|
|
assert secret_file.read_text().strip() == "secret-id-test"
|
|
assert (role_file.stat().st_mode & 0o077) == 0
|
|
assert (secret_file.stat().st_mode & 0o077) == 0
|
|
|
|
|
|
def test_route_missing_decision_blocks_auth_capability(tmp_path):
|
|
result = route_lane(_auth_entry(), hub_url="", repo_root=tmp_path, client=None)
|
|
assert result.kind == "auth-capability"
|
|
assert result.decision_status == "missing"
|
|
assert result.ready is False
|
|
assert result.missing == "approved decision for 'SECRETS-WP-0004'"
|
|
assert result.next_command == "secrets-engine decision inspect SECRETS-WP-0004"
|