secrets-engine/tests/test_layer_conformance.py

"""Layer-model conformance (security-layer-model_v0.8 §3.3, §6, §11).

The declaration must be this repository's own, machine-readable, and equal to
the shipped production fail-closed gate. A published stance map that may drift
from the code is worse than none.
"""
from __future__ import annotations

import importlib.util
import subprocess
import sys
from pathlib import Path
from types import SimpleNamespace

import pytest
import yaml

from secrets_engine.catalog import validate_entry
from secrets_engine.cli import _require_lane_approval
from secrets_engine.errors import DecisionError
from secrets_engine.evidence_class import SHIPPED_RULES, classify, load_classification_rules
from secrets_engine.pep_stance import SHIPPED_STANCE, load_pep_stance

from tests.test_catalog import VALID

ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts" / "check_layer_conformance.py"
DECL = ROOT / "layer.yaml"
STANCE = ROOT / "pep-stance.yaml"
INTENT = ROOT / "INTENT.md"
CLASSIFICATION = ROOT / "evidence-classification.yaml"


def _decl() -> dict:
    return yaml.safe_load(DECL.read_text(encoding="utf-8"))


def _stance() -> dict:
    return yaml.safe_load(STANCE.read_text(encoding="utf-8"))


def _front() -> dict:
    text = INTENT.read_text(encoding="utf-8")
    assert text.startswith("---"), "companion §2 requires INTENT.md frontmatter"
    end = text.find("\n---", 3)
    return yaml.safe_load(text[3:end])


def _fold(value: object) -> str:
    """§3 as amended (A9): comparison is ASCII case-insensitive."""
    return str(value).strip().encode("ascii", "ignore").decode().lower()


def _checker():
    spec = importlib.util.spec_from_file_location("check_layer_conformance", SCRIPT)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def test_declares_engine_lifecycle_in_its_own_voice():
    assert DECL.exists(), "no layer.yaml — §11 requires a machine-readable declaration"
    data = _decl()
    assert data["repository"] == "secrets-engine"
    # Fold, never equality: the ruling declined to order a re-spelling.
    assert _fold(data["layer"]) == "engine"
    assert data["role"] == "lifecycle"
    assert data["declared_by"] == "INTENT.md"
    assert data["derived"] is True
    assert data["derived_from"] == "INTENT.md"
    assert data["decision_surfaces_exposed"] == "none"
    assert data["pep_shaped"] is True


def test_intent_frontmatter_matches_declaration():
    front = _front()
    assert _fold(front["layer"]) == "engine"
    assert _fold(_decl()["layer"]) == _fold(front["layer"]), (
        "INTENT.md governs and layer.yaml must agree after folding (GH-DEC-2026-017 §1-§2)"
    )
    assert front["role"] == "Lifecycle"
    assert front["declaration"] == "layer.yaml"
    assert front["pep_stance"] == "pep-stance.yaml"


def test_no_standard_version_in_either_form():
    """GH-DEC-2026-017 §5 / A12: a layer declaration carries no standard version."""
    assert "standard_version" not in _decl()
    assert "standard_version" not in _front()
    assert not str(_front().get("standard", "")).endswith(".md")


def _run_checker(tmp_path, monkeypatch, intent_layer: str, sidecar_layer: str) -> int:
    checker = _checker()
    front = _front()
    front["layer"] = intent_layer
    intent = tmp_path / "INTENT.md"
    intent.write_text("---\n" + yaml.safe_dump(front) + "---\n\n# INTENT\n", encoding="utf-8")
    decl = _decl()
    decl["layer"] = sidecar_layer
    sidecar = tmp_path / "layer.yaml"
    sidecar.write_text(yaml.safe_dump(decl), encoding="utf-8")
    monkeypatch.setattr(checker, "INTENT", intent)
    monkeypatch.setattr(checker, "DECL", sidecar)
    monkeypatch.setattr(sys, "argv", ["check_layer_conformance.py"])
    try:
        return checker.main()
    except SystemExit as exc:
        return int(exc.code)


def test_checker_folds_case_between_forms(tmp_path, monkeypatch):
    assert _run_checker(tmp_path, monkeypatch, "ENGINE", "engine") == 0
    assert _run_checker(tmp_path, monkeypatch, "Engine", "Engine") == 0


def test_checker_reports_a_real_disagreement(tmp_path, monkeypatch, capsys):
    """A post-fold disagreement is a finding, not resolved by precedence."""
    assert _run_checker(tmp_path, monkeypatch, "Engine", "staff") == 1
    assert "DECLARATION DISAGREEMENT" in capsys.readouterr().out


