acdl/adapters/terraform/policy/checkov_adapter.py

"""Translate Checkov JSON output to Nova 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 Nova tagging standard (D-054, D-043 closure, D-109 hard mode in P3)
is enforced by a custom Checkov rule at
adapters/terraform/policy/custom_rules/nova_tagging.py, loaded via
--external-checks-dir. The adapter therefore maps NOVA_TAG_NAMING as a
real rule (no synthetic SKIPPED record is emitted). Renamed from
ACDL_TAG_NAMING in P2 (REQ-158); the rule is in hard mode as of P3
(REQ-162: hard-fail on missing nova:* or acdl:*-only tags).
"""

import datetime
import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))))
from core.metrics.event_envelope import emit


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, D-109 hard mode (P3): NOVA_TAG_NAMING is a real
    # custom Checkov rule (adapters/terraform/policy/custom_rules/nova_tagging.py),
    # loaded via --external-checks-dir. No synthetic SKIPPED record is emitted.
    # Renamed from ACDL_TAG_NAMING in P2 (REQ-158). Hard mode as of P3
    # (REQ-162: hard-fail on missing nova:* or acdl:*-only tags).
    "NOVA_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, run_id=None, environment="dev"):
    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"))

    # Emit nova.policy.evaluated event (REQ-187).
    if run_id:
        passed = sum(1 for p in out if p["result"] == "pass")
        failed = sum(1 for p in out if p["result"] == "fail")
        skipped = sum(1 for p in out if p["result"] == "skipped")
        severity_breakdown = {}
        for p in out:
            sev = p.get("severity", "info")
            severity_breakdown[sev] = severity_breakdown.get(sev, 0) + 1
        try:
            emit("nova.policy.evaluated", run_id, environment, {
                "passed": passed, "failed": failed, "skipped": skipped,
                "severity_breakdown": severity_breakdown,
                "rule_count": len(out),
            }, contract_id=contract_id)
        except Exception:
            pass  # metrics emission must never break the policy adapter

    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))