diff --git a/core/abac_evaluator.py b/core/abac_evaluator.py new file mode 100644 index 0000000..a032ceb --- /dev/null +++ b/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": [, ...], # C-5.1 + "target_resource": {"type": ..., "id": ..., "owner": ..., "environment": ...}, + "environment": "dev" | "qa" | "prod" | "dr", + "pat_jti": "", + "policy_version": "" + } + +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)) \ No newline at end of file diff --git a/platform/abac/token-vend.policy b/platform/abac/token-vend.policy new file mode 100644 index 0000000..32da8de --- /dev/null +++ b/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 + } + } + ] + } + } + ] + } +} \ No newline at end of file diff --git a/tests/test_abac_policy.py b/tests/test_abac_policy.py new file mode 100644 index 0000000..d462cce --- /dev/null +++ b/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 \ No newline at end of file