diff --git a/README.md b/README.md
index 96201f9..c4ae946 100644
--- a/README.md
+++ b/README.md
@@ -1,787 +1,269 @@
-# LinClean-BE-fastapi (URL 보안 엔진)
+# LinClean FastAPI
-**LinClean의 URL 검역 백엔드입니다.** \b` 단어 경계로만 매치한다. `pineapple` 이 `apple`, `naverstore` 가
-`naver`, `googleblog` 이 `google` 로 잡히는 substring false-positive 를 차단하기
-위함이다. 한국어/한자 라벨은 토큰 경계가 모호하고 본문에 분리 없이 박혀 등장하므로
-substring fallback 으로 받는다. 4자 미만 라벨(`kb`, `sk` 등)은 일반 영단어 오탐이
-크므로 로드 단계에서 제외된다.
-
-#### SSRF 방어
-
-`fetch_page()` 는 두 단계로 막는다.
-
-1. **1선 — lexical 검사**: 호스트가 IP 리터럴이고 사설/loopback/link-local/reserved
- 범위면 `error="blocked_host"` 로 즉시 차단. `localhost` / `metadata.google.internal`
- 등 잘 알려진 내부 호스트네임도 거부.
-2. **2선 — DNS 사전 해석**: 공인 도메인처럼 보이는 호스트네임도 `getaddrinfo` 로 미리
- 풀어 모든 응답 IP 를 1선과 동일 룰로 검증. `evil.com → 127.0.0.1` 처럼 lexical 만으로
- 통과하는 케이스를 connect 전에 차단한다.
-
-`/api/v1/content/*` 엔드포인트는 raw URL 을 받기 때문에 `normalize_url()` 로 한 번 더
-통과시켜 `file://` 같은 비허용 스킴을 fetch 로 흘리지 않는다.
-
-**잔여 위험과 운영 요구사항**: getaddrinfo 시점과 실제 connect 시점 사이에 DNS 가
-다른 IP 로 재해석되는 완전한 DNS rebind 는 본 1·2선만으로는 막을 수 없다. 따라서
-**운영 환경 배포 시에는 분석 엔진의 egress 트래픽을 사설 대역으로 향하지 못하게
-하는 방화벽 룰 또는 분석 전용 프록시(서드파티 룰셋 포함) 를 반드시 함께 둔다**.
-운영 체크리스트의 필수 항목이며, 코드 단의 검사는 이를 대체하지 않는다.
-
-#### 메모리 운영 가이드
-
-`fetch_page()` 가 응답 본문을 `CONTENT_FETCH_MAX_BYTES`(기본 2MiB) 로 잘라주지만,
-`extract_features()` 가 이 본문을 BeautifulSoup 트리로 만들면 인메모리 점유는 보통
-**원본 대비 ~10배까지 부푼다** (파서·태그 객체 오버헤드). 단건 비용을 줄이기 위해
-파서는 C 백엔드인 `lxml` 을 쓴다 — 순수 파이썬 `html.parser` 대비 속도·메모리 모두 유리.
-
-다만 BS4 가 모든 노드를 자기 Tag 래퍼로 감싸는 비용은 그대로라, FastAPI 동시성 N 에서
-이 비용이 곱셈으로 폭주할 수 있다. 추출 단계에 글로벌 세마포어를 둬서 피크 메모리를
-**`CONTENT_EXTRACT_CONCURRENCY` × per-page** 로 제한하고, BS4 가 동기 CPU 작업이라
-이벤트 루프를 막던 문제도 `asyncio.to_thread` 오프로드로 같이 해소된다. 기본값 8 이며,
-메모리 여유가 작은 컨테이너에서는 4 이하로 줄여 천장을 낮춘다 — 부하 시 추출이
-직렬화되어 p95 가 늘 수 있으니 배포 환경별 부하 시나리오에 맞춰 튜닝한다.
-
-`CONTENT_FETCH_MAX_BYTES` 는 단건 입력의 상한 — 2MiB 가 일반 페이지에는 넉넉하지만,
-세마포어와 별개로 워커 메모리에 맞춰 보수적으로 잡는다(예: 1MiB).
-
-#### AI 프로바이더 · 모델 교체
-
-`AIProvider` 는 `async def infer(ctx) -> AIInference | None` 한 개짜리 Protocol
-(`app/services/content_analyzer/ai.py`) 로 좁혀져 있습니다. 기본 구현체는
-`OpenAIProvider` 하나이며, 향후 다른 벤더를 붙일 때도 이 인터페이스만 맞추면
-됩니다.
-
-**같은 OpenAI 안에서 모델 교체**는 환경변수 한 줄로 끝납니다 — API 형상은
-동일하고 Structured Outputs(`response_format=json_schema`, `strict=true`) 도 모든
-gpt-4/gpt-4o/gpt-4.1 계열이 공유합니다.
-
-```bash
-# .env
-OPENAI_MODEL=gpt-4o-mini # 기본 — 저비용·저지연
-# OPENAI_MODEL=gpt-4o # 판정 품질 우선
-# OPENAI_MODEL=gpt-4.1-mini
+DB 의존 파이프라인은 GSB와 URLhaus를 포함합니다. DB 비의존 파이프라인은 외부 위협 DB 없이 정규화, 언체이닝, 도메인 휴리스틱, 콘텐츠/AI 분석만으로 판정합니다.
+
+## 각 단계별 상세
+
+1. URL 정규화
+ 입력 URL의 스킴, 호스트, 기본 포트, 경로, 인코딩, fragment를 정리합니다. 스킴이 없는 URL은 HTTPS 우선으로 분석 가능한 형태를 만듭니다.
+
+2. 리다이렉트 언체이닝
+ 3xx Location 체인을 따라 최종 URL을 찾습니다. 같은 등록 도메인 안의 canonical redirect는 cross-origin으로 보지 않고, 등록 도메인이 달라질 때만 `REDIRECT_CROSS_ORIGIN` 신호를 남깁니다.
+
+3. 외부 위협 DB 조회
+ Google Safe Browsing과 URLhaus 로컬 캐시를 조회합니다. 알려진 악성 URL이면 점수와 관계없이 `danger`로 판정할 수 있습니다.
+
+4. 도메인/URL 휴리스틱
+ IP 직접 접근, userinfo 포함 URL, 오타 도메인, DGA 유사 도메인, suspicious TLD, open redirect 파라미터, 민감 경로, 무료 호스팅 유도 등을 점수화합니다. 신뢰 도메인은 DGA/typo 오탐을 줄이도록 보정합니다.
+
+5. 콘텐츠 분석
+ 최종 URL의 HTML을 가져와 title, form, password field, 외부 form action, 민감정보 필드, 기관/브랜드 사칭 문구, 위험 다운로드, meta refresh, 외부 링크 비율을 분석합니다. 404 등 찾을 수 없는 페이지는 `PAGE_UNAVAILABLE` 실패로 조기 종료합니다.
+
+6. AI 보조 판정
+ 구조화된 페이지 피처와 선행 단계 신호를 OpenAI에 전달합니다. AI 응답은 `phishing`, `suspicious`, `benign` 중 하나이며, `ai_reason`에는 분석 근거와 사용자 행동 가이드를 100자 이내 한 문장으로 담습니다. 예: `브랜드 사칭 정황이 있어 비밀번호나 결제 정보를 입력하지 마세요.`
+
+## 점수 산정 기준
+
+모든 점수는 합산 후 0~100으로 제한합니다. 외부 위협 DB 매치는 고신뢰 신호로 보며, 휴리스틱과 콘텐츠 분석은 보조 신호로 누적합니다.
+
+| 영역 | 신호 | 점수 |
+|---|---:|---:|
+| 외부 DB | GSB match | 50 |
+| 외부 DB | URLhaus match | 50 |
+| 도메인 | IP 직접 접근 | 40 |
+| 도메인 | URL userinfo | 45 |
+| 도메인 | typo domain | 30 |
+| 도메인 | punycode IDN | 35 |
+| 도메인 | open redirect param | 31 |
+| 도메인 | DGA-like | 31 |
+| 도메인 | suspicious TLD | 25 |
+| 도메인 | new domain | 25 |
+| 도메인 | no HTTPS, subdomain/hyphen overuse, hosting platform, sensitive path | 각 20 |
+| 도메인 | brand in URL | 30 |
+| 도메인 | free hosting lure, URL shortener | 각 25 |
+| 콘텐츠 | brand impersonation form | 50 |
+| 콘텐츠 | external credential form | 45 |
+| 콘텐츠 | sensitive ID field | 30 |
+| 콘텐츠 | financial field, external meta refresh | 각 25 |
+| 콘텐츠 | PII form, risky download, public agency lure, meta refresh | 각 20 |
+| 콘텐츠 | Korean lure text | 15 |
+| 콘텐츠 | logo alt impersonation | 10 |
+| 콘텐츠 | external link overuse | 5 |
+| 콘텐츠 | fetch failed | 15 |
+| AI | phishing | 45 |
+| AI | suspicious | 31 |
+
+보정 기준:
+
+- 도메인 휴리스틱 점수는 최대 80점입니다.
+- 콘텐츠 분석 점수는 최대 100점입니다.
+- 종합 점수는 최대 100점입니다.
+- 선행 단계 점수가 61점 이상이면 콘텐츠 분석을 건너뛰고 `danger`로 확정할 수 있습니다.
+- `not_html`, `too_large`, `unexpected_redirect`, `blocked_host`는 악성 근거가 아니라 분석 불가 성격이 강해 fetch failed 신호만 남기고 점수는 올리지 않습니다.
+- 신뢰 도메인에서 AI `suspicious` 단독 판정은 점수를 올리지 않습니다. 단, 외부 form action, 민감정보 필드, 브랜드 사칭 등 강한 신호가 있으면 반영합니다.
+
+## verdict 종류 및 구간별 점수
+
+| Verdict | 점수 구간 | 의미 |
+|---|---:|---|
+| `safe` | 0~30 | 현재 기준에서 뚜렷한 위험 신호가 낮음 |
+| `caution` | 31~60 | 사용자가 링크, 입력 정보, 결제 정보를 한 번 더 확인해야 함 |
+| `danger` | 61~100 | 접속, 로그인, 결제, 다운로드를 피해야 하는 위험 상태 |
+
+GSB 또는 URLhaus에 악성으로 등록된 경우에는 점수 구간과 별개로 `danger` verdict가 우선됩니다.
+
+## API 종류및 상세
+
+공통 헤더:
+
+```http
+X-Internal-Api-Key:
+Content-Type: application/json
```
-모델별 품질을 잠깐 비교하려면 `OpenAIProvider(model="gpt-4o")` 처럼 생성자
-인자로 덮어쓰고 `analyze_content(url, provider=...)` 로 한 번 꽂아 쓰면 됩니다.
-운영 라우트는 전역 프로바이더를 쓰며, 디버그 엔드포인트도 동일하게 전역
-프로바이더만 사용합니다.
-
-환경변수 요약:
-
-| 이름 | 기본값 | 설명 |
-|-----------------------------|-----------------|----------------------------------------------------------------------|
-| `AI_PROVIDER` | `auto` | `auto` (키 있으면 openai) / `openai` / `null` (비활성) |
-| `OPENAI_API_KEY` | *(없음)* | 비워두면 `NullAIProvider` — 규칙 점수만 사용 |
-| `OPENAI_MODEL` | `gpt-4o-mini` | OpenAI 채팅 모델 id |
-| `OPENAI_TIMEOUT_SECONDS` | `5.0` | 단일 호출 타임아웃 |
-| `OPENAI_MAX_OUTPUT_TOKENS` | `120` | verdict + 100자 이내 reason 용 출력 상한 |
-
-#### 응답에 실리는 AI 메타데이터
+| Method | Path | 설명 |
+|---|---|---|
+| GET | `/api/v1/health` | 서버 상태 확인 |
+| GET | `/api/v1/health/ready` | DB 등 readiness 확인 |
+| POST | `/api/v1/analyze` | 비동기 분석 접수. 완료 후 Spring 콜백 전송 |
+| POST | `/api/v1/analyze/sync` | GSB, URLhaus 포함 전체 파이프라인 동기 실행 |
+| POST | `/api/v1/analyze/db-independent/sync` | 외부 DB 없이 동기 실행 |
+| POST | `/api/v1/normalize` | URL 정규화와 리다이렉트 언체이닝 결과 확인 |
+| POST | `/api/v1/threat-db` | GSB, URLhaus 조회만 실행 |
+| POST | `/api/v1/domain-heuristic` | 도메인/URL 휴리스틱만 실행 |
+| POST | `/api/v1/content-analysis` | 콘텐츠 분석과 AI 보조 판정만 실행 |
+| POST | `/api/v1/content/fetch-extract` | HTML fetch와 feature 추출만 실행 |
+
+요청 예시:
-`ai_reason` 은 보안 전문가의 근거 중심 문장으로 요청하되, 비전문가도 이해할 수
-있도록 쉬운 한국어 100자 이내로 제한합니다. 모델이 더 길게 응답해도 클라이언트에서
-100자로 잘라 응답합니다.
-
-`ContentAnalysisResult` 에는 verdict/reason 뿐 아니라 **실제로 응답한 모델 id 와
-토큰 사용량**이 함께 실립니다. 비용 관측, 모델 비교, 프롬프트 튜닝에 그대로
-쓸 수 있게 하기 위함입니다.
-
-```jsonc
+```json
{
- "ai_verdict": "phishing",
- "ai_reason": "URL 이 네이버 공식 도메인이 아니며 ...",
- "ai_error": null,
- "ai_model": "gpt-4o-mini",
- "ai_token_usage": {
- "prompt_tokens": 412,
- "completion_tokens": 47,
- "total_tokens": 459
- }
+ "url": "https://example.com"
}
```
-`ai_token_usage` 는 OpenAI 응답 `usage` 블록이 비어있을 때만 `null` 로 떨어집니다.
-API 호출 자체가 실패하면 `ai_error="ai_unavailable"` 로 기록되고 규칙 점수만으로
-결과가 반환됩니다. `AI_PROVIDER=openai` 로 강제했는데 `OPENAI_API_KEY` 가 비어
-NullProvider 로 폴백된 경우에는 `ai_error="provider_misconfigured"` 로 응답에
-표시되어, 정상 NullProvider 동작과 misconfiguration 상태를 운영자가 응답만으로 구분할
-수 있습니다.
-
-### 최종 점수 산출
-
-각 단계의 점수를 합산(상한 100)해 다음 구간으로 매핑합니다.
-
-| 합산 점수 | verdict | UI |
-|-----------|---------|----|
-| 0 ~ 30 | **safe** (안전) | 초록 — 정상적으로 열기 가능 |
-| 31 ~ 60 | **caution** (주의) | 노랑 — 이유 표시 후 사용자 판단 |
-| 61 이상 | **danger** (위험) | 빨강 — "피싱 의심, 열지 마세요" |
-
-**예외 — blacklist 매치는 score 100 / danger**: `threat_db.is_malicious=True`
-면 GSB / URLhaus 매치 = 알려진 악성 URL 로 보고 score 를 100으로 고정하며,
-verdict 는 `danger` 로 강제됩니다. 이 경우 4단계 fetch/AI 는 수행하지 않습니다.
+비동기 분석 접수는 `analysisId`를 함께 보냅니다.
-판정 근거는 `stages` 내부의 각 단계 원시 결과(`threat_db`, `domain_heuristic`,
-`content_analysis`)에 남습니다. 별도의 `reasons` 배열/자연어 `summary` 필드는
-현재 코드에서는 생성하지 않습니다.
-
-#### 응답 스키마 — verdict / score 가 상단
-
-`PipelineSuccess` 응답은 `verdict` 와 `score` 를 단계별 원시 결과(`stages`) 보다
-**위에** 직렬화합니다. 클라이언트는 stages 트리를 파싱하지 않고도 첫 몇 줄에서
-판정을 즉시 읽을 수 있습니다. 동기 확인은 `/api/v1/analyze/sync` 로 수행하고,
-비동기 `/api/v1/analyze` 는 분석 완료 후 Spring 콜백으로 동일 결론을 전달합니다.
-
-```jsonc
+```json
{
- "status": "success",
- "analysisId": "...",
- "originalUrl": "https://bit.ly/abc123",
- "finalUrl": "https://login-secure-naver-auth.com/signin",
- "verdict": "danger", // ← 상단
- "score": 75, // ← 상단
- "stages": { ... }
+ "analysisId": "analysis-1",
+ "url": "https://example.com"
}
```
----
-
-## 단계별 운영/QA 엔드포인트 (`/api/v1/*`)
-
-현재 코드는 단계별 단독 호출 라우터를 운영 라우터에 항상 마운트합니다. 모든
-엔드포인트는 `X-Internal-Api-Key` 인증을 요구하며, raw URL 을 받아
-`normalize_url()` 로 1차 검증한 뒤 해당 단계만 실행합니다. 전체 파이프라인을
-동기로 확인하려면 `/api/v1/analyze/sync` 를 사용합니다. 외부 위협 DB(GSB/URLhaus)
-없이 URL·리다이렉트·RDAP·콘텐츠/AI만 확인하려면
-`/api/v1/analyze/db-independent/sync` 를 사용합니다.
-
-| Method | Path | 단계 | 호출 함수 |
-|--------|---------------------------------|---------------------|---------------------------------|
-| POST | `/normalize` | Stage 1 | `normalize_url` + `unchain_url` |
-| POST | `/threat-db` | Stage 2 | `check_threat_db` |
-| POST | `/domain-heuristic` | Stage 3 | `check_domain_heuristic` |
-| POST | `/content-analysis` | Stage 4 | `analyze_content` |
-| POST | `/analyze/sync` | 전체 (1~4 + verdict)| `run_pipeline` |
-| POST | `/analyze/db-independent/sync` | DB 비의존 전체 | `run_db_independent_pipeline` |
-| POST | `/content/fetch-extract` | 4단계 보조 확인 | `fetch_page` + `extract_features` |
-
-요청 바디는 모두 `{ "url": "" }` 형태이며, `/analyze/sync` 는
-`analysisId` 를 내부에서 생성한다. `/threat-db` 는 DB 세션이 필요해
-요청 처리 시 `get_db` dependency 가 일반 라우트와 동일하게 주입된다.
-
-**단계별 API 사용 시 주의 — unchainer 는 1단계에서만 동작**: `/threat-db`,
-`/domain-heuristic`, `/content-analysis` 는 `normalize_url` 만 거치고
-unchain 은 거치지 않는다. 단축 URL(`bit.ly/...`) 을 그대로 넣으면 4단계는
-`unexpected_redirect` 로 떨어지고, 2·3단계는 단축 호스트 자체로 매치/조회가
-나간다. 단축 URL 을 풀어 분석하려면 먼저 `/normalize` 로 `unchain.final_url` 을
-확인한 뒤 그 값을 단계별 API 에 다시 넣는다.
-
-**예시 — Stage 1 (정규화 + 언체이닝)**
-
-```bash
-curl -sX POST http://localhost:8000/api/v1/normalize \
- -H 'Content-Type: application/json' \
- -H 'X-Internal-Api-Key: ' \
- -d '{"url":"https://bit.ly/abc123"}'
-```
+## JSON 응답 예시
-응답 (`DevNormalizeResponse`):
+동기 분석 성공:
-```jsonc
+```json
{
- "normalize": {
- "original_url": "https://bit.ly/abc123",
- "normalized_url": "https://bit.ly/abc123"
+ "status": "success",
+ "analysis_id": "analysis-1",
+ "original_url": "https://example.com",
+ "final_url": "https://example.com/",
+ "verdict": "caution",
+ "score": 31,
+ "timings": {
+ "total_seconds": 1.23,
+ "stages": {
+ "normalize": 0.001,
+ "unchain": 0.12,
+ "threat_db": 0.03,
+ "domain_heuristic": 0.2,
+ "content_analysis": 0.87
+ }
},
- "unchain": {
- "input_url": "https://bit.ly/abc123",
- "final_url": "https://login-secure-naver-auth.com/signin",
- "hops": [ /* HopRecord[] */ ],
- "hop_count": 2,
- "timed_out": false,
- "error": null,
- "signals": []
+ "stages": {
+ "content_analysis": {
+ "final_url": "https://example.com/",
+ "fetched": true,
+ "status_code": 200,
+ "score": 31,
+ "signals": [],
+ "ai_verdict": "suspicious",
+ "ai_reason": "로그인 유도 정황이 있어 비밀번호나 결제 정보를 입력하지 마세요.",
+ "ai_error": null,
+ "ai_model": "gpt-4o-mini"
+ }
}
}
```
-**예시 — Stage 4 (콘텐츠 정적 분석)**
-
-```bash
-curl -sX POST http://localhost:8000/api/v1/content-analysis \
- -H 'Content-Type: application/json' \
- -H 'X-Internal-Api-Key: ' \
- -d '{"url":"https://login-secure-naver-auth.com/signin"}'
-```
-
-응답은 `ContentAnalysisResult` 그대로이며, 최종 `verdict` 는 포함하지 않습니다.
-`ai_error` 필드는 다음 의미를
-갖는다:
-
-- `null` — 정상 호출 또는 NullProvider 정상 비활성
-- `ai_unavailable` — OpenAI 호출 실패 (timeout / 네트워크 / 5xx 등)
-- `provider_misconfigured` — `AI_PROVIDER=openai` 강제인데 `OPENAI_API_KEY` 가 비어
- 부팅 시 NullProvider 로 폴백된 상태. 응답으로 misconfiguration 을 식별할 수 있게 노출.
+찾을 수 없는 페이지:
----
-
-
-## 시스템 한계와 보완 방안
-
-어떤 정적 분석 시스템도 100% 탐지는 불가능합니다. 본 엔진이 인지하고 있는
-주요 한계와 대응 방안은 다음과 같습니다.
-
-| 공격 방식 | 한계 | 대응 방안 |
-|----------|------|----------|
-| 지연 활성화 (TOCTOU) | 분석 후 피싱 페이지로 교체 | 클릭 직전 재검사 + 리마인드 시 재검사 |
-| 봇 탐지 | 크롤러엔 정상 페이지 반환 | User-Agent 주기적 교체 |
-| 합법 도메인 악용 (`github.io` 등) | 신뢰 도메인 위에 피싱 호스팅 | 호스팅 플랫폼 가산 점수 |
-| 리버스 프록시 | 진짜 사이트를 실시간 프록시 | 도메인 불일치 경고 |
-| 오픈 리다이렉트 | 합법 도메인 경유 후 피싱 | 리다이렉트 파라미터 감지 |
-
----
-
-## 두 백엔드 간 통신 규약
-
-분석 엔진은 **요청-응답 동기 호출이 아니라 콜백** 방식으로 Spring 과 통신합니다.
-Spring 이 분석 위임 요청을 보내면, 본 엔진은 즉시 `analysisId` 를 반환하고
-실제 결과는 분석이 끝난 뒤 Spring 의 내부 콜백 엔드포인트로 POST 합니다.
-
-```
-Spring FastAPI (본 엔진)
- │ │
- │ POST /api/v1/analyze │
- │ { analysisId, url } │
- ├──────────────────────────────────────►│
- │ │ (4단계 파이프라인 비동기 실행)
- │◄──── 202 Accepted │
- │ { analysisId, status: "queued" } │
- │ │
- │ │ (분석 완료)
- │ POST /internal/analysis-result │
- │ X-Internal-Api-Key: │
- │ { ... AnalysisResultCallback ... } │
- │◄──────────────────────────────────────┤
- │ │
- │ 200 OK { received: true } │
- ├──────────────────────────────────────►│
+```json
+{
+ "status": "failed",
+ "analysis_id": "analysis-1",
+ "original_url": "https://missing.example",
+ "final_url": "https://missing.example",
+ "failed_at_stage": "content_analysis",
+ "error": "페이지를 찾을 수 없습니다.",
+ "error_code": "PAGE_UNAVAILABLE",
+ "status_code": 404
+}
```
-모든 호출에는 `X-Internal-Api-Key` 헤더가 포함되어야 하며, Spring 은 이 헤더가
-없거나 값이 일치하지 않는 요청을 거부합니다. 외부에서 `/internal/*` 경로를
-직접 호출할 수 없게 하기 위함입니다. 이 값은 동적으로 발급·만료되는 토큰이
-아니라, 두 백엔드가 환경변수로 공유하는 **정적 사전 공유 키(pre-shared key)**
-입니다.
-
-### Spring 콜백 메시지 정의
-
-**Endpoint:** `POST {SPRING_INTERNAL_URL}/internal/analysis-result`
-
-**Headers:**
-
-| 헤더 | 필수 | 설명 |
-|------|------|------|
-| `Content-Type` | ✓ | `application/json` |
-| `X-Internal-Api-Key` | ✓ | 두 백엔드가 환경변수로 공유하는 사전 공유 키 |
-| `X-Request-ID` | ✓ | 원 요청의 request id 를 그대로 echo (분산 추적용) |
-
-**Body — 성공 (`AnalysisResultCallback`):**
+비동기 접수 응답:
```json
{
- "analysisId": "9f0e0e3a-2c1f-4b76-9d3e-0e0a5a1cf2a1",
- "requestId": "b4c3a9e7-7c2a-4a1d-9f0b-1c2d3e4f5a6b",
- "status": "succeeded",
- "originalUrl": "https://bit.ly/abc123",
- "finalUrl": "https://login-secure-naver-auth.com/signin",
- "verdict": "danger",
- "score": 82,
- "stages": {
- "normalize": {
- "original_url": "https://bit.ly/abc123",
- "normalized_url": "https://bit.ly/abc123"
- },
- "unchain": {
- "input_url": "https://bit.ly/abc123",
- "final_url": "https://login-secure-naver-auth.com/signin",
- "hops": [],
- "hop_count": 2,
- "timed_out": false,
- "error": null,
- "signals": []
- },
- "threat_db": {
- "final_url": "https://login-secure-naver-auth.com/signin",
- "is_malicious": true,
- "sources_checked": 2,
- "gsb": { "checked": true, "is_threat": true, "matches": [] },
- "urlhaus": { "checked": true, "is_threat": false },
- "threat_types": ["SOCIAL_ENGINEERING"]
- },
- "domain_heuristic": {
- "domain": "login-secure-naver-auth.com",
- "score": 30,
- "signals": ["NEW_DOMAIN", "HYPHEN_OVERUSE"],
- "rdap": {
- "domain": "login-secure-naver-auth.com",
- "registrar": "NameCheap, Inc.",
- "created_date": "2026-03-22T00:00:00Z",
- "expiry_date": null,
- "domain_age_days": 16,
- "is_new_domain": true
- },
- "rdap_error": null
- },
- "content_analysis": {
- "final_url": "https://login-secure-naver-auth.com/signin",
- "fetched": false,
- "status_code": null,
- "score": 0,
- "signals": ["SKIPPED_ALREADY_DANGER"],
- "title": null,
- "has_password_field": false,
- "has_meta_refresh": false,
- "external_link_ratio": null,
- "brand_impersonation": false,
- "logo_alt_impersonation": false,
- "is_spa_shell": false,
- "ai_verdict": null,
- "ai_reason": null,
- "reason": "위험성이 확인된 URL입니다. 페이지를 열지 않는 것이 좋습니다.",
- "ai_error": null,
- "ai_model": null,
- "ai_token_usage": null,
- "error": "skipped_already_danger"
- }
- },
- "engineVersion": "0.1.0",
- "analyzedAt": "2026-04-07T05:42:11Z",
- "elapsedMs": 1843
+ "analysisId": "analysis-1",
+ "status": "queued"
}
```
-**Body — 실패:**
+Spring 성공 콜백은 camelCase로 전송합니다.
```json
{
- "analysisId": "9f0e0e3a-2c1f-4b76-9d3e-0e0a5a1cf2a1",
- "requestId": "b4c3a9e7-7c2a-4a1d-9f0b-1c2d3e4f5a6b",
- "status": "failed",
- "originalUrl": "https://bit.ly/abc123",
- "error": {
- "code": "NORMALIZE_FAILED",
- "stage": "normalize",
- "message": "invalid url"
+ "analysisId": "analysis-1",
+ "requestId": "request-1",
+ "status": "succeeded",
+ "originalUrl": "https://example.com",
+ "finalUrl": "https://example.com/",
+ "verdict": "caution",
+ "score": 31,
+ "summary": "로그인 유도 정황이 있어 비밀번호나 결제 정보를 입력하지 마세요.",
+ "stages": {
+ "contentAnalysis": {
+ "fetched": true,
+ "hasPasswordField": true,
+ "aiVerdict": "suspicious",
+ "aiReason": "로그인 유도 정황이 있어 비밀번호나 결제 정보를 입력하지 마세요."
+ }
},
"engineVersion": "0.1.0",
- "analyzedAt": "2026-04-07T05:42:41Z",
- "elapsedMs": 7
+ "analyzedAt": "2026-05-29T00:00:00Z",
+ "elapsedMs": 1234
}
```
-
-**필드 정의:**
-
-| 필드 | 타입 | 필수 | 설명 |
-|------|------|------|------|
-| `analysisId` | string (uuid) | ✓ | Spring 이 분석 위임 시 발급한 식별자. 콜백에서 그대로 echo |
-| `requestId` | string (uuid) | ✓ | 분산 추적용 request id (`X-Request-ID` 헤더와 동일) |
-| `status` | enum | ✓ | `succeeded` / `failed` |
-| `originalUrl` | string | ✓ | Spring 에게서 받은 원본 URL |
-| `finalUrl` | string | succeeded | 1·2단계를 거친 뒤 확정된 최종 분석 대상 URL |
-| `verdict` | enum | succeeded | `safe` / `caution` / `danger` |
-| `score` | int (0~100) | succeeded | 4단계 합산 점수 (100 cap) |
-| `stages` | object | succeeded | 현재 `PipelineStages` 모델을 JSON 직렬화한 단계별 원시 결과. 내부 키는 코드 모델과 동일한 snake_case |
-| `engineVersion` | string | ✓ | 본 엔진 버전. Spring 이 결과의 호환성을 판단할 때 사용 |
-| `analyzedAt` | string (ISO8601 UTC) | ✓ | 분석 종료 시각 |
-| `elapsedMs` | int | ✓ | 분석에 소요된 wall-clock 시간 (밀리초) |
-| `error` | object | failed | 실패 원인 |
-
-**Spring 측 응답:**
-
-```json
-HTTP/1.1 200 OK
-{ "received": true }
-```
-
-- `200 OK` 외 응답 또는 네트워크 오류 시 본 엔진은 지수 백오프로 최대 3회
- 재시도합니다. 그 후에도 실패하면 dead-letter 로그에 적재하고 운영자가
- 수동 재처리할 수 있게 합니다.
-- 동일 `analysisId` 의 콜백이 중복 도착할 수 있으므로, **Spring 측 처리는
- 멱등(idempotent) 해야 합니다.**
diff --git a/app/core/config.py b/app/core/config.py
index 028f021..878eca9 100644
--- a/app/core/config.py
+++ b/app/core/config.py
@@ -74,6 +74,11 @@ def alembic_database_url(self) -> str:
"gitlab.com": 2,
"bitbucket.org": 2,
"sites.google.com": 2,
+ "dropbox.com": 2,
+ "www.dropbox.com": 2,
+ "dropboxusercontent.com": 2,
+ "dl.dropboxusercontent.com": 2,
+ "www.dropboxusercontent.com": 2,
}
)
@@ -124,10 +129,11 @@ def alembic_database_url(self) -> str:
score_weight_no_https: int = 20
score_weight_new_domain: int = 25
score_weight_subdomain_overuse: int = 20
- score_weight_open_redirect_param: int = 30
+ score_weight_open_redirect_param: int = 31
score_weight_hyphen_overuse: int = 20
score_weight_suspicious_tld: int = 25
- score_weight_dga_like: int = 10
+ score_weight_dga_like: int = 31
+ score_weight_redirect_cross_origin: int = 15
score_weight_hosting_platform: int = 20
score_weight_url_userinfo: int = 45
score_weight_brand_in_url: int = 30
@@ -206,7 +212,7 @@ def alembic_database_url(self) -> str:
# 정상 컨텐츠 또는 파이프라인 정합성 문제로 보고 점수 가산 없이 시그널만 남긴다.
score_weight_content_fetch_failed: int = 15
score_weight_ai_phishing: int = 45
- score_weight_ai_suspicious: int = 20
+ score_weight_ai_suspicious: int = 31
# 4단계 단독 캡 — 컨텐츠 분석 단계 안에서만 적용된다. 전 단계 합산은 별도로 score_total_cap 에서
# 다시 100 으로 클램프되므로, 여기를 낮춰도 합산 상한이 자동으로 같이 낮아지는 게 아니다.
content_analysis_score_cap: int = 100
diff --git a/app/schemas/db_independent_pipeline.py b/app/schemas/db_independent_pipeline.py
index 5e14792..62da8f0 100644
--- a/app/schemas/db_independent_pipeline.py
+++ b/app/schemas/db_independent_pipeline.py
@@ -35,8 +35,11 @@ class DbIndependentPipelineFailure(BaseModel):
status: Literal["failed"] = "failed"
analysis_id: str
original_url: str
+ final_url: str | None = None
failed_at_stage: PipelineStage
error: str
+ error_code: str | None = None
+ status_code: int | None = None
timings: PipelineTimings | None = None
diff --git a/app/schemas/domain_heuristic.py b/app/schemas/domain_heuristic.py
index f3a25da..54b96c8 100644
--- a/app/schemas/domain_heuristic.py
+++ b/app/schemas/domain_heuristic.py
@@ -32,6 +32,7 @@ class DomainHeuristicSignal(StrEnum):
FREE_HOSTING_LURE = "FREE_HOSTING_LURE"
SENSITIVE_PATH = "SENSITIVE_PATH"
URL_SHORTENER = "URL_SHORTENER"
+ REDIRECT_CROSS_ORIGIN = "REDIRECT_CROSS_ORIGIN"
class DomainHeuristicSkippedReason(StrEnum):
diff --git a/app/schemas/pipeline.py b/app/schemas/pipeline.py
index f1b8c6d..0d2ff7b 100644
--- a/app/schemas/pipeline.py
+++ b/app/schemas/pipeline.py
@@ -76,8 +76,11 @@ class PipelineFailure(BaseModel):
status: Literal["failed"] = "failed"
analysis_id: str
original_url: str
+ final_url: str | None = None
failed_at_stage: PipelineStage
error: str
+ error_code: str | None = None
+ status_code: int | None = None
timings: PipelineTimings | None = None
diff --git a/app/services/analysis_callback.py b/app/services/analysis_callback.py
index 66148b5..41e7956 100644
--- a/app/services/analysis_callback.py
+++ b/app/services/analysis_callback.py
@@ -42,6 +42,7 @@
"FREE_HOSTING_LURE": "무료 호스팅 주소에서 신뢰를 유도하는 문구를 사용합니다.",
"SENSITIVE_PATH": "로그인 또는 인증 관련 경로를 사용합니다.",
"URL_SHORTENER": "단축 URL 서비스를 사용합니다.",
+ "REDIRECT_CROSS_ORIGIN": "입력 URL이 다른 사이트로 이동합니다.",
}
_CONTENT_REASON_MESSAGES: dict[str, str] = {
@@ -99,6 +100,7 @@ def _signal_weight(code: str) -> int:
"FREE_HOSTING_LURE": settings.score_weight_free_hosting_lure,
"SENSITIVE_PATH": settings.score_weight_sensitive_path,
"URL_SHORTENER": settings.score_weight_url_shortener,
+ "REDIRECT_CROSS_ORIGIN": settings.score_weight_redirect_cross_origin,
}
content_weights = {
"BRAND_IMPERSONATION_FORM": settings.score_weight_brand_impersonation,
@@ -313,20 +315,26 @@ def _failure_payload(
elapsed_ms: int,
analyzed_at: datetime,
) -> dict[str, Any]:
+ error: dict[str, Any] = {
+ "code": result.error_code or f"{result.failed_at_stage.value.upper()}_FAILED",
+ "stage": _error_stage(result.failed_at_stage),
+ "message": result.error,
+ }
+ if result.status_code is not None:
+ error["statusCode"] = result.status_code
+
payload: dict[str, Any] = {
"analysisId": result.analysis_id,
"requestId": request_id,
"status": "failed",
"originalUrl": result.original_url,
- "error": {
- "code": f"{result.failed_at_stage.value.upper()}_FAILED",
- "stage": _error_stage(result.failed_at_stage),
- "message": result.error,
- },
+ "error": error,
"engineVersion": settings.app_version,
"analyzedAt": _iso_z(analyzed_at),
"elapsedMs": elapsed_ms,
}
+ if result.final_url is not None:
+ payload["finalUrl"] = result.final_url
return payload
diff --git a/app/services/content_analyzer/ai_openai.py b/app/services/content_analyzer/ai_openai.py
index e92fb49..4182d1a 100644
--- a/app/services/content_analyzer/ai_openai.py
+++ b/app/services/content_analyzer/ai_openai.py
@@ -34,7 +34,10 @@
"type": "string",
"enum": [v.value for v in AIVerdict],
},
- "reason": {"type": "string"},
+ "reason": {
+ "type": "string",
+ "description": "Korean analysis reason plus user action guidance, within 100 chars.",
+ },
},
"required": ["verdict", "reason"],
"additionalProperties": False,
@@ -67,7 +70,9 @@
"보수적으로 선택한다. "
"verdict 는 phishing / suspicious / benign 중 하나. reason 은 보안 전문가처럼 "
"근거 중심으로 쓰되, IT와 보안을 모르는 사람도 이해할 수 있는 쉬운 한국어 100자 이내 "
- "1문장으로 작성한다. "
+ "1문장으로 작성한다. reason 에는 별도 필드 없이 분석 근거와 사용자 행동 가이드를 함께 "
+ "담아라. 예: 비밀번호 입력에 주의하세요, 결제 수단을 등록하지 마세요, 첨부 파일을 "
+ "내려받지 마세요. "
"확증이 없으면 benign 또는 suspicious 를 쓰고 phishing 은 보수적으로만 사용한다."
)
diff --git a/app/services/content_analyzer/analyze.py b/app/services/content_analyzer/analyze.py
index f0a9478..986398a 100644
--- a/app/services/content_analyzer/analyze.py
+++ b/app/services/content_analyzer/analyze.py
@@ -24,6 +24,7 @@
from app.services.content_analyzer.fetch import fetch_page
from app.services.content_analyzer.render import render_page
from app.services.content_analyzer.signals import ContentScoring, score_content
+from app.services.domain_heuristic.patterns import is_trusted_registered_domain
logger = get_logger(__name__)
@@ -63,6 +64,45 @@ def _ai_score_weight(verdict: AIVerdict) -> int:
return 0
+_HIGH_RISK_CONTENT_SIGNALS: frozenset[str] = frozenset(
+ {
+ ContentSignal.BRAND_IMPERSONATION_FORM.value,
+ ContentSignal.CREDENTIAL_FORM_EXTERNAL.value,
+ ContentSignal.SENSITIVE_ID_FIELD.value,
+ ContentSignal.FINANCIAL_FIELD.value,
+ ContentSignal.RISKY_DOWNLOAD_LINK.value,
+ ContentSignal.EXTERNAL_META_REFRESH.value,
+ }
+)
+_HIGH_RISK_UPSTREAM_SIGNALS: frozenset[str] = frozenset(
+ {
+ "URL_USERINFO",
+ "OPEN_REDIRECT_PARAM",
+ "BRAND_IN_URL",
+ "FREE_HOSTING_LURE",
+ "SENSITIVE_PATH",
+ "SUSPICIOUS_TLD",
+ "PUNYCODE_IDN",
+ "TYPO_DOMAIN",
+ }
+)
+
+
+def _should_apply_ai_suspicious_score(
+ final_url: str,
+ scoring: ContentScoring,
+ upstream_signals: tuple[str, ...],
+) -> bool:
+ if not is_trusted_registered_domain(final_url):
+ return True
+ content_signal_values = {signal.value for signal in scoring.signals}
+ if content_signal_values & _HIGH_RISK_CONTENT_SIGNALS:
+ return True
+ if set(upstream_signals) & _HIGH_RISK_UPSTREAM_SIGNALS:
+ return True
+ return False
+
+
def _unique_merge(left: list[str], right: list[str]) -> list[str]:
merged = list(left)
for item in right:
@@ -279,7 +319,12 @@ async def analyze_content(
ai_reason = inference.reason
ai_model = inference.model
ai_token_usage = inference.token_usage
- score += _ai_score_weight(inference.verdict)
+ if inference.verdict != AIVerdict.SUSPICIOUS or _should_apply_ai_suspicious_score(
+ final_url,
+ scoring,
+ upstream_tuple,
+ ):
+ score += _ai_score_weight(inference.verdict)
elif ai_error is None:
# 추론이 None 인데 호출 단계 예외도 없었다면 NullAIProvider 동작.
# 부팅 시 misconfiguration 으로 폴백된 NullProvider 면 fallback_reason 을 응답에 노출.
diff --git a/app/services/content_analyzer/extract.py b/app/services/content_analyzer/extract.py
index 0dab603..f02694b 100644
--- a/app/services/content_analyzer/extract.py
+++ b/app/services/content_analyzer/extract.py
@@ -53,8 +53,9 @@
_MAX_CTA_TEXTS = 40
_MAX_DOWNLOAD_LINKS = 40
_RISKY_DOWNLOAD_EXTENSIONS: frozenset[str] = frozenset(
- {".apk", ".ipa", ".exe", ".msi", ".dmg", ".scr", ".bat", ".cmd", ".js", ".vbs"}
+ {".apk", ".ipa", ".exe", ".msi", ".dmg", ".scr", ".bat", ".cmd", ".vbs"}
)
+_RISKY_DOWNLOAD_ATTR_EXTENSIONS: frozenset[str] = frozenset({".js"})
_KOREAN_LURE_KEYWORDS: tuple[str, ...] = (
"지원금",
@@ -341,7 +342,11 @@ def _collect_cta_texts(soup: BeautifulSoup) -> list[str]:
return texts
-def _is_risky_download_url(raw_url: str, base_url: str) -> str | None:
+def _anchor_has_download_attr(anchor: Tag) -> bool:
+ return anchor.has_attr("download")
+
+
+def _is_risky_download_url(raw_url: str, base_url: str, *, has_download_attr: bool) -> str | None:
joined = urljoin(base_url, raw_url.strip())
parsed = urlparse(joined)
if parsed.scheme not in _NAV_SCHEMES:
@@ -349,6 +354,10 @@ def _is_risky_download_url(raw_url: str, base_url: str) -> str | None:
path = parsed.path.lower()
if any(path.endswith(ext) for ext in _RISKY_DOWNLOAD_EXTENSIONS):
return joined
+ if has_download_attr and any(
+ path.endswith(ext) for ext in _RISKY_DOWNLOAD_ATTR_EXTENSIONS
+ ):
+ return joined
return None
@@ -358,7 +367,11 @@ def _collect_download_links(soup: BeautifulSoup, base_url: str) -> list[str]:
href = anchor.get("href")
if not isinstance(href, str):
continue
- resolved = _is_risky_download_url(href, base_url)
+ resolved = _is_risky_download_url(
+ href,
+ base_url,
+ has_download_attr=_anchor_has_download_attr(anchor),
+ )
if resolved is not None:
_append_unique(links, resolved, limit=_MAX_DOWNLOAD_LINKS)
if len(links) >= _MAX_DOWNLOAD_LINKS:
diff --git a/app/services/content_analyzer/render.py b/app/services/content_analyzer/render.py
index 0350f3b..374c2ba 100644
--- a/app/services/content_analyzer/render.py
+++ b/app/services/content_analyzer/render.py
@@ -8,6 +8,7 @@
from __future__ import annotations
import asyncio
+import weakref
from dataclasses import dataclass
from typing import Any
from urllib.parse import urlparse
@@ -34,12 +35,16 @@ class RenderResult:
_render_semaphore: asyncio.Semaphore | None = None
+_render_semaphore_loop_ref: weakref.ReferenceType[asyncio.AbstractEventLoop] | None = None
def _get_render_semaphore() -> asyncio.Semaphore:
- global _render_semaphore
- if _render_semaphore is None:
+ global _render_semaphore, _render_semaphore_loop_ref
+ current_loop = asyncio.get_running_loop()
+ stored_loop = _render_semaphore_loop_ref() if _render_semaphore_loop_ref else None
+ if _render_semaphore is None or stored_loop is not current_loop:
_render_semaphore = asyncio.Semaphore(settings.content_render_concurrency)
+ _render_semaphore_loop_ref = weakref.ref(current_loop)
return _render_semaphore
diff --git a/app/services/db_independent_pipeline.py b/app/services/db_independent_pipeline.py
index 46ce2b7..312d5ca 100644
--- a/app/services/db_independent_pipeline.py
+++ b/app/services/db_independent_pipeline.py
@@ -8,13 +8,13 @@
from app.core.config import settings
from app.core.exceptions import NormalizationError
-from app.schemas.content_analysis import ContentAnalysisResult
+from app.schemas.content_analysis import ContentAnalysisResult, ContentSignal
from app.schemas.db_independent_pipeline import (
DbIndependentPipelineFailure,
DbIndependentPipelineStages,
DbIndependentPipelineSuccess,
)
-from app.schemas.domain_heuristic import DomainHeuristicResult
+from app.schemas.domain_heuristic import DomainHeuristicResult, DomainHeuristicSignal
from app.schemas.pipeline import (
PipelineStage,
PipelineStageTimings,
@@ -22,9 +22,14 @@
Verdict,
)
from app.schemas.unchain import UnchainResult
-from app.services.content_analyzer import analyze_content, skipped_already_danger
+from app.services.content_analyzer import analyze_content
from app.services.domain_heuristic import check_domain_heuristic
from app.services.normalizer import normalize_url
+from app.services.page_unavailability import (
+ PAGE_UNAVAILABLE_CODE,
+ content_page_unavailable,
+ unchain_page_unavailable,
+)
from app.services.pipeline_deadline import (
PipelineDeadline,
PipelineStageTimeoutError,
@@ -73,7 +78,7 @@ def _total_score(
def _redirect_signal_code(raw_signal: str) -> str | None:
if raw_signal.startswith("cross_origin:"):
- return "REDIRECT_CROSS_ORIGIN"
+ return DomainHeuristicSignal.REDIRECT_CROSS_ORIGIN.value
if raw_signal == "scheme_downgrade":
return "REDIRECT_SCHEME_DOWNGRADE"
if raw_signal == "redirect_loop":
@@ -87,6 +92,64 @@ def _redirect_signal_code(raw_signal: str) -> str | None:
return None
+def _augment_heuristic_with_redirect_signals(
+ heuristic: DomainHeuristicResult,
+ unchain: UnchainResult,
+) -> DomainHeuristicResult:
+ signals = list(heuristic.signals)
+ score = heuristic.score
+ if any(signal.startswith("cross_origin:") for signal in unchain.signals):
+ if DomainHeuristicSignal.REDIRECT_CROSS_ORIGIN not in signals:
+ signals.append(DomainHeuristicSignal.REDIRECT_CROSS_ORIGIN)
+ score = min(
+ score + settings.score_weight_redirect_cross_origin,
+ settings.domain_heuristic_score_cap,
+ )
+ if signals == heuristic.signals and score == heuristic.score:
+ return heuristic
+ return heuristic.model_copy(update={"signals": signals, "score": score})
+
+
+def _page_unavailable_content(
+ final_url: str,
+ *,
+ message: str,
+ status_code: int | None,
+) -> ContentAnalysisResult:
+ return ContentAnalysisResult(
+ final_url=final_url,
+ fetched=False,
+ status_code=status_code,
+ score=0,
+ signals=[ContentSignal.FETCH_FAILED],
+ reason=message,
+ error="page_unavailable",
+ )
+
+
+def _page_unavailable_failure(
+ *,
+ analysis_id: str,
+ original_url: str,
+ final_url: str,
+ failed_at_stage: PipelineStage,
+ error: str,
+ status_code: int | None,
+ started: float,
+ stage_timings: PipelineStageTimings,
+) -> DbIndependentPipelineFailure:
+ return DbIndependentPipelineFailure(
+ analysis_id=analysis_id,
+ original_url=original_url,
+ final_url=final_url,
+ failed_at_stage=failed_at_stage,
+ error=error,
+ error_code=PAGE_UNAVAILABLE_CODE,
+ status_code=status_code,
+ timings=_build_timings(started, stage_timings),
+ )
+
+
def _collect_db_independent_signals(
heuristic: DomainHeuristicResult,
unchain: UnchainResult,
@@ -148,6 +211,71 @@ async def run_db_independent_pipeline(
unchain = timed_out_unchain_result(normalize.normalized_url, error="stage_error")
_set_stage_timing(stage_timings, PipelineStage.UNCHAIN, stage_started)
+ if unavailable := unchain_page_unavailable(unchain):
+ message, status_code = unavailable
+ log.info(
+ "db_independent_pipeline.page_unavailable",
+ stage=PipelineStage.UNCHAIN,
+ final_url=unchain.final_url,
+ status_code=status_code,
+ error=unchain.error,
+ )
+ stage_started = time.perf_counter()
+ try:
+ heuristic = await deadline.run(
+ PipelineStage.DOMAIN_HEURISTIC.value,
+ check_domain_heuristic(unchain.final_url),
+ settings.pipeline_domain_timeout_seconds,
+ )
+ except PipelineStageTimeoutError:
+ log.warning(
+ "db_independent_pipeline.stage_timeout",
+ stage=PipelineStage.DOMAIN_HEURISTIC,
+ )
+ heuristic = timed_out_domain_result(unchain.final_url)
+ except Exception as exc:
+ log.warning(
+ "db_independent_pipeline.stage_error",
+ stage=PipelineStage.DOMAIN_HEURISTIC,
+ error=str(exc),
+ error_type=type(exc).__name__,
+ )
+ heuristic = timed_out_domain_result(unchain.final_url)
+ _set_stage_timing(stage_timings, PipelineStage.DOMAIN_HEURISTIC, stage_started)
+ heuristic = _augment_heuristic_with_redirect_signals(heuristic, unchain)
+ content = _page_unavailable_content(
+ unchain.final_url,
+ message=message,
+ status_code=status_code,
+ )
+ score = _total_score(heuristic, content)
+ if score >= settings.score_caution_threshold:
+ verdict = _decide_verdict(score)
+ return DbIndependentPipelineSuccess(
+ analysis_id=analysis_id,
+ original_url=original_url,
+ final_url=unchain.final_url,
+ verdict=verdict,
+ score=score,
+ timings=_build_timings(total_started, stage_timings),
+ stages=DbIndependentPipelineStages(
+ normalize=normalize,
+ unchain=unchain,
+ domain_heuristic=heuristic,
+ content_analysis=content,
+ ),
+ )
+ return _page_unavailable_failure(
+ analysis_id=analysis_id,
+ original_url=original_url,
+ final_url=unchain.final_url,
+ failed_at_stage=PipelineStage.UNCHAIN,
+ error=message,
+ status_code=status_code,
+ started=total_started,
+ stage_timings=stage_timings,
+ )
+
stage_started = time.perf_counter()
try:
heuristic = await deadline.run(
@@ -170,35 +298,68 @@ async def run_db_independent_pipeline(
)
heuristic = timed_out_domain_result(unchain.final_url)
_set_stage_timing(stage_timings, PipelineStage.DOMAIN_HEURISTIC, stage_started)
+ heuristic = _augment_heuristic_with_redirect_signals(heuristic, unchain)
- if heuristic.score >= settings.score_danger_threshold:
- stage_started = time.perf_counter()
- content = skipped_already_danger(unchain.final_url)
- _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, stage_started)
- else:
- upstream = _collect_db_independent_signals(heuristic, unchain)
- stage_started = time.perf_counter()
- try:
- content = await deadline.run(
- PipelineStage.CONTENT_ANALYSIS.value,
- analyze_content(unchain.final_url, upstream_signals=upstream),
- settings.pipeline_content_timeout_seconds,
- )
- except PipelineStageTimeoutError:
- log.warning(
- "db_independent_pipeline.stage_timeout",
- stage=PipelineStage.CONTENT_ANALYSIS,
- )
- content = timed_out_content_result(unchain.final_url)
- except Exception as exc:
- log.warning(
- "db_independent_pipeline.stage_error",
- stage=PipelineStage.CONTENT_ANALYSIS,
- error=str(exc),
- error_type=type(exc).__name__,
+ upstream = _collect_db_independent_signals(heuristic, unchain)
+ stage_started = time.perf_counter()
+ try:
+ content = await deadline.run(
+ PipelineStage.CONTENT_ANALYSIS.value,
+ analyze_content(unchain.final_url, upstream_signals=upstream),
+ settings.pipeline_content_timeout_seconds,
+ )
+ except PipelineStageTimeoutError:
+ log.warning(
+ "db_independent_pipeline.stage_timeout",
+ stage=PipelineStage.CONTENT_ANALYSIS,
+ )
+ content = timed_out_content_result(unchain.final_url)
+ except Exception as exc:
+ log.warning(
+ "db_independent_pipeline.stage_error",
+ stage=PipelineStage.CONTENT_ANALYSIS,
+ error=str(exc),
+ error_type=type(exc).__name__,
+ )
+ content = timed_out_content_result(unchain.final_url)
+ _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, stage_started)
+
+ if unavailable := content_page_unavailable(content):
+ message, status_code = unavailable
+ log.info(
+ "db_independent_pipeline.page_unavailable",
+ stage=PipelineStage.CONTENT_ANALYSIS,
+ final_url=unchain.final_url,
+ status_code=status_code,
+ error=content.error,
+ )
+ score = _total_score(heuristic, content)
+ if score >= settings.score_caution_threshold:
+ verdict = _decide_verdict(score)
+ return DbIndependentPipelineSuccess(
+ analysis_id=analysis_id,
+ original_url=original_url,
+ final_url=unchain.final_url,
+ verdict=verdict,
+ score=score,
+ timings=_build_timings(total_started, stage_timings),
+ stages=DbIndependentPipelineStages(
+ normalize=normalize,
+ unchain=unchain,
+ domain_heuristic=heuristic,
+ content_analysis=content,
+ ),
)
- content = timed_out_content_result(unchain.final_url)
- _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, stage_started)
+ return _page_unavailable_failure(
+ analysis_id=analysis_id,
+ original_url=original_url,
+ final_url=unchain.final_url,
+ failed_at_stage=PipelineStage.CONTENT_ANALYSIS,
+ error=message,
+ status_code=status_code,
+ started=total_started,
+ stage_timings=stage_timings,
+ )
score = _total_score(heuristic, content)
verdict = _decide_verdict(score)
diff --git a/app/services/domain_heuristic/brands.txt b/app/services/domain_heuristic/brands.txt
index 4cc00b9..3e19252 100644
--- a/app/services/domain_heuristic/brands.txt
+++ b/app/services/domain_heuristic/brands.txt
@@ -195,6 +195,7 @@ sktelecom.com
lguplus.com
uplus.co.kr
lg.com
+lg.co.kr
skbroadband.com
ktcs.co.kr
ktwiz.co.kr
@@ -428,11 +429,16 @@ github.com
gitlab.com
bitbucket.org
stackoverflow.com
+python.org
+postgresql.org
+typescriptlang.org
+rust-lang.org
docker.com
dockerhub.com
cloudflare.com
digitalocean.com
notion.so
+notion.com
atlassian.com
jira.atlassian.com
confluence.atlassian.com
@@ -548,6 +554,9 @@ razer.com
# ─────────────────────────────────────────
netflix.com
spotify.com
+ubuntu.com
+apnews.com
+ethz.ch
dropbox.com
wordpress.com
wix.com
diff --git a/app/services/domain_heuristic/check.py b/app/services/domain_heuristic/check.py
index 25aca97..ffb7785 100644
--- a/app/services/domain_heuristic/check.py
+++ b/app/services/domain_heuristic/check.py
@@ -41,6 +41,7 @@ def _signal_scores() -> dict[DomainHeuristicSignal, int]:
DomainHeuristicSignal.FREE_HOSTING_LURE: settings.score_weight_free_hosting_lure,
DomainHeuristicSignal.SENSITIVE_PATH: settings.score_weight_sensitive_path,
DomainHeuristicSignal.URL_SHORTENER: settings.score_weight_url_shortener,
+ DomainHeuristicSignal.REDIRECT_CROSS_ORIGIN: settings.score_weight_redirect_cross_origin,
}
diff --git a/app/services/domain_heuristic/dga.py b/app/services/domain_heuristic/dga.py
index fc583b4..1c62334 100644
--- a/app/services/domain_heuristic/dga.py
+++ b/app/services/domain_heuristic/dga.py
@@ -5,6 +5,7 @@
from app.core.config import settings
from app.core.tld import extract_url_parts
from app.schemas.domain_heuristic import DomainHeuristicSignal
+from app.services.domain_heuristic.patterns import is_trusted_registered_domain
_CONSONANTS = frozenset("bcdfghjklmnpqrstvwxyz")
@@ -31,6 +32,9 @@ def _consonant_ratio(s: str) -> float:
def check_dga(url: str) -> DomainHeuristicSignal | None:
+ if is_trusted_registered_domain(url):
+ return None
+
ext = extract_url_parts(url)
label = ext.domain # 등록 도메인 레이블
if not label or len(label) < _MIN_DGA_LABEL_LEN:
diff --git a/app/services/domain_heuristic/patterns.py b/app/services/domain_heuristic/patterns.py
index c3b3dd9..1e2e583 100644
--- a/app/services/domain_heuristic/patterns.py
+++ b/app/services/domain_heuristic/patterns.py
@@ -93,6 +93,13 @@
}
)
+_TRUSTED_BENIGN_DOMAINS = frozenset(
+ {
+ "example.com",
+ "httpbin.org",
+ }
+)
+
_SENSITIVE_PATH_TOKENS = frozenset(
{
"account",
@@ -232,6 +239,18 @@ def _trusted_brand_domains() -> frozenset[str]:
return frozenset(domains)
+def is_trusted_registered_domain(url: str) -> bool:
+ ext = extract_url_parts(url)
+ registered_domain = (ext.top_domain_under_public_suffix or "").lower()
+ return bool(
+ registered_domain
+ and (
+ registered_domain in _trusted_brand_domains()
+ or registered_domain in _TRUSTED_BENIGN_DOMAINS
+ )
+ )
+
+
def _brand_labels_in_url_text(text: str) -> set[str]:
lowered = text.lower()
tokens = re.findall(r"[a-z0-9]+", lowered)
diff --git a/app/services/page_unavailability.py b/app/services/page_unavailability.py
new file mode 100644
index 0000000..94049d4
--- /dev/null
+++ b/app/services/page_unavailability.py
@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+from app.schemas.content_analysis import ContentAnalysisResult
+from app.schemas.unchain import UnchainResult
+
+PAGE_UNAVAILABLE_CODE = "PAGE_UNAVAILABLE"
+
+_UNAVAILABLE_ERRORS: tuple[str, ...] = (
+ "dns_failure",
+ "timeout",
+ "connect_error",
+ "blocked_host",
+ "invalid_host",
+)
+
+
+def _message_for_status(status_code: int) -> str:
+ if status_code == 404:
+ return "페이지를 찾을 수 없습니다."
+ if 400 <= status_code < 500:
+ return "페이지 요청이 거부되었거나 찾을 수 없습니다."
+ return "대상 서버 오류로 페이지를 확인할 수 없습니다."
+
+
+def _message_for_error(error: str) -> str:
+ if error == "dns_failure":
+ return "도메인 주소를 확인할 수 없습니다."
+ if error == "timeout":
+ return "페이지 응답 시간이 초과되었습니다."
+ if error in {"connect_error"} or error.startswith("connection_refused"):
+ return "페이지에 연결할 수 없습니다."
+ if error in {"blocked_host", "invalid_host"}:
+ return "내부망 또는 차단된 호스트라 분석하지 않았습니다."
+ if error.startswith("server_error_"):
+ return "대상 서버 오류로 페이지를 확인할 수 없습니다."
+ if error.startswith("http_error_"):
+ return "페이지를 가져오지 못했습니다."
+ return "페이지를 확인할 수 없습니다."
+
+
+def _status_from_error(error: str) -> int | None:
+ for prefix in ("server_error_", "http_error_"):
+ if error.startswith(prefix):
+ raw = error.removeprefix(prefix)
+ if raw.isdigit():
+ return int(raw)
+ return None
+
+
+def unchain_page_unavailable(unchain: UnchainResult) -> tuple[str, int | None] | None:
+ """Return user message/status when unchain already proved the page is unavailable."""
+ for hop in reversed(unchain.hops):
+ if hop.status_code >= 400:
+ return _message_for_status(hop.status_code), hop.status_code
+
+ error = unchain.error
+ if error is None:
+ return None
+ if error in _UNAVAILABLE_ERRORS or error.startswith(
+ ("connection_refused", "server_error_", "http_error_")
+ ):
+ status_code = _status_from_error(error)
+ message = _message_for_status(status_code) if status_code else _message_for_error(error)
+ return message, status_code
+ return None
+
+
+def content_page_unavailable(content: ContentAnalysisResult) -> tuple[str, int | None] | None:
+ """Return user message/status when content fetch could not reach an analyzable page."""
+ if content.fetched:
+ return None
+ error = content.error
+ if error is None:
+ return None
+ if error in _UNAVAILABLE_ERRORS or error.startswith("http_error_"):
+ status_code = content.status_code or _status_from_error(error)
+ message = content.reason or (
+ _message_for_status(status_code) if status_code else _message_for_error(error)
+ )
+ return message, status_code
+ return None
diff --git a/app/services/pipeline.py b/app/services/pipeline.py
index 0b0eb01..407f98a 100644
--- a/app/services/pipeline.py
+++ b/app/services/pipeline.py
@@ -14,8 +14,12 @@
from app.core.config import settings
from app.core.exceptions import NormalizationError
from app.core.tld import extract_url_parts
-from app.schemas.content_analysis import ContentAnalysisResult
-from app.schemas.domain_heuristic import DomainHeuristicResult, DomainHeuristicSkippedReason
+from app.schemas.content_analysis import ContentAnalysisResult, ContentSignal
+from app.schemas.domain_heuristic import (
+ DomainHeuristicResult,
+ DomainHeuristicSignal,
+ DomainHeuristicSkippedReason,
+)
from app.schemas.normalize import NormalizeResult
from app.schemas.pipeline import (
PipelineFailure,
@@ -31,6 +35,11 @@
from app.services.content_analyzer import analyze_content, skipped_already_danger
from app.services.domain_heuristic import check_domain_heuristic
from app.services.normalizer import normalize_url
+from app.services.page_unavailability import (
+ PAGE_UNAVAILABLE_CODE,
+ content_page_unavailable,
+ unchain_page_unavailable,
+)
from app.services.pipeline_deadline import (
PipelineDeadline,
PipelineStageTimeoutError,
@@ -99,9 +108,15 @@ async def _stage_unchain(log: structlog.stdlib.BoundLogger, normalized_url: str)
async def _stage_threat_db(
- log: structlog.stdlib.BoundLogger, final_url: str, session: AsyncSession
+ log: structlog.stdlib.BoundLogger,
+ final_url: str,
+ session: AsyncSession,
+ original_url: str | None = None,
) -> ThreatDbResult:
- result = await check_threat_db(session, final_url)
+ if original_url and original_url != final_url:
+ result = await check_threat_db(session, final_url, original_url=original_url)
+ else:
+ result = await check_threat_db(session, final_url)
log.info(
"pipeline.threat_db.done",
is_malicious=result.is_malicious,
@@ -140,6 +155,41 @@ def _collect_upstream_signals(
return tuple(codes)
+def _augment_heuristic_with_redirect_signals(
+ heuristic: DomainHeuristicResult,
+ unchain: UnchainResult,
+) -> DomainHeuristicResult:
+ signals = list(heuristic.signals)
+ score = heuristic.score
+ if any(signal.startswith("cross_origin:") for signal in unchain.signals):
+ if DomainHeuristicSignal.REDIRECT_CROSS_ORIGIN not in signals:
+ signals.append(DomainHeuristicSignal.REDIRECT_CROSS_ORIGIN)
+ score = min(
+ score + settings.score_weight_redirect_cross_origin,
+ settings.domain_heuristic_score_cap,
+ )
+ if signals == heuristic.signals and score == heuristic.score:
+ return heuristic
+ return heuristic.model_copy(update={"signals": signals, "score": score})
+
+
+def _page_unavailable_content(
+ final_url: str,
+ *,
+ message: str,
+ status_code: int | None,
+) -> ContentAnalysisResult:
+ return ContentAnalysisResult(
+ final_url=final_url,
+ fetched=False,
+ status_code=status_code,
+ score=0,
+ signals=[ContentSignal.FETCH_FAILED],
+ reason=message,
+ error="page_unavailable",
+ )
+
+
async def _stage_content_analysis(
log: structlog.stdlib.BoundLogger,
final_url: str,
@@ -214,9 +264,33 @@ def _skipped_heuristic(final_url: str) -> DomainHeuristicResult:
)
+def _page_unavailable_failure(
+ *,
+ analysis_id: str,
+ original_url: str,
+ final_url: str,
+ failed_at_stage: PipelineStage,
+ error: str,
+ status_code: int | None,
+ started: float,
+ stage_timings: PipelineStageTimings,
+) -> PipelineFailure:
+ return PipelineFailure(
+ analysis_id=analysis_id,
+ original_url=original_url,
+ final_url=final_url,
+ failed_at_stage=failed_at_stage,
+ error=error,
+ error_code=PAGE_UNAVAILABLE_CODE,
+ status_code=status_code,
+ timings=_build_timings(started, stage_timings),
+ )
+
+
async def _run_stage_2_and_3(
log: structlog.stdlib.BoundLogger,
final_url: str,
+ original_url: str,
session: AsyncSession,
timings: PipelineStageTimings,
) -> tuple[ThreatDbResult, DomainHeuristicResult, bool]:
@@ -231,7 +305,7 @@ async def _run_stage_2_and_3(
_timed_async_stage(
timings,
PipelineStage.THREAT_DB,
- _stage_threat_db(log, final_url, session),
+ _stage_threat_db(log, final_url, session, original_url),
)
)
heur_task = asyncio.create_task(
@@ -335,13 +409,89 @@ async def run_pipeline(
if stage_timings.unchain is None:
_set_stage_timing(stage_timings, PipelineStage.UNCHAIN, stage_started)
+ if unavailable := unchain_page_unavailable(unchain):
+ message, status_code = unavailable
+ log.info(
+ "pipeline.page_unavailable",
+ stage=PipelineStage.UNCHAIN,
+ final_url=unchain.final_url,
+ status_code=status_code,
+ error=unchain.error,
+ )
+ try:
+ threat, heuristic, _ = await deadline.run(
+ "reputation",
+ _run_stage_2_and_3(
+ log,
+ unchain.final_url,
+ norm.normalized_url,
+ session,
+ stage_timings,
+ ),
+ settings.pipeline_reputation_timeout_seconds,
+ )
+ except PipelineStageTimeoutError:
+ log.warning("pipeline.stage_timeout", stage="reputation")
+ threat = timed_out_threat_db_result(unchain.final_url)
+ heuristic = timed_out_domain_result(unchain.final_url)
+ except Exception as exc:
+ log.warning(
+ "pipeline.stage_error",
+ stage="reputation",
+ error=str(exc),
+ error_type=type(exc).__name__,
+ )
+ threat = timed_out_threat_db_result(unchain.final_url)
+ heuristic = timed_out_domain_result(unchain.final_url)
+
+ heuristic = _augment_heuristic_with_redirect_signals(heuristic, unchain)
+ content = _page_unavailable_content(
+ unchain.final_url,
+ message=message,
+ status_code=status_code,
+ )
+ score = _total_score(threat, heuristic, content)
+ if threat.is_malicious or score >= settings.score_caution_threshold:
+ verdict = _decide_verdict(score, threat)
+ return PipelineSuccess(
+ analysis_id=analysis_id,
+ original_url=original_url,
+ final_url=unchain.final_url,
+ verdict=verdict,
+ score=score,
+ timings=_build_timings(total_started, stage_timings),
+ stages=PipelineStages(
+ normalize=norm,
+ unchain=unchain,
+ threat_db=threat,
+ domain_heuristic=heuristic,
+ content_analysis=content,
+ ),
+ )
+ return _page_unavailable_failure(
+ analysis_id=analysis_id,
+ original_url=original_url,
+ final_url=unchain.final_url,
+ failed_at_stage=PipelineStage.UNCHAIN,
+ error=message,
+ status_code=status_code,
+ started=total_started,
+ stage_timings=stage_timings,
+ )
+
# 2·3단계는 둘 다 unchain.final_url 만 필요하고 서로 독립이라 병렬로 돈다.
# threat_db 가 먼저 malicious 로 끝나면 verdict 가 이미 danger 로 확정이므로
# heuristic 을 cancel 하고 4단계까지 skip — 여기서 조기 종료가 일어난다.
try:
threat, heuristic, short_circuited = await deadline.run(
"reputation",
- _run_stage_2_and_3(log, unchain.final_url, session, stage_timings),
+ _run_stage_2_and_3(
+ log,
+ unchain.final_url,
+ norm.normalized_url,
+ session,
+ stage_timings,
+ ),
settings.pipeline_reputation_timeout_seconds,
)
except PipelineStageTimeoutError:
@@ -349,6 +499,7 @@ async def run_pipeline(
threat = timed_out_threat_db_result(unchain.final_url)
heuristic = timed_out_domain_result(unchain.final_url)
short_circuited = False
+
except Exception as exc:
log.warning(
"pipeline.stage_error",
@@ -360,6 +511,8 @@ async def run_pipeline(
heuristic = timed_out_domain_result(unchain.final_url)
short_circuited = False
+ heuristic = _augment_heuristic_with_redirect_signals(heuristic, unchain)
+
if short_circuited:
log.info(
"pipeline.short_circuit",
@@ -371,13 +524,13 @@ async def run_pipeline(
content = skipped_already_danger(unchain.final_url)
_set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, stage_started)
else:
- # 이미 danger 확정된 URL은 페이지를 받아보지 않는다 — 네트워크·AI 비용 절감.
- # 판정이 바뀌지 않을 단계에 초 단위 지연과 건당 원화를 쓸 이유가 없다.
+ # known malicious 는 verdict 가 이미 외부 DB 로 확정됐으므로 페이지를 받아보지 않는다.
+ # 휴리스틱 danger 는 페이지가 존재하지 않을 수 있으므로 content fetch 로 가용성을 확인한다.
preceding = _preceding_score(threat, heuristic)
- if threat.is_malicious or preceding >= settings.score_danger_threshold:
+ if threat.is_malicious:
log.info(
"pipeline.content_analysis.skipped",
- reason=("threat_db_match" if threat.is_malicious else "already_danger"),
+ reason="threat_db_match",
preceding_score=preceding,
)
stage_started = time.perf_counter()
@@ -407,6 +560,44 @@ async def run_pipeline(
)
content = timed_out_content_result(unchain.final_url)
+ if unavailable := content_page_unavailable(content):
+ message, status_code = unavailable
+ log.info(
+ "pipeline.page_unavailable",
+ stage=PipelineStage.CONTENT_ANALYSIS,
+ final_url=unchain.final_url,
+ status_code=status_code,
+ error=content.error,
+ )
+ score = _total_score(threat, heuristic, content)
+ if threat.is_malicious or score >= settings.score_caution_threshold:
+ verdict = _decide_verdict(score, threat)
+ return PipelineSuccess(
+ analysis_id=analysis_id,
+ original_url=original_url,
+ final_url=unchain.final_url,
+ verdict=verdict,
+ score=score,
+ timings=_build_timings(total_started, stage_timings),
+ stages=PipelineStages(
+ normalize=norm,
+ unchain=unchain,
+ threat_db=threat,
+ domain_heuristic=heuristic,
+ content_analysis=content,
+ ),
+ )
+ return _page_unavailable_failure(
+ analysis_id=analysis_id,
+ original_url=original_url,
+ final_url=unchain.final_url,
+ failed_at_stage=PipelineStage.CONTENT_ANALYSIS,
+ error=message,
+ status_code=status_code,
+ started=total_started,
+ stage_timings=stage_timings,
+ )
+
score = _total_score(threat, heuristic, content)
verdict = _decide_verdict(score, threat)
log.info(
diff --git a/app/services/threat_db/check.py b/app/services/threat_db/check.py
index 2a7cbd8..0b8c636 100644
--- a/app/services/threat_db/check.py
+++ b/app/services/threat_db/check.py
@@ -24,34 +24,81 @@ def _merge_threat_types(gsb: GSBResult, urlhaus: URLhausResult) -> list[str]:
return seen
-async def check_threat_db(session: AsyncSession, final_url: str) -> ThreatDbResult:
+def _candidate_urls(final_url: str, original_url: str | None) -> list[str]:
+ candidates = [final_url]
+ if original_url and original_url != final_url:
+ candidates.append(original_url)
+ return candidates
+
+
+def _merge_gsb(results: list[GSBResult]) -> GSBResult:
+ checked = any(result.checked for result in results)
+ matches: list = []
+ error = None
+ for result in results:
+ if result.matches:
+ matches.extend(result.matches)
+ if error is None and result.error:
+ error = result.error
+ return GSBResult(
+ checked=checked,
+ is_threat=bool(matches),
+ matches=matches,
+ error=None if checked else error,
+ )
+
+
+def _merge_urlhaus(results: list[URLhausResult]) -> URLhausResult:
+ for result in results:
+ if result.is_threat:
+ return result
+ checked = any(result.checked for result in results)
+ error = next((result.error for result in results if result.error), None)
+ return URLhausResult(checked=checked, is_threat=False, error=None if checked else error)
+
+
+async def check_threat_db(
+ session: AsyncSession,
+ final_url: str,
+ *,
+ original_url: str | None = None,
+) -> ThreatDbResult:
"""final_url 을 GSB + URLhaus 와 병렬 대조해 판정 결과 반환.
어느 한쪽이 실패해도 다른 쪽 결과로 판정한다. 두 쪽 다 실패 시
is_malicious=False, sources_checked=0 로 반환하여 상위 레이어가 보수적으로 처리.
"""
- gsb_task = check_gsb(final_url)
- urlhaus_task = check_urlhaus(session, final_url)
+ candidates = _candidate_urls(final_url, original_url)
+ gsb_tasks = [check_gsb(url) for url in candidates]
+ urlhaus_tasks = [check_urlhaus(session, url) for url in candidates]
- gsb_raw, urlhaus_raw = await asyncio.gather(gsb_task, urlhaus_task, return_exceptions=True)
+ raw_results = await asyncio.gather(*gsb_tasks, *urlhaus_tasks, return_exceptions=True)
# CancelledError 는 상위 task 의 취소 신호이므로 절대 삼키지 않는다.
# (shutdown / 요청 timeout 시 degraded 결과를 영속화하는 사고 방지)
- for raw in (gsb_raw, urlhaus_raw):
+ for raw in raw_results:
if isinstance(raw, asyncio.CancelledError):
raise raw
- if isinstance(gsb_raw, BaseException):
- logger.warning("threat_db.gsb_unexpected", error=str(gsb_raw))
- gsb = GSBResult(checked=False, is_threat=False, error="unexpected")
- else:
- gsb = gsb_raw
-
- if isinstance(urlhaus_raw, BaseException):
- logger.warning("threat_db.urlhaus_unexpected", error=str(urlhaus_raw))
- urlhaus = URLhausResult(checked=False, is_threat=False, error="unexpected")
- else:
- urlhaus = urlhaus_raw
+ gsb_results: list[GSBResult] = []
+ urlhaus_results: list[URLhausResult] = []
+ for raw in raw_results[: len(candidates)]:
+ if isinstance(raw, BaseException):
+ logger.warning("threat_db.gsb_unexpected", error=str(raw))
+ gsb_results.append(GSBResult(checked=False, is_threat=False, error="unexpected"))
+ else:
+ gsb_results.append(raw)
+ for raw in raw_results[len(candidates) :]:
+ if isinstance(raw, BaseException):
+ logger.warning("threat_db.urlhaus_unexpected", error=str(raw))
+ urlhaus_results.append(
+ URLhausResult(checked=False, is_threat=False, error="unexpected")
+ )
+ else:
+ urlhaus_results.append(raw)
+
+ gsb = _merge_gsb(gsb_results)
+ urlhaus = _merge_urlhaus(urlhaus_results)
is_malicious = gsb.is_threat or urlhaus.is_threat
sources_checked = sum((gsb.checked, urlhaus.checked))
diff --git a/app/services/threat_db/match_keys.py b/app/services/threat_db/match_keys.py
index 80c2c52..accf888 100644
--- a/app/services/threat_db/match_keys.py
+++ b/app/services/threat_db/match_keys.py
@@ -1,7 +1,7 @@
"""URLhaus 조회·동기화에 쓰는 매칭 키 생성.
-host 한 개가 기본이지만 GitHub/GitLab 같은 다중 테넌트 호스트는
-계정/리포 레벨에서 악성 여부가 갈리므로 host+path-prefix 키도 함께 생성.
+host 한 개가 기본이지만 GitHub/GitLab/Dropbox 같은 다중 테넌트 호스트는
+계정/리포/공유 파일 레벨에서 악성 여부가 갈리므로 host+path-prefix 키만 사용한다.
"""
from __future__ import annotations
@@ -14,7 +14,7 @@
def derive_keys(url: str) -> list[str]:
"""URL 에서 매칭 키 후보를 더 구체적인 순서로 반환.
- 반환: [host_path, host] 또는 [host]
+ 반환: [host_path] 또는 [host]
host 추출 실패 시 빈 리스트.
"""
parsed = urlparse(url)
@@ -28,8 +28,8 @@ def derive_keys(url: str) -> list[str]:
segments = [seg for seg in parsed.path.split("/") if seg]
if len(segments) < required:
- return [host]
+ return []
prefix = "/".join(segments[:required])
host_path = f"{host}/{prefix}"
- return [host_path, host]
+ return [host_path]
diff --git a/app/services/threat_db/urlhaus.py b/app/services/threat_db/urlhaus.py
index bd2ceb9..6602ebd 100644
--- a/app/services/threat_db/urlhaus.py
+++ b/app/services/threat_db/urlhaus.py
@@ -6,6 +6,7 @@
from __future__ import annotations
from typing import Literal
+from urllib.parse import urlsplit, urlunsplit
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError
@@ -41,14 +42,52 @@ def _to_result(
)
+def _url_variants(url: str) -> list[str]:
+ variants: list[str] = []
+
+ def add(candidate: str) -> None:
+ if candidate and candidate not in variants:
+ variants.append(candidate)
+
+ add(url)
+ try:
+ parts = urlsplit(url)
+ except ValueError:
+ return variants
+ if not parts.scheme or not parts.netloc:
+ return variants
+
+ path_variants = [parts.path]
+ if parts.path == "":
+ path_variants.append("/")
+ elif parts.path == "/":
+ path_variants.append("")
+ elif parts.path.endswith("/"):
+ path_variants.append(parts.path.rstrip("/"))
+ else:
+ path_variants.append(parts.path + "/")
+
+ schemes = [parts.scheme]
+ if parts.scheme == "https":
+ schemes.append("http")
+ elif parts.scheme == "http":
+ schemes.append("https")
+
+ for scheme in schemes:
+ for path in path_variants:
+ add(urlunsplit((scheme, parts.netloc, path, parts.query, parts.fragment)))
+ return variants
+
+
async def check_urlhaus(session: AsyncSession, url: str) -> URLhausResult:
"""URLhaus 로컬 스냅샷에서 URL 매칭 여부 조회."""
try:
# 1) URL 완전일치
- stmt = select(URLhausEntry).where(URLhausEntry.url == url)
- row = (await session.execute(stmt)).scalar_one_or_none()
- if row is not None:
- return _to_result(row, "url", row.url)
+ for candidate in _url_variants(url):
+ stmt = select(URLhausEntry).where(URLhausEntry.url == candidate)
+ row = (await session.execute(stmt)).scalar_one_or_none()
+ if row is not None:
+ return _to_result(row, "url", row.url)
# 2) match_key IN (...)
keys = derive_keys(url)
diff --git a/app/services/threat_db/urlhaus_sync.py b/app/services/threat_db/urlhaus_sync.py
index 265352d..16b81aa 100644
--- a/app/services/threat_db/urlhaus_sync.py
+++ b/app/services/threat_db/urlhaus_sync.py
@@ -79,7 +79,7 @@ def _derive_match_key(url: str) -> tuple[str, str] | None:
return None
parsed = urlparse(url)
host = (parsed.hostname or "").lower()
- # derive_keys 는 [host_path, host] 또는 [host] — 첫 원소가 가장 구체적 키.
+ # derive_keys 는 [host_path] 또는 [host] — 첫 원소가 저장할 match_key.
return host, keys[0]
diff --git a/app/services/unchainer/unchain.py b/app/services/unchainer/unchain.py
index cf4c54b..92e81df 100644
--- a/app/services/unchainer/unchain.py
+++ b/app/services/unchainer/unchain.py
@@ -18,6 +18,7 @@
from app.core.config import settings
from app.core.dns_cache import resolve_host_addrs
+from app.core.tld import extract_url_parts
from app.schemas.analysis import HopRecord, UnchainResult
_REDIRECT_CODES: frozenset[int] = frozenset({301, 302, 303, 307, 308})
@@ -97,6 +98,17 @@ def _https_variant(url: str) -> str | None:
return urlunparse(parsed._replace(scheme="https"))
+def _registered_domain_from_host(host: str | None) -> str:
+ if not host:
+ return ""
+ ext = extract_url_parts(f"https://{host.strip('[]')}/")
+ return (ext.top_domain_under_public_suffix or host).lower()
+
+
+def _is_registered_domain_change(left_host: str | None, right_host: str | None) -> bool:
+ return _registered_domain_from_host(left_host) != _registered_domain_from_host(right_host)
+
+
def _is_analyzable_html_response(resp: httpx.Response) -> bool:
content_type = resp.headers.get("content-type", "").lower()
return 200 <= resp.status_code < 400 and "text/html" in content_type
@@ -162,8 +174,10 @@ async def _unchain_url_inner(
"Accept-Language": "ko-KR,ko;q=0.9,en-US;q=0.8,en;q=0.7",
}
- client = _get_client()
- if prefer_https_when_schemeless and await _https_responds(client, current_url, headers):
+ client: httpx.AsyncClient | None = None
+ if prefer_https_when_schemeless:
+ client = _get_client()
+ if client is not None and await _https_responds(client, current_url, headers):
https_url = _https_variant(current_url)
if https_url is not None:
current_url = https_url
@@ -183,6 +197,9 @@ async def _unchain_url_inner(
signals.append("ssrf_blocked")
break
+ if client is None:
+ client = _get_client()
+
hop, next_url, hop_error = await _follow_one_hop(
client,
current_url,
@@ -306,8 +323,8 @@ async def _follow_one_hop(
if parsed_url.scheme == "https" and parsed_next.scheme == "http":
signals.append("scheme_downgrade")
- # 크로스 오리진 호스트 변화
- if parsed_url.hostname != parsed_next.hostname:
+ # 등록 도메인 변화. www → bare 같은 정상 canonical redirect 는 제외한다.
+ if _is_registered_domain_change(parsed_url.hostname, parsed_next.hostname):
signals.append(f"cross_origin:{parsed_url.hostname}->{parsed_next.hostname}")
return hop, next_url, None
diff --git a/pyproject.toml b/pyproject.toml
index 9acba7b..6c4de16 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -20,10 +20,12 @@ dependencies = [
"lxml>=5.3.0",
"cachetools>=5.5.0",
"openai>=1.60.0",
- "playwright>=1.45.0",
]
[project.optional-dependencies]
+render = [
+ "playwright>=1.45.0",
+]
dev = [
"pytest>=8.3.0",
"pytest-asyncio>=0.24.0",
@@ -77,6 +79,7 @@ exclude = ["alembic/"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
+asyncio_default_test_loop_scope = "function"
testpaths = ["tests"]
addopts = "-ra --strict-markers --strict-config"
filterwarnings = ["error"]
diff --git a/tests/services/content_analyzer/test_ai_openai.py b/tests/services/content_analyzer/test_ai_openai.py
index 2336294..c46ce72 100644
--- a/tests/services/content_analyzer/test_ai_openai.py
+++ b/tests/services/content_analyzer/test_ai_openai.py
@@ -132,6 +132,9 @@ async def test_openai_prompt_requests_short_expert_plain_korean_reason() -> None
assert "100자" in system_msg
assert "보안 전문가" in system_msg
assert "쉬운 한국어" in system_msg
+ assert "사용자 행동 가이드" in system_msg
+ assert "비밀번호" in system_msg
+ assert "결제" in system_msg
async def test_openai_truncates_reason_to_100_chars() -> None:
diff --git a/tests/services/content_analyzer/test_analyze.py b/tests/services/content_analyzer/test_analyze.py
index 5601963..3bdae78 100644
--- a/tests/services/content_analyzer/test_analyze.py
+++ b/tests/services/content_analyzer/test_analyze.py
@@ -225,6 +225,44 @@ async def infer(self, ctx: AIPromptContext) -> AIInference:
assert result.ai_verdict == AIVerdict.SUSPICIOUS
assert result.score == settings.score_weight_ai_suspicious
+ assert result.score >= settings.score_caution_threshold
+
+ async def test_ai_suspicious_on_trusted_clean_domain_does_not_score(
+ self, monkeypatch: pytest.MonkeyPatch
+ ) -> None:
+ class StubAI:
+ async def infer(self, ctx: AIPromptContext) -> AIInference:
+ return AIInference(verdict=AIVerdict.SUSPICIOUS, reason="mild")
+
+ monkeypatch.setattr(
+ "app.services.content_analyzer.analyze.get_ai_provider",
+ lambda: StubAI(),
+ )
+ with _mock_fetch(ok=True, html="Reddit "):
+ result = await analyze_content("https://www.reddit.com/")
+
+ assert result.ai_verdict == AIVerdict.SUSPICIOUS
+ assert result.score == 0
+
+ @pytest.mark.parametrize("url", ["https://www.lg.co.kr/", "https://www.python.org/"])
+ async def test_ai_suspicious_on_trusted_redirect_target_does_not_score(
+ self,
+ monkeypatch: pytest.MonkeyPatch,
+ url: str,
+ ) -> None:
+ class StubAI:
+ async def infer(self, ctx: AIPromptContext) -> AIInference:
+ return AIInference(verdict=AIVerdict.SUSPICIOUS, reason="mild")
+
+ monkeypatch.setattr(
+ "app.services.content_analyzer.analyze.get_ai_provider",
+ lambda: StubAI(),
+ )
+ with _mock_fetch(ok=True, html="Trusted "):
+ result = await analyze_content(url)
+
+ assert result.ai_verdict == AIVerdict.SUSPICIOUS
+ assert result.score == 0
async def test_ai_benign_verdict_no_score(self, monkeypatch: pytest.MonkeyPatch) -> None:
class StubAI:
diff --git a/tests/services/content_analyzer/test_extract.py b/tests/services/content_analyzer/test_extract.py
index 5655a22..09bdcce 100644
--- a/tests/services/content_analyzer/test_extract.py
+++ b/tests/services/content_analyzer/test_extract.py
@@ -212,6 +212,32 @@ def test_extracts_risky_download_links_and_lure_text(self) -> None:
assert "카카오톡" in features.korean_lure_keywords
assert "카카오톡 최신버전 다운로드" in features.cta_texts
+ def test_regular_js_anchor_is_not_risky_download(self) -> None:
+ html = """
+
+
+ bundle
+
+
+ """
+
+ features = extract_features(html, base_url="https://normal.test/")
+
+ assert features.download_links == []
+
+ def test_download_js_anchor_is_risky_download(self) -> None:
+ html = """
+
+
+ download script
+
+
+ """
+
+ features = extract_features(html, base_url="https://suspicious.test/")
+
+ assert features.download_links == ["https://suspicious.test/payload.js"]
+
def test_extraction_caps_high_signal_lists(self) -> None:
inputs = "".join(
f'휴대폰 번호 {i} '
diff --git a/tests/services/content_analyzer/test_render.py b/tests/services/content_analyzer/test_render.py
new file mode 100644
index 0000000..a0b438d
--- /dev/null
+++ b/tests/services/content_analyzer/test_render.py
@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+import pytest
+from app.services.content_analyzer import render
+
+
+@pytest.mark.asyncio
+async def test_render_semaphore_is_recreated_per_event_loop(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ first = render._get_render_semaphore()
+ monkeypatch.setattr(render, "_render_semaphore_loop_ref", lambda: object())
+ second = render._get_render_semaphore()
+
+ assert first is not second
diff --git a/tests/services/domain_heuristic/test_check.py b/tests/services/domain_heuristic/test_check.py
index 0ff1da7..c1c081a 100644
--- a/tests/services/domain_heuristic/test_check.py
+++ b/tests/services/domain_heuristic/test_check.py
@@ -139,3 +139,30 @@ async def test_known_safe_domains_not_malicious():
for url in safe_urls:
result = await check_domain_heuristic(url)
assert result.score < 50, f"{url} score={result.score}, signals={result.signals}"
+
+
+@pytest.mark.asyncio
+async def test_known_safe_dga_like_domains_are_not_flagged():
+ safe_urls = [
+ "https://www.stackoverflow.com/",
+ "https://www.postgresql.org/",
+ "https://www.typescriptlang.org/",
+ ]
+ with patch(_RDAP_PATH, new_callable=AsyncMock) as mock_rdap:
+ mock_rdap.return_value = (None, "not_found")
+ for url in safe_urls:
+ result = await check_domain_heuristic(url)
+ assert DomainHeuristicSignal.DGA_LIKE not in result.signals, (
+ f"{url} score={result.score}, signals={result.signals}"
+ )
+ assert result.score < 31
+
+
+@pytest.mark.asyncio
+async def test_known_safe_alias_domain_is_not_typo():
+ with patch(_RDAP_PATH, new_callable=AsyncMock) as mock_rdap:
+ mock_rdap.return_value = (None, "not_found")
+ result = await check_domain_heuristic("https://www.notion.com/")
+
+ assert DomainHeuristicSignal.TYPO_DOMAIN not in result.signals
+ assert result.score < 31
diff --git a/tests/services/test_analysis_callback.py b/tests/services/test_analysis_callback.py
index acbbe85..b8cfcf9 100644
--- a/tests/services/test_analysis_callback.py
+++ b/tests/services/test_analysis_callback.py
@@ -297,3 +297,43 @@ async def test_posts_failure_callback_payload(monkeypatch: pytest.MonkeyPatch) -
}
assert "timings" not in payload
assert "stages" not in payload
+
+
+@pytest.mark.asyncio
+async def test_posts_page_unavailable_callback_without_verdict(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ monkeypatch.setattr(settings, "spring_internal_url", "http://spring.internal")
+ mock_client = AsyncMock()
+ mock_client.__aenter__.return_value = mock_client
+ mock_client.post.return_value = httpx.Response(200)
+ result = PipelineFailure(
+ analysis_id="aid-missing",
+ original_url="https://missing.test/",
+ final_url="https://missing.test/",
+ failed_at_stage=PipelineStage.CONTENT_ANALYSIS,
+ error="페이지를 찾을 수 없습니다.",
+ error_code="PAGE_UNAVAILABLE",
+ status_code=404,
+ )
+
+ with patch("app.services.analysis_callback.httpx.AsyncClient", return_value=mock_client):
+ delivered = await post_analysis_callback(
+ result,
+ request_id="rid-missing",
+ elapsed_ms=17,
+ analyzed_at=datetime(2026, 4, 7, 5, 43, 41, tzinfo=UTC),
+ )
+
+ assert delivered is True
+ payload = mock_client.post.await_args.kwargs["json"]
+ assert payload["status"] == "failed"
+ assert payload["finalUrl"] == "https://missing.test/"
+ assert payload["error"] == {
+ "code": "PAGE_UNAVAILABLE",
+ "stage": 4,
+ "message": "페이지를 찾을 수 없습니다.",
+ "statusCode": 404,
+ }
+ assert "verdict" not in payload
+ assert "score" not in payload
diff --git a/tests/services/test_db_independent_pipeline.py b/tests/services/test_db_independent_pipeline.py
index 9d34db3..7da654f 100644
--- a/tests/services/test_db_independent_pipeline.py
+++ b/tests/services/test_db_independent_pipeline.py
@@ -8,7 +8,7 @@
import pytest
from app.core.config import settings
-from app.schemas.content_analysis import ContentAnalysisResult
+from app.schemas.content_analysis import ContentAnalysisResult, ContentSignal
from app.schemas.db_independent_pipeline import (
DbIndependentPipelineFailure,
DbIndependentPipelineSuccess,
@@ -109,8 +109,9 @@ async def test_db_independent_pipeline_passes_url_and_redirect_signals_to_conten
mock_heuristic.return_value = _make_heuristic(15)
mock_content.return_value = _make_content(final_url)
- await run_db_independent_pipeline("aid-sig", "https://short.test/a")
+ result = await run_db_independent_pipeline("aid-sig", "https://short.test/a")
+ assert isinstance(result, DbIndependentPipelineSuccess)
mock_content.assert_awaited_once()
args, kwargs = mock_content.await_args
assert args == (final_url,)
@@ -121,11 +122,13 @@ async def test_db_independent_pipeline_passes_url_and_redirect_signals_to_conten
"HOSTING_PLATFORM",
"REDIRECT_CROSS_ORIGIN",
)
+ assert DomainHeuristicSignal.REDIRECT_CROSS_ORIGIN in result.stages.domain_heuristic.signals
+ assert result.stages.domain_heuristic.score > 15
assert "provider" not in kwargs
@pytest.mark.asyncio
-async def test_db_independent_pipeline_skips_content_when_heuristic_is_danger() -> None:
+async def test_db_independent_pipeline_checks_content_when_heuristic_is_danger() -> None:
final_url = "https://danger.example.com/login"
content_started = asyncio.Event()
@@ -157,9 +160,86 @@ async def _slow_content(_: str, **__: object) -> ContentAnalysisResult:
result = await run_db_independent_pipeline("aid-parallel", final_url)
assert isinstance(result, DbIndependentPipelineSuccess)
- assert result.score == 65
- assert result.stages.content_analysis.error == "skipped_already_danger"
- assert content_started.is_set() is False
+ assert result.score == 85
+ assert result.verdict == Verdict.DANGER
+ assert result.stages.content_analysis.fetched is True
+ assert content_started.is_set() is True
+ mock_content.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_db_independent_pipeline_returns_failure_when_page_unavailable() -> None:
+ final_url = "https://missing.example.com/"
+
+ with (
+ patch("app.services.db_independent_pipeline.normalize_url") as mock_norm,
+ patch(
+ "app.services.db_independent_pipeline.unchain_url", new_callable=AsyncMock
+ ) as mock_unchain,
+ patch(
+ "app.services.db_independent_pipeline.check_domain_heuristic",
+ new_callable=AsyncMock,
+ ) as mock_heuristic,
+ patch(
+ "app.services.db_independent_pipeline.analyze_content", new_callable=AsyncMock
+ ) as mock_content,
+ ):
+ mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url)
+ mock_unchain.return_value = _make_unchain(final_url)
+ mock_heuristic.return_value = _make_heuristic(0)
+ mock_content.return_value = ContentAnalysisResult(
+ final_url=final_url,
+ fetched=False,
+ status_code=404,
+ score=0,
+ signals=[ContentSignal.FETCH_FAILED],
+ reason="페이지를 찾을 수 없습니다.",
+ error="http_error_404",
+ )
+
+ result = await run_db_independent_pipeline("aid-missing", final_url)
+
+ assert isinstance(result, DbIndependentPipelineFailure)
+ assert result.failed_at_stage == PipelineStage.CONTENT_ANALYSIS
+ assert result.error_code == "PAGE_UNAVAILABLE"
+ assert result.final_url == final_url
+ assert result.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_db_independent_pipeline_returns_verdict_when_unavailable_url_signal_is_strong() -> None:
+ final_url = "http://xj3kq9vbnm2p7zla.com/login"
+ unchain = _make_unchain(final_url)
+ unchain.error = "dns_failure"
+
+ with (
+ patch("app.services.db_independent_pipeline.normalize_url") as mock_norm,
+ patch(
+ "app.services.db_independent_pipeline.unchain_url", new_callable=AsyncMock
+ ) as mock_unchain,
+ patch(
+ "app.services.db_independent_pipeline.check_domain_heuristic",
+ new_callable=AsyncMock,
+ ) as mock_heuristic,
+ patch(
+ "app.services.db_independent_pipeline.analyze_content", new_callable=AsyncMock
+ ) as mock_content,
+ ):
+ mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url)
+ mock_unchain.return_value = unchain
+ mock_heuristic.return_value = DomainHeuristicResult(
+ domain="xj3kq9vbnm2p7zla.com",
+ score=settings.score_caution_threshold,
+ signals=[DomainHeuristicSignal.DGA_LIKE],
+ )
+
+ result = await run_db_independent_pipeline("aid-unavailable-url-signal", final_url)
+
+ assert isinstance(result, DbIndependentPipelineSuccess)
+ assert result.verdict == Verdict.CAUTION
+ assert result.score == settings.score_caution_threshold
+ assert result.stages.content_analysis.fetched is False
+ assert result.stages.content_analysis.error == "page_unavailable"
mock_content.assert_not_awaited()
diff --git a/tests/services/test_pipeline.py b/tests/services/test_pipeline.py
index c8abe95..3958ab3 100644
--- a/tests/services/test_pipeline.py
+++ b/tests/services/test_pipeline.py
@@ -14,15 +14,21 @@
from app.schemas.normalize import NormalizeResult
from app.schemas.pipeline import PipelineFailure, PipelineStage, PipelineSuccess, Verdict
from app.schemas.threat_db import GSBMatch, GSBResult, ThreatDbResult, URLhausResult
-from app.schemas.unchain import UnchainResult
+from app.schemas.unchain import HopRecord, UnchainResult
from app.services.pipeline import run_pipeline
if TYPE_CHECKING:
from sqlalchemy.ext.asyncio import AsyncSession
-def _make_unchain(final_url: str) -> UnchainResult:
- return UnchainResult(input_url=final_url, final_url=final_url, hops=[], hop_count=0, signals=[])
+def _make_unchain(final_url: str, *, signals: list[str] | None = None) -> UnchainResult:
+ return UnchainResult(
+ input_url=final_url,
+ final_url=final_url,
+ hops=[],
+ hop_count=0,
+ signals=signals or [],
+ )
def _make_threat(final_url: str) -> ThreatDbResult:
@@ -49,6 +55,18 @@ def _make_content(final_url: str, *, score: int = 0) -> ContentAnalysisResult:
return ContentAnalysisResult(final_url=final_url, fetched=True, score=score, signals=[])
+def _missing_content(final_url: str, *, status_code: int = 404) -> ContentAnalysisResult:
+ return ContentAnalysisResult(
+ final_url=final_url,
+ fetched=False,
+ status_code=status_code,
+ score=0,
+ signals=[ContentSignal.FETCH_FAILED],
+ reason="페이지를 찾을 수 없습니다.",
+ error=f"http_error_{status_code}",
+ )
+
+
async def _resolve_upstream(value: object) -> object:
if inspect.isawaitable(value):
return await value
@@ -243,6 +261,86 @@ async def test_run_pipeline_runs_content_when_below_danger(
assert await _resolve_upstream(kwargs["upstream_signals"]) == ()
+@pytest.mark.asyncio
+async def test_run_pipeline_returns_failure_when_unchain_sees_404(
+ async_session: AsyncSession,
+) -> None:
+ final_url = "https://missing.test/not-found"
+ unchain = UnchainResult(
+ input_url=final_url,
+ final_url=final_url,
+ hops=[],
+ hop_count=0,
+ signals=[],
+ )
+ unchain.hops.append(HopRecord(url=final_url, status_code=404))
+ unchain.hop_count = 1
+
+ with (
+ patch("app.services.pipeline.normalize_url") as mock_norm,
+ patch("app.services.pipeline.unchain_url", new_callable=AsyncMock) as mock_unchain,
+ patch("app.services.pipeline.check_threat_db", new_callable=AsyncMock) as mock_threat,
+ patch(
+ "app.services.pipeline.check_domain_heuristic", new_callable=AsyncMock
+ ) as mock_heuristic,
+ patch("app.services.pipeline.analyze_content", new_callable=AsyncMock) as mock_content,
+ ):
+ mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url)
+ mock_unchain.return_value = unchain
+ mock_threat.return_value = _make_threat(final_url)
+ mock_heuristic.return_value = _heuristic_with_score(0)
+
+ result = await run_pipeline("aid-404", final_url, async_session)
+
+ assert isinstance(result, PipelineFailure)
+ assert result.failed_at_stage == PipelineStage.UNCHAIN
+ assert result.error_code == "PAGE_UNAVAILABLE"
+ assert result.final_url == final_url
+ assert result.status_code == 404
+ assert "페이지를 찾을 수 없습니다" in result.error
+ mock_threat.assert_awaited_once()
+ mock_heuristic.assert_awaited_once()
+ mock_content.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_run_pipeline_returns_failure_when_content_fetch_cannot_connect(
+ async_session: AsyncSession,
+) -> None:
+ final_url = "https://offline.test/"
+
+ with (
+ patch("app.services.pipeline.normalize_url") as mock_norm,
+ patch("app.services.pipeline.unchain_url", new_callable=AsyncMock) as mock_unchain,
+ patch("app.services.pipeline.check_threat_db", new_callable=AsyncMock) as mock_threat,
+ patch(
+ "app.services.pipeline.check_domain_heuristic", new_callable=AsyncMock
+ ) as mock_heuristic,
+ patch("app.services.pipeline.analyze_content", new_callable=AsyncMock) as mock_content,
+ ):
+ mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url)
+ mock_unchain.return_value = _make_unchain(final_url)
+ mock_threat.return_value = _make_threat(final_url)
+ mock_heuristic.return_value = _heuristic_with_score(0)
+ mock_content.return_value = ContentAnalysisResult(
+ final_url=final_url,
+ fetched=False,
+ score=0,
+ signals=[ContentSignal.FETCH_FAILED],
+ reason="페이지에 연결할 수 없습니다.",
+ error="connect_error",
+ )
+
+ result = await run_pipeline("aid-connect", final_url, async_session)
+
+ assert isinstance(result, PipelineFailure)
+ assert result.failed_at_stage == PipelineStage.CONTENT_ANALYSIS
+ assert result.error_code == "PAGE_UNAVAILABLE"
+ assert result.final_url == final_url
+ assert result.status_code is None
+ assert result.error == "페이지에 연결할 수 없습니다."
+
+
class TestVerdictAndScore:
"""PipelineSuccess.verdict / score 매핑 회귀."""
@@ -387,6 +485,43 @@ async def test_run_pipeline_passes_upstream_signals_to_content_analysis(
assert await _resolve_upstream(kwargs["upstream_signals"]) == ("TYPO_DOMAIN", "NEW_DOMAIN")
+@pytest.mark.asyncio
+async def test_run_pipeline_scores_cross_origin_redirect_signal(
+ async_session: AsyncSession,
+) -> None:
+ final_url = "https://redirected.example.com/login"
+
+ with (
+ patch("app.services.pipeline.normalize_url") as mock_norm,
+ patch("app.services.pipeline.unchain_url", new_callable=AsyncMock) as mock_unchain,
+ patch("app.services.pipeline.check_threat_db", new_callable=AsyncMock) as mock_threat,
+ patch(
+ "app.services.pipeline.check_domain_heuristic", new_callable=AsyncMock
+ ) as mock_heuristic,
+ patch("app.services.pipeline.analyze_content", new_callable=AsyncMock) as mock_content,
+ ):
+ mock_norm.return_value = NormalizeResult(
+ original_url="https://short.test/a",
+ normalized_url="https://short.test/a",
+ )
+ mock_unchain.return_value = _make_unchain(
+ final_url,
+ signals=["cross_origin:short.test->redirected.example.com"],
+ )
+ mock_threat.return_value = _make_threat(final_url)
+ mock_heuristic.return_value = _heuristic_with_score(20)
+ mock_content.return_value = _make_content(final_url)
+
+ result = await run_pipeline("aid-redirect-score", "https://short.test/a", async_session)
+
+ assert isinstance(result, PipelineSuccess)
+ assert DomainHeuristicSignal.REDIRECT_CROSS_ORIGIN in result.stages.domain_heuristic.signals
+ assert result.stages.domain_heuristic.score > 20
+ args, kwargs = mock_content.await_args
+ assert args == (final_url,)
+ assert "REDIRECT_CROSS_ORIGIN" in await _resolve_upstream(kwargs["upstream_signals"])
+
+
@pytest.mark.asyncio
async def test_run_pipeline_runs_threat_and_heuristic_in_parallel(
async_session: AsyncSession,
@@ -595,10 +730,10 @@ async def test_run_pipeline_short_circuits_on_urlhaus_match(
@pytest.mark.asyncio
-async def test_run_pipeline_skips_content_when_heuristic_alone_exceeds_threshold(
+async def test_run_pipeline_checks_content_when_heuristic_alone_exceeds_threshold(
async_session: AsyncSession,
) -> None:
- """위협 DB 미매치여도 휴리스틱만으로 danger 구간이면 건너뛴다."""
+ """위협 DB 미매치인 휴리스틱 danger 는 페이지 존재 확인 후 verdict 를 낸다."""
final_url = "https://typo-naverr.test/"
with (
@@ -614,12 +749,51 @@ async def test_run_pipeline_skips_content_when_heuristic_alone_exceeds_threshold
mock_unchain.return_value = _make_unchain(final_url)
mock_threat.return_value = _make_threat(final_url)
mock_heuristic.return_value = _heuristic_with_score(settings.score_danger_threshold)
+ mock_content.return_value = _make_content(final_url)
result = await run_pipeline("aid-heur", final_url, async_session)
assert isinstance(result, PipelineSuccess)
+ mock_content.assert_awaited_once()
+ assert result.verdict == Verdict.DANGER
+ assert result.stages.content_analysis.fetched is True
+
+
+@pytest.mark.asyncio
+async def test_run_pipeline_returns_verdict_when_page_unavailable_but_url_signal_is_strong(
+ async_session: AsyncSession,
+) -> None:
+ final_url = "http://xj3kq9vbnm2p7zla.com/login"
+ unchain = _make_unchain(final_url)
+ unchain.error = "dns_failure"
+ heuristic = DomainHeuristicResult(
+ domain="xj3kq9vbnm2p7zla.com",
+ score=settings.score_caution_threshold,
+ signals=[DomainHeuristicSignal.DGA_LIKE],
+ )
+
+ with (
+ patch("app.services.pipeline.normalize_url") as mock_norm,
+ patch("app.services.pipeline.unchain_url", new_callable=AsyncMock) as mock_unchain,
+ patch("app.services.pipeline.check_threat_db", new_callable=AsyncMock) as mock_threat,
+ patch(
+ "app.services.pipeline.check_domain_heuristic", new_callable=AsyncMock
+ ) as mock_heuristic,
+ patch("app.services.pipeline.analyze_content", new_callable=AsyncMock) as mock_content,
+ ):
+ mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url)
+ mock_unchain.return_value = unchain
+ mock_threat.return_value = _make_threat(final_url)
+ mock_heuristic.return_value = heuristic
+
+ result = await run_pipeline("aid-unavailable-url-signal", final_url, async_session)
+
+ assert isinstance(result, PipelineSuccess)
+ assert result.verdict == Verdict.CAUTION
+ assert result.score == settings.score_caution_threshold
+ assert result.stages.content_analysis.fetched is False
+ assert result.stages.content_analysis.error == "page_unavailable"
mock_content.assert_not_awaited()
- assert result.stages.content_analysis.error == "skipped_already_danger"
@pytest.mark.asyncio
diff --git a/tests/services/threat_db/test_check.py b/tests/services/threat_db/test_check.py
index 8bd58fd..22e2732 100644
--- a/tests/services/threat_db/test_check.py
+++ b/tests/services/threat_db/test_check.py
@@ -114,3 +114,38 @@ async def test_cancelled_error_propagates(async_session: AsyncSession) -> None:
pytest.raises(asyncio.CancelledError),
):
await check_threat_db(async_session, "https://x.test/")
+
+
+async def test_checks_original_and_final_url_candidates(async_session: AsyncSession) -> None:
+ clean = GSBResult(checked=True, is_threat=False)
+ hit = URLhausResult(
+ checked=True,
+ is_threat=True,
+ match_type="host",
+ matched_key="phish-origin.test",
+ threat="phishing",
+ )
+
+ with (
+ patch("app.services.threat_db.check.check_gsb", AsyncMock(return_value=clean)) as mock_gsb,
+ patch(
+ "app.services.threat_db.check.check_urlhaus",
+ AsyncMock(side_effect=[URLhausResult(checked=True, is_threat=False), hit]),
+ ) as mock_urlhaus,
+ ):
+ result = await check_threat_db(
+ async_session,
+ "https://benign-final.test/",
+ original_url="https://phish-origin.test/login",
+ )
+
+ assert result.is_malicious is True
+ assert result.urlhaus.matched_key == "phish-origin.test"
+ assert [call.args[0] for call in mock_gsb.await_args_list] == [
+ "https://benign-final.test/",
+ "https://phish-origin.test/login",
+ ]
+ assert [call.args[1] for call in mock_urlhaus.await_args_list] == [
+ "https://benign-final.test/",
+ "https://phish-origin.test/login",
+ ]
diff --git a/tests/services/threat_db/test_match_keys.py b/tests/services/threat_db/test_match_keys.py
index 0a0a785..61cd3d1 100644
--- a/tests/services/threat_db/test_match_keys.py
+++ b/tests/services/threat_db/test_match_keys.py
@@ -9,20 +9,33 @@ def test_host_only_for_standard_domain() -> None:
assert derive_keys("https://example.com/path/deep") == ["example.com"]
-def test_github_returns_host_path_and_host() -> None:
+def test_github_returns_host_path_only() -> None:
keys = derive_keys("https://github.com/alice/repo/blob/main/x.exe")
- assert keys == ["github.com/alice/repo", "github.com"]
+ assert keys == ["github.com/alice/repo"]
def test_raw_githubusercontent() -> None:
keys = derive_keys("https://raw.githubusercontent.com/alice/repo/main/x.sh")
- assert keys[0] == "raw.githubusercontent.com/alice/repo"
- assert keys[-1] == "raw.githubusercontent.com"
+ assert keys == ["raw.githubusercontent.com/alice/repo"]
-def test_github_short_path_fallbacks_to_host() -> None:
- # path segment 가 2개 미만이면 host 만.
- assert derive_keys("https://github.com/alice") == ["github.com"]
+def test_github_short_path_does_not_fallback_to_host() -> None:
+ # 다중 테넌트 호스트는 특정 사용자/리포 단위 이하로는 매칭하지 않는다.
+ assert derive_keys("https://github.com/alice") == []
+
+
+def test_dropbox_root_does_not_fallback_to_host() -> None:
+ assert derive_keys("https://www.dropbox.com/") == []
+
+
+def test_dropbox_shared_file_uses_path_prefix() -> None:
+ keys = derive_keys("https://www.dropbox.com/scl/fi/abc/report.exe?dl=0")
+ assert keys == ["www.dropbox.com/scl/fi"]
+
+
+def test_dropboxusercontent_download_host_uses_path_prefix() -> None:
+ keys = derive_keys("https://dl.dropboxusercontent.com/scl/fi/abc/report.exe")
+ assert keys == ["dl.dropboxusercontent.com/scl/fi"]
def test_empty_host_returns_empty_list() -> None:
diff --git a/tests/services/threat_db/test_urlhaus.py b/tests/services/threat_db/test_urlhaus.py
index 81f2ff5..c546055 100644
--- a/tests/services/threat_db/test_urlhaus.py
+++ b/tests/services/threat_db/test_urlhaus.py
@@ -41,6 +41,24 @@ async def test_exact_url_match(async_session: AsyncSession) -> None:
assert result.tags == ["exe", "emotet"]
+async def test_exact_url_match_tolerates_scheme_and_trailing_slash_variants(
+ async_session: AsyncSession,
+) -> None:
+ await _seed(
+ async_session,
+ id=11,
+ url="http://evil.test/login",
+ host="evil.test",
+ match_key="evil.test",
+ )
+
+ result = await check_urlhaus(async_session, "https://evil.test/login/")
+
+ assert result.is_threat is True
+ assert result.match_type == "url"
+ assert result.matched_key == "http://evil.test/login"
+
+
async def test_host_match(async_session: AsyncSession) -> None:
await _seed(
async_session,
@@ -69,24 +87,40 @@ async def test_host_path_match_github(async_session: AsyncSession) -> None:
assert result.matched_key == "github.com/bad/repo"
-async def test_host_path_preferred_over_host(async_session: AsyncSession) -> None:
- # 같은 호스트에 host 키와 host_path 키가 모두 있으면 host_path 우선.
+async def test_multitenant_host_does_not_match_host_only_entry(
+ async_session: AsyncSession,
+) -> None:
await _seed(
async_session,
- id=4,
- url="https://github.com/foo",
- host="github.com",
- match_key="github.com",
+ id=6,
+ url="https://www.dropbox.com/scl/fi/bad/payload.exe",
+ host="www.dropbox.com",
+ match_key="www.dropbox.com",
)
+
+ result = await check_urlhaus(async_session, "https://www.dropbox.com/")
+
+ assert result.checked is True
+ assert result.is_threat is False
+
+
+async def test_multitenant_host_path_match_dropbox(async_session: AsyncSession) -> None:
await _seed(
async_session,
- id=5,
- url="https://github.com/bad/repo/a",
- host="github.com",
- match_key="github.com/bad/repo",
+ id=7,
+ url="https://www.dropbox.com/scl/fi/bad/payload.exe",
+ host="www.dropbox.com",
+ match_key="www.dropbox.com/scl/fi",
+ )
+
+ result = await check_urlhaus(
+ async_session,
+ "https://www.dropbox.com/scl/fi/bad/readme.txt",
)
- result = await check_urlhaus(async_session, "https://github.com/bad/repo/anything")
+
+ assert result.is_threat is True
assert result.match_type == "host_path"
+ assert result.matched_key == "www.dropbox.com/scl/fi"
async def test_no_match(async_session: AsyncSession) -> None:
diff --git a/tests/services/threat_db/test_urlhaus_sync.py b/tests/services/threat_db/test_urlhaus_sync.py
index 8d99735..8daf807 100644
--- a/tests/services/threat_db/test_urlhaus_sync.py
+++ b/tests/services/threat_db/test_urlhaus_sync.py
@@ -16,6 +16,7 @@
# id, dateadded, url, url_status, last_online, threat, tags, urlhaus_link, reporter
1,2026-04-14 00:00:00,https://evil.test/a.exe,online,,malware_download,"exe,emotet",https://urlhaus.abuse.ch/url/1/,tester
2,2026-04-14 00:05:00,https://github.com/bad/repo/raw/main/x.sh,online,,malware_download,"sh",https://urlhaus.abuse.ch/url/2/,tester
+3,2026-04-14 00:10:00,https://www.dropbox.com/scl/fi/bad/payload.exe,online,,malware_download,"exe",https://urlhaus.abuse.ch/url/3/,tester
"""
@@ -55,18 +56,19 @@ async def test_sync_inserts_rows(sync_engine_patch) -> None:
stats = await sync_module.sync_urlhaus()
# 최초 실행 — 모두 insert, update 는 0 이어야 한다(C1 회귀 방지).
- assert stats["total"] == 2
- assert stats["inserted"] == 2
+ assert stats["total"] == 3
+ assert stats["inserted"] == 3
assert stats["updated"] == 0
assert stats["failed"] == 0
async with sync_engine_patch() as session:
rows = (await session.execute(select(URLhausEntry))).scalars().all()
- assert len(rows) == 2
+ assert len(rows) == 3
by_id = {r.id: r for r in rows}
assert by_id[1].host == "evil.test"
assert by_id[1].match_key == "evil.test"
assert by_id[2].match_key == "github.com/bad/repo"
+ assert by_id[3].match_key == "www.dropbox.com/scl/fi"
async def test_sync_idempotent(sync_engine_patch) -> None:
@@ -80,10 +82,10 @@ async def test_sync_idempotent(sync_engine_patch) -> None:
# 두 번째 실행은 모두 update 여야 한다 — insert/update 분류가 맞는지 검증.
async with sync_engine_patch() as session:
rows = (await session.execute(select(URLhausEntry))).scalars().all()
- assert len(rows) == 2
- assert stats2["total"] == 2
+ assert len(rows) == 3
+ assert stats2["total"] == 3
assert stats2["inserted"] == 0
- assert stats2["updated"] == 2
+ assert stats2["updated"] == 3
assert stats2["failed"] == 0
diff --git a/tests/services/unchainer/test_unchain.py b/tests/services/unchainer/test_unchain.py
index 9d46c8e..b3539e6 100644
--- a/tests/services/unchainer/test_unchain.py
+++ b/tests/services/unchainer/test_unchain.py
@@ -278,6 +278,20 @@ async def test_cross_origin_signal(self) -> None:
assert any(s.startswith("cross_origin:") for s in result.signals)
assert "cross_origin:safe.com->evil.com" in result.signals
+ @pytest.mark.asyncio
+ async def test_same_registered_domain_redirect_is_not_cross_origin(self) -> None:
+ responses = [
+ _make_response(301, {"location": "https://github.com/"}),
+ _make_response(200),
+ ]
+ client = _mock_client(responses)
+
+ with patch(_PATCH_TARGET, return_value=client):
+ result = await unchain_url("https://www.github.com/")
+
+ assert not any(s.startswith("cross_origin:") for s in result.signals)
+ assert result.final_url == "https://github.com/"
+
class TestRelativeLocation:
"""상대 경로 Location 해석."""
@@ -554,16 +568,20 @@ class TestSsrfProtection:
@pytest.mark.asyncio
async def test_loopback_blocked(self) -> None:
"""127.0.0.1 등 루프백 주소 차단."""
- with patch(
- "app.services.unchainer.unchain._check_host_safety",
- new_callable=AsyncMock,
- return_value="ssrf_blocked",
+ with (
+ patch(
+ "app.services.unchainer.unchain._check_host_safety",
+ new_callable=AsyncMock,
+ return_value="ssrf_blocked",
+ ),
+ patch("app.services.unchainer.unchain._build_client") as build_client,
):
result = await unchain_url("http://127.0.0.1/admin")
assert result.error == "ssrf_blocked"
assert "ssrf_blocked" in result.signals
assert result.hop_count == 0
+ build_client.assert_not_called()
@pytest.mark.asyncio
async def test_private_ip_blocked(self) -> None: