13. Security Best Practices

Chapter 13 of 22 · 20 min

KEY INSIGHT

MCP servers execute arbitrary code on behalf of AI agents, making security not optional but foundationalΓÇöevery exposed endpoint is a potential attack vector. The primary attack surface for MCP servers includes injection through tool parameters, path traversal in file operations, command injection in shell tools, and resource exhaustion via unbounded operations. Each vector requires specific defenses. Input validation forms the first line of defense. Never trust client-supplied data, regardless of apparent format: ```python import re from pathlib import Path def sanitize_path(user_input: str) -> Path: """Safely resolve paths, preventing traversal.""" base = Path("/data").resolve() # Reject absolute paths and traversal attempts if ".." in user_input or user_input.startswith("/"): raise ValueError("Invalid path format") resolved = (base / user_input).resolve() # Ensure result stays within base directory if not resolved.is_relative_to(base): raise ValueError("Path escapes base directory") return resolved @mcp.tool() def read_file(path: str) -> str: safe_path = sanitize_path(path) return safe_path.read_text() ``` Shell command execution requires extreme caution. If shell tools are necessary, use parameterized commands rather than string interpolation: ```python import shlex @mcp.tool() def safe_grep(pattern: str, directory: str) -> list[str]: """Grep with parameterized execution.""" # Validate pattern is safe regex try: re.compile(pattern) except re.error as e: raise ValueError(f"Invalid regex: {e}") # Escape directory for shell safety safe_dir = shlex.quote(directory) safe_pattern = shlex.quote(pattern) result = subprocess.run( ["grep", "-r", safe_pattern, safe_dir], capture_output=True, text=True, timeout=30 ) if result.returncode not in (0, 1): raise RuntimeError(f"grep failed: {result.stderr}") return result.stdout.strip().split("\n") ``` Rate limiting prevents resource exhaustion. Implement per-client throttling: ```python from collections import defaultdict import time class RateLimiter: def __init__(self, max_calls: int, window: float): self.max_calls = max_calls self.window = window self.calls = defaultdict(list) def check(self, client_id: str) -> bool: now = time.time() self.calls[client_id] = [ t for t in self.calls[client_id] if now - t < self.window ] if len(self.calls[client_id]) >= self.max_calls: return False self.calls[client_id].append(now) return True rate_limiter = RateLimiter(max_calls=100, window=60) async def rate_limited_call(client_id: str, func, *args, **kwargs): if not rate_limiter.check(client_id): raise PermissionError("Rate limit exceeded") return await func(*args, **kwargs) ``` Network exposure requires TLS. Local servers should bind to localhost; public deployments need proper certificates. Never transmit unencrypted credentials.

EXERCISE

Audit an existing MCP tool for injection vulnerabilities. Create a test suite with malicious inputs including path traversal sequences (../../etc/passwd), shell metacharacters (; rm -rf), and oversized inputs. Verify sanitization correctly blocks each attack.