acdl/adapters/kyverno-json/kyverno_json_engine.py

"""Nova KyvernoJsonEngine (REQ-293, v1.25; fixed v1.26 P3 W0.5).

Implements the ``PolicyEngine`` protocol (``core/policy_engine.py``)
by shelling to the ``kj`` CLI (``kyverno-json``). Translates native
kyverno-json scan output to Nova ``PolicyCheckResult`` dicts
(``schemas/policy_check_result.schema.json``).

Engine enum reuse (D-116): records carry ``engine: "kyverno"`` (no new
enum value). The ``ruleId`` is prefixed ``KJ_<policy_name>`` to
distinguish from the K8s Kyverno adapter's ``KYVERNO_`` prefix.

Severity (RESEARCH §2.6, G-Q10a): kyverno-json does not natively assign
severities. Each Nova policy declares its severity via a
``metadata.annotations["nova.cloudinit.dev/severity"]`` field. The
engine reads this annotation from the loaded policy file (not from the
scan result — the result carries the policy spec but the annotation is
read here from disk) and applies it to every result that policy
produces. Default when absent: ``"info"``.

Graceful degradation (D-120): ``is_configured()`` returns ``False`` when
``which kj`` is absent → ``evaluate()`` returns a single SKIPPED PCR
(``ruleId: KJ_ENGINE_NOT_CONFIGURED``). The platform functions without
the binary.

Defensive parsing: any kyverno-json output that doesn't match the
expected shape produces an ``error`` PCR, never an exception. The
engine is read-only against a local policy dir + a temp payload file.

v1.26 P3 W0.5 fix — three substrate bugs uncovered once ``kj`` was
actually installed (the v1.25 test suite ``pytest.skip``-masked them):

1. **``.json`` policy files are not loaded by ``kj`` v0.0.3.** The
   upstream policy loader (``pkg/policy/load.go``) uses
   ``fileinfo.IsYaml()`` which only matches ``.yaml``/``.yml``
   extensions — ``.json`` files are silently skipped, yielding
   ``evaluating N resources against 0 policies``. Nova policies are
   authored as ``.json`` (the ``TestPolicyFilesExist`` tests assert the
   ``.json`` filenames). Fix: ``evaluate()`` materializes a temp policy
   dir that mirrors the source tree with every ``.json`` policy copied
   to a ``.yaml`` twin (JSON is a valid YAML subset — verified against
   ``kj`` v0.0.3). The source ``.json`` files remain untouched.

2. **Bare-list output format.** ``kj scan --output json`` emits a bare
   JSON list at the top level (NOT ``{"results": [...]}``). Each entry
   has ``resource`` (the evaluated payload) + ``results`` (list of
   per-policy result objects, each carrying ``policy.metadata.name``,
   ``rules[]`` with ``rule.name``, ``violations[]`` (present on fail),
   ``error`` (string, present on policy-evaluation error)). The v1.25
   ``_translate`` did ``out.get("results", [])`` on a dict — but
   ``out`` is a list → returned ``[]`` → emitted a single
   ``KJ_NO_RESULTS`` pass PCR. **This is why all failing fixtures showed
   0 fails.** Fix: ``_translate`` handles list (v0.0.3) and dict
   (future-proof) shapes.

3. **``validate`` wrapper + check syntax.** Documented in the policy
   files themselves (see the W0.5 policy edits). The engine itself does
   not enforce policy shape — it only translates ``kj`` output — so
   this fix lives in the policy ``.json`` files.
"""

import datetime
import json
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Any, Union

import yaml


Payload = Union[dict, list, str]

SEVERITY_DEFAULT = "info"
SEVERITY_ANNOTATION = "nova.cloudinit.dev/severity"

RESULT_MAP = {
    "pass": "pass",
    "fail": "fail",
    "error": "error",
    "skip": "skipped",
    "skipped": "skipped",
    "warn": "skipped",
    "warning": "skipped",
}


def _iso8601_now() -> str:
    return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _which_kj() -> str | None:
    """Return the path to ``kj`` if on PATH, else ``None``."""
    return shutil.which("kj")


