Jon Chery
e74a8c2f5d
---ci--- project: acdl phase: 42 milestone: v1.9 status: execute ---/ci--- Phase 42 — stub-implementation (REQ-107..111, D-084): route_halt_artifact (REQ-107): - core/separation_of_duties.py: real SNS publish (ACDL_SOD_HALT_TOPIC_ARN) + outbox fallback (SEPARATION_OF_DUTIES_VIOLATION event via outbox_writer) + stderr emission. No silent print-only stub. - terraform/platform/main.tf: aws_sns_topic.acdl-sod-halt + output. HITL attestation gates (REQ-108): - core/hitl_gates.py: attest(contract_id, env, approver, evidence, outbox_client) records approver_qa/approver_prod/approver_dr to outbox, runs SoD check on prod, invokes attestation matrix, returns (ok, reason). Dev skips (autonomous). approver_from_env() reads GITHUB_ACTOR/GITEA_ACTOR. - scripts/run_platform.sh: Step 7b HITL gate before apply for qa/prod/dr. 8-concern attestation matrix (REQ-109, D-084): - core/attestation_matrix.py: check(env, evidence) runs the 8 concerns from hitl_matrix_design.md §10.4. Offline-testable (contract_nfrs, schema_validity, policy_pass) run for real. Operator-supplied accept signed artifacts validated for freshness (FRESHNESS_DAYS table) + schema. Signature skip when ACDL_ATTESTATION_SIGNING_KEY_ID unset (D-089). Fail loud if missing/expired for prod/dr. Wiz real client (REQ-110): - adapters/wiz/wiz_adapter.py: WizClient (GraphQL API, Bearer auth, pagination via pageInfo.hasNextPage + endCursor). fetch_and_adapt translates issues → PolicyCheckResult; graceful degrade when WIZ_API_TOKEN/WIZ_API_URL unset. Kyverno fleshed out (REQ-111): - adapters/kyverno/kyverno_adapter.py: full PolicyReport → PolicyCheckResult mapping (pass/fail/skip/warn + severity + skip-with- reason + resource ref construction from kind/name/namespace). adapt_inactive() emits KYVERNO_INACTIVE_TF_STACK guard. --kube-version stub parsed for future GitOps. Tests: +47 (test_route_halt_artifact.py, test_hitl_gates.py, test_attestation_matrix.py, test_wiz_adapter_real_client.py, expanded test_kyverno_adapter.py). Existing wiz_adapter tests updated for the real client's control.name ruleId. 493 passed; run_ci.sh green; run_platform.sh --check-only green.
222 lines
9.0 KiB
Python
222 lines
9.0 KiB
Python
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")
|
|
# Empty results emit the inactive-for-TF guard record (REQ-111).
|
|
assert len(results) == 1
|
|
assert results[0]["ruleId"] == "KYVERNO_INACTIVE_TF_STACK"
|
|
|
|
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 len(results) == 1
|
|
assert results[0]["ruleId"] == "KYVERNO_INACTIVE_TF_STACK"
|
|
|
|
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 len(results) == 1
|
|
assert results[0]["ruleId"] == "KYVERNO_INACTIVE_TF_STACK"
|
|
|
|
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)
|
|
|
|
# --- v1.9 REQ-111: fleshed-out translator tests ---
|
|
|
|
class TestFleshedOutTranslator:
|
|
def test_pass_result_emits_pass(self, tmp_path):
|
|
f = tmp_path / "pass.json"
|
|
f.write_text(json.dumps({"results": [
|
|
{"policy": "require-labels", "rule": "check-app-label", "severity": "medium",
|
|
"result": "pass", "resource": "pod/x", "message": "label present"},
|
|
]}))
|
|
results = adapt(str(f), "c1")
|
|
assert results[0]["result"] == "pass"
|
|
assert results[0]["ruleId"] == "require-labels/check-app-label"
|
|
|
|
def test_fail_result_with_severity(self, tmp_path):
|
|
f = tmp_path / "fail.json"
|
|
f.write_text(json.dumps({"results": [
|
|
{"policy": "disallow-privileged", "rule": "no-priv", "severity": "critical",
|
|
"result": "fail", "resource": "pod/y", "message": "privileged container"},
|
|
]}))
|
|
results = adapt(str(f), "c2")
|
|
assert results[0]["result"] == "fail"
|
|
assert results[0]["severity"] == "critical"
|
|
assert results[0]["ruleId"] == "disallow-privileged/no-priv"
|
|
|
|
def test_skip_with_reason(self, tmp_path):
|
|
f = tmp_path / "skip.json"
|
|
f.write_text(json.dumps({"results": [
|
|
{"policy": "require-image-digests", "rule": "digest", "severity": "low",
|
|
"result": "skip", "resource": "pod/z", "skipReason": "no image"},
|
|
]}))
|
|
results = adapt(str(f), "c3")
|
|
assert results[0]["result"] == "skipped"
|
|
assert "no image" in results[0]["message"]
|
|
|
|
def test_warn_result_maps_to_skipped(self, tmp_path):
|
|
f = tmp_path / "warn.json"
|
|
f.write_text(json.dumps({"results": [
|
|
{"policy": "p", "rule": "r", "severity": "info", "result": "warn", "resource": "x"},
|
|
]}))
|
|
results = adapt(str(f), "c4")
|
|
assert results[0]["result"] == "skipped"
|
|
|
|
def test_informational_severity_maps_to_info(self, tmp_path):
|
|
f = tmp_path / "info.json"
|
|
f.write_text(json.dumps({"results": [
|
|
{"policy": "p", "rule": "r", "severity": "informational", "result": "pass", "resource": "x"},
|
|
]}))
|
|
results = adapt(str(f), "c5")
|
|
assert results[0]["severity"] == "info"
|
|
|
|
def test_resource_ref_constructed_from_kind_name(self, tmp_path):
|
|
f = tmp_path / "res.json"
|
|
f.write_text(json.dumps({"results": [
|
|
{"policy": "p", "rule": "r", "severity": "low", "result": "fail",
|
|
"kind": "Pod", "namespace": "default", "name": "my-pod"},
|
|
]}))
|
|
results = adapt(str(f), "c6")
|
|
assert "my-pod" in results[0]["resourceRef"]
|
|
|
|
def test_inactive_guard_directly(self):
|
|
from adapters.kyverno.kyverno_adapter import adapt_inactive
|
|
pcrs = adapt_inactive("c7")
|
|
assert len(pcrs) == 1
|
|
assert pcrs[0]["ruleId"] == "KYVERNO_INACTIVE_TF_STACK"
|
|
assert pcrs[0]["result"] == "skipped"
|
|
assert "Terraform" in pcrs[0]["message"]
|
|
|
|
def test_kube_version_parsed(self, tmp_path):
|
|
"""--kube-version is parsed but not yet used (future GitOps)."""
|
|
f = tmp_path / "k.json"
|
|
f.write_text(json.dumps({"results": [
|
|
{"policy": "p", "rule": "r", "severity": "low", "result": "pass", "resource": "x"},
|
|
]}))
|
|
results = adapt(str(f), "c8", kube_version="1.28")
|
|
assert len(results) == 1
|