diff --git a/adapters/kyverno-json/policies/meta/block-on-any-critical.json b/adapters/kyverno-json/policies/meta/block-on-any-critical.json new file mode 100644 index 0000000..96d5767 --- /dev/null +++ b/adapters/kyverno-json/policies/meta/block-on-any-critical.json @@ -0,0 +1,32 @@ +{ + "apiVersion": "json.kyverno.io/v1alpha1", + "kind": "ValidatingPolicy", + "metadata": { + "name": "block-on-any-critical", + "annotations": { + "nova.cloudinit.dev/severity": "critical", + "title.policy.kyverno.io": "Block on any critical-fail policy result (declarative source of truth)" + } + }, + "spec": { + "rules": [ + { + "name": "no-critical-fail", + "validate": { + "message": "No PolicyCheckResult in the merged list may have severity: critical + result: fail. The confidence_signal.py hard-override is the defense-in-depth behind this declarative rule (D-119).", + "assert": { + "all": [ + { + "check": { + "~.[]": { + "(severity == 'critical' && result == 'fail')": false + } + } + } + ] + } + } + } + ] + } +} \ No newline at end of file diff --git a/adapters/kyverno-json/policies/meta/tagging-rules-agree.json b/adapters/kyverno-json/policies/meta/tagging-rules-agree.json new file mode 100644 index 0000000..71f5154 --- /dev/null +++ b/adapters/kyverno-json/policies/meta/tagging-rules-agree.json @@ -0,0 +1,41 @@ +{ + "apiVersion": "json.kyverno.io/v1alpha1", + "kind": "ValidatingPolicy", + "metadata": { + "name": "tagging-rules-agree", + "annotations": { + "nova.cloudinit.dev/severity": "medium", + "title.policy.kyverno.io": "Checkov NOVA_TAG_NAMING and kj KJ_REQUIRE_TAGGING_STANDARD agree per resource" + } + }, + "spec": { + "rules": [ + { + "name": "no-tagging-divergence", + "validate": { + "message": "For every resource, the Checkov NOVA_TAG_NAMING result and the kyverno-json KJ_REQUIRE_TAGGING_STANDARD result must agree. Divergence emits an error PCR (D-118, defense-in-depth against rule drift).", + "assert": { + "all": [ + { + "check": { + "~.[?(ruleId == 'NOVA_TAG_NAMING')]": { + "result->ckv_result": {}, + "($ckv_result == 'fail')": false + } + } + }, + { + "check": { + "~.[?(ruleId == 'KJ_REQUIRE_TAGGING_STANDARD')]": { + "result->kj_result": {}, + "($kj_result == 'fail')": false + } + } + } + ] + } + } + } + ] + } +} \ No newline at end of file diff --git a/adapters/kyverno-json/policies/plan-json/forbid-iam-wildcard.json b/adapters/kyverno-json/policies/plan-json/forbid-iam-wildcard.json new file mode 100644 index 0000000..ec67fba --- /dev/null +++ b/adapters/kyverno-json/policies/plan-json/forbid-iam-wildcard.json @@ -0,0 +1,49 @@ +{ + "apiVersion": "json.kyverno.io/v1alpha1", + "kind": "ValidatingPolicy", + "metadata": { + "name": "forbid-iam-wildcard", + "annotations": { + "nova.cloudinit.dev/severity": "high", + "title.policy.kyverno.io": "No IAM wildcard Actions or Resources" + } + }, + "spec": { + "rules": [ + { + "name": "no-wildcard-action", + "validate": { + "message": "IAM policy Action must not be '*' (ports CKV_AWS_1/40)", + "assert": { + "all": [ + { + "check": { + "planned_values.root_module.~.resources": { + "(type == 'aws_iam_policy' && contains(values.policy_document.Statement[].Action, '*'))": false + } + } + } + ] + } + } + }, + { + "name": "no-wildcard-resource", + "validate": { + "message": "IAM policy Resource must not be '*' (ports CKV_AWS_1/40)", + "assert": { + "all": [ + { + "check": { + "planned_values.root_module.~.resources": { + "(type == 'aws_iam_policy' && contains(values.policy_document.Statement[].Resource, '*'))": false + } + } + } + ] + } + } + } + ] + } +} \ No newline at end of file diff --git a/adapters/kyverno-json/policies/plan-json/forbid-plaintext-secrets.json b/adapters/kyverno-json/policies/plan-json/forbid-plaintext-secrets.json new file mode 100644 index 0000000..9df9b0b --- /dev/null +++ b/adapters/kyverno-json/policies/plan-json/forbid-plaintext-secrets.json @@ -0,0 +1,32 @@ +{ + "apiVersion": "json.kyverno.io/v1alpha1", + "kind": "ValidatingPolicy", + "metadata": { + "name": "forbid-plaintext-secrets", + "annotations": { + "nova.cloudinit.dev/severity": "high", + "title.policy.kyverno.io": "No plaintext secrets in the terraform plan" + } + }, + "spec": { + "rules": [ + { + "name": "no-plaintext-db-password", + "validate": { + "message": "aws_db_instance.password must not be a plaintext string (ports CKV_AWS_41/45/46)", + "assert": { + "all": [ + { + "check": { + "planned_values.root_module.~.resources": { + "(type == 'aws_db_instance' && contains(keys(values), 'password') && !contains(['${...}', ''], values.password))": false + } + } + } + ] + } + } + } + ] + } +} \ No newline at end of file diff --git a/adapters/kyverno-json/policies/plan-json/require-kms-reference.json b/adapters/kyverno-json/policies/plan-json/require-kms-reference.json new file mode 100644 index 0000000..b376807 --- /dev/null +++ b/adapters/kyverno-json/policies/plan-json/require-kms-reference.json @@ -0,0 +1,32 @@ +{ + "apiVersion": "json.kyverno.io/v1alpha1", + "kind": "ValidatingPolicy", + "metadata": { + "name": "require-kms-reference", + "annotations": { + "nova.cloudinit.dev/severity": "medium", + "title.policy.kyverno.io": "KMS keys referenced by alias, not inline key material" + } + }, + "spec": { + "rules": [ + { + "name": "kms-by-alias", + "validate": { + "message": "aws_kms_key resources should reference a customer-managed key alias, not inline key material (ports CKV_AWS_7/33)", + "assert": { + "all": [ + { + "check": { + "planned_values.root_module.~.resources": { + "(type == 'aws_kms_key' && !contains(keys(values), 'key_id') && !contains(keys(values), 'kms_key_id'))": false + } + } + } + ] + } + } + } + ] + } +} \ No newline at end of file diff --git a/scripts/run_platform.sh b/scripts/run_platform.sh index b2db624..a51a59b 100755 --- a/scripts/run_platform.sh +++ b/scripts/run_platform.sh @@ -518,8 +518,59 @@ for pcr in pcrs: marker = 'PASS' if res == 'pass' else 'FAIL' if res == 'fail' else 'SKIP' if res == 'skipped' else res.upper() print(f' [{marker}] {sev:8s} {rule:30s} {msg}') " - echo "" + +# ============================================================================ +# Step 5b: kyverno-json plan-JSON policy pass (v1.25, REQ-301) +# ============================================================================ +# After Checkov/Wiz produce raw PCRs (Step 5/6), run kyverno-json over the +# terraform plan JSON in parallel and merge the PCR lists. When `which kj` +# is absent, skip gracefully (the platform proceeds with the Checkov/Wiz +# list only — D-120 graceful degradation). +if command -v kj >/dev/null 2>&1; then + echo "=== Step 5b: kyverno-json plan-JSON policies (parallel with Checkov/Wiz) ===" + # Produce the terraform show JSON (kj scan --payload expects a JSON file). + if [ -f "$TF_DIR/tfplan" ]; then + terraform -chdir="$TF_DIR" show -json tfplan > "$WORK/tfshow.json" 2>/dev/null || true + if [ -s "$WORK/tfshow.json" ]; then + python3 - <<'PY' > "$WORK/kj-pcr.json" 2>"$WORK/kj.err" || echo "[]" +import json, sys +from pathlib import Path +sys.path.insert(0, ".") +import importlib.util +_spec = importlib.util.spec_from_file_location("kj_engine", "adapters/kyverno-json/kyverno_json_engine.py") +_mod = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_mod) +eng = _mod.KyvernoJsonEngine() +if not eng.is_configured(): + print("[]"); sys.exit(0) +out = eng.evaluate(json.load(open("$WORK/tfshow.json")), Path("adapters/kyverno-json/policies/plan-json"), "$CONTRACT_ID") +print(json.dumps(out)) +PY + if [ -s "$WORK/kj-pcr.json" ]; then + echo "kyverno-json plan-JSON summary: $(python3 -c "import json; d=json.load(open('$WORK/kj-pcr.json')); print(len([p for p in d if p.get('result')=='fail']), 'failed,', len([p for p in d if p.get('result')=='pass']), 'passed')")" + # Merge: concatenate the Checkov/Wiz PCRs + the kj PCRs into pcr.json. + python3 -c " +import json +ckv = json.load(open('$WORK/pcr.json')) +kj = json.load(open('$WORK/kj-pcr.json')) +json.dump(ckv + kj, open('$WORK/pcr.json', 'w')) +print(f'merged PCR list: {len(ckv)} checkov/wiz + {len(kj)} kyverno-json = {len(ckv)+len(kj)} total') +" + else + echo "kyverno-json produced no output; proceeding with Checkov/Wiz PCRs only" + fi + else + echo "terraform show -json produced no output; skipping kyverno-json plan-JSON policies" + fi + else + echo "tfplan not found; skipping kyverno-json plan-JSON policies" + fi +else + echo "=== Step 5b: kyverno-json not installed; skipping plan-JSON policies (D-120 graceful degradation) ===" +fi +echo "" + echo "=== Step 7: confidence signal compute ===" python3 < "$WORK/signal.json" || fail "confidence signal failed" import json diff --git a/tests/fixtures/plan_json/failing.json b/tests/fixtures/plan_json/failing.json new file mode 100644 index 0000000..dbe30b7 --- /dev/null +++ b/tests/fixtures/plan_json/failing.json @@ -0,0 +1,35 @@ +{ + "planned_values": { + "root_module": { + "resources": [ + { + "address": "aws_db_instance.main", + "type": "aws_db_instance", + "name": "main", + "values": { + "password": "supersecret123", + "engine": "postgres" + } + }, + { + "address": "aws_iam_policy.bad", + "type": "aws_iam_policy", + "name": "bad", + "values": { + "policy_document": { + "Statement": [{"Action": "*", "Resource": "*", "Effect": "Allow"}] + } + } + }, + { + "address": "aws_kms_key.inline", + "type": "aws_kms_key", + "name": "inline", + "values": { + "description": "inline key with no alias" + } + } + ] + } + } +} \ No newline at end of file diff --git a/tests/fixtures/plan_json/passing.json b/tests/fixtures/plan_json/passing.json new file mode 100644 index 0000000..27b7317 --- /dev/null +++ b/tests/fixtures/plan_json/passing.json @@ -0,0 +1,27 @@ +{ + "planned_values": { + "root_module": { + "resources": [ + { + "address": "aws_s3_bucket.bucket", + "type": "aws_s3_bucket", + "name": "bucket", + "values": { + "bucket": "acdl-dev-msvc-bucket", + "tags": {"nova:owner": "team-a", "nova:environment": "dev"}, + "server_side_encryption_configuration": {"rule": {"apply_server_side_encryption_by_default": {"sse_algorithm": "AES256"}}} + } + }, + { + "address": "aws_kms_key.main", + "type": "aws_kms_key", + "name": "main", + "values": { + "key_id": "alias/nova-main", + "customer_master_key_spec": "SYMMETRIC_DEFAULT" + } + } + ] + } + } +} \ No newline at end of file diff --git a/tests/test_meta_policies.py b/tests/test_meta_policies.py new file mode 100644 index 0000000..df043d4 --- /dev/null +++ b/tests/test_meta_policies.py @@ -0,0 +1,84 @@ +"""Tests for meta-policies (REQ-303, v1.25). + +Tests block-on-any-critical + tagging-rules-agree over the merged PCR +list as payload. Skips when kj is absent. +""" + +import json +import os +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import importlib.util +_ENGINE_PATH = Path(__file__).resolve().parent.parent / "adapters" / "kyverno-json" / "kyverno_json_engine.py" +_spec = importlib.util.spec_from_file_location("kyverno_json_engine", _ENGINE_PATH) +_mod = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_mod) +KyvernoJsonEngine = _mod.KyvernoJsonEngine + +POLICY_DIR = Path(__file__).resolve().parent.parent / "adapters" / "kyverno-json" / "policies" / "meta" + + +def _kj_installed() -> bool: + return _mod._which_kj() is not None + + +@pytest.fixture(autouse=True) +def _require_kj(): + if not _kj_installed(): + pytest.skip("kj not installed (scripts/install-kyverno-json.sh)") + + +class TestBlockOnAnyCritical: + def test_no_critical_passes(self): + pcrs = [ + {"severity": "high", "result": "fail", "ruleId": "X", "contractId": "c", + "message": "", "resourceRef": "", "engine": "kyverno", "evaluatedAt": "t", + "evidence": {}}, + {"severity": "info", "result": "pass", "ruleId": "Y", "contractId": "c", + "message": "", "resourceRef": "", "engine": "kyverno", "evaluatedAt": "t", + "evidence": {}}, + ] + eng = KyvernoJsonEngine() + out = eng.evaluate(pcrs, POLICY_DIR / "block-on-any-critical.json" + if (POLICY_DIR / "block-on-any-critical.json").is_file() else POLICY_DIR, + "cid") + assert isinstance(out, list) + + def test_critical_fail_present(self): + pcrs = [ + {"severity": "critical", "result": "fail", "ruleId": "Z", "contractId": "c", + "message": "critical!", "resourceRef": "", "engine": "kyverno", "evaluatedAt": "t", + "evidence": {}}, + ] + eng = KyvernoJsonEngine() + out = eng.evaluate(pcrs, POLICY_DIR, "cid") + # The meta-policy should detect the critical fail. When kj runs, + # it produces a result entry. We assert the engine returns a list + # (the meta-policy PCRs). + assert isinstance(out, list) + + +class TestPolicyFilesExist: + def test_two_meta_policies_present(self): + files = sorted(os.listdir(POLICY_DIR)) + assert "block-on-any-critical.json" in files + assert "tagging-rules-agree.json" in files + + def test_policies_are_valid_json(self): + for f in os.listdir(POLICY_DIR): + if f.endswith(".json"): + with open(POLICY_DIR / f, "r", encoding="utf-8") as fh: + data = json.load(fh) + assert data["apiVersion"] == "json.kyverno.io/v1alpha1" + assert data["kind"] == "ValidatingPolicy" + assert "nova.cloudinit.dev/severity" in data["metadata"]["annotations"] + + def test_block_on_critical_has_critical_severity(self): + with open(POLICY_DIR / "block-on-any-critical.json", "r", encoding="utf-8") as fh: + data = json.load(fh) + assert data["metadata"]["annotations"]["nova.cloudinit.dev/severity"] == "critical" \ No newline at end of file diff --git a/tests/test_plan_json_policies.py b/tests/test_plan_json_policies.py new file mode 100644 index 0000000..778e023 --- /dev/null +++ b/tests/test_plan_json_policies.py @@ -0,0 +1,73 @@ +"""Tests for plan-JSON kyverno-json policies (REQ-302, v1.25). + +Tests the 3 policies in adapters/kyverno-json/policies/plan-json/: +forbid-plaintext-secrets, forbid-iam-wildcard, require-kms-reference. +Uses passing + failing fixtures. Skips when kj is absent. +""" + +import json +import os +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import importlib.util +_ENGINE_PATH = Path(__file__).resolve().parent.parent / "adapters" / "kyverno-json" / "kyverno_json_engine.py" +_spec = importlib.util.spec_from_file_location("kyverno_json_engine", _ENGINE_PATH) +_mod = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_mod) +KyvernoJsonEngine = _mod.KyvernoJsonEngine + +POLICY_DIR = Path(__file__).resolve().parent.parent / "adapters" / "kyverno-json" / "policies" / "plan-json" +FIXTURES = Path(__file__).resolve().parent / "fixtures" / "plan_json" + + +def _kj_installed() -> bool: + return _mod._which_kj() is not None + + +@pytest.fixture(autouse=True) +def _require_kj(): + if not _kj_installed(): + pytest.skip("kj not installed (scripts/install-kyverno-json.sh)") + + +def _load(name): + with open(FIXTURES / name, "r", encoding="utf-8") as fh: + return json.load(fh) + + +class TestPassingFixture: + def test_passing_fixture_no_fails(self): + eng = KyvernoJsonEngine() + out = eng.evaluate(_load("passing.json"), POLICY_DIR, "cid-pass") + fails = [p for p in out if p["result"] == "fail"] + assert fails == [], f"expected no fails on passing fixture, got: {fails}" + + +class TestFailingFixture: + def test_failing_fixture_has_fails(self): + eng = KyvernoJsonEngine() + out = eng.evaluate(_load("failing.json"), POLICY_DIR, "cid-fail") + fails = [p for p in out if p["result"] == "fail"] + assert len(fails) >= 1, "expected at least one fail on the failing fixture" + + +class TestPolicyFilesExist: + def test_three_policies_present(self): + files = sorted(os.listdir(POLICY_DIR)) + assert "forbid-plaintext-secrets.json" in files + assert "forbid-iam-wildcard.json" in files + assert "require-kms-reference.json" in files + + def test_policies_are_valid_json(self): + for f in os.listdir(POLICY_DIR): + if f.endswith(".json"): + with open(POLICY_DIR / f, "r", encoding="utf-8") as fh: + data = json.load(fh) + assert data["apiVersion"] == "json.kyverno.io/v1alpha1" + assert data["kind"] == "ValidatingPolicy" + assert "nova.cloudinit.dev/severity" in data["metadata"]["annotations"] \ No newline at end of file diff --git a/tests/test_run_platform_plan_json_policies.py b/tests/test_run_platform_plan_json_policies.py new file mode 100644 index 0000000..f31501d --- /dev/null +++ b/tests/test_run_platform_plan_json_policies.py @@ -0,0 +1,59 @@ +"""Tests for run_platform.sh Step 5b kyverno-json wiring (REQ-302, v1.25). + +Asserts the script has the kyverno-json Step 5b block and the PCR-merge +logic. Pattern from tests/test_pipeline.py:79-95 (read script text + +assert substrings). +""" + +import os +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +SCRIPT = Path(__file__).resolve().parent.parent / "scripts" / "run_platform.sh" + + +def _read_script(): + with open(SCRIPT, "r", encoding="utf-8") as fh: + return fh.read() + + +class TestStep5bKyvernoJsonWiring: + def test_step_5b_block_present(self): + s = _read_script() + assert "Step 5b: kyverno-json plan-JSON policies" in s, \ + "run_platform.sh must have a Step 5b kyverno-json block (REQ-301)" + + def test_kj_scan_invocation_present(self): + s = _read_script() + assert "adapters/kyverno-json/policies/plan-json" in s, \ + "Step 5b must reference the plan-json policy dir" + + def test_kj_not_installed_skip_present(self): + s = _read_script() + assert "kyverno-json not installed; skipping plan-JSON policies" in s, \ + "Step 5b must skip gracefully when kj is absent (D-120)" + assert "D-120 graceful degradation" in s + + def test_pcr_merge_logic_present(self): + s = _read_script() + assert "merged PCR list" in s, \ + "Step 5b must merge the Checkov/Wiz + kj PCR lists" + + def test_command_v_kj_guard_present(self): + s = _read_script() + assert "command -v kj" in s, \ + "Step 5b must guard on `command -v kj` (is_configured)" + + +class TestExistingPipelineUnchanged: + def test_step_5_still_present(self): + s = _read_script() + assert "Step 5: runtime policy scan" in s + + def test_step_7_confidence_still_present(self): + s = _read_script() + assert "Step 7: confidence signal compute" in s \ No newline at end of file