acdl/core/policy_engine.py

"""Nova Policy Engine Registry (REQ-291, v1.25).

The swappable policy-engine abstraction. A Python Protocol (PEP 544)
defines the engine contract; a registry selects the active engine from
``config.json``'s ``policy.engine`` key. This is the **swap boundary**
(ARCHITECTURE.md §12.7) — the confidence signal and pipeline never
import an engine directly; they go through the registry. A future
``OpaEngine`` implements the same protocol without touching the
confidence signal, the PCR schema, or the pipeline.

The protocol is minimal (3 members) by design:

- ``name`` — the engine's registry key (matches ``config.json.policy.engine``).
- ``is_configured()`` — returns False when the engine's binary is absent
  (the registry's caller must skip gracefully, emitting SKIPPED PCRs).
- ``evaluate(payload, policy_dir, contract_id)`` — runs the engine's
  policies over ``payload`` and returns a ``list[dict]`` where each dict
  conforms to ``schemas/policy_check_result.schema.json``.

A ``NullEngine`` is the fallback when the ``policy`` key is absent from
``config.json`` (backward compatibility for tests that don't set the
key — it emits a single SKIPPED PCR so the confidence signal proceeds
with a neutral ``policy`` input).

Engine enum reuse (D-116): kyverno-json PCR records carry
``engine: "kyverno"`` (no new enum value). The ``engine`` field records
the policy-engine *family*, not the specific binary. The K8s Kyverno
adapter and the kyverno-json engine are distinguished by ``ruleId``
prefix (``KYVERNO_`` vs ``KJ_``).
"""

import json
import os
from pathlib import Path
from typing import Any, Callable, Protocol, Union, runtime_checkable

import datetime


def _iso8601_now() -> str:
    return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


Payload = Union[dict, list, str]


@runtime_checkable
class PolicyEngine(Protocol):
    """The swap boundary for policy engines.

    Implementations: ``KyvernoJsonEngine`` (adapters/kyverno-json/),
    ``NullEngine`` (this module), future ``OpaEngine``.
    """

    @property
    def name(self) -> str: ...

    def is_configured(self) -> bool: ...

    def evaluate(self, payload: Payload, policy_dir: Path,
                 contract_id: str) -> list[dict]: ...


def _skipped_pcr(rule_id: str, message: str, contract_id: str) -> dict:
    return {
        "contractId": contract_id,
        "evaluatedAt": _iso8601_now(),
        "engine": "kyverno",
        "ruleId": rule_id,
        "severity": "info",
        "result": "skipped",
        "message": message,
        "evidence": {},
        "resourceRef": "",
    }


class NullEngine:
    """Fallback when ``config.json.policy`` is absent.

    Emits a single SKIPPED PCR with ``ruleId: NULL_ENGINE_INACTIVE`` so
    the confidence signal's ``policy`` input is non-null (the per-input
    score for a single SKIPPED PCR is 1.0 — skipped counts as pass per
    ``core/confidence_signal.py:84-89``). This keeps existing tests
    passing when the ``policy`` key is not set.
    """

    name = "null"

    def is_configured(self) -> bool:
        return False

    def evaluate(self, payload: Payload, policy_dir: Path,
                 contract_id: str) -> list[dict]:
        return [_skipped_pcr(
            "NULL_ENGINE_INACTIVE",
            "NullEngine active — the `policy` key is absent from config.json. "
            "No policy engine is configured; the confidence signal proceeds with "
            "a neutral SKIPPED policy input.",
            contract_id,
        )]


_REGISTRY: dict[str, Callable[[], PolicyEngine]] = {}


def register(name: str, factory: Callable[[], PolicyEngine]) -> None:
    """Register an engine factory under ``name``.

    The factory is called lazily by ``get_engine()`` so an engine's
    binary dependency (e.g. ``kj``) is not required at import time.
    """
    _REGISTRY[name] = factory


def _load_config_policy() -> dict | None:
    """Read the ``policy`` object from ``.ciagent/config.json``.

    Returns ``None`` when the file is absent or the ``policy`` key is
    missing (the caller falls back to ``NullEngine``).
    """
    repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    cfg = os.path.join(repo_root, ".ciagent", "config.json")
    if not os.path.isfile(cfg):
        return None
    try:
        with open(cfg, "r", encoding="utf-8") as fh:
            data = json.load(fh)
    except (json.JSONDecodeError, OSError):
        return None
    return data.get("policy")


def get_engine() -> PolicyEngine:
    """Return the active ``PolicyEngine`` from ``config.json``.

    Reads ``config.json.policy.engine`` (default ``"kyverno-json"``).
    Falls back to ``NullEngine`` when the ``policy`` key is absent
    (backward compatibility). Raises ``KeyError`` for an unknown engine
    name (a typo in config — fail loud, not silent).
    """
    policy_cfg = _load_config_policy()
    if policy_cfg is None:
        return NullEngine()
    engine_name = policy_cfg.get("engine", "kyverno-json")
    factory = _REGISTRY.get(engine_name)
    if factory is None:
        raise KeyError(
            f"Unknown policy engine '{engine_name}' in config.json. "
            f"Registered engines: {sorted(_REGISTRY.keys()) or ['(none)']}. "
            f"Set policy.engine to a registered name or install the engine adapter."
        )
    return factory()


def get_policy_root() -> Path:
    """Return the configured policy root directory (or a default)."""
    policy_cfg = _load_config_policy()
    if policy_cfg is None:
        return Path("adapters/kyverno-json/policies")
    root = policy_cfg.get("policy_root", "adapters/kyverno-json/policies")
    repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    if os.path.isabs(root):
        return Path(root)
    return Path(repo_root) / root


def _register_builtin(name: str, factory: Callable[[], PolicyEngine]) -> None:
    register(name, factory)


def _autoload_kyverno_json() -> None:
    """Register the kyverno-json engine if its adapter is importable.

    The adapter directory uses a hyphen (``adapters/kyverno-json/``),
    so a plain ``import`` is not possible. Load the module by file path
    via ``importlib.util``. Lazy import so ``core/policy_engine.py``
    does not require ``adapters/kyverno-json/`` at import time (the
    adapter imports ``yaml``, which may be unavailable in minimal test
    envs).
    """
    try:
        import importlib.util
        repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        adapter_path = os.path.join(
            repo_root, "adapters", "kyverno-json", "kyverno_json_engine.py"
        )
        if not os.path.isfile(adapter_path):
            return
        spec = importlib.util.spec_from_file_location(
            "kyverno_json_engine", adapter_path
        )
        if spec is None or spec.loader is None:
            return
        mod = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(mod)
        engine_cls = getattr(mod, "KyvernoJsonEngine")
        _register_builtin("kyverno-json", engine_cls)
    except Exception:
        pass


_autoload_kyverno_json()


if __name__ == "__main__":
    eng = get_engine()
    print(json.dumps({
        "engine": eng.name,
        "is_configured": eng.is_configured(),
        "policy_root": str(get_policy_root()),
    }, indent=2))