feat(P23): tagging standard + Wiz adapter + Kyverno adapter

Phase 23 (v1.7) — tagging standards and security adapters.

* schemas/tagging-standard.json (D-054): canonical required-tags schema
  (acdl:owner, acdl:contract, acdl:environment, acdl:cost-center).
* adapters/terraform/policy/custom_rules/acdl_tagging.py: Checkov custom
  rule (ACDL_TAG_NAMING) loaded via --external-checks-dir; closes D-043
  (synthetic SKIPPED record replaced by real PASS/FAIL records).
* checkov_adapter.py: removed _emit_tag_naming_skipped(), added
  ACDL_TAG_NAMING to RULE_MAP, updated docstring.
* scripts/run_platform.sh: both Checkov invocations pass
  --external-checks-dir adapters/terraform/policy/custom_rules/.
* adapters/wiz/ (D-052): Wiz adapter translating issue records to
  PolicyCheckResult (engine: "wiz"); graceful degradation emits
  WIZ_NOT_CONFIGURED SKIPPED when unconfigured; is_configured() gate.
* adapters/kyverno/ (D-053): Kyverno adapter translating PolicyReport
  results to PolicyCheckResult (engine: "kyverno"); ready but inactive
  for Terraform-only stacks; 3 sample ClusterPolicies in policies/.
* schemas/policy_check_result.schema.json: engine enum += "wiz".
* tests: fixtures + test_wiz_adapter.py (8 tests) + test_kyverno_adapter.py
  (13 tests); updated test_checkov_adapter.py to not expect the removed
  synthetic ACDL_TAG_NAMING SKIPPED record.
* scripts/run_ci.sh: lint stage compiles the new adapter modules.

202 tests pass; CI pipeline OK (lint + test + check-only).

Deviations:
- Wiz adapt() had an AttributeError on bare-list top-level input
  (data.get() on a list); fixed to dispatch on isinstance(data, list)
  before calling .get(). No spec change — bare-list handling is implied
  by the original docstring's "data if isinstance(data, list)" branch.
- Kyverno _to_pcr({}) defaults result to "skipped" (entry.get("result",
  "skip") -> "skip"), not "error"; test expectation corrected. Added an
  explicit unknown-result-string test to cover the "error" fallback.

