ほとんどのチームは、顧客からの苦情やコンプライアンス監査の後にPII漏洩を発見する。その時点で、データはすでにETLパイプラインを横断し、アプリケーションログに着地し、3つの異なるオブザーバビリティツールにインデックスされている。

事後的に発見するのは考古学だ。実際に必要なのは、データがコードに入った瞬間からPIIを追跡し、その知識をすべての変換を通じて伝播させ、間違ったチャネルからの流出をブロックするシステムだ。これは解決済みの問題ではないが、解決可能な問題だ。

「PIIを追跡する」とは実際に何を意味するのか

PIIの追跡は、しばしばデータマスキングやアクセス制御と混同される。これらは関連する関心事だが、本当の問題の下流にある。マスク、暗号化、アクセス制限のいずれも、位置がわからないデータには実行できない。

追跡とは、パイプラインのあらゆる地点で、どのフィールドに機密データが含まれているかを知っていることだ。email や ssn のようなカラム名から推測するのではない。実際に知っていることだ。データが変換、集計、シリアライゼーションを経ても生き残るラベルを持っているからだ。

これには3つのものが必要だ。フィールドを機密としてタグ付けできる型システムまたはメタデータ層、データが関数やサービス間を移動する際にそのタグを伝播させるメカニズム、タグ付けされたデータが承認された宛先にのみ書き込まれることを検証する enforcement ポイントだ。

アーキテクチャ: タグ付け、伝播、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 値を受け取るあらゆる関数は、ペイロードを解析したりキー名から推測したりせずに、その機密度を検査できる。

変換を通じてラベルを伝播させる

ラベル付きデータは、ビジネスロジックを通じてラベルが生き残って初めて有用だ。map、filter、集計を行う際、機密度がどう組み合わさるかのルールが必要だ。

最も単純なルールセットはjoin semilatticeだ。2つの値をマージすると、より制限的なラベルが結果に付与される。

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形式がメタデータを落とすと、データがサービス境界を超えた瞬間にパイプラインは目が見えなくなる。

Enforcement: ラベルが実際に重要になる場所

追跡だけでは高価なオブザーバビリティの見せかけに過ぎない。ラベル付きデータがストレージに書き込まれる前、ネットワーク越しに送信される前、UIにレンダリングされる前に検査される絞り込みポイントが必要だ。

一般的なパターンは、宛先を最大許容機密度にマッピングするsinkレジストリだ。

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違反よりもはるかにマシだ。

ロギングの場合、ロガーレベルでこれを enforcement できる。Pythonの logging モジュールは、ログレコードが出力される前に検査できるフィルタをサポートしている。

import logging

class PiiFilter(logging.Filter):
    def filter(self, record):
        msg = record.getMessage()
        # 実際には、より洗練されたチェックまたは構造化ロギングを使用する
        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())

これは鈍い道具だ。より良いアプローチは構造化ロギングで、ログペイロードが同じラベル付き辞書であり、フォーマッタが正規表現ではなく機密度タグに基づいて編集するものだ。

静的解析によるセーフティネット

実行時のラベリングは強力だが、規律を必要とする。誰かが新しいフィールドをラップし忘れたり、型エラーを避けるために Labeled 値をプリミティブに戻したりする。

静的解析がその隙間を埋める。Semgrep やカスタムリンターのようなツールは、HTTPリクエストボディからの生の文字列が、ラベリングコンストラクタを通さずに直接ロガーやデータベースsinkに渡されないことを enforcement できる。

最小限の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

これですべての漏洩を捉えられるわけではないが、明らかなものは捉えられる。そして、ほとんどの漏洩は明らかなところから来る。

誰も話したがらないトレードオフ

このアプローチは摩擦を追加する。すべてのデータ変換に追加のメタデータフィールドが関与するようになる。シリアライゼーションペイロードが大きくなる。コードレビューは、user_agent がPIIかどうかの議論になる。（普通はそうだ。EUもそう考えている。）

パフォーマンスは本当の懸念事項だ。毎秒数百万イベントを処理している場合、すべてのフィールドに Labeled ラッパーを割り当てるのは測定可能なオーバーヘッドだ。ホットパスでは、ラベルチェックをバッチ処理するか、enforcement をエッジに移動する必要があるかもしれない。

メンテナンスの負担もある。機密度ラベルは、それを維持する人間と同じくらい良いものだ。プライバシー規制が変わったり、ビジネスが新しい管轄区域に拡大したりすると、データを再タグ付けする必要がある。ここに魔法はない。仕事だ。

私たちが選ばなかったもの: 保存後の正規表現スキャン

一部のチームは、事後的に正規表現でデータストアをスキャンし、メールアドレスや社会保障番号のようなパターンを探すことでこれを解決しようとする。これは何もしないよりはマシだが、根本的にリアクティブだ。

正規表現は難読化されたデータ、ハッシュ化された識別子、複合PIIを見逃す。システムへの信頼を損なう誤検出も生じる。私たちは早期にこのアプローチを検討し、拒否した。なぜなら、症状（間違った場所にあるデータ）ではなく原因（ラベルなしでサービスから出ていくデータ）を扱わないからだ。

絞り込みポイントから始めよう

day oneですべてのサービスのすべてのフィールドにラベルを付ける必要はない。境界から始める。データがユーザー、サードパーティAPI、イベントストリームからシステムに入る際にラベル付けする。次に、最もリスクの高いsinkに enforcement を追加する: 分析パイプライン、ロギングインフラ、外部統合。

これらが整ったら、内側に拡張する。目標は完璧なカバレッジではない。目標は、PII漏洩を偶発的に作るのを高価にし、コードレビューで明らかに捉えやすくすることだ。

Pythonでこれを構築する場合、上記の Labeled クラスで始めるのに十分だ。TypeScriptでは、ブランド型や同様のラッパーが同じように機能する。Goでは、誤ってキャストされるのを防ぐために非公開フィールドを持つstructが望ましい。

ツールは単純だ。難しい部分は、追跡されていないPIIをバグと見なし、それをバグとして扱うことを決断することだ。