orchestrata_ultimate.gateway.handlers._policy_gate

Reference › orchestrata_ultimate › gateway › handlers › _policy_gate

PolicyGate asks the governance policy engine whether a workflow RPC verb is allowed (SPEC-014 REQ-014-080). The workflow handler calls it before acting, and rpc supplies the engine factory. It takes an engine factory and builds the engine lazily on first use, once; if construction fails it logs a warning and stops retrying. check takes an action such as workflow.cancel and a context, evaluates it, and returns the decision. If no factory was ever registered it returns None, which is the permissive posture when no governance infrastructure is wired. When a factory exists, a denial or an evaluation error raises RPCError with code -32604. See gateway.

At a glance

Class diagram

Dependency diagram

API

Classes

PolicyGate

Line 27

Lazy-built PolicyEngine gate for a single workflow RPC verb.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/gateway/handlers/_policy_gate.py

"""Governance PAM++ gate for workflow control RPCs (SPEC-014 REQ-014-080).

Same lazy-factory shape as ``GovernanceHandler._resolve_engine``
(gateway/handlers/governance.py) so both consume the ONE
``_make_policy_engine`` factory the gateway already registers at boot
(server.py's ``_init_policy_engine``) -- REQ-014-080 is enforced with the
same real Prolog rule base ``governance.evaluate`` uses.

Fail-closed (D4): an explicit ``PolicyDecision`` DENY, or an evaluation
error, raises when a factory IS configured. Permissive when NO factory has
ever been wired -- matches today's dev-gateway posture (no governance infra
=> ``governance.evaluate`` also answers a permissive stub), so a bare
gateway is never bricked.
"""

import logging
from collections.abc import Callable
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.tracing.tracer import Tracer
from orchestrata_ultimate.gateway.exceptions import RPCError

logger = logging.getLogger(__name__)


class PolicyGate:
    """Lazy-built ``PolicyEngine`` gate for a single workflow RPC verb."""

    def __init__(
        self,
        engine_factory: Callable[[], Any] | None = None,
        tracer: Tracer | None = None,
    ) -> None:
        """Wire the gate to a deferred engine factory.

        Args:
            engine_factory: ``() -> PolicyEngine``, or None for the
                permissive dev posture (no governance infra wired).
            tracer: Injected debug tracer; defaults to the process Singleton.
        """
        self._engine_factory = engine_factory
        self._engine: Any | None = None
        self._tracer = tracer or Tracer.default()

    @traces(spec="014", story="US-014-005", reqs=["REQ-014-080"])
    async def check(self, action: str, context: dict[str, Any]) -> Any:
        """Evaluate *action* against the wired ``PolicyEngine``.

        Args:
            action: The workflow verb being authorized (e.g.
                ``"workflow.cancel"``).
            context: Evaluation context (actor, connection, params).

        Returns:
            The ``PolicyDecision`` when a factory is configured and the
            engine allows the action; ``None`` when no factory has ever
            been registered (permissive dev posture).

        Raises:
            RPCError: The engine denies the action, or evaluation itself
                raised -- only when a factory IS configured (D4).
        """
        with self._tracer.span("PolicyGate.check"):
            engine = self._resolve_engine()
            if engine is None:
                return None
            try:
                decision = await engine.evaluate(action, context)
            except Exception as exc:
                logger.warning("workflow policy evaluation failed: %s", exc)
                raise RPCError(
                    f"governance evaluation error for {action}: {exc}",
                    rpc_code=-32604,
                ) from exc
            if not decision.allowed:
                raise RPCError(
                    f"governance denied {action}: {decision.reason}",
                    rpc_code=-32604,
                )
            return decision

    @traces(spec="014", story="US-014-005", reqs=["REQ-014-080"])
    def _resolve_engine(self) -> Any | None:
        """Build the engine once from the factory (mirrors
        ``GovernanceHandler._resolve_engine``); ``None`` in stub mode."""
        if self._engine is None and self._engine_factory is not None:
            try:
                self._engine = self._engine_factory()
            except Exception as exc:
                logger.warning("workflow policy engine init failed: %s", exc)
                self._engine_factory = None  # don't retry on every call
        return self._engine