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---
106 lines
3.2 KiB
Python
106 lines
3.2 KiB
Python
"""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 <wiz_issues.json> <contract-id>
|
|
"""
|
|
|
|
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 <wiz_issues.json> <contract-id>", file=sys.stderr)
|
|
sys.exit(2)
|
|
print(json.dumps(adapt(sys.argv[1], sys.argv[2]), indent=2)) |