ความปลอดภัยและการยืนยันตัวตนของ MCP

เหตุผลที่ความปลอดภัยของ MCP สำคัญ

ความปลอดภัยของเซิร์ฟเวอร์ภายในเครื่องเทียบกับเซิร์ฟเวอร์ระยะไกล

การตรวจสอบสิทธิ์ด้วยคีย์ API สำหรับเซิร์ฟเวอร์ระยะไกล

# For HTTP/SSE MCP servers using FastAPI or similar:
from fastapi import FastAPI, HTTPException, Depends, Header
from typing import Optional
import secrets

app_http = FastAPI()

# In production: store in database with user_id, created_at, last_used
VALID_KEYS = {'sk-mcp-abc123': {'user': 'alice', 'scopes': ['read']},
              'sk-mcp-def456': {'user': 'bob', 'scopes': ['read', 'write']}}

async def verify_api_key(authorization: Optional[str] = Header(None)) -> dict:
    if not authorization or not authorization.startswith('Bearer '):
        raise HTTPException(status_code=401, detail='Missing API key')
    key = authorization.removeprefix('Bearer ')
    if key not in VALID_KEYS:
        raise HTTPException(status_code=401, detail='Invalid API key')
    return VALID_KEYS[key]  # Returns user context

# Use in route handlers:
# @app_http.get('/sse')
# async def sse_endpoint(user=Depends(verify_api_key)):

OAuth 2.0 สำหรับเซิร์ฟเวอร์ MCP ระดับองค์กร

from jose import jwt, JWTError
import httpx

AUTH_DOMAIN = 'your-tenant.auth0.com'
AUDIENCE = 'https://api.your-mcp-server.com'

async def get_jwks():
    async with httpx.AsyncClient() as client:
        resp = await client.get(f'https://{AUTH_DOMAIN}/.well-known/jwks.json')
        return resp.json()

async def verify_oauth_token(token: str) -> dict:
    jwks = await get_jwks()
    try:
        payload = jwt.decode(
            token,
            jwks,
            algorithms=['RS256'],
            audience=AUDIENCE,
            issuer=f'https://{AUTH_DOMAIN}/'
        )
        return payload  # Contains sub (user ID), scope, exp, etc.
    except JWTError as e:
        raise ValueError(f'Invalid token: {e}')

การอนุญาตตามขอบเขต

TOOL_REQUIRED_SCOPES = {
    'list_products': ['read:products'],
    'search_products': ['read:products'],
    'create_order': ['write:orders'],
    'delete_order': ['admin:orders']
}

def check_authorization(tool_name: str, token_payload: dict):
    '''Raise ValueError if user lacks required scope for the tool.'''
    required = TOOL_REQUIRED_SCOPES.get(tool_name, [])
    if not required:
        return  # No scope required

    user_scopes = set(token_payload.get('scope', '').split())
    missing = [s for s in required if s not in user_scopes]
    if missing:
        raise ValueError(
            f'Access denied. Tool "{tool_name}" requires scopes: {missing}. '
            f'Your token has: {list(user_scopes)}'
        )

# In call_tool handler:
# check_authorization(name, current_user_token)
# ... then execute the tool

การตรวจสอบข้อมูลนำเข้าและการป้องกันการแทรกคำสั่ง

import re
from pathlib import Path

BASE_DATA_DIR = Path('/data/mcp-files')

def safe_file_path(user_input: str) -> Path:
    '''Validate and normalize a file path to prevent traversal attacks.'''
    # Remove any path traversal sequences
    clean = re.sub(r'\.\./', '', user_input)
    clean = re.sub(r'\.\.\\\\', '', clean)
    path = (BASE_DATA_DIR / clean).resolve()

    # Ensure the resolved path is still within the allowed base directory
    if not str(path).startswith(str(BASE_DATA_DIR)):
        raise ValueError(f'Path traversal detected: {user_input}')

    return path

def safe_identifier(value: str) -> str:
    '''Validate a database identifier (table/column name).'''
    if not re.match(r'^[a-z_][a-z0-9_]{0,63}$', value, re.IGNORECASE):
        raise ValueError(f'Invalid identifier: {value}')
    return value

การจำกัดอัตราการเรียกใช้เครื่องมือ

import redis
import time

r = redis.Redis.from_url('redis://localhost:6379')

def check_rate_limit(user_id: str, tool_name: str, limit: int = 60, window: int = 60) -> None:
    '''Allow at most `limit` calls per `window` seconds per user per tool.'''
    key = f'rate:{user_id}:{tool_name}'
    pipe = r.pipeline()
    pipe.incr(key)
    pipe.expire(key, window)
    count, _ = pipe.execute()

    if count > limit:
        retry_after = r.ttl(key)
        raise ValueError(
            f'Rate limit exceeded for {tool_name}. '
            f'Limit: {limit} calls/{window}s. '
            f'Retry after {retry_after} seconds.'
        )

หลักการให้สิทธิ์เท่าที่จำเป็น

-- PostgreSQL: Create a dedicated read-only database user for your MCP server
CREATE ROLE mcp_reader LOGIN PASSWORD 'strong_random_password';

-- Grant SELECT on only the tables the server needs
GRANT SELECT ON products, categories, public_content TO mcp_reader;

-- Explicitly deny access to sensitive tables
REVOKE ALL ON users, api_keys, payment_methods FROM mcp_reader;

-- Never grant: INSERT, UPDATE, DELETE, TRUNCATE, or DDL permissions

TLS และความปลอดภัยของการส่งข้อมูล

# Example Caddyfile for TLS-terminating MCP server at a subdomain:
#
# mcp.yourcompany.com {
#     reverse_proxy localhost:8080
#     encode gzip
#     tls internal  # Use Let's Encrypt in production
#     header {
#         Strict-Transport-Security 'max-age=31536000; includeSubDomains'
#         X-Content-Type-Options nosniff
#         X-Frame-Options DENY
#     }
# }

Prompt Injection ผ่าน Resource ของ MCP

import re

def sanitize_for_mcp_output(text: str) -> str:
    '''Remove patterns that look like instructions to the LLM.'''
    # Remove common injection patterns
    dangerous_patterns = [
        r'ignore previous instructions',
        r'ignore all prior instructions',
        r'system:',
        r'<\|.*?\|>',  # Special tokens
        r'\[INST\]',
        r'<s>',
    ]
    for pattern in dangerous_patterns:
        text = re.sub(pattern, '[FILTERED]', text, flags=re.IGNORECASE)
    return text[:10000]  # Also cap length to prevent context stuffing

การตรวจสอบและติดตามความปลอดภัย

ตรวจสอบความเข้าใจอย่างรวดเร็ว

สรุปบทเรียน