远程服务端企业级认证授权集成实战

import { createRemoteJWKSet, jwtVerify, JWTPayload } from 'jose';
import { Request, Response, NextFunction } from 'express';
import { LRUCache } from 'lru-cache';

// JWKS 缓存配置：5分钟 TTL，防止频繁请求 IdP
const jwksCache = new LRUCache<string, ReturnType<typeof createRemoteJWKSet>>({
  max: 10,
  ttl: 5 * 60 * 1000, // 5 minutes
});

interface MCPAuthConfig {
  issuer: string;
  audience: string;
  jwksUri: string;
  requiredScopes: string[];
  maxTokenAge: string;
}

interface AuthenticatedRequest extends Request {
  auth?: {
    sub: string;
    scopes: string[];
    clientId: string;
    payload: JWTPayload;
  };
}

// JWKS 获取，带 miss-then-refetch 策略
function getJWKS(jwksUri: string, forceRefresh = false) {
  const cacheKey = jwksUri;
  if (!forceRefresh && jwksCache.has(cacheKey)) {
    return jwksCache.get(cacheKey)!;
  }
  const jwks = createRemoteJWKSet(new URL(jwksUri));
  jwksCache.set(cacheKey, jwks);
  return jwks;
}

export function createMCPAuthMiddleware(config: MCPAuthConfig) {
  // 刷新频率限制：每分钟最多 1 次强制刷新
  let lastForceRefresh = 0;
  const REFRESH_COOLDOWN = 60_000;

  return async (req: AuthenticatedRequest, res: Response, next: NextFunction) => {
    const authHeader = req.headers.authorization;
    if (!authHeader?.startsWith('Bearer ')) {
      return res.status(401).json({
        error: 'missing_token',
        error_description: 'Authorization header with Bearer token required',
      });
    }

    const token = authHeader.slice(7);

    try {
      let jwks = getJWKS(config.jwksUri);
      let result;

      try {
        result = await jwtVerify(token, jwks, {
          issuer: config.issuer,
          audience: config.audience,
          maxTokenAge: config.maxTokenAge,
        });
      } catch (err: any) {
        // kid 未匹配时尝试刷新 JWKS（密钥轮转场景）
        if (err.code === 'ERR_JWKS_NO_MATCHING_KEY') {
          const now = Date.now();
          if (now - lastForceRefresh > REFRESH_COOLDOWN) {
            lastForceRefresh = now;
            jwks = getJWKS(config.jwksUri, true);
            result = await jwtVerify(token, jwks, {
              issuer: config.issuer,
              audience: config.audience,
              maxTokenAge: config.maxTokenAge,
            });
          } else {
            throw err; // 冷却期内不重试
          }
        } else {
          throw err;
        }
      }

      // Scope 校验：每个请求都必须校验，不只是登录时
      const tokenScopes = (result.payload.scope as string || '').split(' ');
      const hasRequiredScopes = config.requiredScopes.every(
        (s) => tokenScopes.includes(s)
      );
      if (!hasRequiredScopes) {
        return res.status(403).json({
          error: 'insufficient_scope',
          error_description: `Required scopes: ${config.requiredScopes.join(', ')}`,
        });
      }

      req.auth = {
        sub: result.payload.sub!,
        scopes: tokenScopes,
        clientId: result.payload.azp as string || result.payload.client_id as string || '',
        payload: result.payload,
      };

      next();
    } catch (error: any) {
      return res.status(401).json({
        error: 'invalid_token',
        error_description: error.message,
      });
    }
  };
}

// 使用示例
const mcpAuth = createMCPAuthMiddleware({
  issuer: 'https://login.microsoftonline.com/{tenant-id}/v2.0',
  audience: 'api://mcp-server-production',
  jwksUri: 'https://login.microsoftonline.com/{tenant-id}/discovery/v2.0/keys',
  requiredScopes: ['mcp.tools.execute', 'mcp.resources.read'],
  maxTokenAge: '15m',
});

// Azure AD App Registration 配置
const azureADConfig = {
  // App Registration 中定义 MCP API 权限
  api: {
    oauth2PermissionScopes: [
      {
        id: 'generated-uuid-1',
        value: 'mcp.tools.execute',
        type: 'User',
        adminConsentDescription: 'Execute MCP tools on behalf of the user',
        adminConsentDisplayName: 'Execute MCP Tools',
      },
      {
        id: 'generated-uuid-2',
        value: 'mcp.resources.read',
        type: 'User',
        adminConsentDescription: 'Read MCP resources on behalf of the user',
        adminConsentDisplayName: 'Read MCP Resources',
      },
    ],
  },
  // Federated Credential 配置（用于 MCP Client 注册）
  federatedCredentials: {
    name: 'mcp-client-federation',
    issuer: 'https://mcp-client.example.com',
    subject: 'mcp-client-service-principal',
    audiences: ['api://mcp-server-production'],
  },
};

