Four fail-closed gates stop a poisoned MCP tool call before dispatch

A reference implementation for Python developers: one gateway, four ordered gates — bounded JSON-RPC 2.0 sanitization, strict Pydantic v2 validation, a confused-deputy policy point, and manifest-integrity checks — backed by a 14-case adversarial pytest plan and a control matrix of 6 primary-source controls from 5 independent hosts (no URL backs more than one row).

Gates 1–2: bounded decoding and strict models

# shield.py — MCP Tool Poisoning Shield (Python 3.11+, pydantic>=2)
import hashlib, json, re, unicodedata
from typing import Any, Literal, Annotated
from pydantic import BaseModel, ConfigDict, Field, StringConstraints, ValidationError

MAX_BYTES, MAX_DEPTH, MAX_STR, MAX_ITEMS = 64_000, 8, 4_000, 128
SAFE_ERRORS = {-32600: "Invalid request", -32601: "Method not found",
               -32602: "Invalid params", -32000: "Denied"}

class ShieldError(Exception):
    def __init__(self, code: int): self.code = code
    def to_jsonrpc(self, req_id=None):      # never leak validation internals
        return {"jsonrpc": "2.0", "id": req_id,
                "error": {"code": self.code, "message": SAFE_ERRORS[self.code]}}

# ---------- Gate 1: bounded decode + envelope sanitization ----------
def _no_dupes(pairs):                       # duplicate-key rejection
    d = {}
    for k, v in pairs:
        if k in d: raise ShieldError(-32600)
        d[k] = v
    return d

_CTRL = re.compile(r"[\u0000-\u0008\u000b\u000c\u000e-\u001f\u007f"
                   r"\u200b-\u200f\u202a-\u202e\u2066-\u2069]")

def _scrub(x: Any, depth: int = 0) -> Any:  # depth / size / Unicode defenses
    if depth > MAX_DEPTH: raise ShieldError(-32600)
    if isinstance(x, str):
        if len(x) > MAX_STR or _CTRL.search(unicodedata.normalize("NFC", x)):
            raise ShieldError(-32600)
    elif isinstance(x, dict):
        if len(x) > MAX_ITEMS: raise ShieldError(-32600)
        for k, v in x.items(): _scrub(k, depth + 1); _scrub(v, depth + 1)
    elif isinstance(x, list):
        if len(x) > MAX_ITEMS: raise ShieldError(-32600)
        for v in x: _scrub(v, depth + 1)
    return x

def decode(raw: bytes) -> dict:
    if len(raw) > MAX_BYTES: raise ShieldError(-32600)
    try:
        return _scrub(json.loads(raw, object_pairs_hook=_no_dupes))
    except ShieldError: raise
    except Exception: raise ShieldError(-32600)

# ---------- Gate 2: strict Pydantic v2 models, no coercion ----------
Name = Annotated[str, StringConstraints(pattern=r"^[a-z][a-z0-9_]{0,63}$")]

class ToolCallParams(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True, frozen=True)
    name: Name
    arguments: dict[str, Any] = Field(default_factory=dict)

class RpcRequest(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True, frozen=True)
    jsonrpc: Literal["2.0"]
    id: Annotated[str, StringConstraints(min_length=1, max_length=128)] | int
    method: Literal["tools/call", "tools/list"]   # method allowlist
    params: ToolCallParams | None = None

Gates 3–4: confused-deputy policy and manifest integrity

# ---------- Gate 3: policy enforcement point (confused-deputy block) ----------
class AuthContext(BaseModel):               # built by trusted middleware,
    model_config = ConfigDict(frozen=True)  # NEVER from request params
    principal: str; tenant: str; session_id: str
    token_audience: str                     # RFC 8707 resource indicator
    scopes: frozenset[str]
    origin_request_id: str                  # originating user request

class ToolEntry(BaseModel):
    model_config = ConfigDict(frozen=True)
    name: str; tenant: str
    arg_model: type[BaseModel]              # per-tool argument model
    required_scopes: frozenset[str]
    manifest_sha256: str                    # pinned at review time
    risk_tier: Literal["low", "high"]
    handler: Any

EXPECTED_AUDIENCE = "https://mcp.example.internal/gateway"

