9 분 소요

🔐 문서 속 개인정보, 탐지부터 평가까지 3부작 — 들어오는 문서에서 개인정보를 넓게 잡고 검증기로 확실히 거르는 탐지 게이트(1편), 잡은 값을 마스킹·가역 토큰화·봉투 암호화로 안전하게 다루는 뒷단(2편), 그리고 사람 확인 큐를 정답지 삼아 지표·임계값을 재고 되먹이는 평가 루프(3편)까지, "개인정보는 문 앞에서 거른다"는 한 목표로 이어집니다. 전체 3편.
  1. 들어오는 문서에서 개인정보를 걸러내는 Airflow util — 넓게 잡고, 확실히 거른다
  2. 탐지한 개인정보를 안전하게 다루기 — 마스킹, 가역 토큰화, 격리 저장 ← 지금 글
  3. 개인정보 게이트를 스스로 재게 하기 — 라벨·지표·임계값 스윕과 안 보이는 절반

Summary

앞 글에서 들어오는 문서를 훑어 개인정보를 잡아내는 게이트를 만들었어요. 정규식으로 넓게 잡고 → 체크섬·Luhn 으로 거르고 → 문맥으로 점수를 매겨, 문서를 차단(block)·검토(review)·통과(pass) 셋으로 갈랐죠. 근데 잡아놓고 끝이 아니에요. 잡은 다음에 그걸 어떻게 다루느냐 가 사실 더 어려운 문제예요.

검토 큐로 넘길 문서는 사람에게 보여줘야 하는데 원본 개인정보를 그대로 보여줄 순 없어요. 차단한 문서의 원본은 지워버리자니 나중에 “이거 오탐이었어요” 하면 복구할 방법이 없고, 그냥 두자니 개인정보를 쌓아두는 셈이라 그 자체가 리스크고요. 이 글은 그 뒷단을 다룹니다.

핵심은 세 가지를 구분해서 쓰는 거예요.

💡 이 글에서 다루는 것

  • 마스킹 vs 토큰화 — 되돌릴 수 없는가, 있는가 (언제 뭘 쓰나)
  • 비가역 마스킹 — hit 오프셋으로 안전하게 치환하기 (뒤에서부터)
  • 가역 토큰화 — 원본은 볼트에 잠그고 토큰만 흘리기 (결정적 토큰의 득과 실)
  • 격리 저장 — 차단된 원본을 봉투 암호화로 (KMS · 데이터 키)
  • 접근통제 + 감사로그 — 누가 원본을 열었나
  • 보관기한(TTL)과 파기 — 키를 지워서 소각(crypto-shredding)
  • Airflow review/block 분기에서 이걸 호출하기

앞 글의 pii_scan(text) 이 돌려준 report.hits(종류·좌표·점수)를 그대로 이어받아요. 예시 값은 전부 합성/마스킹 한 것이고 실제 값이 아니에요.



1. 잡은 다음이 진짜 문제다

한 문장으로 정리하면 이래요. 마스킹은 값을 죽이고, 토큰화는 값을 잠그고, 격리는 문서를 잠근다. 셋은 목적이 달라서 티어별로 다르게 씁니다.

티어 원본 처리 왜
pass 그대로 흘림 (감사 로그만) 개인정보 없음
review 마스킹본 을 사람에게 검토엔 원본이 필요 없음
block 원본을 암호화 격리 오탐 복구·근거 보존은 필요, 유출은 막아야

여기서 실수하기 쉬운 게 두 가지예요.

  • 원본을 그냥 지운다 → 오탐이었을 때 복구 불가, “왜 막았는지” 근거도 사라짐. 규제 대응 시 소명 자료가 없어요.
  • 원본을 평문으로 쌓아둔다 → 격리 저장소가 그 자체로 개인정보 유출의 표적이 됨. 잡으려고 만든 게 새 위험이 되는 거죠.

그래서 “지우지도 평문으로 두지도 않는” 중간을 설계해야 해요. 마스킹·토큰화·암호화 격리가 그 중간이에요.



2. 마스킹이냐 토큰화냐 — 되돌릴 수 있는가

