orchestrata_ultimate.orchestrator.intent_validator

Reference › orchestrata_ultimate › orchestrator › intent_validator

The validation stage of the intent pipeline: it decides whether an Intent may proceed. intent_broker calls it first, and intent_broker_factory builds it with a shared Prolog bridge and a per-minute rate limit.

IntentValidator.validate returns a ValidationResult after running four checks in order, stopping at the first that denies:

• _validate_structure requires an id, a requester, and a real action and target;
• _check_security denies writes, deletes and modifies under any prefix in PROTECTED_PATHS, and denies command execution whose path and arguments contain a pattern from DESTRUCTIVE_PATTERNS;
• _check_rate_limit keeps a 60-second window of request times per requester under a lock and denies when the count reaches the limit;
• _evaluate_prolog runs the governance rules in a worker thread: it asks permission_granted and, when that fails, permission_denied to obtain the reason, recording the query and result on the ValidationResult.

The stage is fail-closed. An exception anywhere, including a Prolog error, produces a DENIED outcome with the error in the reason, never an allow.

At a glance

Class diagram

Dependency diagram

API

Classes

IntentValidator

Decorators: traces(spec='003', story='US-003', reqs=['REQ-003-US-010-003', 'REQ-003-US-010-014', 'REQ-003-US-003-001', 'REQ-003-US-… · Line 73

Validates intents using Prolog governance rules.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/orchestrator/intent_validator.py

"""Prolog-based intent validation.

Validates intents against governance rules stored in Prolog.
Uses PrologBridge for deterministic policy evaluation.
Fail-closed: any error results in DENIED.

Spec: 003-intent / US-003
Author: Mohamed
"""

import asyncio
import logging
import threading
import time

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.infra.prolog_bridge import PrologBridge
from orchestrata_ultimate.orchestrator.intent_types import (
    Intent,
    IntentAction,
    IntentTarget,
    ValidationOutcome,
    ValidationResult,
)

logger = logging.getLogger(__name__)

# Protected system paths that always deny writes/deletes
PROTECTED_PATHS = frozenset(
    {
        "/etc",
        "/usr",
        "/var",
        "/root",
        "/boot",
        "/sys",
        "/proc",
        "/dev",
        "/usr/bin",
        "/usr/sbin",
        "/var/log",
    }
)

# Destructive command patterns
DESTRUCTIVE_PATTERNS = frozenset(
    {
        "rm -rf /",
        "dd if=/dev/zero",
        "dd of=/dev/",
        "mkfs",
        "shred",
        "format ",
        "fdisk",
        "parted",
    }
)


@traces(
    spec="003",
    story="US-003",
    reqs=[
        "REQ-003-US-010-003",
        "REQ-003-US-010-014",
        "REQ-003-US-003-001",
        "REQ-003-US-003-005",
        "REQ-003-US-003-008",
        "REQ-003-US-003-011",
        "REQ-003-US-003-012",
    ],
)
class IntentValidator:
    """Validates intents using Prolog governance rules.

    The validation pipeline:
    1. Structural validation (well-formed intent)
    2. Security checks (protected paths, destructive commands)
    3. Prolog policy evaluation (governance rules)
    4. Rate limit check

    Fail-closed: any exception during validation results in DENIED.
    """

    def __init__(
        self,
        prolog_bridge: PrologBridge,
        rate_limit_per_minute: int = 60,
    ):
        self._prolog = prolog_bridge
        self._rate_limit = rate_limit_per_minute
        self._request_counts: dict[str, list[float]] = {}
        # Guards _request_counts against concurrent mutation. The critical
        # section contains no awaits, so a plain threading.Lock is correct for
        # both interleaved coroutines and real threads sharing this instance.
        self._rate_limit_lock = threading.Lock()

    @traces(
        spec="003",
        story="US-003",
        reqs=[
            "REQ-003-US-010-040",
            "REQ-003-US-003-001",
            "REQ-003-US-003-008",
            "REQ-003-US-003-020",
            "REQ-003-US-003-023",
            "REQ-003-US-003-026",
            "REQ-003-US-003-035",
            "REQ-003-US-006-005",
        ],
    )
    async def validate(self, intent: Intent) -> ValidationResult:
        """Validate an intent against all governance rules.

        Returns ValidationResult with outcome and details.
        Fail-closed: returns DENIED on any error.
        """
        start = time.monotonic()
        try:
            structural = self._validate_structure(intent)
            if structural is not None:
                return structural
            security = self._check_security(intent)
            if security is not None:
                return security
            rate = self._check_rate_limit(intent.requester)
            if rate is not None:
                return rate
            result = await self._evaluate_prolog(intent)
            elapsed_ms = (time.monotonic() - start) * 1000
            logger.info(
                f"Intent {intent.id} validation: {result.outcome.value} "
                f"in {elapsed_ms:.1f}ms"
            )
            return result
        except Exception as e:
            logger.error(f"Intent {intent.id} validation error: {e}")
            return ValidationResult(
                outcome=ValidationOutcome.DENIED,
                intent_id=intent.id,
                reason=f"Validation error: {e}",
            )

    @traces(
        spec="003",
        story="US-003",
        reqs=[
            "REQ-003-US-003-004",
            "REQ-003-US-003-006",
            "REQ-003-US-003-013",
            "REQ-003-US-003-014",
            "REQ-003-US-003-015",
            "REQ-003-US-003-016",
            "REQ-003-US-003-018",
            "REQ-003-US-003-029",
            "REQ-003-US-003-031",
        ],
    )
    def _validate_structure(self, intent: Intent) -> ValidationResult | None:
        """Validate intent structure is well-formed."""
        violations = []
        if not intent.id:
            violations.append("Missing intent ID")
        if not intent.requester:
            violations.append("Missing requester")
        if not isinstance(intent.action, IntentAction):
            violations.append(f"Invalid action: {intent.action}")
        if not isinstance(intent.target, IntentTarget):
            violations.append(f"Invalid target: {intent.target}")
        if violations:
            return ValidationResult(
                outcome=ValidationOutcome.DENIED,
                intent_id=intent.id,
                reason="Structural validation failed",
                violations=violations,
            )
        return None

    @traces(
        spec="003",
        story="US-003",
        reqs=[
            "REQ-003-US-003-002",
            "REQ-003-US-003-003",
            "REQ-003-US-003-027",
            "REQ-003-US-003-028",
        ],
    )
    def _check_security(self, intent: Intent) -> ValidationResult | None:
        """Check security constraints (protected paths, destructive commands)."""
        if intent.action in (
            IntentAction.WRITE,
            IntentAction.DELETE,
            IntentAction.MODIFY,
        ):
            path = intent.resource_path
            for protected in PROTECTED_PATHS:
                if path.startswith(protected):
                    return ValidationResult(
                        outcome=ValidationOutcome.DENIED,
                        intent_id=intent.id,
                        reason=f"Protected path: {protected}",
                        violations=[
                            f"Cannot {intent.action.value} to protected path {protected}"
                        ],
                    )
        if (
            intent.action == IntentAction.EXECUTE
            and intent.target == IntentTarget.COMMAND
        ):
            command = (
                intent.resource_path + " " + str(intent.parameters.get("args", ""))
            )
            for pattern in DESTRUCTIVE_PATTERNS:
                if pattern in command.lower():
                    return ValidationResult(
                        outcome=ValidationOutcome.DENIED,
                        intent_id=intent.id,
                        reason=f"Destructive command: {pattern}",
                        violations=[f"Blocked destructive pattern: {pattern}"],
                    )
        return None

    @traces(
        spec="003",
        story="US-003",
        reqs=["REQ-003-US-003-005", "REQ-003-US-003-017", "REQ-003-US-003-030"],
    )
    def _check_rate_limit(self, requester: str) -> ValidationResult | None:
        """Check per-requester rate limit."""
        now = time.monotonic()
        window = 60.0
        with self._rate_limit_lock:
            if requester not in self._request_counts:
                self._request_counts[requester] = []
            self._request_counts[requester] = [
                t for t in self._request_counts[requester] if now - t < window
            ]
            if len(self._request_counts[requester]) >= self._rate_limit:
                return ValidationResult(
                    outcome=ValidationOutcome.DENIED,
                    intent_id="",
                    reason=f"Rate limit exceeded: {self._rate_limit}/min for {requester}",
                )
            self._request_counts[requester].append(now)
        return None

    @traces(
        spec="003",
        story="US-003",
        reqs=[
            "REQ-003-US-010-003",
            "REQ-003-US-010-014",
            "REQ-003-US-003-002",
            "REQ-003-US-003-003",
            "REQ-003-US-003-007",
            "REQ-003-US-003-012",
            "REQ-003-US-003-032",
        ],
    )
    async def _evaluate_prolog(self, intent: Intent) -> ValidationResult:
        """Evaluate intent against Prolog governance rules."""
        prolog_query = (
            f"permission_granted("
            f"intent('{intent.action.value}', '{intent.target.value}', "
            f"'{intent.resource_path}'), Reason)"
        )
        try:
            result = await asyncio.to_thread(
                self._prolog.query_bool,
                f"permission_granted("
                f"intent('{intent.action.value}', '{intent.target.value}', "
                f"'{intent.resource_path}'), _)",
            )
            if result:
                return ValidationResult(
                    outcome=ValidationOutcome.ALLOWED,
                    intent_id=intent.id,
                    reason="Prolog policy: allowed",
                    prolog_query=prolog_query,
                    prolog_result="true",
                )
            denial = await asyncio.to_thread(
                self._prolog.query_one,
                f"permission_denied("
                f"intent('{intent.action.value}', '{intent.target.value}', "
                f"'{intent.resource_path}'), Reason)",
            )
            reason = "Prolog policy: denied"
            if denial and "Reason" in denial:
                reason = f"Prolog policy: {denial['Reason']}"
            return ValidationResult(
                outcome=ValidationOutcome.DENIED,
                intent_id=intent.id,
                reason=reason,
                prolog_query=prolog_query,
                prolog_result="false",
            )
        except Exception as e:
            logger.error(f"Prolog evaluation error for intent {intent.id}: {e}")
            return ValidationResult(
                outcome=ValidationOutcome.DENIED,
                intent_id=intent.id,
                reason=f"Prolog error (fail-closed): {e}",
                prolog_query=prolog_query,
            )