[FEATURE] 외부 DB 비의존 피싱 탐지 파이프라인 개발

Makefile

@@ -8,7 +8,8 @@ install:  ## Install runtime + dev dependencies
 	pre-commit install
 
 run:  ## Run the API locally with hot reload
-	uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
+	uvicorn app.main:app --reload --reload-dir app --reload-exclude .venv \
+		--host 0.0.0.0 --port 8000
 
 test:  ## Run the test suite
 	pytest

README.md

@@ -15,10 +15,10 @@
 | 로컬 캐시 DB | **SQLite + aiosqlite**                          | URLhaus 등 외부 피드 캐시 전용 |
 | ORM | **SQLAlchemy 2.0 (async)**                      | DeclarativeBase + naming convention |
 | Migration | **Alembic**                                     | SQLite batch mode |
-| HTTP Client | **httpx**                                       | 외부 API 호출 (GSB / RDAP / Claude / Spring 콜백) |
+| HTTP Client | **httpx**                                       | 외부 API 호출 (GSB / RDAP / OpenAI / Spring 콜백) |
 | Crawler | **BeautifulSoup4 + requests**                   | 페이지 본문 추출, 피싱 신호 탐지 |
 | Domain Lookup | **RDAP (httpx)**                                | 도메인 등록일·만료일·레지스트라 조회 |
-| 캐시 | **인메모리 dict + TTL + single-flight** / **SQLite 스냅샷** / **`functools.lru_cache`** | RDAP 24h 캐시·동시 요청 합치기 / URLhaus 로컬 캐시 / Settings 싱글톤 |
+| 캐시 | **인메모리 dict + TTL + single-flight** / **SQLite 스냅샷** / **`functools.lru_cache`** | RDAP 7일 캐시·동시 요청 합치기 / URLhaus 로컬 캐시 / Settings 싱글톤 |
 | Scheduler | **APScheduler**                                 | URLhaus 주기 동기화 |
 | Validation | **Pydantic v2 + pydantic-settings**             | 요청·응답·환경변수 |
 | Logging | **structlog**                                   | 구조적 로깅 + request_id 자동 바인딩 |
@@ -179,11 +179,13 @@ linclean-fastapi/
   기반으로 조립합니다.
 - **`services/`** — 4단계 파이프라인을 **단계별 하위 패키지**로 분리합니다.
   각 패키지의 `__init__.py` 가 해당 단계의 public 진입점을 re-export 하며,
-  오케스트레이터(`pipeline.py`)가 이를 순차 호출합니다. `Request` 같은 FastAPI
+  오케스트레이터(`pipeline.py`)가 이를 조립합니다. `Request` 같은 FastAPI
   객체를 받지 않고 `AsyncSession` / 순수 인자만 받습니다.
   - **`normalizer/`** — 1단계. `normalize_url()` 로 URL 을 canonical form 으로
-    정규화합니다 (스킴·호스트 소문자화, 기본 포트 제거, 퍼센트 인코딩 정돈,
-    경로 dot-segment 해소, IDN 디코딩, 프래그먼트 제거, 입력 검증).
+    정규화합니다 (앞뒤 공백 제거, 스킴·호스트 소문자화, 기본 포트 제거,
+    퍼센트 인코딩 정돈, 경로 dot-segment 해소, IDN 디코딩, 프래그먼트 제거,
+    입력 검증). 스킴이 없는 입력은 먼저 `https://` 로 분석 가능한 정상 HTML
+    응답인지 확인하고, 그렇지 않으면 `http://` 로 내려 분석합니다.
   - **`unchainer/`** — 1단계 후반. `unchain_url()` 로 리다이렉트 체인(3xx Location)
     을 끝까지 추적해 최종 URL 을 확정합니다. HEAD 우선 → GET 폴백 전략으로
     대역폭을 절약하면서 호환성을 확보하고, 네트워크 에러 시에도 GET 으로
@@ -200,8 +202,9 @@ linclean-fastapi/
   - **`content_analyzer/`** — 4단계. 최종 URL의 HTML만 fetch 하고, lxml 기반
     정적 추출 결과를 규칙 점수와 AI 보조 판정으로 합성합니다. 네트워크/AI 실패는
     degraded 결과로 흡수하되 `CancelledError` 는 상위로 전파합니다.
