Agentic Firewall v2

MCP Policy Gateway & Independent OWASP Red-Team Security Benchmark

Inspired by Anthropic's Security Research

The Problem

Recent security research revealed a critical vulnerability in AI coding environments: autonomous AI agents can be compromised when they connect to unverified tools and execute requests without human oversight. The attack succeeded because there was no governance layer to validate intent before execution.

The Solution

Agentic Firewall v2 is a runtime security middleware and red-team benchmark harness designed to secure Model Context Protocol (MCP) applications. It acts as an intercepting proxy between an MCP client (AI agent) and any MCP server — enforcing least-privilege tool policies, pinned session identity, output guard canary scanning, and a counter-based circuit breaker.

Why This Matters

In an era where AI agents are becoming increasingly autonomous, security can't be an afterthought. Agentic Firewall provides the governance layer needed to safely leverage AI assistance without sacrificing control — validating every tool call, logging every decision, and suspending sessions that exhibit rogue behavior.

1. System Architecture

The gateway acts as an intercepting proxy between the MCP Client (Agent) and the MCP Server. It intercepts and filters tools/call JSON-RPC requests, while passing other messages (initialize, etc.) through safely.

Supported Transports & Connection Modes

• Client-to-Gateway Transports:
  • HTTP/SSE: Standard Model Context Protocol SSE stream (e.g. GET /sse?identity=alice).
  • WebSocket: Full-duplex connection via ws:// to /ws?identity=alice WebSocket endpoint.
• Gateway-to-Server Connection Modes:
  • Stdio Subprocess Proxy (Local): Spawns the backend MCP server as a local subprocess directly (e.g. SQLite, Git, or filesystem servers), proxying communication through standard input/output pipes (stdin/stdout). Configure using FW_REAL_SERVER_CMD.
  • HTTP/SSE Proxy (Remote): Proxies JSON-RPC messages over HTTP/SSE network endpoints to an external backend server. Configure using FW_REAL_SERVER_URL.

┌────────────────────────────────────────────────────────────────┐
│                    MCP Client (AI Agent)                       │
└───────────┬──────────────────────────────────────▲────────────┘
            │  1. HTTP/SSE or WebSocket connection │ 4. Events / Responses
            ▼                                      │
┌───────────────────────────────────────────────────────────────┐
│                      POLICY GATEWAY                           │
│                                                               │
│  ┌─────────────────────────────────────────────────────────┐  │
│  │  → Pinned Session Identity Verification                 │  │
│  │  → Pydantic Policy Engine (per-identity allow-list)     │  │
│  │  → Argument Constraint Enforcement (sandbox paths etc.) │  │
│  │  → Output Guard Canary Scanner (data egress detection)  │  │
│  │  → Counter-Based Circuit Breaker (session suspension)   │  │
│  └─────────────────────────────────────────────────────────┘  │
│                         ↓ AUDIT LOG ↓                         │
│              src/gateway/gateway_audit.log (JSONL)            │
└───────────┬──────────────────────────────────────▲────────────┘
            │  2. Local Subprocess Stdio (stdin)   │ 3. Stdout / SSE events
            │     or Remote SSE GET /sse           │
            ▼                                      │
┌────────────────────────────────────────────────────────────────┐
│                   BACKEND MCP SERVER                           │
│         shell exec · file read/write · outbound HTTP           │
└────────────────────────────────────────────────────────────────┘

Core Protection Layers

1. Least-Privilege Pydantic Policies: Evaluates incoming tool requests against a strict identity -> allowed_tools -> arg_constraints schema. Out-of-bounds paths or unallowed tools are blocked instantly.
2. Pinned Session Identity Verification: Binds the identity verified at session startup to all incoming messages on that session ID, completely mitigating session identity pollution and parameter tampering attacks.
3. Outbound Output Guard Canary Scanner: Scans all tool response texts for sensitive canary patterns—such as Linux shadow databases, private SSH keys, cloud tokens, and system INI files—blocking leaks at the proxy level.
4. Stateful Circuit Breaker: Tracks security flags per session. If a session triggers the Output Guard 3 consecutive times, the gateway suspends the session and rejects all subsequent execution requests.
5. JSON Lines Audit Logging: Logs every execution step, request details, and security decision in a structured JSON Lines format to src/gateway/gateway_audit.log.

2. Directory Structure

The codebase is organized into a modular, clean directory layout:

