acdl/core/abac_evaluator.py

"""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))