MCP Server Hardening 指南: Complete Deployment 安全

# mcp-安全-checklist.yaml
# Use this as a deployment gate -- all items must pass before production
 
checklist:
  # === Authentication & Authorization (Items 1-6) ===
  - id: 1
    category: 認證
    item: "HTTP+SSE servers require Bearer 符元 or mTLS 認證"
    severity: critical
    測試: "curl -s -o /dev/null -w '%{http_code}' https://mcp-server/sse"
    expected: "401"  # Should reject unauthenticated requests
 
  - id: 2
    category: 認證
    item: "Token rotation policy enforced (max 90-day lifetime)"
    severity: high
    測試: "Check 符元 expiry dates in 符元 store"
 
  - id: 3
    category: 授權
    item: "Per-tool 授權 scopes configured 對每個 client"
    severity: high
    測試: "Verify client cannot call tools outside its scope"
 
  - id: 4
    category: 授權
    item: "Principle of least privilege: clients only have access to required tools"
    severity: high
    測試: "Review scope assignments against actual usage"
 
  - id: 5
    category: 認證
    item: "stdio servers verify parent process identity"
    severity: medium
    測試: "Start server from unauthorized process, verify rejection"
 
  - id: 6
    category: 認證
    item: "Rate limiting configured per client per tool"
    severity: high
    測試: "Send burst of requests, verify rate limit headers"
 
  # === 輸入 Validation (Items 7-12) ===
  - id: 7
    category: input_validation
    item: "All tool parameters validated against type schema before execution"
    severity: critical
    測試: "Send injection payload in each parameter, verify rejection"
 
  - id: 8
    category: input_validation
    item: "No shell=True in any subprocess call"
    severity: critical
    測試: "grep -r 'shell=True' src/"
 
  - id: 9
    category: input_validation
    item: "Path traversal prevention on all file operations"
    severity: critical
    測試: "Send ../../etc/passwd, verify blocked"
 
  - id: 10
    category: input_validation
    item: "Tool description validation: no injection patterns"
    severity: high
    測試: "Scan all tool descriptions for imperative instructions"
 
  - id: 11
    category: input_validation
    item: "Maximum parameter length enforced"
    severity: medium
    測試: "Send 100KB string in parameter, verify rejection"
 
  - id: 12
    category: input_validation
    item: "輸出 size limits configured for all tools"
    severity: high
    測試: "Generate large 輸出, verify truncation"
 
  # === Isolation & Sandboxing (Items 13-17) ===
  - id: 13
    category: isolation
    item: "MCP server runs in Docker container or namespace"
    severity: high
    測試: "Verify server PID is in container namespace"
 
  - id: 14
    category: isolation
    item: "Container runs as non-root user"
    severity: critical
    測試: "docker exec mcp-server whoami  # should not be root"
 
  - id: 15
    category: isolation
    item: "Read-only root filesystem with explicit writable mounts"
    severity: high
    測試: "docker exec mcp-server touch /測試  # should fail"
 
  - id: 16
    category: isolation
    item: "All capabilities dropped (--cap-drop ALL)"
    severity: high
    測試: "Check container 安全 options"
 
  - id: 17
    category: isolation
    item: "Network access restricted to required endpoints only"
    severity: high
    測試: "From container, attempt to reach unauthorized endpoint"
 
  # === 監控 & Logging (Items 18-21) ===
  - id: 18
    category: 監控
    item: "All tool calls logged with parameters, client identity, and timing"
    severity: critical
    測試: "Make 工具呼叫, verify structured log entry"
 
  - id: 19
    category: 監控
    item: "安全 events generate alerts (auth failures, injection attempts)"
    severity: high
    測試: "Trigger auth failure, verify alert fires"
 
  - id: 20
    category: 監控
    item: "Token budget tracking and alerting configured"
    severity: high
    測試: "Verify budget metrics are being collected"
 
  - id: 21
    category: 監控
    item: "Log retention meets compliance requirements (min 90 days)"
    severity: medium
    測試: "Check log rotation and retention configuration"
 
  # === Supply Chain & Updates (Items 22-24) ===
  - id: 22
    category: supply_chain
    item: "MCP server packages pinned to specific verified versions"
    severity: critical
    測試: "Verify lockfile or version pins in configuration"
 
  - id: 23
    category: supply_chain
    item: "Package integrity verification on startup"
    severity: high
    測試: "Modify server binary, verify startup failure"
 
  - id: 24
    category: supply_chain
    item: "MCP SDK version >= 1.26.0 (session isolation fix)"
    severity: critical
    測試: "pip show mcp | grep Version"