가장 먼저 갈리는 결정은 원본을 나중에 되살릴 일이 있는가 예요.

  비가역 마스킹 (redaction) 가역 토큰화 (pseudonymization)
원본 복구 불가 (되돌릴 수 없음) 가능 (권한 + 볼트 있으면)
어디에 쓰나 로그·분석·사람 검토 화면 재식별이 필요한 워크플로
같은 값 매칭 안 됨(다 별표라) 됨(결정적 토큰이면)
유출 시 피해 없음(복구 불가) 볼트/키가 같이 새면 복구됨
저장 부담 없음 볼트 + 키 관리 필요

기본은 마스킹 이에요. 되돌릴 수 없는 게 가장 안전하니까요. 로그에 남기거나, 분석용으로 흘리거나, 사람이 검토 화면에서 보는 값은 전부 마스킹이면 충분해요.

토큰화는 “나중에 원본이 필요한 게 확실할 때만” 써요. 예를 들어 “같은 고객이 여러 문서에 걸쳐 몇 번 나왔나” 를 세야 하는데 값 자체는 숨겨야 한다면, 같은 원본이 항상 같은 토큰이 되는 결정적 토큰 이 필요해요. 대신 볼트와 키라는 관리 부담을 지는 거죠.

⚠️ 되돌릴 수 있다는 건 곧 되돌릴 위험도 있다 는 뜻이에요. 토큰화를 남발하면 사실상 개인정보를 형태만 바꿔 계속 들고 있는 셈이 돼요. “정말 재식별이 필요한가?” 를 먼저 물어야 해요.



3. 비가역 마스킹 — 오프셋으로 안전하게 치환

앞 글의 PiiHit 은 start/end 좌표를 들고 있었죠. 그걸로 문서에서 해당 구간만 골라 별표로 바꿉니다. 종류마다 얼마나 남기고 얼마나 가릴지 정책을 따로 둬요 — 전화번호 뒤 4자리처럼 최소한의 식별 단서는 남겨야 검토자가 “아, 그 번호” 하고 알아볼 수 있거든요.

import re

def mask_value(kind: str, raw: str) -> str:
    if kind == "rrn":     # 생년월일+성별 자리까지만, 뒤 6자리 가림
        return raw[:8] + "*" * (len(raw) - 8)
    if kind == "card":    # 앞 4 · 뒤 4 만
        return raw[:4] + "*" * (len(raw) - 8) + raw[-4:]
    if kind == "phone":   # 가운데만
        return raw[:4] + "****" + raw[-5:]
    if kind == "email":   # 로컬파트 첫 글자만
        return raw[0] + "***@" + raw.split("@")[1]
    return "*" * len(raw)

값 하나씩 넣어보면 이렇게 나와요.

for kind, raw in [("rrn", "900101-1234567"), ("card", "4242424242424242"),
                  ("phone", "010-1234-5678"), ("email", "hong@example.com")]:
    print(f"{kind:6} | {raw:18} → {mask_value(kind, raw)}")

rrn    | 900101-1234567     → 900101-1******
card   | 4242424242424242   → 4242********4242
phone  | 010-1234-5678      → 010-****-5678
email  | hong@example.com   → h***@example.com

이제 문서 전체에 적용해요. 여기서 함정이 하나 있어요 — 앞에서부터 치환하면 글자 수가 바뀌면서 뒤쪽 hit 의 start/end 좌표가 다 밀려버려요. 그래서 뒤에서부터(오프셋 큰 것부터) 치환합니다.

def mask_text(text: str, hits: list) -> str:
    # start 가 큰 것부터 치환 → 앞쪽 오프셋이 안 밀림
    for h in sorted(hits, key=lambda h: h.start, reverse=True):
        text = text[:h.start] + mask_value(h.kind, h.raw) + text[h.end:]
    return text

doc = "고객 900101-1234567 카드 4242424242424242 연락처 010-1234-5678 메일 hong@example.com"
hits = find_candidates(doc)          # 1편의 스캐너
print(mask_text(doc, hits))

고객 900101-1****** 카드 4242********4242 연락처 010-****-5678 메일 h***@example.com

