From 6deec352eb44b6a533427157c0d1248e3ced68bd Mon Sep 17 00:00:00 2001 From: minsoo0506 Date: Sun, 24 May 2026 17:06:52 +0900 Subject: [PATCH 1/6] =?UTF-8?q?[Fix]=20=EB=A6=AC=EB=B7=B0=20=EB=B0=98?= =?UTF-8?q?=EC=98=81=20=EB=B0=8F=20=EB=A0=8C=EB=8D=94=EB=A7=81=20=EC=9D=98?= =?UTF-8?q?=EC=A1=B4=EC=84=B1=20=EA=B0=9C=EC=84=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/services/content_analyzer/extract.py | 19 +++++++++++--- app/services/content_analyzer/render.py | 7 +++-- pyproject.toml | 4 ++- .../services/content_analyzer/test_extract.py | 26 +++++++++++++++++++ .../services/content_analyzer/test_render.py | 15 +++++++++++ 5 files changed, 65 insertions(+), 6 deletions(-) create mode 100644 tests/services/content_analyzer/test_render.py 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..72d23b5 100644 --- a/app/services/content_analyzer/render.py +++ b/app/services/content_analyzer/render.py @@ -34,12 +34,15 @@ class RenderResult: _render_semaphore: asyncio.Semaphore | None = None +_render_semaphore_loop: asyncio.AbstractEventLoop | None = None def _get_render_semaphore() -> asyncio.Semaphore: - global _render_semaphore - if _render_semaphore is None: + global _render_semaphore, _render_semaphore_loop + current_loop = asyncio.get_running_loop() + if _render_semaphore is None or _render_semaphore_loop is not current_loop: _render_semaphore = asyncio.Semaphore(settings.content_render_concurrency) + _render_semaphore_loop = current_loop return _render_semaphore diff --git a/pyproject.toml b/pyproject.toml index 9acba7b..24549f7 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", 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'' diff --git a/tests/services/content_analyzer/test_render.py b/tests/services/content_analyzer/test_render.py new file mode 100644 index 0000000..d4afd1e --- /dev/null +++ b/tests/services/content_analyzer/test_render.py @@ -0,0 +1,15 @@ +from __future__ import annotations + +import asyncio + +from app.services.content_analyzer import render + + +def test_render_semaphore_is_recreated_per_event_loop() -> None: + async def get_sem() -> asyncio.Semaphore: + return render._get_render_semaphore() + + first = asyncio.run(get_sem()) + second = asyncio.run(get_sem()) + + assert first is not second From 1dd73dc8b5017da015e177b2b571e068730833aa Mon Sep 17 00:00:00 2001 From: minsoo0506 Date: Mon, 25 May 2026 00:11:14 +0900 Subject: [PATCH 2/6] =?UTF-8?q?[Fix]=20URLhaus=20=EB=8B=A4=EC=A4=91=20?= =?UTF-8?q?=ED=85=8C=EB=84=8C=ED=8A=B8=20=EB=A7=A4=EC=B9=AD=20=EC=88=98?= =?UTF-8?q?=EC=A0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/core/config.py | 5 +++ app/services/threat_db/match_keys.py | 10 ++--- app/services/threat_db/urlhaus_sync.py | 2 +- tests/services/threat_db/test_match_keys.py | 27 +++++++++---- tests/services/threat_db/test_urlhaus.py | 38 +++++++++++++------ tests/services/threat_db/test_urlhaus_sync.py | 14 ++++--- 6 files changed, 66 insertions(+), 30 deletions(-) diff --git a/app/core/config.py b/app/core/config.py index 028f021..472ae3c 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, } ) 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_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/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..e1b2b2f 100644 --- a/tests/services/threat_db/test_urlhaus.py +++ b/tests/services/threat_db/test_urlhaus.py @@ -69,24 +69,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://github.com/bad/repo/anything") + + result = await check_urlhaus( + async_session, + "https://www.dropbox.com/scl/fi/bad/readme.txt", + ) + + 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 From 9c50b3d82fa6c4690c45fb1c2eb86b740f5e4a06 Mon Sep 17 00:00:00 2001 From: minsoo0506 Date: Fri, 29 May 2026 12:11:04 +0900 Subject: [PATCH 3/6] =?UTF-8?q?[Fix]=20=ED=94=BC=EC=8B=B1=20=ED=83=90?= =?UTF-8?q?=EC=A7=80=20=EC=8B=A0=ED=98=B8=20=EA=B0=9C=EC=84=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/core/config.py | 7 +- app/schemas/domain_heuristic.py | 1 + app/services/analysis_callback.py | 18 +- app/services/db_independent_pipeline.py | 223 +++++++++++++++--- app/services/domain_heuristic/check.py | 1 + app/services/pipeline.py | 211 ++++++++++++++++- app/services/threat_db/check.py | 79 +++++-- app/services/threat_db/urlhaus.py | 47 +++- .../services/content_analyzer/test_analyze.py | 1 + .../services/test_db_independent_pipeline.py | 92 +++++++- tests/services/test_pipeline.py | 186 ++++++++++++++- tests/services/threat_db/test_check.py | 35 +++ tests/services/threat_db/test_urlhaus.py | 18 ++ 13 files changed, 838 insertions(+), 81 deletions(-) diff --git a/app/core/config.py b/app/core/config.py index 472ae3c..878eca9 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -129,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 @@ -211,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/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/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/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/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/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/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/tests/services/content_analyzer/test_analyze.py b/tests/services/content_analyzer/test_analyze.py index 5601963..a5be551 100644 --- a/tests/services/content_analyzer/test_analyze.py +++ b/tests/services/content_analyzer/test_analyze.py @@ -225,6 +225,7 @@ 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_benign_verdict_no_score(self, monkeypatch: pytest.MonkeyPatch) -> None: class StubAI: 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_urlhaus.py b/tests/services/threat_db/test_urlhaus.py index e1b2b2f..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, From 47792efe212caee9c370f5527a3b49f6d76aefb4 Mon Sep 17 00:00:00 2001 From: minsoo0506 Date: Fri, 29 May 2026 12:44:44 +0900 Subject: [PATCH 4/6] =?UTF-8?q?[Fix]=20=ED=8E=98=EC=9D=B4=EC=A7=80=20?= =?UTF-8?q?=EB=AF=B8=EB=B0=9C=EA=B2=AC=20=EC=BD=9C=EB=B0=B1=20=EB=B3=B4?= =?UTF-8?q?=EA=B0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/schemas/db_independent_pipeline.py | 3 ++ app/schemas/pipeline.py | 3 ++ app/services/content_analyzer/render.py | 10 +++-- app/services/unchainer/unchain.py | 9 ++++- .../services/content_analyzer/test_render.py | 16 ++++---- tests/services/test_analysis_callback.py | 40 +++++++++++++++++++ tests/services/unchainer/test_unchain.py | 12 ++++-- 7 files changed, 75 insertions(+), 18 deletions(-) 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/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/content_analyzer/render.py b/app/services/content_analyzer/render.py index 72d23b5..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,15 +35,16 @@ class RenderResult: _render_semaphore: asyncio.Semaphore | None = None -_render_semaphore_loop: asyncio.AbstractEventLoop | None = None +_render_semaphore_loop_ref: weakref.ReferenceType[asyncio.AbstractEventLoop] | None = None def _get_render_semaphore() -> asyncio.Semaphore: - global _render_semaphore, _render_semaphore_loop + global _render_semaphore, _render_semaphore_loop_ref current_loop = asyncio.get_running_loop() - if _render_semaphore is None or _render_semaphore_loop is not current_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 = current_loop + _render_semaphore_loop_ref = weakref.ref(current_loop) return _render_semaphore diff --git a/app/services/unchainer/unchain.py b/app/services/unchainer/unchain.py index cf4c54b..63c8ddb 100644 --- a/app/services/unchainer/unchain.py +++ b/app/services/unchainer/unchain.py @@ -162,8 +162,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 +185,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, diff --git a/tests/services/content_analyzer/test_render.py b/tests/services/content_analyzer/test_render.py index d4afd1e..a0b438d 100644 --- a/tests/services/content_analyzer/test_render.py +++ b/tests/services/content_analyzer/test_render.py @@ -1,15 +1,15 @@ from __future__ import annotations -import asyncio - +import pytest from app.services.content_analyzer import render -def test_render_semaphore_is_recreated_per_event_loop() -> None: - async def get_sem() -> asyncio.Semaphore: - return render._get_render_semaphore() - - first = asyncio.run(get_sem()) - second = asyncio.run(get_sem()) +@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/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/unchainer/test_unchain.py b/tests/services/unchainer/test_unchain.py index 9d46c8e..6dc605c 100644 --- a/tests/services/unchainer/test_unchain.py +++ b/tests/services/unchainer/test_unchain.py @@ -554,16 +554,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: From c4d5bd15417dcf91c664962d3c38592bab97bea6 Mon Sep 17 00:00:00 2001 From: minsoo0506 Date: Fri, 29 May 2026 22:44:23 +0900 Subject: [PATCH 5/6] =?UTF-8?q?[Fix]=20=EC=A0=95=EC=83=81=20URL=20?= =?UTF-8?q?=EC=98=A4=ED=83=90=20=EB=B0=8F=20AI=20=EC=95=88=EB=82=B4=20?= =?UTF-8?q?=EB=AC=B8=EC=84=9C=20=EC=88=98=EC=A0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 963 +++++------------- app/services/content_analyzer/ai_openai.py | 9 +- app/services/content_analyzer/analyze.py | 47 +- app/services/domain_heuristic/brands.txt | 9 + app/services/domain_heuristic/dga.py | 4 + app/services/domain_heuristic/patterns.py | 19 + app/services/page_unavailability.py | 81 ++ app/services/unchainer/unchain.py | 16 +- pyproject.toml | 1 + .../content_analyzer/test_ai_openai.py | 3 + .../services/content_analyzer/test_analyze.py | 37 + tests/services/domain_heuristic/test_check.py | 27 + tests/services/unchainer/test_unchain.py | 14 + 13 files changed, 492 insertions(+), 738 deletions(-) create mode 100644 app/services/page_unavailability.py diff --git a/README.md b/README.md index 96201f9..d2751d8 100644 --- a/README.md +++ b/README.md @@ -1,787 +1,284 @@ -# LinClean-BE-fastapi (URL 보안 엔진) +# LinClean FastAPI -**LinClean의 URL 검역 백엔드입니다.**
-카톡·문자·메일로 받은 링크를 열기 전에 4단계 파이프라인으로 검사해 -**안전 / 주의 / 위험** 판정과 그 근거를 산출하는 **상태 비저장(stateless) 분석 전용 서버** 입니다. +## 핵심 목표 ---- +LinClean FastAPI는 Spring 서비스가 전달한 URL을 열기 전에 분석해 `safe`, `caution`, `danger` verdict와 근거를 반환하는 URL 보안 엔진입니다. + +주요 목표는 다음과 같습니다. + +- 단축 URL과 리다이렉트를 풀어 실제 도착지를 확인합니다. +- Google Safe Browsing, URLhaus, 도메인 휴리스틱, 콘텐츠 분석, AI 보조 판정을 한 파이프라인에서 합산합니다. +- 사라진 페이지나 400번대 페이지는 무리하게 verdict를 만들지 않고 실패 상태로 Spring에 콜백합니다. +- `caution` 이상 판정에서는 AI 분석 근거와 사용자 행동 가이드를 기존 `ai_reason` 안에 100자 이내로 담습니다. ## 기술 스택 -| 분류 | 기술 | 비고 | -|------|-------------------------------------------------|------| -| Framework | **FastAPI** | lifespan, async, app factory | -| Language | **Python 3.11+** | 타입 힌트, async/await | -| 로컬 캐시 DB | **SQLite + aiosqlite** | URLhaus 등 외부 피드 캐시 전용 | -| ORM | **SQLAlchemy 2.0 (async)** | DeclarativeBase + naming convention | -| Migration | **Alembic** | SQLite batch mode | -| HTTP Client | **httpx** | 외부 API 호출 (GSB / RDAP / OpenAI / Spring 콜백) | -| Crawler | **BeautifulSoup4 + requests** | 페이지 본문 추출, 피싱 신호 탐지 | -| Domain Lookup | **RDAP (httpx)** | 도메인 등록일·만료일·레지스트라 조회 | -| 캐시 | **인메모리 dict + TTL + single-flight** / **SQLite 스냅샷** / **`functools.lru_cache`** | RDAP 7일 캐시·동시 요청 합치기 / URLhaus 로컬 캐시 / Settings 싱글톤 | -| Scheduler | **APScheduler** | URLhaus 주기 동기화 | -| Validation | **Pydantic v2 + pydantic-settings** | 요청·응답·환경변수 | -| Logging | **structlog** | 구조적 로깅 + request_id 자동 바인딩 | -| Lint / Format | **Ruff** | E/F/I/B/SIM/S/UP 룰셋 | -| Type Check | **mypy (strict)** | pydantic 플러그인 | -| Test | **pytest + pytest-asyncio + httpx AsyncClient** | ASGI 테스트 | -| Packaging | **hatchling** | PEP 621 | -| AI | **OpenAI Chat Completions (gpt-4o-mini 기본)** | 페이지 콘텐츠 정적 분석 보조 — 모델은 `OPENAI_MODEL` 로 교체 | - ---- +| 구분 | 기술 | +|---|---| +| Language | Python 3.13 | +| API | FastAPI, Pydantic v2 | +| DB | SQLite, SQLAlchemy Async, Alembic | +| HTTP | httpx | +| Scheduler | APScheduler | +| HTML 분석 | BeautifulSoup4, lxml | +| 도메인 분석 | tldextract, RDAP | +| AI | OpenAI Chat Completions, NullAIProvider fallback | +| 테스트 | pytest, pytest-asyncio | +| 품질 도구 | Ruff, mypy | +| 선택 기능 | Playwright 렌더링 분석 | ## 디렉토리 구조 -``` +```text linclean-fastapi/ -├── app/ -│ ├── main.py # FastAPI 앱 팩토리 + lifespan -│ │ -│ ├── api/ # HTTP 계층 -│ │ ├── deps.py # 공용 의존성 (DBSession 등) -│ │ ├── error_handlers.py # 글로벌 예외 → ErrorResponse 변환 -│ │ └── v1/ -│ │ ├── router.py # v1 라우터 집합 -│ │ └── endpoints/ -│ │ ├── analyze.py # /analyze 비동기 접수, /analyze/sync 동기 실행 -│ │ ├── health.py # /health, /health/ready -│ │ └── stages.py # 단계별 운영/QA 엔드포인트 -│ │ -│ ├── core/ # 인프라/공통 -│ │ ├── config.py # 환경변수 (pydantic-settings) -│ │ ├── dns_cache.py # fetch / unchain 공용 DNS TTL 캐시 -│ │ ├── logging.py # structlog + stdlib bridge -│ │ ├── scheduler.py # APScheduler 기반 URLhaus 주기 동기화 -│ │ └── exceptions.py # AppError 도메인 예외 계층 -│ │ -│ ├── db/ # 영속성 계층 (외부 피드 캐시 전용) -│ │ ├── base.py # DeclarativeBase + naming convention -│ │ └── session.py # async engine, get_db, SQLite PRAGMA -│ │ -│ ├── middleware/ -│ │ └── request_context.py # X-Request-ID + 구조적 access log -│ │ -│ ├── models/ # SQLAlchemy ORM 모델 (외부 피드 캐시) -│ │ ├── __init__.py # Alembic autogen 용 import 모음 -│ │ └── urlhaus_entry.py # URLhaus 로컬 캐시 테이블 -│ │ -│ ├── schemas/ # Pydantic DTO -│ │ ├── common.py # HealthResponse, ErrorResponse 등 -│ │ ├── analyze.py # /analyze 요청/접수 응답 -│ │ ├── analysis.py # 하위 호환 re-export -│ │ ├── content_analysis.py # Stage 4 콘텐츠 분석 DTO -│ │ ├── domain_heuristic.py # Stage 3 도메인 휴리스틱 DTO -│ │ ├── normalize.py # Stage 1 정규화 DTO -│ │ ├── pipeline.py # PipelineSuccess / PipelineFailure / Verdict -│ │ ├── threat_db.py # Stage 2 GSB / URLhaus DTO -│ │ └── unchain.py # 리다이렉트 체인 DTO -│ │ -│ └── services/ # 도메인/비즈니스 로직 (파이프라인 단계별 모듈) -│ ├── pipeline.py # 1~4단계 오케스트레이터, 병렬화/short-circuit -│ ├── analysis_callback.py # Spring /internal/analysis-result 콜백 전송 -│ ├── normalizer/ # 1단계: URL 정규화(Canonicalization) -│ │ ├── __init__.py # 진입점 (normalize_url re-export) -│ │ └── normalize.py # 입력 검증, 스킴·호스트 정규화, 포트·프래그먼트 제거, 퍼센트 인코딩 정돈, 경로 정규화, IDN 디코딩 -│ ├── unchainer/ # 1단계 후반: URL 언체이닝(리다이렉트 추적) -│ │ ├── __init__.py # 진입점 (unchain_url re-export) -│ │ └── unchain.py # HEAD+GET 폴백, 체인 총 timeout, SSRF 방어, 의심 신호 수집 -│ ├── threat_db/ # 2단계: 외부 위협 DB 대조 -│ │ ├── __init__.py # 진입점 (check_threat_db re-export) -│ │ ├── check.py # GSB + URLhaus 병렬 조회·병합 -│ │ ├── gsb.py # Google Safe Browsing Lookup API -│ │ ├── urlhaus.py # 로컬 SQLite 조회 -│ │ ├── urlhaus_sync.py # CSV 다운로드 → SQLite upsert -│ │ └── match_keys.py # URLhaus 매칭 키 생성 (host / host+path) -│ ├── domain_heuristic/ # 3단계: 도메인 기반 휴리스틱 분석 -│ │ ├── __init__.py # 진입점 (check_domain_heuristic re-export) -│ │ ├── check.py # 패턴/DGA/타이포/RDAP 조합 및 점수 캡 -│ │ ├── patterns.py # IP 직접 접근, TLD, HTTPS, 하위도메인, 오픈 리다이렉트 -│ │ ├── dga.py # 엔트로피/자음 비율 기반 DGA 후보 탐지 -│ │ ├── typosquatting.py # brands.txt 기반 유사 브랜드 도메인 탐지 -│ │ ├── rdap.py # RDAP 조회, in-flight 병합, TTL/LRU 캐시 -│ │ └── brands.txt # 보호 브랜드 도메인 목록 -│ └── content_analyzer/ # 4단계: 페이지 콘텐츠 정적 분석 + AI 보조 판정 -│ ├── __init__.py # 진입점 (analyze_content re-export) -│ ├── analyze.py # fetch · extract · signals · AI 결과 병합 -│ ├── fetch.py # HTML fetch, content-type/size 컷, SSRF 방어 -│ ├── extract.py # BeautifulSoup+lxml 기반 title/input/meta/link/img 추출 -│ ├── signals.py # 브랜드 위장, meta refresh, 외부 링크 과다 등 규칙 점수 -│ ├── ai.py # AIProvider Protocol, NullAIProvider, 프롬프트 컨텍스트 -│ └── ai_openai.py # OpenAI Structured Outputs 기반 구현체 -│ -├── alembic/ # 외부 피드 캐시 스키마 마이그레이션 -│ ├── env.py # SQLite + batch mode 설정 -│ ├── script.py.mako -│ └── versions/ -│ -├── tests/ # pytest 테스트 -│ ├── conftest.py # 공용 픽스처 -│ ├── demo/ # 데모 스크립트 -│ │ ├── demo_normalize.py # URL 정규화 데모 -│ │ ├── demo_unchain.py # URL 언체이닝 데모 -│ │ ├── demo_threat_db.py # 외부 위협 DB 대조 데모 -│ │ ├── demo_domain_heuristic.py # 도메인 휴리스틱 데모 -│ │ └── demo_content_analysis.py # 콘텐츠 분석 데모 -│ ├── api/ -│ │ ├── test_analyze_callback.py # /analyze background callback 연결 테스트 -│ │ └── test_stages.py # 단계별 API 인증/응답 테스트 -│ └── services/ -│ ├── test_pipeline.py # 전체 파이프라인 오케스트레이션 테스트 -│ ├── test_analysis_callback.py # Spring 콜백 payload/retry 테스트 -│ ├── normalizer/ -│ │ └── test_normalize.py # URL 정규화 단위 테스트 -│ ├── unchainer/ -│ │ └── test_unchain.py # URL 언체이닝 단위 테스트 -│ ├── threat_db/ -│ │ ├── test_match_keys.py # URLhaus 매칭 키 단위 테스트 -│ │ ├── test_gsb.py # GSB Lookup 단위 테스트 -│ │ ├── test_urlhaus.py # URLhaus 조회 단위 테스트 -│ │ ├── test_urlhaus_sync.py # URLhaus 동기화 단위 테스트 -│ │ └── test_check.py # 병렬 조회·판정·폴백 단위 테스트 -│ ├── domain_heuristic/ -│ │ ├── test_check.py # 휴리스틱 통합 점수/신호 테스트 -│ │ ├── test_dga.py # DGA 후보 탐지 테스트 -│ │ ├── test_patterns.py # 도메인 패턴 신호 테스트 -│ │ ├── test_rdap.py # RDAP 파싱/캐시 테스트 -│ │ └── test_typosquatting.py # 브랜드 유사 도메인 테스트 -│ └── content_analyzer/ -│ ├── test_analyze.py # fetch/extract/signals/AI 통합 테스트 -│ ├── test_fetch.py # HTML fetch, SSRF, content-type/size 테스트 -│ ├── test_extract.py # HTML feature 추출 테스트 -│ ├── test_signals.py # 콘텐츠 규칙 점수 테스트 -│ ├── test_ai.py # AI provider protocol/null provider 테스트 -│ └── test_ai_openai.py # OpenAI provider structured output 테스트 -│ -├── data/ # SQLite 캐시 파일 (gitignore) -│ -├── alembic.ini -├── pyproject.toml # 의존성 + ruff/mypy/pytest 설정 -├── Makefile # install / run / test / migrate ... -├── .pre-commit-config.yaml -├── .env.example -└── README.md + app/ + api/ # FastAPI 라우터, 인증 의존성, 에러 핸들러 + core/ # 설정, 로깅, 스케줄러, DNS 캐시 + db/ # SQLAlchemy async engine/session + middleware/ # request_id, access log + models/ # URLhaus 캐시용 ORM 모델 + schemas/ # Pydantic 요청/응답 모델 + services/ + normalizer/ # URL 정규화 + unchainer/ # 리다이렉트 추적 + threat_db/ # GSB, URLhaus 조회 + domain_heuristic/ # 도메인/URL 휴리스틱 + content_analyzer/ # HTML fetch/extract/signals/AI + pipeline.py # DB 의존 전체 파이프라인 + db_independent_pipeline.py + analysis_callback.py # Spring 콜백 + page_unavailability.py + alembic/ # DB migration + data/ # SQLite 파일 위치 + reports/ # 날짜별 평가 결과, 커밋 제외 + scripts/ # 평가/운영 스크립트, 커밋 제외 + tests/ # 단위/통합 테스트 ``` -### 계층별 책임 - -- **`api/`** — HTTP 입출력만 담당. 라우터는 얇게 유지하고, 비즈니스 로직은 - `services/` 에 위임합니다. 의존성(`Depends`)은 `api/deps.py` 에 모아둡니다. -- **`core/`** — 프레임워크에 종속되지 않는 인프라 코드. 설정 로드, 로깅 구성, - 도메인 예외(`AppError`) 등 어디서든 import 해도 안전한 모듈만 둡니다. -- **`db/`** — SQLAlchemy 엔진/세션과 `Base`. **여기서 다루는 것은 외부 위협 - 피드 캐시뿐입니다.** 비즈니스 엔티티(User, Link, Directory 등)는 만들지 - 마세요. SQLite 전용 PRAGMA(`WAL`, `foreign_keys=ON`, `synchronous=NORMAL`) - 가 연결마다 자동 적용됩니다. -- **`models/`** — ORM 모델. 새 모델을 추가하면 반드시 - `app/models/__init__.py` 에서 import 해야 Alembic autogenerate 가 인식합니다. -- **`schemas/`** — 요청·응답 Pydantic 모델. Spring 콜백 본문은 - `services/analysis_callback.py` 가 `PipelineSuccess` / `PipelineFailure` 결과를 - 기반으로 조립합니다. -- **`services/`** — 4단계 파이프라인을 **단계별 하위 패키지**로 분리합니다. - 각 패키지의 `__init__.py` 가 해당 단계의 public 진입점을 re-export 하며, - 오케스트레이터(`pipeline.py`)가 이를 조립합니다. `Request` 같은 FastAPI - 객체를 받지 않고 `AsyncSession` / 순수 인자만 받습니다. - - **`normalizer/`** — 1단계. `normalize_url()` 로 URL 을 canonical form 으로 - 정규화합니다 (앞뒤 공백 제거, 스킴·호스트 소문자화, 기본 포트 제거, - 퍼센트 인코딩 정돈, 경로 dot-segment 해소, IDN 디코딩, 프래그먼트 제거, - 입력 검증). 스킴이 없는 입력은 먼저 `https://` 로 분석 가능한 정상 HTML - 응답인지 확인하고, 그렇지 않으면 `http://` 로 내려 분석합니다. - - **`unchainer/`** — 1단계 후반. `unchain_url()` 로 리다이렉트 체인(3xx Location) - 을 끝까지 추적해 최종 URL 을 확정합니다. HEAD 우선 → GET 폴백 전략으로 - 대역폭을 절약하면서 호환성을 확보하고, 네트워크 에러 시에도 GET 으로 - 재시도합니다. 체인 전체에 총 timeout 을 적용해 악의적 서버 방어가 가능하며, - `javascript:` / `data:` 같은 비허용 스킴 리다이렉트를 차단합니다. - 스킴 다운그레이드·크로스 오리진 등의 의심 신호도 수집합니다. - - **`threat_db/`** — 2단계. GSB 실시간 조회와 URLhaus 로컬 SQLite 조회를 - 병렬로 수행하고 결과를 `ThreatDbResult` 로 병합합니다. URLhaus 동기화는 - CSV 다운로드 후 chunk 단위 upsert 로 부분 진행을 보존합니다. - - **`domain_heuristic/`** — 3단계. 등록 가능 도메인을 기준으로 RDAP 등록일, - 오타 도메인, suspicious TLD, DGA 후보, 오픈 리다이렉트 파라미터, 하위도메인 - 과다 사용 등을 점수화합니다. RDAP 조회는 TTL/LRU 캐시와 in-flight 병합으로 - 외부 호출 수를 제한합니다. - - **`content_analyzer/`** — 4단계. 최종 URL의 HTML만 fetch 하고, lxml 기반 - 정적 추출 결과를 규칙 점수와 AI 보조 판정으로 합성합니다. 네트워크/AI 실패는 - degraded 결과로 흡수하되 `CancelledError` 는 상위로 전파합니다. - - **`pipeline.py`** — 1~4단계를 조립합니다. 2·3단계를 병렬 실행하고, - 외부 위협 DB 매치나 danger 임계 도달 시 4단계를 시작하지 않고 - short-circuit 합니다. AI 판정은 선행 단계 신호가 준비된 뒤에만 수행됩니다. - - **`analysis_callback.py`** — 비동기 `/analyze` 완료 후 Spring 내부 콜백 - 엔드포인트로 결과를 POST 합니다. 2xx 외 응답/네트워크 오류는 최대 3회 - 재시도하고, 최종 실패는 dead-letter 로그로 남깁니다. -- **`middleware/`** — `RequestContextMiddleware` 가 매 요청마다 `X-Request-ID` - 를 생성/전파하고 structlog contextvars 에 바인딩합니다. 응답 헤더로도 echo - 되며 모든 로그 라인에 자동으로 따라붙습니다. -- **`alembic/`** — SQLite 의 제한적 ALTER 지원을 보완하기 위해 - `render_as_batch=True` 로 동작합니다. - ---- - -## 핵심 — URL 안전성 분석 4단계 파이프라인 - -``` -URL 입력 (Spring 으로부터 위임) - │ - ▼ -1단계: URL 정규화 + 단축 URL 언체이닝 ← 스킴·호스트 소문자, 기본 포트/프래그먼트 - │ 제거, 리다이렉트 체인 추적해 최종 URL 확정 - ▼ -2단계: 외부 위협 DB 대조 ← GSB(API) + URLhaus(로컬 SQLite) - │ (최종 URL 기준 대조 — 블랙리스트 매치 시 즉시 short-circuit 가능) - │ - ▼ -3단계: 도메인 휴리스틱 분석 ← RDAP(등록일·레지스트라), 오타 도메인, 패턴 - │ - ▼ -4단계: 페이지 콘텐츠 정적 분석 ← BeautifulSoup + AI API (피싱 신호 추론) - │ - ▼ -종합 위험 점수 산출 → 안전 / 주의 / 위험 판정 - │ - ▼ -Spring `/internal/analysis-result` 콜백 POST +## 파이프라인 구조 + +```text +입력 URL + -> 1. URL 정규화 + -> 2. 리다이렉트 언체이닝 + -> 3. 외부 위협 DB 조회 + -> 4. 도메인/URL 휴리스틱 + -> 5. 콘텐츠 분석 + -> 6. AI 보조 판정 + -> 점수 합산 및 verdict 산출 + -> 동기 응답 또는 Spring 콜백 ``` -### 1단계 — URL 정규화 + 단축 URL 언체이닝 - -외부 DB 대조와 도메인 분석이 의미를 가지려면 **어떤 URL 을 검사할지부터 -확정** 해야 합니다. `bit.ly` 같은 단축 URL 상태로 GSB / URLhaus 를 조회하면 -거의 항상 매치되지 않기 때문에, 모든 후속 단계의 입력이 되는 "최종 URL" -을 먼저 만듭니다. - -- **정규화**: 스킴·호스트 소문자화, 기본 포트(`:80` / `:443`) 제거, - 프래그먼트(`#...`) 제거, 퍼센트 인코딩 정돈, 추적 파라미터(`utm_*` 등) - 정책적 제거, IDN(퓨니코드) → 유니코드 정규화 -- **언체이닝**: `HEAD` 우선 → `GET` 폴백 전략으로 리다이렉트 체인(3xx Location) - 을 끝까지 따라가 **최종 분석 대상 URL** 을 확정합니다. - - 네트워크 에러(타임아웃·연결 실패 등) 발생 시에도 GET 으로 재시도 - - 체인 전체에 총 timeout(기본 30초) 적용 — 악의적 서버의 지연 공격 방어 - - `javascript:`, `data:` 등 비허용 스킴 리다이렉트 차단 - - 스킴 다운그레이드(HTTPS→HTTP), 크로스 오리진, 무한 루프, max hop 초과 감지 - - 각 hop 의 원본 Location 값(`raw_location`)과 절대경로 해석 결과를 모두 기록 -- 이후 2~4 단계는 모두 이 **최종 URL** 을 기준으로 동작합니다. - -### 2단계 — 외부 위협 DB 대조 - -1단계에서 확정된 최종 URL 을 두 개의 위협 피드와 병렬로 대조합니다. 자체 -휴리스틱보다 먼저 실행해 이미 알려진 악성 URL 이면 조기에 `danger` 로 -short-circuit 할 수 있습니다. - -| 소스 | 방식 | 응답 시간 | 탐지 대상 | -|------|------|-----------|-----------| -| **Google Safe Browsing** | Google API 실시간 조회 | 100~300ms | 피싱 / 멀웨어 / 소셜 엔지니어링 | -| **URLhaus (abuse.ch)** | CSV → **로컬 SQLite** 캐시 | 1~5ms | 멀웨어 배포 URL | - -URLhaus 데이터는 APScheduler 가 주기적으로 CSV 를 다운로드해 로컬 SQLite 에 -upsert 합니다. 분석 시에는 외부 호출 없이 로컬 인덱스만 조회합니다. 두 소스 -중 하나라도 매치되면 그 자체로 강한 위험 신호이며, 점수 가산과 함께 후속 -단계에 결과를 그대로 전달합니다. - -**구현 노트 (`services/threat_db/`):** - -- **외부 의존성 실패는 파이프라인을 죽이지 않습니다.** GSB / URLhaus / DB / - 스케줄러 어디서 실패해도 `check_threat_db()` 는 항상 `ThreatDbResult` 를 - 반환하며, 실패 사유는 `error` 필드에 문자열 코드로 기록됩니다. GSB 만 실패한 - 경우 URLhaus 결과 단독으로 `is_malicious` 를 판정하고, 둘 다 실패하면 - `sources_checked=0` 으로 반환해 상위 레이어가 보수적으로 처리할 수 있게 합니다. -- **URLhaus 매칭 키**: 기본적으로 host 한 개를 키로 쓰되, GitHub / GitLab / - Bitbucket / sites.google.com 같은 다중 테넌트 호스트는 - `host + path-prefix(N 세그먼트)` 키를 추가로 생성합니다. 계정·리포 단위에서 - 악성 여부가 갈리는 도메인을 host 전체로 블랙리스트화해 오탐하지 않도록 합니다. - 동기화·조회 모두 동일한 `derive_keys()` 를 사용해 키 일관성을 보장합니다. -- **스케줄러**: `AsyncIOScheduler(timezone=UTC)` 싱글톤이 `urlhaus_sync` 를 - `IntervalTrigger(seconds=urlhaus_refresh_interval_seconds)` 로 주기 실행합니다 - (`coalesce=True, max_instances=1, misfire_grace_time=interval`). 앱 부트 시 - `urlhaus_sync_on_startup=True` 이면 최초 1회 즉시 동기화를 백그라운드로 - 수행합니다. 테스트에서는 `settings.scheduler_enabled=False` 로 전역 비활성화. -- **청크 커밋 동기화**: `sync_urlhaus()` 는 CSV 수만 행을 단일 트랜잭션으로 - 감싸지 않고 `CHUNK_SIZE=500` 단위로 나눠 커밋합니다. 중간 청크에서 DB 오류가 - 나도 직전 청크까지의 결과는 영속화되며, 실패 청크는 롤백되어 `failed` 로 - 누적됩니다. `stats = {inserted, updated, total, failed}` 는 실제 커밋된 - 행 수만 반영하므로 재시도 대상 판정에 그대로 쓸 수 있습니다. insert/update - 분류는 청크 직전에 `SELECT` 로 기존 id 를 조회해 정확히 분리합니다 - (SQLite `ON CONFLICT DO UPDATE` 는 cursor 로 두 경로 구분 불가). -- **CancelledError 전파**: `check_threat_db()` 내부 `asyncio.gather(..., - return_exceptions=True)` 는 일반 예외만 degraded 결과로 흡수하고, - `CancelledError` 는 그대로 re-raise 합니다. 상위 shutdown / 요청 타임아웃 - 신호를 삼키면 degraded 결과가 영속화될 위험이 있기 때문입니다. - -### 3단계 — 도메인 휴리스틱 분석 - -규칙 기반 점수표로 도메인의 위험 신호를 합산합니다. 도메인 등록 정보는 -**RDAP (RFC 7480~7484)** 로 조회합니다. - -**2단계·3단계 동시 실행 + 외부 DB 매치 시 조기 종료**: 2·3단계는 1단계 -최종 URL만 필요하고 서로 독립이라 `run_pipeline` 에서 동시에 띄웁니다. -4단계는 threat DB/RDAP 신호가 확정되고 danger short-circuit 대상이 아닐 때만 -시작합니다. - -- **GSB 또는 URLhaus 매치 (`threat_db.is_malicious=True`)** 가 먼저 떨어지면, - 아직 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 와 RDAP(캐시 미스 시 기본 최대 3s)의 latency 가 - 겹칩니다. 이후 danger 임계 미만일 때만 콘텐츠 fetch/extract 와 AI 분석을 - 수행합니다. -- `CancelledError` 와 stage 내부 예외는 남은 task 를 정리한 뒤 상위로 전파되어 - shutdown / 타임아웃 신호가 degraded 결과로 삼켜지지 않습니다. - -| 검사 항목 | 위험 신호 예시 | 점수 | -|----------|----------------|------| -| IP 직접 접근 | `http://192.168.x.x/login` | +40 | -| 오타 도메인 (레벤슈타인 거리 1~2) | `naverr.com`, `naaver.com` | +40 | -| punycode / IDN 호모글리프 | `xn--naver-xxx.com` | +35 | -| HTTPS 미사용 | `http://` | +30 | -| 신규 도메인 (RDAP 등록 30일 미만) | `created_date` 기준 | +30 | -| 서브도메인 과다 중첩 | `signin.auth.naver.attacker.xyz` | +25 | -| 특수문자·하이픈 과다 | `login-secure-naver-auth.com` | +20 | -| 의심 TLD | `.zip`, `.mov`, `.xyz`, `.top` 등 | +20 | -| 오픈 리다이렉트 파라미터 | `?url=`, `?redirect=` | +20 | -| DGA 의심 도메인 | Shannon 엔트로피 ≥ 3.5 또는 자음 비율 ≥ 0.7 | +15 | -| 합법 호스팅 플랫폼 (공유 호스팅 주의 가중치) | `user.github.io`, `app.netlify.app` | +15 | - -레벤슈타인 거리 함수는 외부 라이브러리에 의존하지 않고 직접 구현합니다 (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` 등)는 정상 운영 도메인이므로 타이포스쿼팅 검사에서 제외됩니다. - -### 4단계 — 페이지 콘텐츠 정적 분석 - -`httpx + BeautifulSoup` 으로 실제 페이지를 크롤링해 HTML 구조를 추출하고, -**OpenAI Chat Completions API** 가 피싱 신호를 정적으로 추론합니다. AI 는 본 -엔진 안에서 이 정적 분석 단계에서만 사용됩니다. - -추출 / 점수화하는 신호: - -- **로그인 폼 + 브랜드 위장**: `` 가 있고 title 에는 - 유명 브랜드명이 있는데 도메인은 그 브랜드와 무관 (+50) -- **브랜드 로고 이미지 위장**: `...` 의 alt 텍스트와 도메인 불일치 (+30) -- **`meta refresh` 자동 리다이렉트** (+20) -- **외부 링크 비율 과다**: 80% 이상이 외부 도메인이면 (+15) -- **AI 정적 추론**: 추출된 텍스트·폼·메타데이터를 AI 에게 넘겨 "이 페이지가 - 특정 브랜드를 사칭하거나 자격 증명을 탈취하려 하는지" 여부와 근거 텍스트를 - 반환받아 점수에 반영 (phishing +40 / suspicious +20 / benign 0) -- **`SPA_SHELL`** (점수 0, 시그널만): 초기 HTML 이 React/Vue/Next/Nuxt/Svelte/Angular - 마운트 셸뿐이라 정적 추출로 폼·입력 판정이 결정적이지 않은 상태. `is_spa_shell=true` - 로 응답에 실리고, AI 프롬프트에도 힌트로 전달돼 모델이 "폼 없음" 으로 단정하지 - 않도록 한다. 정상 SPA 가 압도적으로 많아 점수 가산은 하지 않는다. -- **페이지 접근 실패** — 사유별 가산 분리: - - `timeout` / `connect_error` / `http_error_*` / `unexpected` (도달 자체 실패): **+10** - - `not_html` (PDF/이미지 등 정상 비-HTML), `too_large` (대용량 정상 페이지), - `unexpected_redirect` (unchainer 가 놓친 3xx — 파이프라인 정합성 이슈): **+0** (시그널만) - - `blocked_host` (사설/loopback IP 또는 클라우드 메타데이터 호스트): **+10** (SSRF 1선 차단) - -#### 브랜드 매칭 전략 — 영문은 단어 경계, 한국어는 substring - -`brands.txt` (약 600개) 의 라벨로 title/alt 텍스트를 매칭할 때, 영문 라벨은 -`\b