6.6 Compulsory Security: Privileged Action Gates & Sandbox Enforcement

Canonical Curriculum Reference: docs/plan/06_phase6_stateful_graphs_hitl.md
Security Standards Reference: OWASP Top 10 for LLM & Agentic Applications (2025/2026), ASI01: Excessive Agency & ASI02: Insecure Output Handling.

---

🔐 1. The Threat of "Rogue Agents" & Excessive Agency

When autonomous agents are equipped with tool-calling capabilities and stateful graph orchestration, the primary security threat shifts from simple prompt injection to Excessive Agency (OWASP ASI01):

• An agent tricked by an indirect prompt injection in an incoming email invokes delete_customer_records(force=True).
• A financial analysis agent autonomous loops into issuing unauthorized wire transfers.
• A coding agent executes shell commands outside its designated workspace, leaking ~/.ssh/id_rsa or tampering with system binaries.

To neutralize these catastrophic threats, the architecture must enforce two non-bypassable security invariants:

1. Privileged Action Gates: Native Human-in-the-Loop authorization with immutable audit logging.
2. Deterministic Sandbox Enforcement: OS-level filesystem and process boundary containment.

---

🛡️ 2. Privileged Action Gates: The 3-Tier Security Architecture

Every sensitive tool (financial payouts, data deletion, credential modification, production deployments) must be protected by a 3-Tier Gate:

[ Agent Emits Action Proposal: `refund_customer(amount=$5,000)` ]
                                |
                                v
+---------------------------------------------------------------+
|                    PRIVILEGED ACTION GATE                     |
|                                                               |
| 1. TIER 1: RBAC / ABAC Policy Evaluation                      |
|    - Does the initiating user role have refund authority?     |
|    - Is $5,000 within the authorized transaction threshold?   |
|                                                               |
| 2. TIER 2: Contextual Validation                              |
|    - Is there an active ticket / incident tied to this task?   |
|    - Does the agent's internal monologue substantiate action? |
|                                                               |
| 3. TIER 3: Human Authorization & Cryptographic Signing        |
|    - Halt graph via interrupt()                               |
|    - Require HMAC/RSA signed human supervisor token           |
+---------------------------------------------------------------+
                                |
                   [ All 3 Tiers Verified ]
                                v
[ Execute Sensitive Tool & Commit to Immutable Audit Trail ]

The Audit Trail of Reasoning

For regulatory compliance (SOC2, HIPAA, GDPR), recording what was executed is insufficient; you must prove why the agent chose to execute it. The gate logs:

• Exact User Goal ($G_0$).
• Intermediate Thought Trajectory ($\tau$).
• Full Tool Payload ($A_t$).
• Human Reviewer ID and Digital Signature.
• Execution Timestamp and Cryptographic Hash.

---

📦 3. Sandbox Enforcement: Hardened Tool Boundaries

Agents must never run in an unconstrained shell on the host server. The harness enforces Path Traversal Sandboxing:

import os
from pathlib import Path

class FilesystemSandbox:
    """Enforces strict path containment within designated workspace root."""

    def __init__(self, sandbox_root: Path) -> None:
        self._root: Path = sandbox_root.resolve()

    def resolve_safe_path(self, requested_relative_path: str) -> Path:
        """Resolves path and raises PermissionError if attempting path traversal."""
        target = (self._root / requested_relative_path).resolve()
        
        # Security invariant: target must be within sandbox_root
        if not target.is_relative_to(self._root):
            raise PermissionError(
                f"[SECURITY ALERT]: Path traversal attempt blocked! "
                f"Requested '{requested_relative_path}' escapes sandbox '{self._root}'."
            )
        return target

---

💻 4. Production Implementation: Privileged Action Gate

import hashlib
import hmac
import time
from typing import Any, Final
from pydantic import BaseModel, ConfigDict, Field


class PrivilegedActionProposal(BaseModel):
    model_config = ConfigDict(frozen=True, extra="forbid")
    action_name: str
    target_resource_id: str
    parameters: dict[str, Any]
    agent_reasoning_summary: str
    initiating_user_id: str


