大多數團隊都是在客戶抱怨或合規稽核後才發現 PII 外洩。等到那時，資料早已穿過你的 ETL pipeline，掉進應用程式日誌，並被三種不同的可觀測性工具索引過了。

事後追查根本是考古學。你真正需要的是一個系統：在 PII 進入程式碼的那一刻就開始追蹤，把這個資訊傳播到每一次轉換，並阻止它從錯誤的管道流出。這不是一個已經解決的問題，但它是可以解決的。

「追蹤 PII」到底意味著什麼

PII 追蹤經常與資料遮罩或存取控制混為一談。這些是相關的議題，但它們位於真正問題的下游。你無法對找不到的資料進行編輯、加密或限制存取。

追蹤意味著，在 pipeline 的每一個點上，你都知道哪些欄位包含敏感資料。不是根據 email 或 ssn 這類欄位名稱來猜測。而是真正知道，因為資料帶著一個標籤，這個標籤能撐過轉換、彙總和序列化。

這需要三樣東西：一個能為欄位標記敏感性的 type system 或 metadata layer、一個在資料於函式與服務之間流動時攜帶這些標籤的傳播機制，以及 enforcement points，讓你在那裡驗證被標記的資料只會寫入核准的目的地。

架構：標記、傳播、強制執行

我看過最乾淨的實作是在語言層級使用帶標籤的資料型別。在 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 或 aggregate 時，你需要規則來決定敏感度如何合併。

最簡單的規則集是 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

當你為下游服務序列化一筆記錄時，把標籤包含在 metadata 中。Consumer 接著就能決定要寫入原始資料庫、經過遮罩的數據倉儲，還是稽核日誌。

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)
    }

這是讓人栽跟頭的部分。你不能只在資料進入時標記就祈禱萬事順利。標籤必須是 serialization layer 中的一等公民。如果你的內部 RPC 格式丟掉了 metadata，資料一跨越 service boundary，你的 pipeline 就會失明。

強制執行：標籤真正發揮作用的地方

沒有 enforcement 的追蹤只是昂貴的可觀測性劇場。你需要在資料寫入儲存空間、透過網路傳送或在 UI 中渲染之前，設置檢查點來檢查帶標籤的資料。

一個常見的模式是 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

如果你的分析 pipeline 試圖把包含 Sensitivity.PII 的記錄送進 ClickHouse，寫入會在執行時以明確的錯誤失敗。這不優雅，但很明確。明確總比違反 GDPR 好。

特別是對於日誌記錄，你可以在 logger 層級強制執行。Python 的 logging module 支援 filter，可以在 log records 被發出之前檢查它們。

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())

這是個粗糙的工具。更好的方法是 structured logging：你的日誌 payload 就是同一個帶標籤的字典，而 formatter 根據敏感度標籤而非正規表示式來進行編輯。

靜態分析作為後盾

執行時標記很強大，但需要紀律。總會有人忘記包裝新欄位，或者為了避免 type error 而把 Labeled 值強制轉回原始型別。

靜態分析填補了這個缺口。像 Semgrep 或自訂 linter 這樣的工具可以強制執行：來自 HTTP request body 的原始字串絕不能直接傳給 logger 或資料庫 sink，必須先經過標記建構函式。

一個最小的 Semgrep rule 可能長這樣：

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

這不會抓到所有外洩，但會抓到明顯的那些，而大多數外洩就是從這裡來的。

沒人想談的權衡

這個方法會增加摩擦。現在每一次資料轉換都多了一個 metadata 欄位。序列化 payload 變得更大。Code review 變成爭論 user_agent 算不算 PII。（順帶一提，它通常算。歐盟也這麼認為。）

效能是真正的顧慮。如果你每秒處理數百萬筆事件，為每個欄位配置一個 Labeled wrapper 是可測量的額外負擔。在熱路徑中，你可能需要batch 檢查標籤，或把 enforcement 移到邊緣。

還有維護負擔。敏感度標籤的好壞只取決於維護它們的人。當隱私法規改變，或業務擴展到新的司法管轄區時，你需要重新標記資料。這裡沒有魔法。這就是工作。

我們沒選的方案：靜態 regex 掃描

有些團隊用 regex 事後掃描資料儲存空間，尋找看起來像電子郵件或社會安全號碼的模式。這比什麼都沒有好，但本質上是反應式的。

Regex 會漏掉混淆過的資料、雜湊過的識別碼和複合式 PII。它也會產生誤報，侵蝕對系統的信任。我們早期就考慮過這個方法並否決了它，因為它治的是症狀（資料出現在錯誤的地方），而不是病因（資料未經標記就離開服務）。

從檢查點開始

你不需要第一天就為每個服務的每個欄位加上標籤。從邊界開始。當資料從使用者、第三方 API 和事件串流進入你的系統時，就為它加上標籤。然後在風險最高的 sink 加上 enforcement：分析 pipeline、日誌基礎設施和外部整合。

等這些就位後，再向內擴展。目標不是完美的覆蓋率。目標是讓意外造成 PII 外洩的成本變高，並且讓它在 code review 中顯而易見。

如果你用 Python 實作，上面的 Labeled class 就足以開始。在 TypeScript 中，branded type 或類似的 wrapper 運作方式相同。在 Go 中，你會需要一個帶有未匯出欄位的 struct 來防止意外轉型。

工具很簡單。困難的部分在於決定：未經追蹤的 PII 就是 bug，並且要像對待 bug 一樣對待它。