review 로 분류된 문서는 이 마스킹본을 사람 확인 큐에 넣어요. 검토자는 “어디에 무슨 종류가 걸렸는지” 는 보되, 원본 숫자는 못 봐요. 마스킹은 한 번 하면 되돌릴 수 없으니 로그·분석·화면 어디에 흘러가도 안전해요.

✅ 설명을 위해 find_candidates 로 hit 를 만들었지만, 실제로는 1편의 pii_scan(text).hits 를 그대로 mask_text 에 넘기면 돼요. 마스킹은 검증·점수를 통과해 정말 개인정보로 확정된 구간 에만 적용하는 게 원칙이에요.



4. 가역 토큰화 — 원본은 볼트에 잠근다

되돌릴 필요가 있을 땐 마스킹 대신 토큰 으로 바꿔요. 원본은 암호화해서 볼트(vault) 에 넣고, 문서 자리에는 토큰만 남깁니다.

토큰을 만드는 방법이 두 갈래예요.

  • 랜덤 토큰 — 매번 다른 값. 안전하지만 같은 원본이 매번 다른 토큰이라 매칭이 안 됨.
  • 결정적 토큰 — 같은 원본은 항상 같은 토큰. 재식별 없이 “같은 사람인지” 조인이 됨. 대신 사전공격 위험이 생겨요.

“같은 고객이 몇 번 나왔나” 같은 걸 세려면 결정적이어야 해요. 원본을 노출하지 않으면서 결정성을 주려고 HMAC 을 씁니다(단순 해시가 아니라 비밀키를 섞은 해시).

import hmac, hashlib, re

# 실서비스에선 KMS/Secrets Manager 에서 로드. 절대 코드/리포에 두지 않음
_SECRET = b"<PII_TOKEN_HMAC_KEY>"

def token_of(kind: str, raw: str) -> str:
    norm = raw.lower() if kind == "email" else re.sub(r"\D", "", raw)
    mac = hmac.new(_SECRET, norm.encode(), hashlib.sha256).hexdigest()[:12]
    return f"<{kind}:{mac}>"

같은 값을 두 번 넣으면 같은 토큰이 나와요(그래서 조인이 됨).

print(token_of("phone", "010-1234-5678"))
print(token_of("phone", "01012345678"))   # 구분자만 다른 같은 번호

<phone:541af96ac1da>
<phone:541af96ac1da>

구분자(-)가 달라도 정규화 덕분에 같은 토큰이에요. 이제 토큰과 원본을 잇는 볼트를 둡니다.

_VAULT = {}   # token → 봉투 암호화된 원본 (실서비스는 KMS + 전용 DB)

def tokenize(kind: str, raw: str) -> str:
    tok = token_of(kind, raw)
    _VAULT[tok] = envelope_encrypt(raw)     # 원본은 암호화해서만 저장
    return tok

def detokenize(tok: str, *, actor: str) -> str:
    audit(actor, "detokenize", tok)         # 열람은 반드시 기록 (6장)
    return envelope_decrypt(_VAULT[tok])

detokenize 를 아무 데서나 부르지 못하게 actor(누가 요청했는지)를 필수 인자로 박아뒀어요. 복원은 권한 있는 주체가, 기록을 남기면서만 되게 하는 거예요.

🚨 결정적 토큰의 급소는 저엔트로피 개인정보 예요. 전화번호·주민번호는 가능한 값의 범위가 좁아서, 만약 _SECRET 이 새면 공격자가 모든 후보값을 넣어보며 토큰을 되맞추는 사전공격이 가능해요. 그래서 HMAC 키는 KMS 에 두고 주기적으로 로테이션 하고, 볼트 접근은 토큰화 서비스로만 좁혀야 해요.



5. 격리 저장 — 차단된 원본을 봉투 암호화로

block 문서의 원본은 파이프라인 하류로 절대 안 보내지만, 완전히 버리지도 않아요. 오탐 복구와 “왜 막았는지” 근거 보존 때문에요. 대신 봉투 암호화(envelope encryption) 로 격리 저장소에 넣습니다.