import time
from functools import lru_cache
from typing import Optional

import httpx
from fastapi import Depends, HTTPException, Security
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import jwt, JWTError, ExpiredSignatureError
from pydantic import BaseModel

security = HTTPBearer()


class JWKSManager:
    """JWKS 管理器：带缓存和自动刷新"""

    def __init__(self, jwks_uri: str, cache_ttl: int = 300):
        self.jwks_uri = jwks_uri
        self.cache_ttl = cache_ttl
        self._keys: dict = {}
        self._last_fetch: float = 0
        self._refresh_cooldown: float = 60  # 最小刷新间隔
        self._last_forced_refresh: float = 0

    async def get_signing_keys(self, force_refresh: bool = False) -> dict:
        now = time.time()

        if force_refresh:
            if now - self._last_forced_refresh < self._refresh_cooldown:
                return self._keys  # 冷却期内返回缓存
            self._last_forced_refresh = now

        if not force_refresh and self._keys and (now - self._last_fetch < self.cache_ttl):
            return self._keys

        async with httpx.AsyncClient() as client:
            resp = await client.get(self.jwks_uri, timeout=10)
            resp.raise_for_status()
            jwks_data = resp.json()

        self._keys = {
            key["kid"]: key for key in jwks_data.get("keys", [])
        }
        self._last_fetch = now
        return self._keys

    async def get_key_by_kid(self, kid: str) -> Optional[dict]:
        keys = await self.get_signing_keys()
        if kid not in keys:
            # kid 未命中，尝试强制刷新（密钥轮转场景）
            keys = await self.get_signing_keys(force_refresh=True)
        return keys.get(kid)


class MCPTokenPayload(BaseModel):
    sub: str
    scopes: list[str]
    client_id: str
    exp: int
    iss: str
    aud: str


class MCPAuthVerifier:
    """MCP Server Token 验证器"""

    def __init__(
        self,
        issuer: str,
        audience: str,
        jwks_uri: str,
        required_scopes: list[str],
    ):
        self.issuer = issuer
        self.audience = audience
        self.required_scopes = required_scopes
        self.jwks_manager = JWKSManager(jwks_uri)

    async def verify(
        self, credentials: HTTPAuthorizationCredentials = Security(security)
    ) -> MCPTokenPayload:
        token = credentials.credentials

        # 解码 Header 获取 kid
        try:
            unverified_header = jwt.get_unverified_header(token)
        except JWTError:
            raise HTTPException(status_code=401, detail="Invalid token format")

        kid = unverified_header.get("kid")
        if not kid:
            raise HTTPException(status_code=401, detail="Token missing kid header")

        # 获取对应的签名密钥
        key = await self.jwks_manager.get_key_by_kid(kid)
        if not key:
            raise HTTPException(status_code=401, detail="Unknown signing key")

        # 验证 Token
        try:
            payload = jwt.decode(
                token,
                key,
                algorithms=["RS256"],
                audience=self.audience,
                issuer=self.issuer,
                options={"verify_exp": True, "leeway": 30},
            )
        except ExpiredSignatureError:
            raise HTTPException(status_code=401, detail="Token expired")
        except JWTError as e:
            raise HTTPException(status_code=401, detail=f"Token validation failed: {e}")

        # Scope 校验
        token_scopes = payload.get("scope", "").split()
        if not all(s in token_scopes for s in self.required_scopes):
            raise HTTPException(
                status_code=403,
                detail=f"Insufficient scope. Required: {self.required_scopes}",
            )

        return MCPTokenPayload(
            sub=payload["sub"],
            scopes=token_scopes,
            client_id=payload.get("azp", payload.get("client_id", "")),
            exp=payload["exp"],
            iss=payload["iss"],
            aud=payload["aud"],
        )


# FastAPI 集成示例
mcp_auth = MCPAuthVerifier(
    issuer="https://login.microsoftonline.com/{tenant-id}/v2.0",
    audience="api://mcp-server-production",
    jwks_uri="https://login.microsoftonline.com/{tenant-id}/discovery/v2.0/keys",
    required_scopes=["mcp.tools.execute"],
)

import { ConfidentialClientApplication } from '@azure/msal-node';

interface OBOConfig {
  clientId: string;
  clientSecret: string;
  tenantId: string;
  downstreamScopes: string[];
}

class MCPOnBehalfOfHandler {
  private msalClient: ConfidentialClientApplication;
  private config: OBOConfig;

  constructor(config: OBOConfig) {
    this.config = config;
    this.msalClient = new ConfidentialClientApplication({
      auth: {
        clientId: config.clientId,
        clientSecret: config.clientSecret,
        authority: `https://login.microsoftonline.com/${config.tenantId}`,
      },
    });
  }

