Feat(#47): AI 사기문자 변형 공격 테스트 시스템 구축

.gitignore

@@ -15,3 +15,4 @@ __pycache__/
 secrets/
 *.pem
 *.key
+scripts/adversarial_test/output/

scripts/adversarial_test/aggregate.py

@@ -0,0 +1,108 @@
+import argparse
+import csv
+import logging
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+OUTPUT_DIR = Path(__file__).parent / "output"
+ORIGINAL_LABEL = "ORIGINAL"
+EXCERPT_LENGTH = 60
+
+
+def read_results(path: Path) -> list[dict]:
+    with path.open(encoding="utf-8", newline="") as f:
+        rows = list(csv.DictReader(f))
+    for row in rows:
+        row["rule_detected"] = row["rule_detected"] == "True"
+        row["pipeline_detected"] = row["pipeline_detected"] == "True"
+        row["rule_score"] = int(row["rule_score"])
+        row["pipeline_score"] = int(row["pipeline_score"])
+    return rows
+
+
+def _rate(rows: list[dict], key: str) -> float:
+    return sum(row[key] for row in rows) / len(rows) * 100 if rows else 0.0
+
+
+def _detection_row(label: str, rows: list[dict]) -> str:
+    return (
+        f"| {label} | {len(rows)} | {_rate(rows, 'rule_detected'):.1f}% "
+        f"| {_rate(rows, 'pipeline_detected'):.1f}% |"
+    )
+
+
+def build_summary(rows: list[dict]) -> str:
+    originals = [r for r in rows if r["mutation_type"] == ORIGINAL_LABEL]
+    mutated = [r for r in rows if r["mutation_type"] != ORIGINAL_LABEL]
+
+    lines = [
+        "# SafeFam 사기문자 변형 공격 테스트 결과",
+        "",
+        "탐지 성공 기준: 규칙엔진 rule_score >= 40 / 파이프라인 risk_grade != LOW",
+        "",
+        "## 1. 원본 vs 변형 전체 탐지율",
+        "",
+        "| 구분 | 건수 | 규칙엔진 단독 탐지율 | 전체 파이프라인 탐지율 |",
+        "| --- | ---: | ---: | ---: |",
+        _detection_row("원본 (ORIGINAL)", originals),
+        _detection_row("변형 전체", mutated),
+        "",
+        "## 2. 변형 유형별 탐지율 (규칙엔진 탐지율 낮은 순)",
+        "",
+        "| 변형 유형 | 건수 | 규칙엔진 탐지율 | 파이프라인 탐지율 | 규칙 평균점수 | 파이프라인 평균점수 |",
+        "| --- | ---: | ---: | ---: | ---: | ---: |",
+    ]
+
+    by_mutation: dict[str, list[dict]] = {}
+    for row in mutated:
+        by_mutation.setdefault(row["mutation_type"], []).append(row)
+
+    for name, group in sorted(by_mutation.items(), key=lambda item: _rate(item[1], "rule_detected")):
+        rule_avg = sum(r["rule_score"] for r in group) / len(group)
+        pipe_avg = sum(r["pipeline_score"] for r in group) / len(group)
+        lines.append(
+            f"| {name} | {len(group)} | {_rate(group, 'rule_detected'):.1f}% "
+            f"| {_rate(group, 'pipeline_detected'):.1f}% | {rule_avg:.1f} | {pipe_avg:.1f} |"
+        )
+
+    # 데모 핵심: 규칙엔진은 뚫렸지만 문맥분석 포함 파이프라인은 여전히 잡아낸 사례
+    saved = [r for r in mutated if not r["rule_detected"] and r["pipeline_detected"]]
+    saved_rate = len(saved) / len(mutated) * 100 if mutated else 0.0
+    lines += [
+        "",
+        "## 3. 규칙엔진 실패 -> 파이프라인 방어 성공 사례",
+        "",
+        f"변형 {len(mutated)}건 중 {len(saved)}건 ({saved_rate:.1f}%)이 "
+        "규칙엔진을 우회했으나 파이프라인이 탐지했습니다.",
+        "",
+        "| 샘플 ID | 변형 유형 | 규칙점수 | 파이프라인 점수/등급 | 변형문 발췌 |",
+        "| --- | --- | ---: | --- | --- |",
+    ]
+    for row in saved:
+        excerpt = row["text"][:EXCERPT_LENGTH].replace("\n", " ").replace("|", "/")
+        lines.append(
+            f"| {row['sample_id']} | {row['mutation_type']} | {row['rule_score']} "
+            f"| {row['pipeline_score']} ({row['pipeline_grade']}) | {excerpt}... |"
+        )
+
+    return "\n".join(lines) + "\n"
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="변형 공격 평가 결과 집계")
+    parser.add_argument("--results", type=Path, default=OUTPUT_DIR / "results.csv")
+    parser.add_argument("--out", type=Path, default=OUTPUT_DIR / "summary.md")
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+
+    rows = read_results(args.results)
+    summary = build_summary(rows)
+    args.out.parent.mkdir(parents=True, exist_ok=True)
+    args.out.write_text(summary, encoding="utf-8")
+    logger.info("[Aggregate] %d건 집계 -> %s", len(rows), args.out)
+
+
+if __name__ == "__main__":
+    main()

