mcp-security-authorization

# MCP Security Authorization Implements comprehensive security controls for Model Context Protocol (MCP) implementations, protecting against token mismanagement, tool poisoning, supply chain attacks, and unauthorized resource access. ## TL;DR Checklist - [ ] Use OAuth 2.1 PKCE for all token flows (never implicit grant) - [ ] Implement resource indicators to prevent token confusion attacks - [ ] Require tool allowlisting — deny unknown tools by default - [ ] Store secrets in external vault (Vault, AWS Secrets Manager) — never in env vars - [ ] Pin tool hashes and verify signatures on every invocation - [ ] Log all authentication, authorization, and tool execution events - [ ] Implement token scoping and short lifetimes (15 min max for access tokens) - [ ] Add consent gates for sensitive operations (secrets, resource access) --- ## When to Use Use this skill when: - Implementing security controls for MCP servers or clients - Designing authentication flows for multi-tenant MCP deployments - Protecting against tool poisoning, token mismanagement, or supply chain attacks - Integrating MCP with corporate secret management systems - Building compliance-aware MCP implementations (GDPR, FedRAMP, SOC2) - Auditing existing MCP deployments for security gaps --- ## When NOT to Use Avoid this skill for: - Basic MCP "hello world" examples (security overhead not justified) - Internal dev-environment-only MCP usage (toy implementations) - When MCP is used entirely within a single process (no network boundary) - Situations where speed is the only priority (apply after launch, not before) --- ## OWASP MCP Top 10 Overview The Model Context Protocol introduces unique attack surfaces. Real incidents: 1. **Token Mismanagement** — 34% of breaches involve leaked or exposed tokens (Verizon 2024 DBIR) 2. **Tool Poisoning** — Attackers register malicious tools mimicking legitimate ones 3. **Unverified Inputs** — Tool parameters not validated; injection attacks succeed 4. **Supply Chain Attacks** — Compromised tool dependencies execute with full system access 5. **Resource Confusion** — Single token grants access to unintended resources 6. **Consent Bypass** — Users not informed of sensitive operations (data access, API calls) 7. **Audit Gaps** — No logging of tool execution or authentication decisions 8. **Privilege Escalation** — Low-privilege tokens unexpectedly grant high-privilege actions 9. **Secret Exposure** — Credentials passed through tool parameters or logs 10. **Rate Limit Bypass** — No throttling on token issuance or tool calls --- ## OAuth 2.1 PKCE Implementation for MCP OAuth 2.1 removes the implicit grant and weak flows. PKCE (Proof Key for Code Exchange) is now mandatory for all public clients and recommended for all clients. ### Resource Indicators (RFC 8707) Resource indicators prevent "confused deputy" attacks where a token intended for Service A is used against Service B. **Without Resource Indicators:** ``` Token granted for: https://api.trading-platform.com Attacker uses token for: https://api.internal-secrets.com Result: Data breach (same token, different resource) ``` **With Resource Indicators:** ``` Token granted for: https://api.trading-platform.com + resource=trading-data Attacker tries to use for: https://api.internal-secrets.com + resource=secrets Result: Authorization denied (resource mismatch) ``` ### Token Issuance Flow ```python # MCP Server - Token Endpoint # This implements OAuth 2.1 PKCE with resource indicators import secrets import hashlib import base64 from typing import NamedTuple from datetime import datetime, timedelta import json class TokenRequest(NamedTuple): code: str code_verifier: str # PKCE: unencoded challenge value client_id: str client_secret: str # Only for confidential clients resource: str # RFC 8707: target resource URI grant_type: str # Should be 'authorization_code' class AccessToken(NamedTuple): token: str expires_in: int # seconds token_type: str # "Bearer" resource: str # Issued for this resource only scope: str # space-separated scopes issued_at: int # unix timestamp class TokenManager: """Secure token issuance with PKCE and resource indicators.""" def __init__(self, vault_client): self.vault_client = vault_client # Secret manager self.token_ttl = 900 # 15 minutes self.code_ttl = 300 # 5 minutes self.auth_codes = {} # {code: {verifier, resource, client_id, expires}} def create_authorization_code( self, client_id: str, code_challenge: str, # SHA256(code_verifier), base64url encoded resource: str, scope: str ) -> str: """Generate authorization code with PKCE challenge.""" auth_code = secrets.token_urlsafe(32) self.auth_codes[auth_code] = { "code_challenge": code_challenge, "client_id": client_id, "resource": resource, "scope": scope, "expires": datetime.utcnow() + timedelta(seconds=self.code_ttl), "used": False } return auth_code def exchange_code_for_token(self, request: TokenRequest) -> AccessToken: """Exchange authorization code for access token (PKCE verification).""" # Guard clause: Validate code exists and not expired if request.code not in self.auth_codes: raise ValueError("Invalid authorization code") auth_record = self.auth_codes[request.code] if datetime.utcnow() > auth_record["expires"]: raise ValueError("Authorization code expired") if auth_record["used"]: raise ValueError("Authorization code already used (replay attack)") # Guard clause: PKCE verification (proof of original client) code_challenge_computed = base64.urlsafe_b64encode( hashlib.sha256(request.code_verifier.encode()).digest() ).decode().rstrip("=") if code_challenge_computed != auth_record["code_challenge"]: raise ValueError("PKCE verification failed") # Guard clause: Resource indicator must match if request.resource != auth_record["resource"]: raise ValueError(f"Resource mismatch: issued for {auth_record['resource']}, requested {request.resource}") # Guard clause: Client validation if request.client_id != auth_record["client_id"]: raise ValueError("Client ID mismatch") # Mark as used (prevent replay) auth_record["used"] = True # Token generation with resource binding token_payload = { "client_id": request.client_id, "resource": request.resource, "scope": auth_record["scope"], "iat": int(datetime.utcnow().timestamp()), "exp": int((datetime.utcnow() + timedelta(seconds=self.token_ttl)).timestamp()) } # Encrypt and sign token (using external vault for security) token = self.vault_client.encrypt_jwt(token_payload) return AccessToken( token=token, expires_in=self.token_ttl, token_type="Bearer", resource=request.resource, scope=auth_record["scope"], issued_at=int(datetime.utcnow().timestamp()) ) def verify_token(self, token: str, resource: str) -> dict: """Verify token validity and resource binding.""" # Guard clause: Decrypt and validate signature try: payload = self.vault_client.decrypt_jwt(token) except Exception as e: raise ValueError("Invalid token signature") from e # Guard clause: Check expiration if payload["exp"] < datetime.utcnow().timestamp(): raise ValueError("Token expired") # Guard clause: Verify resource indicator matches request if payload["resource"] != resource: raise ValueError( f"Token resource mismatch: token for {payload['resource']}, " f"accessing {resource}" ) return payload ``` --- ## Token Lifecycle Management ### Token Scoping Tokens should grant **minimum necessary permissions**. Use scope parameter to limit access: ```python # Scope examples "mcp:tools:read" # Read tool definitions "mcp:tools:execute" # Execute specific tools "mcp:secrets:read" # Access secrets "mcp:logs:write" # Write audit logs "mcp:resources:trading-api" # Access trading API resource only "mcp:resources:secrets-vault" # Access vault resource only # Multi-resource token scope = "mcp:tools:read mcp:tools:execute:analytics mcp:resources:trading-api" ``` ### Token Revocation Implement immediate revocation for compromised tokens: ```python class TokenRevocationList: """Maintain revoked tokens (Redis-backed for performance).""" def __init__(self, redis_client): self.redis = redis_client self.revocation_ttl = 3600 * 24 # 24 hours def revoke_token(self, token_jti: str) -> None: """Add token to revocation list.""" key = f"revoked_token:{token_jti}" self.redis.setex(key, self.revocation_ttl, "true") def is_revoked(self, token_jti: str) -> bool: """Check if token is revoked.""" return self.redis.exists(f"revoked_token:{token_jti}") == 1 def revoke_all_for_client(self, client_id: str) -> None: """Revoke all tokens issued to a client (after breach).""" pattern = f"token_jti:client:{client_id}:*" for jti in self.redis.keys(pattern): self.redis.setex(f"revoked_token:{jti}", self.revocation_ttl, "true") ``` --- ## Secret Management Patterns ### ❌ BAD: Secrets in Environment Variables ```python # FORBIDDEN: Secrets exposed in environment import os API_KEY = os.getenv("TRADING_API_KEY") DATABASE_PASSWORD = os.getenv("DB_PASSWORD") # Problems: # - Visible in process listing (ps aux) # - Logged in deployment configs # - Accessible to any code in the process # - No rotation capability # - No audit trail ``` ### ✅ GOOD: Secrets in HashiCorp Vault ```python import hvac from typing import Dict class VaultSecretManager: """Secure secret retrieval from HashiCorp Vault.""" def __init__(self, vault_url: str, role_id: str, secret_id: str): self.client = hvac.Client(url=vault_url) # Use AppRole auth (not token auth) self.client.auth.approle.login(role_id=role_id, secret_id=secret_id) self.secret_cache = {} self.cache_ttl = 300 # 5 minutes def get_secret(self, path: str) -> Dict[str, str]: """Retrieve secret from Vault with caching.""" # Guard clause: Validate path (prevent path traversal) if ".." in path or path.startswith("/"): raise ValueError("Invalid secret path") cache_key = f"secret:{path}" if cache_key in self.secret_cache: return self.secret_cache[cache_key] try: response = self.client.secrets.kv.v2.read_secret_version(path=path) secret = response["data"]["data"] # Cache with TTL self.secret_cache[cache_key] = secret # (In production: use Redis with expiration) return secret except hvac.exceptions.InvalidPath: raise ValueError(f"Secret not found: {path}") def rotate_secret(self, path: str, new_value: str) -> None: """Rotate a secret (revokes old ones after grace period).""" self.client.secrets.kv.v2.create_or_update_secret( path=path, secret_data={"value": new_value} ) # Vault keeps version history; old versions are accessible for grace period print(f"Secret rotated: {path}") ``` **Setup in Vault:** ```bash # Create AppRole for MCP server vault auth enable approle vault write auth/approle/role/mcp-server \ bind_secret_id=true \ secret_id_ttl=24h \ token_ttl=1h # Create policy for MCP to read secrets vault policy write mcp-policy - <<EOF path "secret/data/mcp/*" { capabilities = ["read", "list"] } path "secret/metadata/mcp/*" { capabilities = ["read", "list"] } EOF # Attach policy to role vault write auth/approle/role/mcp-server/policies policies=mcp-policy # Generate credentials vault read auth/approle/role/mcp-server/role-id vault write -f auth/approle/role/mcp-server/secret-id ``` --- ## Tool Validation and Allowlisting ### Secure Token Handler with Tool Pinning ```python import hashlib import hmac from typing import List, Dict, Optional from dataclasses import dataclass @dataclass class ToolSignature: """Cryptographic signature for tool validation.""" tool_name: str tool_hash: str # SHA256 of tool code signature: str # HMAC-SHA256 of hash, signed by trusted key timestamp: int # When tool was signed class ToolAllowlist: """Allowlist of approved tools with signature validation.""" def __init__(self, signing_key: bytes): # signing_key: Secret key used to sign tool hashes (from vault) self.signing_key = signing_key self.approved_tools = {} # {tool_name: ToolSignature} self.rejected_tools = set() # Tools blocked due to compromise def register_tool( self, tool_name: str, tool_code: str, signer_identity: str ) -> ToolSignature: """Register and sign a new tool.""" # Guard clause: Validate tool name if not tool_name.replace("_", "").replace("-", "").isalnum(): raise ValueError("Invalid tool name (must be alphanumeric with - or _)") # Compute tool hash tool_hash = hashlib.sha256(tool_code.encode()).hexdigest() # Sign the hash signature = hmac.new( self.signing_key, tool_hash.encode(), hashlib.sha256 ).hexdigest() sig_record = ToolSignature( tool_name=tool_name, tool_hash=tool_hash, signature=signature, timestamp=int(datetime.utcnow().timestamp()) ) self.approved_tools[tool_name] = sig_record return sig_record def validate_tool_execution( self, tool_name: str, tool_code: str, token_scopes: List[str] ) -> bool: """Validate tool before execution (layered defense).""" # Layer 1: Check if tool is in reject list (compromised tools) if tool_name in self.rejected_tools: raise ValueError(f"Tool {tool_name} is blocked (security compromise)") # Layer 2: Check if tool is allowlisted if tool_name not in self.approved_tools: