diff --git a/CLAUDE.md b/CLAUDE.md index 7bc16a3..71b40db 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -14,52 +14,69 @@ cd charcnn-kit ## 파이프라인 (순서) ```bash pip install -r requirements.txt -python src/download_phiusiil.py # 데이터 다운로드 (UCI PhiUSIIL, 5만 균형 샘플) -python src/preprocess.py # raw -> train/valid/test (8:1:1 stratified) +python src/download_train_mixed.py # 데이터 다운로드 (PhiUSIIL + URLhaus + OpenPhish + PhishTank + Tranco 혼합) +python src/preprocess.py # raw -> train/valid/test (domain-group 8:1:1 split) python src/train.py # 학습 (early stopping + best-model 저장) python src/evaluate.py # 지표 + FN/FP URL 출력 +python src/download_ood.py # 학습 URL 제외 OOD 평가셋 생성 +python src/evaluate_ood.py # OOD 지표 + threshold sweep + hard example 저장 python src/predict.py --url "..." [--xai] # 단일 URL 예측 ``` -대안 데이터 소스: `python src/download_data.py` (URLhaus + Tranco). 두 소스 형식 차이로 모델이 trivially 분리되므로 baseline용으론 부적합하다. +대안 데이터 소스: `python src/download_phiusiil.py` (UCI PhiUSIIL만 사용), `python src/download_data.py` (URLhaus + Tranco만 사용). 단일 출처/두 출처만 쓰면 source bias가 생기기 쉬우므로 현재 기본은 혼합 학습셋이다. PhishTank app key가 있으면 `PHISHTANK_APP_KEY` 환경변수 사용. ## 절대 어겨선 안 되는 컨벤션 - **라벨 규칙**: `0 = 정상`, `1 = 악성`. README에서도 명시한 고정 규칙. PhiUSIIL은 반대 규칙(1=legit, 0=phish)이라 `download_phiusiil.py`에서 라벨을 반전시킨다 — 이 반전 로직은 깨면 안 됨. - **수정 금지 영역** (README 8번): - `dataset.py`의 문자 인코딩 구조 (`build_vocab`, `encode_url`, PAD/UNK 토큰) - - `model.py`의 CharCNN 핵심 구조 + - `model.py`의 `CharCNN` 핵심 구조 (CharCNN 클래스는 그대로 두고 새 클래스 `HybridCharCNN`이 sub-module만 재사용) - 라벨 기준 자체 - **vocab 생애주기**: vocab은 `train.py`에서 train.csv로부터 생성되어 `saved/char_vocab.json`에 저장된다. evaluate/predict는 **반드시 저장된 vocab을 로드**해야 한다 (새로 만들면 character→index 매핑이 달라져 모델이 깨짐). +- **feature 정규화 통계 생애주기**: tabular feature mean/std는 `train.py`에서 train.csv로만 fit하여 `saved/feature_norm.json`에 저장. valid/test/predict는 반드시 저장된 통계를 로드해 적용 — train 외 데이터의 통계를 섞으면 data leakage가 됨. +- **금지 feature**: `URLSimilarityIndex`는 PhiUSIIL에서 라벨과 |corr|=0.86으로 cheat feature 역할. `FEATURE_COLS`에 절대 추가하지 말 것 (이슈 #3 ablation에서 확인). ## 아키텍처 (큰 그림) -**데이터 흐름:** +**데이터 흐름 (이슈 #3 이후):** ``` -data/raw/urls.csv (url, label) - └─ preprocess.py → stratified 8/1/1 split, 라벨 검증, 중복/null 제거 +data/raw/urls.csv (url, label, source) + └─ preprocess.py → domain-group 8/1/1 split, 라벨 검증, 중복/feature-NaN 제거 data/processed/{train,valid,test}.csv - └─ URLDataset (dataset.py) → char-level encode_url(URL) → tensor + └─ URLDataset (dataset.py) → (char-encoded URL, tabular feature 벡터, label) + → train의 mean/std로 표준화 torch.DataLoader - └─ CharCNN (model.py): Embedding → 여러 Conv1d kernel → max-pool over time → concat → Dropout → FC(1) → BCEWithLogitsLoss -saved/charcnn.pt + saved/char_vocab.json + └─ HybridCharCNN (model.py): + ├─ char 분기: Embedding → 여러 Conv1d kernel → max-pool over time → concat → Dropout + ├─ tabular 분기: Linear → ReLU → Dropout (작은 MLP) + └─ 두 분기 concat → FC(1) → BCEWithLogitsLoss +saved/charcnn.pt + saved/char_vocab.json + saved/feature_norm.json ``` -**CharCNN 구조 핵심:** 문자별 임베딩을 Conv1d 여러 kernel size(`KERNEL_SIZES`)로 통과시켜 각각의 max-pool 결과를 concat. URL 길이는 `MAX_LEN`(200)으로 패딩/truncate. +**CharCNN 구조 핵심:** 문자별 임베딩을 Conv1d 여러 kernel size(`KERNEL_SIZES`)로 통과시켜 각각의 max-pool 결과를 concat. URL 길이는 `MAX_LEN`(200)으로 패딩/truncate. CharCNN 클래스 자체는 보존되고, `HybridCharCNN`이 그 sub-module(embedding/convs/dropout)을 재사용해 forward를 새로 구성한다. + +**Tabular feature (이슈 #3):** +- `FEATURE_COLS = URL_FEATURE_COLS` (25개). URL 문자열에서 즉시 계산 가능한 feature만 학습/평가/추론 전부에서 동일하게 사용한다. +- 동적 URL 대응 feature 포함: query/path/token 관련 `NoOfQueryParams`, `QueryLength`, `PathLength`, `NoOfPathSegments`, `HasFragment`, `NoOfEqualsInURL`, `NoOfAmpersandInURL` 등. +- `HTML_FEATURE_COLS`는 현재 비활성화. 실제 predict/evaluate_ood에서 페이지 fetch 없이 mean imputation만 하게 되면 OOD 오탐이 커지므로 다시 추가하지 말 것. **학습 흐름 (`train.py`):** - valid_loss를 epoch마다 계산 - `valid_loss`가 갱신될 때마다 best state를 `copy.deepcopy`로 보관 - `PATIENCE` epoch 동안 개선 없으면 early stopping - 최종 저장은 **마지막 epoch이 아닌 best epoch의 state** (이게 핵심) +- tabular feature의 mean/std도 train에서만 fit해 `saved/feature_norm.json`에 저장 -**Config (`config.py`):** 모든 하이퍼파라미터/경로의 단일 출처. README 9번에 추천 튜닝 순서가 정리되어 있음. +**Config (`config.py`):** 모든 하이퍼파라미터/경로/feature 컬럼 목록의 단일 출처. README 9번에 추천 튜닝 순서가 정리되어 있음. +**Split/Augmentation/Hard mining:** `USE_DOMAIN_GROUP_SPLIT=True`가 기본. 같은 등록 도메인이 train/valid/test에 섞이지 않도록 `StratifiedGroupKFold`를 사용한다. random row split으로 되돌리면 성능은 올라가 보여도 데이터 누수 가능성이 커진다. `download_train_mixed.py`는 정상/악성 각각 5,000개의 동적 URL(query/path/token) synthetic augmentation을 추가한다. `evaluate_ood.py`는 FN/FP를 `data/hard_examples/hard_examples.csv`에 저장하고, 다음 `download_train_mixed.py` 실행 때 클래스별 최대 `HARD_EXAMPLES_MAX_PER_CLASS`개만 재학습에 섞는다. hard example 도메인 group은 train에 고정한다. ## 알려진 제약 -- **CharCNN의 본질적 한계**: URL 문자열만 보므로 "평범해 보이는 신규 phishing 도메인"은 구분 불가. 현재 baseline의 FN 9건이 모두 이 패턴(`https://www.{도메인}.{tld}` 단순 형태). recall을 더 올리려면 WHOIS/DNS feature 통합이 필요 — 모델 구조 튜닝만으로는 한계. +- **CharCNN/URL-feature의 한계**: URL 문자열만 보면 "평범해 보이는 신규 phishing 도메인"은 놓칠 수 있다. 그래서 URLhaus/Tranco/PhishTank/OpenPhish 등 외부 출처 OOD 평가를 계속 돌려야 한다. +- **PhiUSIIL HTML feature의 OOD 위험**: `URLSimilarityIndex`는 라벨과 사실상 1:1 → 학습용 `FEATURE_COLS`에서 제외. `HasSocialNet`, `HasCopyrightInfo`, `HasDescription`도 실제 추론에서 계산하지 않으면 train/test 분포가 어긋난다. 현재는 HTML feature 비활성화 상태. - **Windows 콘솔에서 한글 mojibake**: print의 한글이 깨져 보일 수 있음(예: `악성` → `��`). 코드/저장 파일은 UTF-8로 정상이며, **콘솔 표시 문제일 뿐**이므로 무시해도 된다. - **재현 가능 산출물은 .gitignore됨**: `data/raw/*.csv`, `data/processed/*.csv`, `saved/*.pt`, `saved/*.json`. 새 환경에서 작업할 땐 위 파이프라인을 처음부터 돌려야 한다. ## 현재 baseline (참고용) -- PhiUSIIL test 5,000건: Accuracy 0.9982, Precision 1.0000, Recall 0.9964, FN 9, FP 0 -- README 목표(recall ≥0.90, accuracy ~0.90)는 충족 상태. 다음 작업의 비교 기준점. +- **Mixed + URL-feature HybridCharCNN** (2026-04-28, domain split, dynamic URL augmentation, hard-mining 1회, PhiUSIIL+URLhaus+OpenPhish+PhishTank+Tranco): internal test Accuracy 0.9975, Precision 0.9980, Recall 0.9969, FN 17, FP 11 at threshold 0.2. +- **OOD(URLhaus+PhishTank/Tranco, 학습 URL 제외)**: Accuracy 0.9975, Precision 0.9988, Recall 0.9962, FN 19, FP 6 at threshold 0.2. Hard mining은 오탐 안정성을 개선했지만 FN은 증가했으므로 recall 우선 운영이면 threshold 0.1 후보도 함께 비교할 것. +- README 목표(recall ≥0.90, accuracy ~0.90)는 둘 다 충족. diff --git a/charcnn-kit/.gitignore b/charcnn-kit/.gitignore index 68bfe68..3dea07c 100644 --- a/charcnn-kit/.gitignore +++ b/charcnn-kit/.gitignore @@ -6,8 +6,13 @@ __pycache__/ # 생성물: 다운로드 스크립트로 재생성 가능 data/raw/*.csv data/processed/*.csv +data/ood/*.csv +data/hard_examples/*.csv +data/phiusiil_columns_report.txt !data/raw/.gitkeep !data/processed/.gitkeep +!data/ood/.gitkeep +!data/hard_examples/.gitkeep # 학습 산출물: train.py로 재생성 가능 saved/*.pt diff --git a/charcnn-kit/README.md b/charcnn-kit/README.md index 28306c3..a0c0564 100644 --- a/charcnn-kit/README.md +++ b/charcnn-kit/README.md @@ -59,6 +59,18 @@ http://paypa1-login.xyz,1 * `valid.csv` * `test.csv` +#### data/ood/ + +외부 출처 OOD 평가 데이터 위치 + +* `ood_test.csv` + +#### data/hard_examples/ + +OOD 평가에서 틀린 FN/FP URL을 누적 저장하는 위치 + +* `hard_examples.csv` + --- ### saved/ @@ -73,6 +85,10 @@ http://paypa1-login.xyz,1 문자를 숫자로 바꾸는 사전 파일 +#### feature_norm.json + +tabular feature의 train mean/std 저장 파일 + --- ### src/ @@ -87,10 +103,14 @@ http://paypa1-login.xyz,1 | model.py | CharCNN 모델 구조 | | train.py | 모델 학습 (early stopping + best-model 저장) | | evaluate.py | 성능 평가 + FN/FP 사례 출력 | +| evaluate_ood.py | 외부 출처 OOD 평가 + threshold sweep | | predict.py | URL 단일 예측 | | explain.py | XAI 로그 기능 | +| features.py | URL feature 계산 + feature 정규화 저장/로드 | | download_phiusiil.py | UCI PhiUSIIL 데이터셋 다운로드 (추천) | | download_data.py | URLhaus + Tranco 데이터 다운로드 (대안) | +| download_train_mixed.py | PhiUSIIL + URLhaus + OpenPhish + PhishTank + Tranco 혼합 학습셋 생성 | +| download_ood.py | URLhaus + PhishTank + OpenPhish + Tranco OOD 평가셋 생성 | --- @@ -121,7 +141,24 @@ http://paypa1-login.xyz,1 * `0` = 정상 * `1` = 악성 -#### 옵션 A. 공개 데이터셋 자동 다운로드 (추천) +#### 옵션 A. 혼합 학습셋 자동 다운로드 (추천) + +PhiUSIIL, URLhaus, OpenPhish, PhishTank, Tranco를 섞어 학습 데이터를 만듭니다. +현재 기본 학습 파이프라인은 이 방식을 권장합니다. + +```bash +python src/download_train_mixed.py +``` + +기본 구성: + +* 악성 약 50,000개: PhiUSIIL phishing 25,000 + URLhaus recent 15,000 + PhishTank 10,000 + OpenPhish available feed +* 정상 약 50,000개: PhiUSIIL legitimate + Tranco top 100,000 중 샘플 +* 동적 URL augmentation 10,000개: 정상/악성 각각 5,000개 query/path/token 변형 +* 실행 결과: `data/raw/urls.csv` +* PhishTank app key가 있으면 환경변수 `PHISHTANK_APP_KEY`에 넣으면 됩니다. 없으면 public feed를 시도합니다. + +#### 옵션 B. PhiUSIIL만 사용 UCI ML PhiUSIIL Phishing URL Dataset (235k URL)에서 5만개 균형 샘플링: @@ -133,10 +170,17 @@ python src/download_phiusiil.py > 다른 소스도 가능: `python src/download_data.py` (URLhaus + Tranco). 다만 두 소스 형식 차이로 모델이 trivially 분리되어 baseline용으론 비추천. -#### 옵션 B. 직접 작성 +#### 옵션 C. 직접 작성 CSV 형식에 맞춰 `data/raw/urls.csv`를 수동으로 채워도 됩니다. +#### 추가로 찾아오면 좋은 데이터 출처 + +* 악성 URL: URLhaus `https://urlhaus.abuse.ch/`, PhishTank `https://www.phishtank.org/developer_info.php`, OpenPhish `https://openphish.com/phishing_feeds.html` +* 정상 URL: Tranco `https://tranco-list.eu/`, 회사/학교/정부/뉴스/쇼핑/포털 등 실제 정상 사이트 목록 +* 직접 가져온 CSV는 최소 `url,label` 컬럼을 맞추면 됩니다. 라벨은 `0=정상`, `1=악성`입니다. +* 정상 후보는 악성 feed와 겹치는 URL을 제거해야 합니다. + --- ### 3단계. 전처리 @@ -153,6 +197,9 @@ data/processed/valid.csv data/processed/test.csv ``` +기본 전처리는 같은 등록 도메인이 train/valid/test에 동시에 들어가지 않도록 domain group split을 사용합니다. 이 설정은 과적합과 데이터 누수를 줄이기 위한 것입니다. +URL feature는 query/path/token 같은 동적 URL 대응을 위해 `NoOfQueryParams`, `QueryLength`, `PathLength`, `NoOfPathSegments`, `HasFragment` 등을 포함합니다. + --- ### 4단계. 학습 @@ -166,6 +213,7 @@ python src/train.py ```text saved/charcnn.pt saved/char_vocab.json +saved/feature_norm.json ``` 예상 로그: @@ -195,26 +243,86 @@ python src/evaluate.py ```text === Evaluation Result === -Accuracy : 0.9982 -Precision: 1.0000 -Recall : 0.9964 -F1 Score : 0.9982 +Accuracy : 0.9975 +Precision: 0.9980 +Recall : 0.9969 +F1 Score : 0.9975 Confusion Matrix: -[[2500 0] - [ 9 2489]] +[[5509 11] + [ 17 5504]] -=== False Negatives (미탐: 악성을 정상으로 판단) [9건] === - score=0.2706 https://www.vmailmessage.com +=== False Negatives (미탐: 악성을 정상으로 판단) [17건] === + score=0.1295 https://verificahype.simply.site ... -=== False Positives (오탐: 정상을 악성으로 판단) [0건] === +=== False Positives (오탐: 정상을 악성으로 판단) [11건] === ``` * 지표 외에 **FN/FP 사례가 score와 함께 출력**되어 모델이 어떤 URL을 놓쳤는지 분석 가능합니다. --- -### 6단계. 단일 URL 예측 +### 6단계. 외부 출처 OOD 평가 + hard example 저장 + +PhiUSIIL 내부 split 성능은 실제 인터넷 URL 성능을 보장하지 않습니다. +외부 데이터로 별도 평가합니다. + +```bash +python src/download_ood.py +python src/evaluate_ood.py +``` + +기본 설정: + +* 악성: URLhaus recent 5,000개 +* 정상: Tranco 상위 100,000 도메인 중 5,000개 샘플 +* 저장 위치: `data/ood/ood_test.csv` +* `download_ood.py`는 기본적으로 `data/raw/urls.csv`에 들어간 학습 URL을 제외하고 OOD 평가셋을 만듭니다. +* 현재 모델은 실제 추론에서 계산 가능한 URL 기반 feature만 사용합니다. +* `evaluate_ood.py`는 기본적으로 오분류 FN/FP를 `data/hard_examples/hard_examples.csv`에 누적 저장합니다. + +hard example을 다음 학습에 반영하는 루프: + +```bash +python src/evaluate_ood.py --save-hard +python src/download_train_mixed.py +python src/preprocess.py +python src/train.py +python src/download_ood.py +python src/evaluate_ood.py +``` + +과적합 방지: + +* hard example은 클래스별 최대 `HARD_EXAMPLES_MAX_PER_CLASS`개만 학습에 섞습니다. +* hard example의 등록 도메인 group은 train에 고정하고 valid/test와 겹치지 않게 합니다. +* 같은 OOD 파일에 대해 무한 반복하지 말고, 최신 feed로 새 OOD를 만든 뒤 반복합니다. + +2026-04-28 현재 HybridCharCNN OOD 결과: + +```text +Threshold 0.20 +Accuracy : 0.9975 +Precision: 0.9988 +Recall : 0.9962 +F1 Score : 0.9975 +Confusion Matrix [[TN FP], [FN TP]]: +[[ 4994 6] + [ 19 4981]] +``` + +해석: + +* 이전 PhiUSIIL+HTML-feature 모델은 Tranco 정상 5,000개 중 4,981개를 오탐했습니다. +* 혼합 학습셋 + URL 기반 feature-only 모델로 바꾼 뒤 대량 오탐 문제는 해소됐습니다. +* 현재 OOD 악성은 URLhaus + PhishTank 잔여 URL을 포함합니다. OOD FN은 PhishTank 쪽에서 발생합니다. +* 동적 URL augmentation 후 OOD는 이전 FN 14 / FP 23에서 FN 12 / FP 15로 개선됐습니다. +* hard example 1회 재학습 후 OOD 오탐은 15개에서 6개로 줄었고, 내부 test도 FP 17개에서 11개로 줄었습니다. 대신 OOD FN은 12개에서 19개로 늘어 threshold/정책 선택이 필요합니다. +* OpenPhish community feed는 현재 샘플 수가 작아 추가 검증용으로는 부족합니다. 직접 수집 정상 URL과 최신 phishing feed를 계속 추가해야 합니다. + +--- + +### 7단계. 단일 URL 예측 ```bash python src/predict.py --url "http://paypa1-login.xyz" diff --git a/charcnn-kit/data/hard_examples/.gitkeep b/charcnn-kit/data/hard_examples/.gitkeep new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/charcnn-kit/data/hard_examples/.gitkeep @@ -0,0 +1 @@ + diff --git a/charcnn-kit/data/ood/.gitkeep b/charcnn-kit/data/ood/.gitkeep new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/charcnn-kit/data/ood/.gitkeep @@ -0,0 +1 @@ + diff --git a/charcnn-kit/src/config.py b/charcnn-kit/src/config.py index 8d0a4e4..a35fdee 100644 --- a/charcnn-kit/src/config.py +++ b/charcnn-kit/src/config.py @@ -15,7 +15,7 @@ DEVICE = "cuda" if torch.cuda.is_available() else "cpu" # pridict -THRESHOLD = 0.5 # score가 0.5 이상이면 악성으로 판단 +THRESHOLD = 0.2 # score가 0.2 이상이면 악성으로 판단 (FN/FP 균형) VOCAB_PATH = "saved/char_vocab.json" # 문자 사전 저장 경로 MODEL_PATH = "saved/charcnn.pt" # 모델 저장 경로 @@ -30,4 +30,67 @@ TRAIN_RATIO = 0.8 # 학습 데이터 비율 VALID_RATIO = 0.1 # 검증 데이터 비율 TEST_RATIO = 0.1 # 평가 데이터 비율 -SEED = 42 # 재현성을 위한 random seed \ No newline at end of file +SEED = 42 # 재현성을 위한 random seed + +# === OOD 평가 === +OOD_CSV_PATH = "data/ood/ood_test.csv" # 외부 출처 OOD 평가 데이터 +OOD_SAMPLE_PER_CLASS = 5000 # OOD 평가셋 클래스별 샘플 수 +OOD_BENIGN_TOP_N = 100000 # Tranco 정상 후보를 상위 N개 도메인으로 제한 + +# === 혼합 학습셋 / hard example 재학습 === +HARD_EXAMPLES_PATH = "data/hard_examples/hard_examples.csv" # OOD 오분류 누적 저장 +MIXED_PHIUSIIL_PER_CLASS = 25000 # 혼합 학습셋 PhiUSIIL 클래스별 샘플 수 +MIXED_URLHAUS_SAMPLES = 15000 # 혼합 학습셋 URLhaus 악성 샘플 수 +MIXED_OPENPHISH_SAMPLES = 5000 # 혼합 학습셋 OpenPhish 악성 샘플 수 +MIXED_PHISHTANK_SAMPLES = 10000 # 혼합 학습셋 PhishTank 악성 샘플 수 +MIXED_TRANCO_SAMPLES = 40000 # 혼합 학습셋 Tranco 정상 샘플 수 +MIXED_TRANCO_TOP_N = 100000 # 혼합 학습용 Tranco 정상 후보 상위 N개 +DYNAMIC_AUG_PER_CLASS = 5000 # 동적 URL(query/path/token) synthetic augmentation 수 +HARD_EXAMPLES_MAX_PER_CLASS = 2000 # 재학습에 섞을 hard example 클래스별 최대 수 +USE_DOMAIN_GROUP_SPLIT = True # 같은 등록 도메인이 train/valid/test에 섞이지 않게 분할 +FORCE_HARD_EXAMPLES_TO_TRAIN = True # hard example 도메인 group은 train에 고정 + +# === Tabular feature 통합 (이슈 #3) === +# URL 문자열만으로 즉시 계산 가능 → 학습/평가/추론 모두에서 사용 +URL_FEATURE_COLS = [ + "IsHTTPS", + "URLLength", + "DomainLength", + "NoOfSubDomain", + "IsDomainIP", + "TLDLength", + "NoOfLettersInURL", + "LetterRatioInURL", + "NoOfDegitsInURL", # PhiUSIIL 원본 오타 그대로 + "DegitRatioInURL", + "SpacialCharRatioInURL", + "NoOfOtherSpecialCharsInURL", + "HasObfuscation", + "NoOfEqualsInURL", + "NoOfQMarkInURL", + "NoOfAmpersandInURL", + "NoOfAtInURL", + "NoOfDashInURL", + "NoOfDotInURL", + "NoOfPercentInURL", + "PathLength", + "QueryLength", + "NoOfPathSegments", + "NoOfQueryParams", + "HasFragment", +] +# HTML/페이지 분석이 필요 → 학습/평가만 PhiUSIIL 사전계산값 사용, +# 단일 URL 추론에서는 train mean으로 imputation +# 주의: URLSimilarityIndex는 라벨과 |corr|=0.86으로 cheat-feature이라 제외함 +# (이슈 #3 ablation 결과; 다시 추가하지 말 것) +HTML_FEATURE_COLS = [ + "HasSocialNet", + "HasCopyrightInfo", + "HasDescription", +] +# 현재 배포/외부 평가에서는 HTML을 실제 fetch하지 않으므로 URL 기반 feature만 사용한다. +# HTML feature를 다시 쓰려면 predict/evaluate_ood에서도 동일 feature를 계산해야 한다. +FEATURE_COLS = URL_FEATURE_COLS + +FEATURE_NORM_PATH = "saved/feature_norm.json" # 학습 데이터의 mean/std 저장 +TABULAR_HIDDEN = 32 # tabular 분기의 hidden 차원 diff --git a/charcnn-kit/src/dataset.py b/charcnn-kit/src/dataset.py index 7871ba7..432760e 100644 --- a/charcnn-kit/src/dataset.py +++ b/charcnn-kit/src/dataset.py @@ -1,5 +1,7 @@ # url 문자열을 문자 단위 숫자 배열로 변환 +# (이슈 #3) feature_cols가 주어지면 tabular feature도 함께 반환 import json +import numpy as np import pandas as pd import torch from torch.utils.data import Dataset @@ -55,9 +57,16 @@ def encode_url(url, vocab, max_len=MAX_LEN): # pytorch dataloader가 읽을 수 있는 url데이터셋 클래스 -# 각 데이터는 (url 숫자 배열, label) 형으로 반환됨 +# 기본: (url 숫자 배열, label) +# feature_cols 지정 시: (url 숫자 배열, tabular feature 벡터, label) class URLDataset(Dataset): - def __init__(self, csv_path, vocab=None, build_new_vocab=False): + def __init__( + self, + csv_path, + vocab=None, + build_new_vocab=False, + feature_cols=None, + ): self.df = pd.read_csv(csv_path) self.df = self.df.dropna(subset=["url", "label"]) @@ -71,6 +80,30 @@ def __init__(self, csv_path, vocab=None, build_new_vocab=False): raise ValueError("vocab이 필요합니다.") self.vocab = vocab + self.feature_cols = list(feature_cols) if feature_cols else None + if self.feature_cols: + missing = [c for c in self.feature_cols if c not in self.df.columns] + if missing: + raise ValueError(f"CSV에 feature 컬럼이 없습니다: {missing}") + # 결측 row를 추가로 떨어뜨림 (preprocess에서 이미 처리되지만 안전망) + self.df = self.df.dropna(subset=self.feature_cols).reset_index(drop=True) + self.features = ( + self.df[self.feature_cols].astype("float32").to_numpy() + ) + else: + self.features = None + + def apply_normalization(self, mean: np.ndarray, std: np.ndarray): + """학습 데이터로 fit한 mean/std로 features를 표준화 (in-place). + + 표준화: (x - mean) / std → 각 feature를 평균 0, 표준편차 1로 맞춰 + 스케일이 다른 컬럼들이 학습을 망가뜨리지 않게 함. + """ + if self.features is None: + raise RuntimeError("feature_cols 없이 정규화는 불가합니다.") + std_safe = np.where(std == 0, 1.0, std) + self.features = ((self.features - mean) / std_safe).astype("float32") + def __len__(self): return len(self.df) @@ -79,8 +112,11 @@ def __getitem__(self, idx): label = self.df.iloc[idx]["label"] x = encode_url(url, self.vocab) - x = torch.tensor(x, dtype=torch.long) y = torch.tensor(label, dtype=torch.float32) - return x, y \ No newline at end of file + if self.features is None: + return x, y + + f = torch.from_numpy(self.features[idx]) + return x, f, y diff --git a/charcnn-kit/src/download_ood.py b/charcnn-kit/src/download_ood.py new file mode 100644 index 0000000..5ba38c3 --- /dev/null +++ b/charcnn-kit/src/download_ood.py @@ -0,0 +1,116 @@ +# 외부 출처 OOD 평가셋 생성 +# - 악성: URLhaus recent +# - 정상: Tranco top 1M +# +# 학습용 data/raw/urls.csv를 덮어쓰지 않고 data/ood/ood_test.csv에 저장한다. +import argparse +import os +import random + +import pandas as pd + +from config import OOD_BENIGN_TOP_N, OOD_CSV_PATH, OOD_SAMPLE_PER_CLASS, RAW_CSV_PATH, SEED +from download_data import fetch_benign, fetch_malicious +from download_train_mixed import fetch_openphish, fetch_phishtank, _safe_fetch + + +def _dedupe(urls: list[str]) -> list[str]: + seen = set() + out = [] + for url in urls: + key = str(url).strip() + if not key or key in seen: + continue + seen.add(key) + out.append(key) + return out + + +def _load_excluded_urls(path: str) -> set[str]: + if not path or not os.path.exists(path): + return set() + df = pd.read_csv(path, usecols=["url"]) + excluded = set(df["url"].dropna().astype(str).str.strip()) + print(f"[exclude] loaded {len(excluded)} URLs from {path}") + return excluded + + +def build_ood(csv_path: str, sample_per_class: int, benign_top_n: int, exclude_csv: str): + rng = random.Random(SEED) + + urlhaus = _dedupe(fetch_malicious()) + openphish = _dedupe(_safe_fetch("openphish", fetch_openphish)) + phishtank = _dedupe(_safe_fetch("phishtank", fetch_phishtank)) + malicious_rows = ( + [{"url": url, "source": "urlhaus_recent"} for url in urlhaus] + + [{"url": url, "source": "openphish"} for url in openphish] + + [{"url": url, "source": "phishtank"} for url in phishtank] + ) + benign = _dedupe(fetch_benign()) + if benign_top_n > 0: + benign = benign[:benign_top_n] + print(f"[benign] limited to top {benign_top_n} Tranco domains") + + excluded = _load_excluded_urls(exclude_csv) + if excluded: + before_mal = len(malicious_rows) + before_ben = len(benign) + malicious_rows = [row for row in malicious_rows if row["url"] not in excluded] + benign = [url for url in benign if url not in excluded] + print( + f"[exclude] removed train overlap: " + f"malicious {before_mal - len(malicious_rows)}, benign {before_ben - len(benign)}" + ) + + malicious_rows = list({row["url"]: row for row in malicious_rows}.values()) + malicious = [row["url"] for row in malicious_rows] + malicious_set = set(malicious) + benign = [url for url in benign if url not in malicious_set] + + if not malicious_rows or not benign: + raise RuntimeError("OOD 평가셋을 만들 수 없습니다. 외부 데이터 소스를 확인하세요.") + + target = min(sample_per_class, len(malicious_rows), len(benign)) + if target < sample_per_class: + print( + f"[warn] requested {sample_per_class} per class, " + f"but only {target} can be sampled" + ) + + malicious_rows = rng.sample(malicious_rows, target) + benign = rng.sample(benign, target) + + rows = ( + [{"url": row["url"], "label": 1, "source": row["source"]} for row in malicious_rows] + + [{"url": url, "label": 0, "source": "tranco_top"} for url in benign] + ) + rng.shuffle(rows) + + df = pd.DataFrame(rows) + os.makedirs(os.path.dirname(csv_path), exist_ok=True) + df.to_csv(csv_path, index=False) + + print( + f"saved {len(df)} rows to {csv_path} " + f"(악성 {target} / 정상 {target})" + ) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--csv", default=OOD_CSV_PATH) + parser.add_argument("--sample-per-class", type=int, default=OOD_SAMPLE_PER_CLASS) + parser.add_argument("--benign-top-n", type=int, default=OOD_BENIGN_TOP_N) + parser.add_argument("--exclude-csv", default=RAW_CSV_PATH) + args = parser.parse_args() + + build_ood( + csv_path=args.csv, + sample_per_class=args.sample_per_class, + benign_top_n=args.benign_top_n, + exclude_csv=args.exclude_csv, + ) + + +if __name__ == "__main__": + main() diff --git a/charcnn-kit/src/download_phiusiil.py b/charcnn-kit/src/download_phiusiil.py index 6e8f001..5461d38 100644 --- a/charcnn-kit/src/download_phiusiil.py +++ b/charcnn-kit/src/download_phiusiil.py @@ -12,7 +12,7 @@ import pandas as pd -from config import RAW_CSV_PATH, SEED +from config import RAW_CSV_PATH, SEED, FEATURE_COLS PHIUSIIL_ZIP_URL = ( "https://archive.ics.uci.edu/static/public/967/" @@ -66,8 +66,16 @@ def main(): dist = df["label"].value_counts().to_dict() print(f"PhiUSIIL label dist (raw): {dist}") - df_out = df[[url_col, "label"]].copy() - df_out.columns = ["url", "label"] + # 이슈 #3: URL/label 외에 학습에 사용할 tabular feature 컬럼도 함께 보존 + missing_feat = [c for c in FEATURE_COLS if c not in df.columns] + if missing_feat: + raise RuntimeError( + f"PhiUSIIL CSV에 기대한 feature 컬럼이 없습니다: {missing_feat}" + ) + + keep_cols = [url_col, "label"] + FEATURE_COLS + df_out = df[keep_cols].copy() + df_out.rename(columns={url_col: "url"}, inplace=True) # 라벨 반전: PhiUSIIL(1=legit, 0=phish) -> 본 프로젝트(0=정상, 1=악성) df_out["label"] = 1 - df_out["label"].astype(int) diff --git a/charcnn-kit/src/download_train_mixed.py b/charcnn-kit/src/download_train_mixed.py new file mode 100644 index 0000000..99b90f7 --- /dev/null +++ b/charcnn-kit/src/download_train_mixed.py @@ -0,0 +1,394 @@ +# 혼합 학습 데이터셋 생성 +# - 악성: PhiUSIIL phishing + URLhaus recent +# + OpenPhish community + PhishTank online-valid +# - 정상: PhiUSIIL legitimate + Tranco top domains +# +# 결과는 data/raw/urls.csv에 저장한다. label 규칙은 0=정상, 1=악성. +import argparse +import bz2 +import csv +import hashlib +import io +import os +import random +import zipfile +import urllib.parse +import urllib.request + +import pandas as pd + +from config import ( + DYNAMIC_AUG_PER_CLASS, + HARD_EXAMPLES_MAX_PER_CLASS, + HARD_EXAMPLES_PATH, + MIXED_OPENPHISH_SAMPLES, + MIXED_PHISHTANK_SAMPLES, + MIXED_PHIUSIIL_PER_CLASS, + MIXED_TRANCO_SAMPLES, + MIXED_TRANCO_TOP_N, + MIXED_URLHAUS_SAMPLES, + RAW_CSV_PATH, + SEED, +) +from download_data import fetch_benign, fetch_malicious +from download_phiusiil import PHIUSIIL_ZIP_URL, USER_AGENT, _find_url_col + +OPENPHISH_FEED_URL = "https://raw.githubusercontent.com/openphish/public_feed/refs/heads/main/feed.txt" +PHISHTANK_PUBLIC_CSV_BZ2_URL = "http://data.phishtank.com/data/online-valid.csv.bz2" +PHISHTANK_KEYED_CSV_BZ2_URL = "http://data.phishtank.com/data/{key}/online-valid.csv.bz2" +PHISHTANK_APP_KEY_ENV = "PHISHTANK_APP_KEY" + + +def _http_get_binary(url: str) -> bytes: + req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT}) + with urllib.request.urlopen(req, timeout=180) as resp: + return resp.read() + + +def _http_get_text(url: str) -> str: + return _http_get_binary(url).decode("utf-8", errors="replace") + + +def _dedupe(urls: list[str]) -> list[str]: + seen = set() + out = [] + for url in urls: + key = str(url).strip() + if not key or key in seen: + continue + seen.add(key) + out.append(key) + return out + + +def _sample(urls: list[str], n: int, rng: random.Random, name: str) -> list[str]: + urls = _dedupe(urls) + target = min(n, len(urls)) + if target < n: + print(f"[warn] {name}: requested {n}, available {target}") + return rng.sample(urls, target) + + +def _safe_fetch(name: str, fetcher) -> list[str]: + try: + return fetcher() + except Exception as exc: + print(f"[warn] {name} fetch failed: {exc}") + return [] + + +def fetch_phiusiil_urls(per_class: int, rng: random.Random) -> tuple[list[str], list[str]]: + print(f"[phiusiil] GET {PHIUSIIL_ZIP_URL}") + blob = _http_get_binary(PHIUSIIL_ZIP_URL) + print(f"[phiusiil] downloaded {len(blob) / 1024 / 1024:.1f} MB") + + with zipfile.ZipFile(io.BytesIO(blob)) as zf: + csv_name = next((n for n in zf.namelist() if n.lower().endswith(".csv")), None) + if csv_name is None: + raise RuntimeError("PhiUSIIL zip 안에 CSV가 없습니다.") + with zf.open(csv_name) as f: + df = pd.read_csv(f) + + url_col = _find_url_col(df.columns) + if url_col is None or "label" not in df.columns: + raise RuntimeError( + f"PhiUSIIL CSV에 url/label 컬럼을 찾지 못했습니다: {list(df.columns)}" + ) + + # PhiUSIIL raw label: 1=legitimate, 0=phishing + phish = df[df["label"] == 0][url_col].astype(str).tolist() + legit = df[df["label"] == 1][url_col].astype(str).tolist() + + phish = _sample(phish, per_class, rng, "phiusiil_phish") + legit = _sample(legit, per_class, rng, "phiusiil_legit") + print(f"[phiusiil] sampled phishing={len(phish)}, legitimate={len(legit)}") + return phish, legit + + +def fetch_openphish() -> list[str]: + print(f"[openphish] GET {OPENPHISH_FEED_URL}") + text = _http_get_text(OPENPHISH_FEED_URL) + urls = [ + line.strip() + for line in text.splitlines() + if line.strip() and not line.lstrip().startswith("#") + ] + print(f"[openphish] {len(urls)} URLs parsed") + return urls + + +def fetch_phishtank() -> list[str]: + key = os.environ.get(PHISHTANK_APP_KEY_ENV, "").strip() + if key: + url = PHISHTANK_KEYED_CSV_BZ2_URL.format(key=urllib.parse.quote(key, safe="")) + else: + url = PHISHTANK_PUBLIC_CSV_BZ2_URL + key_note = "with app key" if key else "without app key" + print(f"[phishtank] GET {url} ({key_note})") + blob = _http_get_binary(url) + text = bz2.decompress(blob).decode("utf-8", errors="replace") + + urls = [] + reader = csv.DictReader(io.StringIO(text)) + for row in reader: + if row.get("verified", "yes").lower() != "yes": + continue + if row.get("online", "yes").lower() != "yes": + continue + value = (row.get("url") or "").strip() + if value: + urls.append(value) + + print(f"[phishtank] {len(urls)} URLs parsed") + return urls + + +def _token(seed: str, length: int = 12) -> str: + return hashlib.sha1(seed.encode("utf-8", errors="ignore")).hexdigest()[:length] + + +def _append_query(url: str, params: list[tuple[str, str]]) -> str: + joiner = "&" if "?" in url else "?" + query = "&".join(f"{k}={v}" for k, v in params) + return f"{url}{joiner}{query}" + + +def _append_path(url: str, segment: str) -> str: + if "?" in url: + base, query = url.split("?", 1) + suffix = "?" + query + else: + base, suffix = url, "" + base = base.rstrip("/") + return f"{base}/{segment}{suffix}" + + +def _dynamic_variant(url: str, label: int, idx: int, rng: random.Random) -> str: + token = _token(f"{url}|{idx}|{label}") + if label == 0: + campaign = rng.choice(["spring", "newsletter", "catalog", "app", "promo"]) + mode = rng.randrange(6) + if mode == 0: + return _append_query(url, [("utm_source", "mail"), ("utm_campaign", campaign)]) + if mode == 1: + return _append_query(url, [("page", str((idx % 20) + 1)), ("sort", "recent")]) + if mode == 2: + return _append_path(url, f"products/{int(token[:6], 16) % 1000000}") + if mode == 3: + return _append_query(_append_path(url, "search"), [("q", "support"), ("lang", "en")]) + if mode == 4: + return _append_path(url, f"news/2026/{(idx % 12) + 1:02d}/{token[:6]}") + return _append_query(url, [("ref", token[:8]), ("view", "mobile")]) + + mode = rng.randrange(6) + if mode == 0: + return _append_path(url, f"login/{token[:10]}") + if mode == 1: + return _append_path(url, f"verify/account/{token[:8]}") + if mode == 2: + target = "https%3A%2F%2Fexample.com%2Fcontinue" + return _append_query(url, [("next", target), ("id", token[:10])]) + if mode == 3: + return _append_query(_append_path(url, "secure/update"), [("session", token)]) + if mode == 4: + return _append_path(url, f"invoice/{int(token[:8], 16) % 100000000}") + return _append_query(_append_path(url, f"u/{token[:6]}"), [("token", token)]) + + +def add_dynamic_augments(df: pd.DataFrame, per_class: int, rng: random.Random) -> pd.DataFrame: + if per_class <= 0: + return df + + rows = [] + for label in [0, 1]: + source = df[df["label"] == label] + target = min(per_class, len(source)) + if target <= 0: + continue + sampled = source.sample(n=target, random_state=SEED + label) + for i, (_, row) in enumerate(sampled.iterrows()): + rows.append({ + "url": _dynamic_variant(str(row["url"]), label, i, rng), + "label": label, + "source": f"dynamic_aug_{'malicious' if label else 'benign'}", + }) + + if not rows: + return df + + aug_df = pd.DataFrame(rows) + print( + f"[dynamic] added {len(aug_df)} synthetic dynamic URLs " + f"({int((aug_df['label'] == 1).sum())} malicious / " + f"{int((aug_df['label'] == 0).sum())} benign)" + ) + return pd.concat([df, aug_df], ignore_index=True) + + +def load_hard_examples(path: str, max_per_class: int, rng: random.Random) -> pd.DataFrame: + if not path or not os.path.exists(path): + print(f"[hard] no hard examples found: {path}") + return pd.DataFrame(columns=["url", "label", "source"]) + + df = pd.read_csv(path) + if "url" not in df.columns or "label" not in df.columns: + print(f"[hard] ignored invalid hard example CSV: {path}") + return pd.DataFrame(columns=["url", "label", "source"]) + + df = df.dropna(subset=["url", "label"]).copy() + df["url"] = df["url"].astype(str).str.strip() + df["label"] = df["label"].astype(int) + df = df[(df["url"].str.len() > 0) & (df["label"].isin([0, 1]))] + df = df.drop_duplicates(subset=["url", "label"], keep="last") + + pieces = [] + for label in [0, 1]: + part = df[df["label"] == label] + target = min(max_per_class, len(part)) + if target > 0: + pieces.append(part.sample(n=target, random_state=SEED + 100 + label)) + + if not pieces: + print(f"[hard] no usable hard examples in {path}") + return pd.DataFrame(columns=["url", "label", "source"]) + + out = pd.concat(pieces, ignore_index=True) + if "error_type" in out.columns: + out["source"] = out["error_type"].map({ + "false_negative": "hard_false_negative", + "false_positive": "hard_false_positive", + }).fillna("hard_example") + else: + out["source"] = "hard_example" + + out = out[["url", "label", "source"]].sample(frac=1, random_state=SEED).reset_index(drop=True) + print( + f"[hard] loaded {len(out)} hard examples from {path} " + f"(악성 {int((out['label'] == 1).sum())} / 정상 {int((out['label'] == 0).sum())})" + ) + return out + + +def _rebalance_preserving_hard(df: pd.DataFrame) -> pd.DataFrame: + hard_mask = df["source"].astype(str).str.startswith("hard_") + hard_df = df[hard_mask] + base_df = df[~hard_mask] + + pos_total = int((df["label"] == 1).sum()) + neg_total = int((df["label"] == 0).sum()) + target = min(pos_total, neg_total) + + pieces = [hard_df] + for label in [0, 1]: + hard_n = int((hard_df["label"] == label).sum()) + needed = max(0, target - hard_n) + part = base_df[base_df["label"] == label] + if len(part) <= needed: + pieces.append(part) + else: + pieces.append(part.sample(n=needed, random_state=SEED + 200 + label)) + + return pd.concat(pieces, ignore_index=True).sample(frac=1, random_state=SEED).reset_index(drop=True) + + +def build_mixed_train( + output: str, + phiusiil_per_class: int, + urlhaus_samples: int, + openphish_samples: int, + phishtank_samples: int, + tranco_samples: int, + dynamic_aug_per_class: int, + hard_examples_path: str, + hard_max_per_class: int, + tranco_top_n: int, +): + rng = random.Random(SEED) + + phi_phish, phi_legit = fetch_phiusiil_urls(phiusiil_per_class, rng) + + urlhaus = fetch_malicious() + openphish = _safe_fetch("openphish", fetch_openphish) + phishtank = _safe_fetch("phishtank", fetch_phishtank) + tranco = fetch_benign() + if tranco_top_n > 0: + tranco = tranco[:tranco_top_n] + print(f"[tranco] limited to top {tranco_top_n} domains") + + # 정상 후보에서 악성 URL과 정확히 같은 항목은 제거 + malicious_seen = set(_dedupe(urlhaus + openphish + phishtank + phi_phish)) + tranco = [url for url in tranco if url not in malicious_seen] + phi_legit = [url for url in phi_legit if url not in malicious_seen] + + urlhaus = _sample(urlhaus, urlhaus_samples, rng, "urlhaus") + openphish = _sample(openphish, openphish_samples, rng, "openphish") + phishtank = _sample(phishtank, phishtank_samples, rng, "phishtank") + tranco = _sample(tranco, tranco_samples, rng, "tranco") + + rows = ( + [{"url": url, "label": 1, "source": "phiusiil_phish"} for url in phi_phish] + + [{"url": url, "label": 0, "source": "phiusiil_legit"} for url in phi_legit] + + [{"url": url, "label": 1, "source": "urlhaus_recent"} for url in urlhaus] + + [{"url": url, "label": 1, "source": "openphish"} for url in openphish] + + [{"url": url, "label": 1, "source": "phishtank"} for url in phishtank] + + [{"url": url, "label": 0, "source": "tranco_top"} for url in tranco] + ) + rng.shuffle(rows) + + df = pd.DataFrame(rows).drop_duplicates(subset=["url"], keep="first") + # 중복 제거 후 클래스 균형 재조정 + pos = df[df["label"] == 1] + neg = df[df["label"] == 0] + target = min(len(pos), len(neg)) + df = pd.concat([ + pos.sample(n=target, random_state=SEED), + neg.sample(n=target, random_state=SEED), + ]).sample(frac=1, random_state=SEED).reset_index(drop=True) + df = add_dynamic_augments(df, dynamic_aug_per_class, rng) + hard_df = load_hard_examples(hard_examples_path, hard_max_per_class, rng) + if len(hard_df) > 0: + df = pd.concat([df, hard_df], ignore_index=True) + df = df.drop_duplicates(subset=["url"], keep="first").reset_index(drop=True) + df = _rebalance_preserving_hard(df) + + os.makedirs(os.path.dirname(output), exist_ok=True) + df.to_csv(output, index=False) + + print( + f"saved {len(df)} rows to {output} " + f"(악성 {int((df['label'] == 1).sum())} / 정상 {int((df['label'] == 0).sum())})" + ) + print("source distribution:") + print(df["source"].value_counts().to_string()) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--output", default=RAW_CSV_PATH) + parser.add_argument("--phiusiil-per-class", type=int, default=MIXED_PHIUSIIL_PER_CLASS) + parser.add_argument("--urlhaus-samples", type=int, default=MIXED_URLHAUS_SAMPLES) + parser.add_argument("--openphish-samples", type=int, default=MIXED_OPENPHISH_SAMPLES) + parser.add_argument("--phishtank-samples", type=int, default=MIXED_PHISHTANK_SAMPLES) + parser.add_argument("--tranco-samples", type=int, default=MIXED_TRANCO_SAMPLES) + parser.add_argument("--dynamic-aug-per-class", type=int, default=DYNAMIC_AUG_PER_CLASS) + parser.add_argument("--hard-examples", default=HARD_EXAMPLES_PATH) + parser.add_argument("--hard-max-per-class", type=int, default=HARD_EXAMPLES_MAX_PER_CLASS) + parser.add_argument("--tranco-top-n", type=int, default=MIXED_TRANCO_TOP_N) + args = parser.parse_args() + + build_mixed_train( + output=args.output, + phiusiil_per_class=args.phiusiil_per_class, + urlhaus_samples=args.urlhaus_samples, + openphish_samples=args.openphish_samples, + phishtank_samples=args.phishtank_samples, + tranco_samples=args.tranco_samples, + dynamic_aug_per_class=args.dynamic_aug_per_class, + hard_examples_path=args.hard_examples, + hard_max_per_class=args.hard_max_per_class, + tranco_top_n=args.tranco_top_n, + ) + + +if __name__ == "__main__": + main() diff --git a/charcnn-kit/src/evaluate.py b/charcnn-kit/src/evaluate.py index 293e0fd..ee5f39b 100644 --- a/charcnn-kit/src/evaluate.py +++ b/charcnn-kit/src/evaluate.py @@ -1,5 +1,6 @@ # 모델의 성능을 측정합니다. # accuracy, precision, recall, f1 +# (이슈 #3) HybridCharCNN + tabular feature 정규화 로드 사용 import torch from torch.utils.data import DataLoader from sklearn.metrics import ( @@ -15,34 +16,36 @@ MODEL_PATH, BATCH_SIZE, THRESHOLD, - TEST_CSV_PATH + TEST_CSV_PATH, + FEATURE_COLS, ) from dataset import URLDataset, load_vocab -from model import CharCNN +from model import HybridCharCNN +from features import load_norm def evaluate(): - # 1. vocab 로드 + # 1. vocab + 정규화 통계 로드 vocab = load_vocab() + mean, std = load_norm() - # 2. test dataset 생성 + # 2. test dataset (학습과 동일한 mean/std로 정규화) test_dataset = URLDataset( csv_path=TEST_CSV_PATH, - vocab=vocab + vocab=vocab, + feature_cols=FEATURE_COLS, ) + test_dataset.apply_normalization(mean, std) - test_loader = DataLoader( - test_dataset, - batch_size=BATCH_SIZE, - shuffle=False - ) + test_loader = DataLoader(test_dataset, batch_size=BATCH_SIZE, shuffle=False) # 3. 모델 로드 - model = CharCNN(vocab_size=len(vocab)).to(DEVICE) - model.load_state_dict( - torch.load(MODEL_PATH, map_location=DEVICE) - ) + model = HybridCharCNN( + vocab_size=len(vocab), + num_features=len(FEATURE_COLS), + ).to(DEVICE) + model.load_state_dict(torch.load(MODEL_PATH, map_location=DEVICE)) model.eval() y_true = [] @@ -51,10 +54,11 @@ def evaluate(): # 4. 예측 with torch.no_grad(): - for x, y in test_loader: + for x, f, y in test_loader: x = x.to(DEVICE) + f = f.to(DEVICE) - logits = model(x) + logits = model(x, f) scores = torch.sigmoid(logits) preds = (scores >= THRESHOLD).int().cpu().tolist() @@ -63,7 +67,7 @@ def evaluate(): y_pred.extend(preds) y_score.extend(scores.cpu().tolist()) - # 5. 지표 계산 + # 5. 지표 acc = accuracy_score(y_true, y_pred) prec = precision_score(y_true, y_pred, zero_division=0) rec = recall_score(y_true, y_pred, zero_division=0) @@ -71,17 +75,17 @@ def evaluate(): cm = confusion_matrix(y_true, y_pred) print("=== Evaluation Result ===") - print(f"Accuracy : {acc:.4f}") # 전체 중 맞춘 개수 - print(f"Precision: {prec:.4f}") # 악성으로 분류한 것 중 실제 악성 비율 - print(f"Recall : {rec:.4f}") # 실제 악성 url을 악성으로 분류한 개수 - print(f"F1 Score : {f1:.4f}") # precision+recall 균형 점수 + print(f"Accuracy : {acc:.4f}") + print(f"Precision: {prec:.4f}") + print(f"Recall : {rec:.4f}") + print(f"F1 Score : {f1:.4f}") print("Confusion Matrix:") print(cm) - # 6. 오분류 사례 출력 (FN: 악성을 정상으로, FP: 정상을 악성으로) + # 6. 오분류 출력 urls = test_dataset.df["url"].tolist() - fn_cases = [] # 미탐 (실제 1, 예측 0) - fp_cases = [] # 오탐 (실제 0, 예측 1) + fn_cases = [] + fp_cases = [] for url, t, p, s in zip(urls, y_true, y_pred, y_score): if t == 1 and p == 0: fn_cases.append((s, url)) @@ -89,7 +93,6 @@ def evaluate(): fp_cases.append((s, url)) print(f"\n=== False Negatives (미탐: 악성을 정상으로 판단) [{len(fn_cases)}건] ===") - # 점수 높은 순(0.5에 가까웠던 것부터)으로 출력 for s, url in sorted(fn_cases, key=lambda x: -x[0]): print(f" score={s:.4f} {url}") diff --git a/charcnn-kit/src/evaluate_ood.py b/charcnn-kit/src/evaluate_ood.py new file mode 100644 index 0000000..472e3e3 --- /dev/null +++ b/charcnn-kit/src/evaluate_ood.py @@ -0,0 +1,264 @@ +# 저장된 모델을 외부 출처 OOD 데이터셋으로 평가한다. +# CSV는 최소 url,label 컬럼이 필요하다. source 컬럼이 있으면 출처별 오류율도 출력한다. +import argparse +import os + +import numpy as np +import pandas as pd +import torch +from sklearn.metrics import accuracy_score, confusion_matrix, f1_score, precision_score, recall_score +from torch.utils.data import DataLoader, Dataset + +from config import ( + BATCH_SIZE, + DEVICE, + FEATURE_COLS, + HARD_EXAMPLES_PATH, + HTML_FEATURE_COLS, + MODEL_PATH, + OOD_CSV_PATH, + THRESHOLD, +) +from dataset import encode_url, load_vocab +from features import compute_url_features, load_norm, standardize +from model import HybridCharCNN + + +class OODDataset(Dataset): + def __init__(self, csv_path: str, vocab: dict, mean: np.ndarray, std: np.ndarray): + df = pd.read_csv(csv_path) + if "url" not in df.columns or "label" not in df.columns: + raise ValueError("OOD CSV에는 url,label 컬럼이 필요합니다.") + + df = df.dropna(subset=["url", "label"]).copy() + df["url"] = df["url"].astype(str).str.strip() + df["label"] = df["label"].astype(int) + df = df[(df["url"].str.len() > 0) & (df["label"].isin([0, 1]))].reset_index(drop=True) + if len(df) == 0: + raise ValueError("평가할 OOD row가 없습니다.") + + self.df = df + self.vocab = vocab + self.mean = mean + self.std = std + self.features = self._build_features() + + def _build_features(self) -> np.ndarray: + rows = [] + for _, row in self.df.iterrows(): + url_feats = compute_url_features(row["url"]) + raw = np.zeros(len(FEATURE_COLS), dtype="float32") + + for i, col in enumerate(FEATURE_COLS): + if col in url_feats: + raw[i] = url_feats[col] + elif col in self.df.columns and not pd.isna(row[col]): + raw[i] = float(row[col]) + else: + raw[i] = self.mean[i] + + rows.append(raw) + + features = np.stack(rows) + return standardize(features, self.mean, self.std) + + def __len__(self): + return len(self.df) + + def __getitem__(self, idx: int): + url = self.df.iloc[idx]["url"] + label = self.df.iloc[idx]["label"] + + x = torch.tensor(encode_url(url, self.vocab), dtype=torch.long) + f = torch.from_numpy(self.features[idx]) + y = torch.tensor(label, dtype=torch.float32) + return x, f, y + + +def _metrics(y_true: list[int], y_score: list[float], threshold: float) -> dict: + y_pred = [1 if s >= threshold else 0 for s in y_score] + tn, fp, fn, tp = confusion_matrix(y_true, y_pred, labels=[0, 1]).ravel() + return { + "threshold": threshold, + "accuracy": accuracy_score(y_true, y_pred), + "precision": precision_score(y_true, y_pred, zero_division=0), + "recall": recall_score(y_true, y_pred, zero_division=0), + "f1": f1_score(y_true, y_pred, zero_division=0), + "tn": int(tn), + "fp": int(fp), + "fn": int(fn), + "tp": int(tp), + } + + +def _parse_thresholds(raw: str) -> list[float]: + vals = [] + for part in raw.split(","): + part = part.strip() + if not part: + continue + val = float(part) + if not 0.0 <= val <= 1.0: + raise ValueError(f"threshold는 0~1 사이여야 합니다: {val}") + vals.append(val) + return vals or [THRESHOLD] + + +def _print_metrics(name: str, m: dict): + print(f"=== {name} (threshold={m['threshold']:.2f}) ===") + print(f"Accuracy : {m['accuracy']:.4f}") + print(f"Precision: {m['precision']:.4f}") + print(f"Recall : {m['recall']:.4f}") + print(f"F1 Score : {m['f1']:.4f}") + print("Confusion Matrix [[TN FP], [FN TP]]:") + print(f"[[{m['tn']:>5} {m['fp']:>5}]") + print(f" [{m['fn']:>5} {m['tp']:>5}]]") + + +def _print_score_summary(df: pd.DataFrame): + print("\n=== Score Distribution ===") + group_cols = ["label"] + if "source" in df.columns: + group_cols.append("source") + + for key, group in df.groupby(group_cols): + scores = group["score"].to_numpy() + q = np.quantile(scores, [0.0, 0.25, 0.5, 0.75, 0.9, 0.99, 1.0]) + # group_cols가 1개면 key는 scalar, 2개 이상이면 tuple로 옴 → 항상 tuple로 정규화 + key_tuple = key if isinstance(key, tuple) else (key,) + label_str = " | ".join(f"{c}={v}" for c, v in zip(group_cols, key_tuple)) + print( + f"{label_str}: n={len(group)} " + f"min={q[0]:.4f} p25={q[1]:.4f} med={q[2]:.4f} " + f"p75={q[3]:.4f} p90={q[4]:.4f} p99={q[5]:.4f} max={q[6]:.4f}" + ) + + +def _save_hard_examples(df: pd.DataFrame, csv_path: str, output_path: str): + hard = df[df["label"] != df["pred"]].copy() + if len(hard) == 0: + print(f"\n[hard] no hard examples to save ({output_path})") + return + + hard["error_type"] = np.where(hard["label"] == 1, "false_negative", "false_positive") + hard["eval_csv"] = csv_path + hard["threshold"] = THRESHOLD + + keep_cols = ["url", "label", "source", "score", "pred", "error_type", "eval_csv", "threshold"] + for col in keep_cols: + if col not in hard.columns: + hard[col] = "" + hard = hard[keep_cols] + + if os.path.exists(output_path): + old = pd.read_csv(output_path) + hard = pd.concat([old, hard], ignore_index=True) + + hard = hard.dropna(subset=["url", "label"]) + hard["url"] = hard["url"].astype(str).str.strip() + hard["label"] = hard["label"].astype(int) + hard = hard.drop_duplicates(subset=["url", "label"], keep="last").reset_index(drop=True) + + os.makedirs(os.path.dirname(output_path), exist_ok=True) + hard.to_csv(output_path, index=False) + + dist = hard["error_type"].value_counts().to_dict() + print(f"\n[hard] saved {len(hard)} cumulative hard examples to {output_path}: {dist}") + + +def evaluate_ood(csv_path: str, thresholds: list[float], max_errors: int, save_hard: bool, hard_path: str): + vocab = load_vocab() + mean, std = load_norm() + + dataset = OODDataset(csv_path=csv_path, vocab=vocab, mean=mean, std=std) + loader = DataLoader(dataset, batch_size=BATCH_SIZE, shuffle=False) + + model = HybridCharCNN(vocab_size=len(vocab), num_features=len(FEATURE_COLS)).to(DEVICE) + model.load_state_dict(torch.load(MODEL_PATH, map_location=DEVICE)) + model.eval() + + y_true = [] + y_score = [] + with torch.no_grad(): + for x, f, y in loader: + x = x.to(DEVICE) + f = f.to(DEVICE) + logits = model(x, f) + scores = torch.sigmoid(logits) + + y_true.extend(y.int().tolist()) + y_score.extend(scores.cpu().tolist()) + + print(f"loaded {len(dataset)} OOD rows from {csv_path}") + active_html = [c for c in HTML_FEATURE_COLS if c in FEATURE_COLS] + if active_html: + print(f"HTML features without CSV values are imputed to train mean: {active_html}") + else: + print("HTML features are inactive; OOD evaluation uses URL-derived features only.") + + print("\n=== Threshold Sweep ===") + print("thr acc prec recall f1 FN FP") + for threshold in thresholds: + m = _metrics(y_true, y_score, threshold) + print( + f"{threshold:0.3f} {m['accuracy']:.4f} {m['precision']:.4f} " + f"{m['recall']:.4f} {m['f1']:.4f} {m['fn']:>5} {m['fp']:>5}" + ) + + selected = _metrics(y_true, y_score, THRESHOLD) + print() + _print_metrics("Selected Config Threshold", selected) + + df = dataset.df.copy() + df["score"] = y_score + df["pred"] = (df["score"] >= THRESHOLD).astype(int) + + _print_score_summary(df) + + if "source" in df.columns: + print("\n=== Source Breakdown ===") + for source, group in df.groupby("source"): + n = len(group) + pos = int((group["label"] == 1).sum()) + fp = int(((group["label"] == 0) & (group["pred"] == 1)).sum()) + fn = int(((group["label"] == 1) & (group["pred"] == 0)).sum()) + print(f"{source}: n={n}, malicious={pos}, FN={fn}, FP={fp}") + + fn_cases = df[(df["label"] == 1) & (df["pred"] == 0)].sort_values("score", ascending=False) + fp_cases = df[(df["label"] == 0) & (df["pred"] == 1)].sort_values("score", ascending=False) + + print(f"\n=== False Negatives [{len(fn_cases)}건, top {max_errors}] ===") + for _, row in fn_cases.head(max_errors).iterrows(): + print(f" score={row['score']:.4f} {row['url']}") + + print(f"\n=== False Positives [{len(fp_cases)}건, top {max_errors}] ===") + for _, row in fp_cases.head(max_errors).iterrows(): + print(f" score={row['score']:.4f} {row['url']}") + + if save_hard: + _save_hard_examples(df, csv_path=csv_path, output_path=hard_path) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--csv", default=OOD_CSV_PATH) + parser.add_argument( + "--thresholds", + default="0.10,0.20,0.30,0.40,0.50,0.60,0.70,0.80,0.90,0.95,0.99,0.999", + ) + parser.add_argument("--max-errors", type=int, default=20) + parser.add_argument("--save-hard", action=argparse.BooleanOptionalAction, default=True) + parser.add_argument("--hard-path", default=HARD_EXAMPLES_PATH) + args = parser.parse_args() + + evaluate_ood( + csv_path=args.csv, + thresholds=_parse_thresholds(args.thresholds), + max_errors=args.max_errors, + save_hard=args.save_hard, + hard_path=args.hard_path, + ) + + +if __name__ == "__main__": + main() diff --git a/charcnn-kit/src/explain.py b/charcnn-kit/src/explain.py index 2ffa27e..0c5bc9b 100644 --- a/charcnn-kit/src/explain.py +++ b/charcnn-kit/src/explain.py @@ -6,13 +6,14 @@ from captum.attr import LayerIntegratedGradients -def get_xai_log(model, x, url: str, top_k: int = 10) -> dict: +def get_xai_log(model, x, url: str, top_k: int = 10, additional_forward_args=None) -> dict: lig = LayerIntegratedGradients(model, model.embedding) attributions = lig.attribute( inputs=x, - baselines=torch.zeros_like(x) + baselines=torch.zeros_like(x), + additional_forward_args=additional_forward_args, ) # [1, MAX_LEN, EMBED_DIM] -> [MAX_LEN] diff --git a/charcnn-kit/src/features.py b/charcnn-kit/src/features.py new file mode 100644 index 0000000..8d96727 --- /dev/null +++ b/charcnn-kit/src/features.py @@ -0,0 +1,119 @@ +# Tabular feature 추출 / 정규화 유틸 (이슈 #3) +# +# - URL 문자열만으로 즉시 계산되는 feature(URL_FEATURE_COLS)는 전처리/추론에서 같은 함수로 계산 +# - 페이지 fetch가 필요한 3개(HTML_FEATURE_COLS)는 현재 FEATURE_COLS에서 비활성화 +import json +import re +from urllib.parse import parse_qsl, urlparse + +import numpy as np + +from config import ( + URL_FEATURE_COLS, + HTML_FEATURE_COLS, + FEATURE_COLS, + FEATURE_NORM_PATH, +) + +_IP_RE = re.compile(r"^(?:\d{1,3}\.){3}\d{1,3}$") +_PERCENT_HEX_RE = re.compile(r"%[0-9a-fA-F]{2}") + + +def compute_url_features(url: str) -> dict: + """URL 문자열만으로 계산되는 feature 반환. + + PhiUSIIL과 정확히 동일한 정의를 보장하지는 않지만(공식 정의 미공개), + preprocess.py와 predict.py가 동일 함수를 쓰므로 학습/추론 일관성은 유지된다. + """ + url = str(url) + parsed = urlparse(url if "://" in url else "http://" + url) + domain = parsed.netloc.split(":")[0] + parts = [p for p in domain.split(".") if p] + tld = parts[-1] if parts else "" + path_segments = [p for p in parsed.path.split("/") if p] + query_params = parse_qsl(parsed.query, keep_blank_values=True) + + n = len(url) + n_letters = sum(c.isalpha() for c in url) + n_digits = sum(c.isdigit() for c in url) + # "Other special": 알파벳/숫자/일반 URL 구분자가 아닌 모든 문자 + n_other_special = sum( + 1 for c in url + if not c.isalnum() and c not in "/:?#&=._-" + ) + + return { + "IsHTTPS": float(parsed.scheme.lower() == "https"), + "URLLength": float(n), + "DomainLength": float(len(domain)), + # 서브도메인 수: "www.x.com" -> 1, "a.b.x.com" -> 2 (최소 0 보장) + "NoOfSubDomain": float(max(0, len(parts) - 2)), + "IsDomainIP": float(bool(_IP_RE.match(domain))), + "TLDLength": float(len(tld)), + "NoOfLettersInURL": float(n_letters), + "LetterRatioInURL": round(n_letters / n, 3) if n else 0.0, + # PhiUSIIL이 사용하는 오타 컬럼명 그대로 유지 + "NoOfDegitsInURL": float(n_digits), + "DegitRatioInURL": round(n_digits / n, 3) if n else 0.0, + "SpacialCharRatioInURL": round(n_other_special / n, 3) if n else 0.0, + "NoOfOtherSpecialCharsInURL": float(n_other_special), + "HasObfuscation": float(bool(_PERCENT_HEX_RE.search(url))), + "NoOfEqualsInURL": float(url.count("=")), + "NoOfQMarkInURL": float(url.count("?")), + "NoOfAmpersandInURL": float(url.count("&")), + "NoOfAtInURL": float(url.count("@")), + "NoOfDashInURL": float(url.count("-")), + "NoOfDotInURL": float(url.count(".")), + "NoOfPercentInURL": float(url.count("%")), + "PathLength": float(len(parsed.path)), + "QueryLength": float(len(parsed.query)), + "NoOfPathSegments": float(len(path_segments)), + "NoOfQueryParams": float(len(query_params)), + "HasFragment": float(bool(parsed.fragment)), + } + + +def save_norm(mean: np.ndarray, std: np.ndarray, path: str = FEATURE_NORM_PATH): + payload = { + "cols": list(FEATURE_COLS), + "mean": [float(x) for x in mean], + "std": [float(x) for x in std], + } + with open(path, "w", encoding="utf-8") as f: + json.dump(payload, f, indent=2) + + +def load_norm(path: str = FEATURE_NORM_PATH): + with open(path, "r", encoding="utf-8") as f: + payload = json.load(f) + if payload["cols"] != list(FEATURE_COLS): + raise ValueError( + "feature 컬럼 구성이 저장 시점과 다릅니다.\n" + f" saved={payload['cols']}\n" + f" current={list(FEATURE_COLS)}" + ) + mean = np.array(payload["mean"], dtype="float32") + std = np.array(payload["std"], dtype="float32") + return mean, std + + +def standardize(features: np.ndarray, mean: np.ndarray, std: np.ndarray) -> np.ndarray: + """표준화: (x - mean) / std. std=0인 컬럼은 1로 치환해 division-by-zero 방지.""" + std_safe = np.where(std == 0, 1.0, std) + return ((features - mean) / std_safe).astype("float32") + + +def assemble_inference_vector(url: str, mean: np.ndarray, std: np.ndarray) -> np.ndarray: + """단일 URL 추론용 feature 벡터 생성. + + - URL_FEATURE_COLS: compute_url_features로 직접 계산 + - URL로 계산할 수 없는 active feature: train mean으로 imputation (표준화 후 0) + """ + url_feats = compute_url_features(url) + raw = np.zeros(len(FEATURE_COLS), dtype="float32") + for i, col in enumerate(FEATURE_COLS): + if col in url_feats: + raw[i] = url_feats[col] + else: + raw[i] = mean[i] + return standardize(raw, mean, std) diff --git a/charcnn-kit/src/inspect_phiusiil.py b/charcnn-kit/src/inspect_phiusiil.py new file mode 100644 index 0000000..73de91b --- /dev/null +++ b/charcnn-kit/src/inspect_phiusiil.py @@ -0,0 +1,64 @@ +# PhiUSIIL CSV의 전체 컬럼/타입/샘플값을 한눈에 확인하기 위한 일회성 점검 스크립트. +# 이슈 #3(미사용 feature 통합)에서 어떤 컬럼을 가져올지 결정하는 데에만 사용한다. +import io +import zipfile +import urllib.request + +import pandas as pd + +PHIUSIIL_ZIP_URL = ( + "https://archive.ics.uci.edu/static/public/967/" + "phiusiil+phishing+url+dataset.zip" +) +USER_AGENT = "Mozilla/5.0 (charcnn-kit dataset downloader)" +HTTP_TIMEOUT = 180 + + +REPORT_PATH = "data/phiusiil_columns_report.txt" + + +def main(): + print(f"GET {PHIUSIIL_ZIP_URL}") + req = urllib.request.Request(PHIUSIIL_ZIP_URL, headers={"User-Agent": USER_AGENT}) + with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp: + blob = resp.read() + print(f"downloaded {len(blob) / 1024 / 1024:.1f} MB") + + with zipfile.ZipFile(io.BytesIO(blob)) as zf: + csv_name = next(n for n in zf.namelist() if n.lower().endswith(".csv")) + with zf.open(csv_name) as f: + df = pd.read_csv(f) + + lines = [] + lines.append(f"rows={len(df)}, cols={len(df.columns)}") + + if "label" in df.columns: + lines.append(f"label dist (raw): {df['label'].value_counts().to_dict()}") + + lines.append("") + lines.append("=== columns ===") + lines.append(f"{'#':>3} {'name':<28} {'dtype':<10} {'nans':>6} {'nuniq':>7} sample") + for i, col in enumerate(df.columns): + s = df[col] + sample = s.dropna().head(3).tolist() + sample_str = ", ".join(repr(x)[:30] for x in sample) + lines.append( + f"{i:>3} {col:<28} {str(s.dtype):<10} " + f"{int(s.isna().sum()):>6} {int(s.nunique()):>7} {sample_str}" + ) + + if "label" in df.columns: + num_cols = df.select_dtypes(include="number").columns.drop("label", errors="ignore") + corr = df[num_cols].corrwith(df["label"]).abs().sort_values(ascending=False) + lines.append("") + lines.append("=== |corr(feature, raw_label)| (1=legit, 0=phish) ===") + for name, v in corr.items(): + lines.append(f" {v:.4f} {name}") + + with open(REPORT_PATH, "w", encoding="utf-8") as f: + f.write("\n".join(lines)) + print(f"wrote report -> {REPORT_PATH} ({len(lines)} lines)") + + +if __name__ == "__main__": + main() diff --git a/charcnn-kit/src/model.py b/charcnn-kit/src/model.py index 389164c..ef16b96 100644 --- a/charcnn-kit/src/model.py +++ b/charcnn-kit/src/model.py @@ -7,7 +7,8 @@ EMBED_DIM, NUM_FILTERS, KERNEL_SIZES, - DROPOUT + DROPOUT, + TABULAR_HIDDEN, ) @@ -72,4 +73,47 @@ def forward(self, x): # 다른 곳에서 sigmoid를 씌우면 확률이 됩니다. # 각 url의 점수를 출력 - return x.squeeze(1) \ No newline at end of file + return x.squeeze(1) + + +# 이슈 #3: CharCNN + tabular feature 융합 모델 (late fusion). +# CharCNN 본체 구조는 건드리지 않고, 그 안의 sub-module을 재사용해 자체 forward를 구성. +# 출력 직전에 tabular 분기와 concat → 최종 fc. +class HybridCharCNN(nn.Module): + def __init__(self, vocab_size: int, num_features: int): + super().__init__() + # CharCNN을 만들어 sub-module(embedding/convs/dropout)만 재사용. + # CharCNN.fc는 입력 차원이 다르므로 사용하지 않고 새 fc로 교체. + base = CharCNN(vocab_size) + self.embedding = base.embedding # explain.py가 model.embedding 참조 + self.convs = base.convs + self.dropout = base.dropout + char_out_dim = base.fc.in_features # NUM_FILTERS * len(KERNEL_SIZES) + + # tabular 분기: 표준화된 feature 벡터 → 작은 MLP + self.tab_branch = nn.Sequential( + nn.Linear(num_features, TABULAR_HIDDEN), + nn.ReLU(), + nn.Dropout(DROPOUT), + ) + + # 최종 fc: char vector + tab vector concat + self.fc = nn.Linear(char_out_dim + TABULAR_HIDDEN, 1) + + def forward(self, x_url, x_tab): + e = self.embedding(x_url) + e = e.permute(0, 2, 1) # [B, embed, len] + + pooled = [] + for conv in self.convs: + c = conv(e) + c = torch.relu(c) + c = torch.max(c, dim=2)[0] + pooled.append(c) + char_vec = torch.cat(pooled, dim=1) + char_vec = self.dropout(char_vec) + + tab_vec = self.tab_branch(x_tab) + + z = torch.cat([char_vec, tab_vec], dim=1) + return self.fc(z).squeeze(1) diff --git a/charcnn-kit/src/predict.py b/charcnn-kit/src/predict.py index f96bfcf..c7e5c05 100644 --- a/charcnn-kit/src/predict.py +++ b/charcnn-kit/src/predict.py @@ -1,47 +1,59 @@ # 모델 검사 테스트용. # url 하나를 넣으면 score/label을 출력합니다. +# (이슈 #3) HybridCharCNN: URL-derived feature는 직접 계산. import argparse import torch -from config import DEVICE, MODEL_PATH, THRESHOLD +from config import DEVICE, MODEL_PATH, THRESHOLD, FEATURE_COLS from dataset import load_vocab, encode_url -from model import CharCNN +from model import HybridCharCNN +from features import load_norm, assemble_inference_vector from explain import get_xai_log - def predict_url(url: str, use_xai: bool = False) -> dict: - vocab = load_vocab() # !!예측 시에 새 vocab 만들면 안 됨 + vocab = load_vocab() # !!예측 시에 새 vocab 만들면 안 됨 + mean, std = load_norm() # 모델 구조 생성 - model = CharCNN(vocab_size=len(vocab)).to(DEVICE) - + model = HybridCharCNN( + vocab_size=len(vocab), + num_features=len(FEATURE_COLS), + ).to(DEVICE) + # 학습된 모델 가중치 로드 model.load_state_dict(torch.load(MODEL_PATH, map_location=DEVICE)) - - # 예측 모드로 전환 model.eval() - # 분석할 url 전처리 + # URL 전처리 encoded = encode_url(url, vocab) x = torch.tensor([encoded], dtype=torch.long).to(DEVICE) - # 예측 실행 + # tabular feature: URL은 직접 계산 / HTML은 train mean으로 채워 정규화 + feat_vec = assemble_inference_vector(url, mean, std) + f = torch.from_numpy(feat_vec).unsqueeze(0).to(DEVICE) + + # 예측 with torch.no_grad(): - logit = model(x) + logit = model(x, f) score = torch.sigmoid(logit).item() result = { "url": url, "score": round(score, 4), - "label": "malicious" if score >= THRESHOLD else "normal" + "label": "malicious" if score >= THRESHOLD else "normal", } if use_xai: + # HybridCharCNN.forward(x_url, x_tab)이라 captum에 x_tab을 함께 넘겨야 함. + # 의도적으로 IG는 char embedding에만 적용 (baseline=zeros_like(x_url)). + # tabular feature는 attribution 대상이 아니라 고정값으로 통과시킨다 — + # XAI 결과가 char-level에 흐릿하게 나오면 모델이 tabular 분기에 더 의존한다는 신호. result["xai"] = get_xai_log( model=model, x=x, - url=url + url=url, + additional_forward_args=(f,), ) return result @@ -54,4 +66,4 @@ def predict_url(url: str, use_xai: bool = False) -> dict: args = parser.parse_args() result = predict_url(args.url, use_xai=args.xai) - print(result) \ No newline at end of file + print(result) diff --git a/charcnn-kit/src/preprocess.py b/charcnn-kit/src/preprocess.py index 0ef444b..d9958dc 100644 --- a/charcnn-kit/src/preprocess.py +++ b/charcnn-kit/src/preprocess.py @@ -1,8 +1,10 @@ # 원본 url 데이터를 학습/검증/평가용으로 분할 # data/raw/urls.csv -> data/processed/{train,valid,test}.csv import os +from urllib.parse import urlparse + import pandas as pd -from sklearn.model_selection import train_test_split +from sklearn.model_selection import StratifiedGroupKFold, train_test_split from config import ( RAW_CSV_PATH, @@ -13,7 +15,13 @@ VALID_RATIO, TEST_RATIO, SEED, + FORCE_HARD_EXAMPLES_TO_TRAIN, + USE_DOMAIN_GROUP_SPLIT, + URL_FEATURE_COLS, + HTML_FEATURE_COLS, + FEATURE_COLS, ) +from features import compute_url_features def _validate_ratios(): @@ -61,6 +69,29 @@ def _load_and_clean(raw_path: str) -> pd.DataFrame: # 5. 중복 url 제거 (동일 url에 다른 라벨이 있으면 첫 row만 유지) df = df.drop_duplicates(subset=["url"], keep="first").reset_index(drop=True) + # 6. URL 기반 feature는 학습/평가/추론 모두 같은 함수 정의를 쓰도록 재계산한다. + url_features = pd.DataFrame( + [compute_url_features(url) for url in df["url"]], + index=df.index, + ) + for col in URL_FEATURE_COLS: + df[col] = url_features[col].astype("float32") + + # HTML feature를 FEATURE_COLS에 다시 활성화한 경우, 직접 만든 url,label CSV처럼 + # 값이 없으면 0으로 채워 학습을 계속 진행할 수 있게 한다. + active_html = [c for c in HTML_FEATURE_COLS if c in FEATURE_COLS] + missing_html = [c for c in active_html if c not in df.columns] + if missing_html: + for col in missing_html: + df[col] = 0.0 + print(f"[features] missing HTML features filled with 0.0: {missing_html}") + + # 7. tabular feature NaN row 제거 (주로 HTML feature 안전망) + before_feat = len(df) + df = df.dropna(subset=FEATURE_COLS).reset_index(drop=True) + if len(df) < before_feat: + print(f"[clean] feature NaN row {before_feat - len(df)}개 추가 제거") + after = len(df) print(f"[clean] {before} -> {after} rows ({before - after}개 제거)") @@ -68,6 +99,11 @@ def _load_and_clean(raw_path: str) -> pd.DataFrame: def _stratified_split(df: pd.DataFrame): + if USE_DOMAIN_GROUP_SPLIT: + grouped = _domain_group_split(df) + if grouped is not None: + return grouped + # train vs (valid+test) rest_ratio = VALID_RATIO + TEST_RATIO train_df, rest_df = train_test_split( @@ -89,6 +125,75 @@ def _stratified_split(df: pd.DataFrame): return train_df, valid_df, test_df +def _domain_group(url: str) -> str: + parsed = urlparse(url if "://" in url else "http://" + url) + host = (parsed.hostname or str(url)).lower().strip(".") + if host.startswith("www."): + host = host[4:] + + labels = [p for p in host.split(".") if p] + if len(labels) <= 2: + return host + + # 간단한 eTLD+1 근사. publicsuffix 없이 흔한 ccTLD 2단계 도메인 누수만 줄인다. + second_level = {"ac", "co", "com", "edu", "go", "gov", "ne", "net", "or", "org"} + if len(labels[-1]) == 2 and labels[-2] in second_level and len(labels) >= 3: + return ".".join(labels[-3:]) + return ".".join(labels[-2:]) + + +def _domain_group_split(df: pd.DataFrame): + groups = df["url"].map(_domain_group) + n_groups = groups.nunique() + if n_groups < 10: + print(f"[split] domain group split skipped: only {n_groups} groups") + return None + + # 10-fold로 나눈 뒤 1 fold=test, 1 fold=valid, 나머지=train. + # 같은 등록 도메인이 서로 다른 split에 섞이지 않는다. + splitter = StratifiedGroupKFold(n_splits=10, shuffle=True, random_state=SEED) + fold_id = pd.Series(-1, index=df.index) + for fold, (_, test_idx) in enumerate(splitter.split(df, df["label"], groups)): + fold_id.iloc[test_idx] = fold + + if (fold_id < 0).any(): + print("[split] domain group split failed; falling back to row split") + return None + + hard_groups = set() + if FORCE_HARD_EXAMPLES_TO_TRAIN and "source" in df.columns: + source = df["source"].fillna("").astype(str) + hard_groups = set(groups[source.str.startswith("hard_")]) + if hard_groups: + fold_id.loc[groups.isin(hard_groups)] = 2 + print(f"[split] forced hard example groups to train: {len(hard_groups)}") + + test_df = df[fold_id == 0].copy() + valid_df = df[fold_id == 1].copy() + train_df = df[fold_id >= 2].copy() + + for name, part in [("train", train_df), ("valid", valid_df), ("test", test_df)]: + if len(part) == 0 or part["label"].nunique() < 2: + print(f"[split] domain group split invalid for {name}; falling back to row split") + return None + + train_groups = set(groups.loc[train_df.index]) + valid_groups = set(groups.loc[valid_df.index]) + test_groups = set(groups.loc[test_df.index]) + overlap = ( + len(train_groups & valid_groups) + + len(train_groups & test_groups) + + len(valid_groups & test_groups) + ) + print(f"[split] domain group split enabled: groups={n_groups}, overlap={overlap}") + + return ( + train_df.reset_index(drop=True), + valid_df.reset_index(drop=True), + test_df.reset_index(drop=True), + ) + + def _report(name: str, df: pd.DataFrame): n = len(df) pos = int((df["label"] == 1).sum()) diff --git a/charcnn-kit/src/train.py b/charcnn-kit/src/train.py index 709502e..2908f11 100644 --- a/charcnn-kit/src/train.py +++ b/charcnn-kit/src/train.py @@ -1,4 +1,5 @@ # 모델 학습 및 학습 결과와 완료된 모델을 저장함 +# (이슈 #3) HybridCharCNN: CharCNN + URL 기반 tabular feature 융합 import copy import os import torch @@ -14,51 +15,59 @@ MODEL_PATH, VOCAB_PATH, TRAIN_CSV_PATH, - VALID_CSV_PATH + VALID_CSV_PATH, + FEATURE_COLS, ) from dataset import URLDataset, save_vocab -from model import CharCNN +from model import HybridCharCNN +from features import save_norm def train(): os.makedirs("saved", exist_ok=True) - # 1. 학습 데이터셋 생성 - train_dataset = URLDataset(csv_path=TRAIN_CSV_PATH, build_new_vocab=True) - - # 2. train에서 만든 vocab 저장 + # 1. 학습 데이터셋 (vocab + tabular feature 모두 빌드) + train_dataset = URLDataset( + csv_path=TRAIN_CSV_PATH, + build_new_vocab=True, + feature_cols=FEATURE_COLS, + ) vocab = train_dataset.vocab save_vocab(vocab) - # 3. 검증 데이터셋 생성 (train의 vocab 재사용) - valid_dataset = URLDataset(csv_path=VALID_CSV_PATH, vocab=vocab) - - # 4. DataLoader 생성 - train_loader = DataLoader( - train_dataset, - batch_size=BATCH_SIZE, - shuffle=True - ) - valid_loader = DataLoader( - valid_dataset, - batch_size=BATCH_SIZE, - shuffle=False + # 2. tabular feature 정규화 (mean/std 계산은 train 데이터에서만 — 데이터 누수 방지) + # 표준화: (x - mean) / std → 평균 0, 표준편차 1로 맞춤. 스케일이 다른 feature들이 + # 학습을 한쪽으로 끌어당기지 않게 하는 표준 전처리. + mean = train_dataset.features.mean(axis=0) + std = train_dataset.features.std(axis=0) + train_dataset.apply_normalization(mean, std) + save_norm(mean, std) + print(f"[norm] {len(FEATURE_COLS)}개 feature mean/std 저장됨") + + # 3. 검증 데이터셋 (train의 vocab + 동일 mean/std 사용) + valid_dataset = URLDataset( + csv_path=VALID_CSV_PATH, + vocab=vocab, + feature_cols=FEATURE_COLS, ) + valid_dataset.apply_normalization(mean, std) - # 5. 모델 생성 - model = CharCNN(vocab_size=len(vocab)).to(DEVICE) + # 4. DataLoader + train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True) + valid_loader = DataLoader(valid_dataset, batch_size=BATCH_SIZE, shuffle=False) - # 6. 손실 함수 - criterion = nn.BCEWithLogitsLoss() + # 5. 모델 + model = HybridCharCNN( + vocab_size=len(vocab), + num_features=len(FEATURE_COLS), + ).to(DEVICE) - # 7. optimizer - optimizer = torch.optim.Adam( - model.parameters(), - lr=LEARNING_RATE - ) + # 6. 손실 함수 / optimizer + criterion = nn.BCEWithLogitsLoss() + optimizer = torch.optim.Adam(model.parameters(), lr=LEARNING_RATE) - # 8. 학습 루프 (early stopping + best model 추적) + # 7. 학습 루프 (early stopping + best model 추적) best_valid_loss = float("inf") best_state = None best_epoch = 0 @@ -68,41 +77,34 @@ def train(): model.train() total_loss = 0.0 - for x, y in train_loader: + for x, f, y in train_loader: x = x.to(DEVICE) + f = f.to(DEVICE) y = y.to(DEVICE) - # 이전 gradient 초기화 optimizer.zero_grad() - - # 모델 예측 - logits = model(x) - - # loss 계산 + logits = model(x, f) loss = criterion(logits, y) - - # 역전파 loss.backward() - - # 파라미터 업데이트 optimizer.step() total_loss += loss.item() avg_loss = total_loss / len(train_loader) - # 검증 loss 계산 + # 검증 loss model.eval() valid_loss = 0.0 with torch.no_grad(): - for x, y in valid_loader: + for x, f, y in valid_loader: x = x.to(DEVICE) + f = f.to(DEVICE) y = y.to(DEVICE) - logits = model(x) + logits = model(x, f) valid_loss += criterion(logits, y).item() avg_valid_loss = valid_loss / max(len(valid_loader), 1) - # best 모델 갱신 / patience 체크 + # best 갱신 / patience if avg_valid_loss < best_valid_loss: best_valid_loss = avg_valid_loss best_state = copy.deepcopy(model.state_dict()) @@ -126,7 +128,7 @@ def train(): ) break - # 9. best 모델 복원 후 저장 + # 8. best 모델 복원 후 저장 if best_state is None: raise RuntimeError("학습된 모델 state가 없습니다.") model.load_state_dict(best_state)