scripts/adversarial_test/corpus.py

@@ -0,0 +1,133 @@
+import argparse
+import csv
+import json
+import logging
+import random
+import re
+from pathlib import Path
+
+from app.analysis.rules.analyzer import analyze_text_with_rules
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_CSV_PATH = Path("data_science/Data/SMSData/phishing_total_dataset_2705.csv")
+OUTPUT_DIR = Path(__file__).parent / "output"
+DEFAULT_OUTPUT_PATH = OUTPUT_DIR / "corpus.json"
+
+_RE_NON_SLUG = re.compile(r"[^0-9a-zA-Z가-힣]+")
+
+
+# 파일명/ID로 쓸 수 있게 카테고리 한글명을 정규화 (한글은 그대로 두고 공백/기호만 제거)
+def _slugify(value: str) -> str:
+    return _RE_NON_SLUG.sub("_", value).strip("_") or "unknown"
+
+
+# "금융기관사칭=25,정부공공기관사칭=10" 형태를 카테고리별 목표 건수 dict로 변환
+def parse_category_overrides(raw: str | None) -> dict[str, int] | None:
+    if not raw:
+        return None
+    overrides: dict[str, int] = {}
+    for pair in raw.split(","):
+        name, _, count = pair.partition("=")
+        overrides[name.strip()] = int(count)
+    return overrides
+
+
+# 사기문자만 골라 카테고리별 샘플링. category_overrides에 없는 카테고리는 per_type을 기본값으로 사용 —
+# 특정 시나리오(예: 금융기관사칭)를 집중 검증하면서도 다른 유형과의 비교 기준선은 남겨둔다
+def select_corpus(
+    csv_path: Path,
+    per_type: int = 6,
+    seed: int = 42,
+    category_overrides: dict[str, int] | None = None,
+    keyword_filter: list[str] | None = None,
+    min_rule_score: int | None = None,
+) -> list[dict]:
+    by_type: dict[str, list[dict]] = {}
+
+    with csv_path.open(encoding="utf-8-sig", newline="") as f:
+        for row in csv.DictReader(f):
+            if row.get("label") != "phishing":
+                continue
+            sms_type = (row.get("type") or "").strip()
+            text = (row.get("text") or "").strip()
+            if not sms_type or not text:
+                continue
+            # 키워드 필터: "계좌 정지형" 등 특정 사기 시나리오만 골라 코퍼스를 좁힐 때 사용
+            if keyword_filter and not any(kw in text for kw in keyword_filter):
+                continue
+            # min_rule_score: 변형 실험의 "원본은 규칙엔진이 잡는다"는 전제를 보장하려면 느슨한
+            # 키워드 매칭보다 실제 규칙엔진 점수로 직접 거르는 편이 정확하다
+            if min_rule_score is not None and analyze_text_with_rules(text)["rule_score"] < min_rule_score:
+                continue
+            by_type.setdefault(sms_type, []).append(
+                {"text": text, "has_url": (row.get("has_url") or "").strip().lower() == "true"}
+            )
+
+    rng = random.Random(seed)
+    corpus: list[dict] = []
+    overrides = category_overrides or {}
+
+    for sms_type in sorted(by_type):
+        rows = by_type[sms_type]
+        target = overrides.get(sms_type, per_type)
+        sampled = rng.sample(rows, min(target, len(rows)))
+        slug = _slugify(sms_type)
+        for idx, row in enumerate(sampled):
+            corpus.append(
+                {
+                    "id": f"{slug}_{idx}",
+                    "text": row["text"],
+                    "type": sms_type,
+                    "has_url": row["has_url"],
+                }
+            )
+        logger.info("[Corpus] %s: %d개 샘플링 (전체 %d개)", sms_type, len(sampled), len(rows))
+
+    return corpus
+
+
+def write_corpus(corpus: list[dict], out_path: Path) -> None:
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    out_path.write_text(json.dumps(corpus, ensure_ascii=False, indent=2), encoding="utf-8")
+    logger.info("[Corpus] %d건 저장 -> %s", len(corpus), out_path)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="사기문자 코퍼스 생성")
+    parser.add_argument("--csv", type=Path, default=DEFAULT_CSV_PATH)
+    parser.add_argument("--per-type", type=int, default=6)
+    parser.add_argument(
+        "--category-overrides",
+        type=str,
+        default=None,
+        help="예: 금융기관사칭=25,정부공공기관사칭=10",
+    )
+    parser.add_argument(
+        "--keywords",
+        type=str,
+        default=None,
+        help="쉼표 구분 키워드. 지정 시 텍스트에 하나라도 포함된 행만 사용 (예: 계좌,이체,정지,동결)",
+    )
+    parser.add_argument(
+        "--min-rule-score",
+        type=int,
+        default=None,
+        help="지정 시 규칙엔진 점수가 이 값 이상인 원문만 사용 (원본은 규칙엔진이 이미 탐지하는 케이스로 한정)",
+    )
+    parser.add_argument("--out", type=Path, default=DEFAULT_OUTPUT_PATH)
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+    corpus = select_corpus(
+        args.csv,
+        per_type=args.per_type,
+        category_overrides=parse_category_overrides(args.category_overrides),
+        keyword_filter=args.keywords.split(",") if args.keywords else None,
+        min_rule_score=args.min_rule_score,
+    )
+    write_corpus(corpus, args.out)
+
+
+if __name__ == "__main__":
+    main()

