feat(P04): kyverno-json ABAC policy + evaluator (REQ-339, D-227, C-5.1, security-engineer)

---ci---
project: acdl
phase: 4
milestone: v1.28
status: execute
persona: security-engineer
---

core/abac_evaluator.py

@@ -0,0 +1,145 @@
+"""Nova ABAC evaluator for the token-vend Lambda (REQ-339, C-6.1, D-231).
+
+Wraps :func:`core.policy_engine.get_engine` to evaluate the
+``platform/abac/token-vend.policy`` kyverno-json ``ValidatingPolicy``
+against a token-vend authorization payload and produce an allow/deny
+decision with the policy SHA (D-231).
+
+Payload shape (REQ-339, C-5.1)::
+
+    {
+      "subject":           {"id": ..., "role": ..., "owner": ...},
+      "requested_claims":  [<claim name>, ...],          # C-5.1
+      "target_resource":   {"type": ..., "id": ..., "owner": ..., "environment": ...},
+      "environment":       "dev" | "qa" | "prod" | "dr",
+      "pat_jti":           "<PAT jti>",
+      "policy_version":    "<git SHA>"
+    }
+
+Decision rule (C-6.1 fail-closed): **any** PCR with ``result == "fail"``
+and ``severity == "critical"`` → ``allowed=False``. The caller (the
+token-vend Lambda) is additionally required to fail closed when
+``KyvernoJsonEngine.is_configured()`` returns ``False`` or when this
+function raises — see ``tests/test_abac_fail_closed.py`` (the grill's
+#1 finding, INV-17).
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import Tuple
+
+from core.policy_engine import get_engine
+
+
+_POLICY_DIR = Path("platform/abac")
+_POLICY_FILE = _POLICY_DIR / "token-vend.policy"
+_CONTRACT_ID = "token-vend"
+
+
+def _materialize_policy_dir(src_dir: Path) -> Tuple[Path, bool]:
+    """Mirror ``src_dir`` to a temp dir, copying ``*.policy`` files to
+    ``*.json`` twins (JSON is a valid kyverno-json policy format; the
+    ``KyvernoJsonEngine`` only loads ``.json``/``.yaml``/``.yml``, and
+    Nova ABAC policies use the ``.policy`` extension per REQ-339, so a
+    byte-for-byte copy with a ``.json`` extension is required).
+
+    Returns ``(temp_dir, created)``; ``created`` is ``False`` when no
+    policy files were found. The caller is responsible for removing the
+    temp dir.
+    """
+    tmp = Path(tempfile.mkdtemp(prefix="nova-abac-pol-"))
+    any_policy = False
+    if src_dir.is_dir():
+        for entry in sorted(os.listdir(src_dir)):
+            if entry.startswith(".") or entry.startswith("_"):
+                continue
+            src_file = src_dir / entry
+            if not src_file.is_file():
+                continue
+            if entry.endswith(".policy"):
+                dest = tmp / (entry[: -len(".policy")] + ".json")
+                shutil.copy2(src_file, dest)
+                any_policy = True
+            elif entry.endswith((".json", ".yaml", ".yml")):
+                shutil.copy2(src_file, tmp / entry)
+                any_policy = True
+    return tmp, any_policy
+
+
+def _policy_sha() -> str:
+    """Return the git SHA of the policy file (D-231).
+
+    Uses ``git rev-parse HEAD:platform/abac/token-vend.policy`` so the
+    SHA is stable across checkouts (blob SHA, not commit SHA). Falls
+    back to ``"unknown"`` when git is unavailable or the file is not
+    tracked (e.g. during local development before the first commit).
+    """
+    repo_root = os.environ.get("NOVA_REPO_ROOT") or os.getcwd()
+    try:
+        sha = subprocess.check_output(
+            ["git", "rev-parse", "HEAD:platform/abac/token-vend.policy"],
+            cwd=repo_root,
+            stderr=subprocess.DEVNULL,
+            text=True,
+            timeout=5,
+        ).strip()
+        return sha or "unknown"
+    except Exception:
+        return "unknown"
+
+
+def evaluate_token_vend_policy(
+    payload: dict,
+) -> Tuple[bool, list, str]:
+    """Evaluate the token-vend ABAC policy against ``payload``.
+
+    Args:
+        payload: the ABAC authorization payload (see module docstring).
+
+    Returns:
+        ``(allowed, pcrs, policy_sha)`` where ``allowed`` is ``True``
+        iff no PCR has ``result == "fail"`` with ``severity ==
+        "critical"`` (C-6.1). ``pcrs`` is the raw list of
+        ``PolicyCheckResult`` dicts from the engine. ``policy_sha`` is
+        the git blob SHA of the policy file (D-231).
+
+    Raises:
+        Exception: any engine error propagates — the caller MUST catch
+            and fail closed (403 ``abac_eval_failed``). This function
+            does NOT swallow errors: failing closed is the *caller's*
+            responsibility so the denial audit event is emitted at the
+            Lambda boundary with the right reason code.
+    """
+    engine = get_engine()
+    # Nova ABAC policies use the `.policy` extension (REQ-339), but
+    # KyvernoJsonEngine only loads `.json`/`.yaml`/`.yml`. Materialize a
+    # temp dir with `.policy` → `.json` twins so the engine picks them
+    # up. The temp dir is removed in the `finally` block.
+    pol_dir, _ = _materialize_policy_dir(_POLICY_DIR)
+    try:
+        pcrs = engine.evaluate(payload, pol_dir, _CONTRACT_ID)
+    finally:
+        shutil.rmtree(pol_dir, ignore_errors=True)
+    allowed = not any(
+        p.get("result") == "fail" and str(p.get("severity", "")).lower() == "critical"
+        for p in pcrs
+    )
+    return allowed, pcrs, _policy_sha()
+
+
+if __name__ == "__main__":  # pragma: no cover - CLI inspection helper
+    import json
+    import sys
+
+    if len(sys.argv) > 1:
+        with open(sys.argv[1]) as fh:
+            pl = json.load(fh)
+    else:
+        pl = json.loads(sys.stdin.read())
+    allowed, pcrs, sha = evaluate_token_vend_policy(pl)
+    print(json.dumps({"allowed": allowed, "policy_sha": sha, "pcrs": pcrs}, indent=2))

platform/abac/token-vend.policy

@@ -0,0 +1,51 @@
+{
+  "apiVersion": "json.kyverno.io/v1alpha1",
+  "kind": "ValidatingPolicy",
+  "metadata": {
+    "name": "token-vend",
+    "annotations": {
+      "nova.cloudinit.dev/severity": "critical",
+      "title.policy.kyverno.io": "Token vend ABAC authorization (REQ-339, C-5.1, C-6.1)"
+    }
+  },
+  "spec": {
+    "rules": [
+      {
+        "name": "owner-matches",
+        "assert": {
+          "all": [
+            {
+              "check": {
+                "(target_resource.owner == subject.owner)": true
+              }
+            }
+          ]
+        }
+      },
+      {
+        "name": "role-env-match",
+        "assert": {
+          "all": [
+            {
+              "check": {
+                "((subject.role == 'developer' && environment == 'dev') || (subject.role == 'sre' && contains(['qa','prod','dr'], environment)))": true
+              }
+            }
+          ]
+        }
+      },
+      {
+        "name": "requested-claims-present",
+        "assert": {
+          "all": [
+            {
+              "check": {
+                "(length(requested_claims) > `0`)": true
+              }
+            }
+          ]
+        }
+      }
+    ]
+  }
+}

tests/test_abac_policy.py

@@ -0,0 +1,103 @@
+"""ABAC policy tests for the token-vend Lambda (REQ-339, C-5.1, C-6.1).
+
+Uses the **real** ``kj`` binary at ``/usr/local/bin/kj`` — these are
+real policy-evaluation tests, not mocked. Skipped when ``kj`` is absent
+(graceful, not failed — the binary is a build-host dep).
+"""
+
+from __future__ import annotations
+
+import shutil
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from core.abac_evaluator import evaluate_token_vend_policy
+
+KJ_AVAILABLE = shutil.which("kj") is not None
+skip_no_kj = pytest.mark.skipif(
+    not KJ_AVAILABLE, reason="`kj` binary not on PATH (D-227 build-host dep)"
+)
+
+
+def _payload(role, env, owner="t1", res_owner="t1"):
+    return {
+        "subject": {"id": "u1", "role": role, "owner": owner},
+        "requested_claims": ["sub", "roles"],
+        "target_resource": {
+            "type": "contract",
+            "id": "c1",
+            "owner": res_owner,
+            "environment": env,
+        },
+        "environment": env,
+        "pat_jti": "p1",
+        "policy_version": "test",
+    }
+
+
+@skip_no_kj
+def test_developer_dev_allowed():
+    allowed, pcrs, sha = evaluate_token_vend_policy(_payload("developer", "dev"))
+    assert allowed is True, [p for p in pcrs if p["result"] == "fail"]
+    assert sha  # non-empty SHA
+
+
+@skip_no_kj
+def test_sre_prod_allowed():
+    allowed, pcrs, sha = evaluate_token_vend_policy(_payload("sre", "prod"))
+    assert allowed is True, [p for p in pcrs if p["result"] == "fail"]
+
+
+@skip_no_kj
+def test_sre_qa_allowed():
+    allowed, _, _ = evaluate_token_vend_policy(_payload("sre", "qa"))
+    assert allowed is True
+
+
+@skip_no_kj
+def test_sre_dr_allowed():
+    allowed, _, _ = evaluate_token_vend_policy(_payload("sre", "dr"))
+    assert allowed is True
+
+
+@skip_no_kj
+def test_developer_prod_denied():
+    allowed, pcrs, _ = evaluate_token_vend_policy(_payload("developer", "prod"))
+    assert allowed is False
+    fails = [p for p in pcrs if p["result"] == "fail" and p["severity"] == "critical"]
+    assert fails, "expected at least one critical fail PCR"
+
+
+@skip_no_kj
+def test_wrong_owner_denied():
+    allowed, pcrs, _ = evaluate_token_vend_policy(
+        _payload("developer", "dev", owner="t1", res_owner="t2")
+    )
+    assert allowed is False
+    fails = [p for p in pcrs if p["result"] == "fail"]
+    assert fails
+
+
+@skip_no_kj
+def test_developer_qa_denied():
+    allowed, _, _ = evaluate_token_vend_policy(_payload("developer", "qa"))
+    assert allowed is False
+
+
+@skip_no_kj
+def test_empty_requested_claims_denied():
+    pl = _payload("developer", "dev")
+    pl["requested_claims"] = []
+    allowed, pcrs, _ = evaluate_token_vend_policy(pl)
+    assert allowed is False
+
+
+@skip_no_kj
+def test_policy_sha_is_string():
+    _, _, sha = evaluate_token_vend_policy(_payload("developer", "dev"))
+    assert isinstance(sha, str)
+    assert len(sha) > 0