diff --git a/app/api/v1/endpoints/analyze.py b/app/api/v1/endpoints/analyze.py index 2186c49..bb5febc 100644 --- a/app/api/v1/endpoints/analyze.py +++ b/app/api/v1/endpoints/analyze.py @@ -100,6 +100,30 @@ async def analyze_sync( ) +@router.post( + "/analyze/no-ai/sync", + response_model=PipelineSuccess | PipelineFailure, + summary="전체 파이프라인 — AI 비사용 동기 결과 반환", + description=( + "run_pipeline 을 AI 추론 없이 동기로 실행합니다. normalize, unchain, threat DB, " + "도메인 휴리스틱, 콘텐츠 정적 분석은 수행하지만 OpenAI provider 를 호출하지 않고 " + "AI verdict/score 가산 없이 규칙 기반 verdict 와 summary 를 반환합니다." + ), +) +async def analyze_no_ai_sync( + body: AnalyzeSyncRequest, + session: DBSession, + _: InternalApiKey, +) -> PipelineSuccess | PipelineFailure: + analysis_id = str(uuid.uuid4()) + return await run_pipeline( + analysis_id=analysis_id, + original_url=body.url, + session=session, + use_ai=False, + ) + + @router.post( "/analyze/db-independent/sync", response_model=DbIndependentPipelineSuccess | DbIndependentPipelineFailure, diff --git a/app/schemas/pipeline.py b/app/schemas/pipeline.py index 0d2ff7b..6e5d9c4 100644 --- a/app/schemas/pipeline.py +++ b/app/schemas/pipeline.py @@ -68,6 +68,7 @@ class PipelineSuccess(BaseModel): final_url: str verdict: Verdict score: int = Field(ge=0, le=100) + summary: str | None = None timings: PipelineTimings | None = None stages: PipelineStages diff --git a/app/services/analysis_callback.py b/app/services/analysis_callback.py index 41e7956..5fa47cf 100644 --- a/app/services/analysis_callback.py +++ b/app/services/analysis_callback.py @@ -9,13 +9,13 @@ from app.core.config import settings from app.core.logging import get_logger -from app.schemas.pipeline import PipelineFailure, PipelineStage, PipelineSuccess, Verdict +from app.schemas.pipeline import PipelineFailure, PipelineStage, PipelineSuccess +from app.services.analysis_summary import build_analysis_summary logger = get_logger(__name__) _CALLBACK_PATH = "/internal/analysis-result" _MAX_ATTEMPTS = 3 -_KNOWN_MALICIOUS_SUMMARY = "악성으로 알려진 페이지 입니다." _STAGE_NUMBERS: dict[PipelineStage, int] = { PipelineStage.NORMALIZE: 1, @@ -191,8 +191,7 @@ def _callback_reasons(result: PipelineSuccess) -> list[dict[str, Any]]: stage=4, weight=settings.score_weight_ai_phishing, message=( - stages.content_analysis.ai_reason - or "AI가 피싱 가능성이 높다고 판단했습니다." + stages.content_analysis.ai_reason or "AI가 피싱 가능성이 높다고 판단했습니다." ), ) elif verdict == "suspicious": @@ -202,8 +201,7 @@ def _callback_reasons(result: PipelineSuccess) -> list[dict[str, Any]]: stage=4, weight=settings.score_weight_ai_suspicious, message=( - stages.content_analysis.ai_reason - or "AI가 의심스러운 페이지로 판단했습니다." + stages.content_analysis.ai_reason or "AI가 의심스러운 페이지로 판단했습니다." ), ) @@ -211,27 +209,19 @@ def _callback_reasons(result: PipelineSuccess) -> list[dict[str, Any]]: def _callback_summary(result: PipelineSuccess) -> str: - if result.stages.threat_db.is_malicious: - return _KNOWN_MALICIOUS_SUMMARY - content = result.stages.content_analysis - if content.ai_reason: - return content.ai_reason - if content.reason: - return content.reason - if result.verdict == Verdict.DANGER: - return "위험성이 높은 페이지입니다. 접속과 정보 입력을 피하세요." - if result.verdict == Verdict.CAUTION: - return "주의가 필요한 페이지입니다. 링크와 입력 정보를 한 번 더 확인하세요." - return "현재 분석 기준에서 뚜렷한 위험 신호가 확인되지 않았습니다." + if result.summary: + return result.summary + return build_analysis_summary( + verdict=result.verdict, + threat=result.stages.threat_db, + heuristic=result.stages.domain_heuristic, + content=result.stages.content_analysis, + ) def _spring_stages(result: PipelineSuccess) -> dict[str, Any]: stages = result.stages - gsb_matches = [ - match.threat_type - for match in stages.threat_db.gsb.matches - if match.threat_type - ] + gsb_matches = [match.threat_type for match in stages.threat_db.gsb.matches if match.threat_type] chain = [hop.url for hop in stages.unchain.hops] if not chain: diff --git a/app/services/analysis_summary.py b/app/services/analysis_summary.py new file mode 100644 index 0000000..6eccb28 --- /dev/null +++ b/app/services/analysis_summary.py @@ -0,0 +1,68 @@ +from __future__ import annotations + +from typing import Any + +from app.schemas.content_analysis import ContentAnalysisResult +from app.schemas.domain_heuristic import DomainHeuristicResult +from app.schemas.threat_db import ThreatDbResult + +KNOWN_MALICIOUS_SUMMARY = "악성으로 알려진 페이지 입니다." +SAFE_SUMMARY = "현재 분석 기준에서 뚜렷한 위험 신호가 확인되지 않았습니다." +CAUTION_SUMMARY = "주의가 필요한 페이지입니다. 링크와 입력 정보를 한 번 더 확인하세요." +DANGER_SUMMARY = "위험성이 높은 페이지입니다. 접속과 정보 입력을 피하세요." + + +def _code(value: Any) -> str: + return value.value if hasattr(value, "value") else str(value) + + +def build_analysis_summary( + *, + verdict: Any, + threat: ThreatDbResult, + heuristic: DomainHeuristicResult, + content: ContentAnalysisResult, +) -> str: + if threat.is_malicious: + return KNOWN_MALICIOUS_SUMMARY + if content.ai_reason: + return content.ai_reason + + domain_signals = {_code(signal) for signal in heuristic.signals} + content_signals = {_code(signal) for signal in content.signals} + + if "FETCH_FAILED" in content_signals: + return content.reason or "페이지를 확인할 수 없어 안전 여부를 판단하기 어렵습니다." + if "SKIPPED_ALREADY_DANGER" in content_signals: + return content.reason or DANGER_SUMMARY + + if "BRAND_IMPERSONATION_FORM" in content_signals and ( + content.has_password_field or "SENSITIVE_PATH" in domain_signals + ): + return "브랜드 사칭과 로그인 정보 입력 유도 정황이 있어 비밀번호를 입력하지 마세요." + if "CREDENTIAL_FORM_EXTERNAL" in content_signals: + return "로그인 정보가 외부 주소로 전송될 수 있어 계정 정보를 입력하지 마세요." + if "BRAND_IMPERSONATION_FORM" in content_signals or "LOGO_ALT_IMPERSONATION" in content_signals: + return "브랜드 사칭 정황이 있어 링크와 입력 정보를 신중히 확인하세요." + if {"PII_COLLECTION_FORM", "SENSITIVE_ID_FIELD", "FINANCIAL_FIELD"} & content_signals: + return "민감정보 입력을 유도하는 정황이 있어 개인정보나 결제 정보를 입력하지 마세요." + if "RISKY_DOWNLOAD_LINK" in content_signals: + return "위험한 파일 다운로드를 유도할 수 있어 파일을 열지 마세요." + if {"PUBLIC_AGENCY_LURE", "KOREAN_LURE_TEXT"} & content_signals: + return "국내 사용자 또는 공공기관을 사칭하는 유인 문구가 있어 주의하세요." + + if "TYPO_DOMAIN" in domain_signals: + return "유명 서비스와 유사한 도메인이므로 로그인이나 결제 정보를 입력하지 마세요." + if {"URL_USERINFO", "OPEN_REDIRECT_PARAM", "REDIRECT_CROSS_ORIGIN"} & domain_signals: + return "URL이 다른 사이트 이동이나 위장에 악용될 수 있어 접속 전 주소를 확인하세요." + if {"NEW_DOMAIN", "SUSPICIOUS_TLD", "DGA_LIKE"} & domain_signals: + return "신뢰도가 낮은 도메인 특성이 있어 링크와 입력 정보를 한 번 더 확인하세요." + + if content.reason: + return content.reason + verdict_code = _code(verdict) + if verdict_code == "danger": + return DANGER_SUMMARY + if verdict_code == "caution": + return CAUTION_SUMMARY + return SAFE_SUMMARY diff --git a/app/services/content_analyzer/ai.py b/app/services/content_analyzer/ai.py index 33726c0..65074b2 100644 --- a/app/services/content_analyzer/ai.py +++ b/app/services/content_analyzer/ai.py @@ -55,6 +55,12 @@ class AIProvider(Protocol): async def infer(self, ctx: AIPromptContext) -> AIInference | None: ... +class AIProviderError(Exception): + def __init__(self, code: str, message: str) -> None: + super().__init__(message) + self.code = code + + class NullAIProvider: """모델이 설정되지 않았을 때의 기본 구현. 추론을 시도하지 않고 None 반환. diff --git a/app/services/content_analyzer/analyze.py b/app/services/content_analyzer/analyze.py index f4ca63a..339c8f4 100644 --- a/app/services/content_analyzer/analyze.py +++ b/app/services/content_analyzer/analyze.py @@ -19,7 +19,12 @@ ContentSignal, TokenUsage, ) -from app.services.content_analyzer.ai import AIPromptContext, AIProvider, get_ai_provider +from app.services.content_analyzer.ai import ( + AIPromptContext, + AIProvider, + AIProviderError, + get_ai_provider, +) from app.services.content_analyzer.extract import ExtractedFeatures, extract_features_async from app.services.content_analyzer.fetch import fetch_page from app.services.content_analyzer.render import render_page @@ -114,8 +119,7 @@ def _merge_features(static: ExtractedFeatures, rendered: ExtractedFeatures) -> E title=rendered.title or static.title, has_password_field=static.has_password_field or rendered.has_password_field, has_password_form_external_action=( - static.has_password_form_external_action - or rendered.has_password_form_external_action + static.has_password_form_external_action or rendered.has_password_form_external_action ), has_meta_refresh=static.has_meta_refresh or rendered.has_meta_refresh, has_external_meta_refresh=( @@ -246,6 +250,7 @@ async def analyze_content( *, provider: AIProvider | None = None, upstream_signals: Iterable[str] | Awaitable[Iterable[str]] = (), + use_ai: bool = True, ) -> ContentAnalysisResult: """콘텐츠 정적 분석 진입점. @@ -289,27 +294,38 @@ async def analyze_content( ai_model: str | None = None ai_token_usage: TokenUsage | None = None - active_provider = provider if provider is not None else get_ai_provider() 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) - ) - except asyncio.CancelledError: - # shutdown / 요청 타임아웃 신호는 degraded 결과로 흡수하지 않는다 - raise - except Exception as exc: - logger.warning( - "content_analysis.ai_error", - url=final_url, - error=str(exc), - error_type=type(exc).__name__, - ) - ai_error = "ai_unavailable" - inference = None + active_provider: AIProvider | None = None + inference = None + if use_ai: + active_provider = provider if provider is not None else get_ai_provider() + try: + inference = await active_provider.infer( + _build_prompt_context(final_url, features, upstream_tuple) + ) + except asyncio.CancelledError: + # shutdown / 요청 타임아웃 신호는 degraded 결과로 흡수하지 않는다 + raise + except AIProviderError as exc: + logger.warning( + "content_analysis.ai_error", + url=final_url, + error=str(exc), + error_code=exc.code, + error_type=type(exc).__name__, + ) + ai_error = exc.code + except Exception as exc: + logger.warning( + "content_analysis.ai_error", + url=final_url, + error=str(exc), + error_type=type(exc).__name__, + ) + ai_error = "ai_unavailable" score = scoring.score if inference is not None: @@ -323,7 +339,7 @@ async def analyze_content( upstream_tuple, ): score += _ai_score_weight(inference.verdict) - elif ai_error is None: + elif use_ai and ai_error is None and active_provider is not None: # 추론이 None 인데 호출 단계 예외도 없었다면 NullAIProvider 동작. # 부팅 시 misconfiguration 으로 폴백된 NullProvider 면 fallback_reason 을 응답에 노출. # 정상 NullProvider 는 fallback_reason=None 이라 ai_error 가 None 으로 유지된다. diff --git a/app/services/pipeline.py b/app/services/pipeline.py index ec85997..2e4b10f 100644 --- a/app/services/pipeline.py +++ b/app/services/pipeline.py @@ -32,6 +32,7 @@ ) from app.schemas.threat_db import ThreatDbResult from app.schemas.unchain import UnchainResult +from app.services.analysis_summary import build_analysis_summary 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 @@ -196,14 +197,17 @@ async def _stage_content_analysis( log: structlog.stdlib.BoundLogger, final_url: str, upstream_signals: tuple[str, ...], + *, + use_ai: bool, ) -> ContentAnalysisResult: - result = await analyze_content(final_url, upstream_signals=upstream_signals) + result = await analyze_content(final_url, upstream_signals=upstream_signals, use_ai=use_ai) 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, + use_ai=use_ai, upstream_signals=list(upstream_signals), ) return result @@ -289,6 +293,44 @@ def _page_unavailable_failure( ) +def _pipeline_success( + *, + analysis_id: str, + original_url: str, + final_url: str, + verdict: Verdict, + score: int, + started: float, + stage_timings: PipelineStageTimings, + normalize: NormalizeResult, + unchain: UnchainResult, + threat: ThreatDbResult, + heuristic: DomainHeuristicResult, + content: ContentAnalysisResult, +) -> PipelineSuccess: + return PipelineSuccess( + analysis_id=analysis_id, + original_url=original_url, + final_url=final_url, + verdict=verdict, + score=score, + summary=build_analysis_summary( + verdict=verdict, + threat=threat, + heuristic=heuristic, + content=content, + ), + timings=_build_timings(started, stage_timings), + stages=PipelineStages( + normalize=normalize, + unchain=unchain, + threat_db=threat, + domain_heuristic=heuristic, + content_analysis=content, + ), + ) + + async def _run_stage_2_and_3( log: structlog.stdlib.BoundLogger, final_url: str, @@ -359,6 +401,8 @@ async def run_pipeline( analysis_id: str, original_url: str, session: AsyncSession, + *, + use_ai: bool = True, ) -> PipelineSuccess | PipelineFailure: log = logger.bind(analysis_id=analysis_id) log.info("pipeline.start", url=original_url) @@ -455,20 +499,19 @@ async def run_pipeline( score = _total_score(threat, heuristic, content) if threat.is_malicious or score >= settings.score_caution_threshold: verdict = _decide_verdict(score, threat) - return PipelineSuccess( + return _pipeline_success( analysis_id=analysis_id, original_url=original_url, final_url=unchain.final_url, verdict=verdict, score=score, - timings=_build_timings(total_started, stage_timings), - stages=PipelineStages( - normalize=norm, - unchain=unchain, - threat_db=threat, - domain_heuristic=heuristic, - content_analysis=content, - ), + started=total_started, + stage_timings=stage_timings, + normalize=norm, + unchain=unchain, + threat=threat, + heuristic=heuristic, + content=content, ) return _page_unavailable_failure( analysis_id=analysis_id, @@ -546,7 +589,12 @@ async def run_pipeline( _timed_async_stage( stage_timings, PipelineStage.CONTENT_ANALYSIS, - _stage_content_analysis(log, unchain.final_url, upstream), + _stage_content_analysis( + log, + unchain.final_url, + upstream, + use_ai=use_ai, + ), ), settings.pipeline_content_timeout_seconds, ) @@ -574,20 +622,19 @@ async def run_pipeline( score = _total_score(threat, heuristic, content) if threat.is_malicious or score >= settings.score_caution_threshold: verdict = _decide_verdict(score, threat) - return PipelineSuccess( + return _pipeline_success( analysis_id=analysis_id, original_url=original_url, final_url=unchain.final_url, verdict=verdict, score=score, - timings=_build_timings(total_started, stage_timings), - stages=PipelineStages( - normalize=norm, - unchain=unchain, - threat_db=threat, - domain_heuristic=heuristic, - content_analysis=content, - ), + started=total_started, + stage_timings=stage_timings, + normalize=norm, + unchain=unchain, + threat=threat, + heuristic=heuristic, + content=content, ) return _page_unavailable_failure( analysis_id=analysis_id, @@ -608,18 +655,17 @@ async def run_pipeline( verdict=verdict.value, score=score, ) - return PipelineSuccess( + return _pipeline_success( analysis_id=analysis_id, original_url=original_url, final_url=unchain.final_url, verdict=verdict, score=score, - timings=_build_timings(total_started, stage_timings), - stages=PipelineStages( - normalize=norm, - unchain=unchain, - threat_db=threat, - domain_heuristic=heuristic, - content_analysis=content, - ), + started=total_started, + stage_timings=stage_timings, + normalize=norm, + unchain=unchain, + threat=threat, + heuristic=heuristic, + content=content, )