可逆脱敏管道实战：Python 构建大模型安全网关，敏感信息不裸奔

客服小李在工单系统里选中一条售后记录，请模型起草短信。原文包含手机号、邮箱和订单号：模型需要知道“这两处是同一个客户”，但不需要知道号码本身。更重要的是，工单偶尔会被误粘贴访问令牌、数据库连接串或私钥片段——这类数据不该被替换后继续推理，而应该阻断请求，并由人工轮换凭据。

本文给出一个有意收窄能力的文本安全网关：只接受非流式、纯文本的聊天请求；先阻断凭据，再把手机号和邮箱换成请求内随机占位符；最后按调用方已验证的权限决定是否还原模型文本。它不是“接入一个代理就万事大吉”的方案：图片、附件、OCR、RAG 检索片段、工具调用和流式输出没有同等保护时，必须拒绝，而不是悄悄放行。

本文的目标是让客服起草场景可安全落地，不宣称这是通用 DLP、合规认证或全模态网关。

先明确谁能看到什么

对象或通道	网关行为	原因
工单正文中的手机号、邮箱	同一请求映射为随机 token	模型可保留实体指代与重复关系，不能得到原值
私钥、第三方 Bearer Token、已覆盖格式的 API Key、连接串口令	403 拒绝，记录规则 ID，不记录命中值	凭据不应进入推理上下文；必须轮换或处置
调用网关的 Authorization	只用于网关自身鉴权，绝不转发上游	它是访问控制凭据，不能按“发现即阻断”处理
无 pii:reveal 权限的客服前端	返回 token 化回复	避免 API 响应本身成为越权泄露通道
有 pii:reveal 权限的受控工单后端	在同一请求内还原已知 token	仅供本就被授权查看工单原文的展示路径
模型输出的电话、邮件、付款指令	仅作为文本，不能驱动真实动作	工具执行必须用业务 ID、服务端授权和独立参数校验

这里有两个常被混淆的概念。

• 调用方鉴权回答“谁能调用网关、能否看到原值”。客户端传来的 Authorization 可能是正当凭据，网关必须验证它，但不得透传给模型供应商。
• 业务内容检测回答“工单文本中是否含不应出网的数据”。本例只对 messages[].content 检测；任何未被解析的字段都不在保护承诺内。

占位符映射只存活于单次 HTTP 请求的局部内存：同一请求中相同手机号会变成同一个 token，不同请求一定不同。它不能进入日志、追踪属性、指标标签、数据库或队列。进程内存也可能被调试器或转储读取，所以生产环境仍应限制调试权限、禁用不必要的 core dump，并审查 APM 采样。

场景数据流：客服草稿而不是自动发送

客服页面 → 工单后端（已鉴权） → 安全网关 → 云端模型
                │                   │
                │                   ├─ 凭据命中：拒绝，零次上游调用
                │                   ├─ 手机/邮箱：替换为随机 token
                │                   └─ 上游仅收到 token 化文本
                │
                └─ 有 pii:reveal：显示还原后的“短信草稿”
                   无 pii:reveal：显示 token 化草稿

短信发送：只能由工单后端使用订单/客户业务 ID 再次授权后执行

即使模型说“已向该号码发送”，也只是自然语言。它没有短信工具，也不拥有客户号码；业务系统不能把模型输出中的电话号码当作调用短信服务的参数。

可运行的最小闭环

环境为 Python 3.11+。安装依赖：

pip install 'fastapi>=0.110' 'uvicorn[standard]>=0.27' 'httpx>=0.27' 'pydantic>=2' pytest

保存以下代码为 app.py。示例使用兼容 /v1/chat/completions 的非流式纯文本上游；模型名、上游地址和上游服务密钥只能来自部署配置。调用者只能选择部署方允许的模型，不能提供上游 URL、上游 API Key 或任意请求头。

import hmac
import os
import re
import secrets
from dataclasses import dataclass
from typing import Annotated

import httpx
from fastapi import Depends, FastAPI, Header, HTTPException
from pydantic import BaseModel, ConfigDict, Field

app = FastAPI()

UPSTREAM_URL = os.environ.get("UPSTREAM_URL", "")
UPSTREAM_API_KEY = os.environ.get("UPSTREAM_API_KEY", "")
ALLOWED_MODELS = frozenset(
    model.strip() for model in os.environ.get("ALLOWED_MODELS", "").split(",") if model.strip()
)
# 演示用两种网关调用方；真实生产请替换为 mTLS 或身份提供商签发的短期令牌与 scope。
MASKED_CALLER_TOKEN = os.environ.get("GATEWAY_MASKED_CALLER_TOKEN", "")
REVEAL_CALLER_TOKEN = os.environ.get("GATEWAY_REVEAL_CALLER_TOKEN", "")