-  - **`pipeline.py`** — 1~4단계를 조립합니다. 2·3단계를 병렬 실행하고, 외부
-    위협 DB 매치나 danger 임계 도달 시 비용이 큰 4단계를 short-circuit 합니다.
+  - **`pipeline.py`** — 1~4단계를 조립합니다. 2·3단계를 병렬 실행하고,
+    외부 위협 DB 매치나 danger 임계 도달 시 4단계를 시작하지 않고
+    short-circuit 합니다. AI 판정은 선행 단계 신호가 준비된 뒤에만 수행됩니다.
   - **`analysis_callback.py`** — 비동기 `/analyze` 완료 후 Spring 내부 콜백
     엔드포인트로 결과를 POST 합니다. 2xx 외 응답/네트워크 오류는 최대 3회
     재시도하고, 최종 실패는 dead-letter 로그로 남깁니다.
@@ -307,20 +310,23 @@ upsert 합니다. 분석 시에는 외부 호출 없이 로컬 인덱스만 조
 규칙 기반 점수표로 도메인의 위험 신호를 합산합니다. 도메인 등록 정보는
 **RDAP (RFC 7480~7484)** 로 조회합니다.
 
-**2단계와 동시 실행 + 외부 DB 매치 시 조기 종료**: 두 단계 모두 1단계 최종 URL
-만 필요하고 서로 독립이라 `run_pipeline` 에서 두 task 를 동시에 띄우고
-`asyncio.wait(return_when=FIRST_COMPLETED)` 로 먼저 끝난 쪽을 확인합니다.
+**2단계·3단계 동시 실행 + 외부 DB 매치 시 조기 종료**: 2·3단계는 1단계
+최종 URL만 필요하고 서로 독립이라 `run_pipeline` 에서 동시에 띄웁니다.
+4단계는 threat DB/RDAP 신호가 확정되고 danger short-circuit 대상이 아닐 때만
+시작합니다.
 
 - **GSB 또는 URLhaus 매치 (`threat_db.is_malicious=True`)** 가 먼저 떨어지면,
-  아직 RDAP 대기 중일 수 있는 heuristic task 를 **즉시 `cancel()`** 하고 4단계도
-  `skipped_already_danger` 로 묶어 바로 반환합니다. verdict 가 이미 danger 로
-  확정이므로 RDAP 이 돌아올 때까지 대기할 이유가 없습니다. heuristic 자리에는
-  `rdap_error="skipped_threat_matched"` 인 placeholder 가 채워져 응답 스키마를
-  유지합니다.
+  아직 RDAP 대기 중일 수 있는 task 를 **즉시 `cancel()`** 하고, 4단계는
+  시작하지 않은 채 `skipped_already_danger` 로 묶어 바로 반환합니다. 알려진
+  악성 URL 은 score 100 / verdict danger 로 확정되므로 페이지 fetch 나 AI 호출을
+  수행하지 않습니다.
+  heuristic 자리에는 `skipped_reason="threat_matched"` 인 placeholder 가 채워져
+  응답 스키마를 유지합니다.
 - **heuristic 이 먼저 끝난 경우**는 threat_db 를 마저 기다린 뒤, is_malicious
   이거나 합산 점수가 임계를 넘으면 4단계만 skip 합니다.
-- **정상 경로**에서는 GSB(100~300ms)와 RDAP(캐시 미스 시 최대 ~5s)의 latency 가
-  겹쳐서 4단계 skip 판정까지의 wall clock 이 둘 중 느린 쪽으로 수렴합니다.
+- **정상 경로**에서는 GSB 와 RDAP(캐시 미스 시 기본 최대 3s)의 latency 가
+  겹칩니다. 이후 danger 임계 미만일 때만 콘텐츠 fetch/extract 와 AI 분석을
+  수행합니다.
 - `CancelledError` 와 stage 내부 예외는 남은 task 를 정리한 뒤 상위로 전파되어
   shutdown / 타임아웃 신호가 degraded 결과로 삼켜지지 않습니다.
 
@@ -338,7 +344,7 @@ upsert 합니다. 분석 시에는 외부 호출 없이 로컬 인덱스만 조
 | DGA 의심 도메인 | Shannon 엔트로피 ≥ 3.5 또는 자음 비율 ≥ 0.7 | +15 |
 | 합법 호스팅 플랫폼 (공유 호스팅 주의 가중치) | `user.github.io`, `app.netlify.app` | +15 |
 