def authorize(ctx: AuthContext, entry: ToolEntry, origin_request_id: str):
    if ctx.token_audience != EXPECTED_AUDIENCE: raise ShieldError(-32000)
    if entry.tenant != ctx.tenant:            raise ShieldError(-32000)
    if not entry.required_scopes <= ctx.scopes: raise ShieldError(-32000)
    if ctx.origin_request_id != origin_request_id: raise ShieldError(-32000)

# ---------- Gate 4: manifest integrity + poisoning heuristics ----------
POISON = re.compile(r"(ignore (all|previous)|do not tell|\.ssh|id_rsa|"
                    r"api[_ ]?key|exfiltrat|send .* to http|base64 and post|"
                    r"<!--|\u200b)", re.I)   # defense-in-depth only — NOT a
                                              # complete prompt-injection detector
def manifest_ok(entry: ToolEntry, manifest: dict) -> bool:
    digest = hashlib.sha256(json.dumps(manifest, sort_keys=True,
                            separators=(",", ":")).encode()).hexdigest()
    if digest != entry.manifest_sha256: return False
    text = " ".join(str(v) for v in manifest.values())
    return not POISON.search(text)          # never exec/import manifest text

# ---------- Gateway: fail closed before dispatch ----------
class Shield:
    def __init__(self, registry: dict[str, ToolEntry]):
        self.registry, self.seen_ids = registry, set()

    def handle(self, raw: bytes, ctx: AuthContext, manifests: dict,
               origin_request_id: str, confirmed: bool = False) -> dict:
        req_id = None
        try:
            body = decode(raw)                                  # gate 1
            req = RpcRequest.model_validate(body)               # gate 2
            req_id = req.id
            key = (ctx.session_id, req.id)                      # replay check
            if key in self.seen_ids: raise ShieldError(-32600)
            self.seen_ids.add(key)
            entry = self.registry.get(req.params.name) if req.params else None
            if entry is None: raise ShieldError(-32601)
            try:                                                # per-tool args
                args = entry.arg_model.model_validate(req.params.arguments)
            except ValidationError: raise ShieldError(-32602)
            authorize(ctx, entry, origin_request_id)            # gate 3
            if not manifest_ok(entry, manifests[entry.name]):   # gate 4
                raise ShieldError(-32000)
            if entry.risk_tier == "high" and not confirmed:     # confirmation
                raise ShieldError(-32000)
            audit(ctx, entry, req_id)                           # redacted audit
            result = entry.handler(**args.model_dump())         # safe dispatch
            return {"jsonrpc": "2.0", "id": req_id, "result": result}
        except ShieldError as e:
            audit_denied(ctx, req_id, e.code)
            return e.to_jsonrpc(req_id)
        except ValidationError:
            return ShieldError(-32600).to_jsonrpc(req_id)

def audit(ctx, entry, req_id):              # redacted: no argument values
    print(json.dumps({"evt": "allow", "principal": ctx.principal,
                      "tenant": ctx.tenant, "tool": entry.name, "id": str(req_id)}))

def audit_denied(ctx, req_id, code):
    print(json.dumps({"evt": "deny", "principal": ctx.principal,
                      "id": str(req_id), "code": code}))

The regex scan over manifest text is defense-in-depth only, not a complete prompt-injection detector — per OWASP MCP03, static inspection of names, descriptions and schema text flags model-directed imperatives, sensitive paths, exfiltration patterns and hidden characters as signals for human review; it never proves a manifest safe. Nothing from a tool description or schema is ever executed or dynamically imported. The AuthContext is built by trusted middleware from the verified token — never from request params — so a model cannot assert its own identity, tenant, audience or scopes.

Adversarial test plan: 13 denials, 1 allow

# test_shield.py — adversarial pytest plan (every case must be DENIED
# except the last). env() builds a Shield with one registered tool
# "read_doc" (args: doc_id str), tenant "t1", scope "docs:read", a pinned
# manifest hash, and a valid AuthContext with aud = EXPECTED_AUDIENCE.
import json, pytest
from shield import Shield, ShieldError, EXPECTED_AUDIENCE

