diff --git a/app/dto/schemas.py b/app/dto/schemas.py index 3d47a99..2c5fb1d 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): @@ -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", @@ -29,3 +28,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) + 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): + 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, + "hybrid_url": 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 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 7b62eb6..80d71d4 100644 --- a/app/router/analyze.py +++ b/app/router/analyze.py @@ -1,49 +1,33 @@ -from fastapi import APIRouter, Depends, HTTPException, status -from app.dto.response import ApiResponse -from app.dto.request import AnalyzeRequest -from app.dto.schemas import URLScanResponse -from app.service.security.gemini_text_analyzer import analyze_text_with_gemini +import logging +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=ApiResponse[dict], status_code=status.HTTP_200_OK) +@router.post( + "", + response_model=SmishingAnalysisResponse, + status_code=status.HTTP_200_OK, + summary="[메인 통합 엔진] 문자 본문 기반 3중 스미싱 통합 분석" +) async def analyze_smishing( - payload: AnalyzeRequest, + payload: URLScanRequest, scan_service: ScanService = Depends(get_scan_service) - ): - - try: - text_analysis = await analyze_text_with_gemini(payload.message) - - url_scan_result: URLScanResponse = await scan_service.scan_message_text(payload.message) - - 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 - "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 - } +) -> SmishingAnalysisResponse: - 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="문자 분석이 완료되었습니다.") + logger.info(f"[Router] 통합 스미싱 분석 마스터 파이프라인 진입: {payload.text[:15]}...") + try: + return await scan_service.analyze_pipeline(payload.text) except Exception as e: - return ApiResponse.error(message=f"통합 스미싱 탐지 중 서버 에러가 발생했습니다: {str(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 211b229..46e1a6e 100644 --- a/app/service/scan_service.py +++ b/app/service/scan_service.py @@ -1,145 +1,115 @@ -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, text: 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 + urls = extract_urls(text) + 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(text)) + 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 ) + # 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: + real_url_analysis = None + + return SmishingAnalysisResponse( + status="SUCCESS", + message="3중 가중치 결합 스미싱 통합 분석이 완료되었습니다.", + final_score=final_score, + risk_grade=risk_grade, + contribution_breakdown=breakdown, + text_analysis=text_analysis, + url_analysis=real_url_analysis + ) except Exception as e: - logger.error(f" 파이프라인 수행 중 예외 발생: {str(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), + text_analysis=None, + url_analysis=None + ) + + async def scan_message_text(self, message: str) -> URLScanResponse: + urls = extract_urls(message) + if not urls: 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 + 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..14884a4 --- /dev/null +++ b/app/service/security/hybrid_url_engine.py @@ -0,0 +1,78 @@ +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: + # 쉘 환경변수에 따른 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 = [] + + 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 생략 (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 + 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 + } \ 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