# 规则应有版本、负责人和回归样本；这里只是有限字段的示例，不是“识别所有秘密”。
CREDENTIAL_RULES = {
    "private_key_header": re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
    "github_token": re.compile(r"\b(?:ghp|gho|ghu|ghs|ghr)_[A-Za-z0-9]{36,}\b"),
    "live_key": re.compile(r"\b(?:sk_live|rk_live)_[A-Za-z0-9]{24,}\b"),
    "assignment_secret": re.compile(r"(?i)\b(?:password|passwd|api[_-]?key|secret)\s*[:=]\s*[^\s,;]+"),
    "authorization_header": re.compile(r"(?im)^\s*authorization\s*:\s*bearer\s+\S+"),
    "url_password": re.compile(r"(?i)://[^\s/@:]+:[^\s/@]+@[^\s/]+"),
}
PHONE = re.compile(r"(?<!\d)1[3-9]\d{9}(?!\d)")
EMAIL = re.compile(r"(?<![\w.+-])[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}(?![\w.-])")
TOKEN = re.compile(r"\[\[GW_(?:PHONE|EMAIL)_[A-F0-9]{32}\]\]")

class Message(BaseModel):
    model_config = ConfigDict(extra="forbid")
    role: str
    content: str = Field(min_length=1, max_length=16_000)

class RequestBody(BaseModel):
# 因 extra=forbid，stream、tools、图片 content 数组等未实现字段会被 422 拒绝。
    model_config = ConfigDict(extra="forbid")
    model: str = Field(min_length=1, max_length=128)
    messages: list[Message] = Field(min_length=1, max_length=20)
    temperature: float | None = Field(default=None, ge=0, le=2)

@dataclass(frozen=True)
class Principal:
    can_reveal: bool

@dataclass(frozen=True)
class Span:
    start: int
    end: int
    kind: str
    value: str

def authenticate(authorization: Annotated[str | None, Header()] = None) -> Principal:
    """验证调用网关的身份；不把该头转发给上游，也不记录其值。"""
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(401, "gateway_auth_required")
    presented = authorization.removeprefix("Bearer ")
    if MASKED_CALLER_TOKEN and hmac.compare_digest(presented, MASKED_CALLER_TOKEN):
        return Principal(can_reveal=False)
    if REVEAL_CALLER_TOKEN and hmac.compare_digest(presented, REVEAL_CALLER_TOKEN):
        return Principal(can_reveal=True)
    raise HTTPException(403, "gateway_auth_denied")

def detect(text: str) -> list[Span]:
    """凭据先阻断；普通实体候选随后按位置、长度消解重叠。"""
    for rule_id, rule in CREDENTIAL_RULES.items():
        if rule.search(text):
            # 对外不回显命中内容；内部审计仅可记录 rule_id 与请求 ID。
            raise HTTPException(403, f"credential_detected:{rule_id}")
    if "[[GW_" in text:
        raise HTTPException(422, "reserved_token_namespace")

    candidates = [
        Span(match.start(), match.end(), kind, match.group())
        for kind, pattern in (("EMAIL", EMAIL), ("PHONE", PHONE))
        for match in pattern.finditer(text)
    ]
    selected: list[Span] = []
    for span in sorted(candidates, key=lambda item: (item.start, -(item.end - item.start))):
        if not selected or span.start >= selected[-1].end:
            selected.append(span)
    return selected

def transform(text: str, forward: dict[tuple[str, str], str], reverse: dict[str, str]) -> str:
    for span in reversed(detect(text)):
        key = (span.kind, span.value)
        if key not in forward:
            token = f"[[GW_{span.kind}_{secrets.token_hex(16).upper()}]]"
            while token in reverse:  # 理论上极低概率，仍显式保证唯一性。
                token = f"[[GW_{span.kind}_{secrets.token_hex(16).upper()}]]"
            forward[key] = token
            reverse[token] = span.value
        text = text[:span.start] + forward[key] + text[span.end:]
    return text

def restore(text: str, reverse: dict[str, str]) -> str:
    """只还原本请求生成的完整 token；陌生或损坏 token 一律失败关闭。"""
    tokens = TOKEN.findall(text)
    if any(token not in reverse for token in tokens):
        raise HTTPException(502, "unknown_placeholder")
    if "[[GW_" in TOKEN.sub("", text):
        raise HTTPException(502, "malformed_placeholder")
    return TOKEN.sub(lambda match: reverse[match.group()], text)

