在 Gateway 日志中检测 MCP 流量

组织可能缺乏对 Model Context Protocol (MCP) 流量的可见性，这可能会允许员工连接到 IT 监管之外的远程 MCP 服务器。这些连接会带来敏感内部数据和凭证外泄、工具注入攻击或软件供应链的风险。

作为 IT 管理员，您希望识别影子 MCP 流量以防止未经授权的数据外泄，同时仍然支持受监管的使用案例。在本教程中，您将使用 Cloudflare GraphQL Analytics API 扫描 Gateway HTTP 日志以查找 MCP 流量模式，创建检测 MCP JSON-RPC 方法的 DLP 配置文件，并对流量进行分类以区分发送到授权 MCP 服务器门户的流量与发送到“影子”远程 MCP 服务器的流量。

前提条件

拥有 Zero Trust 组织的 Cloudflare 账户
已启用 HTTP 过滤并正在主动代理用户流量的 Gateway
具有以下权限的 API 令牌：
- 账户级 Zero Trust: Read
- 账户级 DLP: Write
- 账户级 Gateway: Write
您的 Cloudflare 账户 ID（可在 Cloudflare 仪表板 ↗的 Account Home（账户主页） 下找到）
熟悉 GraphQL Analytics API 查询
具备 TypeScript 和 REST API 的实际操作知识

1. 查看 Gateway HTTP 数据集

GraphQL Analytics API 中的 gatewayHttpRequestsAdaptiveGroups 数据集提供了聚合的 Gateway HTTP 日志数据。使用该数据集查询与 MCP 相关的流量模式：

维度：httpHost、httpRequestURI、action、users、dlpProfiles
时间范围：最多 30 天的历史数据
分组：按维度值聚合结果
过滤：支持 OR、AND 和 like 运算符

2. 构建 MCP 检测查询

MCP 流量可以通过三个信号进行识别：

域名模式：包含 mcp 的主机名（例如 mcp.datadog.com）
URL 路径：标准 MCP 端点，例如 /mcp、/mcp/sse 和 /sse
DLP 匹配：请求体中的 JSON-RPC 方法（在后面的步骤中介绍）

以下 GraphQL 查询会扫描 Gateway 日志以查找前两个信号：

const query = `
  query MCPTrafficScan($accountTag: string, $since: string, $until: string) {
    viewer {
      accounts(filter: { accountTag: $accountTag }) {
        gatewayHttpRequestsAdaptiveGroups(
          filter: {
            datetime_geq: $since
            datetime_leq: $until
            OR: [
              { httpHost_like: "%mcp%" }
              { httpRequestURI_like: "%/mcp%" }
              { httpRequestURI_like: "%/sse%" }
            ]
          }
          limit: 10000
        ) {
          dimensions {
            httpHost
            action
            users
          }
          count
        }
      }
    }
  }
`;

const variables = {
	accountTag: "<YOUR_ACCOUNT_ID>",
	since: "<START_DATE>", // ISO-8601 format, for example 2025-03-08T00:00:00Z
	until: "<END_DATE>", // Up to 30 days after start date
};

const response = await fetch("https://api.cloudflare.com/client/v4/graphql", {
	method: "POST",
	headers: {
		Authorization: `Bearer ${apiToken}`,
		"Content-Type": "application/json",
	},
	body: JSON.stringify({ query, variables }),
});

const data = await response.json();
const groups =
	data.data?.viewer?.accounts?.[0]?.gatewayHttpRequestsAdaptiveGroups || [];

const query = `
  query MCPTrafficScan($accountTag: string, $since: string, $until: string) {
    viewer {
      accounts(filter: { accountTag: $accountTag }) {
        gatewayHttpRequestsAdaptiveGroups(
          filter: {
            datetime_geq: $since
            datetime_leq: $until
            OR: [
              { httpHost_like: "%mcp%" }
              { httpRequestURI_like: "%/mcp%" }
              { httpRequestURI_like: "%/sse%" }
            ]
          }
          limit: 10000
        ) {
          dimensions {
            httpHost
            action
            users
          }
          count
        }
      }
    }
  }
`;

const variables = {
	accountTag: "<YOUR_ACCOUNT_ID>",
	since: "<START_DATE>", // ISO-8601 format, for example 2025-03-08T00:00:00Z
	until: "<END_DATE>", // Up to 30 days after start date
};

