continuous-intelligence/acdl
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 128 Wiki Activity

fix(P03 W0.5): kyverno-json substrate works with real kj (engine + policies + install script)

Browse Source 
The v1.25 kyverno-json engine adapter and policies were authored but never
validated against the real `kj` binary — the test suite
`pytest.skip("kj not installed")` when `kj` was absent, masking the bug.
With `kj` v0.0.3 now installed, the 3 failing-fixture tests
(stack-ir/plan-json/regression) showed 0 fails (all passed falsely). Root
causes (3 substrate bugs) and fixes:

1. ENGINE — bare-list output format. `kj scan --output json` emits a bare
   JSON LIST at the top level (NOT `{"results": [...]}`); each entry has
   `resource` + `results[].rules[]` with `violations[]` (fail) / `error`
   string (eval error) / neither (pass). The v1.25 `_translate` did
   `out.get("results", [])` on a dict → `out` is a list → returned `[]` →
   emitted a single KJ_NO_RESULTS pass PCR. Rewrote `_translate` to parse
   the real v0.0.3 nested shape (policy.metadata.name, rule.name,
   violations[].errors[].field/detail/value). Future-proofs to also accept
   the legacy dict shape. Preserves RESULT_MAP, severity-from-annotation,
   is_configured(), _skipped_not_configured, _error_pcr, the temp-file
   payload write, and the subprocess invocation.

2. ENGINE — `.json` policies not loaded by `kj`. The upstream loader
   (pkg/policy/load.go) uses fileinfo.IsYaml() which only matches
   `.yaml`/`.yml` — `.json` files are silently skipped (0 policies).
   Nova policies are authored as `.json` (TestPolicyFilesExist asserts the
   filenames). Added `_materialize_yaml_policy_dir`: mirrors the source
   tree to a temp dir, copying every `.json` policy to a `.yaml` twin
   (JSON is a valid YAML subset, verified against kj v0.0.3). Source
   `.json` files remain untouched.

3. POLICIES — `validate` wrapper + check syntax. Removed the `validate`
   wrapper from all 16 policies (kj v0.0.3 ignores `validate`-wrapped
   rules — `assert` goes directly under the rule). Fixed the check syntax:
   a check entry is `expression: expected_value` (e.g.
   `(regex_match(..., @)): true`), not `field: (expression)` (which
   compared a bool to nothing → "types not comparable"). For per-resource
   checks over stack-IR/plan-JSON, `~.resources` (descendant anchor) is
   required for per-element iteration; a plain path applies to the whole
   array. For type-scoped rules (s3/ebs encryption, iam/db/kms), the type
   guard is folded into the expression (`type == '...' && !<has-prop>`)
   so non-matching resources short-circuit to false. cap-013 dedup uses
   `max(map(&length(@), values(group_by(adapters, &@)))) == `1`` (no
   `duplicates` JMESPath fn exists). Preserved all policy metadata
   (apiVersion, kind, metadata.name, severity + title annotations) —
   TestPolicyValidity/TestPolicyFilesExist still pass.

INSTALL SCRIPT — the v1.25 `go install .../cmd/kj@latest` failed: the
`cmd/kj` path does not exist in v0.0.3 (upstream produces a binary named
`kyverno-json`). Fixed to `go install github.com/kyverno/kyverno-json@latest`
+ symlink `kyverno-json` → `kj` (GOBIN and /usr/local/bin fallbacks).
Idempotent: short-circuits when `kj` is already on PATH and working.