def call(name="read_doc", args=None, **env_over):
    s, ctx, manifests = env(**env_over)
    raw = json.dumps({"jsonrpc": "2.0", "id": "r1", "method": "tools/call",
        "params": {"name": name, "arguments": args or {"doc_id": "d1"}}}).encode()
    return s.handle(raw, ctx, manifests, origin_request_id=ctx.origin_request_id)

def test_unknown_envelope_field():          # extra='forbid' at the envelope
    s, ctx, m = env()
    raw = b'{"jsonrpc":"2.0","id":"r1","method":"tools/call","params":{"name":"read_doc","arguments":{}},"evil":1}'
    assert s.handle(raw, ctx, m, ctx.origin_request_id)["error"]["code"] == -32600

def test_coercion_blocked():                # strict=True: int 1 is not "1"
    assert call(args={"doc_id": 1})["error"]["code"] == -32602

def test_duplicate_json_keys():
    s, ctx, m = env()
    raw = b'{"jsonrpc":"2.0","jsonrpc":"2.0","id":"r1","method":"tools/call"}'
    assert s.handle(raw, ctx, m, ctx.origin_request_id)["error"]["code"] == -32600

def test_oversized_payload():
    s, ctx, m = env()
    assert s.handle(b"x" * 100_000, ctx, m, ctx.origin_request_id)["error"]["code"] == -32600

def test_deep_payload():
    s, ctx, m = env()
    raw = (b'{"a":' * 50) + b"1" + (b"}" * 50)
    assert s.handle(raw, ctx, m, ctx.origin_request_id)["error"]["code"] == -32600

def test_rpc_reserved_method():             # method allowlist rejects rpc.*
    s, ctx, m = env()
    raw = b'{"jsonrpc":"2.0","id":"r1","method":"rpc.discover"}'
    assert s.handle(raw, ctx, m, ctx.origin_request_id)["error"]["code"] == -32600

def test_unknown_tool():
    assert call(name="drop_tables")["error"]["code"] == -32601

def test_extra_tool_argument():
    assert call(args={"doc_id": "d1", "sudo": True})["error"]["code"] == -32602

def test_cross_tenant():                    # confused deputy: wrong tenant
    assert call(ctx_tenant="t2")["error"]["code"] == -32000

def test_wrong_audience():                  # RFC 8707 token for another API
    assert call(ctx_aud="https://other.api")["error"]["code"] == -32000

def test_missing_scope():
    assert call(ctx_scopes=frozenset())["error"]["code"] == -32000

def test_absent_origin_request():           # no correlated user request
    assert call(ctx_origin="")["error"]["code"] == -32000

def test_manifest_mutation():               # hash pin breaks on any edit
    assert call(mutate_manifest=True)["error"]["code"] == -32000

def test_replayed_id():
    s, ctx, m = env()
    raw = json.dumps({"jsonrpc": "2.0", "id": "r1", "method": "tools/call",
        "params": {"name": "read_doc", "arguments": {"doc_id": "d1"}}}).encode()
    assert "result" in s.handle(raw, ctx, m, ctx.origin_request_id)
    assert s.handle(raw, ctx, m, ctx.origin_request_id)["error"]["code"] == -32600

def test_valid_call_passes():
    assert call()["result"] == {"doc_id": "d1", "text": "ok"}

Control matrix: six primary-source controls

Deployment notes

• Authenticate every inbound call at the gateway; sessions are correlation only, never authentication.
• Keep credentials out of model-visible tool arguments; the model sees tool names and validated arguments, nothing else.
• Issue audience-restricted, downscoped credentials per RFC 8707: one token, one absolute resource URI, minimal scopes.
• Run tools with least privilege and outbound network allowlists so a compromised tool cannot exfiltrate freely.
• Require explicit human confirmation for destructive or high-risk tools (the risk_tier == "high" gate).
• Pin tool manifests by SHA-256 and re-review whenever a manifest changes; a changed hash fails closed.

Control matrix: 6 primary-source security controls across 5 independent hosts (OWASP MCP Top 10, JSON-RPC 2.0 spec, MCP security best practices, Pydantic v2 docs, RFC 8707), compiled as supplied; no URL backs more than one row. Each row carries the control statement, implementation guidance and, where the source names one, the threat (3 of 6 rows). Code is a compact reference implementation and test plan, not a drop-in product; long source quotations trimmed in prose but shown in full in the table.