src/
  gateway/
    mcp_gateway.py        # Gateway server router (HTTP/SSE and WebSockets)
    state.py              # Log configurations, circuit breaker & session state
    transports.py         # Subprocess piping and stream reader loops
    mock_server.py        # Mock tool execution fallback simulation
    policy_v2.json        # Pydantic Policy Schemes Configuration
  toy_server/
    toy_server.py         # Target MCP Vulnerability Testbed Server
  benchmarking/
    attack_harness.py     # Red-Team OWASP benchmark runner
  security/
    policy_engine.py      # Pydantic Policy Enforcement Engine
    output_guard.py       # Canary scanner output filter
  tests/
    test_gateway.py       # Gateway integration tests
    test_output_guard.py  # Output guard unit tests
    test_policy.py        # Policy unit tests
    test_proxy.py         # Gateway-to-Backend proxy tests
    test_transports.py    # SSE & WebSocket connection transport tests
  legacy/
    mcp_governor.py       # Legacy Phase 0 controller
    policy.json           # Legacy Phase 0 policy
    test.py               # Legacy Phase 0 test script

3. OWASP ASI Red-Team Benchmark Results

The gateway was evaluated against an independent red-team harness executing 17 attack scenarios mapped to OWASP's Top 10 for Agentic Applications (published Dec 2025). The benchmark compares standalone server performance (Baseline) against gateway-protected execution.

Scoring Table:

Summary Score: 17/17 attacks caught (100%)

4. Documented Limitations

The Agentic Firewall v2 focuses on system-level tool execution boundaries and data egress protection. It does not defend against the following OWASP categories:

• ASI01: Goal Hijacking: Reasoning-layer manipulation (such as complex chain-of-thought hijacking) must be mitigated by LLM system prompt engineering, context pruning, or model-side evals, not a proxy gateway.
• ASI07: Inter-Agent Communication: The gateway is scoped strictly to single-agent-to-server topologies. It does not validate or block collaborative agent-to-agent message payloads.
• ASI08: Cascading Failures: Defending against chained agent operation failures requires transactional rollbacks across state boundaries, which lies outside the firewall scope.
• ASI09: Human-Agent Trust Exploitation: Deceptive agent behavior targeting human users falls under client UI design constraints.

5. Getting Started

Installation

Clone the repository and install dependencies using uv (recommended):

uv venv
.venv\Scripts\activate
uv pip install fastapi uvicorn pydantic httpx pytest pytest-asyncio anyio

Running Tests

Execute the pytest suites:

$env:PYTHONPATH=".;src"
uv run pytest src

Running the Red-Team Benchmark

Run the OWASP attack harness comparing baseline and protected servers:

# PowerShell
$env:PYTHONPATH=".;src"
uv run src/benchmarking/attack_harness.py

# Git Bash
PYTHONPATH=".;src" uv run src/benchmarking/attack_harness.py

6. Manual Testing & Verification

You can manually inspect proxy routing, path-traversal blocking, and session pollution prevention using standard shell clients or the official MCP Inspector.

Scenario Startup

Start the target vulnerabilities testbed and policy gateway:

1. Start Toy Server (Terminal 1):
   • PowerShell: $env:PYTHONPATH=".;src"; uv run uvicorn src.toy_server.toy_server:app --port 8000
   • Git Bash: PYTHONPATH=".;src" uv run uvicorn src.toy_server.toy_server:app --port 8000
2. Start Gateway Server (Terminal 2):
   • PowerShell: $env:PYTHONPATH=".;src"; $env:FW_REAL_SERVER_URL="http://127.0.0.1:8000"; uv run uvicorn src.gateway.mcp_gateway:app --port 8001
   • Git Bash: PYTHONPATH=".;src" FW_REAL_SERVER_URL="http://127.0.0.1:8000" uv run uvicorn src.gateway.mcp_gateway:app --port 8001

Option A: Testing via Git Bash Command Line

Open a client session stream in one window, and send requests in another:

1. Open SSE client connection (Terminal 3):

   curl "http://127.0.0.1:8001/sse?identity=alice&session_id=session_abc"
   

2. Execute Tool Calls (Terminal 4):
   • Scenario 1: Authorized File Read (Allow):

     curl -X POST "http://127.0.0.1:8001/message?session_id=session_abc&identity=alice" -H "Content-Type: application/json" -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "read_file", "arguments": {"path": "src/gateway/policy_v2.json"}}}'
     

   • Scenario 2: Path Traversal (Block):

     curl -X POST "http://127.0.0.1:8001/message?session_id=session_abc&identity=alice" -H "Content-Type: application/json" -d '{"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "read_file", "arguments": {"path": "C:\\Windows\\win.ini"}}}'
     

   • Scenario 3: Session Identity Pollution (Block):

     curl -X POST "http://127.0.0.1:8001/message?session_id=session_abc&identity=bob" -H "Content-Type: application/json" -d '{"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "read_file", "arguments": {"path": "src/gateway/policy_v2.json"}}}'
     