const response = await fetch("https://api.cloudflare.com/client/v4/graphql", {
	method: "POST",
	headers: {
		Authorization: `Bearer ${apiToken}`,
		"Content-Type": "application/json",
	},
	body: JSON.stringify({ query, variables }),
});

const data = await response.json();
const groups =
	data.data?.viewer?.accounts?.[0]?.gatewayHttpRequestsAdaptiveGroups || [];

将 <YOUR_ACCOUNT_ID> 替换为您的 Cloudflare 账户 ID。将 <START_DATE> 和 <END_DATE> 替换为涵盖您所需时间范围（最多 30 天）的 ISO-8601 时间戳。

3. 处理查询结果

响应中的每个组代表特定 httpHost 和 action 组合的聚合流量。解析结果以识别未被阻止的 MCP 连接：

const hits = groups.map((group) => ({
	domain: group.dimensions.httpHost,
	requestCount: group.count,
	users: group.dimensions.users || [],
	actions: {
		allowed: group.dimensions.action === "allow" ? group.count : 0,
		blocked: group.dimensions.action === "block" ? group.count : 0,
	},
}));

const totalMCPRequests = hits.reduce((sum, h) => sum + h.requestCount, 0);
const unblockedHits = hits.filter((h) => h.actions.allowed > 0);

console.log(`Found ${totalMCPRequests} MCP requests`);
console.log(`${unblockedHits.length} destinations are unblocked`);

interface MCPTrafficHit {
	domain: string;
	requestCount: number;
	users: string[];
	actions: {
		allowed: number;
		blocked: number;
	};
}

const hits: MCPTrafficHit[] = groups.map((group: any) => ({
	domain: group.dimensions.httpHost,
	requestCount: group.count,
	users: group.dimensions.users || [],
	actions: {
		allowed: group.dimensions.action === "allow" ? group.count : 0,
		blocked: group.dimensions.action === "block" ? group.count : 0,
	},
}));

const totalMCPRequests = hits.reduce((sum, h) => sum + h.requestCount, 0);
const unblockedHits = hits.filter((h) => h.actions.allowed > 0);

console.log(`Found ${totalMCPRequests} MCP requests`);
console.log(`${unblockedHits.length} destinations are unblocked`);

来自数据的关键洞察：

未阻止的流量（action = allow）- 需要调查或阻止的活动 MCP 连接
已阻止的流量（action = block）- 您现有的策略正在发挥作用
用户归因 - 指示哪些员工正在连接到 MCP 服务器

4. 创建用于检测 MCP JSON-RPC 的 DLP 配置文件

Gateway HTTP 策略可以匹配域和 URL 路径，但它们无法检查请求体。DLP 配置文件会扫描 POST 请求体内容以查找模式，这对于检测影子 MCP 非常有用，因为 MCP 在 HTTP 上使用 JSON-RPC 且具有几个可检测的特征。

每个 MCP 请求都包含一个 "method" 字段：

{
	"jsonrpc": "2.0",
	"id": 1,
	"method": "tools/call",
	"params": { "name": "read_file", "arguments": { "path": "/etc/passwd" } }
}

攻击者可能会在非标准域（例如 internal-tools.company.com/api/assistant）上运行 MCP 服务器，而不会触发基于域或基于路径的规则。您可以使用 DLP 扫描 POST 请求体中的 "method": "tools/call" 和其他 MCP 特定的模式，从而为 MCP 流量提供更强大的保护。

查看 DLP 限制

在构建检测模式之前，请注意以下 DLP 限制：

正则表达式语法 — Rust 正则表达式（与 JavaScript 和 PCRE 略有不同）
扫描深度 — 仅限请求体的前 1,024 个字节
仅限 POST — DLP 仅扫描 POST 请求
性能 — 正则表达式模式必须高效，以避免灾难性回溯

构建 MCP 检测模式

可以在 JSON-RPC 方法字段中找到 MCP 指示符。以下正则表达式模式涵盖了核心 MCP 协议方法：

