Jon Chery
3d9dd06411
---ci--- project: acdl phase: 13 milestone: v1.14 status: complete requirements: covered: [REQ-147] partial: [] ---/ci---
131 lines
4.5 KiB
Python
131 lines
4.5 KiB
Python
"""Kyverno adapter — translate Kyverno PolicyReport results to ACDL 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)) |