acdl/adapters/kyverno-json/kyverno_json_engine.py

"""Nova KyvernoJsonEngine (REQ-293, v1.25).

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 YAML (not from the
scan result — the result doesn't carry it) 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.
"""

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 _to_pcr(entry: dict, contract_id: str, severity: str) -> dict:
    """Translate a kyverno-json scan result entry to a PCR dict."""
    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,
    }


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": "",
    }


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)
        # 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(policy_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

    def _translate(self, out: dict, contract_id: str,
                   severities: dict[str, str]) -> list[dict]:
        results = out.get("results", []) if isinstance(out, dict) else []
        if not isinstance(results, list):
            results = []
        pcrs: list[dict] = []
        for entry in results:
            if not isinstance(entry, dict):
                continue
            policy_name = entry.get("policy", "") or "UNKNOWN"
            severity = severities.get(policy_name, SEVERITY_DEFAULT)
            pcrs.append(_to_pcr(entry, contract_id, severity))
        if not pcrs:
            # No results — kyverno-json produced nothing (no match, or
            # all policies passed with 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).
            pcrs.append({
                "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": "",
            })
        return pcrs


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))