def safe_response(result: object, reverse: dict[str, str], can_reveal: bool) -> dict:
    """只返回被验证过的响应字段，避免把供应商扩展字段原样反射给客户端。"""
    if not isinstance(result, dict) or not isinstance(result.get("choices"), list):
        raise ValueError("invalid_upstream_response")

    choices = []
    for choice in result["choices"]:
        if not isinstance(choice, dict) or not isinstance(choice.get("message"), dict):
            raise ValueError("invalid_choice")
        upstream_message = choice["message"]
        content = upstream_message.get("content")
        if not isinstance(content, str) or upstream_message.get("tool_calls"):
            raise ValueError("unsupported_upstream_output")
        # 未获授权时保留 token；获授权时才在当前请求内还原。
        visible_content = restore(content, reverse) if can_reveal else content
        choices.append({
            "index": choice.get("index", len(choices)),
            "message": {"role": "assistant", "content": visible_content},
            "finish_reason": choice.get("finish_reason"),
        })
    return {
        "id": result.get("id"),
        "object": "chat.completion",
        "created": result.get("created"),
        "model": result.get("model"),
        "choices": choices,
    }

@app.post("/v1/chat/completions")
async def chat(body: RequestBody, principal: Annotated[Principal, Depends(authenticate)]):
    if not UPSTREAM_URL or not UPSTREAM_API_KEY or not ALLOWED_MODELS:
        raise HTTPException(503, "gateway_not_configured")
    if body.model not in ALLOWED_MODELS:
        raise HTTPException(403, "model_not_allowed")
    if any(message.role not in {"system", "user", "assistant"} for message in body.messages):
        raise HTTPException(422, "unsupported_role")
    if sum(len(message.content) for message in body.messages) > 32_000:
        raise HTTPException(413, "input_too_large")

    # 先完整检测，确认无凭据后才能产生任何出网调用。
    for message in body.messages:
        detect(message.content)

    forward: dict[tuple[str, str], str] = {}
    reverse: dict[str, str] = {}
    outbound = body.model_dump(exclude_none=True)
    for message in outbound["messages"]:
        message["content"] = transform(message["content"], forward, reverse)

    # 上游头由网关固定构造；调用方 Authorization 不在此列表中。
    headers = {"Authorization": f"Bearer {UPSTREAM_API_KEY}", "Content-Type": "application/json"}
    try:
        async with httpx.AsyncClient(
            timeout=httpx.Timeout(30.0, connect=3.0), follow_redirects=False
        ) as client:
            response = await client.post(UPSTREAM_URL, json=outbound, headers=headers)
        response.raise_for_status()
        return safe_response(response.json(), reverse, principal.can_reveal)
    except HTTPException:
        raise
    except (httpx.HTTPError, ValueError, KeyError, TypeError):
        # 不返回上游正文或异常对象，避免错误回显将输入带回客户端/日志。
        raise HTTPException(502, "upstream_or_response_error") from None

启动前注入配置。示例中的两个调用方 token 仅用于本地验证权限路径；生产环境应由密钥管理系统或身份提供商注入，并使用短期令牌、mTLS 或工作负载身份替代长期静态 token。

export UPSTREAM_URL='https://api.openai.com/v1/chat/completions'
export UPSTREAM_API_KEY='由部署系统注入的上游服务密钥'
export ALLOWED_MODELS='部署批准的模型名'
export GATEWAY_MASKED_CALLER_TOKEN='只可查看脱敏草稿的后端令牌'
export GATEWAY_REVEAL_CALLER_TOKEN='可展示工单原值的受控后端令牌'
uvicorn app:app --host 127.0.0.1 --port 8080

无原值查看权限的客服渠道调用：

curl -s http://127.0.0.1:8080/v1/chat/completions \
  -H 'Authorization: Bearer 只可查看脱敏草稿的后端令牌' \
  -H 'Content-Type: application/json' \
  -d '{"model":"部署批准的模型名","messages":[{"role":"user","content":"为订单 A-20260928-17 的客户 13800138000 起草回访短信，抄送 a@example.com。"}]}'

模型实际接收的是类似“客户 [[GW_PHONE_…]]、抄送 [[GW_EMAIL_…]]”的文本。无 pii:reveal 权限时，响应仍包含这些 token；有权限的工单后端才会收到同一请求内还原的草稿。无论哪种路径，模型都不具备发送短信或邮件的能力。

必须覆盖的测试

将以下内容保存为 test_app.py。这些测试不调用真实模型；端到端测试还应在隔离网络中抓取网关到上游的请求，确认原值不出现。

import pytest
from fastapi import HTTPException

from app import detect, restore, safe_response, transform

def test_same_request_reuses_token_but_requests_are_isolated():
    f1, r1 = {}, {}
    masked = transform("13800138000 和 13800138000", f1, r1)
    assert len(r1) == 1
    assert "13800138000" not in masked
    assert restore(masked, r1) == "13800138000 和 13800138000"

    f2, r2 = {}, {}
    transform("13800138000", f2, r2)
    assert next(iter(r1)) != next(iter(r2))
    with pytest.raises(HTTPException):
        restore(next(iter(r2)), r1)