def _load_policy_severities(policy_dir: Path) -> dict[str, str]:
    """Load each ``.json``/``.yaml``/``.yml`` policy in ``policy_dir``
    (non-recursive) and return ``{policy_name: severity}``.

    kyverno-json policies are Kubernetes-style ``ValidatingPolicy``
    resources. The severity is read from
    ``metadata.annotations["nova.cloudinit.dev/severity"]``. Policies
    in subdirectories (e.g. ``contract/``, ``stack-ir/``) are loaded
    when the caller passes that subdirectory as ``policy_dir``.
    """
    severities: dict[str, str] = {}
    if not policy_dir.is_dir():
        return severities
    for entry in sorted(os.listdir(policy_dir)):
        if entry.startswith("_") or entry.startswith("."):
            continue
        full = policy_dir / entry
        if not full.is_file():
            continue
        if entry.endswith((".json", ".yaml", ".yml")):
            try:
                with open(full, "r", encoding="utf-8") as fh:
                    doc = yaml.safe_load(fh)
                if not isinstance(doc, dict):
                    continue
                name = doc.get("metadata", {}).get("name") or entry.rsplit(".", 1)[0]
                ann = doc.get("metadata", {}).get("annotations", {}) or {}
                sev = ann.get(SEVERITY_ANNOTATION, SEVERITY_DEFAULT)
                severities[name] = str(sev).lower()
            except Exception:
                continue
    return severities


def _materialize_yaml_policy_dir(src: Path) -> tuple[Path, bool]:
    """Mirror ``src`` (recursively) into a temp dir, copying every
    ``.json`` policy to a ``.yaml`` twin and copying ``.yaml``/``.yml``
    files verbatim. Returns ``(temp_dir, created)``.

    ``kj`` v0.0.3's policy loader (``pkg/policy/load.go``) only matches
    ``.yaml``/``.yml`` extensions — ``.json`` files are silently
    skipped. Nova policies are authored as ``.json`` (the
    ``TestPolicyFilesExist`` tests assert the ``.json`` filenames, so
    they cannot be renamed in-place). JSON is a valid YAML subset, so
    a byte-for-byte copy with a ``.yaml`` extension loads cleanly.

    ``created`` is ``False`` when ``src`` contains no policy files at
    all (empty dir) — in that case the temp dir is still returned (the
    caller invokes ``kj`` against it and gets the no-results path).
    """
    tmp = Path(tempfile.mkdtemp(prefix="nova-kj-pol-"))
    any_policy = False
    if src.is_dir():
        for root, _dirs, files in os.walk(src):
            rel = Path(root).relative_to(src)
            dest_root = tmp / rel
            dest_root.mkdir(parents=True, exist_ok=True)
            for fn in files:
                if fn.startswith(".") or fn.startswith("_"):
                    continue
                src_file = Path(root) / fn
                if fn.endswith(".json"):
                    dest_file = dest_root / (fn.rsplit(".", 1)[0] + ".yaml")
                    shutil.copy2(src_file, dest_file)
                    any_policy = True
                elif fn.endswith((".yaml", ".yml")):
                    shutil.copy2(src_file, dest_root / fn)
                    any_policy = True
    return tmp, any_policy


def _skipped_not_configured(contract_id: str) -> dict:
    return {
        "contractId": contract_id,
        "evaluatedAt": _iso8601_now(),
        "engine": "kyverno",
        "ruleId": "KJ_ENGINE_NOT_CONFIGURED",
        "severity": "info",
        "result": "skipped",
        "message": (
            "kyverno-json engine not configured — `which kj` returned no path. "
            "Install via scripts/install-kyverno-json.sh. The platform proceeds "
            "with a neutral SKIPPED policy input (is_configured() guard, D-120)."
        ),
        "evidence": {},
        "resourceRef": "",
    }


def _error_pcr(contract_id: str, message: str) -> dict:
    return {
        "contractId": contract_id,
        "evaluatedAt": _iso8601_now(),
        "engine": "kyverno",
        "ruleId": "KJ_ENGINE_ERROR",
        "severity": "info",
        "result": "error",
        "message": message,
        "evidence": {},
        "resourceRef": "",
    }


def _no_results_pass(contract_id: str) -> dict:
    """No result entries — emit a single pass PCR so the confidence
    signal's policy input is non-empty (a non-empty list of passes →
    score 1.0)."""
    return {
        "contractId": contract_id,
        "evaluatedAt": _iso8601_now(),
        "engine": "kyverno",
        "ruleId": "KJ_NO_RESULTS",
        "severity": "info",
        "result": "pass",
        "message": "kyverno-json scan produced no result entries (all policies passed or no match).",
        "evidence": {},
        "resourceRef": "",
    }


