ops-warden/tests/test_layer_conformance.py

"""Layer-model conformance (security-layer-model_v0.4 §5, §11).

Two things are checked here. §11 makes one of them mechanical: every direct
Tooling client maps to a declared shape. §5.2 asks for the other: the conduit's
supplied-authority property covered by a test.

Deliberately absent: any assertion on a §5.3 review date. A date-triggered
failure breaks the build on a calendar day with no code change, punishing
whoever commits next rather than whoever owns the gap — the same reasoning
recorded in WARDEN-WP-0033-T05 for blocker staleness.
"""
from __future__ import annotations

import os
import subprocess
import sys
from pathlib import Path

import yaml

ROOT = Path(__file__).resolve().parents[1]


def _decl() -> dict:
    return yaml.safe_load((ROOT / "layer.yaml").read_text())


class TestDeclaration:
    def test_declares_staff_layer_in_its_own_voice(self):
        d = _decl()
        assert d["repository"] == "ops-warden"
        assert d["layer"] == "staff"
        assert d["standard_version"] == "0.7"
        # §11: "only the repository's own file, in its own voice, conforms."
        assert d["declared_by"] == "docs/adr/ADR-0010"

    def test_every_tooling_contact_maps_to_a_declared_shape(self):
        """§11 mechanical check — the guard against a new undeclared client."""
        result = subprocess.run(
            [sys.executable, str(ROOT / "scripts" / "check_layer_conformance.py")],
            capture_output=True,
            text=True,
        )
        assert result.returncode == 0, (
            f"undeclared Tooling contact — a finding under §11, not a tracked gap:\n"
            f"{result.stdout}{result.stderr}"
        )

    def test_declared_gaps_carry_all_four_fields(self):
        """§5.3 is machine-readable or it is prose wearing a schema."""
        for c in _decl()["tooling_contacts"]:
            if c["shape"] == "5.3":
                for field in ("capability", "intended_owner", "blocked_on", "review"):
                    assert c.get(field), f"{c['id']} missing {field}"

    def test_gaps_are_not_counted_as_conformance(self):
        """§11: a declared gap is tracked non-conformance. Keep that visible."""
        text = (ROOT / "layer.yaml").read_text()
        assert "TRACKED NON-CONFORMANCE" in text.upper()


class TestConduitSuppliesNoAuthority:
    """§5.2: 'MUST NOT present its own credential, MUST NOT widen what the
    caller could already do.' The standard says this SHOULD be covered by a
    test; this is that test."""

    def test_conduit_supplies_no_authority_of_its_own(self, monkeypatch):
        from warden import proxy

        monkeypatch.setenv("VAULT_TOKEN", "caller-own-token")
        monkeypatch.setenv("HOME", "/home/nobody")
        before = dict(os.environ)

        env = proxy._caller_env()

        # The child environment IS the caller's environment — nothing added,
        # nothing removed, no ops-warden credential injected.
        assert env == before, (
            "conduit altered the caller's environment; §5.2 requires it to "
            "supply no authority of its own"
        )
        assert env["VAULT_TOKEN"] == "caller-own-token"

    def test_conduit_declares_supplied_authority_none(self):
        conduits = [c for c in _decl()["tooling_contacts"] if c["shape"] == "5.2"]
        assert conduits, "no §5.2 conduit declared — proxy.py is one"
        for c in conduits:
            assert c["supplied_authority"] == "none"

    def test_proxy_holds_no_credential_constant(self):
        """A conduit that presents its own token is not a conduit (§5.2)."""
        src = (ROOT / "src" / "warden" / "proxy.py").read_text()
        # It may name token ENV VARS to detect caller auth; it must not carry a
        # token value or mint one.
        for forbidden in ("X-Vault-Token", "auth/approle/login", "token create"):
            assert forbidden not in src, (
                f"proxy.py references {forbidden!r} — that is presenting or "
                f"minting authority, not conducting the caller's"
            )


class TestPepStanceMap:
    """§6.4: every PEP-shaped consumer MUST publish its unreachable-engine
    stance map, total and per zone, 'published rather than held in code'.
    ADR-0009 is named as the reference shape, so it should actually hold."""

    def _stance(self) -> dict:
        return yaml.safe_load((ROOT / "pep-stance.yaml").read_text())

    def test_published_map_equals_shipped_behaviour(self):
        """The whole point. A published map that may drift from the code is
        worse than none, because it invites reliance it cannot support."""
        from warden.config import PolicyConfig

        assert self._stance()["stance"] == PolicyConfig().failure_modes

    def test_stance_is_total_over_the_zone_model(self):
        """§6.4 obligation 3: total, no implicit default."""
        stance = self._stance()["stance"]
        required = {
            "z0-experimental", "z1-operational", "z2-protected",
            "z2-continuity", "z3-critical", "unknown", "not-applicable",
        }
        assert required <= set(stance), f"stance not total; missing {required - set(stance)}"
        assert set(stance.values()) <= {"fail_open", "fail_closed"}

    def test_critical_zone_fails_closed(self):
        """ADR-0009's one non-negotiable row."""
        assert self._stance()["stance"]["z3-critical"] == "fail_closed"

    def test_verdict_is_never_cached(self):
        """§6.4 obligation 2: caching an input claim is permitted; caching the
        answer is a second decision point deciding early (§6.1)."""
        assert self._stance()["verdict_caching"] == "none"

    def test_revocation_visibility_deadline_equals_enforced_ttl_policy(self):
        """§9.7.2: a published replay window must not drift from issuance."""
        from warden.models import ActorType, MAX_TTL_HOURS

        published = self._stance()["revocation_visibility"]
        expected = {actor.value: MAX_TTL_HOURS[actor] for actor in ActorType}
        assert published["deadline_hours"] == expected
        assert published["mechanism"] == "ttl_expiry"
        assert published["revocation_channel"] == "none"

    def test_attributive_emission_cadence_deferral_carries_measurement(self):
        cadence = self._stance()["emission_cadence"]
        assert cadence["classification"] == "attributive"
        assert cadence["status"] == "deferred"
        assert cadence["observed_window"]["signature_records"] == 3
        assert cadence["observed_window"]["active_signature_days"] == 2
        assert cadence["reason"]