  /**
   * 使用用户的 access_token 交换下游服务的委托 token
   * 权限边界：下游 token 的权限 <= 用户原始 token 的权限
   */
  async acquireTokenOnBehalfOf(userAccessToken: string): Promise<string> {
    const result = await this.msalClient.acquireTokenOnBehalfOf({
      oboAssertion: userAccessToken,
      scopes: this.config.downstreamScopes,
    });

    if (!result?.accessToken) {
      throw new Error('OBO token acquisition failed - user may lack consent');
    }

    return result.accessToken;
  }

  /**
   * 在 MCP Tool 执行中使用 OBO
   */
  async executeWithDelegation(
    userToken: string,
    toolFn: (delegatedToken: string) => Promise<any>
  ) {
    const delegatedToken = await this.acquireTokenOnBehalfOf(userToken);
    return toolFn(delegatedToken);
  }
}

// MCP Tool 注册中使用 OBO
const oboHandler = new MCPOnBehalfOfHandler({
  clientId: process.env.MCP_CLIENT_ID!,
  clientSecret: process.env.MCP_CLIENT_SECRET!,
  tenantId: process.env.AZURE_TENANT_ID!,
  downstreamScopes: ['https://management.azure.com/.default'],
});

// Tool 实现
async function queryBillingTool(params: any, context: { userToken: string }) {
  return oboHandler.executeWithDelegation(context.userToken, async (token) => {
    const response = await fetch(
      'https://management.azure.com/subscriptions/{sub-id}/providers/Microsoft.CostManagement/query?api-version=2023-11-01',
      {
        method: 'POST',
        headers: {
          Authorization: `Bearer ${token}`,
          'Content-Type': 'application/json',
        },
        body: JSON.stringify(params.query),
      }
    );
    return response.json();
  });
}

# Nginx 配置：mTLS for MCP Server
server {
    listen 443 ssl;
    server_name mcp.enterprise.com;

    # 服务端证书
    ssl_certificate /etc/nginx/certs/server.crt;
    ssl_certificate_key /etc/nginx/certs/server.key;

    # 客户端证书验证
    ssl_client_certificate /etc/nginx/certs/ca.crt;
    ssl_verify_client on;
    ssl_verify_depth 2;

    # TLS 1.3 强制
    ssl_protocols TLSv1.3;
    ssl_ciphers TLS_AES_256_GCM_SHA384:TLS_CHACHA20_POLY1305_SHA256;

    # 传递客户端证书信息到上游
    location /mcp {
        proxy_pass http://mcp-server:3000;
        proxy_set_header X-Client-Cert-DN $ssl_client_s_dn;
        proxy_set_header X-Client-Cert-Fingerprint $ssl_client_fingerprint;
    }
}

// MCP Server 暴露的 /.well-known/oauth-protected-resource 端点
app.get('/.well-known/oauth-protected-resource', (req, res) => {
  res.json({
    resource: 'https://mcp.enterprise.com',
    authorization_servers: [
      'https://login.microsoftonline.com/{tenant-id}/v2.0'
    ],
    scopes_supported: [
      'mcp.tools.execute',
      'mcp.resources.read',
      'mcp.prompts.use',
    ],
    bearer_methods_supported: ['header'],
    resource_documentation: 'https://docs.enterprise.com/mcp-api',
  });
});

version: "3.9"

services:
  mcp-server:
    image: enterprise/mcp-server:latest
    environment:
      - NODE_ENV=production
      - MCP_AUTH_ISSUER=https://login.microsoftonline.com/{tenant-id}/v2.0
      - MCP_AUTH_AUDIENCE=api://mcp-server-production
      - MCP_AUTH_JWKS_URI=https://login.microsoftonline.com/{tenant-id}/discovery/v2.0/keys
      - MCP_REQUIRED_SCOPES=mcp.tools.execute,mcp.resources.read
      - MCP_TOKEN_MAX_AGE=15m
      - MCP_RATE_LIMIT_WINDOW=60000
      - MCP_RATE_LIMIT_MAX=100
      - LOG_LEVEL=info
      - LOG_FORMAT=json
    ports:
      - "3000:3000"
    deploy:
      replicas: 3
      resources:
        limits:
          cpus: "1.0"
          memory: 512M
        reservations:
          cpus: "0.25"
          memory: 128M
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:3000/health"]
      interval: 30s
      timeout: 5s
      retries: 3
    networks:
      - mcp-internal

  nginx:
    image: nginx:alpine
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
      - ./certs:/etc/nginx/certs:ro
    ports:
      - "443:443"
    depends_on:
      - mcp-server
    networks:
      - mcp-internal
      - mcp-external

  # 结构化日志收集
  fluent-bit:
    image: fluent/fluent-bit:latest
    volumes:
      - ./fluent-bit/fluent-bit.conf:/fluent-bit/etc/fluent-bit.conf:ro
    depends_on:
      - mcp-server
    networks:
      - mcp-internal

networks:
  mcp-internal:
    driver: bridge
    internal: true
  mcp-external:
    driver: bridge