orchestrata_ultimate.orchestrator.intent_security

Reference › orchestrata_ultimate › orchestrator › intent_security

Security checks that an intent can be put through before it is accepted: identity, permission, hostile content and request rate. IntentSecurity is an optional collaborator of intent_broker, which is its only importer. It uses the shared PrologBridge for permission rules and RedisStateManager for counters and the blacklist, and works on the Intent record.

The constructor takes an optional Prolog bridge, an optional Redis manager, a per-minute rate limit (default 60) and an auth_enabled flag, which set_auth_enabled can change. When authentication is disabled every check passes with the reason auth_disabled. authenticate returns an AuthResult carrying a requester id and checks that id against the blacklist; blacklist, unblacklist and is_blacklisted manage that list. authorize returns an AuthzResult: for the actions in DANGEROUS_ACTIONS it asks the Prolog rule permission_granted and denies on a negative answer or on any error; other actions pass. check_injection tests text against five pattern families (SQL, command, path traversal, XSS and LDAP injection) and returns a SecurityResult listing the threats and a sanitized copy, from which _sanitize strips backtick spans, $() substitutions, script tags and null bytes. check_rate_limit keeps a one-minute sliding window per requester as a Redis sorted set with a 120-second expiry, and denies when Redis is missing or errors. The validation stage in intent_validator applies its own protected-path and rate checks.

At a glance

Class diagram

Dependency diagram

API

Classes

AuthResult (dataclass)

Decorators: traces(spec='003', story='US-010', reqs=['REQ-003-US-010-002']), dataclass · Line 30

Authentication result.

AuthzResult (dataclass)

Decorators: traces(spec='003', story='US-010', reqs=['REQ-003-US-010-003']), dataclass · Line 40

Authorization result.

SecurityResult (dataclass)

Decorators: traces(spec='003', story='US-010', reqs=['REQ-003-US-010-007']), dataclass · Line 50

Injection check result.

IntentSecurity

