feat(P1): kyverno-json engine core + PolicyEngine protocol (REQ-291..294, 308, 309)

core/policy_engine.py: PolicyEngine Protocol (PEP 544, runtime_checkable) + PolicyEngineRegistry (selects from config.json.policy.engine) + NullEngine fallback (NULL_ENGINE_INACTIVE when policy key absent). adapters/kyverno-json/: KyvernoJsonEngine — shells to , translates native output → list[dict] PCR records (engine: "kyverno", ruleId KJ_ prefix, severity via nova.cloudinit.dev/severity annotation, default info). is_configured() guards on → KJ_ENGINE_NOT_CONFIGURED SKIPPED PCR (distinct from NullEngine). Defensive parsing (malformed → error PCR). config.json: new object {engine: kyverno-json, policy_root}. scripts/install-kyverno-json.sh: go install kj@latest (D-115). CI (.gitea + .github): install Go + kj for policy-engine tests (best-effort; tests skip when kj absent). tests: 24 pass, 2 skip (kj not installed). 132 existing tests unchanged. NullEngine satisfies PolicyEngine Protocol (G-Q8a — proves swap boundary). ---ci--- project: acdl phase: 1 milestone: v1.25 status: execute phase_role: execution requirements: covered: [REQ-291, REQ-292, REQ-293, REQ-294, REQ-308, REQ-309] partial: [] ---/ci---
2026-08-12 18:19:16 +00:00
parent ba816f69ae
commit ac18c98385
10 changed files with 948 additions and 1 deletions
@@ -0,0 +1,213 @@
+"""Tests for adapters/kyverno-json/kyverno_json_engine.py (REQ-309, v1.25).
+
+PCR schema validity (jsonschema validation), defensive parsing
+(malformed output → error PCR, never exception), is_configured()
+guard, severity annotation reading (G-Q10a), and pytest.skip when
+kj is absent.
+"""
+
+import json
+import os
+import sys
+from pathlib import Path
+from unittest import mock
+
+import jsonschema
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+# Load the engine module by file path (the dir has a hyphen).
+import importlib.util
+_ENGINE_PATH = Path(__file__).resolve().parent.parent / "adapters" / "kyverno-json" / "kyverno_json_engine.py"
+_spec = importlib.util.spec_from_file_location("kyverno_json_engine", _ENGINE_PATH)
+_mod = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(_mod)
+KyvernoJsonEngine = _mod.KyvernoJsonEngine
+_to_pcr = _mod._to_pcr
+_load_policy_severities = _mod._load_policy_severities
+
+PCR_SCHEMA_PATH = Path(__file__).resolve().parent.parent / "schemas" / "policy_check_result.schema.json"
+
+
+def _load_pcr_schema():
+    with open(PCR_SCHEMA_PATH, "r", encoding="utf-8") as fh:
+        return json.load(fh)
+
+
+PCR_SCHEMA = _load_pcr_schema()
+
+
+def _kj_installed() -> bool:
+    """Return True if the kj binary is on PATH."""
+    return _mod._which_kj() is not None
+
+
+def _smoke_policy_dir() -> Path:
+    return Path(__file__).resolve().parent.parent / "adapters" / "kyverno-json" / "policies"
+
+
+class TestToPcr:
+    def test_pass_entry(self):
+        entry = {"policy": "require-contract-id", "rule": "require-id",
+                 "result": "pass", "message": "ok", "resource": "res-1"}
+        pcr = _to_pcr(entry, "cid", "high")
+        assert pcr["contractId"] == "cid"
+        assert pcr["engine"] == "kyverno"
+        assert pcr["ruleId"] == "KJ_require-contract-id/require-id"
+        assert pcr["result"] == "pass"
+        assert pcr["severity"] == "high"
+        assert pcr["resourceRef"] == "res-1"
+
+    def test_fail_entry(self):
+        entry = {"policy": "forbid-public-ingress", "rule": "no-public",
+                 "result": "fail", "message": "public ingress not allowed",
+                 "resource": "s3/x"}
+        pcr = _to_pcr(entry, "cid", "critical")
+        assert pcr["result"] == "fail"
+        assert pcr["severity"] == "critical"
+        assert pcr["message"] == "public ingress not allowed"
+
+    def test_skip_entry(self):
+        entry = {"policy": "p", "rule": "r", "result": "skip"}
+        pcr = _to_pcr(entry, "cid", "info")
+        assert pcr["result"] == "skipped"
+
+    def test_unknown_result_becomes_error(self):
+        entry = {"policy": "p", "rule": "r", "result": "garbled"}
+        pcr = _to_pcr(entry, "cid", "info")
+        assert pcr["result"] == "error"
+
+    def test_pcr_validates_against_schema(self):
+        entry = {"policy": "p", "rule": "r", "result": "pass",
+                 "message": "ok", "resource": "r"}
+        pcr = _to_pcr(entry, "cid-uuid", "medium")
+        jsonschema.validate(pcr, PCR_SCHEMA)
+
+
+class TestSeverityAnnotation:
+    """G-Q10a: severity is read from the policy's metadata.annotation."""
+
+    def test_policy_with_severity_annotation(self, tmp_path):
+        policy = {
+            "apiVersion": "json.kyverno.io/v1alpha1",
+            "kind": "ValidatingPolicy",
+            "metadata": {
+                "name": "test-sev",
+                "annotations": {"nova.cloudinit.dev/severity": "high"},
+            },
+            "spec": {"rules": [{"name": "r", "validate": {"assert": {"all": []}}}]},
+        }
+        p = tmp_path / "test-sev.json"
+        p.write_text(json.dumps(policy))
+        sevs = _load_policy_severities(tmp_path)
+        assert sevs.get("test-sev") == "high"
+
+    def test_policy_without_severity_defaults_info(self, tmp_path):
+        policy = {
+            "apiVersion": "json.kyverno.io/v1alpha1",
+            "kind": "ValidatingPolicy",
+            "metadata": {"name": "no-sev"},
+            "spec": {"rules": [{"name": "r", "validate": {"assert": {"all": []}}}]},
+        }
+        p = tmp_path / "no-sev.json"
+        p.write_text(json.dumps(policy))
+        sevs = _load_policy_severities(tmp_path)
+        assert sevs.get("no-sev") == "info"
+
+    def test_underscore_files_skipped(self, tmp_path):
+        # _smoke.json starts with _ — should be skipped.
+        (tmp_path / "_smoke.json").write_text("{}")
+        sevs = _load_policy_severities(tmp_path)
+        assert sevs == {}
+
+
+class TestIsConfigured:
+    def test_is_configured_returns_bool(self):
+        eng = KyvernoJsonEngine()
+        assert isinstance(eng.is_configured(), bool)
+
+    def test_is_configured_false_when_kj_absent(self, monkeypatch):
+        monkeypatch.setattr(_mod, "_which_kj", lambda: None)
+        eng = KyvernoJsonEngine()
+        assert eng.is_configured() is False
+
+
+class TestEvaluateNotConfigured:
+    """When kj is absent, evaluate() returns KJ_ENGINE_NOT_CONFIGURED."""
+
+    def test_evaluate_returns_skipped_when_not_configured(self, monkeypatch):
+        monkeypatch.setattr(_mod, "_which_kj", lambda: None)
+        eng = KyvernoJsonEngine()
+        out = eng.evaluate({"id": "x"}, Path("/tmp/policies"), "cid-1")
+        assert len(out) == 1
+        assert out[0]["ruleId"] == "KJ_ENGINE_NOT_CONFIGURED"
+        assert out[0]["result"] == "skipped"
+        jsonschema.validate(out[0], PCR_SCHEMA)
+
+
+class TestEvaluateWithKj:
+    """Tests that run the real kj binary. Skip when kj is not installed."""
+
+    @pytest.fixture(autouse=True)
+    def _require_kj(self):
+        if not _kj_installed():
+            pytest.skip("kj not installed (scripts/install-kyverno-json.sh)")
+
+    def test_smoke_policy_round_trip(self, tmp_path):
+        eng = KyvernoJsonEngine()
+        if not eng.is_configured():
+            pytest.skip("kj not configured")
+        # Use the real smoke policy dir.
+        out = eng.evaluate({"id": "msvc"}, _smoke_policy_dir(), "cid-smoke")
+        assert isinstance(out, list)
+        assert len(out) >= 1
+        for pcr in out:
+            jsonschema.validate(pcr, PCR_SCHEMA)
+            assert pcr["engine"] == "kyverno"
+            assert pcr["contractId"] == "cid-smoke"
+
+    def test_no_results_returns_pass(self, tmp_path):
+        # An empty policy dir → no results → KJ_NO_RESULTS pass PCR.
+        eng = KyvernoJsonEngine()
+        empty_dir = tmp_path / "empty"
+        empty_dir.mkdir()
+        out = eng.evaluate({"id": "x"}, empty_dir, "cid-empty")
+        assert len(out) == 1
+        assert out[0]["ruleId"] == "KJ_NO_RESULTS"
+        assert out[0]["result"] == "pass"
+
+
+class TestDefensiveParsing:
+    """Malformed kyverno-json output → error PCR, never exception."""
+
+    def test_malformed_output_produces_error_pcr(self, monkeypatch):
+        eng = KyvernoJsonEngine()
+        # Mock is_configured → True, then mock subprocess to return
+        # garbage output.
+        monkeypatch.setattr(_mod, "_which_kj", lambda: "/fake/kj")
+        monkeypatch.setattr(eng, "is_configured", lambda: True)
+
+        class FakeProc:
+            returncode = 0
+            stdout = "not valid json {"
+            stderr = ""
+
+        def fake_run(*a, **kw):
+            return FakeProc()
+
+        monkeypatch.setattr(_mod.subprocess, "run", fake_run)
+        out = eng.evaluate({"id": "x"}, _smoke_policy_dir(), "cid-bad")
+        assert len(out) == 1
+        assert out[0]["result"] == "error"
+        assert out[0]["ruleId"] == "KJ_ENGINE_ERROR"
+        jsonschema.validate(out[0], PCR_SCHEMA)
+
+    def test_missing_policy_dir_produces_error_pcr(self, monkeypatch):
+        eng = KyvernoJsonEngine()
+        monkeypatch.setattr(_mod, "_which_kj", lambda: "/fake/kj")
+        monkeypatch.setattr(eng, "is_configured", lambda: True)
+        out = eng.evaluate({"id": "x"}, Path("/nonexistent/dir"), "cid-miss")
+        assert len(out) == 1
+        assert out[0]["result"] == "error"
+        assert "not found" in out[0]["message"]
@@ -0,0 +1,125 @@
+"""Tests for core/policy_engine.py (REQ-308, v1.25).
+
+Protocol conformance, registry selection, NullEngine fallback,
+unknown-engine KeyError, and the NullEngine-satisfies-Protocol
+assertion (G-Q8a — proves the swap boundary is real without
+implementing OPA).
+"""
+
+import json
+import os
+import sys
+from pathlib import Path
+from unittest import mock
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import core.policy_engine as pe
+
+
+class TestPolicyEngineProtocol:
+    def test_null_engine_satisfies_protocol(self):
+        # G-Q8a: NullEngine satisfies the PolicyEngine Protocol — proves
+        # the swap boundary is real (a second engine implements it).
+        eng = pe.NullEngine()
+        assert isinstance(eng, pe.PolicyEngine)
+
+    def test_null_engine_is_configured_false(self):
+        assert pe.NullEngine().is_configured() is False
+
+    def test_null_engine_evaluate_returns_skipped(self):
+        out = pe.NullEngine().evaluate({}, Path("/tmp"), "cid-123")
+        assert len(out) == 1
+        pcr = out[0]
+        assert pcr["ruleId"] == "NULL_ENGINE_INACTIVE"
+        assert pcr["result"] == "skipped"
+        assert pcr["engine"] == "kyverno"
+        assert pcr["contractId"] == "cid-123"
+
+    def test_null_engine_severity_is_info(self):
+        out = pe.NullEngine().evaluate({}, Path("/tmp"), "cid")
+        assert out[0]["severity"] == "info"
+
+
+class TestRegistry:
+    def test_register_and_get(self, tmp_path, monkeypatch):
+        # Register a stub engine and verify get_engine() returns it.
+        class StubEngine:
+            name = "stub"
+
+            def is_configured(self) -> bool:
+                return True
+
+            def evaluate(self, payload, policy_dir, contract_id):
+                return [{"contractId": contract_id, "engine": "kyverno",
+                         "ruleId": "STUB", "result": "pass", "severity": "info",
+                         "message": "", "evaluatedAt": "t", "resourceRef": "",
+                         "evidence": {}}]
+
+        pe._REGISTRY.clear()
+        pe.register("stub", StubEngine)
+        monkeypatch.setattr(pe, "_load_config_policy", lambda: {"engine": "stub"})
+        eng = pe.get_engine()
+        assert eng.name == "stub"
+        pe._REGISTRY.clear()
+        pe._autoload_kyverno_json()
+
+    def test_unknown_engine_raises_keyerror(self, monkeypatch):
+        pe._REGISTRY.clear()
+        monkeypatch.setattr(pe, "_load_config_policy",
+                            lambda: {"engine": "nonexistent"})
+        with pytest.raises(KeyError, match="Unknown policy engine"):
+            pe.get_engine()
+        pe._autoload_kyverno_json()
+
+    def test_null_engine_fallback_when_policy_key_absent(self, monkeypatch):
+        # G-Q4: policy key absent → NullEngine (distinct from kj-not-configured).
+        monkeypatch.setattr(pe, "_load_config_policy", lambda: None)
+        eng = pe.get_engine()
+        assert isinstance(eng, pe.NullEngine)
+        assert eng.is_configured() is False
+
+    def test_kyverno_json_registered_via_autoload(self):
+        # The autoload should register kyverno-json if the adapter file exists.
+        pe._autoload_kyverno_json()
+        assert "kyverno-json" in pe._REGISTRY or len(pe._REGISTRY) == 0
+
+
+class TestConfigPolicyLoad:
+    def test_load_config_policy_returns_dict(self):
+        out = pe._load_config_policy()
+        if out is not None:
+            assert "engine" in out
+            assert out["engine"] == "kyverno-json"
+
+    def test_get_policy_root_is_path(self):
+        root = pe.get_policy_root()
+        assert isinstance(root, Path)
+        assert root.name == "policies" or str(root).endswith("policies")
+
+
+class TestKjNotConfiguredPath:
+    """G-Q4: when policy key is present but kj is absent, the engine
+    returns KJ_ENGINE_NOT_CONFIGURED (distinct from NullEngine's
+    NULL_ENGINE_INACTIVE)."""
+
+    def test_kj_not_configured_returns_distinct_ruleid(self, monkeypatch):
+        # Force the registry to return KyvernoJsonEngine, then mock
+        # `which kj` to return None.
+        pe._autoload_kyverno_json()
+        if "kyverno-json" not in pe._REGISTRY:
+            pytest.skip("kyverno-json adapter not loadable in this env")
+        monkeypatch.setattr(pe, "_load_config_policy",
+                            lambda: {"engine": "kyverno-json"})
+        eng = pe.get_engine()
+        # Mock is_configured → False
+        with mock.patch.object(eng, "is_configured", return_value=False):
+            out = eng.evaluate({}, Path("/tmp"), "cid-456")
+        assert len(out) == 1
+        assert out[0]["ruleId"] == "KJ_ENGINE_NOT_CONFIGURED"
+        assert out[0]["result"] == "skipped"
+        assert out[0]["contractId"] == "cid-456"
+        # Distinct from NullEngine
+        assert out[0]["ruleId"] != "NULL_ENGINE_INACTIVE"