class KyvernoJsonEngine:
    """``PolicyEngine`` impl that shells to the ``kj`` CLI."""

    name = "kyverno-json"

    def is_configured(self) -> bool:
        return _which_kj() is not None

    def evaluate(self, payload: Payload, policy_dir: Path,
                 contract_id: str) -> list[dict]:
        if not self.is_configured():
            return [_skipped_not_configured(contract_id)]
        kj = _which_kj()
        policy_dir = Path(policy_dir)
        if not policy_dir.is_dir():
            return [_error_pcr(
                contract_id,
                f"kyverno-json policy dir not found: {policy_dir}",
            )]
        severities = _load_policy_severities(policy_dir)
        # kj v0.0.3 only loads .yaml/.yml policy files. Mirror the tree
        # to a temp dir with .json policies copied to .yaml twins.
        yaml_dir, _any_policy = _materialize_yaml_policy_dir(policy_dir)
        # Write payload to temp file (kj scan --payload expects a file path).
        payload_tmp = tempfile.NamedTemporaryFile(
            mode="w", suffix=".json", delete=False, encoding="utf-8"
        )
        try:
            json.dump(payload, payload_tmp)
            payload_tmp.flush()
            payload_tmp.close()
            cmd = [
                kj, "scan",
                "--policy", str(yaml_dir),
                "--payload", payload_tmp.name,
                "--output", "json",
            ]
            try:
                proc = subprocess.run(
                    cmd, capture_output=True, text=True, timeout=60,
                )
            except subprocess.TimeoutExpired:
                return [_error_pcr(contract_id, "kyverno-json scan timed out (60s)")]
            if proc.returncode not in (0, 1):
                return [_error_pcr(
                    contract_id,
                    f"kyverno-json scan exited {proc.returncode}: {proc.stderr[:200]}",
                )]
            try:
                out = json.loads(proc.stdout) if proc.stdout.strip() else []
            except json.JSONDecodeError as e:
                return [_error_pcr(
                    contract_id,
                    f"kyverno-json output not JSON: {e}",
                )]
            return self._translate(out, contract_id, severities)
        finally:
            try:
                os.unlink(payload_tmp.name)
            except OSError:
                pass
            shutil.rmtree(yaml_dir, ignore_errors=True)

    def _translate(self, out: Any, contract_id: str,
                   severities: dict[str, str]) -> list[dict]:
        # kj v0.0.3 emits a BARE JSON LIST at the top level: each entry
        # has `resource` (the evaluated payload) + `results` (list of
        # per-policy result objects). Future-proof: also accept the
        # legacy {"results": [...]} dict shape.
        if isinstance(out, list):
            entries = out
        elif isinstance(out, dict):
            entries = out.get("results", [])
            if not isinstance(entries, list):
                entries = []
        else:
            entries = []
        pcrs: list[dict] = []
        for entry in entries:
            if not isinstance(entry, dict):
                continue
            resource = entry.get("resource", {})
            results = entry.get("results", [])
            if not isinstance(results, list):
                results = []
            for pol_result in results:
                if not isinstance(pol_result, dict):
                    continue
                policy_obj = pol_result.get("policy", {}) or {}
                policy_name = (
                    policy_obj.get("metadata", {}).get("name") if isinstance(policy_obj, dict)
                    else None
                ) or "UNKNOWN"
                severity = severities.get(policy_name, SEVERITY_DEFAULT)
                rules = pol_result.get("rules", [])
                if not isinstance(rules, list):
                    rules = []
                for rule_entry in rules:
                    if not isinstance(rule_entry, dict):
                        continue
                    rule_obj = rule_entry.get("rule", {}) or {}
                    rule_name = rule_obj.get("name", "") if isinstance(rule_obj, dict) else ""
                    rule_id = f"KJ_{policy_name}"
                    if rule_name:
                        rule_id = f"{rule_id}/{rule_name}"
                    violations = rule_entry.get("violations")
                    error_str = rule_entry.get("error")
                    if isinstance(violations, list) and violations:
                        # Fail: build a message from the violations' errors.
                        msg_parts: list[str] = []
                        for v in violations:
                            if not isinstance(v, dict):
                                continue
                            for err in v.get("errors", []) or []:
                                if not isinstance(err, dict):
                                    continue
                                field = err.get("field", "")
                                detail = err.get("detail", "")
                                value = err.get("value", "")
                                msg_parts.append(
                                    f"{field}: value={value!r} detail={detail}"
                                )
                        message = "; ".join(msg_parts) if msg_parts else "policy rule failed"
                        pcrs.append({
                            "contractId": contract_id,
                            "evaluatedAt": _iso8601_now(),
                            "engine": "kyverno",
                            "ruleId": rule_id,
                            "severity": severity,
                            "result": "fail",
                            "message": message,
                            "evidence": {
                                "resource": resource,
                                "policy": policy_name,
                                "rule": rule_name,
                                "violations": violations,
                            },
                            "resourceRef": _resource_ref(resource),
                        })
                    elif isinstance(error_str, str) and error_str:
                        # Policy-evaluation error (e.g. bad JMESPath).
                        pcrs.append({
                            "contractId": contract_id,
                            "evaluatedAt": _iso8601_now(),
                            "engine": "kyverno",
                            "ruleId": rule_id,
                            "severity": severity,
                            "result": "error",
                            "message": error_str,
                            "evidence": {
                                "resource": resource,
                                "policy": policy_name,
                                "rule": rule_name,
                            },
                            "resourceRef": _resource_ref(resource),
                        })
                    else:
                        # Pass: no violations, no error.
                        pcrs.append({
                            "contractId": contract_id,
                            "evaluatedAt": _iso8601_now(),
                            "engine": "kyverno",
                            "ruleId": rule_id,
                            "severity": severity,
                            "result": "pass",
                            "message": "",
                            "evidence": {
                                "resource": resource,
                                "policy": policy_name,
                                "rule": rule_name,
                            },
                            "resourceRef": _resource_ref(resource),
                        })
        if not pcrs:
            pcrs.append(_no_results_pass(contract_id))
        return pcrs