const DLP_REGEX_PATTERNS = [
	{
		name: "MCP Initialize Method",
		regex: '"method"\\s{0,5}:\\s{0,5}"initialize"',
	},
	{
		name: "MCP Tools Call",
		regex: '"method"\\s{0,5}:\\s{0,5}"tools/call"',
	},
	{
		name: "MCP Tools List",
		regex: '"method"\\s{0,5}:\\s{0,5}"tools/list"',
	},
	{
		name: "MCP Resources Read",
		regex: '"method"\\s{0,5}:\\s{0,5}"resources/read"',
	},
	{
		name: "MCP Resources List",
		regex: '"method"\\s{0,5}:\\s{0,5}"resources/list"',
	},
	{
		name: "MCP Prompts List",
		regex: '"method"\\s{0,5}:\\s{0,5}"prompts/(list|get)"',
	},
	{
		name: "MCP Sampling Create Message",
		regex: '"method"\\s{0,5}:\\s{0,5}"sampling/createMessage"',
	},
	{
		name: "MCP Protocol Version",
		regex: '"protocolVersion"\\s{0,5}:\\s{0,5}"202[4-9]',
	},
	{
		name: "MCP Notifications Initialized",
		regex: '"method"\\s{0,5}:\\s{0,5}"notifications/initialized"',
	},
	{
		name: "MCP Roots List",
		regex: '"method"\\s{0,5}:\\s{0,5}"roots/list"',
	},
];

const DLP_REGEX_PATTERNS = [
	{
		name: "MCP Initialize Method",
		regex: '"method"\\s{0,5}:\\s{0,5}"initialize"',
	},
	{
		name: "MCP Tools Call",
		regex: '"method"\\s{0,5}:\\s{0,5}"tools/call"',
	},
	{
		name: "MCP Tools List",
		regex: '"method"\\s{0,5}:\\s{0,5}"tools/list"',
	},
	{
		name: "MCP Resources Read",
		regex: '"method"\\s{0,5}:\\s{0,5}"resources/read"',
	},
	{
		name: "MCP Resources List",
		regex: '"method"\\s{0,5}:\\s{0,5}"resources/list"',
	},
	{
		name: "MCP Prompts List",
		regex: '"method"\\s{0,5}:\\s{0,5}"prompts/(list|get)"',
	},
	{
		name: "MCP Sampling Create Message",
		regex: '"method"\\s{0,5}:\\s{0,5}"sampling/createMessage"',
	},
	{
		name: "MCP Protocol Version",
		regex: '"protocolVersion"\\s{0,5}:\\s{0,5}"202[4-9]',
	},
	{
		name: "MCP Notifications Initialized",
		regex: '"method"\\s{0,5}:\\s{0,5}"notifications/initialized"',
	},
	{
		name: "MCP Roots List",
		regex: '"method"\\s{0,5}:\\s{0,5}"roots/list"',
	},
];

模式解释：

\\s{0,5} — 允许零到五个空白字符，以处理压缩（minified）和美化（pretty-printed）的 JSON
"method" — 双引号是字面量，因为 JSON 需要它们
"tools/call" — 匹配精确的 MCP 方法名称
202[4-9] — 匹配 2024 到 2029 年的 MCP 协议版本

通过 API 创建 DLP 配置文件

发送 POST 请求以创建一个包含所有检测模式的自定义 DLP 配置文件：

const dlpProfile = {
	name: "MCP-Shield: MCP JSON-RPC Detection",
	description: "Detects MCP protocol JSON-RPC methods in HTTP request bodies.",
	type: "custom",
	entries: DLP_REGEX_PATTERNS.map((p) => ({
		name: p.name,
		enabled: true,
		pattern: {
			regex: p.regex,
			validation: "luhn",
		},
	})),
};

const response = await fetch(
	`https://api.cloudflare.com/client/v4/accounts/${accountId}/gateway/rules`,
	{
		method: "POST",
		headers: {
			Authorization: `Bearer ${apiToken}`,
			"Content-Type": "application/json",
		},
		body: JSON.stringify(dlpRule),
	},
);

const data = await response.json();
if (data.success) {
	console.log(`Created DLP profile: ${data.result.id}`);
}

const dlpProfile = {
	name: "MCP-Shield: MCP JSON-RPC Detection",
	description: "Detects MCP protocol JSON-RPC methods in HTTP request bodies.",
	type: "custom",
	entries: DLP_REGEX_PATTERNS.map((p) => ({
		name: p.name,
		enabled: true,
		pattern: {
			regex: p.regex,
			validation: "luhn",
		},
	})),
};

