Claude API の PII マスキングは台帳運用で決まる — 復元・暗号化・漏洩率の本番設計メモ

// pii-ledger.ts — 復元用台帳を AES-256-GCM で暗号化して共有ストアへ保存する
// 鍵は KMS から注入する想定。プロセス内に平文鍵を長く保持しない
import { createCipheriv, createDecipheriv, randomBytes, createHash } from "node:crypto";
 
const KEY = Buffer.from(process.env.LEDGER_KEY_BASE64!, "base64"); // 32 bytes
 
export function sealLedger(ledger: Record<string, string>): string {
  const iv = randomBytes(12);
  const cipher = createCipheriv("aes-256-gcm", KEY, iv);
  const body = Buffer.concat([cipher.update(JSON.stringify(ledger), "utf8"), cipher.final()]);
  const tag = cipher.getAuthTag();
  // iv(12) + tag(16) + body を base64 で1本化
  return Buffer.concat([iv, tag, body]).toString("base64");
}
 
export function openLedger(sealed: string): Record<string, string> {
  const buf = Buffer.from(sealed, "base64");
  const iv = buf.subarray(0, 12), tag = buf.subarray(12, 28), body = buf.subarray(28);
  const decipher = createDecipheriv("aes-256-gcm", KEY, iv);
  decipher.setAuthTag(tag);
  const out = Buffer.concat([decipher.update(body), decipher.final()]);
  return JSON.parse(out.toString("utf8"));
}
 
// 会話単位のキー。会話IDは推測されにくいよう、生IDではなくハッシュを使う
export const ledgerKey = (conversationId: string) =>
  `pii:ledger:${createHash("sha256").update(conversationId).digest("hex").slice(0, 32)}`;

// pii-pipeline.ts — Claude API ラッパー最終形。会話単位で台帳を継承し、応答を復元する
import Anthropic from "@anthropic-ai/sdk";
import { detectLayer1 } from "./pii-detect";
import { detectLayer2 } from "./pii-ner";
import { sealLedger, openLedger, ledgerKey } from "./pii-ledger";
import type { Redis } from "ioredis";
 
const client = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY! });
 
// 原文 → トークンの逆引きを作りつつ、台帳(トークン → 原文)を更新する
function applyMasks(text: string, ledger: Record<string, string>): string {
  const reverse = new Map(Object.entries(ledger).map(([t, o]) => [o, t]));
  const counters: Record<string, number> = {};
  for (const t of Object.keys(ledger)) {
    const cat = t.slice(1, t.lastIndexOf("_"));
    counters[cat] = Math.max(counters[cat] ?? 0, parseInt(t.slice(t.lastIndexOf("_") + 1), 10) || 0);
  }
  const register = (raw: string, category: string): string => {
    const existing = reverse.get(raw);
    if (existing) return existing;
    counters[category] = (counters[category] ?? 0) + 1;
    const token = `<${category}_${String(counters[category]).padStart(3, "0")}>`;
    ledger[token] = raw; reverse.set(raw, token);
    return token;
  };
  let masked = text;
  // Layer1 と Layer2 の検出結果を、長い文字列から順に全置換（オフセットずれ回避）
  const items = [
    ...detectLayer1(text).map((s) => ({ text: s.text, category: s.category })),
  ];
  // 同期関数内では Layer2 は呼べないため、呼び出し側で結合する（下の callClaude を参照）
  for (const it of items.sort((a, b) => b.text.length - a.text.length)) {
    masked = masked.split(it.text).join(register(it.text, it.category));
  }
  return masked;
}
 
export async function callClaude(
  redis: Redis, conversationId: string,
  messages: { role: "user" | "assistant"; content: string }[],
): Promise<string> {
  const key = ledgerKey(conversationId);
  const sealed = await redis.get(key);
  const ledger: Record<string, string> = sealed ? openLedger(sealed) : {};
 
  const masked = [];
  for (const m of messages) {
    // Layer2(NER) を先に流し、原文へ反映してから Layer1 を含めて一括マスク
    const ents = await detectLayer2(m.content);
    let pre = m.content;
    const reverse = new Map(Object.entries(ledger).map(([t, o]) => [o, t]));
    const counters: Record<string, number> = {};
    for (const e of ents.sort((a, b) => b.text.length - a.text.length)) {
      const ex = reverse.get(e.text);
      const token = ex ?? `<${e.category}_${String((counters[e.category] = (counters[e.category] ?? 0) + 1)).padStart(3, "0")}>`;
      if (!ex) { ledger[token] = e.text; reverse.set(e.text, token); }
      pre = pre.split(e.text).join(token);
    }
    masked.push({ role: m.role, content: applyMasks(pre, ledger) });
  }
 
  // 台帳を暗号化保存（TTLは会話の想定最大時間。例: 24h）
  await redis.set(key, sealLedger(ledger), "EX", 60 * 60 * 24);
 
  const res = await client.messages.create({
    model: "claude-sonnet-4-6", max_tokens: 2048,
    system: "応答中の <PERSON_001> のような <カテゴリ_番号> 形式の文字列は翻訳・変形せず、必ずそのまま出力に含めてください。",
    messages: masked,
  });
  const raw = res.content.filter((b) => b.type === "text").map((b: any) => b.text).join("");
 
  // 復元: 台帳の全トークンを原文へ戻す
  let restored = raw;
  for (const [token, original] of Object.entries(ledger)) restored = restored.split(token).join(original);
  return restored;
}

# golden_eval.py — マスク後本文に PII の痕跡が残っていないかを毎日CIで測る
# 合成データなので生PIIを扱わず、検出漏れ(false negative)を再現率として可視化する
import json, re, sys
from statistics import mean
 
EMAIL = re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b")
PHONE = re.compile(r"(?:\+?81[-\s]?|0)\d{1,4}[-\s]?\d{1,4}[-\s]?\d{3,4}")
 
def leaked(masked: str) -> bool:
    # マスク後に識別子パターンが残っていれば漏れ
    return bool(EMAIL.search(masked) or PHONE.search(masked))
 
def run(path: str, threshold: float = 0.99) -> int:
    cases = [json.loads(l) for l in open(path, encoding="utf-8")]
    recalls = []
    for c in cases:
        masked = c["masked"]            # マスク器の出力(事前生成)
        expected_tokens = c["expected"] # 埋め込んだPII件数
        found = masked.count("<") - leaked_count(masked)
        recalls.append(min(found, expected_tokens) / max(expected_tokens, 1))
    r = mean(recalls) if recalls else 0.0
    leaks = sum(1 for c in cases if leaked(c["masked"]))
    print(f"recall={r:.4f} leaks={leaks}/{len(cases)}")
    return 0 if (r >= threshold and leaks == 0) else 1
 
def leaked_count(masked: str) -> int:
    return len(EMAIL.findall(masked)) + len(PHONE.findall(masked))
 
if __name__ == "__main__":
    sys.exit(run(sys.argv[1]))