Antigravity × Gemma 4 で構築するオンプレ PII マスキングエージェント — 機密データを社外に出さない本番設計

# pii_prescreen.py — 決定論的 PII 検出器
# 期待動作: 入力文字列から確実に PII と判定できる箇所を全て検出し、
# 後段の LLM が処理する前に置換マーカーを差し込む
 
import re
from dataclasses import dataclass
from typing import Iterable, List
 
@dataclass(frozen=True)
class PIIMatch:
    kind: str          # "email" | "phone_jp" | "credit_card" | "ipv4"
    start: int
    end: int
    value: str
 
# 国際的に使えるメールアドレス検出（RFC 5321 を厳密に再現せず、誤検出を許容）
EMAIL_RE = re.compile(
    r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}"
)
 
# 日本の電話番号（市外局番 2〜5 桁、ハイフン任意）
PHONE_JP_RE = re.compile(
    r"(?<!\d)(?:0\d{1,4}[-(\s]?\d{1,4}[-)\s]?\d{3,4})(?!\d)"
)
 
# クレジットカード（Luhn はこの場では検証せず、後段の find_pii で行う）
# 末尾を \d で閉じないと "4532 0151 1283 0366 " のように区切り文字まで飲み込む
CREDIT_RE = re.compile(r"(?<!\d)\d(?:[ -]?\d){12,18}(?!\d)")
 
IPV4_RE = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b")
 
# スパンが重なったときに、どちらを残すかの優先度（小さいほど強い）
SPAN_PRIORITY = {"credit_card": 0, "email": 1, "ipv4": 2, "phone_jp": 3}
 
def luhn_valid(num: str) -> bool:
    """カード番号らしき数字列が Luhn を通るかだけ確認する。"""
    digits = [int(c) for c in num if c.isdigit()]
    if not 13 <= len(digits) <= 19:
        return False
    checksum = 0
    for i, d in enumerate(reversed(digits)):
        if i % 2 == 1:
            d *= 2
            if d > 9:
                d -= 9
        checksum += d
    return checksum % 10 == 0
 
def scan(text: str) -> Iterable[PIIMatch]:
    """各パターンを独立に走査する。スパンの重なりはここでは解決しない。"""
    for m in EMAIL_RE.finditer(text):
        yield PIIMatch("email", m.start(), m.end(), m.group())
    for m in PHONE_JP_RE.finditer(text):
        yield PIIMatch("phone_jp", m.start(), m.end(), m.group())
    for m in CREDIT_RE.finditer(text):
        if luhn_valid(m.group()):
            yield PIIMatch("credit_card", m.start(), m.end(), m.group())
    for m in IPV4_RE.finditer(text):
        yield PIIMatch("ipv4", m.start(), m.end(), m.group())
 
def resolve_overlaps(matches: Iterable[PIIMatch]) -> List[PIIMatch]:
    """重なったスパンを 1 つに畳む。長い方を優先し、同長なら優先度で決める。"""
    ordered = sorted(
        matches,
        key=lambda m: (m.start, -(m.end - m.start), SPAN_PRIORITY.get(m.kind, 99)),
    )
    kept: List[PIIMatch] = []
    for m in ordered:
        if kept and m.start < kept[-1].end:
            prev = kept[-1]
            longer = (m.end - m.start) > (prev.end - prev.start)
            same_len = (m.end - m.start) == (prev.end - prev.start)
            stronger = SPAN_PRIORITY.get(m.kind, 99) < SPAN_PRIORITY.get(prev.kind, 99)
            if longer or (same_len and stronger):
                kept[-1] = m
            continue
        kept.append(m)
    return kept
 
def find_pii(text: str) -> List[PIIMatch]:
    return resolve_overlaps(scan(text))
 
if __name__ == "__main__":
    sample = "注文番号 4532 0151 1283 0366 と TEL 090-1234-5678、接続元 192.168.0.1"
    print("重なり解決なし:", [(m.kind, m.value) for m in scan(sample)])
    print("重なり解決あり:", [(m.kind, m.value) for m in find_pii(sample)])
    # 実行結果:
    # 重なり解決なし: [('phone_jp', '0151 1283 0366'), ('phone_jp', '090-1234-5678'),
    #                 ('credit_card', '4532 0151 1283 0366'), ('ipv4', '192.168.0.1')]
    # 重なり解決あり: [('credit_card', '4532 0151 1283 0366'),
    #                 ('phone_jp', '090-1234-5678'), ('ipv4', '192.168.0.1')]