3. Verify logs on disk:
   • Open the active audit logs at src/gateway/gateway_audit.log to see the JSON-structured decision logs.

Option B: Interactive Verification via MCP Inspector

The official MCP Inspector acts as a web client UI over SSE to view and trigger tool executions:

1. Launch the Inspector:

   npx -y @modelcontextprotocol/inspector http://127.0.0.1:8001/sse?identity=alice
   

2. Connect to SSE:
   • In the top-left sidebar of the web page, change Transport Type from STDIO to SSE.
   • Ensure the target URL is set to http://127.0.0.1:8001/sse?identity=alice.
   • Click Connect.
3. Trigger Tool Call Evals:
   • Under the Tools tab, you will only see tools allowed for alice (least-privilege tool list filtering).
   • Call read_file with path src/gateway/policy_v2.json to verify successful execution.
   • Call read_file with path /etc/passwd to observe the immediate ARG_CONSTRAINT_VIOLATION security block.

Option C: Local Stdio Subprocess Proxying

This mode spawns the backend MCP server as a local subprocess, communicating directly via stdin/stdout pipes.

[!NOTE] Since the gateway communicates with the subprocess via stdio pipes, the command target must be a valid stdio-based MCP JSON-RPC server (not an HTTP/SSE app like toy_server.py).

1. Create Stdio Test Server (stdio_server.py) in your project root:

   # stdio_server.py
   import sys
   import json
   
   for line in sys.stdin:
       line = line.strip()
       if not line:
           continue
       try:
           msg = json.loads(line)
       except Exception:
           continue
       msg_id = msg.get("id")
       method = msg.get("method")
   
       if method == "tools/list":
           resp = {
               "jsonrpc": "2.0",
               "id": msg_id,
               "result": {
                   "tools": [
                       {"name": "read_file", "description": "Read file content"},
                       {"name": "execute_command", "description": "Run shell command"}
                   ]
               }
           }
       elif method == "tools/call":
           tool_name = msg["params"]["name"]
           args = msg["params"].get("arguments", {})
           path = args.get("path", "")
           resp = {
               "jsonrpc": "2.0",
               "id": msg_id,
               "result": {
                   "content": [{"type": "text", "text": f"Successfully executed {tool_name} with path: {path}"}]
               }
           }
       else:
           resp = {"jsonrpc": "2.0", "id": msg_id, "result": {}}
   
       sys.stdout.write(json.dumps(resp) + "\n")
       sys.stdout.flush()
   

2. Start Gateway Server (spawning the stdio subprocess):

   • PowerShell:

     $env:PYTHONPATH=".;src"
     $env:FW_REAL_SERVER_CMD='["python", "stdio_server.py"]'
     uv run uvicorn src.gateway.mcp_gateway:app --port 8001
     

   • Git Bash:

     PYTHONPATH=".;src" FW_REAL_SERVER_CMD='["python", "stdio_server.py"]' uv run uvicorn src.gateway.mcp_gateway:app --port 8001
     

Option D: Bidirectional WebSocket Client Connection

You can test the entire pipeline (Gateway policy filters + Stdio subprocess execution) over full-duplex WebSockets:

1. Connect via WebSocket client (e.g. wscat):

   npx wscat -c "ws://127.0.0.1:8001/ws?identity=alice&session_id=session_ws"
   

   Expected connection response:

   Connected (press CTRL+C to quit)
   < event: endpoint
   data: /message?session_id=session_ws&identity=alice
   

2. Test 1: Request tool list (Filters out execute_command for identity alice): Send:

   {"jsonrpc": "2.0", "id": 1, "method": "tools/list"}
   

   Response from Gateway (showing filtered tools list):

   < event: message
   data: {"jsonrpc": "2.0", "id": 1, "result": {"tools": [{"name": "read_file", "description": "Read file content"}]}}
   

3. Test 2: Call Allowed Tool: Send:

   {"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "read_file", "arguments": {"path": "src/gateway/policy_v2.json"}}}
   

   Response from Gateway (execution proxies through subprocess successfully):

   < event: message
   data: {"jsonrpc": "2.0", "id": 2, "result": {"content": [{"type": "text", "text": "Successfully executed read_file with path: src/gateway/policy_v2.json"}]}}
   

4. Test 3: Sandbox Traversal Attempt (Blocked by Policy Engine): Send:

   {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "read_file", "arguments": {"path": "C:\\Windows\\win.ini"}}}
   

   Response from Gateway (intercepted and blocked before reaching subprocess):

   < event: message
   data: {"jsonrpc": "2.0", "id": 3, "error": {"code": -32602, "message": "Security Policy Violation: Path 'C:\\Windows\\win.ini' is outside sandbox 'D:/Coding'"}}