Verification: `which kj` → /usr/local/bin/kj; `kj version` → v0.0.3.
test_kyverno_json_engine + test_stack_ir_policies + test_plan_json_policies
+ test_meta_policies + test_regression_policies: 36 passed, 0 skips
(_require_kj no longer skips). Full suite (excluding pre-existing hang in
test_verify_regression_mode.py): 776 passed, 6 failed — all 6 failures are
pre-existing (confirmed by stashing this commit's diff and re-running);
the only in-scope-acceptable failure is
test_module_standards.py::test_all_l1_have_required_files (dynamodb
extension drift, data-engineer's later wave).

---ci---
project: acdl
phase: 3
milestone: v1.26
status: execute
wave: W0.5
---
This commit is contained in:
Jon Chery
2026-08-18 21:29:12 +00:00
parent a63c85bc51
commit 59d837f6e7
18 changed files with 469 additions and 280 deletions
Download Patch File Download Diff File Expand all files Collapse all files
adapters/kyverno-json/kyverno_json_engine.py
+263 -62
View File
@@ -1,4 +1,4 @@
"""Nova KyvernoJsonEngine (REQ-293, v1.25). """Nova KyvernoJsonEngine (REQ-293, v1.25; fixed v1.26 P3 W0.5).
Implements the ``PolicyEngine`` protocol (``core/policy_engine.py``) Implements the ``PolicyEngine`` protocol (``core/policy_engine.py``)
by shelling to the ``kj`` CLI (``kyverno-json``). Translates native by shelling to the ``kj`` CLI (``kyverno-json``). Translates native
@@ -12,9 +12,10 @@ distinguish from the K8s Kyverno adapter's ``KYVERNO_`` prefix.
Severity (RESEARCH §2.6, G-Q10a): kyverno-json does not natively assign Severity (RESEARCH §2.6, G-Q10a): kyverno-json does not natively assign
severities. Each Nova policy declares its severity via a severities. Each Nova policy declares its severity via a
``metadata.annotations["nova.cloudinit.dev/severity"]`` field. The ``metadata.annotations["nova.cloudinit.dev/severity"]`` field. The
engine reads this annotation from the loaded policy YAML (not from the engine reads this annotation from the loaded policy file (not from the
scan result — the result doesn't carry it) and applies it to every scan result — the result carries the policy spec but the annotation is
result that policy produces. Default when absent: ``"info"``. read here from disk) and applies it to every result that policy
produces. Default when absent: ``"info"``.
Graceful degradation (D-120): ``is_configured()`` returns ``False`` when Graceful degradation (D-120): ``is_configured()`` returns ``False`` when
``which kj`` is absent → ``evaluate()`` returns a single SKIPPED PCR ``which kj`` is absent → ``evaluate()`` returns a single SKIPPED PCR
@@ -24,6 +25,37 @@ the binary.
Defensive parsing: any kyverno-json output that doesn't match the Defensive parsing: any kyverno-json output that doesn't match the
expected shape produces an ``error`` PCR, never an exception. The expected shape produces an ``error`` PCR, never an exception. The
engine is read-only against a local policy dir + a temp payload file. engine is read-only against a local policy dir + a temp payload file.
v1.26 P3 W0.5 fix — three substrate bugs uncovered once ``kj`` was
actually installed (the v1.25 test suite ``pytest.skip``-masked them):
1. **``.json`` policy files are not loaded by ``kj`` v0.0.3.** The
upstream policy loader (``pkg/policy/load.go``) uses
``fileinfo.IsYaml()`` which only matches ``.yaml``/``.yml``
extensions — ``.json`` files are silently skipped, yielding
``evaluating N resources against 0 policies``. Nova policies are
authored as ``.json`` (the ``TestPolicyFilesExist`` tests assert the
``.json`` filenames). Fix: ``evaluate()`` materializes a temp policy
dir that mirrors the source tree with every ``.json`` policy copied
to a ``.yaml`` twin (JSON is a valid YAML subset — verified against
``kj`` v0.0.3). The source ``.json`` files remain untouched.
2. **Bare-list output format.** ``kj scan --output json`` emits a bare
JSON list at the top level (NOT ``{"results": [...]}``). Each entry
has ``resource`` (the evaluated payload) + ``results`` (list of
per-policy result objects, each carrying ``policy.metadata.name``,
``rules[]`` with ``rule.name``, ``violations[]`` (present on fail),
``error`` (string, present on policy-evaluation error)). The v1.25
``_translate`` did ``out.get("results", [])`` on a dict — but
``out`` is a list → returned ``[]`` → emitted a single
``KJ_NO_RESULTS`` pass PCR. **This is why all failing fixtures showed
0 fails.** Fix: ``_translate`` handles list (v0.0.3) and dict
(future-proof) shapes.
3. **``validate`` wrapper + check syntax.** Documented in the policy
files themselves (see the W0.5 policy edits). The engine itself does
not enforce policy shape — it only translates ``kj`` output — so
this fix lives in the policy ``.json`` files.
""" """
import datetime import datetime
@@ -98,39 +130,41 @@ def _load_policy_severities(policy_dir: Path) -> dict[str, str]:
return severities return severities
def _to_pcr(entry: dict, contract_id: str, severity: str) -> dict: def _materialize_yaml_policy_dir(src: Path) -> tuple[Path, bool]:
"""Translate a kyverno-json scan result entry to a PCR dict.""" """Mirror ``src`` (recursively) into a temp dir, copying every
policy_name = entry.get("policy", "") or "UNKNOWN" ``.json`` policy to a ``.yaml`` twin and copying ``.yaml``/``.yml``
rule_name = entry.get("rule", "") or "" files verbatim. Returns ``(temp_dir, created)``.
rule_id = f"KJ_{policy_name}"
if rule_name: ``kj`` v0.0.3's policy loader (``pkg/policy/load.go``) only matches
rule_id = f"{rule_id}/{rule_name}" ``.yaml``/``.yml`` extensions — ``.json`` files are silently
result_raw = entry.get("result", "skip") skipped. Nova policies are authored as ``.json`` (the
result = RESULT_MAP.get(str(result_raw).lower(), "error") ``TestPolicyFilesExist`` tests assert the ``.json`` filenames, so
message = entry.get("message", "") or "" they cannot be renamed in-place). JSON is a valid YAML subset, so
resource = entry.get("resource", "") a byte-for-byte copy with a ``.yaml`` extension loads cleanly.
if not resource and entry.get("name"):
kind = entry.get("kind", "") ``created`` is ``False`` when ``src`` contains no policy files at
ns = entry.get("namespace", "") all (empty dir) — in that case the temp dir is still returned (the
resource = f"{kind}/{ns}/{entry.get('name')}" if kind else entry.get("name", "") caller invokes ``kj`` against it and gets the no-results path).
return { """
"contractId": contract_id, tmp = Path(tempfile.mkdtemp(prefix="nova-kj-pol-"))
"evaluatedAt": _iso8601_now(), any_policy = False
"engine": "kyverno", if src.is_dir():
"ruleId": rule_id, for root, _dirs, files in os.walk(src):
"severity": severity, rel = Path(root).relative_to(src)
"result": result, dest_root = tmp / rel
"message": message, dest_root.mkdir(parents=True, exist_ok=True)
"evidence": { for fn in files:
"resource": resource, if fn.startswith(".") or fn.startswith("_"):
"policy": policy_name, continue
"rule": rule_name, src_file = Path(root) / fn
"namespace": entry.get("namespace", ""), if fn.endswith(".json"):
"kind": entry.get("kind", ""), dest_file = dest_root / (fn.rsplit(".", 1)[0] + ".yaml")
"name": entry.get("name", ""), shutil.copy2(src_file, dest_file)
}, any_policy = True
"resourceRef": resource, elif fn.endswith((".yaml", ".yml")):
} shutil.copy2(src_file, dest_root / fn)
any_policy = True
return tmp, any_policy
def _skipped_not_configured(contract_id: str) -> dict: def _skipped_not_configured(contract_id: str) -> dict:
@@ -165,6 +199,23 @@ def _error_pcr(contract_id: str, message: str) -> dict:
} }
def _no_results_pass(contract_id: str) -> dict:
"""No result entries — emit a single pass PCR so the confidence
signal's policy input is non-empty (a non-empty list of passes →
score 1.0)."""
return {
"contractId": contract_id,
"evaluatedAt": _iso8601_now(),
"engine": "kyverno",
"ruleId": "KJ_NO_RESULTS",
"severity": "info",
"result": "pass",
"message": "kyverno-json scan produced no result entries (all policies passed or no match).",
"evidence": {},
"resourceRef": "",
}
class KyvernoJsonEngine: class KyvernoJsonEngine:
"""``PolicyEngine`` impl that shells to the ``kj`` CLI.""" """``PolicyEngine`` impl that shells to the ``kj`` CLI."""
@@ -185,6 +236,9 @@ class KyvernoJsonEngine:
f"kyverno-json policy dir not found: {policy_dir}", f"kyverno-json policy dir not found: {policy_dir}",
)] )]
severities = _load_policy_severities(policy_dir) severities = _load_policy_severities(policy_dir)
# kj v0.0.3 only loads .yaml/.yml policy files. Mirror the tree
# to a temp dir with .json policies copied to .yaml twins.
yaml_dir, _any_policy = _materialize_yaml_policy_dir(policy_dir)
# Write payload to temp file (kj scan --payload expects a file path). # Write payload to temp file (kj scan --payload expects a file path).
payload_tmp = tempfile.NamedTemporaryFile( payload_tmp = tempfile.NamedTemporaryFile(
mode="w", suffix=".json", delete=False, encoding="utf-8" mode="w", suffix=".json", delete=False, encoding="utf-8"
@@ -195,7 +249,7 @@ class KyvernoJsonEngine:
payload_tmp.close() payload_tmp.close()
cmd = [ cmd = [
kj, "scan", kj, "scan",
"--policy", str(policy_dir), "--policy", str(yaml_dir),
"--payload", payload_tmp.name, "--payload", payload_tmp.name,
"--output", "json", "--output", "json",
] ]
@@ -211,7 +265,7 @@ class KyvernoJsonEngine:
f"kyverno-json scan exited {proc.returncode}: {proc.stderr[:200]}", f"kyverno-json scan exited {proc.returncode}: {proc.stderr[:200]}",
)] )]
try: try:
out = json.loads(proc.stdout) if proc.stdout.strip() else {} out = json.loads(proc.stdout) if proc.stdout.strip() else []
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
return [_error_pcr( return [_error_pcr(
contract_id, contract_id,
@@ -223,38 +277,185 @@ class KyvernoJsonEngine:
os.unlink(payload_tmp.name) os.unlink(payload_tmp.name)
except OSError: except OSError:
pass pass
shutil.rmtree(yaml_dir, ignore_errors=True)
def _translate(self, out: dict, contract_id: str, def _translate(self, out: Any, contract_id: str,
severities: dict[str, str]) -> list[dict]: severities: dict[str, str]) -> list[dict]:
results = out.get("results", []) if isinstance(out, dict) else [] # kj v0.0.3 emits a BARE JSON LIST at the top level: each entry
if not isinstance(results, list): # has `resource` (the evaluated payload) + `results` (list of
results = [] # per-policy result objects). Future-proof: also accept the
# legacy {"results": [...]} dict shape.
if isinstance(out, list):
entries = out
elif isinstance(out, dict):
entries = out.get("results", [])
if not isinstance(entries, list):
entries = []
else:
entries = []
pcrs: list[dict] = [] pcrs: list[dict] = []
for entry in results: for entry in entries:
if not isinstance(entry, dict): if not isinstance(entry, dict):
continue continue
policy_name = entry.get("policy", "") or "UNKNOWN" resource = entry.get("resource", {})
severity = severities.get(policy_name, SEVERITY_DEFAULT) results = entry.get("results", [])
pcrs.append(_to_pcr(entry, contract_id, severity)) if not isinstance(results, list):
results = []
for pol_result in results:
if not isinstance(pol_result, dict):
continue
policy_obj = pol_result.get("policy", {}) or {}
policy_name = (
policy_obj.get("metadata", {}).get("name") if isinstance(policy_obj, dict)
else None
) or "UNKNOWN"
severity = severities.get(policy_name, SEVERITY_DEFAULT)
rules = pol_result.get("rules", [])
if not isinstance(rules, list):
rules = []
for rule_entry in rules:
if not isinstance(rule_entry, dict):
continue
rule_obj = rule_entry.get("rule", {}) or {}
rule_name = rule_obj.get("name", "") if isinstance(rule_obj, dict) else ""
rule_id = f"KJ_{policy_name}"
if rule_name:
rule_id = f"{rule_id}/{rule_name}"
violations = rule_entry.get("violations")
error_str = rule_entry.get("error")
if isinstance(violations, list) and violations:
# Fail: build a message from the violations' errors.
msg_parts: list[str] = []
for v in violations:
if not isinstance(v, dict):
continue
for err in v.get("errors", []) or []:
if not isinstance(err, dict):
continue
field = err.get("field", "")
detail = err.get("detail", "")
value = err.get("value", "")
msg_parts.append(
f"{field}: value={value!r} detail={detail}"
)
message = "; ".join(msg_parts) if msg_parts else "policy rule failed"
pcrs.append({
"contractId": contract_id,
"evaluatedAt": _iso8601_now(),
"engine": "kyverno",
"ruleId": rule_id,
"severity": severity,
"result": "fail",
"message": message,
"evidence": {
"resource": resource,
"policy": policy_name,
"rule": rule_name,
"violations": violations,
},
"resourceRef": _resource_ref(resource),
})
elif isinstance(error_str, str) and error_str:
# Policy-evaluation error (e.g. bad JMESPath).
pcrs.append({
"contractId": contract_id,
"evaluatedAt": _iso8601_now(),
"engine": "kyverno",
"ruleId": rule_id,
"severity": severity,
"result": "error",
"message": error_str,
"evidence": {
"resource": resource,
"policy": policy_name,
"rule": rule_name,
},
"resourceRef": _resource_ref(resource),
})
else:
# Pass: no violations, no error.
pcrs.append({
"contractId": contract_id,
"evaluatedAt": _iso8601_now(),
"engine": "kyverno",
"ruleId": rule_id,
"severity": severity,
"result": "pass",
"message": "",
"evidence": {
"resource": resource,
"policy": policy_name,
"rule": rule_name,
},
"resourceRef": _resource_ref(resource),
})
if not pcrs: if not pcrs:
# No results — kyverno-json produced nothing (no match, or pcrs.append(_no_results_pass(contract_id))
# all policies passed with no result entries). Emit a
# single pass PCR so the confidence signal's policy input
# is non-empty (a non-empty list of passes → score 1.0).
pcrs.append({
"contractId": contract_id,
"evaluatedAt": _iso8601_now(),
"engine": "kyverno",
"ruleId": "KJ_NO_RESULTS",
"severity": "info",
"result": "pass",
"message": "kyverno-json scan produced no result entries (all policies passed or no match).",
"evidence": {},
"resourceRef": "",
})
return pcrs return pcrs
def _resource_ref(resource: Any) -> str:
"""Best-effort resource ref from the evaluated payload."""
if isinstance(resource, dict):
for key in ("id", "name", "address"):
v = resource.get(key)
if isinstance(v, str) and v:
return v
return ""
# --- Legacy _to_pcr kept for the existing TestToPcr unit tests ---
# (test_kyverno_json_engine.py::TestToPcr constructs flat `entry`
# dicts with `policy`/`rule`/`result`/`message`/`resource` keys and
# asserts the translated PCR shape. The production _translate path no
# longer calls this helper — it inlines the translation against the
# real kj v0.0.3 nested output — but the unit tests pin the helper's
# contract, so it stays.)
def _to_pcr(entry: dict, contract_id: str, severity: str) -> dict:
"""Translate a flat kyverno-json scan result entry to a PCR dict.
Legacy shape (kept for unit-test backwards compatibility): the
entry is a flat dict with ``policy``/``rule``/``result``/``message``/
``resource`` string keys. The production ``_translate`` path no
longer calls this — it inlines translation against the real kj
v0.0.3 nested ``resource``+``results``+``rules`` shape — but the
``TestToPcr`` unit tests pin this contract.
"""
policy_name = entry.get("policy", "") or "UNKNOWN"
rule_name = entry.get("rule", "") or ""
rule_id = f"KJ_{policy_name}"
if rule_name:
rule_id = f"{rule_id}/{rule_name}"
result_raw = entry.get("result", "skip")
result = RESULT_MAP.get(str(result_raw).lower(), "error")
message = entry.get("message", "") or ""
resource = entry.get("resource", "")
if not resource and entry.get("name"):
kind = entry.get("kind", "")
ns = entry.get("namespace", "")
resource = f"{kind}/{ns}/{entry.get('name')}" if kind else entry.get("name", "")
return {
"contractId": contract_id,
"evaluatedAt": _iso8601_now(),
"engine": "kyverno",
"ruleId": rule_id,
"severity": severity,
"result": result,
"message": message,
"evidence": {
"resource": resource,
"policy": policy_name,
"rule": rule_name,
"namespace": entry.get("namespace", ""),
"kind": entry.get("kind", ""),
"name": entry.get("name", ""),
},
"resourceRef": resource,
}
if __name__ == "__main__": if __name__ == "__main__":
if len(sys.argv) < 4: if len(sys.argv) < 4:
print( print(
adapters/kyverno-json/policies/_smoke.json
+8 -9
View File
@@ -12,17 +12,16 @@
"rules": [ "rules": [
{ {
"name": "require-id", "name": "require-id",
"validate": { "assert": {
"message": "contract id is required", "all": [
"assert": { {
"all": [ "check": {
{ "id": {
"check": { "(regex_match('^[a-z][a-z0-9-]{2,5}$', @))": true
"id": "(regex_match('^[a-z][a-z0-9-]{2,5}$', @))"
} }
} }
] }
} ]
} }
} }
] ]
adapters/kyverno-json/policies/contract/forbid-unknown-fields.json
+9 -10
View File
@@ -12,18 +12,17 @@
"rules": [ "rules": [
{ {
"name": "no-unknown-fields", "name": "no-unknown-fields",
"validate": { "assert": {
"message": "contract may only contain id, name, environment, infrastructure (schema-allowed fields)", "all": [
"assert": { {
"all": [ "check": {
{ "(length(keys(@)) == `4`)": true,
"check": { "keys(@)": {
"(length(keys(@)) == `4`)": true, "(contains(['id','name','environment','infrastructure'], @))": true
"keys(@)": "(contains(['id','name','environment','infrastructure'], @))"
} }
} }
] }
} ]
} }
} }
] ]
adapters/kyverno-json/policies/contract/require-env-in-enum.json
+8 -9
View File
@@ -12,17 +12,16 @@
"rules": [ "rules": [
{ {
"name": "env-enum", "name": "env-enum",
"validate": { "assert": {
"message": "contract.environment must be one of dev, qa, prod, dr", "all": [
"assert": { {
"all": [ "check": {
{ "environment": {
"check": { "(contains(['dev','qa','prod','dr'], @))": true
"environment": "(contains(['dev','qa','prod','dr'], @))"
} }
} }
] }
} ]
} }
} }
] ]
adapters/kyverno-json/policies/contract/require-id-pattern.json
+8 -9
View File
@@ -12,17 +12,16 @@
"rules": [ "rules": [
{ {
"name": "id-pattern", "name": "id-pattern",
"validate": { "assert": {
"message": "contract.id must match ^[a-z][a-z0-9-]{2,5}$ (3-6 char operational acronym)", "all": [
"assert": { {
"all": [ "check": {
{ "id": {
"check": { "(regex_match('^[a-z][a-z0-9-]{2,5}$', @))": true
"id": "(regex_match('^[a-z][a-z0-9-]{2,5}$', @))"
} }
} }
] }
} ]
} }
} }
] ]
adapters/kyverno-json/policies/contract/require-infrastructure-min-1.json
+8 -9
View File
@@ -12,17 +12,16 @@
"rules": [ "rules": [
{ {
"name": "infra-min-1", "name": "infra-min-1",
"validate": { "assert": {
"message": "contract.infrastructure must have at least one module entry", "all": [
"assert": { {
"all": [ "check": {
{ "infrastructure": {
"check": { "(length(keys(@)) > `0`)": true
"infrastructure": "(length(keys(@)) > `0`)"
} }
} }
] }
} ]
} }
} }
] ]
adapters/kyverno-json/policies/meta/block-on-any-critical.json
+7 -12
View File
@@ -12,19 +12,14 @@
"rules": [ "rules": [
{ {
"name": "no-critical-fail", "name": "no-critical-fail",
"validate": { "assert": {
"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).", "all": [
"assert": { {
"all": [ "check": {
{ "(severity == 'critical' && result == 'fail')": false
"check": {
"~.[]": {
"(severity == 'critical' && result == 'fail')": false
}
}
} }
] }
} ]
} }
} }
] ]
adapters/kyverno-json/policies/meta/tagging-rules-agree.json
+12 -21
View File
@@ -12,28 +12,19 @@
"rules": [ "rules": [
{ {
"name": "no-tagging-divergence", "name": "no-tagging-divergence",
"validate": { "assert": {
"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).", "all": [
"assert": { {
"all": [ "check": {
{ "(ruleId == 'NOVA_TAG_NAMING' && result == 'fail')": false
"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
}
}
} }
] },
} {
"check": {
"(ruleId == 'KJ_REQUIRE_TAGGING_STANDARD' && result == 'fail')": false
}
}
]
} }
} }
] ]
adapters/kyverno-json/policies/plan-json/forbid-iam-wildcard.json
+22 -20
View File
@@ -12,36 +12,38 @@
"rules": [ "rules": [
{ {
"name": "no-wildcard-action", "name": "no-wildcard-action",
"validate": { "assert": {
"message": "IAM policy Action must not be '*' (ports CKV_AWS_1/40)", "all": [
"assert": { {
"all": [ "check": {
{ "planned_values": {
"check": { "root_module": {
"planned_values.root_module.~.resources": { "~.resources": {
"(type == 'aws_iam_policy' && contains(values.policy_document.Statement[].Action, '*'))": false "(type == 'aws_iam_policy' && contains(values.policy_document.Statement[].Action, '*'))": false
}
} }
} }
} }
] }
} ]
} }
}, },
{ {
"name": "no-wildcard-resource", "name": "no-wildcard-resource",
"validate": { "assert": {
"message": "IAM policy Resource must not be '*' (ports CKV_AWS_1/40)", "all": [
"assert": { {
"all": [ "check": {
{ "planned_values": {
"check": { "root_module": {
"planned_values.root_module.~.resources": { "~.resources": {
"(type == 'aws_iam_policy' && contains(values.policy_document.Statement[].Resource, '*'))": false "(type == 'aws_iam_policy' && contains(values.policy_document.Statement[].Resource, '*'))": false
}
} }
} }
} }
] }
} ]
} }
} }
] ]
adapters/kyverno-json/policies/plan-json/forbid-plaintext-secrets.json
+11 -10
View File
@@ -12,19 +12,20 @@
"rules": [ "rules": [
{ {
"name": "no-plaintext-db-password", "name": "no-plaintext-db-password",
"validate": { "assert": {
"message": "aws_db_instance.password must not be a plaintext string (ports CKV_AWS_41/45/46)", "all": [
"assert": { {
"all": [ "check": {
{ "planned_values": {
"check": { "root_module": {
"planned_values.root_module.~.resources": { "~.resources": {
"(type == 'aws_db_instance' && contains(keys(values), 'password') && !contains(['${...}', ''], values.password))": false "(type == 'aws_db_instance' && contains(keys(values), 'password') && !contains(['${...}', ''], values.password))": false
}
} }
} }
} }
] }
} ]
} }
} }
] ]
adapters/kyverno-json/policies/plan-json/require-kms-reference.json
+11 -10
View File
@@ -12,19 +12,20 @@
"rules": [ "rules": [
{ {
"name": "kms-by-alias", "name": "kms-by-alias",
"validate": { "assert": {
"message": "aws_kms_key resources should reference a customer-managed key alias, not inline key material (ports CKV_AWS_7/33)", "all": [
"assert": { {
"all": [ "check": {
{ "planned_values": {
"check": { "root_module": {
"planned_values.root_module.~.resources": { "~.resources": {
"(type == 'aws_kms_key' && !contains(keys(values), 'key_id') && !contains(keys(values), 'kms_key_id'))": false "(type == 'aws_kms_key' && !contains(keys(values), 'key_id') && !contains(keys(values), 'kms_key_id'))": false
}
} }
} }
} }
] }
} ]
} }
} }
] ]
adapters/kyverno-json/policies/regression/cap-013-adapter-dedup.json
+7 -10
View File
@@ -12,17 +12,14 @@
"rules": [ "rules": [
{ {
"name": "no-duplicate-adapters", "name": "no-duplicate-adapters",
"validate": { "assert": {
"message": "Each adapter must be registered exactly once (no duplicate adapter names in the capability inventory). Declarative mirror of core/regression_verify.py CAP-013.", "all": [
"assert": { {
"all": [ "check": {
{ "(max(map(&length(@), values(group_by(adapters, &@)))) == `1`)": true
"check": {
"adapters": "(length(duplicates(@)) == `0`)"
}
} }
] }
} ]
} }
} }
] ]
adapters/kyverno-json/policies/regression/cap-023-metrics-collector.json
+8 -11
View File
@@ -12,19 +12,16 @@
"rules": [ "rules": [
{ {
"name": "every-metric-has-status", "name": "every-metric-has-status",
"validate": { "assert": {
"message": "Every metric in docs/METRICS.md must declare a status (grounded, derived, or deferred). Declarative mirror of core/regression_verify.py CAP-023.", "all": [
"assert": { {
"all": [ "check": {
{ "~.metrics": {
"check": { "(contains(['grounded','derived','deferred'], status))": true
"~.metrics": {
"(contains(['grounded','derived','deferred'], status))": true
}
} }
} }
] }
} ]
} }
} }
] ]
adapters/kyverno-json/policies/regression/cap-024-deck-structure.json
+11 -14
View File
@@ -12,22 +12,19 @@
"rules": [ "rules": [
{ {
"name": "deck-has-4-beats", "name": "deck-has-4-beats",
"validate": { "assert": {
"message": "The deck must have the 4-beat arc: Problem, Solution, Proof, Roadmap+Ask. Declarative mirror of core/regression_verify.py CAP-024.", "all": [
"assert": { {
"all": [ "check": {
{ "deck": {
"check": { "beats": {
"deck.beats": "(length(@) >= `4`)" "(length(@) >= `4`)": true,
} "(contains(@, 'Problem') && contains(@, 'Solution') && contains(@, 'Proof') && contains(@, 'Roadmap+Ask'))": true
}, }
{
"check": {
"deck.beats": "(contains(@, 'Problem') && contains(@, 'Solution') && contains(@, 'Proof') && contains(@, 'Roadmap+Ask'))"
} }
} }
] }
} ]
} }
} }
] ]
adapters/kyverno-json/policies/stack-ir/forbid-public-ingress.json
+8 -11
View File
@@ -13,19 +13,16 @@
{ {
"name": "no-public-ingress", "name": "no-public-ingress",
"identifier": "id", "identifier": "id",
"validate": { "assert": {
"message": "public_ingress: true is not allowed on any resource (v1.0 demo rule, now declarative)", "all": [
"assert": { {
"all": [ "check": {
{ "~.resources": {
"check": { "(inputs.public_ingress || `false`)": false
"~.resources": {
"(inputs.public_ingress || `false`)": false
}
} }
} }
] }
} ]
} }
} }
] ]
adapters/kyverno-json/policies/stack-ir/require-encryption-by-default.json
+16 -28
View File
@@ -13,43 +13,31 @@
{ {
"name": "s3-encryption", "name": "s3-encryption",
"identifier": "id", "identifier": "id",
"match": { "assert": {
"any": [ "all": [
{"type": "aws:s3:bucket"} {
] "check": {
}, "~.resources": {
"validate": { "(type == 'aws:s3:bucket' && !(contains(keys(inputs), 'bucket_encryption') || contains(keys(inputs), 'kms_key_id')))": false
"message": "S3 buckets must declare encryption config (inputs.bucket_encryption or inputs.kms_key_id)",
"assert": {
"all": [
{
"check": {
"(contains(keys(inputs), 'bucket_encryption') || contains(keys(inputs), 'kms_key_id'))": true
} }
} }
] }
} ]
} }
}, },
{ {
"name": "ebs-encryption", "name": "ebs-encryption",
"identifier": "id", "identifier": "id",
"match": { "assert": {
"any": [ "all": [
{"type": "aws:ebs:volume"} {
] "check": {
}, "~.resources": {
"validate": { "(type == 'aws:ebs:volume' && !(contains(keys(inputs), 'encrypted') || contains(keys(inputs), 'kms_key_id')))": false
"message": "EBS volumes must declare encryption (inputs.encrypted or inputs.kms_key_id)",
"assert": {
"all": [
{
"check": {
"(contains(keys(inputs), 'encrypted') || contains(keys(inputs), 'kms_key_id'))": true
} }
} }
] }
} ]
} }
} }
] ]
adapters/kyverno-json/policies/stack-ir/require-tagging-standard.json
+11 -14
View File
@@ -13,22 +13,19 @@
{ {
"name": "require-nova-tags", "name": "require-nova-tags",
"identifier": "id", "identifier": "id",
"validate": { "assert": {
"message": "Every taggable resource must carry nova:owner, nova:contract, nova:environment, nova:cost-center tags", "all": [
"assert": { {
"all": [ "check": {
{ "~.resources": {
"check": { "(contains(keys(inputs.tags || `{}`), 'nova:owner'))": true,
"~.resources": { "(contains(keys(inputs.tags || `{}`), 'nova:contract'))": true,
"(contains(keys(tags || `[]`), 'nova:owner'))": true, "(contains(keys(inputs.tags || `{}`), 'nova:environment'))": true,
"(contains(keys(tags || `[]`), 'nova:contract'))": true, "(contains(keys(inputs.tags || `{}`), 'nova:cost-center'))": true
"(contains(keys(tags || `[]`), 'nova:environment'))": true,
"(contains(keys(tags || `[]`), 'nova:cost-center'))": true
}
} }
} }
] }
} ]
} }
} }
] ]
scripts/install-kyverno-json.sh
+41 -11
View File
@@ -1,9 +1,18 @@
#!/usr/bin/env bash #!/usr/bin/env bash
# scripts/install-kyverno-json.sh — install the kj CLI (v1.25, REQ-294) # scripts/install-kyverno-json.sh — install the kj CLI (v1.25, REQ-294;
# fixed v1.26 P3 W0.5).
# #
# Installs the kyverno-json CLI (`kj`) via `go install` (D-115). The # Installs the kyverno-json CLI via `go install` (D-115). The binary is a
# binary is a Go project — not a Python package. Cached via the Go # Go project — not a Python package. Cached via the Go module cache.
# module cache. #
# v1.26 P3 W0.5 fix: the v1.25 script ran
# go install github.com/kyverno/kyverno-json/cmd/kj@latest
# but the `cmd/kj` path does NOT exist in v0.0.3 — the upstream
# `go install github.com/kyverno/kyverno-json@latest` produces a binary
# named `kyverno-json`, NOT `kj`. The v1.25 invocation failed silently
# (the test suite masked it via `pytest.skip("kj not installed")`). This
# script now installs the real module and symlinks `kyverno-json` → `kj`
# so the engine's `which kj` check passes.
# #
# Usage: bash scripts/install-kyverno-json.sh # Usage: bash scripts/install-kyverno-json.sh
# Exits 0 on success, 1 if Go is not installed, 2 if `kj version` fails. # Exits 0 on success, 1 if Go is not installed, 2 if `kj version` fails.
@@ -11,19 +20,40 @@ set -euo pipefail
if ! command -v go >/dev/null 2>&1; then if ! command -v go >/dev/null 2>&1; then
echo "ERROR: Go toolchain not found. Install Go (https://go.dev/dl/) first." >&2 echo "ERROR: Go toolchain not found. Install Go (https://go.dev/dl/) first." >&2
echo " kyverno-json is a Go binary — `go install` is the upstream-blessed path (D-115)." >&2 echo " kyverno-json is a Go binary — \`go install\` is the upstream-blessed path (D-115)." >&2
exit 1 exit 1
fi fi
echo "Installing kyverno-json CLI (kj) via go install..."
GOBIN="${GOBIN:-${HOME}/go/bin}" GOBIN="${GOBIN:-${HOME}/go/bin}"
go install github.com/kyverno/kyverno-json/cmd/kj@latest
# Idempotent: if kj is already on PATH and working, short-circuit.
if command -v kj >/dev/null 2>&1 && kj version >/dev/null 2>&1; then
echo "kj installed:"
kj version
echo "DONE"
exit 0
fi
echo "Installing kyverno-json CLI (kyverno-json) via go install..."
# The upstream module produces a binary named `kyverno-json` (NOT `kj`).
# The v1.25 `go install .../cmd/kj@latest` path does not exist in v0.0.3.
go install github.com/kyverno/kyverno-json@latest
# The binary is named `kyverno-json`, not `kj`. Symlink it as `kj` for
# the engine's `which kj` check (kyverno_json_engine.py::_which_kj).
if [ -x "${GOBIN}/kyverno-json" ] && ! command -v kj >/dev/null 2>&1; then
ln -sf "${GOBIN}/kyverno-json" "${GOBIN}/kj"
# If GOBIN not on PATH, try /usr/local/bin so `which kj` resolves.
if ! command -v kj >/dev/null 2>&1; then
ln -sf "${GOBIN}/kyverno-json" /usr/local/bin/kj 2>/dev/null || true
fi
fi
if ! command -v kj >/dev/null 2>&1; then if ! command -v kj >/dev/null 2>&1; then
if [ -x "${GOBIN}/kj" ]; then if [ -x "${GOBIN}/kyverno-json" ]; then
echo "kj installed to ${GOBIN}/kj (not on PATH)" echo "kyverno-json installed to ${GOBIN}/kyverno-json but 'kj' is not on PATH." >&2
echo "add ${GOBIN} to PATH or symlink: ln -s ${GOBIN}/kj /usr/local/bin/kj" echo "add ${GOBIN} to PATH or symlink: ln -sf ${GOBIN}/kyverno-json /usr/local/bin/kj" >&2
"${GOBIN}/kj" version "${GOBIN}/kyverno-json" version
exit 0 exit 0
fi fi
echo "ERROR: kj not found on PATH after go install (checked ${GOBIN})." >&2 echo "ERROR: kj not found on PATH after go install (checked ${GOBIN})." >&2