class AuditTrailEntry(BaseModel):
    model_config = ConfigDict(frozen=True, extra="forbid")
    entry_id: str
    timestamp_utc: float
    action_proposal: PrivilegedActionProposal
    supervisor_approver_id: str
    cryptographic_signature: str


class PrivilegedActionGate:
    """Enforces mandatory human authorization and cryptographic audit logging."""

    def __init__(self, hmac_secret_key: bytes) -> None:
        self._secret: Final[bytes] = hmac_secret_key
        self._audit_log: list[AuditTrailEntry] = []

    def verify_and_authorize(
        self,
        proposal: PrivilegedActionProposal,
        approver_id: str,
        auth_token: str
    ) -> AuditTrailEntry:
        """Validates supervisor credentials and creates tamper-evident audit record."""
        # 1. Compute expected HMAC signature over action payload + approver
        payload_bytes = f"{proposal.action_name}:{proposal.target_resource_id}:{approver_id}".encode("utf-8")
        expected_sig = hmac.new(self._secret, payload_bytes, hashlib.sha256).hexdigest()

        if not hmac.compare_digest(expected_sig, auth_token):
            raise PermissionError(
                f"[SECURITY GATE]: Invalid authorization signature for action '{proposal.action_name}'."
            )

        # 2. Construct immutable audit entry
        entry = AuditTrailEntry(
            entry_id=hashlib.sha256(f"{time.time()}:{expected_sig}".encode()).hexdigest()[:16],
            timestamp_utc=time.time(),
            action_proposal=proposal,
            supervisor_approver_id=approver_id,
            cryptographic_signature=expected_sig
        )
        
        self._audit_log.append(entry)
        return entry

    @property
    def audit_trail(self) -> list[AuditTrailEntry]:
        return list(self._audit_log)

---

🧪 5. Automated Verification with Pytest

import hmac
import hashlib
import pytest
from pathlib import Path
from tempfile import TemporaryDirectory

SECRET_KEY = b"enterprise-audit-secret-2026"

def test_filesystem_sandbox_blocks_path_traversal():
    with TemporaryDirectory() as tmp_dir:
        sandbox = FilesystemSandbox(sandbox_root=Path(tmp_dir))
        
        # Valid path inside sandbox
        safe_path = sandbox.resolve_safe_path("src/app.py")
        assert safe_path.is_relative_to(Path(tmp_dir))
        
        # Malicious traversal attempts must be blocked
        with pytest.raises(PermissionError, match="Path traversal attempt blocked"):
            sandbox.resolve_safe_path("../../../etc/passwd")
            
        with pytest.raises(PermissionError, match="Path traversal attempt blocked"):
            sandbox.resolve_safe_path("/Users/root/.ssh/id_rsa")


def test_privileged_action_gate_authorization_and_audit():
    gate = PrivilegedActionGate(hmac_secret_key=SECRET_KEY)
    
    proposal = PrivilegedActionProposal(
        action_name="DELETE_CUSTOMER_DATA",
        target_resource_id="cust-882",
        parameters={"purge_backups": True},
        agent_reasoning_summary="GDPR Right to be Forgotten request verified.",
        initiating_user_id="agent-worker-01"
    )
    
    approver = "security-director-huy"
    
    # Forge invalid token -> Rejected
    with pytest.raises(PermissionError, match="Invalid authorization signature"):
        gate.verify_and_authorize(proposal, approver, auth_token="forged-signature-token")
        
    # Generate valid signature
    payload = f"{proposal.action_name}:{proposal.target_resource_id}:{approver}".encode()
    valid_sig = hmac.new(SECRET_KEY, payload, hashlib.sha256).hexdigest()
    
    entry = gate.verify_and_authorize(proposal, approver, auth_token=valid_sig)
    assert entry.supervisor_approver_id == approver
    assert len(gate.audit_trail) == 1
    assert entry.action_proposal.action_name == "DELETE_CUSTOMER_DATA"