scripts/adversarial_test/daily_batch.py

@@ -0,0 +1,124 @@
+import argparse
+import asyncio
+import csv
+import json
+import logging
+from dataclasses import asdict
+from pathlib import Path
+
+from app.analysis.service import SmishingAnalysisService
+from scripts.adversarial_test.evaluate import ORIGINAL_LABEL, EvalResult, _evaluate_one
+from scripts.adversarial_test.mutations import MUTATIONS
+
+logger = logging.getLogger(__name__)
+
+OUTPUT_DIR = Path(__file__).parent / "output"
+DEFAULT_CORPUS_PATH = OUTPUT_DIR / "corpus.json"
+DEFAULT_RESULTS_PATH = OUTPUT_DIR / "results.csv"
+
+# LLM 기반 변형은 변형 생성(1콜) + 파이프라인 평가(1콜) = 2콜, 그 외는 파이프라인 평가만 1콜.
+# Gemini 무료 티어 일일 한도(모델당 20건)를 며칠에 걸쳐 나눠 쓰기 위한 하루치 예산 산정에 사용.
+_GEMINI_MUTATION_NAMES = {"urgency_softening", "tone_normalization", "shortening", "phone_call_redirect"}
+
+
+def _estimate_cost(mutation_type: str) -> int:
+    if mutation_type == ORIGINAL_LABEL:
+        return 1
+    return 2 if mutation_type in _GEMINI_MUTATION_NAMES else 1
+
+
+def _load_done(results_path: Path) -> set[tuple[str, str]]:
+    if not results_path.exists():
+        return set()
+    with results_path.open(encoding="utf-8", newline="") as f:
+        return {(row["sample_id"], row["mutation_type"]) for row in csv.DictReader(f)}
+
+
+def _append_result(result: EvalResult, out_path: Path) -> None:
+    is_new = not out_path.exists()
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    with out_path.open("a", encoding="utf-8", newline="") as f:
+        writer = csv.DictWriter(f, fieldnames=list(EvalResult.__dataclass_fields__))
+        if is_new:
+            writer.writeheader()
+        writer.writerow(asdict(result))
+
+
+# 이미 results.csv에 성공 기록된 (샘플, 변형) 조합은 건너뛰고, 남은 대기열에서 예산이 허용하는 만큼만
+# 오늘치로 처리한다 — 프로세스가 중간에 죽어도(429 등) append 방식이라 진행분은 보존됨
+async def run_daily_batch(
+    corpus: list[dict],
+    results_path: Path,
+    budget: int,
+    mutation_names: list[str] | None = None,
+) -> list[EvalResult]:
+    names = mutation_names if mutation_names is not None else list(MUTATIONS)
+    done = _load_done(results_path)
+
+    queue = [
+        (sample, mutation_type)
+        for sample in corpus
+        for mutation_type in [ORIGINAL_LABEL, *names]
+        if (sample["id"], mutation_type) not in done
+    ]
+
+    if not queue:
+        logger.info("[DailyBatch] 남은 작업 없음 - 전체 매트릭스 완료")
+        return []
+
+    selected: list[tuple[dict, str]] = []
+    spent = 0
+    for sample, mutation_type in queue:
+        cost = _estimate_cost(mutation_type)
+        if spent + cost > budget:
+            break
+        selected.append((sample, mutation_type))
+        spent += cost
+
+    logger.info(
+        "[DailyBatch] 오늘 처리 대상: %d건 (예상 호출 %d/%d), 이후 남는 대기열 %d건",
+        len(selected),
+        spent,
+        budget,
+        len(queue) - len(selected),
+    )
+
+    service = SmishingAnalysisService()
+    semaphore = asyncio.Semaphore(1)
+
+    results: list[EvalResult] = []
+    for sample, mutation_type in selected:
+        result = await _evaluate_one(service, sample, mutation_type, semaphore)
+        if result is not None:
+            results.append(result)
+            _append_result(result, results_path)
+
+    logger.info("[DailyBatch] 이번 배치 결과 %d/%d건 성공 저장", len(results), len(selected))
+    return results
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Gemini 일일 무료 할당량 내에서 변형 평가 매트릭스를 하루치씩 이어서 채운다"
+    )
+    parser.add_argument("--corpus", type=Path, default=DEFAULT_CORPUS_PATH)
+    parser.add_argument("--results", type=Path, default=DEFAULT_RESULTS_PATH)
+    parser.add_argument(
+        "--budget",
+        type=int,
+        default=18,
+        help="오늘 소비할 예상 Gemini 호출 수 상한 (일일 한도 20보다 낮게 잡아 여유분 확보 권장)",
+    )
+    parser.add_argument("--mutations", type=str, default=None, help="쉼표 구분 변형 이름 (미지정 시 전체 10종)")
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+
+    corpus = json.loads(args.corpus.read_text(encoding="utf-8"))
+    names = args.mutations.split(",") if args.mutations else None
+
+    asyncio.run(run_daily_batch(corpus, args.results, args.budget, names))
+
+
+if __name__ == "__main__":
+    main()