const response = await fetch(
	`https://api.cloudflare.com/client/v4/accounts/${accountId}/gateway/rules`,
	{
		method: "POST",
		headers: {
			Authorization: `Bearer ${apiToken}`,
			"Content-Type": "application/json",
		},
		body: JSON.stringify(dlpRule),
	},
);

const data = await response.json();
if (data.success) {
	console.log(`Created DLP profile: ${data.result.id}`);
}

将 ${accountId} 替换为您的 Cloudflare 账户 ID，将 ${apiToken} 替换为您的 API 令牌。

在 Gateway 规则中引用该 DLP 配置文件

在 DLP 配置文件存在后，创建一条 Gateway HTTP 策略以阻止匹配该配置文件的请求：

const dlpRule = {
	name: "MCP-Shield: Block MCP JSON-RPC via DLP",
	description: "Blocks requests with MCP JSON-RPC patterns detected by DLP",
	precedence: 85,
	enabled: true,
	action: "block",
	filters: ["http"],
	traffic:
		'any(http.request.body.scan.dlp.profiles[*] == "MCP-Shield: MCP JSON-RPC Detection")',
};

const dlpRule = {
	name: "MCP-Shield: Block MCP JSON-RPC via DLP",
	description: "Blocks requests with MCP JSON-RPC patterns detected by DLP",
	precedence: 85,
	enabled: true,
	action: "block",
	filters: ["http"],
	traffic:
		'any(http.request.body.scan.dlp.profiles[*] == "MCP-Shield: MCP JSON-RPC Detection")',
};

当 DLP 配置文件匹配请求体中的任何正则表达式模式时，将触发此规则。

5. 分类门户流量和影子 MCP 流量

Cloudflare MCP Server Portals 为您组织内经批准的 MCP 访问提供了受监管的基础设施，包括：

受监管的访问 — 由您的 IT 团队管理的集中式 MCP 基础设施
审计跟踪 — 所有通过 Gateway 记录并附带用户归因的 MCP 请求
策略实施 — 自动应用 Zero Trust 策略，包括身份验证和 DLP
批准的工具 — 经安全部门审查的精选 MCP 工具和资源集

分析 Gateway 日志时，区分以下两种类型的 MCP 流量会很有帮助：

流量类型	特征	风险等级	操作
MCP 门户流量	`httpHost` 匹配您的门户域（例如 `mcp.yourcompany.com` 或 `mcp-portal.pages.dev`）	已授权	监控
影子 MCP 流量	`httpHost` 不匹配任何门户域（例如 `mcp.datadog.com`、`api.stripe.com/mcp`）	调查	阻止、重定向或审查

扩展来自处理查询结果的查询处理，通过将主机名与您批准的门户域名列表进行比较来对流量进行分类：

const portalDomains = [
	"mcp.yourcompany.com",
	"mcp-portal.pages.dev",
	"approved-mcp.workers.dev",
];

const results = groups.map((group) => {
	const isPortalTraffic = portalDomains.some((domain) =>
		group.dimensions.httpHost.includes(domain),
	);

	return {
		domain: group.dimensions.httpHost,
		requestCount: group.count,
		users: group.dimensions.users || [],
		trafficType: isPortalTraffic ? "portal" : "shadow",
		riskLevel: isPortalTraffic ? "low" : "high",
	};
});

const portalTraffic = results.filter((r) => r.trafficType === "portal");
const shadowTraffic = results.filter((r) => r.trafficType === "shadow");

console.log("Portal traffic:", portalTraffic);
console.log("Shadow MCP traffic:", shadowTraffic);

const portalDomains = [
	"mcp.yourcompany.com",
	"mcp-portal.pages.dev",
	"approved-mcp.workers.dev",
];

const results = groups.map((group) => {
	const isPortalTraffic = portalDomains.some((domain) =>
		group.dimensions.httpHost.includes(domain),
	);

	return {
		domain: group.dimensions.httpHost,
		requestCount: group.count,
		users: group.dimensions.users || [],
		trafficType: isPortalTraffic ? "portal" : "shadow",
		riskLevel: isPortalTraffic ? "low" : "high",
	};
});

const portalTraffic = results.filter((r) => r.trafficType === "portal");
const shadowTraffic = results.filter((r) => r.trafficType === "shadow");

console.log("Portal traffic:", portalTraffic);
console.log("Shadow MCP traffic:", shadowTraffic);

将 portalDomains 数组替换为您批准的 MCP Server Portals 的实际域。