From 86cba838638671ffa9f155a079588134c879201c Mon Sep 17 00:00:00 2001 From: minsoo0506 Date: Thu, 21 May 2026 18:29:06 +0900 Subject: [PATCH 1/5] =?UTF-8?q?[Feat]=20DB=20=EB=B9=84=EC=9D=98=EC=A1=B4?= =?UTF-8?q?=20=EB=B6=84=EC=84=9D=20=ED=8C=8C=EC=9D=B4=ED=94=84=EB=9D=BC?= =?UTF-8?q?=EC=9D=B8=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/v1/endpoints/analyze.py | 26 +++ app/schemas/__init__.py | 13 ++ app/schemas/analysis.py | 10 + app/schemas/db_independent_pipeline.py | 43 ++++ app/services/db_independent_pipeline.py | 195 ++++++++++++++++++ tests/api/test_analyze_db_independent.py | 105 ++++++++++ .../services/test_db_independent_pipeline.py | 189 +++++++++++++++++ 7 files changed, 581 insertions(+) create mode 100644 app/schemas/db_independent_pipeline.py create mode 100644 app/services/db_independent_pipeline.py create mode 100644 tests/api/test_analyze_db_independent.py create mode 100644 tests/services/test_db_independent_pipeline.py diff --git a/app/api/v1/endpoints/analyze.py b/app/api/v1/endpoints/analyze.py index fe13632..2186c49 100644 --- a/app/api/v1/endpoints/analyze.py +++ b/app/api/v1/endpoints/analyze.py @@ -20,8 +20,13 @@ from app.api.deps import DBSession, InternalApiKey from app.db.session import SessionLocal from app.schemas.analyze import AnalyzeAccepted, AnalyzeRequest +from app.schemas.db_independent_pipeline import ( + DbIndependentPipelineFailure, + DbIndependentPipelineSuccess, +) from app.schemas.pipeline import PipelineFailure, PipelineSuccess from app.services.analysis_callback import post_analysis_callback +from app.services.db_independent_pipeline import run_db_independent_pipeline from app.services.pipeline import run_pipeline router = APIRouter() @@ -93,3 +98,24 @@ async def analyze_sync( original_url=body.url, session=session, ) + + +@router.post( + "/analyze/db-independent/sync", + response_model=DbIndependentPipelineSuccess | DbIndependentPipelineFailure, + summary="DB 비의존 파이프라인 — 동기 결과 반환", + description=( + "GSB, URLhaus 등 외부 threat DB 조회 없이 URL 정규화, 리다이렉트 체인, " + "도메인 휴리스틱, 콘텐츠 정적 분석 결과만으로 verdict/score 를 산출합니다. " + "외부 DB 의존도를 제거한 실험·QA 용 경로입니다." + ), +) +async def analyze_db_independent_sync( + body: AnalyzeSyncRequest, + _: InternalApiKey, +) -> DbIndependentPipelineSuccess | DbIndependentPipelineFailure: + analysis_id = str(uuid.uuid4()) + return await run_db_independent_pipeline( + analysis_id=analysis_id, + original_url=body.url, + ) diff --git a/app/schemas/__init__.py b/app/schemas/__init__.py index e69de29..3f490a0 100644 --- a/app/schemas/__init__.py +++ b/app/schemas/__init__.py @@ -0,0 +1,13 @@ +from app.schemas.db_independent_pipeline import ( + DbIndependentPipelineFailure, + DbIndependentPipelineResult, + DbIndependentPipelineStages, + DbIndependentPipelineSuccess, +) + +__all__ = [ + "DbIndependentPipelineFailure", + "DbIndependentPipelineResult", + "DbIndependentPipelineStages", + "DbIndependentPipelineSuccess", +] diff --git a/app/schemas/analysis.py b/app/schemas/analysis.py index 9a610d0..abf8554 100644 --- a/app/schemas/analysis.py +++ b/app/schemas/analysis.py @@ -8,6 +8,12 @@ FetchExtractResponse, FetchStatusView, ) +from app.schemas.db_independent_pipeline import ( + DbIndependentPipelineFailure, + DbIndependentPipelineResult, + DbIndependentPipelineStages, + DbIndependentPipelineSuccess, +) from app.schemas.domain_heuristic import DomainHeuristicResult, DomainHeuristicSignal, RdapInfo from app.schemas.normalize import NormalizeResult from app.schemas.pipeline import ( @@ -28,6 +34,10 @@ "AnalyzeRequest", "ContentAnalysisResult", "ContentSignal", + "DbIndependentPipelineFailure", + "DbIndependentPipelineResult", + "DbIndependentPipelineStages", + "DbIndependentPipelineSuccess", "DomainHeuristicResult", "DomainHeuristicSignal", "ExtractedFeaturesView", diff --git a/app/schemas/db_independent_pipeline.py b/app/schemas/db_independent_pipeline.py new file mode 100644 index 0000000..5e14792 --- /dev/null +++ b/app/schemas/db_independent_pipeline.py @@ -0,0 +1,43 @@ +from typing import Literal + +from pydantic import BaseModel, Field + +from app.schemas.content_analysis import ContentAnalysisResult +from app.schemas.domain_heuristic import DomainHeuristicResult +from app.schemas.normalize import NormalizeResult +from app.schemas.pipeline import PipelineStage, PipelineTimings, Verdict +from app.schemas.unchain import UnchainResult + + +class DbIndependentPipelineStages(BaseModel): + """외부 threat DB를 제외한 독립 파이프라인 단계 결과.""" + + normalize: NormalizeResult + unchain: UnchainResult + domain_heuristic: DomainHeuristicResult + content_analysis: ContentAnalysisResult + + +class DbIndependentPipelineSuccess(BaseModel): + """DB 비의존 파이프라인 성공 응답.""" + + status: Literal["success"] = "success" + analysis_id: str + original_url: str + final_url: str + verdict: Verdict + score: int = Field(ge=0, le=100) + timings: PipelineTimings | None = None + stages: DbIndependentPipelineStages + + +class DbIndependentPipelineFailure(BaseModel): + status: Literal["failed"] = "failed" + analysis_id: str + original_url: str + failed_at_stage: PipelineStage + error: str + timings: PipelineTimings | None = None + + +DbIndependentPipelineResult = DbIndependentPipelineSuccess | DbIndependentPipelineFailure diff --git a/app/services/db_independent_pipeline.py b/app/services/db_independent_pipeline.py new file mode 100644 index 0000000..019ce13 --- /dev/null +++ b/app/services/db_independent_pipeline.py @@ -0,0 +1,195 @@ +"""외부 threat DB 비의존 URL 분석 파이프라인.""" + +from __future__ import annotations + +import asyncio +import time +from contextlib import suppress + +import structlog + +from app.core.config import settings +from app.core.exceptions import NormalizationError +from app.schemas.content_analysis import ContentAnalysisResult +from app.schemas.db_independent_pipeline import ( + DbIndependentPipelineFailure, + DbIndependentPipelineStages, + DbIndependentPipelineSuccess, +) +from app.schemas.domain_heuristic import DomainHeuristicResult +from app.schemas.pipeline import ( + PipelineStage, + PipelineStageTimings, + PipelineTimings, + Verdict, +) +from app.schemas.unchain import UnchainResult +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.unchainer import unchain_url + +logger = structlog.get_logger(__name__) + + +def _elapsed_seconds(started: float) -> float: + return round(time.perf_counter() - started, 6) + + +def _set_stage_timing( + timings: PipelineStageTimings, + stage: PipelineStage, + started: float, +) -> None: + setattr(timings, stage.value, _elapsed_seconds(started)) + + +def _build_timings(started: float, stage_timings: PipelineStageTimings) -> PipelineTimings: + return PipelineTimings( + total_seconds=_elapsed_seconds(started), + stages=stage_timings, + ) + + +def _decide_verdict(score: int) -> Verdict: + if score >= settings.score_danger_threshold: + return Verdict.DANGER + if score >= settings.score_caution_threshold: + return Verdict.CAUTION + return Verdict.SAFE + + +def _total_score( + heuristic: DomainHeuristicResult, + content: ContentAnalysisResult, +) -> int: + return min(heuristic.score + content.score, settings.score_total_cap) + + +def _redirect_signal_code(raw_signal: str) -> str | None: + if raw_signal.startswith("cross_origin:"): + return "REDIRECT_CROSS_ORIGIN" + if raw_signal == "scheme_downgrade": + return "REDIRECT_SCHEME_DOWNGRADE" + if raw_signal == "redirect_loop": + return "REDIRECT_LOOP" + if raw_signal == "max_hops_reached": + return "REDIRECT_MAX_HOPS_REACHED" + if raw_signal.startswith("unsafe_scheme:"): + return "REDIRECT_UNSAFE_SCHEME" + if raw_signal == "ssrf_blocked": + return "REDIRECT_SSRF_BLOCKED" + return None + + +def _collect_db_independent_signals( + heuristic: DomainHeuristicResult, + unchain: UnchainResult, +) -> tuple[str, ...]: + codes: list[str] = [signal.value for signal in heuristic.signals] + for raw_signal in unchain.signals: + code = _redirect_signal_code(raw_signal) + if code is not None and code not in codes: + codes.append(code) + return tuple(codes) + + +async def _collect_db_independent_signals_after_heuristic( + heuristic_task: asyncio.Task[DomainHeuristicResult], + unchain: UnchainResult, +) -> tuple[str, ...]: + heuristic = await heuristic_task + return _collect_db_independent_signals(heuristic, unchain) + + +async def run_db_independent_pipeline( + analysis_id: str, + original_url: str, +) -> DbIndependentPipelineSuccess | DbIndependentPipelineFailure: + """GSB/URLhaus 조회 없이 URL·리다이렉트·도메인 신호로 판정한다.""" + log = logger.bind(analysis_id=analysis_id, pipeline="db_independent") + log.info("db_independent_pipeline.start", url=original_url) + total_started = time.perf_counter() + stage_timings = PipelineStageTimings() + + stage_started = time.perf_counter() + try: + normalize = normalize_url(original_url) + except NormalizationError as exc: + _set_stage_timing(stage_timings, PipelineStage.NORMALIZE, stage_started) + log.warning("db_independent_pipeline.failed", stage=PipelineStage.NORMALIZE, error=str(exc)) + return DbIndependentPipelineFailure( + analysis_id=analysis_id, + original_url=original_url, + failed_at_stage=PipelineStage.NORMALIZE, + error=str(exc), + timings=_build_timings(total_started, stage_timings), + ) + _set_stage_timing(stage_timings, PipelineStage.NORMALIZE, stage_started) + + stage_started = time.perf_counter() + unchain = await unchain_url( + normalize.normalized_url, + prefer_https_when_schemeless=normalize.scheme_was_added, + ) + _set_stage_timing(stage_timings, PipelineStage.UNCHAIN, stage_started) + + stage_started = time.perf_counter() + heuristic_task = asyncio.create_task(check_domain_heuristic(unchain.final_url)) + + async def _timed_heuristic() -> DomainHeuristicResult: + try: + return await heuristic_task + finally: + _set_stage_timing(stage_timings, PipelineStage.DOMAIN_HEURISTIC, stage_started) + + timed_heuristic_task = asyncio.create_task(_timed_heuristic()) + + content_started = time.perf_counter() + upstream_task = asyncio.create_task( + _collect_db_independent_signals_after_heuristic(heuristic_task, unchain) + ) + content_task = asyncio.create_task( + analyze_content( + unchain.final_url, + upstream_signals=upstream_task, + ) + ) + + heuristic = await timed_heuristic_task + + if heuristic.score >= settings.score_danger_threshold: + content_task.cancel() + with suppress(asyncio.CancelledError): + await content_task + stage_started = time.perf_counter() + content = skipped_already_danger(unchain.final_url) + _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, stage_started) + else: + try: + content = await content_task + finally: + _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, content_started) + + score = _total_score(heuristic, content) + verdict = _decide_verdict(score) + log.info( + "db_independent_pipeline.done", + final_url=unchain.final_url, + verdict=verdict.value, + score=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, + ), + ) diff --git a/tests/api/test_analyze_db_independent.py b/tests/api/test_analyze_db_independent.py new file mode 100644 index 0000000..15ebe16 --- /dev/null +++ b/tests/api/test_analyze_db_independent.py @@ -0,0 +1,105 @@ +"""DB 비의존 분석 엔드포인트 테스트.""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from unittest.mock import AsyncMock + +import httpx +import pytest_asyncio +from app.api.v1.endpoints import analyze as analyze_endpoint +from app.core.config import settings +from app.schemas.content_analysis import ContentAnalysisResult +from app.schemas.db_independent_pipeline import ( + DbIndependentPipelineStages, + DbIndependentPipelineSuccess, +) +from app.schemas.domain_heuristic import DomainHeuristicResult +from app.schemas.normalize import NormalizeResult +from app.schemas.pipeline import PipelineStageTimings, PipelineTimings, Verdict +from app.schemas.unchain import UnchainResult +from fastapi import FastAPI +from httpx import ASGITransport + + +@pytest_asyncio.fixture +async def client() -> AsyncIterator[httpx.AsyncClient]: + app = FastAPI() + app.include_router(analyze_endpoint.router, prefix=settings.api_v1_prefix) + transport = ASGITransport(app=app) + headers = {"X-Internal-Api-Key": settings.internal_api_key} + async with httpx.AsyncClient( + transport=transport, + base_url="http://test", + headers=headers, + ) as c: + yield c + + +def _success_result() -> DbIndependentPipelineSuccess: + final_url = "https://example.com/login" + return DbIndependentPipelineSuccess( + analysis_id="aid-api", + original_url=final_url, + final_url=final_url, + verdict=Verdict.SAFE, + score=0, + timings=PipelineTimings( + total_seconds=0.001, + stages=PipelineStageTimings( + normalize=0.001, + unchain=0.001, + domain_heuristic=0.001, + content_analysis=0.001, + ), + ), + stages=DbIndependentPipelineStages( + normalize=NormalizeResult(original_url=final_url, normalized_url=final_url), + unchain=UnchainResult(input_url=final_url, final_url=final_url), + domain_heuristic=DomainHeuristicResult( + domain="example.com", + score=0, + signals=[], + ), + content_analysis=ContentAnalysisResult( + final_url=final_url, + fetched=True, + score=0, + signals=[], + ), + ), + ) + + +async def test_analyze_db_independent_sync_returns_result( + client: httpx.AsyncClient, + monkeypatch, +) -> None: + mock_run = AsyncMock(return_value=_success_result()) + monkeypatch.setattr(analyze_endpoint, "run_db_independent_pipeline", mock_run) + + resp = await client.post( + f"{settings.api_v1_prefix}/analyze/db-independent/sync", + json={"url": "https://example.com/login"}, + ) + + assert resp.status_code == 200 + body = resp.json() + assert body["status"] == "success" + assert body["verdict"] == "safe" + assert "threat_db" not in body["stages"] + mock_run.assert_awaited_once() + assert mock_run.await_args.kwargs["original_url"] == "https://example.com/login" + + +async def test_analyze_db_independent_sync_requires_internal_key() -> None: + app = FastAPI() + app.include_router(analyze_endpoint.router, prefix=settings.api_v1_prefix) + transport = ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: + resp = await c.post( + f"{settings.api_v1_prefix}/analyze/db-independent/sync", + json={"url": "https://example.com/login"}, + ) + + assert resp.status_code == 401 diff --git a/tests/services/test_db_independent_pipeline.py b/tests/services/test_db_independent_pipeline.py new file mode 100644 index 0000000..be98d20 --- /dev/null +++ b/tests/services/test_db_independent_pipeline.py @@ -0,0 +1,189 @@ +"""외부 threat DB 비의존 파이프라인 회귀 테스트.""" + +from __future__ import annotations + +import asyncio +import inspect +import time +from unittest.mock import AsyncMock, patch + +import pytest +from app.schemas.content_analysis import ContentAnalysisResult +from app.schemas.db_independent_pipeline import ( + DbIndependentPipelineFailure, + DbIndependentPipelineSuccess, +) +from app.schemas.domain_heuristic import DomainHeuristicResult, DomainHeuristicSignal +from app.schemas.normalize import NormalizeResult +from app.schemas.pipeline import PipelineStage, Verdict +from app.schemas.unchain import UnchainResult +from app.services.db_independent_pipeline import run_db_independent_pipeline + + +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_heuristic(score: int) -> DomainHeuristicResult: + return DomainHeuristicResult( + domain="example.com", + score=score, + signals=[DomainHeuristicSignal.HOSTING_PLATFORM] if score else [], + rdap=None, + rdap_error=None, + ) + + +def _make_content(final_url: str, *, score: int = 0) -> ContentAnalysisResult: + return ContentAnalysisResult(final_url=final_url, fetched=True, score=score, signals=[]) + + +@pytest.mark.asyncio +async def test_db_independent_pipeline_never_calls_threat_db() -> None: + final_url = "https://example.com/login" + + 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, + patch("app.services.pipeline.check_threat_db", new_callable=AsyncMock) as mock_threat_db, + ): + 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(15) + mock_content.return_value = _make_content(final_url, score=20) + + result = await run_db_independent_pipeline("aid-db-free", final_url) + + assert isinstance(result, DbIndependentPipelineSuccess) + assert result.analysis_id == "aid-db-free" + assert result.final_url == final_url + assert result.score == 35 + assert result.verdict == Verdict.CAUTION + assert result.stages.domain_heuristic.score == 15 + assert result.stages.content_analysis.score == 20 + assert result.timings is not None + assert result.timings.stages.threat_db is None + mock_threat_db.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_db_independent_pipeline_passes_url_and_redirect_signals_to_content() -> None: + final_url = "https://redirected.example.com/login" + + 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="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_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") + + mock_content.assert_awaited_once() + args, kwargs = mock_content.await_args + assert args == (final_url,) + upstream = kwargs["upstream_signals"] + if inspect.isawaitable(upstream): + upstream = await upstream + assert upstream == ( + "HOSTING_PLATFORM", + "REDIRECT_CROSS_ORIGIN", + ) + assert "provider" not in kwargs + + +@pytest.mark.asyncio +async def test_db_independent_pipeline_runs_heuristic_and_content_concurrently() -> None: + final_url = "https://parallel.example.com/login" + content_started = asyncio.Event() + heuristic_started = asyncio.Event() + + async def _slow_heuristic(_: str) -> DomainHeuristicResult: + heuristic_started.set() + await content_started.wait() + await asyncio.sleep(0.01) + return _make_heuristic(15) + + async def _slow_content(_: str, **__: object) -> ContentAnalysisResult: + content_started.set() + await heuristic_started.wait() + await asyncio.sleep(0.01) + return _make_content(final_url, score=20) + + 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=AsyncMock(side_effect=_slow_heuristic), + ), + patch( + "app.services.db_independent_pipeline.analyze_content", + new=AsyncMock(side_effect=_slow_content), + ) as mock_content, + ): + mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url) + mock_unchain.return_value = _make_unchain(final_url) + + started = time.perf_counter() + result = await run_db_independent_pipeline("aid-parallel", final_url) + + assert isinstance(result, DbIndependentPipelineSuccess) + assert result.score == 35 + assert time.perf_counter() - started < 0.08 + mock_content.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_db_independent_pipeline_returns_failure_on_normalize_error() -> None: + from app.core.exceptions import NormalizationError + + 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, + ): + mock_norm.side_effect = NormalizationError("invalid") + + result = await run_db_independent_pipeline("aid-bad", "not a url") + + assert isinstance(result, DbIndependentPipelineFailure) + assert result.failed_at_stage == PipelineStage.NORMALIZE + assert result.timings is not None + assert result.timings.stages.normalize is not None + mock_unchain.assert_not_awaited() From b2e413c8b551e2dd6fe18e793b1c57e59e62add9 Mon Sep 17 00:00:00 2001 From: minsoo0506 Date: Thu, 21 May 2026 18:29:39 +0900 Subject: [PATCH 2/5] =?UTF-8?q?[Fix]=20=EB=B6=84=EC=84=9D=20=EC=9D=91?= =?UTF-8?q?=EB=8B=B5=20=EC=A7=80=EC=97=B0=EA=B3=BC=20URL=20=EC=8A=A4?= =?UTF-8?q?=ED=82=B4=20=EC=B2=98=EB=A6=AC=20=EC=88=98=EC=A0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/v1/endpoints/stages.py | 5 +- app/core/config.py | 11 +++-- app/schemas/normalize.py | 4 ++ app/services/normalizer/normalize.py | 11 +++-- app/services/pipeline.py | 52 ++++++++++++++++----- app/services/unchainer/unchain.py | 44 +++++++++++++++-- tests/services/normalizer/test_normalize.py | 17 +++++-- tests/services/test_pipeline.py | 39 +++++++++------- 8 files changed, 137 insertions(+), 46 deletions(-) diff --git a/app/api/v1/endpoints/stages.py b/app/api/v1/endpoints/stages.py index da504a3..b7f8a09 100644 --- a/app/api/v1/endpoints/stages.py +++ b/app/api/v1/endpoints/stages.py @@ -98,7 +98,10 @@ async def stage_normalize( status_code=status.HTTP_400_BAD_REQUEST, detail=f"invalid url: {exc.message}", ) from exc - unchain_result = await unchain_url(normalize_result.normalized_url) + unchain_result = await unchain_url( + normalize_result.normalized_url, + prefer_https_when_schemeless=normalize_result.scheme_was_added, + ) return StageNormalizeResponse(normalize=normalize_result, unchain=unchain_result) diff --git a/app/core/config.py b/app/core/config.py index 6ddc9ad..c5fd216 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -83,7 +83,7 @@ def alembic_database_url(self) -> str: # RDAP rdap_bootstrap_url: str = "https://rdap.org/domain/" - rdap_timeout_seconds: float = 5.0 + rdap_timeout_seconds: float = 3.0 rdap_cache_ttl_seconds: int = 60 * 60 * 24 * 7 # 7d # TTL 동안 누적될 수 있는 도메인 엔트리 상한. 무작위 도메인 트래픽이 들어와도 # 메모리가 무한 성장하지 않도록 LRU 로 끊는다. 일 100만 URL 기준 도메인 수 5만 이하 가정. @@ -103,7 +103,8 @@ def alembic_database_url(self) -> str: unchain_max_hops: int = 5 unchain_timeout_seconds: float = 5.0 unchain_connect_timeout_seconds: float = 3.0 - unchain_chain_timeout_seconds: float = 20.0 + unchain_chain_timeout_seconds: float = 6.0 + schemeless_https_probe_timeout_seconds: float = 1.0 unchain_user_agent: str = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " @@ -144,7 +145,7 @@ def alembic_database_url(self) -> str: domain_label_length_threshold: int = 20 # 페이지 콘텐츠 정적 분석 (4단계) - content_fetch_timeout_seconds: float = 8.0 + content_fetch_timeout_seconds: float = 4.0 content_fetch_connect_timeout_seconds: float = 3.0 content_fetch_max_bytes: int = 2 * 1024 * 1024 # 2MiB 이상이면 끊고 분석 # DNS rebinding 잔여 위험은 앱 레벨 사전 해석만으로 완전히 닫을 수 없다. 운영에서 분석 전용 @@ -209,8 +210,8 @@ def alembic_database_url(self) -> str: # OpenAI — 모델 교체는 OPENAI_MODEL 한 줄로 끝난다 (gpt-4o-mini / gpt-4o / gpt-4.1-mini). openai_api_key: str | None = None openai_model: str = "gpt-4o-mini" - openai_timeout_seconds: float = 10.0 - openai_max_output_tokens: int = 300 + openai_timeout_seconds: float = 5.0 + openai_max_output_tokens: int = 120 # Spring 통신 internal_api_key: str diff --git a/app/schemas/normalize.py b/app/schemas/normalize.py index b19b4e8..9f93b97 100644 --- a/app/schemas/normalize.py +++ b/app/schemas/normalize.py @@ -4,3 +4,7 @@ class NormalizeResult(BaseModel): original_url: str = Field(description="Original URL after trimming") normalized_url: str = Field(description="Canonicalized URL") + scheme_was_added: bool = Field( + default=False, + description="True when the request URL had no explicit scheme and normalizer added one", + ) diff --git a/app/services/normalizer/normalize.py b/app/services/normalizer/normalize.py index 3e2df69..a65bbd4 100644 --- a/app/services/normalizer/normalize.py +++ b/app/services/normalizer/normalize.py @@ -36,8 +36,9 @@ def normalize_url(raw_url: str) -> NormalizeResult: # 길이 체크 전에 스킴 확정해야 함 — 스킴 붙이면 바이트 늘어남. # 단순 "://" 체크는 "example.com/path://weird" 같은 케이스 놓침. - if not re.match(r"^[a-zA-Z][a-zA-Z0-9+\-.]*://", cleaned): - cleaned = "https://" + cleaned + scheme_was_added = not re.match(r"^[a-zA-Z][a-zA-Z0-9+\-.]*://", cleaned) + if scheme_was_added: + cleaned = "http://" + cleaned max_len = settings.normalizer_max_url_length if len(cleaned) > max_len: @@ -76,7 +77,11 @@ def normalize_url(raw_url: str) -> NormalizeResult: query = _normalize_pct_encoding(parsed.query) normalized = urlunparse((scheme, netloc, path, params, query, "")) - return NormalizeResult(original_url=original, normalized_url=normalized) + return NormalizeResult( + original_url=original, + normalized_url=normalized, + scheme_was_added=scheme_was_added, + ) def _normalize_idn(hostname: str) -> str: diff --git a/app/services/pipeline.py b/app/services/pipeline.py index 2aea8e8..561b6e5 100644 --- a/app/services/pipeline.py +++ b/app/services/pipeline.py @@ -3,8 +3,9 @@ from __future__ import annotations import asyncio +import inspect import time -from collections.abc import Awaitable +from collections.abc import Awaitable, Iterable from contextlib import suppress from typing import TYPE_CHECKING, TypeVar from urllib.parse import urlparse @@ -135,16 +136,20 @@ def _collect_upstream_signals( async def _stage_content_analysis( log: structlog.stdlib.BoundLogger, final_url: str, - upstream_signals: tuple[str, ...], + upstream_signals: Iterable[str] | Awaitable[Iterable[str]], ) -> ContentAnalysisResult: result = await analyze_content(final_url, upstream_signals=upstream_signals) + if inspect.isawaitable(upstream_signals): + upstream_for_log: list[str] | str = "deferred" + else: + upstream_for_log = list(upstream_signals) log.info( "pipeline.content_analysis.done", fetched=result.fetched, score=result.score, signals=[s.value for s in result.signals], ai_verdict=result.ai_verdict.value if result.ai_verdict else None, - upstream_signals=list(upstream_signals), + upstream_signals=upstream_for_log, ) return result @@ -269,6 +274,15 @@ async def _run_stage_2_and_3( raise +async def _collect_upstream_after_stage_2_and_3( + stage_task: asyncio.Task[tuple[ThreatDbResult, DomainHeuristicResult, bool]], +) -> tuple[str, ...]: + threat, heuristic, short_circuited = await stage_task + if short_circuited: + return () + return _collect_upstream_signals(threat, heuristic) + + async def run_pipeline( analysis_id: str, original_url: str, @@ -297,14 +311,27 @@ async def run_pipeline( unchain: UnchainResult = await _timed_async_stage( stage_timings, PipelineStage.UNCHAIN, - _stage_unchain(log, norm.normalized_url), + unchain_url(norm.normalized_url, prefer_https_when_schemeless=norm.scheme_was_added), ) # 2·3단계는 둘 다 unchain.final_url 만 필요하고 서로 독립이라 병렬로 돈다. # threat_db 가 먼저 malicious 로 끝나면 verdict 가 이미 danger 로 확정이므로 # heuristic 을 cancel 하고 4단계까지 skip — 여기서 조기 종료가 일어난다. - threat, heuristic, short_circuited = await _run_stage_2_and_3( - log, unchain.final_url, session, stage_timings + stage_2_3_task = asyncio.create_task( + _run_stage_2_and_3(log, unchain.final_url, session, stage_timings) ) + upstream_task = asyncio.create_task(_collect_upstream_after_stage_2_and_3(stage_2_3_task)) + content_task = asyncio.create_task( + _timed_async_stage( + stage_timings, + PipelineStage.CONTENT_ANALYSIS, + _stage_content_analysis( + log, + unchain.final_url, + upstream_task, + ), + ) + ) + threat, heuristic, short_circuited = await stage_2_3_task if short_circuited: log.info( @@ -313,6 +340,9 @@ async def run_pipeline( gsb_threat=threat.gsb.is_threat, urlhaus_threat=threat.urlhaus.is_threat, ) + content_task.cancel() + with suppress(asyncio.CancelledError): + await content_task stage_started = time.perf_counter() content = skipped_already_danger(unchain.final_url) _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, stage_started) @@ -326,16 +356,14 @@ async def run_pipeline( reason=("threat_db_match" if threat.is_malicious else "already_danger"), preceding_score=preceding, ) + content_task.cancel() + with suppress(asyncio.CancelledError): + await content_task 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_upstream_signals(threat, heuristic) - content = await _timed_async_stage( - stage_timings, - PipelineStage.CONTENT_ANALYSIS, - _stage_content_analysis(log, unchain.final_url, upstream), - ) + content = await content_task score = _total_score(threat, heuristic, content) verdict = _decide_verdict(score, threat) diff --git a/app/services/unchainer/unchain.py b/app/services/unchainer/unchain.py index 25287c5..425c67d 100644 --- a/app/services/unchainer/unchain.py +++ b/app/services/unchainer/unchain.py @@ -12,7 +12,7 @@ import asyncio import ipaddress -from urllib.parse import urljoin, urlparse +from urllib.parse import urljoin, urlparse, urlunparse import httpx @@ -90,11 +90,37 @@ async def _check_host_safety(url: str) -> str | None: return None -async def unchain_url(url: str) -> UnchainResult: +def _https_variant(url: str) -> str | None: + parsed = urlparse(url) + if parsed.scheme != "http" or not parsed.hostname: + return None + return urlunparse(parsed._replace(scheme="https")) + + +async def _https_responds(client: httpx.AsyncClient, url: str, headers: dict[str, str]) -> bool: + https_url = _https_variant(url) + if https_url is None: + return False + safety_error = await _check_host_safety(https_url) + if safety_error is not None: + return False + try: + req = client.build_request("HEAD", https_url, headers=headers) + resp = await asyncio.wait_for( + client.send(req, stream=True), + timeout=settings.schemeless_https_probe_timeout_seconds, + ) + await resp.aclose() + except (TimeoutError, httpx.HTTPError): + return False + return resp.status_code < 500 + + +async def unchain_url(url: str, *, prefer_https_when_schemeless: bool = False) -> UnchainResult: """리다이렉트 체인을 추적하고 최종 URL·hop 기록·의심 신호를 반환.""" try: return await asyncio.wait_for( - _unchain_url_inner(url), + _unchain_url_inner(url, prefer_https_when_schemeless=prefer_https_when_schemeless), timeout=settings.unchain_chain_timeout_seconds, ) except TimeoutError: @@ -109,7 +135,11 @@ async def unchain_url(url: str) -> UnchainResult: ) -async def _unchain_url_inner(url: str) -> UnchainResult: +async def _unchain_url_inner( + url: str, + *, + prefer_https_when_schemeless: bool = False, +) -> UnchainResult: """실제 체인 추적 로직. unchain_url에서 총 timeout으로 감싸서 호출.""" hops: list[HopRecord] = [] signals: list[str] = [] @@ -125,6 +155,12 @@ async def _unchain_url_inner(url: str) -> UnchainResult: } client = _get_client() + if prefer_https_when_schemeless and await _https_responds(client, current_url, headers): + https_url = _https_variant(current_url) + if https_url is not None: + current_url = https_url + signals.append("schemeless_https_upgrade") + for _ in range(settings.unchain_max_hops): if current_url in visited: signals.append("redirect_loop") diff --git a/tests/services/normalizer/test_normalize.py b/tests/services/normalizer/test_normalize.py index 9c886c2..fa8c2ca 100644 --- a/tests/services/normalizer/test_normalize.py +++ b/tests/services/normalizer/test_normalize.py @@ -37,10 +37,10 @@ def test_max_length_boundary_passes(self) -> None: assert result.normalized_url.startswith("https://example.com/") def test_max_length_checked_after_scheme_prepend(self) -> None: - padding = 1024 - len("https://") - len("example.com/") + padding = 1024 - len("http://") - len("example.com/") url_without_scheme = "example.com/" + "a" * padding result = normalize_url(url_without_scheme) - assert result.normalized_url.startswith("https://example.com/") + assert result.normalized_url.startswith("http://example.com/") class TestOriginalPreservation: @@ -63,13 +63,20 @@ def test_https_preserved(self) -> None: result = normalize_url("HTTPS://Example.com/path") assert result.normalized_url == "https://example.com/path" - def test_no_scheme_defaults_to_https(self) -> None: + def test_no_scheme_defaults_to_http_and_marks_added_scheme(self) -> None: result = normalize_url("example.com/path") - assert result.normalized_url == "https://example.com/path" + assert result.normalized_url == "http://example.com/path" + assert result.scheme_was_added is True + + def test_explicit_http_does_not_mark_added_scheme(self) -> None: + result = normalize_url("http://example.com/path") + assert result.normalized_url == "http://example.com/path" + assert result.scheme_was_added is False def test_scheme_like_string_in_path_gets_scheme_prepended(self) -> None: result = normalize_url("example.com/path://weird") - assert result.normalized_url.startswith("https://example.com/") + assert result.normalized_url.startswith("http://example.com/") + assert result.scheme_was_added is True def test_unsupported_scheme_raises(self) -> None: with pytest.raises(NormalizationError, match="지원하지 않는 스킴"): diff --git a/tests/services/test_pipeline.py b/tests/services/test_pipeline.py index b177117..d2cb34d 100644 --- a/tests/services/test_pipeline.py +++ b/tests/services/test_pipeline.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import inspect from typing import TYPE_CHECKING from unittest.mock import AsyncMock, patch @@ -48,6 +49,12 @@ def _make_content(final_url: str, *, score: int = 0) -> ContentAnalysisResult: return ContentAnalysisResult(final_url=final_url, fetched=True, score=score, signals=[]) +async def _resolve_upstream(value: object) -> object: + if inspect.isawaitable(value): + return await value + return value + + async def _run_with_scores( async_session: AsyncSession, *, @@ -119,7 +126,7 @@ async def test_run_pipeline_includes_domain_heuristic_stage(async_session: Async args, kwargs = mock_content.await_args assert args == (final_url,) # _make_heuristic 가 HOSTING_PLATFORM 시그널을 가지므로 그대로 전달돼야 한다 - assert kwargs["upstream_signals"] == ("HOSTING_PLATFORM",) + assert await _resolve_upstream(kwargs["upstream_signals"]) == ("HOSTING_PLATFORM",) @pytest.mark.asyncio @@ -186,7 +193,7 @@ async def test_run_pipeline_skips_content_when_gsb_malicious( 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, + patch("app.services.pipeline.analyze_content", new_callable=AsyncMock), ): mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url) mock_unchain.return_value = _make_unchain(final_url) @@ -197,8 +204,7 @@ async def test_run_pipeline_skips_content_when_gsb_malicious( result = await run_pipeline("aid-skip", final_url, async_session) assert isinstance(result, PipelineSuccess) - # 네트워크·AI 비용을 아끼기 위해 analyze_content 는 호출되지 않아야 한다 - mock_content.assert_not_awaited() + # 병렬 선시작 이후에도 최종 응답은 skip 결과로 고정돼야 한다. content = result.stages.content_analysis assert content.fetched is False assert content.error == "skipped_already_danger" @@ -231,7 +237,10 @@ async def test_run_pipeline_runs_content_when_below_danger( await run_pipeline("aid-run", final_url, async_session) # heuristic 시그널 없는 경우 upstream_signals 는 빈 튜플로 전달돼야 한다 - mock_content.assert_awaited_once_with(final_url, upstream_signals=()) + mock_content.assert_awaited_once() + args, kwargs = mock_content.await_args + assert args == (final_url,) + assert await _resolve_upstream(kwargs["upstream_signals"]) == () class TestVerdictAndScore: @@ -376,7 +385,7 @@ async def test_run_pipeline_passes_upstream_signals_to_content_analysis( mock_content.assert_awaited_once() args, kwargs = mock_content.await_args assert args == (final_url,) - assert kwargs["upstream_signals"] == ("TYPO_DOMAIN", "NEW_DOMAIN") + assert await _resolve_upstream(kwargs["upstream_signals"]) == ("TYPO_DOMAIN", "NEW_DOMAIN") @pytest.mark.asyncio @@ -479,7 +488,7 @@ async def slow_heuristic(_url: str) -> DomainHeuristicResult: 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", side_effect=slow_heuristic), - patch("app.services.pipeline.analyze_content", new_callable=AsyncMock) as mock_content, + patch("app.services.pipeline.analyze_content", new_callable=AsyncMock), ): mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url) mock_unchain.return_value = _make_unchain(final_url) @@ -490,8 +499,7 @@ async def slow_heuristic(_url: str) -> DomainHeuristicResult: assert isinstance(result, PipelineSuccess) # heuristic 은 cancel 되어 본문이 끝까지 돌지 않았어야 한다 assert heuristic_finished.is_set() is False - # 4단계 분석은 skip - mock_content.assert_not_awaited() + # 4단계 응답은 skip content = result.stages.content_analysis assert content.fetched is False assert content.error == "skipped_already_danger" @@ -530,7 +538,7 @@ async def fast_heuristic(_url: str) -> DomainHeuristicResult: patch("app.services.pipeline.unchain_url", new_callable=AsyncMock) as mock_unchain, patch("app.services.pipeline.check_threat_db", side_effect=slow_threat), patch("app.services.pipeline.check_domain_heuristic", side_effect=fast_heuristic), - patch("app.services.pipeline.analyze_content", new_callable=AsyncMock) as mock_content, + patch("app.services.pipeline.analyze_content", new_callable=AsyncMock), ): mock_norm.return_value = NormalizeResult( original_url=final_url, normalized_url=final_url @@ -547,7 +555,6 @@ async def fast_heuristic(_url: str) -> DomainHeuristicResult: # GSB(+50) + heuristic placeholder(0) + content skip(0) = 50, verdict 는 DANGER 강제. assert result.score == 50 assert result.verdict == Verdict.DANGER - mock_content.assert_not_awaited() @pytest.mark.asyncio @@ -572,7 +579,7 @@ async def test_run_pipeline_short_circuits_on_urlhaus_match( 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, + patch("app.services.pipeline.analyze_content", new_callable=AsyncMock), ): mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url) mock_unchain.return_value = _make_unchain(final_url) @@ -582,7 +589,6 @@ async def test_run_pipeline_short_circuits_on_urlhaus_match( result = await run_pipeline("aid-urlhaus", final_url, async_session) assert isinstance(result, PipelineSuccess) - mock_content.assert_not_awaited() assert result.stages.content_analysis.error == "skipped_already_danger" @@ -600,13 +606,14 @@ async def test_run_pipeline_skips_content_when_heuristic_alone_exceeds_threshold 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, + patch("app.services.pipeline.analyze_content", new_callable=AsyncMock), ): 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(settings.score_danger_threshold) - await run_pipeline("aid-heur", final_url, async_session) + result = await run_pipeline("aid-heur", final_url, async_session) - mock_content.assert_not_awaited() + assert isinstance(result, PipelineSuccess) + assert result.stages.content_analysis.error == "skipped_already_danger" From a0feec5c95b38db111d54b5dad86f828fea86023 Mon Sep 17 00:00:00 2001 From: minsoo0506 Date: Thu, 21 May 2026 18:30:13 +0900 Subject: [PATCH 3/5] =?UTF-8?q?[Fix]=20=EC=BD=98=ED=85=90=EC=B8=A0=20?= =?UTF-8?q?=EB=B6=84=EC=84=9D=20=EC=83=81=ED=83=9C=EC=BD=94=EB=93=9C?= =?UTF-8?q?=EC=99=80=20AI=20=EC=82=AC=EC=9C=A0=20=EC=9D=91=EB=8B=B5=20?= =?UTF-8?q?=EC=88=98=EC=A0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/schemas/content_analysis.py | 11 + app/services/content_analyzer/ai_openai.py | 77 +++--- app/services/content_analyzer/analyze.py | 41 ++- .../content_analyzer/test_ai_openai.py | 251 +++++++++--------- .../services/content_analyzer/test_analyze.py | 19 +- 5 files changed, 239 insertions(+), 160 deletions(-) diff --git a/app/schemas/content_analysis.py b/app/schemas/content_analysis.py index b7b9b9a..e44e407 100644 --- a/app/schemas/content_analysis.py +++ b/app/schemas/content_analysis.py @@ -36,6 +36,10 @@ class TokenUsage(BaseModel): class ContentAnalysisResult(BaseModel): final_url: str fetched: bool + status_code: int | None = Field( + default=None, + description="HTTP status code observed when fetching final_url for content analysis", + ) score: int = 0 signals: list[ContentSignal] = Field( default_factory=list, @@ -59,6 +63,13 @@ class ContentAnalysisResult(BaseModel): ai_verdict: AIVerdict | None = None ai_reason: str | None = None + reason: str | None = Field( + default=None, + description=( + "AI 여부와 무관한 사용자 표시용 설명. 예: http_error_404 는 " + "'페이지를 찾을 수 없습니다.' 로 내려간다." + ), + ) ai_error: str | None = Field( default=None, description=( diff --git a/app/services/content_analyzer/ai_openai.py b/app/services/content_analyzer/ai_openai.py index ef885f0..750fd7b 100644 --- a/app/services/content_analyzer/ai_openai.py +++ b/app/services/content_analyzer/ai_openai.py @@ -15,6 +15,8 @@ import json from typing import Any +import httpx + from app.core.config import settings from app.core.logging import get_logger from app.schemas.content_analysis import AIVerdict, TokenUsage @@ -22,18 +24,6 @@ logger = get_logger(__name__) -AsyncOpenAI: Any | None = None - - -def _get_async_openai_class() -> Any: - global AsyncOpenAI - if AsyncOpenAI is None: - from openai import AsyncOpenAI as _AsyncOpenAI - - AsyncOpenAI = _AsyncOpenAI - return AsyncOpenAI - - _VERDICT_SCHEMA: dict[str, Any] = { "name": "phishing_verdict", "strict": True, @@ -73,7 +63,9 @@ def _get_async_openai_class() -> Any: "'폼이 없다' 가 아니라 '정적 추출로는 판정 불가' 를 의미한다. 이 경우 남은 단서" "(title/URL/이미지 alt/upstream_signals 등)만으로 단정하지 말고 suspicious 또는 benign 을 " "보수적으로 선택한다. " - "verdict 는 phishing / suspicious / benign 중 하나. reason 은 한국어 1~2문장. " + "verdict 는 phishing / suspicious / benign 중 하나. reason 은 보안 전문가처럼 " + "근거 중심으로 쓰되, IT와 보안을 모르는 사람도 이해할 수 있는 쉬운 한국어 100자 이내 " + "1문장으로 작성한다. " "확증이 없으면 benign 또는 suspicious 를 쓰고 phishing 은 보수적으로만 사용한다." ) @@ -98,13 +90,21 @@ def _build_user_prompt(ctx: AIPromptContext) -> str: def _extract_token_usage(completion: object) -> TokenUsage | None: - # completion.usage 는 OpenAI SDK 가 채워주지만 스트리밍/에러 응답 등에선 비어있을 수 있다. - usage = getattr(completion, "usage", None) + # HTTP 응답 dict 또는 SDK 호환 객체를 모두 받는다. 스트리밍/에러 응답 등에선 비어있을 수 있다. + if isinstance(completion, dict): + usage = completion.get("usage") + else: + usage = getattr(completion, "usage", None) if usage is None: return None - prompt = getattr(usage, "prompt_tokens", None) - completion_tokens = getattr(usage, "completion_tokens", None) - total = getattr(usage, "total_tokens", None) + if isinstance(usage, dict): + prompt = usage.get("prompt_tokens") + completion_tokens = usage.get("completion_tokens") + total = usage.get("total_tokens") + else: + prompt = getattr(usage, "prompt_tokens", None) + completion_tokens = getattr(usage, "completion_tokens", None) + total = getattr(usage, "total_tokens", None) if prompt is None or completion_tokens is None or total is None: return None return TokenUsage( @@ -132,6 +132,9 @@ def _parse_verdict( reason = data.get("reason") if not isinstance(verdict_str, str) or not isinstance(reason, str): return None + reason = reason.strip() + if len(reason) > 100: + reason = reason[:100] try: verdict = AIVerdict(verdict_str) @@ -175,7 +178,7 @@ def __init__( else settings.openai_max_output_tokens ) # 클라이언트는 호출 시점에 지연 생성 — 키가 없으면 아예 만들지 않는다. - self._client: Any | None = None + self._client: httpx.AsyncClient | None = None # aclose() 후 재사용 방지. 더블 콜은 no-op, 이후 infer() 는 None 반환. self._closed: bool = False @@ -183,7 +186,7 @@ def __init__( def model(self) -> str: return self._model - def _get_client(self) -> Any | None: + def _get_client(self) -> httpx.AsyncClient | None: if self._closed: # 정상 시나리오에선 lifespan 종료 후 호출이 없어야 한다. 호출이 들리면 # provider 재바인딩이 누락된 신호이므로 silent 폴백 대신 한 번 경고로 남긴다. @@ -192,8 +195,15 @@ def _get_client(self) -> Any | None: if not settings.openai_api_key: return None if self._client is None: - client_class = _get_async_openai_class() - self._client = client_class(api_key=settings.openai_api_key) + self._client = httpx.AsyncClient( + base_url="https://api.openai.com/v1", + timeout=self._timeout, + headers={ + "Authorization": f"Bearer {settings.openai_api_key}", + "Content-Type": "application/json", + }, + trust_env=False, + ) return self._client async def aclose(self) -> None: @@ -201,7 +211,7 @@ async def aclose(self) -> None: return self._closed = True if self._client is not None: - await self._client.close() + await self._client.aclose() self._client = None async def infer(self, ctx: AIPromptContext) -> AIInference | None: @@ -213,15 +223,20 @@ async def infer(self, ctx: AIPromptContext) -> AIInference | None: {"role": "system", "content": _SYSTEM_PROMPT}, {"role": "user", "content": _build_user_prompt(ctx)}, ] + payload = { + "model": self._model, + "messages": messages, + "response_format": _RESPONSE_FORMAT, + "max_tokens": self._max_output_tokens, + "temperature": 0, + } try: - completion = await client.chat.completions.create( - model=self._model, - messages=messages, - response_format=_RESPONSE_FORMAT, - max_tokens=self._max_output_tokens, - temperature=0, + response = await asyncio.wait_for( + client.post("/chat/completions", json=payload), timeout=self._timeout, ) + response.raise_for_status() + completion = response.json() except asyncio.CancelledError: raise except Exception as exc: @@ -234,8 +249,8 @@ async def infer(self, ctx: AIPromptContext) -> AIInference | None: return None try: - raw = completion.choices[0].message.content - except (AttributeError, IndexError) as exc: + raw = completion["choices"][0]["message"]["content"] + except (KeyError, TypeError, IndexError) as exc: logger.warning("openai_ai.unexpected_shape", error=str(exc)) return None diff --git a/app/services/content_analyzer/analyze.py b/app/services/content_analyzer/analyze.py index db905e2..b73b88e 100644 --- a/app/services/content_analyzer/analyze.py +++ b/app/services/content_analyzer/analyze.py @@ -8,7 +8,8 @@ from __future__ import annotations import asyncio -from collections.abc import Iterable +import inspect +from collections.abc import Awaitable, Iterable from app.core.config import settings from app.core.logging import get_logger @@ -69,12 +70,38 @@ def _fetch_failed_score(error: str | None) -> int: return settings.score_weight_content_fetch_failed +def _fetch_failed_reason(error: str | None) -> str: + if error == "http_error_404": + return "페이지를 찾을 수 없습니다." + if error and error.startswith("http_error_4"): + return "페이지 요청이 거부되었거나 찾을 수 없습니다." + if error and error.startswith("http_error_5"): + return "대상 서버 오류로 페이지를 확인할 수 없습니다." + if error == "timeout": + return "페이지 응답 시간이 초과되었습니다." + if error == "connect_error": + return "페이지에 연결할 수 없습니다." + if error == "dns_failure": + return "도메인 주소를 확인할 수 없습니다." + if error == "not_html": + return "분석 가능한 HTML 페이지가 아닙니다." + if error == "too_large": + return "페이지가 너무 커서 분석하지 않았습니다." + if error == "blocked_host": + return "내부망 또는 차단된 호스트라 분석하지 않았습니다." + if error == "unexpected_redirect": + return "예상하지 못한 리다이렉트 응답으로 페이지를 분석하지 못했습니다." + return "페이지를 가져오지 못했습니다." + + def _fetch_failed_result(final_url: str, error: str | None) -> ContentAnalysisResult: return ContentAnalysisResult( final_url=final_url, fetched=False, + status_code=None, score=_fetch_failed_score(error), signals=[ContentSignal.FETCH_FAILED], + reason=_fetch_failed_reason(error), error=error, ) @@ -98,7 +125,7 @@ async def analyze_content( final_url: str, *, provider: AIProvider | None = None, - upstream_signals: Iterable[str] = (), + upstream_signals: Iterable[str] | Awaitable[Iterable[str]] = (), ) -> ContentAnalysisResult: """콘텐츠 정적 분석 진입점. @@ -115,7 +142,9 @@ async def analyze_content( error=fetch_result.error, status=fetch_result.status_code, ) - return _fetch_failed_result(final_url, fetch_result.error) + result = _fetch_failed_result(final_url, fetch_result.error) + result.status_code = fetch_result.status_code + return result features = await extract_features_async(fetch_result.html, base_url=final_url) scoring: ContentScoring = score_content(features, final_url) @@ -127,7 +156,10 @@ async def analyze_content( ai_token_usage: TokenUsage | None = None active_provider = provider if provider is not None else get_ai_provider() - upstream_tuple = tuple(upstream_signals) + if inspect.isawaitable(upstream_signals): + upstream_tuple = tuple(await upstream_signals) + else: + upstream_tuple = tuple(upstream_signals) try: inference = await active_provider.infer( _build_prompt_context(final_url, features, upstream_tuple) @@ -163,6 +195,7 @@ async def analyze_content( return ContentAnalysisResult( final_url=final_url, fetched=True, + status_code=fetch_result.status_code, score=score, signals=list(scoring.signals), title=features.title, diff --git a/tests/services/content_analyzer/test_ai_openai.py b/tests/services/content_analyzer/test_ai_openai.py index d2b8413..28f6ddc 100644 --- a/tests/services/content_analyzer/test_ai_openai.py +++ b/tests/services/content_analyzer/test_ai_openai.py @@ -1,11 +1,13 @@ -"""OpenAIProvider — gpt-4o-mini 기반 피싱 추론 어댑터.""" +"""OpenAIProvider — Chat Completions 기반 피싱 추론 어댑터.""" from __future__ import annotations import asyncio import json +import sys +import time from types import SimpleNamespace -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest from app.core.config import settings @@ -30,48 +32,48 @@ def _ctx() -> AIPromptContext: ) -def _mock_completion( +def _mock_http_response( content: str, *, - usage: SimpleNamespace | None = None, + usage: dict[str, int] | None = None, ) -> SimpleNamespace: - """chat.completions.create 가 돌려주는 객체 모양을 흉내낸 간단 스텁.""" return SimpleNamespace( - choices=[SimpleNamespace(message=SimpleNamespace(content=content))], - usage=usage, + raise_for_status=lambda: None, + json=lambda: { + "choices": [{"message": {"content": content}}], + "usage": usage, + }, ) -def _usage(prompt: int, completion: int) -> SimpleNamespace: +def _mock_http_client( + content: str, + *, + usage: dict[str, int] | None = None, +) -> SimpleNamespace: return SimpleNamespace( - prompt_tokens=prompt, - completion_tokens=completion, - total_tokens=prompt + completion, + post=AsyncMock(return_value=_mock_http_response(content, usage=usage)), + aclose=AsyncMock(), ) +def _usage(prompt: int, completion: int) -> dict[str, int]: + return { + "prompt_tokens": prompt, + "completion_tokens": completion, + "total_tokens": prompt + completion, + } + + async def test_openai_returns_verdict_on_success() -> None: payload = json.dumps({"verdict": "phishing", "reason": "브랜드 불일치"}) - client = SimpleNamespace( - chat=SimpleNamespace( - completions=SimpleNamespace( - create=AsyncMock( - return_value=_mock_completion(payload, usage=_usage(120, 18)) - ) - ) - ) - ) - with patch( - "app.services.content_analyzer.ai_openai.AsyncOpenAI", - return_value=client, - ): - provider = OpenAIProvider() - result = await provider.infer(_ctx()) + client = _mock_http_client(payload, usage=_usage(120, 18)) + with patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", return_value=client): + result = await OpenAIProvider().infer(_ctx()) assert result is not None assert result.verdict == AIVerdict.PHISHING assert result.reason == "브랜드 불일치" - # 모델 id 와 토큰 사용량이 추론 결과에 실려 나와야 한다 assert result.model == settings.openai_model assert result.token_usage is not None assert result.token_usage.prompt_tokens == 120 @@ -81,18 +83,8 @@ async def test_openai_returns_verdict_on_success() -> None: async def test_openai_missing_usage_returns_none_token_usage() -> None: """usage 가 비어 있으면 token_usage=None 으로 떨어뜨린다 (추론 자체는 성공).""" - payload = json.dumps({"verdict": "benign", "reason": "ok"}) - client = SimpleNamespace( - chat=SimpleNamespace( - completions=SimpleNamespace( - create=AsyncMock(return_value=_mock_completion(payload, usage=None)) - ) - ) - ) - with patch( - "app.services.content_analyzer.ai_openai.AsyncOpenAI", - return_value=client, - ): + client = _mock_http_client(json.dumps({"verdict": "benign", "reason": "ok"})) + with patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", return_value=client): result = await OpenAIProvider().infer(_ctx()) assert result is not None @@ -101,18 +93,12 @@ async def test_openai_missing_usage_returns_none_token_usage() -> None: async def test_openai_custom_model_overrides_settings() -> None: """생성자에 model 을 넘기면 settings.openai_model 보다 우선한다 — 모델 비교용.""" - payload = json.dumps({"verdict": "benign", "reason": "ok"}) - create = AsyncMock(return_value=_mock_completion(payload, usage=_usage(10, 5))) - client = SimpleNamespace( - chat=SimpleNamespace(completions=SimpleNamespace(create=create)) - ) - with patch( - "app.services.content_analyzer.ai_openai.AsyncOpenAI", - return_value=client, - ): + client = _mock_http_client(json.dumps({"verdict": "benign", "reason": "ok"})) + with patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", return_value=client): result = await OpenAIProvider(model="gpt-4o").infer(_ctx()) - assert create.call_args.kwargs["model"] == "gpt-4o" + request_json = client.post.await_args.kwargs["json"] + assert request_json["model"] == "gpt-4o" assert result is not None assert result.model == "gpt-4o" @@ -126,33 +112,77 @@ async def test_openai_returns_none_without_api_key(monkeypatch: pytest.MonkeyPat async def test_openai_uses_configured_model(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(settings, "openai_model", "gpt-4o-mini") - create = AsyncMock( - return_value=_mock_completion(json.dumps({"verdict": "benign", "reason": "ok"})) - ) - client = SimpleNamespace( - chat=SimpleNamespace(completions=SimpleNamespace(create=create)) - ) - with patch( - "app.services.content_analyzer.ai_openai.AsyncOpenAI", - return_value=client, - ): + client = _mock_http_client(json.dumps({"verdict": "benign", "reason": "ok"})) + with patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", return_value=client): + await OpenAIProvider().infer(_ctx()) + + request_json = client.post.await_args.kwargs["json"] + assert request_json["model"] == "gpt-4o-mini" + assert request_json["response_format"]["type"] == "json_schema" + assert request_json["response_format"]["json_schema"]["strict"] is True + assert request_json["max_tokens"] == settings.openai_max_output_tokens + + +async def test_openai_prompt_requests_short_expert_plain_korean_reason() -> None: + client = _mock_http_client(json.dumps({"verdict": "benign", "reason": "ok"})) + with patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", return_value=client): + await OpenAIProvider().infer(_ctx()) + + system_msg = client.post.await_args.kwargs["json"]["messages"][0]["content"] + assert "100자" in system_msg + assert "보안 전문가" in system_msg + assert "쉬운 한국어" in system_msg + + +async def test_openai_truncates_reason_to_100_chars() -> None: + long_reason = "가" * 150 + client = _mock_http_client(json.dumps({"verdict": "suspicious", "reason": long_reason})) + with patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", return_value=client): + result = await OpenAIProvider().infer(_ctx()) + + assert result is not None + assert len(result.reason) == 100 + + +async def test_openai_uses_http_client_without_sdk_resource_import() -> None: + client = _mock_http_client(json.dumps({"verdict": "benign", "reason": "ok"})) + httpx_client_cls = MagicMock(return_value=client) + with patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", httpx_client_cls): + await OpenAIProvider(timeout_seconds=3.0).infer(_ctx()) + + assert httpx_client_cls.call_args.kwargs["base_url"] == "https://api.openai.com/v1" + assert httpx_client_cls.call_args.kwargs["timeout"] == 3.0 + assert httpx_client_cls.call_args.kwargs["trust_env"] is False + assert client.post.await_args.args == ("/chat/completions",) + + +async def test_openai_request_has_hard_timeout() -> None: + async def never_returns(*_: object, **__: object) -> object: + await asyncio.sleep(10) + return _mock_http_response(json.dumps({"verdict": "benign", "reason": "late"})) + + client = SimpleNamespace(post=never_returns, aclose=AsyncMock()) + with patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", return_value=client): + started = time.perf_counter() + result = await OpenAIProvider(timeout_seconds=0.01).infer(_ctx()) + + assert result is None + assert time.perf_counter() - started < 0.5 + + +async def test_openai_provider_does_not_import_openai_sdk() -> None: + sys.modules.pop("openai", None) + client = _mock_http_client(json.dumps({"verdict": "benign", "reason": "ok"})) + with patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", return_value=client): await OpenAIProvider().infer(_ctx()) - kwargs = create.call_args.kwargs - assert kwargs["model"] == "gpt-4o-mini" - # structured output 스키마 사용 - assert kwargs["response_format"]["type"] == "json_schema" - assert kwargs["response_format"]["json_schema"]["strict"] is True - # timeout 전달 - assert kwargs["timeout"] == settings.openai_timeout_seconds + assert "openai" not in sys.modules async def test_openai_user_prompt_includes_upstream_signals() -> None: """ctx.upstream_signals 가 user payload 의 upstream_signals 배열로 직렬화돼야 한다.""" - payload = json.dumps({"verdict": "phishing", "reason": "타이포 + 비밀번호 폼"}) - create = AsyncMock(return_value=_mock_completion(payload, usage=_usage(50, 10))) - client = SimpleNamespace( - chat=SimpleNamespace(completions=SimpleNamespace(create=create)) + client = _mock_http_client( + json.dumps({"verdict": "phishing", "reason": "타이포 + 비밀번호 폼"}) ) ctx = AIPromptContext( final_url="https://evil-naverr.test/signin", @@ -163,89 +193,62 @@ async def test_openai_user_prompt_includes_upstream_signals() -> None: external_link_ratio=0.9, upstream_signals=("TYPO_DOMAIN", "NEW_DOMAIN"), ) - with patch( - "app.services.content_analyzer.ai_openai.AsyncOpenAI", - return_value=client, - ): + with patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", return_value=client): await OpenAIProvider().infer(ctx) - user_msg = create.call_args.kwargs["messages"][1]["content"] + user_msg = client.post.await_args.kwargs["json"]["messages"][1]["content"] body = json.loads(user_msg) assert body["upstream_signals"] == ["TYPO_DOMAIN", "NEW_DOMAIN"] async def test_openai_user_prompt_omits_empty_upstream_signals() -> None: """기본값(빈 튜플)이면 빈 배열로 직렬화돼 단독 페이지 분석 모드와 동일하게 동작.""" - payload = json.dumps({"verdict": "benign", "reason": "ok"}) - create = AsyncMock(return_value=_mock_completion(payload)) - client = SimpleNamespace( - chat=SimpleNamespace(completions=SimpleNamespace(create=create)) - ) - with patch( - "app.services.content_analyzer.ai_openai.AsyncOpenAI", - return_value=client, - ): + client = _mock_http_client(json.dumps({"verdict": "benign", "reason": "ok"})) + with patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", return_value=client): await OpenAIProvider().infer(_ctx()) - body = json.loads(create.call_args.kwargs["messages"][1]["content"]) + body = json.loads(client.post.await_args.kwargs["json"]["messages"][1]["content"]) assert body["upstream_signals"] == [] async def test_openai_parse_error_returns_none() -> None: - client = SimpleNamespace( - chat=SimpleNamespace( - completions=SimpleNamespace( - create=AsyncMock(return_value=_mock_completion("not-json")) - ) - ) - ) - with patch( - "app.services.content_analyzer.ai_openai.AsyncOpenAI", - return_value=client, - ): + client = _mock_http_client("not-json") + with patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", return_value=client): result = await OpenAIProvider().infer(_ctx()) assert result is None async def test_openai_unknown_verdict_returns_none() -> None: """응답 JSON 의 verdict 값이 enum 범위를 벗어나면 None 으로 떨어뜨린다.""" - payload = json.dumps({"verdict": "unknown_value", "reason": "x"}) - client = SimpleNamespace( - chat=SimpleNamespace( - completions=SimpleNamespace(create=AsyncMock(return_value=_mock_completion(payload))) - ) - ) - with patch( - "app.services.content_analyzer.ai_openai.AsyncOpenAI", - return_value=client, - ): + client = _mock_http_client(json.dumps({"verdict": "unknown_value", "reason": "x"})) + with patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", return_value=client): assert await OpenAIProvider().infer(_ctx()) is None async def test_openai_api_exception_returns_none() -> None: - client = SimpleNamespace( - chat=SimpleNamespace( - completions=SimpleNamespace(create=AsyncMock(side_effect=RuntimeError("5xx"))) - ) - ) - with patch( - "app.services.content_analyzer.ai_openai.AsyncOpenAI", - return_value=client, - ): + client = SimpleNamespace(post=AsyncMock(side_effect=RuntimeError("5xx")), aclose=AsyncMock()) + with patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", return_value=client): result = await OpenAIProvider().infer(_ctx()) assert result is None async def test_openai_cancelled_propagates() -> None: client = SimpleNamespace( - chat=SimpleNamespace( - completions=SimpleNamespace( - create=AsyncMock(side_effect=asyncio.CancelledError()) - ) - ) + post=AsyncMock(side_effect=asyncio.CancelledError()), + aclose=AsyncMock(), ) - with patch( - "app.services.content_analyzer.ai_openai.AsyncOpenAI", - return_value=client, - ), pytest.raises(asyncio.CancelledError): + with ( + patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", return_value=client), + pytest.raises(asyncio.CancelledError), + ): await OpenAIProvider().infer(_ctx()) + + +async def test_openai_provider_closes_http_client_with_aclose() -> None: + client = _mock_http_client(json.dumps({"verdict": "benign", "reason": "ok"})) + with patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", return_value=client): + provider = OpenAIProvider() + await provider.infer(_ctx()) + await provider.aclose() + + client.aclose.assert_awaited_once() diff --git a/tests/services/content_analyzer/test_analyze.py b/tests/services/content_analyzer/test_analyze.py index 513cdfe..ee2a3c5 100644 --- a/tests/services/content_analyzer/test_analyze.py +++ b/tests/services/content_analyzer/test_analyze.py @@ -20,7 +20,7 @@ def _mock_fetch(ok: bool, html: str = "", error: str | None = None, status: int return_value=FetchResult( ok=ok, url="https://x.test/", - status_code=status if ok else None, + status_code=status, html=html, error=error, ) @@ -54,6 +54,23 @@ async def test_fetch_failure_degraded_result(self) -> None: assert result.score == settings.score_weight_content_fetch_failed assert result.ai_verdict is None + async def test_http_404_fetch_failure_has_human_readable_reason(self) -> None: + with _mock_fetch(ok=False, error="http_error_404", status=404): + result = await analyze_content("https://missing.test/") + + assert result.fetched is False + assert result.error == "http_error_404" + assert result.reason == "페이지를 찾을 수 없습니다." + assert result.status_code == 404 + assert result.ai_verdict is None + + async def test_success_exposes_final_url_status_code(self) -> None: + with _mock_fetch(ok=True, html="", status=204): + result = await analyze_content("https://ok.test/") + + assert result.fetched is True + assert result.status_code == 204 + @pytest.mark.parametrize("error", ["not_html", "too_large", "unexpected_redirect"]) async def test_benign_fetch_errors_score_zero(self, error: str) -> None: """정상 컨텐츠(PDF/이미지/대형 정적)·파이프라인 이슈는 시그널만 남기고 가산은 0.""" From bad7aabda707bad9ba51594f7414aed56c92f6ae Mon Sep 17 00:00:00 2001 From: minsoo0506 Date: Thu, 21 May 2026 18:30:32 +0900 Subject: [PATCH 4/5] =?UTF-8?q?[Fix]=20=EA=B0=9C=EB=B0=9C=20=EC=84=9C?= =?UTF-8?q?=EB=B2=84=20=EC=8B=A4=ED=96=89=EA=B3=BC=20=EB=B6=84=EC=84=9D=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 --- Makefile | 3 +- README.md | 57 +++++++++++++++++---------- app/core/dns_cache.py | 8 ++-- app/services/domain_heuristic/rdap.py | 2 +- tests/test_makefile.py | 11 ++++++ 5 files changed, 54 insertions(+), 27 deletions(-) create mode 100644 tests/test_makefile.py diff --git a/Makefile b/Makefile index 0c4654a..1ab9232 100644 --- a/Makefile +++ b/Makefile @@ -8,7 +8,8 @@ install: ## Install runtime + dev dependencies pre-commit install run: ## Run the API locally with hot reload - uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 + uvicorn app.main:app --reload --reload-dir app --reload-exclude .venv \ + --host 0.0.0.0 --port 8000 test: ## Run the test suite pytest diff --git a/README.md b/README.md index 1af6c9a..c025ea4 100644 --- a/README.md +++ b/README.md @@ -15,10 +15,10 @@ | 로컬 캐시 DB | **SQLite + aiosqlite** | URLhaus 등 외부 피드 캐시 전용 | | ORM | **SQLAlchemy 2.0 (async)** | DeclarativeBase + naming convention | | Migration | **Alembic** | SQLite batch mode | -| HTTP Client | **httpx** | 외부 API 호출 (GSB / RDAP / Claude / Spring 콜백) | +| HTTP Client | **httpx** | 외부 API 호출 (GSB / RDAP / OpenAI / Spring 콜백) | | Crawler | **BeautifulSoup4 + requests** | 페이지 본문 추출, 피싱 신호 탐지 | | Domain Lookup | **RDAP (httpx)** | 도메인 등록일·만료일·레지스트라 조회 | -| 캐시 | **인메모리 dict + TTL + single-flight** / **SQLite 스냅샷** / **`functools.lru_cache`** | RDAP 24h 캐시·동시 요청 합치기 / URLhaus 로컬 캐시 / Settings 싱글톤 | +| 캐시 | **인메모리 dict + TTL + single-flight** / **SQLite 스냅샷** / **`functools.lru_cache`** | RDAP 7일 캐시·동시 요청 합치기 / URLhaus 로컬 캐시 / Settings 싱글톤 | | Scheduler | **APScheduler** | URLhaus 주기 동기화 | | Validation | **Pydantic v2 + pydantic-settings** | 요청·응답·환경변수 | | Logging | **structlog** | 구조적 로깅 + request_id 자동 바인딩 | @@ -179,11 +179,13 @@ linclean-fastapi/ 기반으로 조립합니다. - **`services/`** — 4단계 파이프라인을 **단계별 하위 패키지**로 분리합니다. 각 패키지의 `__init__.py` 가 해당 단계의 public 진입점을 re-export 하며, - 오케스트레이터(`pipeline.py`)가 이를 순차 호출합니다. `Request` 같은 FastAPI + 오케스트레이터(`pipeline.py`)가 이를 조립합니다. `Request` 같은 FastAPI 객체를 받지 않고 `AsyncSession` / 순수 인자만 받습니다. - **`normalizer/`** — 1단계. `normalize_url()` 로 URL 을 canonical form 으로 - 정규화합니다 (스킴·호스트 소문자화, 기본 포트 제거, 퍼센트 인코딩 정돈, - 경로 dot-segment 해소, IDN 디코딩, 프래그먼트 제거, 입력 검증). + 정규화합니다 (앞뒤 공백 제거, 스킴·호스트 소문자화, 기본 포트 제거, + 퍼센트 인코딩 정돈, 경로 dot-segment 해소, IDN 디코딩, 프래그먼트 제거, + 입력 검증). 스킴이 없는 입력은 먼저 `http://` 를 붙이고, 사용자가 스킴을 + 생략한 경우에만 짧은 HTTPS probe 로 응답 가능하면 `https://` 로 올립니다. - **`unchainer/`** — 1단계 후반. `unchain_url()` 로 리다이렉트 체인(3xx Location) 을 끝까지 추적해 최종 URL 을 확정합니다. HEAD 우선 → GET 폴백 전략으로 대역폭을 절약하면서 호환성을 확보하고, 네트워크 에러 시에도 GET 으로 @@ -200,8 +202,10 @@ linclean-fastapi/ - **`content_analyzer/`** — 4단계. 최종 URL의 HTML만 fetch 하고, lxml 기반 정적 추출 결과를 규칙 점수와 AI 보조 판정으로 합성합니다. 네트워크/AI 실패는 degraded 결과로 흡수하되 `CancelledError` 는 상위로 전파합니다. - - **`pipeline.py`** — 1~4단계를 조립합니다. 2·3단계를 병렬 실행하고, 외부 - 위협 DB 매치나 danger 임계 도달 시 비용이 큰 4단계를 short-circuit 합니다. + - **`pipeline.py`** — 1~4단계를 조립합니다. 2·3단계와 4단계의 fetch/extract 를 + 겹쳐 실행하고, 외부 위협 DB 매치나 danger 임계 도달 시 실행 중인 4단계를 + 취소해 short-circuit 합니다. AI 판정은 선행 단계 신호가 준비된 뒤에만 + 수행됩니다. - **`analysis_callback.py`** — 비동기 `/analyze` 완료 후 Spring 내부 콜백 엔드포인트로 결과를 POST 합니다. 2xx 외 응답/네트워크 오류는 최대 3회 재시도하고, 최종 실패는 dead-letter 로그로 남깁니다. @@ -307,20 +311,22 @@ upsert 합니다. 분석 시에는 외부 호출 없이 로컬 인덱스만 조 규칙 기반 점수표로 도메인의 위험 신호를 합산합니다. 도메인 등록 정보는 **RDAP (RFC 7480~7484)** 로 조회합니다. -**2단계와 동시 실행 + 외부 DB 매치 시 조기 종료**: 두 단계 모두 1단계 최종 URL -만 필요하고 서로 독립이라 `run_pipeline` 에서 두 task 를 동시에 띄우고 -`asyncio.wait(return_when=FIRST_COMPLETED)` 로 먼저 끝난 쪽을 확인합니다. +**2단계·3단계·4단계 일부 동시 실행 + 외부 DB 매치 시 조기 종료**: 2·3단계는 +1단계 최종 URL만 필요하고 서로 독립이라 `run_pipeline` 에서 동시에 띄웁니다. +4단계도 HTML fetch/extract 까지는 동시에 시작하지만, AI 판정은 threat DB/RDAP +신호가 확정된 뒤 그 신호를 프롬프트에 실어 수행합니다. - **GSB 또는 URLhaus 매치 (`threat_db.is_malicious=True`)** 가 먼저 떨어지면, - 아직 RDAP 대기 중일 수 있는 heuristic task 를 **즉시 `cancel()`** 하고 4단계도 - `skipped_already_danger` 로 묶어 바로 반환합니다. verdict 가 이미 danger 로 - 확정이므로 RDAP 이 돌아올 때까지 대기할 이유가 없습니다. heuristic 자리에는 - `rdap_error="skipped_threat_matched"` 인 placeholder 가 채워져 응답 스키마를 - 유지합니다. + 아직 RDAP 또는 콘텐츠 fetch 대기 중일 수 있는 task 를 **즉시 `cancel()`** 하고 + 4단계는 `skipped_already_danger` 로 묶어 바로 반환합니다. verdict 가 이미 + danger 로 확정이므로 RDAP/AI 가 돌아올 때까지 대기할 이유가 없습니다. + heuristic 자리에는 `skipped_reason="threat_matched"` 인 placeholder 가 채워져 + 응답 스키마를 유지합니다. - **heuristic 이 먼저 끝난 경우**는 threat_db 를 마저 기다린 뒤, is_malicious 이거나 합산 점수가 임계를 넘으면 4단계만 skip 합니다. -- **정상 경로**에서는 GSB(100~300ms)와 RDAP(캐시 미스 시 최대 ~5s)의 latency 가 - 겹쳐서 4단계 skip 판정까지의 wall clock 이 둘 중 느린 쪽으로 수렴합니다. +- **정상 경로**에서는 GSB, RDAP(캐시 미스 시 기본 최대 3s), 콘텐츠 fetch/extract 의 + latency 가 겹칩니다. 따라서 총 응답 시간은 각 단계를 단순 합산하지 않고, + `unchain + max(2·3단계, fetch/extract) + AI` 에 가깝게 수렴합니다. - `CancelledError` 와 stage 내부 예외는 남은 task 를 정리한 뒤 상위로 전파되어 shutdown / 타임아웃 신호가 degraded 결과로 삼켜지지 않습니다. @@ -338,7 +344,7 @@ upsert 합니다. 분석 시에는 외부 호출 없이 로컬 인덱스만 조 | DGA 의심 도메인 | Shannon 엔트로피 ≥ 3.5 또는 자음 비율 ≥ 0.7 | +15 | | 합법 호스팅 플랫폼 (공유 호스팅 주의 가중치) | `user.github.io`, `app.netlify.app` | +15 | -레벤슈타인 거리 함수는 외부 라이브러리에 의존하지 않고 직접 구현합니다 (DP). 약 500개 브랜드 화이트리스트(`brands.txt`)와 비교합니다. DGA 탐지는 Shannon 엔트로피와 자음 비율 통계만 사용하며 외부 모델이 필요 없습니다. RDAP 응답은 도메인 단위로 인메모리 캐싱(24h, `rdap_cache_ttl_seconds`)하여 동일 도메인 재조회 비용을 줄입니다. 캐시 만료·미스 순간에도 같은 도메인으로 몰리는 요청은 `_inflight` dict + `asyncio.Future` 로 합쳐(**single-flight / request coalescing**) RDAP 서버로 나가는 HTTP 호출을 1건으로 수렴시킵니다. RDAP 실패 시 신생 도메인 신호를 발동하지 않습니다 ("모름"을 "위험"으로 취급하지 않는 원칙). +레벤슈타인 거리 함수는 외부 라이브러리에 의존하지 않고 직접 구현합니다 (DP). 약 500개 브랜드 화이트리스트(`brands.txt`)와 비교합니다. DGA 탐지는 Shannon 엔트로피와 자음 비율 통계만 사용하며 외부 모델이 필요 없습니다. RDAP 응답은 도메인 단위로 인메모리 캐싱(7일, `rdap_cache_ttl_seconds`)하여 동일 도메인 재조회 비용을 줄입니다. 캐시 만료·미스 순간에도 같은 도메인으로 몰리는 요청은 `_inflight` dict + `asyncio.Future` 로 합쳐(**single-flight / request coalescing**) RDAP 서버로 나가는 HTTP 호출을 1건으로 수렴시킵니다. RDAP 서버가 429 를 반환하면 `Retry-After` 또는 기본 쿨다운 동안 추가 RDAP 호출을 건너뛰고 `rdap_error="rate_limited"` 로 내려 호출량을 제한합니다. RDAP 실패 시 신생 도메인 신호를 발동하지 않습니다 ("모름"을 "위험"으로 취급하지 않는 원칙). `HOSTING_PLATFORM` 은 "이 도메인이 악성이다" 라는 신호가 아니라 **공유 호스팅 컨텍스트**(GitHub Pages·Netlify·Vercel·Heroku 등 다수 테넌트가 같은 상위 도메인을 공유)를 나타내는 주의 가중치입니다. URLhaus 매칭 키가 `host + path-prefix` 로 확장되는 것과 같은 맥락에서, 계정·리포 단위로 악성 여부가 갈리는 환경이므로 +15 를 가산합니다. 플랫폼 루트 도메인 자체(`netlify.app`, `vercel.app` 등)는 정상 운영 도메인이므로 타이포스쿼팅 검사에서 제외됩니다. @@ -444,11 +450,15 @@ OPENAI_MODEL=gpt-4o-mini # 기본 — 저비용·저지연 | `AI_PROVIDER` | `auto` | `auto` (키 있으면 openai) / `openai` / `null` (비활성) | | `OPENAI_API_KEY` | *(없음)* | 비워두면 `NullAIProvider` — 규칙 점수만 사용 | | `OPENAI_MODEL` | `gpt-4o-mini` | OpenAI 채팅 모델 id | -| `OPENAI_TIMEOUT_SECONDS` | `10.0` | 단일 호출 타임아웃 | -| `OPENAI_MAX_OUTPUT_TOKENS` | `300` | verdict + reason 용으로 여유 있는 상한 | +| `OPENAI_TIMEOUT_SECONDS` | `5.0` | 단일 호출 타임아웃 | +| `OPENAI_MAX_OUTPUT_TOKENS` | `120` | verdict + 100자 이내 reason 용 출력 상한 | #### 응답에 실리는 AI 메타데이터 +`ai_reason` 은 보안 전문가의 근거 중심 문장으로 요청하되, 비전문가도 이해할 수 +있도록 쉬운 한국어 100자 이내로 제한합니다. 모델이 더 길게 응답해도 클라이언트에서 +100자로 잘라 응답합니다. + `ContentAnalysisResult` 에는 verdict/reason 뿐 아니라 **실제로 응답한 모델 id 와 토큰 사용량**이 함께 실립니다. 비용 관측, 모델 비교, 프롬프트 튜닝에 그대로 쓸 수 있게 하기 위함입니다. @@ -518,7 +528,9 @@ NullProvider 로 폴백된 경우에는 `ai_error="provider_misconfigured"` 로 현재 코드는 단계별 단독 호출 라우터를 운영 라우터에 항상 마운트합니다. 모든 엔드포인트는 `X-Internal-Api-Key` 인증을 요구하며, raw URL 을 받아 `normalize_url()` 로 1차 검증한 뒤 해당 단계만 실행합니다. 전체 파이프라인을 -동기로 확인하려면 `/api/v1/analyze/sync` 를 사용합니다. +동기로 확인하려면 `/api/v1/analyze/sync` 를 사용합니다. 외부 위협 DB(GSB/URLhaus) +없이 URL·리다이렉트·RDAP·콘텐츠/AI만 확인하려면 +`/api/v1/analyze/db-independent/sync` 를 사용합니다. | Method | Path | 단계 | 호출 함수 | |--------|---------------------------------|---------------------|---------------------------------| @@ -527,6 +539,7 @@ NullProvider 로 폴백된 경우에는 `ai_error="provider_misconfigured"` 로 | POST | `/domain-heuristic` | Stage 3 | `check_domain_heuristic` | | POST | `/content-analysis` | Stage 4 | `analyze_content` | | POST | `/analyze/sync` | 전체 (1~4 + verdict)| `run_pipeline` | +| POST | `/analyze/db-independent/sync` | DB 비의존 전체 | `run_db_independent_pipeline` | | POST | `/content/fetch-extract` | 4단계 보조 확인 | `fetch_page` + `extract_features` | 요청 바디는 모두 `{ "url": "" }` 형태이며, `/analyze/sync` 는 @@ -699,6 +712,7 @@ Spring FastAPI (본 엔진) "content_analysis": { "final_url": "https://login-secure-naver-auth.com/signin", "fetched": false, + "status_code": null, "score": 0, "signals": ["SKIPPED_ALREADY_DANGER"], "title": null, @@ -710,6 +724,7 @@ Spring FastAPI (본 엔진) "is_spa_shell": false, "ai_verdict": null, "ai_reason": null, + "reason": null, "ai_error": null, "ai_model": null, "ai_token_usage": null, diff --git a/app/core/dns_cache.py b/app/core/dns_cache.py index 0e1e9b5..5fc4f36 100644 --- a/app/core/dns_cache.py +++ b/app/core/dns_cache.py @@ -16,9 +16,9 @@ import asyncio import socket -from typing import Any, cast +from typing import Any -from cachetools import TTLCache # type: ignore[import-untyped] +from cachetools import TTLCache from app.core.config import settings @@ -35,13 +35,13 @@ async def resolve_host_addrs(hostname: str) -> tuple[AddrInfoTuple, ...]: """hostname 의 getaddrinfo 결과를 캐시. 실패 시 OSError 그대로 raise.""" try: - return cast(tuple[AddrInfoTuple, ...], _cache[hostname]) + return _cache[hostname] except KeyError: pass loop = asyncio.get_running_loop() infos = await loop.getaddrinfo(hostname, None, type=socket.SOCK_STREAM) - result = cast(tuple[AddrInfoTuple, ...], tuple(infos)) + result = tuple(infos) _cache[hostname] = result return result diff --git a/app/services/domain_heuristic/rdap.py b/app/services/domain_heuristic/rdap.py index e060378..eb3172a 100644 --- a/app/services/domain_heuristic/rdap.py +++ b/app/services/domain_heuristic/rdap.py @@ -6,7 +6,7 @@ from email.utils import parsedate_to_datetime import httpx -from cachetools import TTLCache # type: ignore[import-untyped] +from cachetools import TTLCache from app.core.config import settings from app.core.logging import get_logger diff --git a/tests/test_makefile.py b/tests/test_makefile.py new file mode 100644 index 0000000..24147f5 --- /dev/null +++ b/tests/test_makefile.py @@ -0,0 +1,11 @@ +from pathlib import Path + + +def test_run_target_reloads_app_only() -> None: + makefile = Path("Makefile").read_text() + run_block = makefile.split("run: ## Run the API locally with hot reload", 1)[1].split( + "\n\n", 1 + )[0] + + assert "--reload-dir app" in run_block + assert "--reload-exclude .venv" in run_block From 42f6a62435083958ff6891074f871841ea59dd96 Mon Sep 17 00:00:00 2001 From: minsoo0506 Date: Thu, 21 May 2026 19:07:07 +0900 Subject: [PATCH 5/5] =?UTF-8?q?[Fix]=20=EC=95=85=EC=84=B1=20URL=20?= =?UTF-8?q?=EC=A1=B0=EA=B8=B0=20=EC=A2=85=EB=A3=8C=EC=99=80=20=EC=8A=A4?= =?UTF-8?q?=ED=82=B4=20=ED=8C=90=EB=B3=84=20=EC=88=98=EC=A0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 40 +++++++------- app/services/content_analyzer/analyze.py | 2 + app/services/db_independent_pipeline.py | 45 +++------------- app/services/pipeline.py | 54 +++++-------------- app/services/unchainer/unchain.py | 12 ++++- .../services/content_analyzer/test_analyze.py | 10 +++- .../services/test_db_independent_pipeline.py | 51 +++++++++++++----- tests/services/test_pipeline.py | 48 ++++++++++++++--- tests/services/unchainer/test_unchain.py | 47 ++++++++++++++++ 9 files changed, 186 insertions(+), 123 deletions(-) diff --git a/README.md b/README.md index c025ea4..96201f9 100644 --- a/README.md +++ b/README.md @@ -184,8 +184,8 @@ linclean-fastapi/ - **`normalizer/`** — 1단계. `normalize_url()` 로 URL 을 canonical form 으로 정규화합니다 (앞뒤 공백 제거, 스킴·호스트 소문자화, 기본 포트 제거, 퍼센트 인코딩 정돈, 경로 dot-segment 해소, IDN 디코딩, 프래그먼트 제거, - 입력 검증). 스킴이 없는 입력은 먼저 `http://` 를 붙이고, 사용자가 스킴을 - 생략한 경우에만 짧은 HTTPS probe 로 응답 가능하면 `https://` 로 올립니다. + 입력 검증). 스킴이 없는 입력은 먼저 `https://` 로 분석 가능한 정상 HTML + 응답인지 확인하고, 그렇지 않으면 `http://` 로 내려 분석합니다. - **`unchainer/`** — 1단계 후반. `unchain_url()` 로 리다이렉트 체인(3xx Location) 을 끝까지 추적해 최종 URL 을 확정합니다. HEAD 우선 → GET 폴백 전략으로 대역폭을 절약하면서 호환성을 확보하고, 네트워크 에러 시에도 GET 으로 @@ -202,10 +202,9 @@ linclean-fastapi/ - **`content_analyzer/`** — 4단계. 최종 URL의 HTML만 fetch 하고, lxml 기반 정적 추출 결과를 규칙 점수와 AI 보조 판정으로 합성합니다. 네트워크/AI 실패는 degraded 결과로 흡수하되 `CancelledError` 는 상위로 전파합니다. - - **`pipeline.py`** — 1~4단계를 조립합니다. 2·3단계와 4단계의 fetch/extract 를 - 겹쳐 실행하고, 외부 위협 DB 매치나 danger 임계 도달 시 실행 중인 4단계를 - 취소해 short-circuit 합니다. AI 판정은 선행 단계 신호가 준비된 뒤에만 - 수행됩니다. + - **`pipeline.py`** — 1~4단계를 조립합니다. 2·3단계를 병렬 실행하고, + 외부 위협 DB 매치나 danger 임계 도달 시 4단계를 시작하지 않고 + short-circuit 합니다. AI 판정은 선행 단계 신호가 준비된 뒤에만 수행됩니다. - **`analysis_callback.py`** — 비동기 `/analyze` 완료 후 Spring 내부 콜백 엔드포인트로 결과를 POST 합니다. 2xx 외 응답/네트워크 오류는 최대 3회 재시도하고, 최종 실패는 dead-letter 로그로 남깁니다. @@ -311,22 +310,23 @@ upsert 합니다. 분석 시에는 외부 호출 없이 로컬 인덱스만 조 규칙 기반 점수표로 도메인의 위험 신호를 합산합니다. 도메인 등록 정보는 **RDAP (RFC 7480~7484)** 로 조회합니다. -**2단계·3단계·4단계 일부 동시 실행 + 외부 DB 매치 시 조기 종료**: 2·3단계는 -1단계 최종 URL만 필요하고 서로 독립이라 `run_pipeline` 에서 동시에 띄웁니다. -4단계도 HTML fetch/extract 까지는 동시에 시작하지만, AI 판정은 threat DB/RDAP -신호가 확정된 뒤 그 신호를 프롬프트에 실어 수행합니다. +**2단계·3단계 동시 실행 + 외부 DB 매치 시 조기 종료**: 2·3단계는 1단계 +최종 URL만 필요하고 서로 독립이라 `run_pipeline` 에서 동시에 띄웁니다. +4단계는 threat DB/RDAP 신호가 확정되고 danger short-circuit 대상이 아닐 때만 +시작합니다. - **GSB 또는 URLhaus 매치 (`threat_db.is_malicious=True`)** 가 먼저 떨어지면, - 아직 RDAP 또는 콘텐츠 fetch 대기 중일 수 있는 task 를 **즉시 `cancel()`** 하고 - 4단계는 `skipped_already_danger` 로 묶어 바로 반환합니다. verdict 가 이미 - danger 로 확정이므로 RDAP/AI 가 돌아올 때까지 대기할 이유가 없습니다. + 아직 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), 콘텐츠 fetch/extract 의 - latency 가 겹칩니다. 따라서 총 응답 시간은 각 단계를 단순 합산하지 않고, - `unchain + max(2·3단계, fetch/extract) + AI` 에 가깝게 수렴합니다. +- **정상 경로**에서는 GSB 와 RDAP(캐시 미스 시 기본 최대 3s)의 latency 가 + 겹칩니다. 이후 danger 임계 미만일 때만 콘텐츠 fetch/extract 와 AI 분석을 + 수행합니다. - `CancelledError` 와 stage 내부 예외는 남은 task 를 정리한 뒤 상위로 전파되어 shutdown / 타임아웃 신호가 degraded 결과로 삼켜지지 않습니다. @@ -494,9 +494,9 @@ NullProvider 로 폴백된 경우에는 `ai_error="provider_misconfigured"` 로 | 31 ~ 60 | **caution** (주의) | 노랑 — 이유 표시 후 사용자 판단 | | 61 이상 | **danger** (위험) | 빨강 — "피싱 의심, 열지 마세요" | -**예외 — blacklist 매치는 점수와 무관하게 danger**: `threat_db.is_malicious=True` -면 합산 점수가 임계 미만이어도 verdict 가 `danger` 로 강제됩니다. GSB / URLhaus -매치 = 알려진 악성 URL 이라 점수 합산 결과보다 우선해서 결정합니다. +**예외 — blacklist 매치는 score 100 / danger**: `threat_db.is_malicious=True` +면 GSB / URLhaus 매치 = 알려진 악성 URL 로 보고 score 를 100으로 고정하며, +verdict 는 `danger` 로 강제됩니다. 이 경우 4단계 fetch/AI 는 수행하지 않습니다. 판정 근거는 `stages` 내부의 각 단계 원시 결과(`threat_db`, `domain_heuristic`, `content_analysis`)에 남습니다. 별도의 `reasons` 배열/자연어 `summary` 필드는 @@ -724,7 +724,7 @@ Spring FastAPI (본 엔진) "is_spa_shell": false, "ai_verdict": null, "ai_reason": null, - "reason": null, + "reason": "위험성이 확인된 URL입니다. 페이지를 열지 않는 것이 좋습니다.", "ai_error": null, "ai_model": null, "ai_token_usage": null, diff --git a/app/services/content_analyzer/analyze.py b/app/services/content_analyzer/analyze.py index b73b88e..6beb49f 100644 --- a/app/services/content_analyzer/analyze.py +++ b/app/services/content_analyzer/analyze.py @@ -62,6 +62,7 @@ def _ai_score_weight(verdict: AIVerdict) -> int: _FETCH_ERROR_NO_SCORE: frozenset[str] = frozenset( {"not_html", "too_large", "unexpected_redirect", "blocked_host"} ) +_SKIPPED_ALREADY_DANGER_REASON = "위험성이 확인된 URL입니다. 페이지를 열지 않는 것이 좋습니다." def _fetch_failed_score(error: str | None) -> int: @@ -117,6 +118,7 @@ def skipped_already_danger(final_url: str) -> ContentAnalysisResult: fetched=False, score=0, signals=[ContentSignal.SKIPPED_ALREADY_DANGER], + reason=_SKIPPED_ALREADY_DANGER_REASON, error="skipped_already_danger", ) diff --git a/app/services/db_independent_pipeline.py b/app/services/db_independent_pipeline.py index 019ce13..b9773ca 100644 --- a/app/services/db_independent_pipeline.py +++ b/app/services/db_independent_pipeline.py @@ -2,9 +2,7 @@ from __future__ import annotations -import asyncio import time -from contextlib import suppress import structlog @@ -94,14 +92,6 @@ def _collect_db_independent_signals( return tuple(codes) -async def _collect_db_independent_signals_after_heuristic( - heuristic_task: asyncio.Task[DomainHeuristicResult], - unchain: UnchainResult, -) -> tuple[str, ...]: - heuristic = await heuristic_task - return _collect_db_independent_signals(heuristic, unchain) - - async def run_db_independent_pipeline( analysis_id: str, original_url: str, @@ -135,41 +125,18 @@ async def run_db_independent_pipeline( _set_stage_timing(stage_timings, PipelineStage.UNCHAIN, stage_started) stage_started = time.perf_counter() - heuristic_task = asyncio.create_task(check_domain_heuristic(unchain.final_url)) - - async def _timed_heuristic() -> DomainHeuristicResult: - try: - return await heuristic_task - finally: - _set_stage_timing(stage_timings, PipelineStage.DOMAIN_HEURISTIC, stage_started) - - timed_heuristic_task = asyncio.create_task(_timed_heuristic()) - - content_started = time.perf_counter() - upstream_task = asyncio.create_task( - _collect_db_independent_signals_after_heuristic(heuristic_task, unchain) - ) - content_task = asyncio.create_task( - analyze_content( - unchain.final_url, - upstream_signals=upstream_task, - ) - ) - - heuristic = await timed_heuristic_task + heuristic = await check_domain_heuristic(unchain.final_url) + _set_stage_timing(stage_timings, PipelineStage.DOMAIN_HEURISTIC, stage_started) if heuristic.score >= settings.score_danger_threshold: - content_task.cancel() - with suppress(asyncio.CancelledError): - await content_task stage_started = time.perf_counter() content = skipped_already_danger(unchain.final_url) _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, stage_started) else: - try: - content = await content_task - finally: - _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, content_started) + upstream = _collect_db_independent_signals(heuristic, unchain) + stage_started = time.perf_counter() + content = await analyze_content(unchain.final_url, upstream_signals=upstream) + _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, stage_started) score = _total_score(heuristic, content) verdict = _decide_verdict(score) diff --git a/app/services/pipeline.py b/app/services/pipeline.py index 561b6e5..f92b421 100644 --- a/app/services/pipeline.py +++ b/app/services/pipeline.py @@ -3,9 +3,8 @@ from __future__ import annotations import asyncio -import inspect import time -from collections.abc import Awaitable, Iterable +from collections.abc import Awaitable from contextlib import suppress from typing import TYPE_CHECKING, TypeVar from urllib.parse import urlparse @@ -136,20 +135,16 @@ def _collect_upstream_signals( async def _stage_content_analysis( log: structlog.stdlib.BoundLogger, final_url: str, - upstream_signals: Iterable[str] | Awaitable[Iterable[str]], + upstream_signals: tuple[str, ...], ) -> ContentAnalysisResult: result = await analyze_content(final_url, upstream_signals=upstream_signals) - if inspect.isawaitable(upstream_signals): - upstream_for_log: list[str] | str = "deferred" - else: - upstream_for_log = list(upstream_signals) log.info( "pipeline.content_analysis.done", fetched=result.fetched, score=result.score, signals=[s.value for s in result.signals], ai_verdict=result.ai_verdict.value if result.ai_verdict else None, - upstream_signals=upstream_for_log, + upstream_signals=list(upstream_signals), ) return result @@ -170,6 +165,8 @@ def _total_score( content: ContentAnalysisResult, ) -> int: """전 단계 합산 후 100 으로 캡. content.score 는 4단계가 실제로 돌았을 때만 비-0.""" + if threat.is_malicious: + return settings.score_total_cap return min( _preceding_score(threat, heuristic) + content.score, settings.score_total_cap, @@ -177,7 +174,7 @@ def _total_score( def _decide_verdict(score: int, threat: ThreatDbResult) -> Verdict: - """blacklist 매치는 점수와 무관하게 danger. 나머지는 점수 구간으로 매핑.""" + """Known malicious 매치는 danger. 나머지는 점수 구간으로 매핑.""" if threat.is_malicious: return Verdict.DANGER if score >= settings.score_danger_threshold: @@ -274,15 +271,6 @@ async def _run_stage_2_and_3( raise -async def _collect_upstream_after_stage_2_and_3( - stage_task: asyncio.Task[tuple[ThreatDbResult, DomainHeuristicResult, bool]], -) -> tuple[str, ...]: - threat, heuristic, short_circuited = await stage_task - if short_circuited: - return () - return _collect_upstream_signals(threat, heuristic) - - async def run_pipeline( analysis_id: str, original_url: str, @@ -316,22 +304,9 @@ async def run_pipeline( # 2·3단계는 둘 다 unchain.final_url 만 필요하고 서로 독립이라 병렬로 돈다. # threat_db 가 먼저 malicious 로 끝나면 verdict 가 이미 danger 로 확정이므로 # heuristic 을 cancel 하고 4단계까지 skip — 여기서 조기 종료가 일어난다. - stage_2_3_task = asyncio.create_task( - _run_stage_2_and_3(log, unchain.final_url, session, stage_timings) + threat, heuristic, short_circuited = await _run_stage_2_and_3( + log, unchain.final_url, session, stage_timings ) - upstream_task = asyncio.create_task(_collect_upstream_after_stage_2_and_3(stage_2_3_task)) - content_task = asyncio.create_task( - _timed_async_stage( - stage_timings, - PipelineStage.CONTENT_ANALYSIS, - _stage_content_analysis( - log, - unchain.final_url, - upstream_task, - ), - ) - ) - threat, heuristic, short_circuited = await stage_2_3_task if short_circuited: log.info( @@ -340,9 +315,6 @@ async def run_pipeline( gsb_threat=threat.gsb.is_threat, urlhaus_threat=threat.urlhaus.is_threat, ) - content_task.cancel() - with suppress(asyncio.CancelledError): - await content_task stage_started = time.perf_counter() content = skipped_already_danger(unchain.final_url) _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, stage_started) @@ -356,14 +328,16 @@ async def run_pipeline( reason=("threat_db_match" if threat.is_malicious else "already_danger"), preceding_score=preceding, ) - content_task.cancel() - with suppress(asyncio.CancelledError): - await content_task stage_started = time.perf_counter() content = skipped_already_danger(unchain.final_url) _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, stage_started) else: - content = await content_task + upstream = _collect_upstream_signals(threat, heuristic) + content = await _timed_async_stage( + stage_timings, + PipelineStage.CONTENT_ANALYSIS, + _stage_content_analysis(log, unchain.final_url, upstream), + ) score = _total_score(threat, heuristic, content) verdict = _decide_verdict(score, threat) diff --git a/app/services/unchainer/unchain.py b/app/services/unchainer/unchain.py index 425c67d..cf4c54b 100644 --- a/app/services/unchainer/unchain.py +++ b/app/services/unchainer/unchain.py @@ -97,6 +97,11 @@ def _https_variant(url: str) -> str | None: return urlunparse(parsed._replace(scheme="https")) +def _is_analyzable_html_response(resp: httpx.Response) -> bool: + content_type = resp.headers.get("content-type", "").lower() + return 200 <= resp.status_code < 400 and "text/html" in content_type + + async def _https_responds(client: httpx.AsyncClient, url: str, headers: dict[str, str]) -> bool: https_url = _https_variant(url) if https_url is None: @@ -104,16 +109,19 @@ async def _https_responds(client: httpx.AsyncClient, url: str, headers: dict[str safety_error = await _check_host_safety(https_url) if safety_error is not None: return False + resp: httpx.Response | None = None try: req = client.build_request("HEAD", https_url, headers=headers) resp = await asyncio.wait_for( client.send(req, stream=True), timeout=settings.schemeless_https_probe_timeout_seconds, ) - await resp.aclose() + return _is_analyzable_html_response(resp) except (TimeoutError, httpx.HTTPError): return False - return resp.status_code < 500 + finally: + if resp is not None: + await resp.aclose() async def unchain_url(url: str, *, prefer_https_when_schemeless: bool = False) -> UnchainResult: diff --git a/tests/services/content_analyzer/test_analyze.py b/tests/services/content_analyzer/test_analyze.py index ee2a3c5..8e165aa 100644 --- a/tests/services/content_analyzer/test_analyze.py +++ b/tests/services/content_analyzer/test_analyze.py @@ -9,7 +9,7 @@ from app.core.config import settings from app.schemas.content_analysis import AIVerdict, ContentSignal, TokenUsage from app.services.content_analyzer.ai import AIInference, AIPromptContext, NullAIProvider -from app.services.content_analyzer.analyze import analyze_content +from app.services.content_analyzer.analyze import analyze_content, skipped_already_danger from app.services.content_analyzer.fetch import FetchResult @@ -94,6 +94,14 @@ async def test_fetch_failure_skips_extract_and_ai(self) -> None: sig_mock.assert_not_called() ai_mock.assert_not_called() + def test_skipped_already_danger_has_fixed_user_reason(self) -> None: + result = skipped_already_danger("https://danger.test/") + + assert result.fetched is False + assert result.ai_reason is None + assert result.reason == "위험성이 확인된 URL입니다. 페이지를 열지 않는 것이 좋습니다." + assert result.error == "skipped_already_danger" + class TestBrandImpersonationEndToEnd: async def test_brand_impersonation_form_scored(self) -> None: diff --git a/tests/services/test_db_independent_pipeline.py b/tests/services/test_db_independent_pipeline.py index be98d20..6f2028b 100644 --- a/tests/services/test_db_independent_pipeline.py +++ b/tests/services/test_db_independent_pipeline.py @@ -4,7 +4,6 @@ import asyncio import inspect -import time from unittest.mock import AsyncMock, patch import pytest @@ -125,21 +124,16 @@ async def test_db_independent_pipeline_passes_url_and_redirect_signals_to_conten @pytest.mark.asyncio -async def test_db_independent_pipeline_runs_heuristic_and_content_concurrently() -> None: - final_url = "https://parallel.example.com/login" +async def test_db_independent_pipeline_skips_content_when_heuristic_is_danger() -> None: + final_url = "https://danger.example.com/login" content_started = asyncio.Event() - heuristic_started = asyncio.Event() async def _slow_heuristic(_: str) -> DomainHeuristicResult: - heuristic_started.set() - await content_started.wait() await asyncio.sleep(0.01) - return _make_heuristic(15) + return _make_heuristic(65) async def _slow_content(_: str, **__: object) -> ContentAnalysisResult: content_started.set() - await heuristic_started.wait() - await asyncio.sleep(0.01) return _make_content(final_url, score=20) with ( @@ -159,13 +153,44 @@ async def _slow_content(_: str, **__: object) -> ContentAnalysisResult: mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url) mock_unchain.return_value = _make_unchain(final_url) - started = time.perf_counter() result = await run_db_independent_pipeline("aid-parallel", final_url) assert isinstance(result, DbIndependentPipelineSuccess) - assert result.score == 35 - assert time.perf_counter() - started < 0.08 - mock_content.assert_awaited_once() + assert result.score == 65 + assert result.stages.content_analysis.error == "skipped_already_danger" + assert content_started.is_set() is False + mock_content.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_db_independent_pipeline_does_not_start_content_when_heuristic_fails() -> None: + final_url = "https://error.example.com/login" + + async def _failing_heuristic(_: str) -> DomainHeuristicResult: + await asyncio.sleep(0.01) + raise RuntimeError("heuristic failed") + + 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=AsyncMock(side_effect=_failing_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) + + with pytest.raises(RuntimeError, match="heuristic failed"): + await run_db_independent_pipeline("aid-error-cleanup", final_url) + + mock_content.assert_not_awaited() @pytest.mark.asyncio diff --git a/tests/services/test_pipeline.py b/tests/services/test_pipeline.py index d2cb34d..225ca19 100644 --- a/tests/services/test_pipeline.py +++ b/tests/services/test_pipeline.py @@ -300,7 +300,7 @@ async def test_score_capped_at_100(self, async_session: AsyncSession) -> None: async def test_danger_when_threat_matches_even_below_threshold( self, async_session: AsyncSession ) -> None: - """GSB 매치 시 합산 점수가 임계 미만이어도 verdict 는 강제로 danger.""" + """GSB 매치 시 known malicious 로 보고 score=100, verdict=danger.""" final_url = "https://blacklist.test/" # short-circuit 경로 — heuristic 은 placeholder, content skip with ( @@ -322,8 +322,7 @@ async def test_danger_when_threat_matches_even_below_threshold( result = await run_pipeline("aid-blk", final_url, async_session) assert isinstance(result, PipelineSuccess) - # GSB +50 만 합산되고 임계 61 미달이지만 is_malicious=True 라 verdict=danger - assert result.score == 50 + assert result.score == 100 assert result.verdict == Verdict.DANGER async def test_verdict_score_appear_before_stages_in_response( @@ -488,7 +487,7 @@ async def slow_heuristic(_url: str) -> DomainHeuristicResult: 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", side_effect=slow_heuristic), - patch("app.services.pipeline.analyze_content", new_callable=AsyncMock), + 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) @@ -497,6 +496,8 @@ async def slow_heuristic(_url: str) -> DomainHeuristicResult: result = await run_pipeline("aid-sc", final_url, async_session) assert isinstance(result, PipelineSuccess) + assert result.score == 100 + mock_content.assert_not_awaited() # heuristic 은 cancel 되어 본문이 끝까지 돌지 않았어야 한다 assert heuristic_finished.is_set() is False # 4단계 응답은 skip @@ -552,8 +553,7 @@ async def fast_heuristic(_url: str) -> DomainHeuristicResult: assert result.stages.domain_heuristic.score == 0 assert result.stages.domain_heuristic.rdap_error is None assert result.stages.domain_heuristic.skipped_reason == "threat_matched" - # GSB(+50) + heuristic placeholder(0) + content skip(0) = 50, verdict 는 DANGER 강제. - assert result.score == 50 + assert result.score == 100 assert result.verdict == Verdict.DANGER @@ -579,7 +579,7 @@ async def test_run_pipeline_short_circuits_on_urlhaus_match( patch( "app.services.pipeline.check_domain_heuristic", new_callable=AsyncMock ) as mock_heuristic, - patch("app.services.pipeline.analyze_content", new_callable=AsyncMock), + 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) @@ -589,6 +589,8 @@ async def test_run_pipeline_short_circuits_on_urlhaus_match( result = await run_pipeline("aid-urlhaus", final_url, async_session) assert isinstance(result, PipelineSuccess) + assert result.score == 100 + mock_content.assert_not_awaited() assert result.stages.content_analysis.error == "skipped_already_danger" @@ -606,7 +608,7 @@ async def test_run_pipeline_skips_content_when_heuristic_alone_exceeds_threshold patch( "app.services.pipeline.check_domain_heuristic", new_callable=AsyncMock ) as mock_heuristic, - patch("app.services.pipeline.analyze_content", new_callable=AsyncMock), + 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) @@ -616,4 +618,34 @@ async def test_run_pipeline_skips_content_when_heuristic_alone_exceeds_threshold result = await run_pipeline("aid-heur", final_url, async_session) assert isinstance(result, PipelineSuccess) + mock_content.assert_not_awaited() assert result.stages.content_analysis.error == "skipped_already_danger" + + +@pytest.mark.asyncio +async def test_run_pipeline_does_not_start_content_when_stage_2_3_fails( + async_session: AsyncSession, +) -> None: + final_url = "https://error.test/" + + async def failing_threat(_session: AsyncSession, _url: str) -> ThreatDbResult: + raise RuntimeError("threat db failed") + + 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", side_effect=failing_threat), + patch( + "app.services.pipeline.check_domain_heuristic", + new_callable=AsyncMock, + return_value=_heuristic_with_score(0), + ), + 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) + + with pytest.raises(RuntimeError, match="threat db failed"): + await run_pipeline("aid-error-cleanup", final_url, async_session) + + mock_content.assert_not_awaited() diff --git a/tests/services/unchainer/test_unchain.py b/tests/services/unchainer/test_unchain.py index 6389748..9d46c8e 100644 --- a/tests/services/unchainer/test_unchain.py +++ b/tests/services/unchainer/test_unchain.py @@ -132,6 +132,53 @@ async def test_multi_hop_chain(self) -> None: assert result.final_url == "https://a.com/final" assert result.hop_count == 4 + @pytest.mark.asyncio + async def test_schemeless_prefers_https_when_https_is_analyzable(self) -> None: + def _side_effect(method: str, url: str) -> httpx.Response: + assert method == "HEAD" + if url == "https://example.com/path": + return _make_response(200, {"content-type": "text/html; charset=utf-8"}) + raise AssertionError(f"unexpected request: {method} {url}") + + client = _mock_client(side_effect=_side_effect) + + with patch(_PATCH_TARGET, return_value=client): + result = await unchain_url( + "http://example.com/path", + prefer_https_when_schemeless=True, + ) + + assert result.final_url == "https://example.com/path" + assert "schemeless_https_upgrade" in result.signals + + @pytest.mark.asyncio + async def test_schemeless_falls_back_to_http_when_https_is_not_analyzable(self) -> None: + requests: list[tuple[str, str]] = [] + + def _side_effect(method: str, url: str) -> httpx.Response: + requests.append((method, url)) + if url == "https://example.com/path": + return _make_response(404) + if url == "http://example.com/path": + return _make_response(200) + raise AssertionError(f"unexpected request: {method} {url}") + + client = _mock_client(side_effect=_side_effect) + + with patch(_PATCH_TARGET, return_value=client): + result = await unchain_url( + "http://example.com/path", + prefer_https_when_schemeless=True, + ) + + assert result.final_url == "http://example.com/path" + assert result.hops[0].status_code == 200 + assert "schemeless_https_upgrade" not in result.signals + assert requests == [ + ("HEAD", "https://example.com/path"), + ("HEAD", "http://example.com/path"), + ] + class TestRedirectLoop: """무한 루프 감지."""