# gemma_ner.py — Gemma 4 による NER 抽出器
# 期待動作: 正規表現プリスクリーンで掴めなかった PII を文脈から抽出する
 
from llama_cpp import Llama
import json
from typing import List, TypedDict
 
class Entity(TypedDict):
    kind: str   # "person" | "location" | "org"
    text: str
    confidence: float
 
NER_PROMPT = """\
以下のテキストから日本語の固有表現を抽出してください。
出力は JSON 配列のみ。各要素は { "kind": "person|location|org", "text": "...", "confidence": 0.0-1.0 } 。
text が原文に存在しない場合は出力しないこと。
 
テキスト:
\"\"\"
{text}
\"\"\"
 
JSON:"""
 
class GemmaNER:
    def __init__(self, model_path: str, n_ctx: int = 8192):
        self.llm = Llama(
            model_path=model_path,
            n_ctx=n_ctx,
            n_threads=8,
            verbose=False,
        )
 
    def extract(self, text: str, min_confidence: float = 0.6) -> List[Entity]:
        if not text.strip():
            return []
        prompt = NER_PROMPT.format(text=text[:4000])  # 安全のため切り詰め
        try:
            resp = self.llm(
                prompt,
                max_tokens=512,
                temperature=0.0,   # 再現性を最優先
                top_p=0.95,
                stop=["\n\n"],
            )
            raw = resp["choices"][0]["text"].strip()
            # JSON 抽出（モデルが時々前置きを足すため）
            start = raw.find("[")
            end = raw.rfind("]") + 1
            if start == -1 or end == 0:
                return []
            entities = json.loads(raw[start:end])
        except (json.JSONDecodeError, KeyError, IndexError) as e:
            # 本番では構造化ログに出して通知。ここでは握りつぶさない方針
            raise RuntimeError(f"NER 失敗: {e}") from e
 
        # 検証: confidence 閾値 + 原文存在確認（幻覚対策）
        return [
            e for e in entities
            if e.get("confidence", 0) >= min_confidence
            and e.get("text") in text
        ]
 
if __name__ == "__main__":
    ner = GemmaNER("models/gemma-4-9b-it.Q4_K_M.gguf")
    text = "廣川 政樹様、ドリス株式会社の件で東京本社にお越しいただけますか。"
    for e in ner.extract(text):
        print(e)
    # 期待出力例:
    # {'kind': 'person', 'text': '廣川 政樹', 'confidence': 0.92}
    # {'kind': 'org', 'text': 'ドリス株式会社', 'confidence': 0.88}
    # {'kind': 'location', 'text': '東京本社', 'confidence': 0.71}

# masking_router.py — マスキング戦略の振り分け
import base64
import hashlib
import hmac
from cryptography.fernet import Fernet, InvalidToken
from dataclasses import dataclass
 
@dataclass
class MaskedSpan:
    original_kind: str
    placeholder: str
    encrypted: bytes | None  # 可逆時のみ
 
class MaskingRouter:
    """
    purpose に応じて可逆/不可逆を切り替える。
    - 'support': 可逆（後で復号して顧客連絡）
    - 'analytics': 不可逆（再識別不可能であるべき）
    - 'archive': 不可逆 + ハッシュで再現性確保
 
    プレースホルダは暗号文からではなく、pepper 付き HMAC から作る。
    Fernet はランダム IV と時刻を含むため、同じ値でも毎回異なる暗号文になり、
    暗号文由来のプレースホルダは同一人物を会話ごとに別人へ分裂させてしまう。
    """
    def __init__(self, key: bytes | None = None, pepper: bytes = b"", key_id: str = "k1"):
        self.fernet = Fernet(key) if key else None
        self.pepper = pepper
        self.key_id = key_id
 
    def _placeholder(self, value: str, kind: str) -> str:
        """同じ (種別, 値) には必ず同じ札を返す。pepper は鍵と同じ場所で管理する。"""
        if not self.pepper:
            raise RuntimeError("プレースホルダ生成には pepper が必須です")
        mac = hmac.new(
            self.pepper, f"{kind}\x1f{value}".encode("utf-8"), hashlib.sha256
        ).digest()
        tag = base64.b32encode(mac)[:8].decode("ascii")
        return f"[{kind.upper()}_{self.key_id}_{tag}]"
 
    def mask(self, value: str, kind: str, purpose: str) -> MaskedSpan:
        if purpose == "support":
            if self.fernet is None:
                raise RuntimeError("可逆マスキングには鍵が必須です")
            token = self.fernet.encrypt(value.encode("utf-8"))
            return MaskedSpan(kind, self._placeholder(value, kind), token)
        elif purpose == "analytics":
            placeholder = f"[{kind.upper()}_REDACTED]"
            return MaskedSpan(kind, placeholder, None)
        elif purpose == "archive":
            import hashlib
            digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:12]
            return MaskedSpan(kind, f"[{kind.upper()}_{digest}]", None)
        else:
            raise ValueError(f"未知の purpose: {purpose}")
 
    def unmask(self, span: MaskedSpan) -> str:
        if span.encrypted is None or self.fernet is None:
            raise InvalidToken("このスパンは不可逆マスクのため復号できません")
        return self.fernet.decrypt(span.encrypted).decode("utf-8")
 
