大多数团队在客户投诉或合规审计后才发现 PII 泄露。到那时候，数据已经穿越了 ETL 流水线，落入了应用日志，并被三种不同的可观测性工具索引了。

事后发现属于考古学。你真正需要的是一套系统，在数据进入代码的瞬间就追踪 PII，将这一知识传播到每一次转换中，并阻止它从错误的渠道流出。这不算已解决的问题，但它是可解决的。

“追踪 PII” 到底意味着什么

PII 追踪经常与数据脱敏或访问控制混为一谈。这些是相关但下游的问题。你无法对找不到的数据进行脱敏、加密或限制访问。

追踪意味着在流水线的每一个节点上，你都知道哪些字段包含敏感数据。不是根据 email 或 ssn 这样的列名去猜，而是真正知道——因为数据携带着一个能经受住转换、聚合和序列化的标签。

这需要三样东西：一个能在语言层面将字段标记为敏感的类型系统或元数据层，一个能在数据在函数和服务之间移动时携带这些标签的传播机制，以及用于验证被标记数据是否只被写入经批准的目的地的 enforcement 点。

架构：标记、传播、强制执行

我见过最干净的实现是在语言层面使用带标签的数据类型。在 Python 中，你可以把值包装在一个携带敏感度标签的类里。

from dataclasses import dataclass
from enum import Enum, auto
from typing import Generic, TypeVar

T = TypeVar("T")

class Sensitivity(Enum):
    PLAIN = auto()
    PII = auto()
    PCI = auto()

@dataclass(frozen=True)
class Labeled(Generic[T]):
    value: T
    sensitivity: Sensitivity = Sensitivity.PLAIN

    def map(self, fn):
        return Labeled(fn(self.value), self.sensitivity)

当用户注册账户时，你在边界处对原始输入打标签。

from labeled import Labeled, Sensitivity

def create_user(payload: dict) -> dict:
    email = Labeled(payload["email"], Sensitivity.PII)
    name = Labeled(payload["name"], Sensitivity.PII)
    age = Labeled(int(payload["age"]), Sensitivity.PLAIN)

    user = {
        "id": generate_uuid(),
        "email": email,
        "name": name,
        "age": age,
    }

    return user

关键洞察在于 email 和 name 现在具备了自描述性。任何函数接收 Labeled 值时，都可以在不解析 payload 或不根据键名猜测的情况下检查其敏感度。

在转换中传播标签

打了标签的数据只有在标签能经受住业务逻辑时才真正有用。当你 map、filter 或聚合时，你需要一套规则来决定敏感度如何合并。

最简单的规则集是一个 join semilattice。如果你合并两个值，结果获得更严格的标签。

class Sensitivity(Enum):
    PLAIN = 1
    PII = 2
    PCI = 3

    def join(self, other: "Sensitivity") -> "Sensitivity":
        return self if self.value >= other.value else other

当你为下游服务序列化一条记录时，把标签包含在元数据中。消费者可以据此决定写入原始数据库、脱敏后的分析数仓，还是审计日志。

def serialize(record: dict) -> dict:
    return {
        key: {
            "value": serialize_value(val.value),
            "sensitivity": val.sensitivity.name.lower(),
        }
        for key, val in record.items()
        if isinstance(val, Labeled)
    }

这是最容易绊倒人的地方。你不能只在数据摄入时打标签就指望万事大吉。标签必须是你序列化层的一等公民。如果你的内部 RPC 格式丢弃了元数据，那么数据一旦跨越 service boundary，你的流水线就会失明。

强制执行：标签真正发挥作用的地方

只追踪不强制执行，就是昂贵的可观测性表演。你需要在数据被写入存储、通过网络发送或在界面中渲染之前设置检查点，对打了标签的数据进行检查。

一种常见模式是 sink registry，它将目的地映射到允许的最大敏感度。

SINK_POLICIES = {
    "raw_postgres": {Sensitivity.PLAIN, Sensitivity.PII, Sensitivity.PCI},
    "analytics_clickhouse": {Sensitivity.PLAIN},
    "application_logs": {Sensitivity.PLAIN},
    "third_party_api": set(),
}

