Jon Chery
1fd37a2843
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---
90 lines
3.5 KiB
Python
90 lines
3.5 KiB
Python
"""Translate Checkov JSON output to ACDL PolicyCheckResult records.
|
|
|
|
Reads Checkov's JSON output (one framework key, e.g. terraform_plan),
|
|
emits a list of PolicyCheckResult dicts conforming to
|
|
schemas/policy_check_result.schema.json. Run Checkov with --soft-fail so
|
|
Checkov never exits non-zero; the confidence signal decides the gate, not
|
|
Checkov's exit code.
|
|
|
|
The ACDL tagging standard (D-054, D-043 closure) is enforced by a custom
|
|
Checkov rule at adapters/terraform/policy/custom_rules/acdl_tagging.py,
|
|
loaded via --external-checks-dir. The adapter therefore maps
|
|
ACDL_TAG_NAMING as a real rule (no synthetic SKIPPED record is emitted).
|
|
"""
|
|
|
|
import datetime
|
|
import json
|
|
import sys
|
|
|
|
|
|
RULE_MAP = {
|
|
"CKV_AWS_41": ("secrets-in-plaintext", "high"),
|
|
"CKV_AWS_45": ("secrets-in-plaintext", "high"),
|
|
"CKV_AWS_46": ("secrets-in-plaintext", "high"),
|
|
"CKV_AWS_20": ("public-ingress", "high"),
|
|
"CKV_AWS_57": ("public-ingress", "high"),
|
|
"CKV_AWS_24": ("public-ingress", "medium"),
|
|
"CKV_AWS_25": ("public-ingress", "medium"),
|
|
"CKV_AWS_1": ("iam-wildcard", "high"),
|
|
"CKV_AWS_40": ("iam-wildcard", "medium"),
|
|
"CKV_AWS_7": ("kms-key-reference", "medium"),
|
|
"CKV_AWS_33": ("kms-key-reference", "medium"),
|
|
# D-054 / D-043 closure: ACDL_TAG_NAMING is now a real custom Checkov
|
|
# rule (adapters/terraform/policy/custom_rules/acdl_tagging.py), loaded
|
|
# via --external-checks-dir. No synthetic SKIPPED record is emitted.
|
|
"ACDL_TAG_NAMING": ("tagging-standard", "medium"),
|
|
}
|
|
|
|
_RESULT_MAP = {"PASSED": "pass", "FAILED": "fail", "SKIPPED": "skipped"}
|
|
|
|
|
|
def _iso8601_now():
|
|
return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
def _to_pcr(checkov_record, contract_id, result_str):
|
|
check_id = checkov_record.get("check_id", "")
|
|
default_sev = RULE_MAP.get(check_id, (check_id, "info"))[1]
|
|
severity = checkov_record.get("severity", default_sev)
|
|
if isinstance(severity, str):
|
|
severity = severity.lower()
|
|
return {
|
|
"contractId": contract_id,
|
|
"evaluatedAt": _iso8601_now(),
|
|
"engine": "checkov",
|
|
"ruleId": check_id,
|
|
"severity": severity,
|
|
"result": _RESULT_MAP.get(result_str, "error"),
|
|
"message": checkov_record.get("check_name", ""),
|
|
"evidence": {
|
|
"file_path": checkov_record.get("file_path"),
|
|
"resource": checkov_record.get("resource"),
|
|
"resource_address": checkov_record.get("resource_address"),
|
|
"code_block": checkov_record.get("code_block"),
|
|
},
|
|
"resourceRef": checkov_record.get("resource_address") or checkov_record.get("resource", ""),
|
|
}
|
|
|
|
|
|
def adapt(checkov_json_path, contract_id):
|
|
with open(checkov_json_path, "r", encoding="utf-8") as fh:
|
|
data = json.load(fh)
|
|
out = []
|
|
for framework, body in data.items():
|
|
results = body.get("results", body) if isinstance(body, dict) else {}
|
|
if not isinstance(results, dict):
|
|
continue
|
|
for rec in results.get("passed_checks", []):
|
|
out.append(_to_pcr(rec, contract_id, "PASSED"))
|
|
for rec in results.get("failed_checks", []):
|
|
out.append(_to_pcr(rec, contract_id, "FAILED"))
|
|
for rec in results.get("skipped_checks", []):
|
|
out.append(_to_pcr(rec, contract_id, "SKIPPED"))
|
|
return out
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 3:
|
|
print("usage: checkov_adapter.py <checkov.json> <contract-id>", file=sys.stderr)
|
|
sys.exit(2)
|
|
print(json.dumps(adapt(sys.argv[1], sys.argv[2]), indent=2)) |