Securing Autonomous Agents: Stopping the Confused Deputy Attack

import re
from typing import Dict, Any, List, Optional
from dataclasses import dataclass

@dataclass
class AgentMessage:
    sender_id: str
    content: str
    integrity: float          # Range [0.0, 1.0]; 1.0 is internal trusted instruction
    source_provenance: str   # e.g., "internal_supervisor", "web_scrape_untrusted"

class TaintBoundaryGuard:
    # High-risk prompt override patterns
    INJECTION_HEURISTICS = [
        re.compile(r"ignore\s+(previous|all)\s+instructions", re.IGNORECASE),
        re.compile(r"system\s+(override|note|prompt)", re.IGNORECASE),
        re.compile(r"drop\s+table|delete\s+from|alter\s+user", re.IGNORECASE),
        re.compile(r"exec\s*\(\s*['\"]", re.IGNORECASE),
        re.compile(r"curl\s+|webhook\s+|exfil", re.IGNORECASE),
    ]

    def __init__(self, strict_threshold: float = 0.75):
        self.strict_threshold = strict_threshold
        self.taint_audit_log: List[Dict[str, Any]] = []

    def evaluate_provenance(self, text: str, source: str) -> float:
        # Check for direct injection markers
        for pattern in self.INJECTION_HEURISTICS:
            if pattern.search(text):
                return 0.05  # Severe taint detected
        
        # Default untrusted penalty for raw external documents/scrapes
        if "external" in source or "document" in source:
            return 0.25
        return 1.0   # Fully verified internal agent state

    def propagate_taint(self, inputs: List[AgentMessage], output_content: str) -> AgentMessage:
        # Bell-LaPadula contraction: output integrity is bounded by minimum input integrity
        min_input_integrity = min([msg.integrity for msg in inputs]) if inputs else 1.0
        
        # Check if newly generated content introduced untrusted strings
        output_direct_integrity = self.evaluate_provenance(output_content, "agent_generation")
        final_integrity = min(min_input_integrity, output_direct_integrity)

        return AgentMessage(
            sender_id="synthesizer_agent",
            content=output_content,
            integrity=final_integrity,
            source_provenance="propagated_cluster"
        )

    def authorize_tool_invocation(self, message: AgentMessage, tool_tier: int) -> Dict[str, Any]:
        # Tier 1 (Read-only) always authorized
        if tool_tier == 1:
            return {"authorized": True, "reason": "Tier 1 Read-only autonomy permitted."}
        
        # Tier 2 (Sandboxed Code) requires clean taint score
        if tool_tier == 2:
            if message.integrity < self.strict_threshold:
                return {"authorized": False, "reason": "Blocked: Tainted inputs cannot invoke Tier 2 sandboxes."}
            return {"authorized": True, "reason": "Authorized under gVisor isolated microVM."}

        # Tier 3 (State Mutation) requires human gating (covered in Part 4)
        return {
            "authorized": False,
            "reason": "Halted: Tier 3 mutations require cryptographic Dual-Key HITL verification."
        }