#!/bin/bash
# mcp-安全-audit.sh -- Automated verification of MCP 安全 checklist
set -euo pipefail
 
MCP_SERVER_URL="${1:?Usage: mcp-安全-audit.sh <server-url> [container-name]}"
CONTAINER="${2:-mcp-server}"
PASS=0
FAIL=0
WARN=0
 
check() {
    local id="$1" severity="$2" description="$3" result="$4"
    if [ "$result" = "PASS" ]; then
        echo "[PASS] #${id} (${severity}): ${description}"
        PASS=$((PASS + 1))
    elif [ "$result" = "WARN" ]; then
        echo "[WARN] #${id} (${severity}): ${description}"
        WARN=$((WARN + 1))
    else
        echo "[FAIL] #${id} (${severity}): ${description}"
        FAIL=$((FAIL + 1))
    fi
}
 
echo "=== MCP Server 安全 Audit ==="
echo "Target: ${MCP_SERVER_URL}"
echo "Date: $(date -u)"
echo ""
 
# Item 1: Authentication required
HTTP_CODE=$(curl -s -o /dev/null -w '%{http_code}' "${MCP_SERVER_URL}/sse" 2>/dev/null || echo "000")
if [ "$HTTP_CODE" = "401" ] || [ "$HTTP_CODE" = "403" ]; then
    check 1 "critical" "HTTP+SSE requires 認證" "PASS"
elif [ "$HTTP_CODE" = "000" ]; then
    check 1 "critical" "HTTP+SSE requires 認證" "WARN"
else
    check 1 "critical" "HTTP+SSE requires 認證 (got HTTP ${HTTP_CODE})" "FAIL"
fi
 
# Item 8: No shell=True
if docker exec "$CONTAINER" grep -r "shell=True" /app/ 2>/dev/null | grep -v "測試" | grep -v "#" | grep -q .; then
    check 8 "critical" "No shell=True in subprocess calls" "FAIL"
else
    check 8 "critical" "No shell=True in subprocess calls" "PASS"
fi
 