Decorators: traces(spec='003', story='US-010', reqs=['REQ-003-US-010-002', 'REQ-003-US-010-003', 'REQ-003-US-010-007', 'REQ-003-US-… · Line 70

Security layer for intent processing.

Source

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

# mypy: ignore-errors
"""Intent security: authentication, authorization, injection prevention, rate limiting.

Provides security checks for the intent processing pipeline.
"""

from __future__ import annotations

import hashlib
import logging
import re
import time
from dataclasses import dataclass, field
from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.infra.redis_state import RedisStateManager

if TYPE_CHECKING:
    from orchestrata_ultimate.infra.prolog_bridge import PrologBridge
    from orchestrata_ultimate.orchestrator.intent_types import Intent

logger = logging.getLogger(__name__)

SECURITY_PREFIX = "orchestrata:security:intent"


@traces(spec="003", story="US-010", reqs=["REQ-003-US-010-002"])
@dataclass
class AuthResult:
    """Authentication result."""

    authenticated: bool
    requester_id: str = ""
    reason: str = ""


@traces(spec="003", story="US-010", reqs=["REQ-003-US-010-003"])
@dataclass
class AuthzResult:
    """Authorization result."""

    authorized: bool
    reason: str = ""
    required_permission: str = ""


@traces(spec="003", story="US-010", reqs=["REQ-003-US-010-007"])
@dataclass
class SecurityResult:
    """Injection check result."""

    safe: bool
    threats_detected: list[str] = field(default_factory=list)
    sanitized_text: str = ""


@traces(
    spec="003",
    story="US-010",
    reqs=[
        "REQ-003-US-010-002",
        "REQ-003-US-010-003",
        "REQ-003-US-010-007",
        "REQ-003-US-010-008",
        "REQ-003-US-010-009",
        "REQ-003-US-010-011",
    ],
)
class IntentSecurity:
    """Security layer for intent processing."""

    # SQL injection patterns
    SQL_PATTERNS = [
        re.compile(r"['\"]\s*;\s*DROP", re.IGNORECASE),
        re.compile(r"OR\s+1\s*=\s*1", re.IGNORECASE),
        re.compile(r"UNION\s+SELECT", re.IGNORECASE),
        re.compile(r"--\s*$", re.MULTILINE),
        re.compile(r"/\*.*\*/", re.DOTALL),
        re.compile(r";\s*DELETE\s+FROM", re.IGNORECASE),
        re.compile(r";\s*UPDATE\s+.*SET", re.IGNORECASE),
    ]

    # Command injection patterns
    CMD_PATTERNS = [
        re.compile(r";\s*rm\s", re.IGNORECASE),
        re.compile(r"&&\s*cat\s", re.IGNORECASE),
        re.compile(r"\|\s*nc\s", re.IGNORECASE),
        re.compile(r"`[^`]+`"),
        re.compile(r"\$\([^)]+\)"),
        re.compile(r";\s*wget\s", re.IGNORECASE),
        re.compile(r";\s*curl\s", re.IGNORECASE),
    ]

    # Path traversal patterns
    PATH_PATTERNS = [
        re.compile(r"\.\./"),
        re.compile(r"/etc/(?:passwd|shadow|sudoers)", re.IGNORECASE),
        re.compile(r"/proc/", re.IGNORECASE),
        re.compile(r"%00"),  # null byte
        re.compile(r"%2e%2e", re.IGNORECASE),  # encoded ../
    ]

    # XSS patterns
    XSS_PATTERNS = [
        re.compile(r"<script", re.IGNORECASE),
        re.compile(r"javascript:", re.IGNORECASE),
        re.compile(r"onerror\s*=", re.IGNORECASE),
        re.compile(r"onload\s*=", re.IGNORECASE),
    ]

    # LDAP injection patterns
    LDAP_PATTERNS = [
        re.compile(r"\*\)\("),
        re.compile(r"\)\(cn=\*"),
    ]

    # Dangerous actions requiring explicit authorization
    DANGEROUS_ACTIONS = frozenset({"delete", "execute", "modify"})

    def __init__(
        self,
        prolog_bridge: PrologBridge | None = None,
        redis: RedisStateManager | None = None,
        rate_limit: int = 60,
        auth_enabled: bool = True,
    ):
        self._prolog = prolog_bridge
        self._redis = redis
        self._rate_limit = rate_limit  # per minute
        self._auth_enabled = auth_enabled

    # ── Authentication ────────────────────────────────────────────

    @property
    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-025"])
    def auth_enabled(self) -> bool:
        """Return whether authentication/authorization is enabled."""
        return self._auth_enabled

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-025"])
    def set_auth_enabled(self, enabled: bool) -> None:
        """Toggle authentication and authorization at runtime."""
        self._auth_enabled = enabled

    @traces(
        spec="003",
        story="US-003",
        reqs=[
            "REQ-003-US-010-001",
            "REQ-003-US-010-002",
            "REQ-003-US-010-012",
            "REQ-003-US-010-013",
            "REQ-003-US-010-026",
            "REQ-003-US-010-027",
            "REQ-003-US-010-028",
            "REQ-003-US-010-029",
            "REQ-003-US-003-006",
        ],
    )
    def authenticate(self, token: str) -> AuthResult:
        """Authenticate a requester by token."""
        if not self._auth_enabled:
            return AuthResult(
                authenticated=True, requester_id="anonymous", reason="auth_disabled"
            )

        if not token or not isinstance(token, str):
            return AuthResult(authenticated=False, reason="empty_token")

        if len(token) < 8:
            return AuthResult(authenticated=False, reason="invalid_token_length")

        # Simple token validation — hash-based lookup
        token_hash = hashlib.sha256(token.encode()).hexdigest()[:16]
        requester_id = f"user-{token_hash[:8]}"

        # Check blacklist
        if self._redis and self.is_blacklisted(requester_id):
            return AuthResult(
                authenticated=False,
                requester_id=requester_id,
                reason="blacklisted",
            )

        return AuthResult(
            authenticated=True,
            requester_id=requester_id,
        )

    # ── Authorization ─────────────────────────────────────────────

    @traces(
        spec="003",
        story="US-010",
        reqs=[
            "REQ-003-US-010-003",
            "REQ-003-US-010-014",
            "REQ-003-US-010-030",
            "REQ-003-US-010-031",
            "REQ-003-US-010-039",
        ],
    )
    def authorize(self, requester: str, intent: Intent) -> AuthzResult:
        """Check if requester is authorized for the intent action."""
        if not self._auth_enabled:
            return AuthzResult(authorized=True, reason="auth_disabled")

        action = getattr(intent, "action", None)
        if action is None:
            return AuthzResult(authorized=False, reason="no_action")

        action_str = action.value if hasattr(action, "value") else str(action)

        # Check dangerous actions
        if action_str.lower() in self.DANGEROUS_ACTIONS:
            # Try Prolog authorization
            if self._prolog is not None:
                try:
                    target_str = ""
                    if hasattr(intent, "target"):
                        target_str = (
                            intent.target.value
                            if hasattr(intent.target, "value")
                            else str(intent.target)
                        )
                    query = f"permission_granted(intent('{action_str}', '{target_str}', _), _)"
                    result = self._prolog.query(query)
                    if not result.success:
                        return AuthzResult(
                            authorized=False,
                            reason=f"prolog_denied_{action_str}",
                            required_permission=f"{action_str}_permission",
                        )
                except Exception as e:
                    logger.warning("Prolog authorization failed: %s", e)
                    return AuthzResult(
                        authorized=False,
                        reason="authorization_error",
                        required_permission=f"{action_str}_permission",
                    )

        return AuthzResult(authorized=True)

    # ── Injection Detection ───────────────────────────────────────

    @traces(
        spec="003",
        story="US-010",
        reqs=[
            "REQ-003-US-010-007",
            "REQ-003-US-010-010",
            "REQ-003-US-010-019",
            "REQ-003-US-010-034",
            "REQ-003-US-010-039",
        ],
    )
    def check_injection(self, text: str) -> SecurityResult:
        """Check text for injection attacks."""
        if not text:
            return SecurityResult(safe=True, sanitized_text="")

        # (threat label, patterns) — evaluated in declaration order so the
        # resulting threat list is stable across categories.
        categories = (
            ("sql_injection", self.SQL_PATTERNS),
            ("command_injection", self.CMD_PATTERNS),
            ("path_traversal", self.PATH_PATTERNS),
            ("xss", self.XSS_PATTERNS),
            ("ldap_injection", self.LDAP_PATTERNS),
        )

        threats = [
            f"{label}:{pattern.pattern[:30]}"
            for label, patterns in categories
            for pattern in patterns
            if pattern.search(text)
        ]

        sanitized = self._sanitize(text) if threats else text

        return SecurityResult(
            safe=len(threats) == 0,
            threats_detected=threats,
            sanitized_text=sanitized,
        )

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-019"])
    def _sanitize(self, text: str) -> str:
        """Basic sanitization of dangerous characters."""
        # Remove backticks and $()
        sanitized = re.sub(r"`[^`]*`", "", text)
        sanitized = re.sub(r"\$\([^)]*\)", "", sanitized)
        # Remove script tags
        sanitized = re.sub(
            r"<script[^>]*>.*?</script>", "", sanitized, flags=re.IGNORECASE | re.DOTALL
        )
        # Remove null bytes
        sanitized = sanitized.replace("\x00", "")
        sanitized = sanitized.replace("%00", "")
        return sanitized.strip()

    # ── Rate Limiting ─────────────────────────────────────────────

    @traces(
        spec="003",
        story="US-003",
        reqs=[
            "REQ-003-US-010-008",
            "REQ-003-US-010-020",
            "REQ-003-US-010-035",
            "REQ-003-US-003-017",
        ],
    )
    def check_rate_limit(self, requester: str) -> bool:
        """Check if requester is within rate limit. Returns True if allowed."""
        if not self._auth_enabled:
            return True  # Allow all when auth is disabled

        if self._redis is None:
            return False  # Deny by default without Redis

        try:
            key = f"{SECURITY_PREFIX}:rate:{requester}"
            now = time.time()
            window_start = now - 60  # 1 minute window

            pipe = self._redis.client.pipeline()
            pipe.zremrangebyscore(key, 0, window_start)
            pipe.zadd(key, {str(now): now})
            pipe.zcard(key)
            pipe.expire(key, 120)
            results = pipe.execute()

            count = results[2]
            return count <= self._rate_limit
        except Exception as e:
            logger.warning("Rate limit check failed: %s", e)
            return False  # Deny by default on error

    # ── Blacklist Management ──────────────────────────────────────

    @traces(
        spec="003", story="US-010", reqs=["REQ-003-US-010-009", "REQ-003-US-010-021"]
    )
    def is_blacklisted(self, requester: str) -> bool:
        """Check if requester is blacklisted."""
        if self._redis is None:
            return False
        try:
            return bool(
                self._redis.client.sismember(f"{SECURITY_PREFIX}:blacklist", requester)
            )
        except Exception as exc:
            logger.warning(
                "Blacklist check failed for requester '%s': %s", requester, exc
            )
            return False

    @traces(
        spec="003", story="US-010", reqs=["REQ-003-US-010-009", "REQ-003-US-010-021"]
    )
    def blacklist(self, requester: str, reason: str = "") -> None:
        """Add requester to blacklist."""
        if self._redis is None:
            return
        self._redis.client.sadd(f"{SECURITY_PREFIX}:blacklist", requester)

    @traces(
        spec="003", story="US-010", reqs=["REQ-003-US-010-009", "REQ-003-US-010-021"]
    )
    def unblacklist(self, requester: str) -> None:
        """Remove requester from blacklist."""
        if self._redis is None:
            return
        self._redis.client.srem(f"{SECURITY_PREFIX}:blacklist", requester)