def test_credential_blocks_before_any_redaction():
    with pytest.raises(HTTPException) as error:
        detect("password=secret123；联系 a@example.com")
    assert error.value.status_code == 403
    assert "assignment_secret" in error.value.detail

def test_client_cannot_inject_gateway_token_namespace():
    with pytest.raises(HTTPException):
        detect("请保留 [[GW_PHONE_" + "A" * 32 + "]]" )

def test_unknown_or_damaged_upstream_token_fails_closed():
    with pytest.raises(HTTPException):
        restore("[[GW_EMAIL_" + "A" * 32 + "]]", {})
    with pytest.raises(HTTPException):
        restore("[[GW_EMAIL_broken", {})

def test_response_is_masked_without_reveal_scope():
    forward, reverse = {}, {}
    tokenized = transform("13800138000", forward, reverse)
    upstream = {"id": "x", "choices": [{"message": {"content": tokenized}}]}
    masked = safe_response(upstream, reverse, can_reveal=False)
    revealed = safe_response(upstream, reverse, can_reveal=True)
    assert "13800138000" not in masked["choices"][0]["message"]["content"]
    assert revealed["choices"][0]["message"]["content"] == "13800138000"

def test_tool_output_is_not_silently_accepted():
    with pytest.raises(ValueError):
        safe_response({"choices": [{"message": {"content": "x", "tool_calls": [{}]}}]}, {}, False)

运行：

pytest -q

生产落地清单

1. 把权限判定接入现有身份体系。 演示 token 只说明控制点位置；真实系统应把工单后端的租户、用户、工单归属和 pii:reveal 权限写入已签名身份声明，并在网关验证。不能由浏览器提交 can_reveal=true 这类字段自行决定。

2. 为字段、规则和路径建立资产清单。 手机号、邮箱只是第一批允许处理的字段。身份证、银行卡、地址、姓名、内部项目名、附件内容和 RAG 片段必须逐一决定“检测并替换、阻断、或不接入”，不能默认为已保护。

3. 管理检测规则而非迷信正则。 给每条规则分配 ID、版本、负责人、回归样本和误报/漏报指标。用已获授权的标注数据评估覆盖率；发现疑似凭据的策略应比普通实体更保守。未知格式的秘密仍可能漏检，因此高敏业务更适合结合来源字段、分类标签和业务 DLP。

4. 限制网络与协议。 UPSTREAM_URL 固定在部署配置，网络出口只允许批准域名/IP，关闭重定向并限制 DNS 解析。网关只构造白名单请求头，绝不透传客户端头。不要允许请求指定 URL、模型服务商或代理。

5. 保持失败关闭。 检测依赖不可用、规则配置加载失败、上游报错、上游返回未知 token、响应格式变更时，都返回错误；绝不发送原文作为“兼容降级”。

6. 流式与多模态单独设计。 SSE 中 token 可跨事件边界，安全实现需要事件级状态缓冲、最终完成事件校验、取消处理和断连策略；完成这些之前，保持 extra="forbid" 拒绝 stream。OpenAI 的流式接口也以服务器发送事件逐段交付输出，不能把每个 delta 当成完整可还原文本。官方流式事件文档

7. 观测但不留原文。 只记录请求 ID、租户 ID、规则 ID、规则版本、脱敏数量、权限决策、耗时和错误类别。禁止在访问日志、异常堆栈、APM attributes、trace baggage、死信队列和支持工单中存入原文或 token 映射。

8. 把自动化动作隔离在模型之外。 短信、邮件、退款、查单等动作由服务端使用业务 ID 再读取受控数据，校验当前操作者权限、金额/频率/收件人策略，并要求适当的人机确认。不要从模型自由文本提取号码后执行。

上线验收标准

• 抓取网关到上游的真实测试请求，确认手机号、邮箱和调用方 Authorization 均不出现。
• 每个凭据测试样本都得到拒绝，且审计显示该请求零次上游调用。
• 无 pii:reveal 的调用方无法从任何响应字段、错误消息或日志获得原值；有权限者只能还原本请求的 token。
• 不同请求中的相同手机号产生不同 token；一个请求的映射不能还原另一个请求的 token。
• stream、tools、附件/图片 content、未支持角色及额外字段均被拒绝。
• 上游返回陌生、截断或变形 token 时失败关闭；上游新增字段不会被原样反射给客户端。
• 压测分别记录检测、上游与总耗时的 P50/P95/P99，并使用接近真实业务的混合文本；没有实测数据时不承诺“网关 P95 小于某毫秒”。

当且仅当上述路径都通过，且每个业务入口均被纳入清单，才应把该网关接入真实客服流量。

参考资料

• Python secrets：安全随机值生成
• OpenAI API：流式响应事件