봉투 암호화는 이래요 — 문서마다 데이터 키(DEK) 를 새로 만들어 원본을 암호화하고, 그 데이터 키를 다시 마스터 키(KMS) 로 감싸서(wrap) 같이 저장해요. 원본을 열려면 KMS 로 데이터 키를 풀어야 하니, KMS 권한이 없으면 저장소를 통째로 훔쳐도 못 읽어요.

from dataclasses import dataclass

@dataclass
class QuarantineRecord:
    doc_id: str
    ciphertext: bytes      # DEK 로 암호화된 원본
    dek_wrapped: bytes     # KMS 로 감싼 데이터 키
    reason: list[str]      # 격리 사유 = 걸린 개인정보 "종류" 만 (값 아님)
    created_at: str
    expires_at: str        # 보관기한 (7장)

def quarantine_put(doc: dict) -> QuarantineRecord:
    dek = kms_generate_data_key()                 # 문서 전용 1회용 키
    rec = QuarantineRecord(
        doc_id      = doc["doc_id"],
        ciphertext  = aes_encrypt(doc["text"], dek.plaintext),
        dek_wrapped = dek.ciphertext,             # 감싼 키만 저장, 평문 키는 버림
        reason      = [h["kind"] for h in doc["pii_hits"]],
        created_at  = now_iso(),
        expires_at  = plus_days(now_iso(), 90),
    )
    del dek                                        # 평문 데이터 키는 메모리에서 즉시 폐기
    audit("system", "quarantine_put", doc["doc_id"])
    return rec

여기서 놓치기 쉬운 두 가지.

  • reason 에 값이 아니라 종류만 담아요. ["rrn", "card"] 처럼요. 사유 메타데이터에 원본을 넣으면 격리의 의미가 없어져요.
  • 평문 데이터 키는 즉시 버려요(del dek). 저장소엔 dek_wrapped(감싼 것)만 남고, 열 때마다 KMS 를 호출해 그때만 잠깐 평문 키를 얻어요.



6. 접근통제 + 감사로그 — 누가 원본을 열었나

격리 저장소와 볼트는 최소권한 이 생명이에요. “전 직원 읽기 가능” 같은 건 없어요. 복원·열람은 지정된 서비스 계정이나 승인된 담당자만, 그리고 모든 열람은 남깁니다.

def audit(actor: str, action: str, target: str) -> None:
    event = {
        "ts": now_iso(),
        "actor": actor,          # 누가
        "action": action,        # detokenize / quarantine_open / ...
        "target": target,        # 어떤 doc_id / token
    }
    append_only_log(event)       # 추가만 되는(수정·삭제 불가) 저장소로

감사로그의 핵심은 append-only 예요. 나중에 고치거나 지울 수 있으면 감사로그가 아니에요. 그리고 로그에도 원본 값이 아니라 doc_id·토큰 같은 식별자만 남겨요 — 감사로그가 개인정보 사본이 되면 안 되니까요.

💡 “누가 언제 무엇을 열었나” 가 남으면, 오탐 복구로 원본을 한 번 꺼내 본 것도 추적돼요. 사고가 났을 때 영향 범위 를 이 로그로 되짚습니다.



7. 보관기한과 파기 — 키를 지워서 소각한다

개인정보는 오래 들고 있는 것 자체가 리스크 예요. 목적을 다했으면 지워야 해요. 격리 레코드마다 expires_at 을 박아뒀으니, 기한이 지난 걸 주기적으로 파기합니다.

그런데 암호화된 원본을 “진짜로” 지우는 건 생각보다 까다로워요. 백업·복제본까지 다 찾아 덮어써야 하거든요. 그래서 봉투 암호화의 묘수를 써요 — 원본이 아니라 키를 지웁니다.

def quarantine_expire(rec: QuarantineRecord) -> None:
    # ciphertext 를 일일이 안 지워도 됨:
    # 감싼 데이터 키를 없애면 KMS 로도 못 풀어 ciphertext 는 영구 복호 불가
    destroy_wrapped_key(rec.dek_wrapped)
    audit("system", "quarantine_expire", rec.doc_id)

이걸 crypto-shredding(암호 소각) 이라고 해요. 데이터 키만 파기하면 남은 ciphertext 는 아무도 못 여는 난수 덩어리가 돼요. 백업 어딘가에 복제본이 남아 있어도 열쇠가 없으니 안전 하죠. 개인정보를 대량으로 확실하게 파기하는 표준적인 방법이에요.

