From 8384f7609f0fdebcc811f239f897b530cffe7ac7 Mon Sep 17 00:00:00 2001 From: pearseona Date: Mon, 20 Jul 2026 16:03:43 +0900 Subject: [PATCH 1/8] feat: extend Pydantic schemas for final score and contribution breakdown --- app/dto/schemas.py | 51 +++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 50 insertions(+), 1 deletion(-) diff --git a/app/dto/schemas.py b/app/dto/schemas.py index 3d47a99..9a2c97a 100644 --- a/app/dto/schemas.py +++ b/app/dto/schemas.py @@ -1,6 +1,6 @@ +from enum import Enum from pydantic import BaseModel, Field from typing import Optional -from app.core.config import settings # Spring Boot Gateway에서 Python FastAPI로 검사를 요청할 때의 바디 규격 class URLScanRequest(BaseModel): @@ -29,3 +29,52 @@ class Config: "error_message": None } } + +# 3중 가중치 시스템 스펙 정의 +class RiskGrade(str, Enum): + HIGH = "HIGH" + MEDIUM = "MEDIUM" + LOW = "LOW" + +class ContributionBreakdown(BaseModel): + llm: int = Field(..., description="LLM 문맥 분석 기여 점수 (0~50)", ge=0, le=50) + virus_total: int = Field(..., description="VirusTotal 엔진 기여 점수 (0~30)", ge=0, le=30) + rules: int = Field(..., description="로컬 가드 규칙 기반 기여 점수 (0~20)", ge=0, le=20) + +class SmishingAnalysisResponse(BaseModel): + status: str = Field(..., description="응답 상태 (SUCCESS / ERROR)") + message: str = Field(..., description="응답 메시지 설명") + + # 핵심 합성 스코어 필드 + final_score: int = Field(..., description="3중 가중치 합성 최종 위험 점수 (0~100)", ge=0, le=100) + risk_grade: RiskGrade = Field(..., description="최종 점수 기반 위험 등급 분류 (HIGH/MEDIUM/LOW)") + contribution_breakdown: ContributionBreakdown = Field(..., description="3개 레이어별 점수 기여도 명세") + + # 세부 분석 트랙 데이터 + text_analysis: Optional[dict] = Field(None, description="LLM 실시간 문맥 분석 상세 결과") + url_analysis: Optional[dict] = Field(None, description="하이브리드 URL 보안 엔진 상세 분석 결과") + + class Config: + use_enum_values = True + json_schema_extra = { + "example": { + "status": "SUCCESS", + "message": "통합 스미싱 분석이 완료되었습니다.", + "final_score": 95, + "risk_grade": "HIGH", + "contribution_breakdown": { + "llm": 45, + "virus_total": 30, + "rules": 20 + }, + "text_analysis": { + "risk_score": 90, + "reason": "지인을 사칭한 금전 요구 문맥 감지" + }, + "url_analysis": { + "has_url": True, + "is_url_malicious": True, + "url_risk_score": 0.95 + } + } + } \ No newline at end of file From 7991f7d47dc414df8e97a31fcca0dff777ebdade Mon Sep 17 00:00:00 2001 From: pearseona Date: Mon, 20 Jul 2026 16:35:47 +0900 Subject: [PATCH 2/8] feat: implement triple-layered weighted scoring algorithm and risk grading --- app/utils/scoring_engine | 0 1 file changed, 0 insertions(+), 0 deletions(-) create mode 100644 app/utils/scoring_engine diff --git a/app/utils/scoring_engine b/app/utils/scoring_engine new file mode 100644 index 0000000..e69de29 From 7970a561419baf3afd6533ef215244c5084a7518 Mon Sep 17 00:00:00 2001 From: pearseona Date: Mon, 20 Jul 2026 16:37:59 +0900 Subject: [PATCH 3/8] feat: implement triple-layered weighted scoring algorithm and risk grading --- app/dto/schemas.py | 2 +- app/utils/scoring_engine | 60 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 1 deletion(-) diff --git a/app/dto/schemas.py b/app/dto/schemas.py index 9a2c97a..e3d0975 100644 --- a/app/dto/schemas.py +++ b/app/dto/schemas.py @@ -38,7 +38,7 @@ class RiskGrade(str, Enum): class ContributionBreakdown(BaseModel): llm: int = Field(..., description="LLM 문맥 분석 기여 점수 (0~50)", ge=0, le=50) - virus_total: int = Field(..., description="VirusTotal 엔진 기여 점수 (0~30)", ge=0, le=30) + hybrid_url: int = Field(..., description="하이브리드 URL 보안 엔진 기여 점수 (0~30)", ge=0, le=30) rules: int = Field(..., description="로컬 가드 규칙 기반 기여 점수 (0~20)", ge=0, le=20) class SmishingAnalysisResponse(BaseModel): diff --git a/app/utils/scoring_engine b/app/utils/scoring_engine index e69de29..60726b6 100644 --- a/app/utils/scoring_engine +++ b/app/utils/scoring_engine @@ -0,0 +1,60 @@ +import logging +from typing import Tuple, Dict, Any +from app.dto.schemas import RiskGrade, ContributionBreakdown + +logger = logging.getLogger(__name__) + +class scoringEngine: + """ + 3중 가중치 스코어링 시스템 + - LLM 문맥 분석 (50%) + - 하이브리드 URL 보안 엔진 [GSB + VT 백업] (30%) + - 로컬 가드 규칙 기반 패널티 (20%) + """ + + @staticmethod + def calculate_score( + llm_score: int, # LLM이 반환한 위험도 점수 (0~100) + is_url_maliciout: bool, # 하이브리드 URL 엔진의 최종 악성 판정 여부 + url_risk_score: float, # 하이브리드 URL 엔진이 계산한 위험도 점수 (0.0~1.0) + has_rule_violation: bool # 로컬 패널티 룰(.ru 등) 적발 여부 + ) -> Tuple[int, RiskGrade, ContributionBreakdown]: + + # 텍스트 문맥 점수와 하이브리드 URL 엔진의 결과값을 결합하여 최종 위험도를 산출 + + # 1. LLM 문맥 분석 기여 점수 (최대 50점) + llm_contrib = round(llm_score * 0.5) + + # 2. 하이브리드 URL 보안 엔진 기여 점수 (최대 30점) + if is_url_malicious or url_risk_score > 0: + url_contrib = round(url_risk_score * 30) + url_contrib = min(max(url_contrib, 0), 30) + else: + url_contrib = 0 + + # 3. 로컬 가드 규칙 기반 기여 점수 (최대 20점) + rules_contrib = 20 if has_rule_violation else 0 + + # 최종 위험도 점수 합산 + final_score = min(llm_contrib + url_contrib + rules_contrib, 100) + + # 최종 점수 기반 임계치 등급 분기 + if final_score >= 70: + risk_grade = RiskGrade.HIGH + elif final_score >= 40: + risk_grade = RiskGrade.MEDIUM + else: + risk_grade = RiskGrade.LOW + + logger.ingo( + f"[Scoring Engine] 통합 연산 완료 -> 최종 점수: {final_score} | 등급: {risk_grade} " + f"(LLM: {llm_contrib}, Hybrid-URL: {url_contrib}, Rules: {rules_contrib})" + ) + + breakdown = ContributionBreakdown( + llm=llm_contrib, + hybrid_url=url_contrib, + rules=rules_contrib + ) + + return final_score, risk_grade, breakdown \ No newline at end of file From facfa6c09e545eb9cc760b6eb9e347c1ca483093 Mon Sep 17 00:00:00 2001 From: pearseona Date: Mon, 20 Jul 2026 16:45:22 +0900 Subject: [PATCH 4/8] feat: create unified mobile message analysis endpoint --- app/router/analyze.py | 58 +++++++++++++++++++++++++++++++------------ 1 file changed, 42 insertions(+), 16 deletions(-) diff --git a/app/router/analyze.py b/app/router/analyze.py index 7b62eb6..3706a8e 100644 --- a/app/router/analyze.py +++ b/app/router/analyze.py @@ -1,3 +1,4 @@ +import logging from fastapi import APIRouter, Depends, HTTPException, status from app.dto.response import ApiResponse from app.dto.request import AnalyzeRequest @@ -5,6 +6,8 @@ from app.service.security.gemini_text_analyzer import analyze_text_with_gemini from app.service.scan_service import ScanService +logger = logging.getLogger(__name__) + # 문자 분석 전용 라우터 생성 router = APIRouter(prefix="/analyze", tags=["Analyze"]) @@ -13,37 +16,60 @@ def get_scan_service() -> ScanService: return ScanService() # 통합 스미싱 탐지 API -@router.post("", response_model=ApiResponse[dict], status_code=status.HTTP_200_OK) +@router.post( + "", + response_model=ApiResponse[dict], + status_code=status.HTTP_200_OK, + summary="[메인 통합 엔진] 문자 본문 기반 3중 스미싱 통합 분석", + description="문자 본문 전체를 분석하여 LLM 문맥, 하이브리드 URL 검사, 로컬 규칙을 합성한 0~100점 점수를 반환합니다." + ) + async def analyze_smishing( payload: AnalyzeRequest, scan_service: ScanService = Depends(get_scan_service) - ): + ) -> SmishingAnalysisResponse: try: - text_analysis = await analyze_text_with_gemini(payload.message) + logger.info(f"[Router] 통합 스미싱 분석 요청 접수: {payload.message[:15]}...") - url_scan_result: URLScanResponse = await scan_service.scan_message_text(payload.message) + # 파이프라인 호출 + text_analysis = await analyze_text_with_gemini(payload.message) + url_scan_result = await scan_service.scan_message_text(payload.message) + # 하이브리드 url 엔진 결과 규격 real_url_analysis = { "has_url": url_scan_result.has_url, "is_shortened": url_scan_result.original_url != url_scan_result.traced_url if url_scan_result.has_url else False, - "origin_url": url_scan_result.traced_url, # 최종 목적지 URL - "original_url": url_scan_result.original_url, # 최초 추출 URL + "origin_url": url_scan_result.traced_url, + "original_url": url_scan_result.original_url, "is_url_malicious": url_scan_result.is_url_malicious, "url_risk_score": url_scan_result.url_risk_score, "engine_source": url_scan_result.engine_source, "error_message": url_scan_result.error_message } - is_smishing_detected = (text_analysis.get("result", {}).get("grade") != "SAFE") or url_scan_result.is_url_malicious - - result = { - "smishing_detected": is_smishing_detected, - "text_analysis": text_analysis, - "url_analysis": real_url_analysis - } - - return ApiResponse.success(data=result, message="문자 분석이 완료되었습니다.") + return SmishingAnalysisResponse( + status="SUCCESS", + message="통합 스미싱 분석이 정상 수행되었습니다.", + final_score=0, + risk_grade=RiskGrade.LOW, + contribution_breakdown=ContributionBreakdown( + llm=0, + hybrid_url=0, + rules=0 + ), + text_analysis=text_analysis, + url_analysis=real_url_analysis + ) except Exception as e: - return ApiResponse.error(message=f"통합 스미싱 탐지 중 서버 에러가 발생했습니다: {str(e)}") + logger.error(f"[Router Error] 통합 분석 중 예외 발생: {str(e)}") + return SmishingAnalysisResponse( + status="ERROR", + message=f"통합 스미싱 탐지 중 서버 에러가 발생했습니다: {str(e)}", + final_score=0, + risk_grade=RiskGrade.LOW, + contribution_breakdown=ContributionBreakdown(llm=0, hybrid_url=0, rules=0), + text_analysis=None, + url_analysis=None + ) \ No newline at end of file From aefd2b506e24116da4265c4dbbc21db14cec5e2d Mon Sep 17 00:00:00 2001 From: pearseona Date: Mon, 20 Jul 2026 16:58:20 +0900 Subject: [PATCH 5/8] feat: extract hybrid url engine and orchestrate pipelines asynchronously in ScanService --- app/service/scan_service.py | 188 ++++++++-------------- app/service/security/hybrid_url_engine.py | 67 ++++++++ 2 files changed, 131 insertions(+), 124 deletions(-) create mode 100644 app/service/security/hybrid_url_engine.py diff --git a/app/service/scan_service.py b/app/service/scan_service.py index 211b229..03ab69b 100644 --- a/app/service/scan_service.py +++ b/app/service/scan_service.py @@ -1,145 +1,85 @@ -import os import logging import asyncio -from app.dto.schemas import URLScanResponse +from app.dto.schemas import SmishingAnalysisResponse, URLScanResponse, RiskGrade, ContributionBreakdown from app.utils.url_tracker import extract_urls, trace_url - -from app.service.security.virustotal import VirusTotalEngine -from app.service.security.google_safe_browsing import GoogleSafeBrowsingEngine +from app.utils.scoring_engine import ScoringEngine +from app.service.security.gemini_text_analyzer import analyze_text_with_gemini +from app.service.security.hybrid_url_engine import HybridUrlEngine logger = logging.getLogger(__name__) -MOCK_ENABLED = os.getenv("MOCK_SECURITY_API", "False").lower() in ("true", "1", "t") - class ScanService: def __init__(self): - self.vt_engine = VirusTotalEngine() - self.gsb_engine = GoogleSafeBrowsingEngine() + self.hybrid_url_engine = HybridUrlEngine() - async def scan_message_text(self, message: str) -> URLScanResponse: - + # 텍스트 트랙과 URL 트랙을 병렬 조립하고 스코어링을 매핑 + async def analyze_pipeline(self, message: str) -> SmishingAnalysisResponse: + try: - # 본문 텍스트에서 URL 정규식 추출 urls = extract_urls(message) - - if not urls: - logger.info(" 본문 내에 추출된 URL이 없어 안전한 상태로 판정합니다.") - return URLScanResponse( - has_url=False, - original_url=None, - traced_url=None, - is_url_malicious=False, - url_risk_score=0.0, - engine_source="Pre-Processing-Filter", - error_message=None - ) - - # URL이 존재하면 첫 번째 URL을 추출하여 단축 URL 리다이렉트 비동기 추적 - original_url = urls[0] - logger.info(f" URL 추출 완료: {original_url} -> 리다이렉트 추적 시작") - - traced_url = await trace_url(original_url) - logger.info(f" 최종 도달 URL 추적 완료: {traced_url}") - - if MOCK_ENABLED: - logger.info(f"[MOCK MODE] 목 데이터를 로드합니다. Target: {traced_url}") - - gsb_malicious = False - vt_malicious_count = 4 - - is_malicious = gsb_malicious or (vt_malicious_count > 0) - risk_score = 0.95 if gsb_malicious else min(0.1 + (vt_malicious_count * 0.1), 1.0) if is_malicious else 0.0 + has_url = len(urls) > 0 - return URLScanResponse( - has_url=True, - original_url=original_url, - traced_url=traced_url, - is_url_malicious=is_malicious, - url_risk_score=risk_score, - engine_source="Hybrid-Engine (MOCK)", - error_message=None - ) + # 비동기 Task 스케줄링 + text_task = asyncio.create_task(analyze_text_with_gemini(message)) + url_task = None - logger.info(f"[PROD MODE] 실제 VT 및 GSB API를 비동기 병렬 호출합니다.") - - # 비동기 병렬 태스크 스케줄링 - vt_task = asyncio.create_task(self.vt_engine.scan_url(traced_url)) - gsb_task = asyncio.create_task(self.gsb_engine.scan_url(traced_url)) - - try: - results = await asyncio.wait_for( - asyncio.gather(vt_task, gsb_task, return_exceptions=True), - timeout=6.0 - ) - vt_result, gsb_result = results - except asyncio.TimeoutException: - logger.error("[Pipeline Timeout] 외부 보안 API 응답 시간 초과로 하이브리드 스캔이 강제 타임아웃 처리되었습니다.") - vt_result = {"error": "Timeout"} - gsb_result = {"error": "Timeout"} + if has_url: + original_url = urls[0] + async def url_track(): + traced = await trace_url(original_url) + res = await self.hybrid_url_engine.scan_url(traced) + return traced, res + url_task = asyncio.create_task(url_track()) - print("\n" + "="*60) - print(f" [VIRUSTOTAL RAW RESPONSE]: {vt_result}") - print(f" [SAFE BROWSING RAW RESPONSE]: {gsb_result}") - print("="*60 + "\n") - - # 엔진별 결과값 예외 캡처 및 복구 정책 - error_logs = [] - if isinstance(vt_result, Exception) or "error" in str(vt_result): - err_msg = str(vt_result) if isinstance(vt_result, Exception) else vt_result.get("error") - logger.error(f"[Pipeline Error] VirusTotal 엔진 통신 실패: {err_msg}") - vt_result = {"is_malicious": False, "detected_count": 0} - error_logs.append(f"VT Fail ({err_msg[:15]})") - - if isinstance(gsb_result, Exception) or "error" in str(gsb_result): - err_msg = str(gsb_result) if isinstance(gsb_result, Exception) else gsb_result.get("error") - logger.error(f"[Pipeline Error] Google Safe Browsing 엔진 통신 실패: {err_msg}") - gsb_result = {"is_malicious": False} - error_logs.append(f"GSB Fail ({err_msg[:15]})") - - # 공통 dict 규격에서 안전하게 결과 파싱 - vt_malicious_count = vt_result.get("detected_count", 0) - is_gsb_blocked = gsb_result.get("is_malicious", False) - - # 일차적인 외부 인프라 스캔 결과 취합 - is_malicious = is_gsb_blocked or (vt_malicious_count >= 1) or vt_result.get("is_malicious", False) - - # 기본 위험도 점수 계산 레이어 우선 적용 - if is_gsb_blocked: - risk_score = gsb_result.get("raw_score", 0.95) - if risk_score > 1.0: risk_score /= 100.0 - elif vt_malicious_count > 0: - base_score = vt_result.get("raw_score", 0.0) - if base_score > 1.0: base_score /= 100.0 - risk_score = max(base_score, min(0.1 + (vt_malicious_count * 0.15), 0.95)) + # 병렬 실행 공정 + if url_task: + text_analysis, (traced_url, hybrid_res) = await asyncio.gather(text_task, url_task) else: - risk_score = 0.0 + text_analysis = await text_task + traced_url, hybrid_res = None, {"is_malicious": False, "url_risk_score": 0.0, "source": "Pre-Processing-Filter", "error_message": None} - if not is_malicious and (".ru" in traced_url or "testsafebrowsing" in traced_url): - logger.warning("[Infrastructure Guard] 외부 API 응답 공백 감지 - 로컬 정밀 위협 룰셋에 의해 악성 URL로 강제 전환합니다.") - is_malicious = True - risk_score = 0.75 + llm_score = text_analysis.get("result", {}).get("risk_score", 0) if isinstance(text_analysis, dict) else 0 - # 파이프라인 에러 리포트 구성 - combined_error = " | ".join(error_logs) if error_logs else None + # 로컬 규칙 패널티 가드 + has_rule_violation = False + if has_url and traced_url and (".ru" in traced_url or "testsafebrowsing" in traced_url): + has_rule_violation = True + hybrid_res["is_malicious"] = True + hybrid_res["url_risk_score"] = max(hybrid_res["url_risk_score"], 0.75) - return URLScanResponse( - has_url=True, - original_url=original_url, - traced_url=traced_url, - is_url_malicious=is_malicious, - url_risk_score=round(risk_score, 2), - engine_source="Hybrid-Engine (VT+GSB)", - error_message=combined_error + # 3중 스코어링 최종 계산 + final_score, risk_grade, breakdown = ScoringEngine.calculate_score( + llm_score=int(llm_score), + is_url_malicious=hybrid_res["is_malicious"], + url_risk_score=hybrid_res["url_risk_score"], + has_rule_violation=has_rule_violation ) + return SmishingAnalysisResponse( + status="SUCCESS", + message="3중 가중치 결합 스미싱 통합 분석이 완료되었습니다.", + final_score=final_score, + risk_grade=risk_grade, + contribution_breakdown=breakdown, + text_analysis=text_analysis, + url_analysis={ + "has_url": has_url, + "is_shortened": original_url != traced_url if has_url else False, + "origin_url": traced_url, + "original_url": original_url if has_url else None, + "is_url_malicious": hybrid_res["is_malicious"], + "url_risk_score": hybrid_res["url_risk_score"], + "engine_source": hybrid_res["source"], + "error_message": hybrid_res["error_message"] + } + ) except Exception as e: - logger.error(f" 파이프라인 수행 중 예외 발생: {str(e)}") - return URLScanResponse( - has_url=True, - original_url=message[:20] + "...", - traced_url=None, - is_url_malicious=False, - url_risk_score=0.0, - engine_source="Hybrid-Engine-Failure", - error_message=str(e) - ) \ No newline at end of file + logger.error(f"파이프라인 에러: {str(e)}") + return SmishingAnalysisResponse(status="ERROR", message=str(e), final_score=0, risk_grade=RiskGrade.LOW, contribution_breakdown=ContributionBreakdown(llm=0, hybrid_url=0, rules=0)) + + async def scan_message_text(self, message: str) -> URLScanResponse: + urls = extract_urls(message) + if not urls: return URLScanResponse(has_url=False, original_url=None, traced_url=None, is_url_malicious=False, url_risk_score=0.0, engine_source="Pre-Processing-Filter") + traced_url = await trace_url(urls[0]) + res = await self.hybrid_url_engine.scan_url(traced_url) + return URLScanResponse(has_url=True, original_url=urls[0], traced_url=traced_url, is_url_malicious=res["is_malicious"], url_risk_score=res["url_risk_score"], engine_source=res["source"], error_message=res["error_message"]) \ No newline at end of file diff --git a/app/service/security/hybrid_url_engine.py b/app/service/security/hybrid_url_engine.py new file mode 100644 index 0000000..491768d --- /dev/null +++ b/app/service/security/hybrid_url_engine.py @@ -0,0 +1,67 @@ +import os +import logging +import asyncio +from app.service.security.virustotal import VirusTotalEngine +from app.service.security.google_safe_browsing import GoogleSafeBrowsingEngine + +logger = logging.getLogger(__name__) +MOCK_ENABLED = os.getenv("MOCK_SECURITY_API", "False").lower() in ("true", "1", "t") + +# Google Safe Browsing(1차)과 VirusTotal(2차 백업)을 제어하는 하이브리드 URL 분석 코어 엔진 +class HybridUrlEngine: + + def __init__(self): + self.vt_engine = VirusTotalEngine() + self.gsb_engine = GoogleSafeBrowsingEngine() + + async def scan_url(self, traced_url: str) -> dict: + logger.info(f"[MOCK MODE] 하이브리드 URL 스캔 -> Target: {traced_url}") + return {"is_malicious": True, "url_risk_score": 0.85, "source": "Hybrid-Engine (MOCK)", "detected_count": 4, "error_message": None} + + logger.info("[PROD MODE] 1차 방어선: Google Safe Browsing API 가동") + error_logs = [] + + try: + gsb_result = await self.gsb_engine.scan_url(traced_url) + is_gsb_blocked = gsb_result.get("is_malicious", False) + except Exception as e: + logger.error(f"GSB 통신 실패: {str(e)}") + gsb_result = {"is_malicious": False} + is_gsb_blocked = False + error_logs.append(f"GSB Fail ({str(e)[:15]})") + + vt_result = {"is_malicious": False, "detected_count": 0} + + # GSB 악성 확정 시 VT 생략 + if is_gsb_blocked: + logger.info(" GSB 악성 판정으로 VirusTotal 호출 생략 (Quota 절약)") + engine_source = "Hybrid-Engine (GSB)" + risk_score = gsb_result.get("raw_score", 0.95) + if risk_score > 1.0: risk_score /= 100.0 + else: + logger.info(" GSB 청정/불확실로 인한 2차 방어선 VirusTotal 백업 가동") + engine_source = "Hybrid-Engine (GSB+VT)" + try: + vt_result = await asyncio.wait_for(self.vt_engine.scan_url(traced_url), timeout=4.0) + except Exception as e: + logger.error(f"VirusTotal 통신 실패: {str(e)}") + vt_result = {"is_malicious": False, "detected_count": 0} + error_logs.append(f"VT Fail ({str(e)[:15]})") + + vt_malicious_count = vt_result.get("detected_count", 0) + if vt_malicious_count > 0: + base_score = vt_result.get("raw_score", 0.0) + if base_score > 1.0: base_score /= 100.0 + risk_score = max(base_score, min(0.1 + (vt_malicious_count * 0.15), 0.95)) + else: + risk_score = 0.0 + + is_final_malicious = is_gsb_blocked or vt_result.get("is_malicious", False) or vt_result.get("detected_count", 0) >= 1 + combined_error = " | ".join(error_logs) if error_logs else None + + return { + "is_malicious": is_final_malicious, + "url_risk_score": round(risk_score, 2), + "source": engine_source, + "error_message": combined_error + } From f01227b907f446160c2976bc3c08a3896b8304da Mon Sep 17 00:00:00 2001 From: pearseona Date: Mon, 20 Jul 2026 17:05:10 +0900 Subject: [PATCH 6/8] feat: connect unified smishing analysis endpoint to pipeline --- app/router/analyze.py | 56 ++++++------------------------------------- 1 file changed, 7 insertions(+), 49 deletions(-) diff --git a/app/router/analyze.py b/app/router/analyze.py index 3706a8e..fe82764 100644 --- a/app/router/analyze.py +++ b/app/router/analyze.py @@ -1,9 +1,7 @@ import logging -from fastapi import APIRouter, Depends, HTTPException, status -from app.dto.response import ApiResponse +from fastapi import APIRouter, Depends, status from app.dto.request import AnalyzeRequest -from app.dto.schemas import URLScanResponse -from app.service.security.gemini_text_analyzer import analyze_text_with_gemini +from app.dto.schemas import SmishingAnalysisResponse from app.service.scan_service import ScanService logger = logging.getLogger(__name__) @@ -18,58 +16,18 @@ def get_scan_service() -> ScanService: # 통합 스미싱 탐지 API @router.post( "", - response_model=ApiResponse[dict], + response_model=SmishingAnalysisResponse, status_code=status.HTTP_200_OK, summary="[메인 통합 엔진] 문자 본문 기반 3중 스미싱 통합 분석", description="문자 본문 전체를 분석하여 LLM 문맥, 하이브리드 URL 검사, 로컬 규칙을 합성한 0~100점 점수를 반환합니다." - ) - +) async def analyze_smishing( payload: AnalyzeRequest, scan_service: ScanService = Depends(get_scan_service) ) -> SmishingAnalysisResponse: - try: - logger.info(f"[Router] 통합 스미싱 분석 요청 접수: {payload.message[:15]}...") - - # 파이프라인 호출 - text_analysis = await analyze_text_with_gemini(payload.message) - url_scan_result = await scan_service.scan_message_text(payload.message) - - # 하이브리드 url 엔진 결과 규격 - real_url_analysis = { - "has_url": url_scan_result.has_url, - "is_shortened": url_scan_result.original_url != url_scan_result.traced_url if url_scan_result.has_url else False, - "origin_url": url_scan_result.traced_url, - "original_url": url_scan_result.original_url, - "is_url_malicious": url_scan_result.is_url_malicious, - "url_risk_score": url_scan_result.url_risk_score, - "engine_source": url_scan_result.engine_source, - "error_message": url_scan_result.error_message - } + # Spring Boot에서 전달된 문제 메시지를 접수하여 비동기 파이프라인(LLM + 하이브리드 URL + 로컬 룰)으로 정밀 스캔 - return SmishingAnalysisResponse( - status="SUCCESS", - message="통합 스미싱 분석이 정상 수행되었습니다.", - final_score=0, - risk_grade=RiskGrade.LOW, - contribution_breakdown=ContributionBreakdown( - llm=0, - hybrid_url=0, - rules=0 - ), - text_analysis=text_analysis, - url_analysis=real_url_analysis - ) + logger.info(f"[Router] 통합 스미싱 분석 마스터 파이프라인 진입: {payload.message[:15]}...") - except Exception as e: - logger.error(f"[Router Error] 통합 분석 중 예외 발생: {str(e)}") - return SmishingAnalysisResponse( - status="ERROR", - message=f"통합 스미싱 탐지 중 서버 에러가 발생했습니다: {str(e)}", - final_score=0, - risk_grade=RiskGrade.LOW, - contribution_breakdown=ContributionBreakdown(llm=0, hybrid_url=0, rules=0), - text_analysis=None, - url_analysis=None - ) \ No newline at end of file + return await scan_service.analyze_pipeline(payload.message) \ No newline at end of file From 97066a329a7c4b3fdf0eb519cce840118ebe0d5e Mon Sep 17 00:00:00 2001 From: pearseona Date: Mon, 20 Jul 2026 17:11:34 +0900 Subject: [PATCH 7/8] fix: resolve edge cases for messages without URLs and secure fail-safe fallback --- app/service/scan_service.py | 37 ++++++++++++++++++++++++++----------- 1 file changed, 26 insertions(+), 11 deletions(-) diff --git a/app/service/scan_service.py b/app/service/scan_service.py index 03ab69b..bd35fdb 100644 --- a/app/service/scan_service.py +++ b/app/service/scan_service.py @@ -55,6 +55,22 @@ async def url_track(): has_rule_violation=has_rule_violation ) + # URL 부재 시 예외 방어 및 스켈레톤 분기벽 구축 + if has_url: + real_url_analysis = { + "has_url": True, + "is_shortened": original_url != traced_url, + "origin_url": traced_url, + "original_url": original_url, + "is_url_malicious": hybrid_res["is_malicious"], + "url_risk_score": hybrid_res["url_risk_score"], + "engine_source": hybrid_res["source"], + "error_message": hybrid_res["error_message"] + } + else: + # URL이 없는 평문 문자일 경우 null(None) 처리 + real_url_analysis = None + return SmishingAnalysisResponse( status="SUCCESS", message="3중 가중치 결합 스미싱 통합 분석이 완료되었습니다.", @@ -62,20 +78,19 @@ async def url_track(): risk_grade=risk_grade, contribution_breakdown=breakdown, text_analysis=text_analysis, - url_analysis={ - "has_url": has_url, - "is_shortened": original_url != traced_url if has_url else False, - "origin_url": traced_url, - "original_url": original_url if has_url else None, - "is_url_malicious": hybrid_res["is_malicious"], - "url_risk_score": hybrid_res["url_risk_score"], - "engine_source": hybrid_res["source"], - "error_message": hybrid_res["error_message"] - } + url_analysis=real_url_analysis ) except Exception as e: logger.error(f"파이프라인 에러: {str(e)}") - return SmishingAnalysisResponse(status="ERROR", message=str(e), final_score=0, risk_grade=RiskGrade.LOW, contribution_breakdown=ContributionBreakdown(llm=0, hybrid_url=0, rules=0)) + return SmishingAnalysisResponse( + status="ERROR", + message=str(e), + final_score=0, + risk_grade=RiskGrade.LOW, + contribution_breakdown=ContributionBreakdown(llm=0, hybrid_url=0, rules=0), + text_analysis=None, + url_analysis=None + ) async def scan_message_text(self, message: str) -> URLScanResponse: urls = extract_urls(message) From 3cdfbf21722d0443b560c25f6187d5f3c73305e7 Mon Sep 17 00:00:00 2001 From: pearseona Date: Mon, 20 Jul 2026 17:48:59 +0900 Subject: [PATCH 8/8] refactor: fix pipeline bugs and stabilize core smishing detection workflow --- app/dto/schemas.py | 5 +- app/main.py | 4 +- app/router/analyze.py | 28 +++++------ app/service/scan_service.py | 33 +++++++++---- app/service/security/hybrid_url_engine.py | 23 ++++++--- app/utils/scoring_engine | 60 ----------------------- app/utils/scoring_engine.py | 60 +++++++++++++++++++++++ 7 files changed, 120 insertions(+), 93 deletions(-) delete mode 100644 app/utils/scoring_engine create mode 100644 app/utils/scoring_engine.py diff --git a/app/dto/schemas.py b/app/dto/schemas.py index e3d0975..2c5fb1d 100644 --- a/app/dto/schemas.py +++ b/app/dto/schemas.py @@ -17,8 +17,7 @@ class URLScanResponse(BaseModel): error_message: Optional[str] = Field(None, description="에러 발생 시 메시지 기록용") class Config: - - json_schema_extra = { + json_schema_extra = { "example": { "has_url": True, "original_url": "https://bit.ly/suspect-link", @@ -64,7 +63,7 @@ class Config: "risk_grade": "HIGH", "contribution_breakdown": { "llm": 45, - "virus_total": 30, + "hybrid_url": 30, "rules": 20 }, "text_analysis": { diff --git a/app/main.py b/app/main.py index 2c973ac..8a33ffe 100644 --- a/app/main.py +++ b/app/main.py @@ -8,6 +8,7 @@ version=settings.VERSION ) +# Spring Boot 및 프론트엔드 연동을 위한 CORS 미들웨어 설정 app.add_middleware( CORSMiddleware, allow_origins=["*"], @@ -16,7 +17,8 @@ allow_headers=["*"], ) -app.include_router(analyze.router, prefix=settings.API_V1_STR) +# 라우터 통합 등록 +app.include_router(analyze.router, prefix="/api") @app.get("/", tags=["Root"]) def root_check(): diff --git a/app/router/analyze.py b/app/router/analyze.py index fe82764..80d71d4 100644 --- a/app/router/analyze.py +++ b/app/router/analyze.py @@ -1,33 +1,33 @@ import logging -from fastapi import APIRouter, Depends, status -from app.dto.request import AnalyzeRequest -from app.dto.schemas import SmishingAnalysisResponse +from fastapi import APIRouter, Depends, status, HTTPException +from app.dto.schemas import URLScanRequest, SmishingAnalysisResponse from app.service.scan_service import ScanService logger = logging.getLogger(__name__) -# 문자 분석 전용 라우터 생성 router = APIRouter(prefix="/analyze", tags=["Analyze"]) -# 서비스 인스턴스 생성 유틸 def get_scan_service() -> ScanService: return ScanService() -# 통합 스미싱 탐지 API @router.post( "", response_model=SmishingAnalysisResponse, status_code=status.HTTP_200_OK, - summary="[메인 통합 엔진] 문자 본문 기반 3중 스미싱 통합 분석", - description="문자 본문 전체를 분석하여 LLM 문맥, 하이브리드 URL 검사, 로컬 규칙을 합성한 0~100점 점수를 반환합니다." + summary="[메인 통합 엔진] 문자 본문 기반 3중 스미싱 통합 분석" ) async def analyze_smishing( - payload: AnalyzeRequest, + payload: URLScanRequest, scan_service: ScanService = Depends(get_scan_service) - ) -> SmishingAnalysisResponse: +) -> SmishingAnalysisResponse: - # Spring Boot에서 전달된 문제 메시지를 접수하여 비동기 파이프라인(LLM + 하이브리드 URL + 로컬 룰)으로 정밀 스캔 - - logger.info(f"[Router] 통합 스미싱 분석 마스터 파이프라인 진입: {payload.message[:15]}...") + logger.info(f"[Router] 통합 스미싱 분석 마스터 파이프라인 진입: {payload.text[:15]}...") - return await scan_service.analyze_pipeline(payload.message) \ No newline at end of file + try: + return await scan_service.analyze_pipeline(payload.text) + except Exception as e: + logger.error(f"[Router] 스캔 처리 중 장애 발생: {str(e)}") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"서버 내부 스캔 파이프라인 연산 중 오류: {str(e)}" + ) \ No newline at end of file diff --git a/app/service/scan_service.py b/app/service/scan_service.py index bd35fdb..46e1a6e 100644 --- a/app/service/scan_service.py +++ b/app/service/scan_service.py @@ -12,15 +12,15 @@ class ScanService: def __init__(self): self.hybrid_url_engine = HybridUrlEngine() - # 텍스트 트랙과 URL 트랙을 병렬 조립하고 스코어링을 매핑 - async def analyze_pipeline(self, message: str) -> SmishingAnalysisResponse: + # 텍스트 트랙과 URL 트랙을 병렬 조립하고 스코어링을 매핑 + async def analyze_pipeline(self, text: str) -> SmishingAnalysisResponse: try: - urls = extract_urls(message) + urls = extract_urls(text) has_url = len(urls) > 0 - # 비동기 Task 스케줄링 - text_task = asyncio.create_task(analyze_text_with_gemini(message)) + # 비동기 Task 스케줄링 + text_task = asyncio.create_task(analyze_text_with_gemini(text)) url_task = None if has_url: @@ -47,7 +47,7 @@ async def url_track(): hybrid_res["is_malicious"] = True hybrid_res["url_risk_score"] = max(hybrid_res["url_risk_score"], 0.75) - # 3중 스코어링 최종 계산 + # 3중 스코어링 최종 계산 final_score, risk_grade, breakdown = ScoringEngine.calculate_score( llm_score=int(llm_score), is_url_malicious=hybrid_res["is_malicious"], @@ -68,7 +68,6 @@ async def url_track(): "error_message": hybrid_res["error_message"] } else: - # URL이 없는 평문 문자일 경우 null(None) 처리 real_url_analysis = None return SmishingAnalysisResponse( @@ -94,7 +93,23 @@ async def url_track(): async def scan_message_text(self, message: str) -> URLScanResponse: urls = extract_urls(message) - if not urls: return URLScanResponse(has_url=False, original_url=None, traced_url=None, is_url_malicious=False, url_risk_score=0.0, engine_source="Pre-Processing-Filter") + if not urls: + return URLScanResponse( + has_url=False, + original_url=None, + traced_url=None, + is_url_malicious=False, + url_risk_score=0.0, + engine_source="Pre-Processing-Filter" + ) traced_url = await trace_url(urls[0]) res = await self.hybrid_url_engine.scan_url(traced_url) - return URLScanResponse(has_url=True, original_url=urls[0], traced_url=traced_url, is_url_malicious=res["is_malicious"], url_risk_score=res["url_risk_score"], engine_source=res["source"], error_message=res["error_message"]) \ No newline at end of file + return URLScanResponse( + has_url=True, + original_url=urls[0], + traced_url=traced_url, + is_url_malicious=res["is_malicious"], + url_risk_score=res["url_risk_score"], + engine_source=res["source"], + error_message=res["error_message"] + ) \ No newline at end of file diff --git a/app/service/security/hybrid_url_engine.py b/app/service/security/hybrid_url_engine.py index 491768d..14884a4 100644 --- a/app/service/security/hybrid_url_engine.py +++ b/app/service/security/hybrid_url_engine.py @@ -15,9 +15,18 @@ def __init__(self): self.gsb_engine = GoogleSafeBrowsingEngine() async def scan_url(self, traced_url: str) -> dict: - logger.info(f"[MOCK MODE] 하이브리드 URL 스캔 -> Target: {traced_url}") - return {"is_malicious": True, "url_risk_score": 0.85, "source": "Hybrid-Engine (MOCK)", "detected_count": 4, "error_message": None} + # 쉘 환경변수에 따른 MOCK 모드 분기 로직 정상화 + if MOCK_ENABLED: + logger.info(f"[MOCK MODE] 하이브리드 URL 스캔 -> Target: {traced_url}") + return { + "is_malicious": True, + "url_risk_score": 0.85, + "source": "Hybrid-Engine (MOCK)", + "detected_count": 4, + "error_message": None + } + # PROD 운영 모드 가동 logger.info("[PROD MODE] 1차 방어선: Google Safe Browsing API 가동") error_logs = [] @@ -32,12 +41,13 @@ async def scan_url(self, traced_url: str) -> dict: vt_result = {"is_malicious": False, "detected_count": 0} - # GSB 악성 확정 시 VT 생략 + # GSB 악성 확정 시 VT 생략 (Quota 절약) if is_gsb_blocked: logger.info(" GSB 악성 판정으로 VirusTotal 호출 생략 (Quota 절약)") engine_source = "Hybrid-Engine (GSB)" risk_score = gsb_result.get("raw_score", 0.95) - if risk_score > 1.0: risk_score /= 100.0 + if risk_score > 1.0: + risk_score /= 100.0 else: logger.info(" GSB 청정/불확실로 인한 2차 방어선 VirusTotal 백업 가동") engine_source = "Hybrid-Engine (GSB+VT)" @@ -51,7 +61,8 @@ async def scan_url(self, traced_url: str) -> dict: vt_malicious_count = vt_result.get("detected_count", 0) if vt_malicious_count > 0: base_score = vt_result.get("raw_score", 0.0) - if base_score > 1.0: base_score /= 100.0 + if base_score > 1.0: + base_score /= 100.0 risk_score = max(base_score, min(0.1 + (vt_malicious_count * 0.15), 0.95)) else: risk_score = 0.0 @@ -64,4 +75,4 @@ async def scan_url(self, traced_url: str) -> dict: "url_risk_score": round(risk_score, 2), "source": engine_source, "error_message": combined_error - } + } \ No newline at end of file diff --git a/app/utils/scoring_engine b/app/utils/scoring_engine deleted file mode 100644 index 60726b6..0000000 --- a/app/utils/scoring_engine +++ /dev/null @@ -1,60 +0,0 @@ -import logging -from typing import Tuple, Dict, Any -from app.dto.schemas import RiskGrade, ContributionBreakdown - -logger = logging.getLogger(__name__) - -class scoringEngine: - """ - 3중 가중치 스코어링 시스템 - - LLM 문맥 분석 (50%) - - 하이브리드 URL 보안 엔진 [GSB + VT 백업] (30%) - - 로컬 가드 규칙 기반 패널티 (20%) - """ - - @staticmethod - def calculate_score( - llm_score: int, # LLM이 반환한 위험도 점수 (0~100) - is_url_maliciout: bool, # 하이브리드 URL 엔진의 최종 악성 판정 여부 - url_risk_score: float, # 하이브리드 URL 엔진이 계산한 위험도 점수 (0.0~1.0) - has_rule_violation: bool # 로컬 패널티 룰(.ru 등) 적발 여부 - ) -> Tuple[int, RiskGrade, ContributionBreakdown]: - - # 텍스트 문맥 점수와 하이브리드 URL 엔진의 결과값을 결합하여 최종 위험도를 산출 - - # 1. LLM 문맥 분석 기여 점수 (최대 50점) - llm_contrib = round(llm_score * 0.5) - - # 2. 하이브리드 URL 보안 엔진 기여 점수 (최대 30점) - if is_url_malicious or url_risk_score > 0: - url_contrib = round(url_risk_score * 30) - url_contrib = min(max(url_contrib, 0), 30) - else: - url_contrib = 0 - - # 3. 로컬 가드 규칙 기반 기여 점수 (최대 20점) - rules_contrib = 20 if has_rule_violation else 0 - - # 최종 위험도 점수 합산 - final_score = min(llm_contrib + url_contrib + rules_contrib, 100) - - # 최종 점수 기반 임계치 등급 분기 - if final_score >= 70: - risk_grade = RiskGrade.HIGH - elif final_score >= 40: - risk_grade = RiskGrade.MEDIUM - else: - risk_grade = RiskGrade.LOW - - logger.ingo( - f"[Scoring Engine] 통합 연산 완료 -> 최종 점수: {final_score} | 등급: {risk_grade} " - f"(LLM: {llm_contrib}, Hybrid-URL: {url_contrib}, Rules: {rules_contrib})" - ) - - breakdown = ContributionBreakdown( - llm=llm_contrib, - hybrid_url=url_contrib, - rules=rules_contrib - ) - - return final_score, risk_grade, breakdown \ No newline at end of file diff --git a/app/utils/scoring_engine.py b/app/utils/scoring_engine.py new file mode 100644 index 0000000..ad52b4a --- /dev/null +++ b/app/utils/scoring_engine.py @@ -0,0 +1,60 @@ +import logging +from typing import Tuple, Dict, Any +from app.dto.schemas import RiskGrade, ContributionBreakdown + +logger = logging.getLogger(__name__) + +class ScoringEngine: + """ + 3중 가중치 스코어링 시스템 + - LLM 문맥 분석 (50%) + - 하이브리드 URL 보안 엔진 [GSB + VT 백업] (30%) + - 로컬 가드 규칙 기반 패널티 (20%) + """ + + @staticmethod + def calculate_score( + llm_score: int, # LLM이 반환한 위험도 점수 (0~100) + is_url_malicious: bool, # 하이브리드 URL 엔진의 최종 악성 판정 여부 + url_risk_score: float, # 하이브리드 URL 엔진이 계산한 위험도 점수 (0.0~1.0) + has_rule_violation: bool # 로컬 패널티 룰(.ru 등) 적발 여부 + ) -> Tuple[int, RiskGrade, ContributionBreakdown]: + + # 텍스트 문맥 점수와 하이브리드 URL 엔진의 결과값을 결합하여 최종 위험도를 산출 + + # 1. LLM 문맥 분석 기여 점수 (최대 50점) + llm_contrib = round(llm_score * 0.5) + + # 2. 하이브리드 URL 보안 엔진 기여 점수 (최대 30점) + if is_url_malicious or url_risk_score > 0: + url_contrib = round(url_risk_score * 30) + url_contrib = min(max(url_contrib, 0), 30) + else: + url_contrib = 0 + + # 3. 로컬 가드 규칙 기반 기여 점수 (최대 20점) + rules_contrib = 20 if has_rule_violation else 0 + + # 최종 위험도 점수 합산 + final_score = min(llm_contrib + url_contrib + rules_contrib, 100) + + # 최종 점수 기반 임계치 등급 분기 + if final_score >= 70: + risk_grade = RiskGrade.HIGH + elif final_score >= 40: + risk_grade = RiskGrade.MEDIUM + else: + risk_grade = RiskGrade.LOW + + logger.info( + f"[Scoring Engine] 통합 연산 완료 -> 최종 점수: {final_score} | 등급: {risk_grade} " + f"(LLM: {llm_contrib}, Hybrid-URL: {url_contrib}, Rules: {rules_contrib})" + ) + + breakdown = ContributionBreakdown( + llm=llm_contrib, + hybrid_url=url_contrib, + rules=rules_contrib + ) + + return final_score, risk_grade, breakdown \ No newline at end of file