-레벤슈타인 거리 함수는 외부 라이브러리에 의존하지 않고 직접 구현합니다 (DP). 약 500개 브랜드 화이트리스트(`brands.txt`)와 비교합니다. DGA 탐지는 Shannon 엔트로피와 자음 비율 통계만 사용하며 외부 모델이 필요 없습니다. RDAP 응답은 도메인 단위로 인메모리 캐싱(24h, `rdap_cache_ttl_seconds`)하여 동일 도메인 재조회 비용을 줄입니다. 캐시 만료·미스 순간에도 같은 도메인으로 몰리는 요청은 `_inflight` dict + `asyncio.Future` 로 합쳐(**single-flight / request coalescing**) RDAP 서버로 나가는 HTTP 호출을 1건으로 수렴시킵니다. RDAP 실패 시 신생 도메인 신호를 발동하지 않습니다 ("모름"을 "위험"으로 취급하지 않는 원칙).
+레벤슈타인 거리 함수는 외부 라이브러리에 의존하지 않고 직접 구현합니다 (DP). 약 500개 브랜드 화이트리스트(`brands.txt`)와 비교합니다. DGA 탐지는 Shannon 엔트로피와 자음 비율 통계만 사용하며 외부 모델이 필요 없습니다. RDAP 응답은 도메인 단위로 인메모리 캐싱(7일, `rdap_cache_ttl_seconds`)하여 동일 도메인 재조회 비용을 줄입니다. 캐시 만료·미스 순간에도 같은 도메인으로 몰리는 요청은 `_inflight` dict + `asyncio.Future` 로 합쳐(**single-flight / request coalescing**) RDAP 서버로 나가는 HTTP 호출을 1건으로 수렴시킵니다. RDAP 서버가 429 를 반환하면 `Retry-After` 또는 기본 쿨다운 동안 추가 RDAP 호출을 건너뛰고 `rdap_error="rate_limited"` 로 내려 호출량을 제한합니다. RDAP 실패 시 신생 도메인 신호를 발동하지 않습니다 ("모름"을 "위험"으로 취급하지 않는 원칙).
 
 `HOSTING_PLATFORM` 은 "이 도메인이 악성이다" 라는 신호가 아니라 **공유 호스팅 컨텍스트**(GitHub Pages·Netlify·Vercel·Heroku 등 다수 테넌트가 같은 상위 도메인을 공유)를 나타내는 주의 가중치입니다. URLhaus 매칭 키가 `host + path-prefix` 로 확장되는 것과 같은 맥락에서, 계정·리포 단위로 악성 여부가 갈리는 환경이므로 +15 를 가산합니다. 플랫폼 루트 도메인 자체(`netlify.app`, `vercel.app` 등)는 정상 운영 도메인이므로 타이포스쿼팅 검사에서 제외됩니다.
 
@@ -444,11 +450,15 @@ OPENAI_MODEL=gpt-4o-mini   # 기본 — 저비용·저지연
 | `AI_PROVIDER`               | `auto`          | `auto` (키 있으면 openai) / `openai` / `null` (비활성)               |
 | `OPENAI_API_KEY`            | *(없음)*        | 비워두면 `NullAIProvider` — 규칙 점수만 사용                         |
 | `OPENAI_MODEL`              | `gpt-4o-mini`   | OpenAI 채팅 모델 id                                                  |
-| `OPENAI_TIMEOUT_SECONDS`    | `10.0`          | 단일 호출 타임아웃                                                   |
-| `OPENAI_MAX_OUTPUT_TOKENS`  | `300`           | verdict + reason 용으로 여유 있는 상한                               |
+| `OPENAI_TIMEOUT_SECONDS`    | `5.0`           | 단일 호출 타임아웃                                                   |
+| `OPENAI_MAX_OUTPUT_TOKENS`  | `120`           | verdict + 100자 이내 reason 용 출력 상한                             |
 
 #### 응답에 실리는 AI 메타데이터
 
+`ai_reason` 은 보안 전문가의 근거 중심 문장으로 요청하되, 비전문가도 이해할 수
+있도록 쉬운 한국어 100자 이내로 제한합니다. 모델이 더 길게 응답해도 클라이언트에서
+100자로 잘라 응답합니다.
+
 `ContentAnalysisResult` 에는 verdict/reason 뿐 아니라 **실제로 응답한 모델 id 와
 토큰 사용량**이 함께 실립니다. 비용 관측, 모델 비교, 프롬프트 튜닝에 그대로
 쓸 수 있게 하기 위함입니다.
@@ -484,9 +494,9 @@ NullProvider 로 폴백된 경우에는 `ai_error="provider_misconfigured"` 로
 | 31 ~ 60 | **caution** (주의) | 노랑 — 이유 표시 후 사용자 판단 |
 | 61 이상 | **danger** (위험) | 빨강 — "피싱 의심, 열지 마세요" |
 