---ci---
project: acdl
phase: 23
milestone: v1.7
status: execute
---/ci---
This commit is contained in:
Jon Chery
2026-07-22 20:00:46 +00:00
parent dca35c78ec
commit 1fd37a2843
22 changed files with 935 additions and 31 deletions
+40
View File
@@ -0,0 +1,40 @@
{
"apiVersion": "wgpolicyk8s.io/v1alpha1",
"kind": "PolicyReport",
"metadata": {
"name": "acdl-policy-report",
"namespace": "default"
},
"results": [
{
"policy": "disallow-privileged-containers",
"severity": "high",
"result": "pass",
"message": "Pod spec is compliant (no privileged container).",
"resource": "default/Pod/acdl-app",
"namespace": "default",
"kind": "Pod",
"name": "acdl-app"
},
{
"policy": "require-resource-labels",
"severity": "medium",
"result": "fail",
"message": "Pod missing required label acdl:owner.",
"resource": "default/Pod/acdl-bad-app",
"namespace": "default",
"kind": "Pod",
"name": "acdl-bad-app"
},
{
"policy": "require-image-digests",
"severity": "high",
"result": "warn",
"message": "Container image uses a mutable tag; consider pinning to a digest.",
"resource": "default/Pod/acdl-app",
"namespace": "default",
"kind": "Pod",
"name": "acdl-app"
}
]
}
+52
View File
@@ -0,0 +1,52 @@
{
"issues": [
{
"id": "wiz-issue-001",
"title": "Publicly exposed S3 bucket with sensitive data",
"severity": "CRITICAL",
"status": "OPEN",
"control": {
"id": "wiz-control-public-s3",
"name": "Public S3 bucket exposure"
},
"entity": {
"id": "arn:aws:s3:::acdl-leaked-bucket",
"name": "acdl-leaked-bucket",
"cloudPlatform": "AWS",
"subscriptionId": "111111111111"
}
},
{
"id": "wiz-issue-002",
"title": "IAM role with overly broad permissions",
"severity": "HIGH",
"status": "RESOLVED",
"control": {
"id": "wiz-control-iam-broad",
"name": "Overly broad IAM role"
},
"entity": {
"id": "arn:aws:iam::111111111111:role/acdl-broad-role",
"name": "acdl-broad-role",
"cloudPlatform": "AWS",
"subscriptionId": "111111111111"
}
},
{
"id": "wiz-issue-003",
"title": "Security group allows 0.0.0.0/0 on port 22",
"severity": "MEDIUM",
"status": "IN_PROGRESS",
"control": {
"id": "wiz-control-ssh-open",
"name": "SSH open to the world"
},
"entity": {
"id": "arn:aws:ec2:us-east-1:111111111111:security-group/sg-abcdef",
"name": "acdl-ssh-sg",
"cloudPlatform": "AWS",
"subscriptionId": "111111111111"
}
}
]
}
+12 -11
View File
@@ -8,7 +8,7 @@ import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from adapters.terraform.policy.checkov_adapter import (
RULE_MAP, _to_pcr, _emit_tag_naming_skipped, adapt,
RULE_MAP, _to_pcr, adapt,
)
@@ -66,12 +66,11 @@ class TestToPcr:
jsonschema.validate(pcr, policy_check_result_schema)
class TestTagNamingSkipped:
def test_skipped_pcr(self):
pcr = _emit_tag_naming_skipped("c-1")
assert pcr["result"] == "skipped"
assert pcr["ruleId"] == "ACDL_TAG_NAMING"
assert pcr["severity"] == "info"
class TestRuleMapTagging:
def test_acdl_tag_naming_is_real_rule(self):
# D-054 / D-043 closure: ACDL_TAG_NAMING is now a real custom Checkov
# rule, not a synthetic SKIPPED record.
assert RULE_MAP["ACDL_TAG_NAMING"] == ("tagging-standard", "medium")
class TestAdapt:
@@ -99,14 +98,16 @@ class TestAdapt:
results = adapt(str(f), "c-1")
assert isinstance(results, list)
def test_adapt_includes_tag_naming(self, tmp_path):
def test_adapt_does_not_emit_synthetic_tag_naming(self, tmp_path):
# D-043 closure: adapt() no longer appends a synthetic SKIPPED
# ACDL_TAG_NAMING record. The custom Checkov rule (loaded via
# --external-checks-dir) produces real PASS/FAIL records instead.
data = self._sample_checkov_json()
f = tmp_path / "checkov.json"
f.write_text(json.dumps(data))
results = adapt(str(f), "c-1")
tag = [r for r in results if r["ruleId"] == "ACDL_TAG_NAMING"]
assert len(tag) == 1
assert tag[0]["result"] == "skipped"
assert tag == [] # no synthetic record
def test_adapt_has_passed_and_failed(self, tmp_path):
data = self._sample_checkov_json()
@@ -123,4 +124,4 @@ class TestAdapt:
f = tmp_path / "checkov.json"
f.write_text(json.dumps(data))
results = adapt(str(f), "c-1")
assert len(results) == 1 # just the tag naming skipped
assert results == [] # no synthetic tag-naming record anymore
+141
View File
@@ -0,0 +1,141 @@
import json
import sys
from pathlib import Path
import jsonschema
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from adapters.kyverno.kyverno_adapter import (
SEVERITY_MAP, RESULT_MAP, _to_pcr, adapt,
)
FIXTURES = Path(__file__).resolve().parent.parent / "tests" / "fixtures"
class TestSeverityResultMaps:
def test_severity_map(self):
assert SEVERITY_MAP["critical"] == "critical"
assert SEVERITY_MAP["high"] == "high"
assert SEVERITY_MAP["medium"] == "medium"
assert SEVERITY_MAP["low"] == "low"
assert SEVERITY_MAP["info"] == "info"
def test_result_map(self):
assert RESULT_MAP["pass"] == "pass"
assert RESULT_MAP["fail"] == "fail"
assert RESULT_MAP["warn"] == "skipped"
assert RESULT_MAP["error"] == "error"
assert RESULT_MAP["skip"] == "skipped"
class TestToPcr:
def test_translates_pass(self):
entry = {"policy": "p1", "severity": "high", "result": "pass",
"message": "ok", "resource": "ns/Pod/x"}
pcr = _to_pcr(entry, "c-1")
assert pcr["engine"] == "kyverno"
assert pcr["ruleId"] == "p1"
assert pcr["severity"] == "high"
assert pcr["result"] == "pass"
assert pcr["resourceRef"] == "ns/Pod/x"
def test_warn_maps_to_skipped(self):
entry = {"policy": "p1", "severity": "medium", "result": "warn",
"resource": "r"}
pcr = _to_pcr(entry, "c-1")
assert pcr["result"] == "skipped"
def test_unknown_severity_defaults_info(self):
entry = {"policy": "p1", "severity": "BOGUS", "result": "fail",
"resource": "r"}
pcr = _to_pcr(entry, "c-1")
assert pcr["severity"] == "info"
def test_unknown_result_defaults_error(self):
entry = {"policy": "p1", "severity": "low", "result": "BOGUS",
"resource": "r"}
pcr = _to_pcr(entry, "c-1")
assert pcr["result"] == "error"
def test_missing_policy_defaults_unknown(self):
entry = {"severity": "low", "result": "pass", "resource": "r"}
pcr = _to_pcr(entry, "c-1")
assert pcr["ruleId"] == "KYVERNO_UNKNOWN"
def test_pcr_validates_against_schema(self, policy_check_result_schema):
entry = {"policy": "p1", "severity": "high", "result": "fail",
"message": "m", "resource": "ns/Pod/x", "namespace": "ns",
"kind": "Pod", "name": "x"}
pcr = _to_pcr(entry, "11111111-1111-1111-1111-111111111111")
jsonschema.validate(pcr, policy_check_result_schema)
class TestAdapt:
def test_translates_fixture(self, tmp_path, policy_check_result_schema):
src = FIXTURES / "kyverno_policyreport.json"
f = tmp_path / "policyreport.json"
f.write_text(src.read_text())
results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
assert len(results) == 3
# result 1: pass/high
assert results[0]["ruleId"] == "disallow-privileged-containers"
assert results[0]["severity"] == "high"
assert results[0]["result"] == "pass"
# result 2: fail/medium
assert results[1]["ruleId"] == "require-resource-labels"
assert results[1]["severity"] == "medium"
assert results[1]["result"] == "fail"
# result 3: warn/high -> skipped/high
assert results[2]["ruleId"] == "require-image-digests"
assert results[2]["severity"] == "high"
assert results[2]["result"] == "skipped"
for pcr in results:
jsonschema.validate(pcr, policy_check_result_schema)
def test_empty_results(self, tmp_path):
f = tmp_path / "empty.json"
f.write_text(json.dumps({"results": []}))
results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
assert results == []
def test_missing_results_key(self, tmp_path):
f = tmp_path / "noresults.json"
f.write_text(json.dumps({"apiVersion": "x", "kind": "PolicyReport"}))
results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
assert results == []
def test_non_list_results_treated_as_empty(self, tmp_path):
f = tmp_path / "bad.json"
f.write_text(json.dumps({"results": "not-a-list"}))
results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
assert results == []
def test_missing_fields_in_entry(self, tmp_path, policy_check_result_schema):
f = tmp_path / "sparse.json"
f.write_text(json.dumps({"results": [{}]}))
results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
assert len(results) == 1
pcr = results[0]
assert pcr["ruleId"] == "KYVERNO_UNKNOWN"
assert pcr["severity"] == "info"
# entry.get("result", "skip") -> default "skip" -> "skipped"
assert pcr["result"] == "skipped"
jsonschema.validate(pcr, policy_check_result_schema)
def test_unknown_result_string_defaults_error(self, tmp_path, policy_check_result_schema):
# An explicit but unmapped result string falls back to "error".
f = tmp_path / "unknownresult.json"
f.write_text(json.dumps({"results": [
{"policy": "p1", "severity": "low", "result": "BOGUS",
"resource": "r"},
]}))
results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
assert results[0]["result"] == "error"
jsonschema.validate(results[0], policy_check_result_schema)
+141
View File
@@ -0,0 +1,141 @@
import json
import os
import sys
from pathlib import Path
import jsonschema
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from adapters.wiz.wiz_adapter import (
SEVERITY_MAP, RESULT_MAP, _to_pcr, _emit_not_configured, adapt, is_configured,
)
FIXTURES = Path(__file__).resolve().parent.parent / "tests" / "fixtures"
class TestSeverityResultMaps:
def test_severity_map_critical(self):
assert SEVERITY_MAP["CRITICAL"] == "critical"
assert SEVERITY_MAP["HIGH"] == "high"
assert SEVERITY_MAP["MEDIUM"] == "medium"
assert SEVERITY_MAP["LOW"] == "low"
assert SEVERITY_MAP["INFO"] == "info"
def test_result_map_open_is_fail(self):
assert RESULT_MAP["OPEN"] == "fail"
assert RESULT_MAP["RESOLVED"] == "pass"
assert RESULT_MAP["IN_PROGRESS"] == "skipped"
assert RESULT_MAP["DISMISSED"] == "skipped"
class TestToPcr:
def test_translates_open_critical(self):
issue = {
"id": "wiz-1",
"title": "a critical issue",
"severity": "CRITICAL",
"status": "OPEN",
"entity": {"id": "arn:aws:s3:::b", "name": "b"},
}
pcr = _to_pcr(issue, "c-1")
assert pcr["engine"] == "wiz"
assert pcr["ruleId"] == "wiz-1"
assert pcr["severity"] == "critical"
assert pcr["result"] == "fail"
assert pcr["contractId"] == "c-1"
assert pcr["resourceRef"] == "arn:aws:s3:::b"
def test_severity_case_insensitive(self):
issue = {"id": "wiz-1", "severity": "high", "status": "open",
"entity": {"id": "r"}}
pcr = _to_pcr(issue, "c-1")
assert pcr["severity"] == "high"
assert pcr["result"] == "fail"
def test_unknown_severity_defaults_info(self):
issue = {"id": "wiz-1", "severity": "BOGUS", "status": "OPEN",
"entity": {"id": "r"}}
pcr = _to_pcr(issue, "c-1")
assert pcr["severity"] == "info"
def test_unknown_status_defaults_error(self):
issue = {"id": "wiz-1", "severity": "INFO", "status": "BOGUS",
"entity": {"id": "r"}}
pcr = _to_pcr(issue, "c-1")
assert pcr["result"] == "error"
def test_pcr_validates_against_schema(self, policy_check_result_schema):
issue = {"id": "wiz-1", "title": "t", "severity": "CRITICAL",
"status": "OPEN", "entity": {"id": "r", "name": "n"}}
pcr = _to_pcr(issue, "11111111-1111-1111-1111-111111111111")
jsonschema.validate(pcr, policy_check_result_schema)
class TestEmitNotConfigured:
def test_not_configured_pcr(self, policy_check_result_schema):
pcr = _emit_not_configured("11111111-1111-1111-1111-111111111111")
assert pcr["ruleId"] == "WIZ_NOT_CONFIGURED"
assert pcr["result"] == "skipped"
assert pcr["engine"] == "wiz"
jsonschema.validate(pcr, policy_check_result_schema)
class TestAdapt:
def test_translates_fixture(self, tmp_path, policy_check_result_schema):
# adapt() reads a file path, copy fixture to a writable temp path
src = FIXTURES / "wiz_issues.json"
f = tmp_path / "wiz_issues.json"
f.write_text(src.read_text())
results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
assert len(results) == 3
# issue 1: OPEN critical -> fail/critical
assert results[0]["ruleId"] == "wiz-issue-001"
assert results[0]["severity"] == "critical"
assert results[0]["result"] == "fail"
# issue 2: RESOLVED high -> pass/high
assert results[1]["ruleId"] == "wiz-issue-002"
assert results[1]["severity"] == "high"
assert results[1]["result"] == "pass"
# issue 3: IN_PROGRESS medium -> skipped/medium
assert results[2]["ruleId"] == "wiz-issue-003"
assert results[2]["severity"] == "medium"
assert results[2]["result"] == "skipped"
for pcr in results:
jsonschema.validate(pcr, policy_check_result_schema)
def test_empty_issues_emits_not_configured(self, tmp_path):
f = tmp_path / "wiz_empty.json"
f.write_text(json.dumps({"issues": []}))
results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
assert len(results) == 1
assert results[0]["ruleId"] == "WIZ_NOT_CONFIGURED"
assert results[0]["result"] == "skipped"
def test_top_level_list_input(self, tmp_path):
# data is a bare list (no "issues" wrapper)
f = tmp_path / "wiz_list.json"
f.write_text(json.dumps([
{"id": "w-1", "severity": "LOW", "status": "RESOLVED",
"entity": {"id": "r"}},
]))
results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
assert len(results) == 1
assert results[0]["severity"] == "low"
assert results[0]["result"] == "pass"
class TestIsConfigured:
def test_not_configured_when_env_unset(self, monkeypatch):
monkeypatch.delenv("WIZ_API_TOKEN", raising=False)
assert is_configured() is False
def test_configured_when_env_set(self, monkeypatch):
monkeypatch.setenv("WIZ_API_TOKEN", "token-abc")
assert is_configured() is True