def test_checker_vocabulary_is_closed_at_four_tokens(tmp_path, monkeypatch, capsys):
    checker = _checker()
    assert checker.LAYER_VOCABULARY == {"taxonomy", "tooling", "engine", "staff"}
    # Taxonomy is in the vocabulary: a finding (wrong layer for this repo), not MALFORMED.
    assert _run_checker(tmp_path, monkeypatch, "Taxonomy", "taxonomy") == 1
    assert _run_checker(tmp_path, monkeypatch, "surface", "surface") == 2
    assert "outside §3's closed vocabulary" in capsys.readouterr().out


def test_checker_rejects_a_returning_standard_version(tmp_path, monkeypatch):
    checker = _checker()
    decl = _decl()
    decl["standard_version"] = "0.8"
    sidecar = tmp_path / "layer.yaml"
    sidecar.write_text(yaml.safe_dump(decl), encoding="utf-8")
    monkeypatch.setattr(checker, "DECL", sidecar)
    with pytest.raises(SystemExit) as raised:
        checker.load_declaration()
    assert raised.value.code == 2


def _run_with(tmp_path, monkeypatch, front_patch: dict, decl_patch: dict) -> int:
    checker = _checker()
    front = {**_front(), **front_patch}
    decl = {**_decl(), **decl_patch}
    intent = tmp_path / "INTENT.md"
    intent.write_text("---\n" + yaml.safe_dump(front) + "---\n\n# INTENT\n", encoding="utf-8")
    sidecar = tmp_path / "layer.yaml"
    sidecar.write_text(yaml.safe_dump(decl), encoding="utf-8")
    monkeypatch.setattr(checker, "INTENT", intent)
    monkeypatch.setattr(checker, "DECL", sidecar)
    monkeypatch.setattr(sys, "argv", ["check_layer_conformance.py"])
    try:
        return checker.main()
    except SystemExit as exc:
        return int(exc.code)


@pytest.mark.parametrize(
    ("front_patch", "decl_patch"),
    [
        # GH-DEC-2026-020 §1: a version-bearing standard: path counts.
        ({"standard": "net-kingdom/canon/standards/security-layer-model_v0.7.md"}, {}),
        ({"standard": "net-kingdom/canon/standards/security-layer-model_v0.8"}, {}),
        ({}, {"standard": "net-kingdom/canon/standards/security-layer-model_v0.7.md"}),
        ({"companion": "net-kingdom/SECURITY-COMPANION_v0.2.md"}, {}),
        # GH-DEC-2026-020 §2: companion_version counts.
        ({"companion_version": "0.2"}, {}),
        ({}, {"companion_version": "0.2"}),
        ({"standard_version": "0.7"}, {}),
        ({}, {"standard_version": "0.8"}),
    ],
)
def test_checker_rejects_a_version_anywhere_in_the_declaration(
    tmp_path, monkeypatch, capsys, front_patch, decl_patch
):
    """A12 r2 reaches content, not a key name (GH-DEC-2026-020 §1-§2)."""
    assert _run_with(tmp_path, monkeypatch, front_patch, decl_patch) == 2
    out = capsys.readouterr().out
    assert "standard or companion version" in out
    # The version belongs to the run: stated even when the run fails.
    assert "validated against: net-kingdom/canon/standards/security-layer-model_v" in out
    assert "scope:" in out


def test_schema_version_is_not_reached(tmp_path, monkeypatch):
    """GH-DEC-2026-020 §1: a declaration file's own schema version is not reached."""
    assert _run_with(tmp_path, monkeypatch, {}, {"schema_version": "0.2"}) == 0


@pytest.mark.parametrize("patch", [
    {"intent_version": "0.1.0"},
    {"rationale": "The v0.5 scope rule is historical provenance."},
    {"framework": "Historical v0.7 reference; standard identity is separate."},
])
def test_reference_detector_allows_own_versions_and_prose(tmp_path, monkeypatch, patch):
    assert _run_with(tmp_path, monkeypatch, patch, {}) == 0


@pytest.mark.parametrize("patch", [
    {"standard_version_reviewed": "0.7"},
    {"version_of_companion": "0.2"},
    {"standard": "security-layer-model v0.7"},
    {"companion": ["security companion 0.2"]},
    {"references": [{"path": "unrelated-document-v1.2.md"}]},
    {"references": ["security-layer-model@0.7"]},
])
def test_reference_detector_rejects_nested_and_identity_pins(tmp_path, monkeypatch, patch):
    assert _run_with(tmp_path, monkeypatch, {}, patch) == 2