이 파기 잡을 Airflow 스케줄로 매일 돌리면, 격리 저장소가 무한정 쌓이지 않고 기한이 지난 건 자동으로 소각돼요.



8. 파이프라인에 꿰기

이제 1편의 게이트 뒤에 붙이면 돼요. pii_gate 가 붙여준 pii_tier 를 보고 분기합니다.

from airflow.decorators import task

@task
def handle_review(doc: dict) -> dict:
    # 사람에겐 마스킹본만 보여준다
    doc["text_masked"] = mask_text(doc["text"], hits_from(doc))
    push_to_review_queue(doc)
    return doc

@task
def handle_block(doc: dict) -> None:
    # 원본은 암호화 격리, 하류로는 아무것도 안 흘림
    quarantine_put(doc)

전체 흐름을 한 줄로 요약하면 다음과 같아요.

문서 → pii_scan → tier?
   ├─ pass   → 적재 (감사 로그만)
   ├─ review → mask_text → 사람 확인 큐 (마스킹본)
   └─ block  → quarantine_put → 암호화 격리 → 90일 후 crypto-shred

토큰화는 이 분기와 별개로, 적재하되 재식별 여지를 남겨야 하는 특정 필드에만 골라 써요(예: 고객 단위 집계가 필요한 파이프라인). 남발하지 않는 게 포인트예요.



9. 마무리

개인정보는 잡는 것보다 잡은 다음이 어려워요. 그리고 그 뒷단은 결국 “되돌릴 수 있느냐”를 기준으로 갈려요.

  • 마스킹 — 되돌릴 수 없게 값을 죽인다. 기본값. 로그·분석·검토 화면
  • 토큰화 — 볼트에 원본을 잠그고 토큰만 흘린다. 재식별이 확실히 필요할 때만
  • 격리 — 차단 원본을 봉투 암호화로. 오탐 복구·근거는 남기되 유출은 막고
  • 감사·파기 — 열람은 append-only 로 남기고, 기한이 지나면 키를 지워 소각

관통하는 원칙은 하나예요 — 필요한 최소한만, 필요한 만큼만 들고 있는다. 마스킹으로 값을 죽일 수 있으면 죽이고, 꼭 살려야 하면 잠그고, 다 쓴 건 소각한다. 개인정보를 다루는 파이프라인은 “얼마나 잘 모으느냐” 가 아니라 “얼마나 안 남기느냐” 로 안전해져요.

일단 오늘은 여기까지…..
다음 글에서는 이 게이트·마스킹이 얼마나 잘 맞는지 스스로 재는 평가 루프(정탐/오탐 라벨을 모아 임계값을 조정하는)를 정리해볼게요.


← 이전 글: (1/3) 들어오는 문서에서 개인정보를 걸러내는 Airflow util ｜ 다음 글 →: (3/3) 개인정보 게이트를 스스로 재게 하기 — 라벨·지표·임계값 스윕

참고

(1/3) 들어오는 문서에서 개인정보를 걸러내는 Airflow util — 넓게 잡고, 확실히 거른다

11 분 소요

매일 수천 장 들어오는 문서에서 개인정보(주민번호·카드·전화·이메일)를 확실하게 잡아내는 Airflow util 을 만듭니다. 정규식으로 후보를 넓게 잡고, 체크섬·Luhn 검증기로 오탐을 걷어내고, 주변 문맥으로 점수를 매겨 3티어(자동차단·사람확인·통과)로 라우팅. 그리고 오탐...

(3/3) 검토 로그를 되먹여 양식별 파서 붙이기 — 양식 클러스터링·앵커 추출·회귀 테스트

8 분 소요

2편의 사람 확인 큐에 쌓인 검토 로그를 되먹여, 거절이 몰리는 양식에 전용 파서를 붙이는 법을 정리합니다. 양식으로 문서를 묶고, 앵커(라벨) 기반으로 값을 정확히 읽고, 사람이 고친 값을 골든셋 회귀 테스트로 만들고, 양식 드리프트를 감지하는 피드백 루프를 Python 으로 짭...