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