def test_stance_and_classification_versions_are_not_reached():
    """GH-DEC-2026-020 §3: stance maps and classifications keep their version,
    and the checker never applies A12 to them."""
    checker = _checker()
    assert "pep-stance.yaml" in checker.SCOPE and "evidence-classification.yaml" in checker.SCOPE
    for path in (STANCE, CLASSIFICATION):
        data = yaml.safe_load(path.read_text(encoding="utf-8"))
        # Each carries a version A12 would reject if it were applied there...
        assert checker.find_version_pins(data), f"{path.name} keeps its standard_version"
    # ...and the real-tree run still passes: A12 is not applied to them.
    result = subprocess.run(
        [sys.executable, str(SCRIPT)], cwd=ROOT, capture_output=True, text=True, check=False
    )
    assert result.returncode == 0, result.stdout + result.stderr


def test_every_run_states_version_and_scope():
    """GH-DEC-2026-020 §4: the version and scope are printed on every run, PASS included."""
    checker = _checker()
    for argv in ([], ["--report"]):
        result = subprocess.run(
            [sys.executable, str(SCRIPT), *argv],
            cwd=ROOT, capture_output=True, text=True, check=False,
        )
        assert result.returncode == 0, result.stdout + result.stderr
        assert f"validated against: {checker.VALIDATED_AGAINST}" in result.stdout
        assert "scope: single repository secrets-engine (catalogued in §4); declaration = INTENT.md frontmatter + layer.yaml" in result.stdout
        pass_line = [l for l in result.stdout.splitlines() if l.startswith("PASS")]
        assert pass_line and checker.VALIDATED_AGAINST in pass_line[0]


def test_checker_passes_on_the_real_tree():
    result = subprocess.run(
        [sys.executable, str(SCRIPT)],
        capture_output=True,
        text=True,
        check=False,
    )
    assert result.returncode == 0, result.stdout + result.stderr


def test_proposed_capabilities_carry_gap_record_fields():
    for cap in _decl()["proposed_capabilities"]:
        for field in ("capability", "intended_owner", "blocked_on", "review", "state"):
            assert cap.get(field), f"{cap.get('id')} missing {field}"
        assert cap["state"] == "unowned-capability"
        assert cap["owner_status"] == "proposed"
        assert cap.get("contract"), f"{cap.get('id')} missing contract"


def test_stance_map_is_total_over_catalog_stages():
    stance = _stance()["stance"]
    required = {"build", "test", "prod", "unknown"}
    assert required <= set(stance), f"stance not total; missing {required - set(stance)}"
    assert set(stance.values()) <= {"fail_open", "fail_closed"}
    assert stance["prod"] == "fail_closed"
    assert stance["unknown"] == "fail_closed"
    assert _stance()["verdict_caching"] == "none"


def test_published_map_equals_shipped_constant_and_loader():
    """Changing the YAML without changing SHIPPED_STANCE fails, and the reverse."""
    published = _stance()["stance"]
    loaded = load_pep_stance().stance
    assert published == SHIPPED_STANCE
    assert loaded == SHIPPED_STANCE


def test_published_prod_stance_equals_shipped_fail_closed(monkeypatch):
    """Runtime reads pep-stance.yaml; prod fail_closed must equal the gate."""
    assert load_pep_stance().stance["prod"] == "fail_closed"
    entry = validate_entry(dict(VALID, stage="prod", approval={"model": "bootstrap-only"}))
    cfg = SimpleNamespace(hub_url="http://127.0.0.1:8000", bao_addr="http://127.0.0.1:8200")
    monkeypatch.delenv("SECRETS_ENGINE_UNSAFE_DEMO", raising=False)
    with pytest.raises(DecisionError, match="live production remains disabled") as raised:
        _require_lane_approval(cfg, entry, "apply")
    assert raised.value.stance["stance_stage"] == "prod"
    assert raised.value.stance["stance_failure_mode"] == "fail_closed"
    assert "stance_decision_id" not in raised.value.stance


