phase: 7, status: plan-as-execute, persona: security-engineer, task: T-7.4..T-7.8

---ci--- project: acdl phase: 7 milestone: v1.1 status: plan-as-execute persona: security-engineer task: [T-7.4, T-7.5, T-7.6, T-7.7, T-7.8] requirements.covered: [REQ-18, REQ-20, REQ-21] ---/ci--- Wave 3 (security-engineer, 5 files sequential): - T-7.4: schemas/policy_check_result.schema.json (REQ-18 schema half) — canonical shape from ARCHITECTURE.md §12.6; engine enum [checkov,kyverno,opa]; severity enum [critical,high,medium,low,info]; result enum [pass,fail,skipped,error]. Validates as Draft 2020-12; valid instance validates. - T-7.5: adapters/terraform/policy/checkov_adapter.py (REQ-18 adapter half) — Checkov JSON -> PolicyCheckResult; RULE_MAP has all 11 Checkov rule IDs (CKV_AWS_41/45/46/20/57/24/25/1/40/7/33) mapped to the 4 L2 checks + tag/naming; emits ACDL_TAG_NAMING SKIPPED per D-043; stdlib only; tolerates both Checkov JSON shapes. Synthetic fixture produces 3 records all valid against the schema. - T-7.6: platform/audit_ledger_design.md (REQ-20) — three tiers (S3 Object Lock compliance 7yr, acdl-evidence hot index, DynamoDB outbox RPO=0); spike scope (D-041) = hash chain + outbox write; v1.2 build-out = Object Lock + JWS (KMS key, quarterly rotation) + async worker + DLQ + daily checkpoints. Outbox item shape, RPO/RTO table, decision trail. - T-7.7: platform/hitl_matrix_design.md (REQ-21 design half) — pre-execution gate model; Gitea-specific mechanics (workflow_dispatch + gitea.actor per D-042, no Environments API); full 8-concern matrix verbatim from §10.4; timeout 1d warn / 2d freeze; rejection -> HELD + supersedes; CODEOWNERS routing; SoD pointer to the .py. - T-7.8: platform/separation_of_duties.py (REQ-21 impl half) — check(outbox_client, contract_id, current_prod_approver) -> (ok, reason); None outbox -> no-op; equal -> SEPARATION_OF_DUTIES_VIOLATION; distinct -> ok; route_halt_artifact stub; stdlib only (duck-typed outbox_client). All 5 SoD cases verified.
2026-07-21 18:45:28 +00:00
parent f8e99ed906
commit 6ed93f0311
7 changed files with 412 additions and 0 deletions
@@ -0,0 +1,99 @@
+"""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.
+
+Spike scope (D-043): tag/naming is a single SKIPPED record. A custom
+Checkov YAML rule for tag presence lands in v1.2.
+"""
+
+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"),
+}
+
+_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 _emit_tag_naming_skipped(contract_id):
+    return {
+        "contractId": contract_id,
+        "evaluatedAt": _iso8601_now(),
+        "engine": "checkov",
+        "ruleId": "ACDL_TAG_NAMING",
+        "severity": "info",
+        "result": "skipped",
+        "message": "tag/naming check deferred to v1.2 (D-043)",
+        "evidence": {},
+        "resourceRef": "",
+    }
+
+
+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"))
+    out.append(_emit_tag_naming_skipped(contract_id))
+    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))