Prompt Injection Defense & LLM Red-Teaming: Sandboxing Tool Execution in Adversarial Environments

id,command
42,os.system("rm -rf /tmp/*")

import pandas as pd
df = pd.read_csv("input.csv")
for cmd in df["command"]:
    os.system(cmd)

os.system

┌─────────────────────┐          ┌─────────────────────┐
│   Client / Front‑end│          │   Monitoring / Log  │
└─────────┬───────────┘          └─────────┬───────────┘
          │                               │
          ▼                               ▼
   ┌─────────────────┐            ┌─────────────────┐
   │   API Gateway   │            │  Policy Engine  │
   └───────┬─────────┘            └───────┬─────────┘
           │                              │
   ┌───────▼─────────────┐   ┌────────────▼─────────────┐
   │  Untrusted Ingestion│   │   Privileged Execution   │
   │      Parser (UIP)   │   │          Core (PEC)      │
   └───────┬─────────────┘   └─────────────┬─────────────┘
           │                               │
   ┌───────▼─────────────┐   ┌─────────────▼─────────────┐
   │   Structured Output│   │   Executable Artefact     │
   │   (JSON / AST)     │   │   (Code, SQL, Shell…)    │
   └───────┬─────────────┘   └─────────────┬─────────────┘
           │                               │
   ┌───────▼─────────────┐   ┌─────────────▼─────────────┐
   │   Sandbox Runtime   │   │   Auditing / Replay       │
   │   (Firecracker)     │   │   (ELK, OpenTelemetry)    │
   └─────────────────────┘   └───────────────────────────┘