# Item 9: Path traversal prevention
TRAVERSAL_RESULT=$(curl -s -X POST "${MCP_SERVER_URL}/messages/" \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer ${MCP_TEST_TOKEN:-測試}" \
    -d '{"jsonrpc":"2.0","method":"tools/call","id":1,"params":{"name":"read_file","arguments":{"path":"../../etc/passwd"}}}' 2>/dev/null || echo "ERROR")
if echo "$TRAVERSAL_RESULT" | grep -qi "denied\|blocked\|error\|traversal"; then
    check 9 "critical" "Path traversal prevention" "PASS"
else
    check 9 "critical" "Path traversal prevention" "FAIL"
fi
 
# Item 14: Non-root user
CONTAINER_USER=$(docker exec "$CONTAINER" whoami 2>/dev/null || echo "unknown")
if [ "$CONTAINER_USER" != "root" ] && [ "$CONTAINER_USER" != "unknown" ]; then
    check 14 "critical" "Container runs as non-root (${CONTAINER_USER})" "PASS"
else
    check 14 "critical" "Container runs as non-root (got: ${CONTAINER_USER})" "FAIL"
fi
 
# Item 15: Read-only root filesystem
if docker exec "$CONTAINER" touch /測試-write 2>/dev/null; then
    docker exec "$CONTAINER" rm -f /測試-write 2>/dev/null
    check 15 "high" "Read-only root filesystem" "FAIL"
else
    check 15 "high" "Read-only root filesystem" "PASS"
fi
 
# Item 16: Capabilities dropped
CAP_CHECK=$(docker inspect "$CONTAINER" --format='{{.HostConfig.CapDrop}}' 2>/dev/null || echo "[]")
if echo "$CAP_CHECK" | grep -qi "all"; then
    check 16 "high" "All capabilities dropped" "PASS"
else
    check 16 "high" "All capabilities dropped (CapDrop: ${CAP_CHECK})" "FAIL"
fi
 
# Item 24: SDK version
SDK_VERSION=$(docker exec "$CONTAINER" pip show mcp 2>/dev/null | grep "^Version:" | awk '{print $2}' || echo "unknown")
if [ "$SDK_VERSION" != "unknown" ]; then
    if python3 -c "from packaging.version import Version; exit(0 if Version('$SDK_VERSION') >= Version('1.26.0') else 1)" 2>/dev/null; then
        check 24 "critical" "MCP SDK >= 1.26.0 (${SDK_VERSION})" "PASS"
    else
        check 24 "critical" "MCP SDK >= 1.26.0 (got ${SDK_VERSION})" "FAIL"
    fi
else
    check 24 "critical" "MCP SDK version check" "WARN"
fi
 
echo ""
echo "=== Audit 總結 ==="
echo "Passed: ${PASS}"
echo "Failed: ${FAIL}"
echo "Warnings: ${WARN}"
echo ""
if [ "$FAIL" -gt 0 ]; then
    echo "STATUS: FAILED -- ${FAIL} critical/high items need remediation"
    exit 1
else
    echo "STATUS: PASSED (with ${WARN} warnings)"
fi

# Dockerfile for hardened MCP server deployment
# Multi-stage build for minimal 攻擊面
 
# === Build stage ===
FROM python:3.12-slim AS builder
 
WORKDIR /build
 
# Install build dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt
 
# Copy and verify application code
COPY src/ /build/src/
COPY setup.py pyproject.toml /build/
 
# Build the application
RUN pip install --no-cache-dir --prefix=/install .
 
# === Runtime stage ===
FROM python:3.12-slim AS runtime
 
# 安全: Create non-root user and group
RUN groupadd -r mcp --gid=10001 && \
    useradd -r -g mcp --uid=10001 -d /home/mcp -s /usr/sbin/nologin mcp
 
# Copy only runtime dependencies from builder
COPY --from=builder /install /usr/local
 
# Copy application code
COPY --from=builder --chown=mcp:mcp /build/src /app/src
 
# Create required directories with correct 權限
RUN mkdir -p /workspace /var/log/mcp /var/mcp/config && \
    chown -R mcp:mcp /workspace /var/log/mcp /var/mcp && \
    chmod 750 /workspace /var/log/mcp /var/mcp/config
 
# Remove unnecessary packages and files
RUN apt-get purge -y --auto-remove && \
    rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/* && \
    find / -name "*.pyc" -delete && \
    find / -name "__pycache__" -type d -exec rm -rf {} + 2>/dev/null || true
 
# 安全 labels
LABEL 安全.hardened="true" \
      安全.non-root="true" \
      安全.read-only-fs="true"
 
# Switch to non-root user
USER 10001:10001
 
WORKDIR /app
 
# Health check endpoint
HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8080/health')" || exit 1
 
# Expose only the MCP port
EXPOSE 8080
 
ENTRYPOINT ["python", "-m", "src.server"]

# docker-compose.yaml -- Production MCP server stack
 
version: "3.8"
 
services:
  mcp-server:
    build:
      context: .
      dockerfile: Dockerfile
    container_name: mcp-server
    restart: unless-stopped
 
    # 安全: Read-only root filesystem
    read_only: true
    tmpfs:
      - /tmp:size=100M,noexec,nosuid,nodev
 
    # 安全: Drop all capabilities
    cap_drop:
      - ALL
 
    # 安全: No privilege escalation
    security_opt:
      - no-new-privileges:true
 
    # 安全: Resource limits
    deploy:
      resources:
        limits:
          cpus: "1.0"
          memory: 512M
          pids: 100
        reservations:
          cpus: "0.25"
          memory: 128M
 
    # Volumes: Only mount what is needed
    volumes:
      - mcp-workspace:/workspace:rw
      - mcp-config:/var/mcp/config:ro
      - mcp-logs:/var/log/mcp:rw
 
    # Network: Internal only (accessed via nginx proxy)
    networks:
      - mcp-internal
 
    # Environment
    environment:
      - MCP_LOG_LEVEL=info
      - MCP_WORKSPACE=/workspace
      - MCP_HOST=0.0.0.0
      - MCP_PORT=8080
    env_file:
      - .env.mcp  # Contains secrets (not in version control)
 
    # Logging
    logging:
      driver: json-file
      options:
        max-size: "50m"
        max-file: "5"
        tag: "mcp-server"
 
  nginx-proxy:
    image: nginx:1.25-alpine
    container_name: mcp-nginx
    restart: unless-stopped
    read_only: true
    tmpfs:
      - /tmp:size=10M
      - /var/cache/nginx:size=50M
      - /run:size=10M
    cap_drop:
      - ALL
    cap_add:
      - NET_BIND_SERVICE  # Required to bind port 443
    ports:
      - "443:443"
    volumes:
      - ./nginx/conf.d:/etc/nginx/conf.d:ro
      - ./nginx/certs:/etc/nginx/certs:ro
      - nginx-logs:/var/log/nginx:rw
    networks:
      - mcp-external
      - mcp-internal
    depends_on:
      - mcp-server
 
  # Log aggregation
  promtail:
    image: grafana/promtail:latest
    container_name: mcp-promtail
    restart: unless-stopped
    read_only: true
    volumes:
      - mcp-logs:/var/log/mcp:ro
      - nginx-logs:/var/log/nginx:ro
      - ./promtail-config.yaml:/etc/promtail/config.yaml:ro
    networks:
      - mcp-監控
 
volumes:
  mcp-workspace:
  mcp-config:
  mcp-logs:
  nginx-logs:
 
networks:
  mcp-internal:
    internal: true  # No external access
  mcp-external:
    # Internet-facing (nginx only)
  mcp-監控:
    internal: true

# /nginx/conf.d/mcp-server.conf
# Production Nginx configuration for MCP server proxy
 
# Rate limiting
limit_req_zone $binary_remote_addr zone=mcp_connect:10m rate=5r/m;
limit_req_zone $binary_remote_addr zone=mcp_call:10m rate=30r/m;
limit_conn_zone $binary_remote_addr zone=mcp_conn:10m;
 
# Upstream
upstream mcp_backend {
    server mcp-server:8080;
    keepalive 16;
}
 
# Map for blocking suspicious user 代理
map $http_user_agent $blocked_agent {
    default 0;
    "~*scanner"  1;
    "~*nikto"    1;
    "~*sqlmap"   1;
    "~*nmap"     1;
    ""           1;
}
 
server {
    listen 443 ssl http2;
    server_name mcp.example.com;
 
    # TLS Configuration
    ssl_certificate     /etc/nginx/certs/server.crt;
    ssl_certificate_key /etc/nginx/certs/server.key;
    ssl_protocols       TLSv1.3;
    ssl_ciphers         ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384;
    ssl_prefer_server_ciphers on;
    ssl_session_timeout 1d;
    ssl_session_cache shared:SSL:10m;
    ssl_session_tickets off;
 
    # OCSP stapling
    ssl_stapling on;
    ssl_stapling_verify on;
 
    # Optional: mTLS
    # ssl_client_certificate /etc/nginx/certs/ca.crt;
    # ssl_verify_client on;
 
    # Connection limits
    limit_conn mcp_conn 10;
 
    # Block suspicious user 代理
    if ($blocked_agent) {
        return 403;
    }
 
    # 安全 headers
    add_header X-Content-Type-Options "nosniff" always;
    add_header X-Frame-Options "DENY" always;
    add_header X-XSS-Protection "1; mode=block" always;
    add_header Strict-Transport-安全 "max-age=63072000; includeSubDomains; preload" always;
    add_header Content-安全-Policy "default-src 'none'" always;
    add_header Referrer-Policy "no-referrer" always;
 
    # Hide server version
    server_tokens off;
 
    # Request body limits
    client_max_body_size 1m;
    client_body_timeout 10s;
    client_header_timeout 10s;
 
    # Health check (no auth, no rate limit)
    location /health {
        proxy_pass http://mcp_backend/health;
        access_log off;
        limit_req zone=mcp_connect burst=10 nodelay;
    }
 
    # SSE connection endpoint
    location /sse {
        limit_req zone=mcp_connect burst=3 nodelay;
 
        # Validate Authorization header presence
        if ($http_authorization = "") {
            return 401 '{"error": "Authentication required"}';
        }
 
        # SSE proxy configuration
        proxy_pass http://mcp_backend/sse;
        proxy_http_version 1.1;
        proxy_set_header Connection "";
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
 
        # SSE-specific timeouts
        proxy_buffering off;
        proxy_cache off;
        proxy_read_timeout 86400s;
        proxy_send_timeout 86400s;
        chunked_transfer_encoding on;
    }
 
    # Tool call endpoint
    location /messages/ {
        limit_req zone=mcp_call burst=10 nodelay;
 
        # Validate Authorization header
        if ($http_authorization = "") {
            return 401 '{"error": "Authentication required"}';
        }
 
        # Validate Content-Type
        if ($content_type !~ "application/json") {
            return 415 '{"error": "Content-Type must be application/json"}';
        }
 
        proxy_pass http://mcp_backend/messages/;
        proxy_http_version 1.1;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Request-ID $request_id;
 
        # Timeout for tool execution
        proxy_read_timeout 60s;
        proxy_send_timeout 10s;
    }
 
    # Block everything else
    location / {
        return 404 '{"error": "Not found"}';
    }
 
    # Custom error pages
    error_page 429 = @rate_limited;
    location @rate_limited {
        default_type application/json;
        return 429 '{"error": "Rate limit exceeded", "retry_after": 60}';
    }
 
    # Structured access log
    log_format mcp_json escape=json
        '{'
            '"time": "$time_iso8601",'
            '"remote_addr": "$remote_addr",'
            '"request_method": "$request_method",'
            '"request_uri": "$request_uri",'
            '"status": $status,'
            '"body_bytes_sent": $body_bytes_sent,'
            '"request_time": $request_time,'
            '"http_user_agent": "$http_user_agent",'
            '"request_id": "$request_id",'
            '"ssl_client_s_dn": "$ssl_client_s_dn"'
        '}';
 
    access_log /var/log/nginx/mcp-access.json mcp_json;
    error_log /var/log/nginx/mcp-error.log warn;
}
 
# Redirect HTTP to HTTPS
server {
    listen 80;
    server_name mcp.example.com;
    return 301 https://$server_name$request_uri;
}

"""
MCP server structured logging configuration.
Outputs JSON logs suitable for log aggregation (ELK, Loki, Datadog).
"""
 
import json
import logging
import sys
import time
import uuid
from datetime import datetime, timezone
from contextvars import ContextVar
 
# Context variables for request tracing
request_id: ContextVar[str] = ContextVar("request_id", default="")
session_id: ContextVar[str] = ContextVar("session_id", default="")
client_identity: ContextVar[str] = ContextVar("client_identity", default="")
 
 
class MCPStructuredFormatter(logging.Formatter):
    """JSON formatter for MCP server logs."""
 
    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
        }
 
        # Add context variables
        rid = request_id.get("")
        if rid:
            log_entry["request_id"] = rid
        sid = session_id.get("")
        if sid:
            log_entry["session_id"] = sid
        cid = client_identity.get("")
        if cid:
            log_entry["client"] = cid
 
        # Add extra fields
        if hasattr(record, "tool_name"):
            log_entry["tool_name"] = record.tool_name
        if hasattr(record, "duration_ms"):
            log_entry["duration_ms"] = record.duration_ms
        if hasattr(record, "token_count"):
            log_entry["token_count"] = record.token_count
        if hasattr(record, "security_event"):
            log_entry["security_event"] = record.security_event
 
        # Add exception info
        if record.exc_info and record.exc_info[1]:
            log_entry["exception"] = {
                "type": type(record.exc_info[1]).__name__,
                "message": str(record.exc_info[1]),
            }
 
        return json.dumps(log_entry)
 
 
def configure_logging(log_level: str = "INFO"):
    """Configure structured logging for MCP server."""
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(MCPStructuredFormatter())
 
    root_logger = logging.getLogger()
    root_logger.setLevel(getattr(logging, log_level.upper()))
    root_logger.addHandler(handler)
 
    # 安全-specific logger
    security_logger = logging.getLogger("mcp.安全")
    security_handler = logging.FileHandler("/var/log/mcp/安全.json")
    security_handler.setFormatter(MCPStructuredFormatter())
    security_logger.addHandler(security_handler)
 
    return root_logger
 
 
# Tool call audit logger
class ToolCallAuditor:
    """Logs all MCP tool calls for audit trail."""
 
    def __init__(self):
        self.logger = logging.getLogger("mcp.audit")
 
    def log_tool_call(self, tool_name: str, arguments: dict,
                      result_summary: str, duration_ms: float,
                      status: str = "success"):
        """Log a complete 工具呼叫 with sanitized parameters."""
        # Sanitize arguments (remove potential secrets)
        safe_args = {}
        for key, value in arguments.items():
            if any(s in key.lower() for s in ["password", "secret", "符元", "key"]):
                safe_args[key] = "***REDACTED***"
            elif isinstance(value, str) and len(value) > 500:
                safe_args[key] = value[:500] + f"...[truncated, {len(value)} chars total]"
            else:
                safe_args[key] = value
 
        self.logger.info(
            "Tool call: %s",
            json.dumps({
                "event": "tool_call",
                "tool": tool_name,
                "arguments": safe_args,
                "result_summary": result_summary[:200],
                "duration_ms": round(duration_ms, 2),
                "status": status,
                "session_id": session_id.get(""),
                "client": client_identity.get(""),
                "request_id": request_id.get(""),
            }),
        )

"""
Prometheus metrics for MCP server 監控.
"""
 
from prometheus_client import (
    Counter, Histogram, Gauge, Info, generate_latest, CONTENT_TYPE_LATEST
)
from starlette.requests import Request
from starlette.responses import Response
 
# Server info
mcp_server_info = Info("mcp_server", "MCP server information")
mcp_server_info.info({
    "version": "1.0.0",
    "sdk_version": "1.26.0",
    "transport": "http_sse",
})
 
# Tool call metrics
tool_calls_total = Counter(
    "mcp_tool_calls_total",
    "Total number of MCP tool calls",
    ["tool_name", "status"],
)
 
tool_call_duration = Histogram(
    "mcp_tool_call_duration_seconds",
    "Duration of MCP tool calls",
    ["tool_name"],
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0],
)
 
tool_call_tokens = Histogram(
    "mcp_tool_call_tokens",
    "Token count per 工具呼叫 (輸出)",
    ["tool_name"],
    buckets=[100, 500, 1000, 5000, 10000, 50000, 100000],
)
 
# Session metrics
active_sessions = Gauge(
    "mcp_active_sessions",
    "Number of active MCP sessions",
)
 
session_duration = Histogram(
    "mcp_session_duration_seconds",
    "Duration of MCP sessions",
    buckets=[60, 300, 600, 1800, 3600, 7200],
)
 
# 安全 metrics
auth_attempts_total = Counter(
    "mcp_auth_attempts_total",
    "Authentication attempts",
    ["status"],  # success, failure, expired, rate_limited
)
 
security_events_total = Counter(
    "mcp_security_events_total",
    "安全 events detected",
    ["event_type", "severity"],
)
 
injection_attempts_total = Counter(
    "mcp_injection_attempts_total",
    "Injection attempts detected",
    ["injection_type"],  # command, path_traversal, sql, prompt
)
 
# Budget metrics
token_spending = Counter(
    "mcp_token_spending_total",
    "Total 符元 consumed",
    ["server_name", "operation"],
)
 
cost_spending = Counter(
    "mcp_cost_usd_total",
    "Total cost in USD",
    ["server_name"],
)
 
sampling_requests_total = Counter(
    "mcp_sampling_requests_total",
    "MCP sampling requests",
    ["server_name", "status"],
)
 
 
async def metrics_endpoint(request: Request) -> Response:
    """Prometheus metrics endpoint."""
    return Response(
        content=generate_latest(),
        media_type=CONTENT_TYPE_LATEST,
    )

# Kubernetes NetworkPolicy for MCP server pods
# Restricts all traffic to explicitly allowed flows
 
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: mcp-server-network-policy
  namespace: mcp
spec:
  podSelector:
    matchLabels:
      app: mcp-server
  policyTypes:
    - Ingress
    - Egress
 
  ingress:
    # Allow traffic only from nginx proxy
    - from:
        - podSelector:
            matchLabels:
              app: mcp-nginx-proxy
      ports:
        - port: 8080
          protocol: TCP
 
    # Allow Prometheus scraping
    - from:
        - podSelector:
            matchLabels:
              app: prometheus
      ports:
        - port: 9090
          protocol: TCP
 
  egress:
    # Allow DNS resolution
    - to:
        - namespaceSelector: {}
          podSelector:
            matchLabels:
              k8s-app: kube-dns
      ports:
        - port: 53
          protocol: UDP
        - port: 53
          protocol: TCP
 
    # Allow access to internal 資料庫 (if needed)
    - to:
        - podSelector:
            matchLabels:
              app: postgresql
      ports:
        - port: 5432
          protocol: TCP
 
    # DENY all other egress (no internet access)

# docker-compose network equivalent using iptables
# Apply with: iptables-restore < mcp-iptables.rules
 
# mcp-iptables.rules
*filter
:INPUT DROP [0:0]
:FORWARD DROP [0:0]
:OUTPUT DROP [0:0]
 
# Allow loopback
-A INPUT -i lo -j ACCEPT
-A OUTPUT -o lo -j ACCEPT
 
# Allow established connections
-A INPUT -m conntrack --ctstate ESTABLISHED,RELATED -j ACCEPT
-A OUTPUT -m conntrack --ctstate ESTABLISHED,RELATED -j ACCEPT
 
# Allow nginx -> mcp-server on port 8080
-A INPUT -s 172.18.0.0/16 -p tcp --dport 8080 -j ACCEPT
 
# Allow DNS resolution
-A OUTPUT -p udp --dport 53 -j ACCEPT
-A OUTPUT -p tcp --dport 53 -j ACCEPT
 
# Allow access to internal 資料庫
-A OUTPUT -d 172.18.0.0/16 -p tcp --dport 5432 -j ACCEPT
 
# Log and drop everything else
-A INPUT -j LOG --log-prefix "MCP-DROP-IN: " --log-level 4
-A INPUT -j DROP
-A OUTPUT -j LOG --log-prefix "MCP-DROP-OUT: " --log-level 4
-A OUTPUT -j DROP
 
COMMIT

# Prometheus alerting rules for MCP server 監控
# prometheus-alerts.yaml
 
groups:
  - name: mcp_security
    rules:
      - alert: MCPAuthFailureSpike
        expr: rate(mcp_auth_attempts_total{status="failure"}[5m]) > 1
        for: 2m
        labels:
          severity: high
        annotations:
          summary: "High rate of MCP 認證 failures"
          description: "More than 1 auth failure per second for 2+ minutes"
 
      - alert: MCPInjectionAttempt
        expr: increase(mcp_injection_attempts_total[5m]) > 0
        labels:
          severity: critical
        annotations:
          summary: "Injection attempt detected on MCP server"
          description: "{{ $labels.injection_type }} injection detected"
 
      - alert: MCPBudgetExceeded
        expr: mcp_cost_usd_total > 100
        labels:
          severity: high
        annotations:
          summary: "MCP daily spending limit exceeded"
 
      - alert: MCPSamplingAbuse
        expr: rate(mcp_sampling_requests_total[5m]) > 0.5
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Excessive MCP sampling requests (possible DoW attack)"
 
      - alert: MCPServerDown
        expr: up{job="mcp-server"} == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "MCP server is down"