secrets-engine/tests/test_guards.py

"""Negative checks: a plan that would grant broad power must fail closed."""
import pytest

from secrets_engine.catalog import validate_entry
from secrets_engine.errors import PolicyGuardError
from secrets_engine.plan import build_plan
from secrets_engine.roles import assert_path_in_stage, assert_policy_safe

from tests.test_catalog import VALID


def _entry(**over):
    d = dict(VALID)
    d.update(over)
    return validate_entry(d)


def test_wildcard_policy_path_refused():
    with pytest.raises(PolicyGuardError):
        assert_policy_safe("se-test-x", {"secret/*": ["read"]})


def test_sys_path_refused():
    with pytest.raises(PolicyGuardError):
        assert_policy_safe("se-test-x", {"sys/policies/acl/x": ["read"]})


def test_identity_path_refused():
    with pytest.raises(PolicyGuardError):
        assert_policy_safe("se-test-x", {"identity/entity/x": ["read"]})


def test_admin_policy_name_refused():
    with pytest.raises(PolicyGuardError):
        assert_policy_safe("platform-admin", {"secret/data/x": ["read"]})


def test_broad_capability_refused():
    with pytest.raises(PolicyGuardError):
        assert_policy_safe("se-test-x", {"secret/data/x": ["sudo"]})


def test_out_of_stage_path_refused():
    # a 'test' lane pointing into the build prefix is rejected
    e = _entry(stage="test", path="build/sneaky/thing")
    with pytest.raises(PolicyGuardError):
        assert_path_in_stage(e)


def test_build_lane_must_use_build_prefix():
    e = _entry(stage="build", path="random/thing")
    with pytest.raises(PolicyGuardError):
        assert_path_in_stage(e)


def test_stage_mismatch_in_plan_refused():
    e = _entry(stage="test", path="test/team/thing")
    with pytest.raises(PolicyGuardError):
        build_plan(e, "prod")


def test_valid_plan_builds():
    e = _entry(stage="test", path="test/team/thing")
    plan = build_plan(e, "test", decision_id="d1")
    assert plan.policy_name == "se-test-test-lane"
    assert any(a.kind == "approle" for a in plan.actions)
    assert "secret/data/test/team/thing" in plan.policy_hcl


def test_existing_mount_plan_has_check_not_mount_mutation():
    e = _entry(
        stage="prod",
        mount="platform",
        path="workloads/example/runtime",
        mount_management="existing",
    )
    plan = build_plan(e, "prod", decision_id="d1")
    assert any(a.kind == "kv-mount-check" for a in plan.actions)
    assert not any(a.kind == "kv-mount" for a in plan.actions)
    assert "platform/data/workloads/example/runtime" in plan.policy_hcl


def test_existing_auth_plan_has_checks_not_auth_mutations():
    e = _entry(
        delivery_auth={
            "method": "approle",
            "management": "existing",
            "role_name": "existing-exact-role",
            "policy_name": "existing-exact-policy",
        }
    )
    plan = build_plan(e, "test", decision_id="d1")
    assert any(a.kind == "policy-check" for a in plan.actions)
    assert any(a.kind == "approle-check" for a in plan.actions)
    assert not any(a.kind == "policy" for a in plan.actions)
    assert not any(a.kind == "approle" for a in plan.actions)


def test_every_admitted_lane_renders_existing_mount_check_and_exact_policy():
    from secrets_engine.catalog import load_catalog
    from secrets_engine.config import repo_root

    entries = load_catalog(repo_root() / "catalog")
    for lane_id in (
        "issue-core-ingestion-api-key",
        "reuse-surface-hub-write-token",
        "openrouter-llm-connect",
        "forgejo-admin-api-token",
        "email-connect-transactional",
    ):
        entry = entries[lane_id]
        plan = build_plan(entry, "prod", decision_id=entry.approval["decision_ref"])
        assert [a.kind for a in plan.actions] == ["kv-mount-check", "policy", "approle"]
        assert f'path "{entry.kv_data_path}"' in plan.policy_hcl
        assert "*" not in entry.kv_data_path
        assert plan.role_name.startswith("se-prod-")