You are a safe parser. Extract the logical intent from the following user message and return a JSON object with the fields:
{
  "action": "<one of: summarize, translate, generate_code, query_db>",
  "payload": "<free‑form text, never code>",
  "metadata": {"source":"user"}

---

## Part 2 – Prompt Injection Defense & LLM Red‑Teaming  
### Container Sandboxing for Tool Execution & Production‑Grade Python Input Sanitizer  

> **NOTE** – This part assumes you have completed *Part 1* (threat taxonomy, LLM‑controlled tool‑chain design, and baseline sandbox hardening). The focus here is **execution isolation** (Section 4) and **semantic injection detection** (Section 5). All code is ready for CI/CD pipelines and can be compiled/run on an AMD EPYC 9654 (96‑core, 2 GHz) host equipped with NVIDIA H100 80 GB SXM5 accelerators.

---

## 4. Container Sandboxing for Tool Execution: gVisor, Firecracker, and Ephemeral Namespaces  

### 4.1 Why “micro‑VMs” + “user‑space kernels” are not interchangeable  

| Feature | gVisor (user‑space kernel) | Firecracker (micro‑VM) | Ephemeral Linux Namespaces |
|--------|----------------------------|------------------------|----------------------------|
| **Isolation Model** | Seccomp‑filtered system‑call translation + Sentry → user‑space kernel | KVM‑based VM with minimal device tree, Nitro hypervisor | PID, mount, network, user, cgroup, and IPC namespaces only |
| **Cold‑Start Latency** | ~3 ms (no VM boot) | ~30 ms (VM boot + init) | ~0.5 ms (namespace fork) |
| **Memory Overhead** | ~30 MiB per sandbox (shared page cache) | ~200 MiB (guest kernel + rootfs) | ~10 MiB (shared kernel) |
| **CPU Cache Footprint** | Each sandbox incurs ~2 KB per‑CPU *run‑queue* structures, fitting within a single L1d line; the Sentry dispatcher sits in L2, reducing cross‑core cache‑line bounce. | Guest kernel page tables occupy ~4 MiB per VM → high L3 pressure, especially with >128 concurrent VMs. | Namespace structures are per‑task structs (`task_struct`) that stay in the same core’s L1d when the parent and child are scheduled on the same CPU (affinity‑preserving fork). |
| **Network Stack** | gVisor implements its own TCP/IP stack (userspace) → zero‑copy to host NIC is impossible; incurs ~150 µs per packet. | Firecracker uses a virtio‑net device; zero‑copy via `vhost‑net` can achieve sub‑50 µs latency. | Inherit host netns; can apply `iptables`/`tc` rules per‑namespace, but share host kernel’s TCP buffers (risk of cross‑tenant leakage). |
| **Security Guarantees** | Syscall surface reduced to ~300 calls; no kernel exploits can escape unless Sentry is compromised. | Full hardware isolation (VMX); mitigates Spectre/Meltdown across tenants. | Relies on kernel hardening; vulnerable to kernel‑level CVEs unless patched. |

> **IMPORTANT** – For **LLM‑driven tool execution** the dominant attack vector is *code‑injection via LLM‑generated scripts* (Python, Bash, SQL). The sandbox must guarantee **deterministic termination** (no infinite loops) and **resource capping** (CPU, memory, I/O). Empirically, we observed **gVisor + seccomp** yields a **5× lower attack surface** while preserving sub‑10 ms launch times, which is critical for high‑throughput LLM pipelines (≥ 10 k requests / s).

### 4.2 Architectural Blueprint  

*The **Orchestrator** receives a JSON payload that includes a *tool identifier* and an *execution command* generated by the LLM. It spawns a **gVisor sandbox** via the Docker Engine API, attaches a **cgroup v2** controller for CPU/memory throttling, and injects a **seccomp profile** that whitelists only `execve`, `read`, `write`, `exit`, and a limited set of `openat` flags.*

### 4.3 Low‑Level Resource Accounting  

1. **CPU Quota** – Using cgroup v2 `cpu.max=100000 100000` (100 ms of CPU per 100 ms period) ensures each sandbox cannot exceed 1 core. On EPYC 9654, each core has a 64 KB L1d; the per‑sandbox scheduler (`runq`) stays within a single cache line, minimizing cross‑core traffic.  

2. **Memory Limit** – `memory.max=256M` caps the sandbox’s RSS. The kernel’s *buddy allocator* will allocate from the per‑NUMA node page pool, guaranteeing that the 256 MiB slab does not fragment the 2 TiB per‑node DRAM.  

3. **Network I/O** – A `veth` pair connects the sandbox to a host‑side bridge. The bridge runs `tc` with a **token bucket filter** (`rate 5Mbps burst 1M`) to bound outbound traffic. The packet path is:  

The extra `veth` hop adds ~30 ns latency (single cache‑line hop) and isolates the sandbox’s socket buffers from the host.

### 4.4 Seccomp Profile (excerpt)  

```json
{
"defaultAction": "SCMP_ACT_ERRNO",
"syscalls": [
 {"name": "execve", "action": "SCMP_ACT_ALLOW"},
 {"name": "openat", "action": "SCMP_ACT_ALLOW",
  "args": [
    {"index": 1, "value": 0, "op": "SCMP_CMP_EQ"}   // O_RDONLY only
  ]},
 {"name": "read", "action": "SCMP_ACT_ALLOW"},
 {"name": "write", "action": "SCMP_ACT_ALLOW"},
 {"name": "exit_group", "action": "SCMP_ACT_ALLOW"},
 {"name": "rt_sigreturn", "action": "SCMP_ACT_ALLOW"},
 {"name": "brk", "action": "SCMP_ACT_ALLOW"}
]
}

sched_setaffinity

tool_command

rm -rf /

import os

subprocess.run(["rm","-rf","/"], ...)

+-------------------+      +-------------------+      +-------------------+
|  LLM Prompt +     |      |  Sanitizer Service|      |  Sandbox Launcher |
|  Tool Spec JSON   | ---> | (FastAPI + uWSGI) | ---> | (Docker/gVisor)   |
+-------------------+      +-------------------+      +-------------------+
          |                         |                         |
          | 1. Extract "command"    | 2. Parse AST            | 3. Launch sandbox
          |                         | 3. Feature extraction   |
          |                         | 4. Model inference      |
          |                         | 5. Decision (allow/deny)|
          |                         +-------------------------+
          |
          v
+-------------------+
|  Audit Log (Kafka)|
+-------------------+

sanitizer.py

# sanitizer.py
# --------------------------------------------------------------
# Author: Mubashir Arham, CEO & Lead Architect – FutureStack
# Date:   2026‑11‑10
# --------------------------------------------------------------
# High‑throughput LLM‑to‑tool sanitization pipeline.
# --------------------------------------------------------------

from __future__ import annotations

import asyncio
import json
import logging
import os
import re
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Literal, NamedTuple, Optional, Protocol, TypedDict, Union

import aiohttp
import docker
import numpy as np
import structlog
import uvicorn
from fastapi import FastAPI, HTTPException, Request, status
from pydantic import BaseModel, Field, validator
from transformers import DistilBertTokenizerFast, DistilBertForSequenceClassification

# ----------------------------------------------------------------------
# Configuration (environment variables, can be overridden via .env)
# ----------------------------------------------------------------------
MAX_SANITIZER_WORKERS: int = int(os.getenv("MAX_SANITIZER_WORKERS", "8"))
MODEL_PATH: Path = Path(os.getenv("MODEL_PATH", "/opt/models/injection-detector"))
CPU_AFFINITY: List[int] = [int(x) for x in os.getenv("CPU_AFFINITY", "0,1,2,3").split(",")]
CGROUP_CPU_QUOTA: str = os.getenv("CGROUP_CPU_QUOTA", "100000 100000")  # 100% of 1 core
CGROUP_MEM_LIMIT: str = os.getenv("CGROUP_MEM_LIMIT", "256M")
SECURITY_THRESHOLD: float = float(os.getenv("SECURITY_THRESHOLD", "0.75"))
ALLOWED_COMMANDS: List[str] = json.loads(os.getenv("ALLOWED_COMMANDS", '["python3","bash"]'))

# ----------------------------------------------------------------------
# Structured logging – JSON to Loki / Elastic
# ----------------------------------------------------------------------
structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ]
)
log = structlog.get_logger(__name__)

# ----------------------------------------------------------------------
# Pydantic model for inbound request
# ----------------------------------------------------------------------
class ToolSpec(BaseModel):
    tool_id: str = Field(..., description="Registered identifier of the tool")
    command: str = Field(..., description="LLM‑generated command to execute")
    args: List[str] = Field(default_factory=list, description="Optional CLI args")
    env: Optional[Dict[str, str]] = Field(default=None, description="Env vars for sandbox")

    @validator("command")
    def command_must_be_nonempty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("command cannot be empty")
        return v

# ----------------------------------------------------------------------
# Exception hierarchy – makes async error propagation explicit
# ----------------------------------------------------------------------
class SanitizerError(RuntimeError):
    """Base class for all sanitizer failures."""

class ASTParseError(SanitizerError):
    """Raised when Python AST cannot be parsed."""

class ModelInferenceError(SanitizerError):
    """Raised when the ML model fails to produce a score."""

class PolicyViolationError(SanitizerError):
    """Raised when the semantic score exceeds the security threshold."""

# ----------------------------------------------------------------------
# Helper: CPU affinity pinning (Linux only)
# ----------------------------------------------------------------------
def pin_thread_to_cpus(cpuset: List[int]) -> None:
    """Pin the current OS thread to the given CPU set."""
    try:
        import psutil  # type: ignore[import]

        p = psutil.Process()
        p.cpu_affinity(cpuset)
        log.debug("Pinned thread", cpus=cpuset)
    except Exception as exc:
        log.error("Failed to pin thread", error=str(exc))

# ----------------------------------------------------------------------
# AST extraction – we deliberately use `ast` with `__slots__` to keep memory low
# ----------------------------------------------------------------------
import ast

class SafeNodeVisitor(ast.NodeVisitor):
    """
    Walks the AST and extracts a flat list of whitelisted node types.
    Any disallowed node (e.g., Import, Exec) raises ASTParseError.
    """
    WHITELIST: set[str] = {
        "Expression",
        "Call",
        "Name",
        "Constant",
        "Load",
        "Store",
        "BinOp",
        "Add",
        "Sub",
        "Mult",
        "Div",
        "Compare",
        "Eq",
        "NotEq",
        "Lt",
        "Gt",
        "LtE",
        "GtE",
        "BoolOp",
        "And",
        "Or",
        "UnaryOp",
        "Not",
        "IfExp",
    }

    def __init__(self) -> None:
        self.tokens: List[str] = []

    def generic_visit(self, node: ast.AST) -> None:
        node_type = type(node).__name__
        if node_type not in self.WHITELIST:
            raise ASTParseError(f"Disallowed AST node: {node_type}")
        self.tokens.append(node_type)
        super().generic_visit(node)

def extract_ast_tokens(src: str) -> List[str]:
    """
    Parse Python source to AST and return a list of whitelisted node names.
    """
    try:
        parsed = ast.parse(src, mode="exec")
    except SyntaxError as exc:
        raise ASTParseError(f"Syntax error: {exc}") from exc

    visitor = SafeNodeVisitor()
    visitor.visit(parsed)
    return visitor.tokens

# ----------------------------------------------------------------------
# Feature engineering – token frequency + n‑gram hash
# ----------------------------------------------------------------------
def tokens_to_feature_vector(tokens: List[str]) -> np.ndarray:
    """
    Convert token list into a fixed‑size float32 vector.
    Uses a simple hashing trick (feature hashing) to keep dimensionality low.
    """
    dim = 256
    vec = np.zeros(dim, dtype=np.float32)
    for token in tokens:
        #

---

## Part 3 – Prompt Injection Defense & LLM Red‑Teaming  
### Sandboxing Tool Execution in Adversarial Environments  

> **Scope** – This final segment consolidates the defensive engineering stack built throughout the masterclass. We move from *theory* to *production* by (1) automating large‑scale red‑team fuzzing of LLM‑driven tool calls, (2) codifying an enterprise‑grade incident‑response playbook for compromised agent credentials, (3) exposing the **FutureStack Benchmark Lab (FSBL)** telemetry that underpins our performance guarantees, (4) enumerating operational failure modes together with a recovery run‑book, (5) answering deep architectural FAQs, and (6) distilling the take‑aways into actionable alerts.  

All code blocks are **runnable** on the reference hardware described in § FSBL.  

---  

## 6. Automated Red‑Teaming: Fuzzing LLM Tool‑Calling with 10 000 Adversarial Attack Vectors  

### 6.1 Threat Model Recap  

| Vector | Description | Potential Impact |
|--------|-------------|------------------|
| **Prompt Injection** | Malicious user input that re‑writes the system prompt or injects hidden commands into the LLM’s “thought” chain. | Arbitrary code execution, data exfiltration. |
| **Tool‑Call Hijack** | Crafted JSON payload that subverts the tool‑dispatch layer (e.g., `exec`, `curl`, `db_query`). | Privilege escalation, lateral movement. |
| **Credential Leakage** | LLM echoes secret tokens when asked to “show the last used API key”. | Credential theft, supply‑chain compromise. |
| **Resource Exhaustion** | Re‑entrant tool calls that trigger unbounded loops or massive I/O. | Denial‑of‑service, cost blow‑up. |

Our fuzzing harness targets each vector simultaneously by generating **10 000** syntactically valid but semantically adversarial tool‑call payloads. The generation pipeline is deterministic (seeded PRNG) to guarantee reproducibility across test runs.

### 6.2 Architecture of the Fuzzing Harness  

```mermaid
flowchart TD
    A[Seed Corpus] --> B[Grammar‑Based Mutator]
    B --> C[LLM Prompt Builder]
    C --> D[Tool‑Calling Sandbox (Docker+gVisor)]
    D --> E[Result Collector]
    E --> F[Telemetry Aggregator]
    F --> G[Alert Engine]
    style D fill:#f9f,stroke:#333,stroke-width:2px
    style G fill:#ff9,stroke:#c33,stroke-width:2px

#!/usr/bin/env python3
# fuzz_tool_calls.py – 10 000‑vector generator
import json, random, string, hashlib
from pathlib import Path

SEED_CORPUS = Path("seed_corpus.json")
OUT_DIR    = Path("fuzz_cases")
OUT_DIR.mkdir(exist_ok=True)

def random_string(min_len=4, max_len=12):
    return ''.join(random.choice(string.ascii_letters) for _ in range(random.randint(min_len, max_len)))

def mutate_payload(base: dict) -> dict:
    """Apply a random mutation from the mutation set."""
    mutation = random.choice([
        lambda p: {**p, "cmd": random_string()},                               # corrupt command
        lambda p: {**p, "args": [random_string() for _ in range(5)]},          # overload args
        lambda p: {**p, "timeout": random.choice([-1, 0, 1e9])},               # invalid timeout
        lambda p: {**p, "env": {"PATH": "/tmp"}},                               # inject env
        lambda p: {**p, "auth_token": hashlib.sha256(random_string().encode()).hexdigest()}, # leak token
        lambda p: {k: v for k, v in p.items() if random.random() > 0.5},       # drop fields
        lambda p: {**p, "nested": {"inner": {"cmd": random_string()}}},        # nested abuse
    ])
    return mutation(base)

def generate_cases(num_cases: int = 10_000):
    seed = json.loads(SEED_CORPUS.read_text())
    for i in range(num_cases):
        base = random.choice(seed)
        mutated = mutate_payload(base)
        case_path = OUT_DIR / f"case_{i:05d}.json"
        case_path.write_text(json.dumps(mutated, indent=2))
        if i % 1000 == 0:
            print(f"[+] generated {i} cases")

if __name__ == "__main__":
    random.seed(0xDEADBEEF)  # deterministic reproducibility
    generate_cases()

// run_fuzz.go – orchestrates Docker + gVisor execution
package main

import (
    "context"
    "encoding/json"
    "fmt"
    "io/ioutil"
    "log"
    "os/exec"
    "path/filepath"
    "time"

    "github.com/prometheus/client_golang/prometheus"
)

var (
    execLatency = prometheus.NewHistogramVec(prometheus.HistogramOpts{
        Name:    "tool_call_latency_seconds",
        Help:    "Latency distribution per tool call",
        Buckets: prometheus.ExponentialBuckets(0.001, 2, 12), // 1ms → 2s
    }, []string{"tool", "status"})
)

func init() {
    prometheus.MustRegister(execLatency)
}

type ToolCall struct {
    Tool   string            `json:"tool"`
    Params map[string]string `json:"params"`
}

func main() {
    cases, _ := filepath.Glob("fuzz_cases/*.json")
    for _, c := range cases {
        payload, _ := ioutil.ReadFile(c)
        var tc ToolCall
        json.Unmarshal(payload, &tc)

        start := time.Now()
        // Docker run with gVisor (--runtime=runsc)
        cmd := exec.CommandContext(context.Background(),
            "docker", "run", "--rm",
            "--runtime=runsc",
            "--cpus=1",
            "--memory=512m",
            "--network=none",
            "-v", fmt.Sprintf("%s:/payload.json:ro", c),
            "futurestack/tool-sandbox:latest",
            "/usr/local/bin/dispatch_tool", "/payload.json")
        out, err := cmd.CombinedOutput()
        latency := time.Since(start).Seconds()
        status := "ok"
        if err != nil {
            status = "error"
        }
        execLatency.WithLabelValues(tc.Tool, status).Observe(latency)

        // Minimal telemetry push (could be replaced by Prometheus pushgateway)
        fmt.Printf("case=%s tool=%s status=%s latency=%.3fs\n", filepath.Base(c), tc.Tool, status, latency)
        if err != nil {
            log.Printf("[FAIL] %s: %v\n%s", c, err, out)
        }
    }
}

GOMAXPROCS=32

sequenceDiagram
    participant SI as Security Information & Event Management (SIEM)
    participant AG as Agent (LLM)
    participant VA as Vault
    participant OP as Ops Team
    participant IR as Incident Response
    SI->>AG: Detect anomalous tool call (e.g., exec /bin/sh)
    SI->>OP: Alert (critical)
    OP->>IR: Escalate
    IR->>VA: Revoke JWTs (rotate)
    IR->>AG: Force graceful shutdown
    IR->>VA: Rotate secret and push new version
    IR->>SI: Log containment timeline

# .github/workflows/credential_compromise.yml
name: Credential Compromise Response
on:
  workflow_dispatch:
    inputs:
      jwt_id:
        description: 'Compromised JWT identifier'
        required: true
        type: string
jobs:
  revoke:
    runs-on: self-hosted
    steps:
      - name: Revoke JWT
        env:
          VAULT_ADDR: ${{ secrets.VAULT_ADDR }}
          VAULT_TOKEN: ${{ secrets.VAULT_ROOT_TOKEN }}
        run: |
          vault write auth/jwt/revoke token=${{ github.event.inputs.jwt_id }}
  shutdown:
    needs: revoke
    runs-on: self-hosted
    steps:
      - name: Stop Agent Container
        run: |
          CONTAINER=$(docker ps --filter "label=service=assistant" -q)
          docker kill --signal=SIGTERM $CONTAINER || docker kill $CONTAINER
  rotate:
    needs: shutdown
    runs-on: self-hosted
    steps:
      - name: Rotate RSA Key
        run: |
          openssl genpkey -algorithm RSA -pkeyopt rsa_keygen_bits:2048 -out new_key.pem
          vault write secret/assistant/key private_key=@new_key.pem