def test_yaml_is_the_runtime_source(tmp_path, monkeypatch):
    """A published map the pin does not match is a test failure; runtime follows YAML."""
    path = tmp_path / "pep-stance.yaml"
    path.write_text(
        yaml.safe_dump(
            {
                "stance": {
                    "build": "fail_open",
                    "test": "fail_open",
                    "prod": "fail_open",
                    "unknown": "fail_closed",
                }
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setenv("SECRETS_ENGINE_PEP_STANCE", str(path))
    entry = validate_entry(dict(VALID, stage="prod", approval={"model": "bootstrap-only"}))
    cfg = SimpleNamespace(hub_url="http://127.0.0.1:8000", bao_addr="http://127.0.0.1:8200")
    monkeypatch.delenv("SECRETS_ENGINE_UNSAFE_DEMO", raising=False)
    # Runtime follows the YAML (prod is no longer the shipped fail_closed
    # residue). GH-DEC-2026-003 still refuses OpenBao without CAS consume.
    with pytest.raises(DecisionError, match="no durable consume binding"):
        _require_lane_approval(cfg, entry, "apply")


def test_build_stage_is_not_the_production_fail_closed_gate(tmp_path, monkeypatch):
    """build is fail_open relative to access-engine: lane approval still applies."""
    assert load_pep_stance().stance["build"] == "fail_open"
    (tmp_path / ".decisions").mkdir()
    (tmp_path / ".decisions" / "x.yaml").write_text(
        "id: x\ntitle: approved\nstatus: resolved\nsuperseded_by: null\n"
    )
    import secrets_engine.cli as cli

    monkeypatch.setattr(cli, "repo_root", lambda: tmp_path)
    entry = validate_entry(
        dict(
            VALID,
            stage="build",
            path="build/team/thing",
            approval={"model": "decision", "decision_ref": "x"},
        )
    )
    cfg = SimpleNamespace(hub_url="", bao_addr="http://127.0.0.1:8200")
    decision = _require_lane_approval(cfg, entry, "apply")
    assert decision.id == "x"


def test_classification_yaml_equals_shipped_rules():
    loaded = load_classification_rules()
    assert tuple(rule["id"] for rule in loaded) == tuple(rule["id"] for rule in SHIPPED_RULES)
    assert tuple(rule["kind"] for rule in loaded) == tuple(rule["kind"] for rule in SHIPPED_RULES)


def test_classify_does_not_grant_permission():
    prod_provision = classify("provision", "prod")
    test_provision = classify("provision", "test")
    destroy = classify("lifecycle-destroy", "build")
    apply_prod = classify("apply", "prod")
    heartbeat = classify("evidence-heartbeat", "prod")
    session_revoke = classify("session-revoke", "prod")
    wrap = classify("wrap", "prod")
    rotate = classify("rotate", "prod")
    compromise = classify("lifecycle-compromise", "prod")
    assert prod_provision.kind == "load-bearing"
    assert test_provision.kind == "attributive"
    assert destroy.kind == "load-bearing"
    assert apply_prod.kind == "attributive"
    assert heartbeat.kind == "heartbeat"
    assert session_revoke.kind == "load-bearing"
    assert wrap.kind == "load-bearing"
    assert rotate.kind == "load-bearing"
    assert compromise.kind == "load-bearing"
    assert prod_provision.completeness_claimed is False
    assert CLASSIFICATION.exists()


def test_proposed_contracts_exist_and_forbid_secret_material():
    ssh = (ROOT / "docs/ssh-ca-signing-contract.md").read_text(encoding="utf-8")
    secret_use = (ROOT / "docs/secret-use-evidence-contract.md").read_text(encoding="utf-8")
    assert "proposed" in ssh.lower()
    assert "proposed" in secret_use.lower()
    assert "warden sign" in ssh
    assert "private key" in ssh.lower() or "private keys" in ssh.lower()
    assert "secret values" in secret_use.lower() or "secret value" in secret_use.lower()
    assert "audit-core" in secret_use
    assert "completeness is not claimed" in secret_use.lower()


def test_accepted_v08_run_provenance_and_catalog_scope():
    checker = _checker()
    assert "security-layer-model_v0.8.md" in checker.VALIDATED_AGAINST
    assert "sha256:3d8298560069ef471fba28d2b1245fb2309328f231dad1a667cdf0ce5a5a7c8b" in checker.VALIDATED_AGAINST
    assert "single repository secrets-engine (catalogued in §4)" in checker.SCOPE


def test_reference_semantics_propagate_identity_through_nested_sequences():
    checker = _checker()
    for data in (
        {"nested": {"standard": ["Security Layer Model v0.8"]}},
        {"companion": [["Security Companion 0.3"]]},
        {"standard": {"alternate": ["Security Layer Model v0.7"]}},
    ):
        # A dictionary begins a fresh identity key; only sequence descendants
        # inherit the immediate standard:/companion: identity in the reference.
        if isinstance(data.get("standard"), dict):
            assert checker.find_version_pins(data) == []
        else:
            assert checker.find_version_pins(data), data
    assert checker.find_version_pins({"rationale": "Assented at v0.7; migrated at v0.8."}) == []
    assert checker.find_version_pins({"intent_version": "0.2.0", "schema_version": "0.1"}) == []


def test_ascii_folding_never_discards_non_ascii_letters():
    checker = _checker()
    assert checker._fold("Staff") == "staff"
    assert checker._fold("Engine") == "engine"
    # Deleting non-ASCII bytes would turn this fifth token into Staff.
    assert checker._fold("Stαaff") not in checker.LAYER_VOCABULARY