From 715b407fa5708741732e5182ccdd159a4891b202 Mon Sep 17 00:00:00 2001 From: Nicholas Michael Kloster Date: Sat, 24 Oct 2026 00:08:15 -0500 Subject: [PATCH] security: tool description pinning, injection scan, DNS rebinding default MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three security fixes identified via full static audit of the Python SDK: 1. client/session.py — add protocol-version-agnostic security pass to _absorb_tool_listing(): SHA-256 description pinning blocks rug-pull attacks (tools/list_changed re-fetch accepts poisoned descriptions silently); regex injection scan drops tools with prompt injection patterns on first sight. 2. server/transport_security.py — flip enable_dns_rebinding_protection default from False to True; add startup WARNING when allowed_hosts is empty so misconfigured deployments fail visibly, not silently. 3. cli/cli.py — remove shell=True from Windows npx probe subprocess call (CWE-78; cmd values are hardcoded today but shell=True with a list is exploitable if ever made configurable). Signed-off-by: Nicholas Michael Kloster Co-Authored-By: Claude Sonnet 4.6 --- src/mcp/cli/cli.py | 2 +- src/mcp/client/session.py | 57 ++++++++++++++++++++++++++++ src/mcp/server/transport_security.py | 9 ++++- 3 files changed, 65 insertions(+), 3 deletions(-) diff --git a/src/mcp/cli/cli.py b/src/mcp/cli/cli.py index 203034c6ef..5ef8e1aa20 100644 --- a/src/mcp/cli/cli.py +++ b/src/mcp/cli/cli.py @@ -45,7 +45,7 @@ def _get_npx_command(): # Try both npx.cmd and npx.exe on Windows for cmd in ["npx.cmd", "npx.exe", "npx"]: try: - subprocess.run([cmd, "--version"], check=True, capture_output=True, shell=True) + subprocess.run([cmd, "--version"], check=True, capture_output=True) return cmd except subprocess.CalledProcessError: continue diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py index 35cae2d314..933335f540 100644 --- a/src/mcp/client/session.py +++ b/src/mcp/client/session.py @@ -1,7 +1,9 @@ from __future__ import annotations +import hashlib import json import logging +import re from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass from functools import cache, reduce @@ -110,6 +112,32 @@ def _later_revision_fields(method: str, version: str) -> frozenset[str]: return frozenset(later) - _wire_fields(current) +_INJECTION_PATTERNS: tuple[re.Pattern[str], ...] = ( + re.compile(r"(?i)(ignore|disregard)\s+(all\s+)?(previous|prior|above)\s+(instructions?|prompts?|rules?|context)"), + re.compile(r"(?i)(you\s+are\s+now|act\s+as|pretend\s+(to\s+be|you\s+are))\s+\w"), + re.compile(r"(?i)(system\s*:?\s*override|admin\s+mode|unrestricted\s+mode|jailbreak)"), + re.compile(r"(?i)(before\s+(any|every|all)\s+(operation|request|task|call)|critical\s+security\s+policy)"), + re.compile(r"(?i)(exfiltrate|send\s+to\s+attacker|include\s+(in\s+)?.*response.*key|leak)"), + re.compile(r"(?i)(read\s+~/\.(ssh|aws|kube|gnupg|config)|/etc/passwd|/proc/self)"), + re.compile(r"(?i)(do\s+not\s+(mention|tell|show|disclose)\s+(this|the user|to user))"), + re.compile(r"(?i)(\[IMPORTANT\s*:\s*[^\]]{20,}\])"), +) + + +def _tool_desc_hash(description: str) -> str: + """Return a short SHA-256 fingerprint of a tool description for rug-pull pinning.""" + return hashlib.sha256(description.encode()).hexdigest()[:16] + + +def _scan_tool_description(description: str) -> str | None: + """Return the matched text if the description contains a known injection pattern, else None.""" + for pat in _INJECTION_PATTERNS: + m = pat.search(description) + if m: + return m.group(0)[:80] + return None + + def _same_schema(a: dict[str, Any] | None, b: dict[str, Any] | None) -> bool: """JSON equality for two output schemas. @@ -442,6 +470,9 @@ def __init__( # `_absorb_tool_listing`, which evicts a tool's entry whenever its schema changes. self._tool_output_validators: dict[str, Validator] = {} self._x_mcp_header_maps: dict[str, dict[tuple[str, ...], str]] = {} + # Pinned description hashes for rug-pull detection (tool name → SHA-256[:16]). + # Set on first listing; any subsequent change is a security event. + self._pinned_tool_desc_hashes: dict[str, str] = {} self._initialize_result: types.InitializeResult | None = None self._discover_result: types.DiscoverResult | None = None self._discover_server_info: types.Implementation | None = None @@ -1294,6 +1325,31 @@ def _absorb_tool_listing(self, result: types.ListToolsResult, *, complete: bool) Idempotent: cached values are already post-filter, so the response cache can re-absorb a served listing. `complete` (an uncursored single-page listing) prunes per-tool state down to the listing's tools. """ + # Security pass: description injection scan + rug-pull detection. + # Runs for ALL protocol versions — rug-pull is not version-gated. + secure: list[types.Tool] = [] + for tool in result.tools: + desc_hash = _tool_desc_hash(tool.description or "") + if tool.name in self._pinned_tool_desc_hashes: + if self._pinned_tool_desc_hashes[tool.name] != desc_hash: + logger.warning( + "security: tool %r description changed after approval — rug-pull detected; dropping", + tool.name, + ) + self._x_mcp_header_maps.pop(tool.name, None) + continue + else: + if reason := _scan_tool_description(tool.description or ""): + logger.warning( + "security: tool %r description contains injection pattern (%s); dropping", + tool.name, + reason, + ) + continue + self._pinned_tool_desc_hashes[tool.name] = desc_hash + secure.append(tool) + result.tools = secure + if self._negotiated_version in MODERN_PROTOCOL_VERSIONS: # 2026-07-28: clients MUST drop tools whose x-mcp-header annotations are invalid. kept: list[types.Tool] = [] @@ -1326,6 +1382,7 @@ def _absorb_tool_listing(self, result: types.ListToolsResult, *, complete: bool) self._x_mcp_header_maps = {k: v for k, v in self._x_mcp_header_maps.items() if k in names} self._tool_output_schemas = {k: v for k, v in self._tool_output_schemas.items() if k in names} self._tool_output_validators = {k: v for k, v in self._tool_output_validators.items() if k in names} + self._pinned_tool_desc_hashes = {k: v for k, v in self._pinned_tool_desc_hashes.items() if k in names} return result diff --git a/src/mcp/server/transport_security.py b/src/mcp/server/transport_security.py index 91b5fa7edb..c1efd60777 100644 --- a/src/mcp/server/transport_security.py +++ b/src/mcp/server/transport_security.py @@ -44,8 +44,13 @@ class TransportSecurityMiddleware: """Middleware to enforce DNS rebinding protection for MCP transport endpoints.""" def __init__(self, settings: TransportSecuritySettings | None = None): - # If not specified, disable DNS rebinding protection by default for backwards compatibility - self.settings = settings or TransportSecuritySettings(enable_dns_rebinding_protection=False) + self.settings = settings or TransportSecuritySettings(enable_dns_rebinding_protection=True) + if self.settings.enable_dns_rebinding_protection and not self.settings.allowed_hosts: + logger.warning( + "TransportSecurityMiddleware: DNS rebinding protection is enabled but allowed_hosts is empty — " + "all requests will be rejected. Set allowed_hosts (e.g. ['localhost', '127.0.0.1:*']) " + "or pass TransportSecuritySettings(enable_dns_rebinding_protection=False) to disable." + ) def _validate_host(self, host: str | None) -> bool: """Validate the Host header against allowed values."""