-**예외 — blacklist 매치는 점수와 무관하게 danger**: `threat_db.is_malicious=True`
-면 합산 점수가 임계 미만이어도 verdict 가 `danger` 로 강제됩니다. GSB / URLhaus
-매치 = 알려진 악성 URL 이라 점수 합산 결과보다 우선해서 결정합니다.
+**예외 — blacklist 매치는 score 100 / danger**: `threat_db.is_malicious=True`
+면 GSB / URLhaus 매치 = 알려진 악성 URL 로 보고 score 를 100으로 고정하며,
+verdict 는 `danger` 로 강제됩니다. 이 경우 4단계 fetch/AI 는 수행하지 않습니다.
 
 판정 근거는 `stages` 내부의 각 단계 원시 결과(`threat_db`, `domain_heuristic`,
 `content_analysis`)에 남습니다. 별도의 `reasons` 배열/자연어 `summary` 필드는
@@ -518,7 +528,9 @@ NullProvider 로 폴백된 경우에는 `ai_error="provider_misconfigured"` 로
 현재 코드는 단계별 단독 호출 라우터를 운영 라우터에 항상 마운트합니다. 모든
 엔드포인트는 `X-Internal-Api-Key` 인증을 요구하며, raw URL 을 받아
 `normalize_url()` 로 1차 검증한 뒤 해당 단계만 실행합니다. 전체 파이프라인을
-동기로 확인하려면 `/api/v1/analyze/sync` 를 사용합니다.
+동기로 확인하려면 `/api/v1/analyze/sync` 를 사용합니다. 외부 위협 DB(GSB/URLhaus)
+없이 URL·리다이렉트·RDAP·콘텐츠/AI만 확인하려면
+`/api/v1/analyze/db-independent/sync` 를 사용합니다.
 
 | Method | Path                            | 단계                | 호출 함수                       |
 |--------|---------------------------------|---------------------|---------------------------------|
@@ -527,6 +539,7 @@ NullProvider 로 폴백된 경우에는 `ai_error="provider_misconfigured"` 로
 | 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": "<raw url>" }` 형태이며, `/analyze/sync` 는