def write(sink_name: str, record: dict):
    allowed = SINK_POLICIES[sink_name]
    for key, val in record.items():
        if isinstance(val, Labeled) and val.sensitivity not in allowed:
            raise ValueError(
                f"Field {key} with sensitivity {val.sensitivity.name} "
                f"is not allowed in sink {sink_name}"
            )
    # proceed with write

如果你的分析流水线试图将包含 Sensitivity.PII 的记录摄入 ClickHouse，写入会在运行时以明确的错误失败。这不算优雅，但足够清晰。清晰总比 GDPR 违规好。

特别是对于日志，你可以在 logger 层面强制执行。Python 的 logging 模块支持 filters，可以在日志记录被发出前检查它们。

import logging

class PiiFilter(logging.Filter):
    def filter(self, record):
        msg = record.getMessage()
        # In practice, use a more sophisticated check or structured logging
        if any(label in msg for label in ["email", "ssn", "phone"]):
            record.msg = "[REDACTED: PII detected]"
            record.args = ()
        return True

logger = logging.getLogger(__name__)
logger.addFilter(PiiFilter())

这是一种粗暴的手段。更好的方法是结构化日志，让你的日志 payload 就是同一个带标签的字典，formatter 根据敏感度标签而不是正则表达式来进行脱敏。

静态分析作为后备防线

运行时打标签很强大，但它需要纪律。总有人会忘记包装一个新字段，或者会把 Labeled 值强制转回原始类型来避免类型错误。

静态分析填补了这个缺口。像 Semgrep 或自定义 linter 这样的工具可以强制执行：来自 HTTP 请求体的原始字符串绝不能未经 labeling 构造函数就直接传给 logger 或数据库 sink。

一个最小的 Semgrep 规则看起来像这样：

rules:
  - id: unlabeled-pii-log
    patterns:
      - pattern: logger.info($X)
      - metavariable-pattern:
          metavariable: $X
          pattern-either:
            - pattern: request.json[$FIELD]
            - pattern: request.form[$FIELD]
    message: "Logging raw request data without sensitivity labeling"
    languages: [python]
    severity: ERROR

这不会捕获所有泄露，但会捕获最明显的那些——而大多数泄露恰恰来自这些最明显的疏漏。

没人愿意谈的权衡

这种方法会增加摩擦。每一次数据转换现在都涉及一个额外的元数据字段。序列化 payload 会变得更大。代码审查会变成关于 user_agent 算不算 PII 的争论。（它通常是，顺便说一句。欧盟是这么认为的。）

性能是一个真实的顾虑。如果你每秒处理数百万事件，为每个字段分配 Labeled 包装器是可测量的开销。在热路径中，你可能需要批量检查标签，或者把 enforcement 移到边缘。

还有维护负担。敏感度标签的好坏取决于维护它们的人。当隐私法规改变，或者你的业务扩展到新的司法管辖区时，你需要重新打标签。这里没有魔法。这是工作。

我们没有选择的方案：静态正则扫描

有些团队通过在事后用正则表达式扫描数据存储来解决这个问题，寻找看起来像邮箱或社保号的模式。这比什么都没有好，但它本质上是反应式的。

正则表达式会漏掉混淆数据、哈希标识符和复合 PII。它还会产生侵蚀系统信任度的误报。我们早期考虑过这种方法并拒绝了它，因为它治疗的是症状（数据出现在错误的地方），而不是病因（数据未经标签就流出服务）。

从检查点开始

你不需要在第一天就给每个服务中的每个字段打标签。从边界开始。在数据从用户、第三方 API 和事件流进入系统时为其打标签。然后在最高风险的 sink 处添加 enforcement：分析流水线、日志基础设施和外部集成。

一旦这些就位，再向内扩展。目标不是完美覆盖。目标是让 PII 泄露在无意中变得昂贵，并在代码审查中变得容易被发现。

如果你用 Python 构建这个系统，上面的 Labeled 类足以让你起步。在 TypeScript 中，branded type 或类似的包装器也能起到同样的作用。在 Go 中，你会想要一个带有未导出字段的 struct，以防止意外类型转换。

工具很简单。困难的部分是决定未追踪的 PII 是一个 bug，并像对待 bug 一样处理它。