acdl/adapters/kyverno/kyverno_adapter.py

"""Kyverno adapter — translate Kyverno PolicyReport results to Nova PolicyCheckResult records.

Kyverno is a Kubernetes-native policy engine. It evaluates K8s manifests
and produces PolicyReport resources. This adapter translates those results
to the normalized PolicyCheckResult schema (engine: "kyverno").

v1.9 (REQ-111): the translator is fleshed out — full PolicyReport →
PolicyCheckResult mapping with severity + skip-with-reason handling. It
remains inactive for Terraform-only stacks (guard preserved — emits a
single SKIPPED `KYVERNO_INACTIVE_TF_STACK` record when no K8s manifests).
A `--kube-version` flag was previously parsed but never used. It has been
removed (v1.14, G-103) to resolve the stub. Version-aware policy selection
will be added when the GitOps reconciler emits K8s manifests (D-053
roadmap). The adapter is inactive for Terraform-only stacks today.

D-053: the platform emits Terraform, not K8s manifests. This adapter
activates when the GitOps reconciler (roadmap) emits K8s manifests.
Sample policies are included as documentation at adapters/kyverno/policies/.

CLI: kyverno_adapter.py <policyreport.json> <contract-id>
"""

import datetime
import json
import os
import sys


SEVERITY_MAP = {
    "critical": "critical",
    "high": "high",
    "medium": "medium",
    "low": "low",
    "info": "info",
    "informational": "info",
}

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


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


def _to_pcr(entry, contract_id):
    severity_raw = entry.get("severity", "info")
    severity = SEVERITY_MAP.get(str(severity_raw).lower(), "info")
    result_raw = entry.get("result", "skip")
    result = RESULT_MAP.get(str(result_raw).lower(), "error")
    # Skip-with-reason: a skipped result carries a message that explains why.
    message = entry.get("message", "")
    if result == "skipped" and not message:
        message = entry.get("skipReason", entry.get("skippedMessage", "skipped (no reason)"))
    policy = entry.get("policy", "")
    rule = entry.get("rule", "")
    rule_id = f"{policy}/{rule}" if rule else (policy or "KYVERNO_UNKNOWN")
    resource = entry.get("resource", "")
    if not resource and entry.get("name"):
        # Construct a resource ref from kind/name/namespace when present.
        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,
            "namespace": entry.get("namespace", ""),
            "kind": entry.get("kind", ""),
            "name": entry.get("name", ""),
            "policy": policy,
            "rule": rule,
        },
        "resourceRef": resource,
    }


def _emit_inactive_tf(contract_id):
    """Emit a SKIPPED record when the platform emits Terraform, not K8s manifests."""
    return {
        "contractId": contract_id,
        "evaluatedAt": _iso8601_now(),
        "engine": "kyverno",
        "ruleId": "KYVERNO_INACTIVE_TF_STACK",
        "severity": "info",
        "result": "skipped",
        "message": "Kyverno inactive — the platform emits Terraform, not K8s manifests. Activates when the GitOps reconciler emits K8s manifests (D-053).",
        "evidence": {},
        "resourceRef": "",
    }


def adapt(policyreport_json_path, contract_id):
    with open(policyreport_json_path, "r", encoding="utf-8") as fh:
        data = json.load(fh)
    out = []
    # Kyverno PolicyReport has a .results[] array.
    results = data.get("results", [])
    if not isinstance(results, list):
        results = []
    for entry in results:
        out.append(_to_pcr(entry, contract_id))
    if not out:
        out.append(_emit_inactive_tf(contract_id))
    return out


def adapt_inactive(contract_id):
    """Convenience: emit the inactive-for-TF record directly (no report file)."""
    return [_emit_inactive_tf(contract_id)]


if __name__ == "__main__":
    args = sys.argv[1:]
    if len(args) != 2:
        print("usage: kyverno_adapter.py <policyreport.json> <contract-id>", file=sys.stderr)
        sys.exit(2)
    print(json.dumps(adapt(args[0], args[1]), indent=2))