if __name__ == "__main__":
    key = Fernet.generate_key()
    router = MaskingRouter(key, pepper=b"pepper-from-hsm")
    spans = [router.mask("山田 太郎", "person", "support") for _ in range(5)]
    print({sp.placeholder for sp in spans})   # {'[PERSON_k1_XGTDHEVF]'} — 5 回とも同じ
    print(router.unmask(spans[0]))            # 山田 太郎
    # 暗号文からプレースホルダを作っていた旧実装では、同じ入力 5 回で
    # [PERSON_Xfdw_Q==] [PERSON_pdXOeg==] [PERSON_d32cfQ==] … と毎回変わっていました

# audit_log.py — 監査ログの WORM 追記
import json
import time
import hashlib
from pathlib import Path
 
class WORMAuditLog:
    """
    追記専用ログ。各エントリに前エントリのハッシュを含めることで、
    途中改ざんを検出可能にする（簡易ブロックチェーン構造）。
    """
    def __init__(self, path: Path):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
 
    def _last_hash(self) -> str:
        if not self.path.exists() or self.path.stat().st_size == 0:
            return "0" * 64
        with self.path.open("rb") as f:
            f.seek(0, 2)
            size = f.tell()
            f.seek(max(0, size - 4096))
            tail = f.read().decode("utf-8", errors="ignore").strip().splitlines()
            if not tail:
                return "0" * 64
            return json.loads(tail[-1])["hash"]
 
    def append(self, event: dict) -> str:
        prev = self._last_hash()
        entry = {
            "ts": time.time(),
            "prev": prev,
            "event": event,
        }
        payload = json.dumps(entry, separators=(",", ":"), sort_keys=True)
        digest = hashlib.sha256(payload.encode("utf-8")).hexdigest()
        line = json.dumps({**entry, "hash": digest}, ensure_ascii=False)
        with self.path.open("a", encoding="utf-8") as f:
            f.write(line + "\n")
        return digest
 
    def verify(self) -> None:
        """先頭から連鎖を辿り直す。中間行の削除・改変はここでしか検出できない。"""
        prev = "0" * 64
        with self.path.open(encoding="utf-8") as f:
            for lineno, line in enumerate(f, start=1):
                line = line.strip()
                if not line:
                    continue
                record = json.loads(line)
                stored = record.pop("hash")
                payload = json.dumps(record, separators=(",", ":"), sort_keys=True)
                if hashlib.sha256(payload.encode("utf-8")).hexdigest() != stored:
                    raise ValueError(f"{lineno} 行目のハッシュが一致しません")
                if record["prev"] != prev:
                    raise ValueError(f"{lineno} 行目で連鎖が切れています")
                prev = stored
 
if __name__ == "__main__":
    # 書き込み先は root 権限が要らない場所に置く（/var/log は権限で弾かれます）
    log = WORMAuditLog(Path("./var/pii_audit.jsonl"))
    log.append({"action": "mask", "kind": "person", "purpose": "support"})
    log.append({"action": "unmask", "user": "support_ticket_42"})
    log.verify()
    print("チェーン検証 OK")
    # 2 行目を手で削除してから再実行すると:
    #   ValueError: 2 行目で連鎖が切れています