def _resource_ref(resource: Any) -> str:
    """Best-effort resource ref from the evaluated payload."""
    if isinstance(resource, dict):
        for key in ("id", "name", "address"):
            v = resource.get(key)
            if isinstance(v, str) and v:
                return v
    return ""


# --- Legacy _to_pcr kept for the existing TestToPcr unit tests ---
# (test_kyverno_json_engine.py::TestToPcr constructs flat `entry`
# dicts with `policy`/`rule`/`result`/`message`/`resource` keys and
# asserts the translated PCR shape. The production _translate path no
# longer calls this helper — it inlines the translation against the
# real kj v0.0.3 nested output — but the unit tests pin the helper's
# contract, so it stays.)


def _to_pcr(entry: dict, contract_id: str, severity: str) -> dict:
    """Translate a flat kyverno-json scan result entry to a PCR dict.

    Legacy shape (kept for unit-test backwards compatibility): the
    entry is a flat dict with ``policy``/``rule``/``result``/``message``/
    ``resource`` string keys. The production ``_translate`` path no
    longer calls this — it inlines translation against the real kj
    v0.0.3 nested ``resource``+``results``+``rules`` shape — but the
    ``TestToPcr`` unit tests pin this contract.
    """
    policy_name = entry.get("policy", "") or "UNKNOWN"
    rule_name = entry.get("rule", "") or ""
    rule_id = f"KJ_{policy_name}"
    if rule_name:
        rule_id = f"{rule_id}/{rule_name}"
    result_raw = entry.get("result", "skip")
    result = RESULT_MAP.get(str(result_raw).lower(), "error")
    message = entry.get("message", "") or ""
    resource = entry.get("resource", "")
    if not resource and entry.get("name"):
        kind = entry.get("kind", "")
        ns = entry.get("namespace", "")
        resource = f"{kind}/{ns}/{entry.get('name')}" if kind else entry.get("name", "")
    return {
        "contractId": contract_id,
        "evaluatedAt": _iso8601_now(),
        "engine": "kyverno",
        "ruleId": rule_id,
        "severity": severity,
        "result": result,
        "message": message,
        "evidence": {
            "resource": resource,
            "policy": policy_name,
            "rule": rule_name,
            "namespace": entry.get("namespace", ""),
            "kind": entry.get("kind", ""),
            "name": entry.get("name", ""),
        },
        "resourceRef": resource,
    }


if __name__ == "__main__":
    if len(sys.argv) < 4:
        print(
            "usage: kyverno_json_engine.py <payload.json> <policy_dir> <contract-id>",
            file=sys.stderr,
        )
        sys.exit(2)
    with open(sys.argv[1], "r", encoding="utf-8") as fh:
        pl = json.load(fh)
    engine = KyvernoJsonEngine()
    out = engine.evaluate(pl, Path(sys.argv[2]), sys.argv[3])
    print(json.dumps(out, indent=2))