ops-warden/tests/test_agent_read_boundary_check.py

"""Tests for scripts/check_agent_read_boundary.py (WARDEN-WP-0032-T06).

This script is a control, not a report: it is the invariant RISK-F-0009 asked for
("a check that fails when a high-risk lane has no corresponding deny"). So the
parsing has to be right about the two things that would make it lie -- treating a
non-deny grant as a deny, and treating a path pattern as a concrete address.
"""

import importlib.util
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
spec = importlib.util.spec_from_file_location(
    "check_agent_read_boundary", REPO / "scripts" / "check_agent_read_boundary.py"
)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)


class TestDeniedDataPaths:
    def test_extracts_denied_paths(self):
        policy = """
        path "platform/data/workloads/forgejo/forgejo-admin" {
          capabilities = ["deny"]
        }
        """
        assert mod.denied_data_paths(policy) == {"platform/data/workloads/forgejo/forgejo-admin"}

    def test_metadata_read_is_not_a_deny(self):
        """The policy permits metadata read alongside every data deny.

        Counting those as denies would double the apparent coverage.
        """
        policy = """
        path "platform/metadata/workloads/forgejo/forgejo-admin" {
          capabilities = ["read"]
        }
        """
        assert mod.denied_data_paths(policy) == set()

    def test_deny_is_matched_exactly_not_by_substring(self):
        """A capability merely containing 'deny' must not register as a deny."""
        policy = """
        path "platform/data/workloads/x/y" {
          capabilities = ["denylist-read"]
        }
        """
        assert mod.denied_data_paths(policy) == set()

    def test_multiple_blocks(self):
        policy = """
        path "a/data/one" { capabilities = ["deny"] }
        path "a/metadata/one" { capabilities = ["read"] }
        path "b/data/two" { capabilities = ["deny"] }
        """
        assert mod.denied_data_paths(policy) == {"a/data/one", "b/data/two"}


class TestToDataPath:
    def test_inserts_kv_v2_data_segment(self):
        assert (
            mod.to_data_path("platform/workloads/forgejo/forgejo-admin")
            == "platform/data/workloads/forgejo/forgejo-admin"
        )

    def test_tenant_mount(self):
        assert (
            mod.to_data_path("tenants/binky/company-email/imap")
            == "tenants/data/binky/company-email/imap"
        )

    def test_placeholder_pattern_has_no_address(self):
        """`openbao-api-key` is a routing pattern, not one secret.

        RISK-F-0009 counted it among the uncovered lanes; there is nothing for a
        policy to deny, and reporting it as a gap overstates the exposure.
        """
        assert mod.to_data_path("platform/workloads/<domain>/<workload>/<bundle>") is None

    def test_non_kv_lane_has_no_address(self):
        """`ops-warden-warden-sign-token` is a broker grant, not a KV path."""
        assert mod.to_data_path("credential-grants/catalog.yaml grant ops-warden/warden-sign") is None


class TestAgainstTheRealCatalog:
    def test_every_high_risk_lane_resolves_or_is_explicitly_pattern(self):
        """No high-risk lane may fall through the classifier silently.

        Each is either a concrete data path the policy can deny, or a pattern --
        never an unhandled third case, which is the ADR-0007 failure mode.
        """
        import yaml

        entries = yaml.safe_load((REPO / "registry" / "routing" / "catalog.yaml").read_text())["entries"]
        for entry in (e for e in entries if e.get("risk") == "high"):
            template = entry.get("path_template")
            if not template:
                continue
            resolved = mod.to_data_path(template)
            assert resolved is None or resolved.count("/data/") == 1, entry["id"]


class TestGeneratedArtifact:
    """The artifact railiance-platform consumes (WARDEN-WP-0033-T03).

    A consumer applies this to a live deny set, so staleness is the failure that
    matters -- a path graded high after the last emit would silently not reach them.
    """

    def test_artifact_is_current(self):
        import subprocess

        result = subprocess.run(
            ["python3", str(REPO / "scripts" / "emit_high_risk_paths.py"), "--check"],
            capture_output=True, text=True, timeout=60,
        )
        assert result.returncode == 0, (
            f"{result.stdout}{result.stderr}\n"
            "Re-run scripts/emit_high_risk_paths.py and commit the result."
        )

    def test_every_concrete_high_risk_lane_is_in_the_artifact(self):
        import yaml

        catalog = yaml.safe_load((REPO / "registry" / "routing" / "catalog.yaml").read_text())
        artifact = yaml.safe_load(
            (REPO / "registry" / "generated" / "high-risk-data-paths.yaml").read_text()
        )
        emitted = {row["id"] for row in artifact["paths"]} | set(artifact["no_concrete_path"] or [])
        graded_high = {e["id"] for e in catalog["entries"] if e.get("risk") == "high"}
        assert graded_high == emitted, "a high-risk lane is missing from the generated artifact"