[REFACTOR] URL 분석 정확도 개선

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

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
 
 

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):

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
 
 

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
 
 

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 은 보수적으로만 사용한다."
 )
 

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 을 응답에 노출.

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,14 +342,22 @@ 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:
         return 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:

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