@@ -699,6 +712,7 @@ Spring                                FastAPI (본 엔진)
     "content_analysis": {
       "final_url": "https://login-secure-naver-auth.com/signin",
       "fetched": false,
+      "status_code": null,
       "score": 0,
       "signals": ["SKIPPED_ALREADY_DANGER"],
       "title": null,
@@ -710,6 +724,7 @@ Spring                                FastAPI (본 엔진)
       "is_spa_shell": false,
       "ai_verdict": null,
       "ai_reason": null,
+      "reason": "위험성이 확인된 URL입니다. 페이지를 열지 않는 것이 좋습니다.",
       "ai_error": null,
       "ai_model": null,
       "ai_token_usage": null,

app/api/v1/endpoints/analyze.py

@@ -20,8 +20,13 @@
 from app.api.deps import DBSession, InternalApiKey
 from app.db.session import SessionLocal
 from app.schemas.analyze import AnalyzeAccepted, AnalyzeRequest
+from app.schemas.db_independent_pipeline import (
+    DbIndependentPipelineFailure,
+    DbIndependentPipelineSuccess,
+)
 from app.schemas.pipeline import PipelineFailure, PipelineSuccess
 from app.services.analysis_callback import post_analysis_callback
+from app.services.db_independent_pipeline import run_db_independent_pipeline
 from app.services.pipeline import run_pipeline
 
 router = APIRouter()
@@ -93,3 +98,24 @@ async def analyze_sync(
         original_url=body.url,
         session=session,
     )
+
+
+@router.post(
+    "/analyze/db-independent/sync",
+    response_model=DbIndependentPipelineSuccess | DbIndependentPipelineFailure,
+    summary="DB 비의존 파이프라인 — 동기 결과 반환",
+    description=(
+        "GSB, URLhaus 등 외부 threat DB 조회 없이 URL 정규화, 리다이렉트 체인, "
+        "도메인 휴리스틱, 콘텐츠 정적 분석 결과만으로 verdict/score 를 산출합니다. "
+        "외부 DB 의존도를 제거한 실험·QA 용 경로입니다."
+    ),
+)
+async def analyze_db_independent_sync(
+    body: AnalyzeSyncRequest,
+    _: InternalApiKey,
+) -> DbIndependentPipelineSuccess | DbIndependentPipelineFailure:
+    analysis_id = str(uuid.uuid4())
+    return await run_db_independent_pipeline(
+        analysis_id=analysis_id,
+        original_url=body.url,
+    )

app/api/v1/endpoints/stages.py

@@ -98,7 +98,10 @@ async def stage_normalize(
             status_code=status.HTTP_400_BAD_REQUEST,
             detail=f"invalid url: {exc.message}",
         ) from exc
-    unchain_result = await unchain_url(normalize_result.normalized_url)
+    unchain_result = await unchain_url(
+        normalize_result.normalized_url,
+        prefer_https_when_schemeless=normalize_result.scheme_was_added,
+    )
     return StageNormalizeResponse(normalize=normalize_result, unchain=unchain_result)
 
 

app/core/config.py

@@ -83,7 +83,7 @@ def alembic_database_url(self) -> str:
 
     # RDAP
     rdap_bootstrap_url: str = "https://rdap.org/domain/"
-    rdap_timeout_seconds: float = 5.0
+    rdap_timeout_seconds: float = 3.0
     rdap_cache_ttl_seconds: int = 60 * 60 * 24 * 7  # 7d
     # TTL 동안 누적될 수 있는 도메인 엔트리 상한. 무작위 도메인 트래픽이 들어와도
     # 메모리가 무한 성장하지 않도록 LRU 로 끊는다. 일 100만 URL 기준 도메인 수 5만 이하 가정.
@@ -103,7 +103,8 @@ def alembic_database_url(self) -> str:
     unchain_max_hops: int = 5
     unchain_timeout_seconds: float = 5.0
     unchain_connect_timeout_seconds: float = 3.0
-    unchain_chain_timeout_seconds: float = 20.0
+    unchain_chain_timeout_seconds: float = 6.0
+    schemeless_https_probe_timeout_seconds: float = 1.0
     unchain_user_agent: str = (
         "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
         "AppleWebKit/537.36 (KHTML, like Gecko) "
@@ -144,7 +145,7 @@ def alembic_database_url(self) -> str:
     domain_label_length_threshold: int = 20
 
     # 페이지 콘텐츠 정적 분석 (4단계)
-    content_fetch_timeout_seconds: float = 8.0
+    content_fetch_timeout_seconds: float = 4.0
     content_fetch_connect_timeout_seconds: float = 3.0
     content_fetch_max_bytes: int = 2 * 1024 * 1024  # 2MiB 이상이면 끊고 분석
     # DNS rebinding 잔여 위험은 앱 레벨 사전 해석만으로 완전히 닫을 수 없다. 운영에서 분석 전용
@@ -209,8 +210,8 @@ def alembic_database_url(self) -> str:
     # OpenAI — 모델 교체는 OPENAI_MODEL 한 줄로 끝난다 (gpt-4o-mini / gpt-4o / gpt-4.1-mini).
     openai_api_key: str | None = None
     openai_model: str = "gpt-4o-mini"
-    openai_timeout_seconds: float = 10.0
-    openai_max_output_tokens: int = 300
+    openai_timeout_seconds: float = 5.0
+    openai_max_output_tokens: int = 120
 
     # Spring 통신
     internal_api_key: str

app/core/dns_cache.py

@@ -16,9 +16,9 @@
 
 import asyncio
 import socket
-from typing import Any, cast
+from typing import Any
 
-from cachetools import TTLCache  # type: ignore[import-untyped]
+from cachetools import TTLCache
 
 from app.core.config import settings
 
@@ -35,13 +35,13 @@
 async def resolve_host_addrs(hostname: str) -> tuple[AddrInfoTuple, ...]:
     """hostname 의 getaddrinfo 결과를 캐시. 실패 시 OSError 그대로 raise."""
     try:
-        return cast(tuple[AddrInfoTuple, ...], _cache[hostname])
+        return _cache[hostname]
     except KeyError:
         pass
 
     loop = asyncio.get_running_loop()
     infos = await loop.getaddrinfo(hostname, None, type=socket.SOCK_STREAM)
-    result = cast(tuple[AddrInfoTuple, ...], tuple(infos))
+    result = tuple(infos)
     _cache[hostname] = result
     return result
 

app/schemas/__init__.py

@@ -0,0 +1,13 @@
+from app.schemas.db_independent_pipeline import (
+    DbIndependentPipelineFailure,
+    DbIndependentPipelineResult,
+    DbIndependentPipelineStages,
+    DbIndependentPipelineSuccess,
+)
+
+__all__ = [
+    "DbIndependentPipelineFailure",
+    "DbIndependentPipelineResult",
+    "DbIndependentPipelineStages",
+    "DbIndependentPipelineSuccess",
+]