""" Bartholomew Trust Protocol (BTP) — Execution Dispatch-Seam Interceptor ====================================================================== Provides runtime execution-seam enforcement for AI agent tool calls, protecting against runtime indirection, dynamic string construction (shlex, getattr, eval/exec results), and multi-argument command arrays (subprocess). Unlike static AST pre-filtering (which only inspects raw code blocks before compilation), the Dispatch Seam intercepts fully evaluated, concrete runtime arguments (*args, **kwargs) at the exact point of system dispatch. """ import functools import hashlib import inspect import json import logging import shlex import time from typing import Any, Callable, Dict, List, Optional, Tuple, Union logger = logging.getLogger("btp.dispatch_seam") try: from src.polyglot_ast_validator import PolyglotASTValidator except ImportError: PolyglotASTValidator = None class DispatchViolationError(PermissionError): """ Exception raised when a tool invocation violates safety invariants at the runtime execution dispatch seam. """ def __init__( self, reason: str, rule_id: str = "BTP-DISPATCH-001", blocked_payload: str = "", function_name: str = "", latency_us: float = 0.0, receipt: Optional[Dict[str, Any]] = None, ): super().__init__( f"[BTP-DISPATCH-SEAM-VETO] Call to '{function_name}' blocked by {rule_id}: {reason}" ) self.reason = reason self.rule_id = rule_id self.blocked_payload = blocked_payload self.function_name = function_name self.latency_us = latency_us self.receipt = receipt or {} def to_diagnostics(self) -> Dict[str, Any]: return { "status": "BLOCKED_AT_DISPATCH_SEAM", "function": self.function_name, "rule_id": self.rule_id, "reason": self.reason, "blocked_payload": ( self.blocked_payload[:120] + "..." if len(self.blocked_payload) > 120 else self.blocked_payload ), "latency_us": round(self.latency_us, 2), "receipt": self.receipt, } def extract_evaluated_payloads(arg: Any) -> List[str]: """ Recursively extracts and canonicalizes executable string representations from concrete runtime arguments (*args, **kwargs). """ payloads: List[str] = [] if isinstance(arg, str): payloads.append(arg) elif isinstance(arg, (list, tuple)): # 1. Reconstructed unified string for command arrays str_tokens = [str(x) for x in arg if isinstance(x, (str, int, float, bool))] if str_tokens: try: unified_cmd = shlex.join(str_tokens) except Exception: unified_cmd = " ".join(str_tokens) payloads.append(unified_cmd) # 2. Inspect elements individually if complex or structured for item in arg: if isinstance(item, str): if any(c in item for c in (" ", "\n", ";", "|", "`", "$", "(", ")", "'", '"')): if item not in payloads: payloads.append(item) elif not isinstance(item, (int, float, bool)): payloads.extend(extract_evaluated_payloads(item)) elif isinstance(arg, dict): for k, v in arg.items(): if isinstance(k, str) and any(c in k for c in (" ", ";", "|", "&", "$")): payloads.append(k) payloads.extend(extract_evaluated_payloads(v)) elif hasattr(arg, "command"): payloads.extend(extract_evaluated_payloads(getattr(arg, "command"))) elif hasattr(arg, "query"): payloads.extend(extract_evaluated_payloads(getattr(arg, "query"))) return payloads class DispatchSeamInterceptor: """ High-performance, sub-15µs runtime execution seam interceptor. Binds directly to function dispatch, inspects evaluated runtime arguments, and produces Ed25519 non-repudiation receipts. """ def __init__( self, guard: Optional[Any] = None, agent_id: str = "agent-seam-guard", workspace_id: str = "bartholomew-core/production", strict: bool = True, sync_cloud: bool = False, ): if guard is not None: self.guard = guard else: try: from src import Guard self.guard = Guard(strict=strict) except Exception: self.guard = PolyglotASTValidator() if PolyglotASTValidator else None self.agent_id = agent_id self.workspace_id = workspace_id self.strict = strict self.sync_cloud = sync_cloud def inspect_dispatch( self, func_name: str, args: Tuple[Any, ...], kwargs: Dict[str, Any], ) -> Tuple[bool, str, str, float, Optional[Dict[str, Any]]]: """ Inspects evaluated arguments at the execution seam. Returns: (is_allowed, reason, rule_id, latency_us, receipt) """ t0 = time.perf_counter_ns() all_payloads: List[str] = [] for arg in args: all_payloads.extend(extract_evaluated_payloads(arg)) for k, v in kwargs.items(): all_payloads.extend(extract_evaluated_payloads(v)) # Evaluate extracted payloads against BTP AST invariants for payload in all_payloads: if not payload or not isinstance(payload, str): continue if self.guard and hasattr(self.guard, "evaluate_ast"): eval_res = self.guard.evaluate_ast(payload) if not eval_res.get("allowed", True): latency_us = (time.perf_counter_ns() - t0) / 1000.0 reason = eval_res.get("reason", "Destructive pattern detected at dispatch seam") violations = eval_res.get("violations", ["BTP-DISPATCH-001"]) rule_id = violations[0].split(":")[0] if violations else "BTP-DISPATCH-001" receipt = eval_res.get("receipt") return False, reason, rule_id, latency_us, receipt elif self.guard and hasattr(self.guard, "check"): res = self.guard.check(payload, agent_id=self.agent_id) if not res.get("allowed", True): latency_us = (time.perf_counter_ns() - t0) / 1000.0 return False, res.get("reason", "Denied"), "BTP-DISPATCH-001", latency_us, res.get("receipt") latency_us = (time.perf_counter_ns() - t0) / 1000.0 return True, "Approved", "BTP-DISPATCH-PASS", latency_us, None def protect( self, fn: Optional[Callable] = None, *, on_violation: Optional[Callable[[DispatchViolationError], Any]] = None, ): """ Decorator that wraps any tool or execution function at the dispatch seam. """ def decorator(target_func: Callable): func_name = getattr(target_func, "__name__", "anonymous_tool") @functools.wraps(target_func) def wrapper(*args, **kwargs): is_allowed, reason, rule_id, latency_us, receipt = self.inspect_dispatch( func_name=func_name, args=args, kwargs=kwargs, ) if not is_allowed: # Synthesize blocked payload representation for debugging/audit payload_summary = str(args)[:100] if args else str(kwargs)[:100] err = DispatchViolationError( reason=reason, rule_id=rule_id, blocked_payload=payload_summary, function_name=func_name, latency_us=latency_us, receipt=receipt, ) logger.warning("Execution Seam Veto: %s", err.to_diagnostics()) if on_violation: return on_violation(err) raise err return target_func(*args, **kwargs) return wrapper if fn is not None: return decorator(fn) return decorator _default_seam_interceptor = None def get_default_seam_interceptor() -> DispatchSeamInterceptor: global _default_seam_interceptor if _default_seam_interceptor is None: _default_seam_interceptor = DispatchSeamInterceptor() return _default_seam_interceptor def dispatch_seam_guard( fn: Optional[Callable] = None, *, on_violation: Optional[Callable[[DispatchViolationError], Any]] = None, ): """ Drop-in decorator for runtime execution seam protection. Usage: @dispatch_seam_guard def execute_shell(cmd_args: List[str]): subprocess.run(cmd_args) # Rejects: execute_shell(["rm", "-rf", "/var/lib"]) # -> DispatchViolationError (<15µs) """ return get_default_seam_interceptor().protect(fn, on_violation=on_violation)