From 1fd37a2843a4d447192b07e8c4e87f9448e8d4d5 Mon Sep 17 00:00:00 2001 From: Jon Chery Date: Wed, 22 Jul 2026 20:00:46 +0000 Subject: [PATCH] feat(P23): tagging standard + Wiz adapter + Kyverno adapter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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--- --- adapters/kyverno/README.md | 68 +++++++++ adapters/kyverno/__init__.py | 0 adapters/kyverno/kyverno_adapter.py | 81 ++++++++++ .../disallow-privileged-containers.yaml | 27 ++++ .../policies/require-image-digests.yaml | 26 ++++ .../policies/require-resource-labels.yaml | 37 +++++ adapters/terraform/policy/checkov_adapter.py | 25 +--- .../terraform/policy/custom_rules/README.md | 34 +++++ .../terraform/policy/custom_rules/__init__.py | 0 .../policy/custom_rules/acdl_tagging.py | 55 +++++++ adapters/wiz/README.md | 55 +++++++ adapters/wiz/__init__.py | 0 adapters/wiz/wiz_adapter.py | 106 +++++++++++++ schemas/policy_check_result.schema.json | 2 +- schemas/tagging-standard.json | 46 ++++++ scripts/run_ci.sh | 3 + scripts/run_platform.sh | 4 +- tests/fixtures/kyverno_policyreport.json | 40 +++++ tests/fixtures/wiz_issues.json | 52 +++++++ tests/test_checkov_adapter.py | 23 +-- tests/test_kyverno_adapter.py | 141 ++++++++++++++++++ tests/test_wiz_adapter.py | 141 ++++++++++++++++++ 22 files changed, 935 insertions(+), 31 deletions(-) create mode 100644 adapters/kyverno/README.md create mode 100644 adapters/kyverno/__init__.py create mode 100644 adapters/kyverno/kyverno_adapter.py create mode 100644 adapters/kyverno/policies/disallow-privileged-containers.yaml create mode 100644 adapters/kyverno/policies/require-image-digests.yaml create mode 100644 adapters/kyverno/policies/require-resource-labels.yaml create mode 100644 adapters/terraform/policy/custom_rules/README.md create mode 100644 adapters/terraform/policy/custom_rules/__init__.py create mode 100644 adapters/terraform/policy/custom_rules/acdl_tagging.py create mode 100644 adapters/wiz/README.md create mode 100644 adapters/wiz/__init__.py create mode 100644 adapters/wiz/wiz_adapter.py create mode 100644 schemas/tagging-standard.json create mode 100644 tests/fixtures/kyverno_policyreport.json create mode 100644 tests/fixtures/wiz_issues.json create mode 100644 tests/test_kyverno_adapter.py create mode 100644 tests/test_wiz_adapter.py diff --git a/adapters/kyverno/README.md b/adapters/kyverno/README.md new file mode 100644 index 0000000..fee1802 --- /dev/null +++ b/adapters/kyverno/README.md @@ -0,0 +1,68 @@ +# Kyverno Adapter + +The Kyverno adapter translates Kyverno `PolicyReport` results to the +normalized ACDL +[`PolicyCheckResult`](../../schemas/policy_check_result.schema.json) schema +(engine: `"kyverno"`), mirroring the Checkov/Wiz adapter pattern. + +## What Kyverno is + +[Kyverno](https://kyverno.io/) is a Kubernetes-native policy engine. It +runs as an admission controller inside a cluster, validates / mutates / +generates K8s resources against declarative `ClusterPolicy` rules, and +publishes results to `PolicyReport` resources. + +## When to use it + +Kyverno is the right engine **when the platform emits Kubernetes +manifests** (a K8s-native stack). The ACDL platform today emits Terraform +only (D-053), so this adapter is **ready but inactive**: it ships now so +the schema path, severity/result mapping and sample policies are in place +ahead of the GitOps reconciler that will emit K8s manifests (roadmap). + +## How the adapter translates PolicyReport results + +`kyverno_adapter.py ` reads a JSON file +containing a Kyverno `PolicyReport` (or just its `.results[]` array) and +emits a list of `PolicyCheckResult` dicts: + +| Kyverno PolicyReport result field | PolicyCheckResult field | +|-----------------------------------|-------------------------| +| `policy` | `ruleId` (default `KYVERNO_UNKNOWN`) | +| `severity` | `severity` (lower-cased, mapped) | +| `result` | `result` (`pass`/`fail`/`error` as-is, `warn`/`skip`→`skipped`) | +| `message` | `message` | +| `resource` | `resourceRef` + `evidence.resource` | +| `namespace`, `kind`, `name` | `evidence.*` | + +The adapter is read-only against a local JSON fixture; the GitOps +reconciler is responsible for fetching the live `PolicyReport` and writing +the file. When there are zero results, the adapter returns an empty list +(unlike Wiz it does not synthesize a SKIPPED record — Kyverno not running +is a deployment state, not a configuration gap). + +## Roadmap dependency + +This adapter activates when the GitOps reconciler (roadmap) emits K8s +manifests. Until then it is documentation-only; the pipeline does not +invoke it. The `engine: "kyverno"` enum value is present in +`schemas/policy_check_result.schema.json` so future records validate. + +## Sample policies + +The `policies/` directory holds three valid Kyverno `ClusterPolicy` +manifests (documentation-only today — the platform does not run them): + +- `disallow-privileged-containers.yaml` — fail pods with + `securityContext.privileged: true`. +- `require-resource-labels.yaml` — require `acdl:owner` and + `acdl:environment` labels on all pods (mirrors the ACDL tagging standard + in [`schemas/tagging-standard.json`](../../schemas/tagging-standard.json)). +- `require-image-digests.yaml` — require container images to reference a + digest (`image@sha256:...`), not a mutable tag. + +## Schema path + +The output records validate against +[`schemas/policy_check_result.schema.json`](../../schemas/policy_check_result.schema.json) +(`engine: "kyverno"` was already in the enum and is retained in Phase 23). \ No newline at end of file diff --git a/adapters/kyverno/__init__.py b/adapters/kyverno/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/adapters/kyverno/kyverno_adapter.py b/adapters/kyverno/kyverno_adapter.py new file mode 100644 index 0000000..1c17b48 --- /dev/null +++ b/adapters/kyverno/kyverno_adapter.py @@ -0,0 +1,81 @@ +"""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"). + +D-053: the platform emits Terraform, not K8s manifests. This adapter is +ready but inactive for Terraform-only stacks. It activates when the GitOps +reconciler (roadmap) emits K8s manifests. Sample policies are included as +documentation at adapters/kyverno/policies/. + +CLI: kyverno_adapter.py +""" + +import datetime +import json +import sys + + +SEVERITY_MAP = { + "critical": "critical", + "high": "high", + "medium": "medium", + "low": "low", + "info": "info", +} + +RESULT_MAP = { + "pass": "pass", + "fail": "fail", + "warn": "skipped", + "error": "error", + "skip": "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") + return { + "contractId": contract_id, + "evaluatedAt": _iso8601_now(), + "engine": "kyverno", + "ruleId": entry.get("policy", "KYVERNO_UNKNOWN"), + "severity": severity, + "result": result, + "message": entry.get("message", ""), + "evidence": { + "resource": entry.get("resource", ""), + "namespace": entry.get("namespace", ""), + "kind": entry.get("kind", ""), + "name": entry.get("name", ""), + }, + "resourceRef": entry.get("resource", ""), + } + + +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)) + return out + + +if __name__ == "__main__": + if len(sys.argv) != 3: + print("usage: kyverno_adapter.py ", file=sys.stderr) + sys.exit(2) + print(json.dumps(adapt(sys.argv[1], sys.argv[2]), indent=2)) \ No newline at end of file diff --git a/adapters/kyverno/policies/disallow-privileged-containers.yaml b/adapters/kyverno/policies/disallow-privileged-containers.yaml new file mode 100644 index 0000000..a1a754c --- /dev/null +++ b/adapters/kyverno/policies/disallow-privileged-containers.yaml @@ -0,0 +1,27 @@ +apiVersion: kyverno.io/v1 +kind: ClusterPolicy +metadata: + name: disallow-privileged-containers + annotations: + policies.kyverno.io/title: Disallow Privileged Containers + policies.kyverno.io/category: Security + policies.kyverno.io/severity: high + policies.kyverno.io/subject: Pod +spec: + validationFailureAction: audit + background: true + rules: + - name: require-non-privileged + match: + any: + - resources: + kinds: + - Pod + validate: + message: "Privileged containers are not allowed. Set securityContext.privileged to false." + pattern: + spec: + containers: + - name: "*" + securityContext: + privileged: "false" \ No newline at end of file diff --git a/adapters/kyverno/policies/require-image-digests.yaml b/adapters/kyverno/policies/require-image-digests.yaml new file mode 100644 index 0000000..268c68f --- /dev/null +++ b/adapters/kyverno/policies/require-image-digests.yaml @@ -0,0 +1,26 @@ +apiVersion: kyverno.io/v1 +kind: ClusterPolicy +metadata: + name: require-image-digests + annotations: + policies.kyverno.io/title: Require Image Digests + policies.kyverno.io/category: Supply Chain + policies.kyverno.io/severity: high + policies.kyverno.io/subject: Pod +spec: + validationFailureAction: audit + background: true + rules: + - name: require-digest-reference + match: + any: + - resources: + kinds: + - Pod + validate: + message: "Container images must reference a digest (e.g. image@sha256:...), not a mutable tag." + pattern: + spec: + containers: + - name: "*" + image: "*@sha256:*" \ No newline at end of file diff --git a/adapters/kyverno/policies/require-resource-labels.yaml b/adapters/kyverno/policies/require-resource-labels.yaml new file mode 100644 index 0000000..a58d559 --- /dev/null +++ b/adapters/kyverno/policies/require-resource-labels.yaml @@ -0,0 +1,37 @@ +apiVersion: kyverno.io/v1 +kind: ClusterPolicy +metadata: + name: require-resource-labels + annotations: + policies.kyverno.io/title: Require ACDL Resource Labels + policies.kyverno.io/category: Governance + policies.kyverno.io/severity: medium + policies.kyverno.io/subject: Pod +spec: + validationFailureAction: audit + background: true + rules: + - name: require-acdl-owner-label + match: + any: + - resources: + kinds: + - Pod + validate: + message: "Pods must carry the acdl:owner label (ACDL tagging standard)." + pattern: + metadata: + labels: + acdl:owner: "?*" + - name: require-acdl-environment-label + match: + any: + - resources: + kinds: + - Pod + validate: + message: "Pods must carry the acdl:environment label (ACDL tagging standard)." + pattern: + metadata: + labels: + acdl:environment: "?*" \ No newline at end of file diff --git a/adapters/terraform/policy/checkov_adapter.py b/adapters/terraform/policy/checkov_adapter.py index 529bee5..1d05bce 100644 --- a/adapters/terraform/policy/checkov_adapter.py +++ b/adapters/terraform/policy/checkov_adapter.py @@ -6,8 +6,10 @@ 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. +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 @@ -27,6 +29,10 @@ RULE_MAP = { "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"} @@ -60,20 +66,6 @@ def _to_pcr(checkov_record, contract_id, result_str): } -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) @@ -88,7 +80,6 @@ def adapt(checkov_json_path, contract_id): 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 diff --git a/adapters/terraform/policy/custom_rules/README.md b/adapters/terraform/policy/custom_rules/README.md new file mode 100644 index 0000000..e4f33b1 --- /dev/null +++ b/adapters/terraform/policy/custom_rules/README.md @@ -0,0 +1,34 @@ +# ACDL Custom Checkov Rules + +This directory holds ACDL-authored Checkov custom rules, written in the +[Checkov Python custom-rule framework](https://www.checkov.io/4.Contributing/Custom%20Policies.html). + +## Files + +- `acdl_tagging.py` — `ACDL_TAG_NAMING` (D-054): ensures every taggable AWS + resource carries the four required ACDL tags + (`acdl:owner`, `acdl:contract`, `acdl:environment`, `acdl:cost-center`). + This rule replaces the synthetic SKIPPED `ACDL_TAG_NAMING` record that the + Checkov adapter previously emitted (D-043 closure). The canonical tag set + is declared in [`schemas/tagging-standard.json`](../../../schemas/tagging-standard.json). + +## How Checkov loads them + +Checkov custom rules are discovered via the `--external-checks-dir` flag. +`scripts/run_platform.sh` invokes Checkov with: + +``` +checkov -f terraform/spike/main.tf --framework terraform -o json --soft-fail \ + --external-checks-dir adapters/terraform/policy/custom_rules/ +``` + +Checkov imports each `*.py` file in the directory and instantiates the +module-level `check` object (see the `check = AcdlTaggingStandard()` line at +the bottom of `acdl_tagging.py`). + +## Severity / result mapping + +The Checkov adapter (`adapters/terraform/policy/checkov_adapter.py`) +maps `ACDL_TAG_NAMING` to `(tagging-standard, medium)` in `RULE_MAP`. The +custom rule therefore produces real `PASS`/`FAIL` PolicyCheckResult records, +feeding the confidence signal instead of the old SKIPPED placeholder. \ No newline at end of file diff --git a/adapters/terraform/policy/custom_rules/__init__.py b/adapters/terraform/policy/custom_rules/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/adapters/terraform/policy/custom_rules/acdl_tagging.py b/adapters/terraform/policy/custom_rules/acdl_tagging.py new file mode 100644 index 0000000..3adc40a --- /dev/null +++ b/adapters/terraform/policy/custom_rules/acdl_tagging.py @@ -0,0 +1,55 @@ +"""ACDL tagging standard custom Checkov rule (D-054). + +Checks that all taggable AWS resources have the required ACDL tags: + acdl:owner, acdl:contract, acdl:environment, acdl:cost-center + +Fails (severity medium) when any required tag is missing. +Closes the D-043 deferral (the SKIPPED ACDL_TAG_NAMING placeholder +becomes a real check). +""" + +from __future__ import annotations + +from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck +from checkov.common.models.enums import CheckResult, CheckCategories +from checkov.common.models.consts import graph_resource_name_utils + +REQUIRED_TAGS = ("acdl:owner", "acdl:contract", "acdl:environment", "acdl:cost-center") + +# Resources that support tags (exclude resources that have no tags attribute) +NON_TAGGABLE_TYPES = ( + "aws_cloudfront_origin_access_control", + "aws_lambda_function_url", + "aws_route_table_association", + "aws_internet_gateway", +) + +class AcdlTaggingStandard(BaseResourceCheck): + def __init__(self): + name = "Ensure all taggable AWS resources have required ACDL tags" + check_id = "ACDL_TAG_NAMING" + supported_resources = ["*"] # all resources + categories = [CheckCategories.GENERAL_SECURITY] + super().__init__(name=name, check_id=check_id, categories=categories, supported_resources=supported_resources) + + def scan_resource_conf(self, conf, entity_type): + # Skip non-taggable resources + if entity_type in NON_TAGGABLE_TYPES: + return CheckResult.PASSED + # Check for a tags block + tags = conf.get("tags") + if not tags: + return CheckResult.FAILED + tag_keys = set() + if isinstance(tags, list) and tags: + tag_block = tags[0] + if isinstance(tag_block, dict): + tag_keys = set(tag_block.keys()) + elif isinstance(tags, dict): + tag_keys = set(tags.keys()) + missing = [t for t in REQUIRED_TAGS if t not in tag_keys] + if missing: + return CheckResult.FAILED + return CheckResult.PASSED + +check = AcdlTaggingStandard() \ No newline at end of file diff --git a/adapters/wiz/README.md b/adapters/wiz/README.md new file mode 100644 index 0000000..fa9082b --- /dev/null +++ b/adapters/wiz/README.md @@ -0,0 +1,55 @@ +# Wiz Adapter + +The Wiz adapter translates Wiz API issue records to the normalized ACDL +[`PolicyCheckResult`](../../schemas/policy_check_result.schema.json) schema +(engine: `"wiz"`), mirroring the Checkov adapter pattern. + +## What Wiz is + +[Wiz](https://www.wiz.io/) is a cloud security SaaS platform that +continuously scans CSPM / CWPP / KSPM findings across AWS, Azure, GCP and +Kubernetes. It exposes a GraphQL/REST API for fetching issue records. + +## Adapter behaviour + +`wiz_adapter.py ` reads a JSON file of Wiz +issue records (the shape returned by the Wiz `issues` GraphQL query / +list endpoint) and emits a list of `PolicyCheckResult` dicts: + +| Wiz field | PolicyCheckResult field | +|------------------|------------------------------------------------------------| +| `id` / `control.id` | `ruleId` | +| `severity` | `severity` (mapped `CRITICAL/HIGH/MEDIUM/LOW/INFO`) | +| `status` | `result` (`OPEN→fail`, `RESOLVED→pass`, `IN_PROGRESS/DISMISSED→skipped`) | +| `title` / `control.name` | `message` | +| `entity.id` | `resourceRef` + `evidence.resource` | +| `entity.{name,cloudPlatform,subscriptionId}` | `evidence.*` | + +The adapter is read-only against a local JSON fixture; the pipeline is +responsible for fetching from Wiz (when configured) and writing the file. + +## Offline / degraded behaviour (D-052) + +When Wiz is not configured the pipeline passes an empty issues payload (or +simply does not invoke the adapter). The adapter degrades gracefully: + +- an empty `issues` list → the adapter emits a single `WIZ_NOT_CONFIGURED` + `PolicyCheckResult` with `result: "skipped"` so the confidence policy + input stays non-empty (and does not falsely inflate the score). + +`is_configured()` returns `True` only when the `WIZ_API_TOKEN` +environment variable is set; the pipeline uses it to decide whether to +fetch and invoke the adapter at all. + +## Configuration + +| Env var | Required | Purpose | +|-----------------|----------|--------------------------------------------------| +| `WIZ_API_TOKEN` | yes | Bearer token for the Wiz REST API. When unset, `is_configured()` returns `False`. | +| `WIZ_ENDPOINT` | no | Wiz API endpoint (defaults to `https://api.wiz.io` when implemented). | + +## Schema path + +The output records validate against +[`schemas/policy_check_result.schema.json`](../../schemas/policy_check_result.schema.json) +(`engine: "wiz"` was added to the enum in Phase 23). \ No newline at end of file diff --git a/adapters/wiz/__init__.py b/adapters/wiz/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/adapters/wiz/wiz_adapter.py b/adapters/wiz/wiz_adapter.py new file mode 100644 index 0000000..5c0f327 --- /dev/null +++ b/adapters/wiz/wiz_adapter.py @@ -0,0 +1,106 @@ +"""Wiz adapter — translate Wiz API results to ACDL PolicyCheckResult records. + +Wiz is a SaaS security platform with a REST API (issues, security graph +queries). This adapter translates Wiz issue records to the normalized +PolicyCheckResult schema (engine: "wiz"), matching the Checkov adapter +pattern. + +D-052: stub + schema path. The adapter degrades gracefully when Wiz is +not configured — it emits a single SKIPPED record (WIZ_NOT_CONFIGURED) +so the confidence policy input stays non-empty. The pipeline invokes it +optionally when WIZ_API_TOKEN is set. + +CLI: wiz_adapter.py +""" + +import datetime +import json +import os +import sys + + +SEVERITY_MAP = { + "CRITICAL": "critical", + "HIGH": "high", + "MEDIUM": "medium", + "LOW": "low", + "INFO": "info", +} + +RESULT_MAP = { + "OPEN": "fail", + "RESOLVED": "pass", + "IN_PROGRESS": "skipped", + "DISMISSED": "skipped", +} + + +def _iso8601_now(): + return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _to_pcr(wiz_issue, contract_id): + severity_raw = wiz_issue.get("severity", "INFO") + severity = SEVERITY_MAP.get(str(severity_raw).upper(), "info") + status = wiz_issue.get("status", "OPEN") + result = RESULT_MAP.get(str(status).upper(), "error") + control = wiz_issue.get("control", {}) + return { + "contractId": contract_id, + "evaluatedAt": _iso8601_now(), + "engine": "wiz", + "ruleId": wiz_issue.get("id", control.get("id", "WIZ_UNKNOWN")), + "severity": severity, + "result": result, + "message": wiz_issue.get("title", control.get("name", "")), + "evidence": { + "resource": wiz_issue.get("entity", {}).get("id"), + "resource_name": wiz_issue.get("entity", {}).get("name"), + "cloud_platform": wiz_issue.get("entity", {}).get("cloudPlatform"), + "subscription_id": wiz_issue.get("entity", {}).get("subscriptionId"), + }, + "resourceRef": wiz_issue.get("entity", {}).get("id", ""), + } + + +def _emit_not_configured(contract_id): + return { + "contractId": contract_id, + "evaluatedAt": _iso8601_now(), + "engine": "wiz", + "ruleId": "WIZ_NOT_CONFIGURED", + "severity": "info", + "result": "skipped", + "message": "Wiz adapter not configured (WIZ_API_TOKEN not set); degraded gracefully (D-052).", + "evidence": {}, + "resourceRef": "", + } + + +def adapt(wiz_json_path, contract_id): + with open(wiz_json_path, "r", encoding="utf-8") as fh: + data = json.load(fh) + out = [] + # Accept either a bare list of issues or an object with an "issues" key. + if isinstance(data, list): + issues = data + else: + issues = data.get("issues", []) + if not isinstance(issues, list): + issues = [] + for issue in issues: + out.append(_to_pcr(issue, contract_id)) + if not out: + out.append(_emit_not_configured(contract_id)) + return out + + +def is_configured(): + return bool(os.environ.get("WIZ_API_TOKEN")) + + +if __name__ == "__main__": + if len(sys.argv) != 3: + print("usage: wiz_adapter.py ", file=sys.stderr) + sys.exit(2) + print(json.dumps(adapt(sys.argv[1], sys.argv[2]), indent=2)) \ No newline at end of file diff --git a/schemas/policy_check_result.schema.json b/schemas/policy_check_result.schema.json index f231e28..33269fb 100644 --- a/schemas/policy_check_result.schema.json +++ b/schemas/policy_check_result.schema.json @@ -19,7 +19,7 @@ }, "engine": { "type": "string", - "enum": ["checkov", "kyverno", "opa"], + "enum": ["checkov", "kyverno", "opa", "wiz"], "description": "Policy engine that produced this result." }, "ruleId": { diff --git a/schemas/tagging-standard.json b/schemas/tagging-standard.json new file mode 100644 index 0000000..b8e0dbb --- /dev/null +++ b/schemas/tagging-standard.json @@ -0,0 +1,46 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://acdl.dev/schemas/tagging-standard.json", + "title": "ACDL Tagging Standard", + "description": "Required tags for all taggable AWS resources created by the platform. Enforced by a Checkov custom YAML rule (adapters/terraform/policy/custom_rules/acdl_tagging.yaml). The checkov adapter maps ACDL_TAG_NAMING as a real rule (D-054, D-043 closure).", + "type": "object", + "properties": { + "required_tags": { + "type": "object", + "description": "The set of tags that must be present on every taggable AWS resource.", + "properties": { + "acdl:owner": { + "type": "string", + "description": "The consumer repository name (e.g. 'consumer-repo'). Injected from the ABAC session." + }, + "acdl:contract": { + "type": "string", + "description": "The contract ID (UUID)." + }, + "acdl:environment": { + "type": "string", + "enum": ["dev", "qa", "prod", "dr"], + "description": "The environment name." + }, + "acdl:cost-center": { + "type": "string", + "description": "The cost center (consumer-provided or platform-default 'acdl-default')." + } + }, + "required": ["acdl:owner", "acdl:contract", "acdl:environment", "acdl:cost-center"], + "additionalProperties": false + }, + "default_values": { + "type": "object", + "description": "Default values used when the consumer does not supply the tag.", + "properties": { + "acdl:cost-center": { + "type": "string", + "default": "acdl-default" + } + } + } + }, + "required": ["required_tags"], + "additionalProperties": false +} \ No newline at end of file diff --git a/scripts/run_ci.sh b/scripts/run_ci.sh index 043c010..f4f2624 100755 --- a/scripts/run_ci.sh +++ b/scripts/run_ci.sh @@ -47,6 +47,9 @@ python3 -m py_compile \ core/contract_resolver.py \ adapters/terraform/adapter.py \ adapters/terraform/policy/checkov_adapter.py \ + adapters/terraform/policy/custom_rules/acdl_tagging.py \ + adapters/wiz/wiz_adapter.py \ + adapters/kyverno/kyverno_adapter.py \ scripts/push_consumer_image.py \ || fail "lint: py_compile failed" echo "lint: OK" diff --git a/scripts/run_platform.sh b/scripts/run_platform.sh index 7a69328..ed4b662 100755 --- a/scripts/run_platform.sh +++ b/scripts/run_platform.sh @@ -195,9 +195,9 @@ fi echo "" echo "=== Step 5: run Checkov on terraform/spike/main.tf ===" if [ "$QUIET" = "0" ]; then - checkov -f terraform/spike/main.tf --framework terraform -o json --soft-fail 2>&1 | tee "$WORK/checkov.json" + checkov -f terraform/spike/main.tf --framework terraform -o json --soft-fail --external-checks-dir adapters/terraform/policy/custom_rules/ 2>&1 | tee "$WORK/checkov.json" else - checkov -f terraform/spike/main.tf --framework terraform -o json --soft-fail > "$WORK/checkov.json" 2> "$WORK/checkov.err" + checkov -f terraform/spike/main.tf --framework terraform -o json --soft-fail --external-checks-dir adapters/terraform/policy/custom_rules/ > "$WORK/checkov.json" 2> "$WORK/checkov.err" fi [ -s "$WORK/checkov.json" ] || fail "checkov produced no output" echo "" diff --git a/tests/fixtures/kyverno_policyreport.json b/tests/fixtures/kyverno_policyreport.json new file mode 100644 index 0000000..0d6b413 --- /dev/null +++ b/tests/fixtures/kyverno_policyreport.json @@ -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" + } + ] +} \ No newline at end of file diff --git a/tests/fixtures/wiz_issues.json b/tests/fixtures/wiz_issues.json new file mode 100644 index 0000000..cac470a --- /dev/null +++ b/tests/fixtures/wiz_issues.json @@ -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" + } + } + ] +} \ No newline at end of file diff --git a/tests/test_checkov_adapter.py b/tests/test_checkov_adapter.py index e479d1c..123ee87 100644 --- a/tests/test_checkov_adapter.py +++ b/tests/test_checkov_adapter.py @@ -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 \ No newline at end of file + assert results == [] # no synthetic tag-naming record anymore \ No newline at end of file diff --git a/tests/test_kyverno_adapter.py b/tests/test_kyverno_adapter.py new file mode 100644 index 0000000..89af729 --- /dev/null +++ b/tests/test_kyverno_adapter.py @@ -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) \ No newline at end of file diff --git a/tests/test_wiz_adapter.py b/tests/test_wiz_adapter.py new file mode 100644 index 0000000..5c6662d --- /dev/null +++ b/tests/test_wiz_adapter.py @@ -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 \ No newline at end of file