Feat(#47): AI 사기문자 변형 공격 테스트 시스템 구축 - #48
Conversation
규칙 엔진 단독으로는 탐지 실패하지만 SafeFam 전체 파이프라인(나이브베이즈+ Gemini 하이브리드 문맥분석)은 여전히 탐지함을 정량 검증하는 내부 레드팀 테스트 하네스. 실제 사기문자 코퍼스에 10종 변형(규칙기반 6종 + Gemini 패러프레이즈 4종)을 적용해 탐지율을 비교한다. - corpus.py: 카테고리/키워드/규칙엔진 점수 기준 코퍼스 선정 - mutations.py: 변형 10종 생성기 - evaluate.py / aggregate.py: 규칙엔진 단독 vs 전체 파이프라인 평가 및 집계 - daily_batch.py / import_manual_mutations.py / rate_limit.py: Gemini 무료 티어 일일 할당량 제약 내에서 평가를 재개 가능한 배치로 분할 수행 app/, data_science/SMSModel/ 등 프로덕션 코드/모델은 무수정.
📝 WalkthroughWalkthroughAdded an adversarial SMS testing workflow. It selects corpus records, generates deterministic and Gemini-based mutations, evaluates rule and pipeline detection, supports budgeted execution, and writes CSV and Markdown reports. ChangesAdversarial SMS evaluation
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Runner
participant CorpusSelector
participant MutationRegistry
participant AnalysisService
participant ResultStore
participant ReportBuilder
Runner->>CorpusSelector: select and persist corpus
Runner->>MutationRegistry: generate original and mutated messages
MutationRegistry->>AnalysisService: submit messages for analysis
AnalysisService->>ResultStore: write evaluation results
ResultStore->>ReportBuilder: load CSV results
ReportBuilder->>Runner: write Markdown summary
Possibly related issues
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (5)
scripts/adversarial_test/mutations.py (1)
1-9: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valuePrefer
collections.abc.Callable.
typing.Callableis deprecated since Python 3.9. Use thecollections.abcalias for new code.♻️ Proposed refactor
import logging import random import re -from typing import Callable +from collections.abc import Callable🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@scripts/adversarial_test/mutations.py` around lines 1 - 9, Replace the typing.Callable import in mutations.py with Callable imported from collections.abc, preserving its existing usage throughout the module.scripts/adversarial_test/run.py (1)
20-45: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚖️ Poor tradeoffShare the corpus CLI options through a parent parser.
Five arguments here (
--csv,--per-type,--category-overrides,--keywords,--min-rule-score) are copied verbatim fromscripts/adversarial_test/corpus.py, including the help strings. A change in one file will not reach the other, and the two entry points will accept different defaults.Define the shared options once in
corpus.pyas a parser withadd_help=False, then pass it throughparents=[...]here.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@scripts/adversarial_test/run.py` around lines 20 - 45, Define the shared corpus CLI options once in corpus.py using an argparse parent parser with add_help=False, including --csv, --per-type, --category-overrides, --keywords, and --min-rule-score with their existing defaults and help text. Update main in run.py to use that parser via parents=[...] and remove the duplicated argument definitions, leaving run.py-specific options unchanged.scripts/adversarial_test/import_manual_mutations.py (1)
30-41: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winValidate
mutation_typeagainst the known mutation names.The keys come from a hand-written JSON file. No check rejects an unknown name. A typo such as
shortenningis written toresults.csvandbuild_summaryinscripts/adversarial_test/aggregate.pygroups it as a separate mutation type. The report then shows a spurious row and understates the real mutation's sample count.The key
ORIGINALis also accepted. It would count manual text as a baseline original and distort the comparison in section 1 of the report.♻️ Proposed fix
+from scripts.adversarial_test.evaluate import ORIGINAL_LABEL +from scripts.adversarial_test.mutations import MUTATIONS ... for mutation_type, text in mutations.items(): + if mutation_type == ORIGINAL_LABEL or mutation_type not in MUTATIONS: + logger.warning( + "[Import] 알 수 없는 변형 유형 건너뜀: %s/%s", sample_id, mutation_type + ) + continue if not text or (sample_id, mutation_type) in done: continue queue.append((sample, mutation_type, text))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@scripts/adversarial_test/import_manual_mutations.py` around lines 30 - 41, Validate each mutation_type in the manual mutation import loop against the established known mutation-name set before queueing it; skip unknown names and reject ORIGINAL so invalid keys cannot reach results.csv or build_summary. Keep valid, non-empty, non-completed mutations queued as before.scripts/adversarial_test/evaluate.py (1)
44-50: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueReusable workflow steps are exposed under private names.
_evaluate_one,_load_done, and_append_resultare the shared building blocks of three entry points, but the underscore prefix marks them as module-private. Two modules must therefore import private symbols across module boundaries, and a future rename insideevaluate.pyordaily_batch.pybreaks them with no signal.
scripts/adversarial_test/evaluate.py#L44-L50: rename_evaluate_onetoevaluate_oneand keep it as the public single-case entry point.scripts/adversarial_test/daily_batch.py#L30-L44: rename_load_donetoload_doneand_append_resulttoappend_result, or move both intoevaluate.pynext towrite_results_csvso persistence lives in one module.scripts/adversarial_test/daily_batch.py#L10: update the import to the renamed publicevaluate_one.scripts/adversarial_test/import_manual_mutations.py#L8-L9: update both imports to the renamed publicload_done,append_result, andevaluate_one.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@scripts/adversarial_test/evaluate.py` around lines 44 - 50, Expose the shared workflow helpers under public names: rename evaluate.py’s _evaluate_one to evaluate_one, daily_batch.py’s _load_done to load_done, and _append_result to append_result. Update imports and call sites in scripts/adversarial_test/daily_batch.py lines 30-44 and 10, and scripts/adversarial_test/import_manual_mutations.py lines 8-9, to use the renamed symbols; no direct change is needed beyond these renames and references.scripts/adversarial_test/daily_batch.py (1)
19-27: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDerive the Gemini mutation set from
MUTATIONSinstead of duplicating it.
_GEMINI_MUTATION_NAMESrestates which mutations call Gemini.MUTATIONSalready encodes this: the Gemini mutations are the coroutine functions. If a new async mutation is added toscripts/adversarial_test/mutations.pyand this set is not updated,_estimate_costunder-counts by one call per case and the batch exceeds the free-tier daily limit.♻️ Proposed refactor
+import inspect ... -# LLM 기반 변형은 변형 생성(1콜) + 파이프라인 평가(1콜) = 2콜, 그 외는 파이프라인 평가만 1콜. -# Gemini 무료 티어 일일 한도(모델당 20건)를 며칠에 걸쳐 나눠 쓰기 위한 하루치 예산 산정에 사용. -_GEMINI_MUTATION_NAMES = {"urgency_softening", "tone_normalization", "shortening", "phone_call_redirect"} +# LLM 기반 변형은 변형 생성(1콜) + 파이프라인 평가(1콜) = 2콜, 그 외는 파이프라인 평가만 1콜. +# Gemini 호출 여부는 MUTATIONS의 코루틴 여부로 판별해 목록 중복을 없앤다. +_GEMINI_MUTATION_NAMES = { + name for name, fn in MUTATIONS.items() if inspect.iscoroutinefunction(fn) +}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@scripts/adversarial_test/daily_batch.py` around lines 19 - 27, Replace the duplicated _GEMINI_MUTATION_NAMES set with logic in _estimate_cost that derives Gemini usage from the coroutine-function entries in MUTATIONS. Preserve ORIGINAL_LABEL as one call, count coroutine mutations as two calls, and keep synchronous mutations at one call so newly added async mutations are included automatically.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@scripts/adversarial_test/daily_batch.py`:
- Around line 37-44: Update _append_result to determine header writing from the
existing output file’s size, treating a missing or zero-byte file as requiring a
header; preserve append behavior and write the header before the first data row
for non-empty results.csv files.
In `@scripts/adversarial_test/evaluate.py`:
- Around line 130-137: Update write_results_csv to refuse overwriting an
existing output file by default, adding an explicit overwrite parameter and only
using truncating write mode when it is enabled. Add the matching --overwrite CLI
option in the evaluate.py argument parser and pass args.overwrite at every
write_results_csv call, preserving normal writes for new files.
---
Nitpick comments:
In `@scripts/adversarial_test/daily_batch.py`:
- Around line 19-27: Replace the duplicated _GEMINI_MUTATION_NAMES set with
logic in _estimate_cost that derives Gemini usage from the coroutine-function
entries in MUTATIONS. Preserve ORIGINAL_LABEL as one call, count coroutine
mutations as two calls, and keep synchronous mutations at one call so newly
added async mutations are included automatically.
In `@scripts/adversarial_test/evaluate.py`:
- Around line 44-50: Expose the shared workflow helpers under public names:
rename evaluate.py’s _evaluate_one to evaluate_one, daily_batch.py’s _load_done
to load_done, and _append_result to append_result. Update imports and call sites
in scripts/adversarial_test/daily_batch.py lines 30-44 and 10, and
scripts/adversarial_test/import_manual_mutations.py lines 8-9, to use the
renamed symbols; no direct change is needed beyond these renames and references.
In `@scripts/adversarial_test/import_manual_mutations.py`:
- Around line 30-41: Validate each mutation_type in the manual mutation import
loop against the established known mutation-name set before queueing it; skip
unknown names and reject ORIGINAL so invalid keys cannot reach results.csv or
build_summary. Keep valid, non-empty, non-completed mutations queued as before.
In `@scripts/adversarial_test/mutations.py`:
- Around line 1-9: Replace the typing.Callable import in mutations.py with
Callable imported from collections.abc, preserving its existing usage throughout
the module.
In `@scripts/adversarial_test/run.py`:
- Around line 20-45: Define the shared corpus CLI options once in corpus.py
using an argparse parent parser with add_help=False, including --csv,
--per-type, --category-overrides, --keywords, and --min-rule-score with their
existing defaults and help text. Update main in run.py to use that parser via
parents=[...] and remove the duplicated argument definitions, leaving
run.py-specific options unchanged.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: d8dd6066-beaf-42a5-a1ed-53bdbdfe7c8d
📒 Files selected for processing (10)
.gitignorescripts/adversarial_test/__init__.pyscripts/adversarial_test/aggregate.pyscripts/adversarial_test/corpus.pyscripts/adversarial_test/daily_batch.pyscripts/adversarial_test/evaluate.pyscripts/adversarial_test/import_manual_mutations.pyscripts/adversarial_test/mutations.pyscripts/adversarial_test/rate_limit.pyscripts/adversarial_test/run.py
| 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)) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Write the header when the file exists but is empty.
is_new is False for a zero-byte results.csv. A crash or an external touch between creation and the first write leaves such a file. _load_done and read_results in scripts/adversarial_test/aggregate.py then read the first data row as the header, and every field lookup fails.
Base the decision on the file size.
🛡️ Proposed fix
def _append_result(result: EvalResult, out_path: Path) -> None:
- is_new = not out_path.exists()
out_path.parent.mkdir(parents=True, exist_ok=True)
+ is_new = not out_path.exists() or out_path.stat().st_size == 0
with out_path.open("a", encoding="utf-8", newline="") as f:📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| 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)) | |
| def _append_result(result: EvalResult, out_path: Path) -> None: | |
| out_path.parent.mkdir(parents=True, exist_ok=True) | |
| is_new = not out_path.exists() or out_path.stat().st_size == 0 | |
| 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)) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@scripts/adversarial_test/daily_batch.py` around lines 37 - 44, Update
_append_result to determine header writing from the existing output file’s size,
treating a missing or zero-byte file as requiring a header; preserve append
behavior and write the header before the first data row for non-empty
results.csv files.
| def write_results_csv(results: list[EvalResult], out_path: Path) -> None: | ||
| out_path.parent.mkdir(parents=True, exist_ok=True) | ||
| with out_path.open("w", encoding="utf-8", newline="") as f: | ||
| writer = csv.DictWriter(f, fieldnames=list(EvalResult.__dataclass_fields__)) | ||
| writer.writeheader() | ||
| for result in results: | ||
| writer.writerow(asdict(result)) | ||
| logger.info("[Evaluate] 결과 %d건 저장 -> %s", len(results), out_path) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
write_results_csv truncates the file that the daily batch workflow accumulates.
out_path.open("w", ...) truncates. The default out_path is output/results.csv, which is the same default path that _append_result in scripts/adversarial_test/daily_batch.py and scripts/adversarial_test/import_manual_mutations.py append to across days.
A single run of evaluate.py or run.py with default arguments therefore deletes every result already collected under the Gemini daily budget. Those results cannot be regenerated cheaply, because each row consumed quota.
Guard the overwrite. Refuse to truncate an existing file unless the caller opts in.
🛡️ Proposed fix: require an explicit overwrite opt-in
-def write_results_csv(results: list[EvalResult], out_path: Path) -> None:
+def write_results_csv(results: list[EvalResult], out_path: Path, overwrite: bool = False) -> None:
out_path.parent.mkdir(parents=True, exist_ok=True)
+ # daily_batch/import_manual_mutations는 같은 파일에 append로 예산을 며칠에 걸쳐 쌓는다.
+ # 기본 경로를 말없이 덮어쓰면 재생성 비용이 큰 결과가 사라진다.
+ if out_path.exists() and not overwrite:
+ raise FileExistsError(
+ f"{out_path}가 이미 존재합니다. 덮어쓰려면 --overwrite를 지정하세요."
+ )
with out_path.open("w", encoding="utf-8", newline="") as f:Add the matching CLI flag:
parser.add_argument("--overwrite", action="store_true", help="기존 results.csv를 덮어쓴다")
...
write_results_csv(results, args.out, overwrite=args.overwrite)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def write_results_csv(results: list[EvalResult], out_path: Path) -> None: | |
| out_path.parent.mkdir(parents=True, exist_ok=True) | |
| with out_path.open("w", encoding="utf-8", newline="") as f: | |
| writer = csv.DictWriter(f, fieldnames=list(EvalResult.__dataclass_fields__)) | |
| writer.writeheader() | |
| for result in results: | |
| writer.writerow(asdict(result)) | |
| logger.info("[Evaluate] 결과 %d건 저장 -> %s", len(results), out_path) | |
| def write_results_csv(results: list[EvalResult], out_path: Path, overwrite: bool = False) -> None: | |
| out_path.parent.mkdir(parents=True, exist_ok=True) | |
| # daily_batch/import_manual_mutations는 같은 파일에 append로 예산을 며칠에 걸쳐 쌓는다. | |
| # 기본 경로를 말없이 덮어쓰면 재생성 비용이 큰 결과가 사라진다. | |
| if out_path.exists() and not overwrite: | |
| raise FileExistsError( | |
| f"{out_path}가 이미 존재합니다. 덮어쓰려면 --overwrite를 지정하세요." | |
| ) | |
| with out_path.open("w", encoding="utf-8", newline="") as f: | |
| writer = csv.DictWriter(f, fieldnames=list(EvalResult.__dataclass_fields__)) | |
| writer.writeheader() | |
| for result in results: | |
| writer.writerow(asdict(result)) | |
| logger.info("[Evaluate] 결과 %d건 저장 -> %s", len(results), out_path) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@scripts/adversarial_test/evaluate.py` around lines 130 - 137, Update
write_results_csv to refuse overwriting an existing output file by default,
adding an explicit overwrite parameter and only using truncating write mode when
it is enabled. Add the matching --overwrite CLI option in the evaluate.py
argument parser and pass args.overwrite at every write_results_csv call,
preserving normal writes for new files.
1 participant
📝 개요
규칙 엔진 단독으로는 탐지에 실패하지만 SafeFam 전체 파이프라인(나이브베이즈+Gemini 하이브리드 문맥분석)은 여전히 탐지함을 정량적으로 검증하는 내부 레드팀 테스트 하네스를 추가합니다. 실제 신고 데이터셋(
phishing_total_dataset_2705.csv) 기반 코퍼스에 10종 변형(띄어쓰기 삽입, 특수문자 삽입, 기관명 변형, 숫자→문자 치환, URL 제거, 오탈자, 그리고 Gemini 패러프레이즈 기반 긴급성 순화/문체 정상화/축약/전화상담 유도)을 적용해 원본 vs 변형의 규칙엔진/전체 파이프라인 탐지율을 비교합니다.app/,data_science/SMSModel/등 프로덕션 코드·모델은 무수정이며, 기존 테스트(tests/analysis,tests/security) 84건 통과 확인했습니다.🔗 관련 이슈
🎯 주요 변경 사항
scripts/adversarial_test/corpus.py: 카테고리/키워드/규칙엔진 점수 기준으로 사기문자 코퍼스 선정scripts/adversarial_test/mutations.py: 변형 10종 생성기 (규칙기반 6종 + Gemini 패러프레이즈 4종)scripts/adversarial_test/evaluate.py,aggregate.py: 규칙엔진 단독 vs 전체 파이프라인 평가 및 탐지율 집계, 요약 리포트 생성scripts/adversarial_test/daily_batch.py,import_manual_mutations.py,rate_limit.py: Gemini 무료 티어 일일 할당량(모델당 20건) 제약 안에서 평가 매트릭스를 재개 가능한 배치로 나눠 수행.gitignore:scripts/adversarial_test/output/(코퍼스·결과·리포트 산출물) 추가📸 사진
생략
✅ PR 체크리스트
uvicorn구동 또는 테스트 코드)를 통과했습니다.Summary by CodeRabbit