분석엔진 1차

.github/workflows/deploy.yml

@@ -0,0 +1,29 @@
+name: FastAPI Auto Deploy
+
+on:
+  push:
+    branches: [ "develop" ]
+
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+    steps:
+      - name: SSH Remote Commands for FastAPI
+        uses: appleboy/ssh-action@master
+        with:
+          host: ${{ secrets.GCP_IP }}
+          username: ${{ secrets.GCP_USER }}
+          key: ${{ secrets.GCP_SSH_KEY }}
+          command_timeout: 30m
+          script: |
+            # 1. 파이썬 공용 폴더로 이동해서 최신 코드 땡겨오기
+            cd /opt/analysis-engine
+            git pull origin develop
+
+            # 2. 도커 설계도가 있는 메인 서버 폴더로 이동!
+            cd /opt/main-server
+
+            # 3. 파이썬(veriq-analysis) 컨테이너만 콕 집어서 다시 빌드하고 실행
+            sudo docker-compose up --build -d veriq-analysis
+
+

.gitignore

@@ -143,6 +143,7 @@ venv/
 ENV/
 env.bak/
 venv.bak/
+.vscode
 
 # Spyder project settings
 .spyderproject
@@ -173,7 +174,7 @@ cython_debug/
 #  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
-#.idea/
+.idea/
 
 # Abstra
 # Abstra is an AI-powered process automation framework.
@@ -186,7 +187,7 @@ cython_debug/
 #  that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
 #  and can be added to the global gitignore or merged into this file. However, if you prefer, 
 #  you could uncomment the following to ignore the entire vscode folder
-# .vscode/
+.vscode/
 
 # Ruff stuff:
 .ruff_cache/
@@ -205,3 +206,12 @@ cython_debug/
 marimo/_static/
 marimo/_lsp/
 __marimo__/
+
+# 검사기 테스트 결과 및 테스트 도구
+test
+test/
+test-data
+test-data/
+eval_results
+tools
+test_results

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "git.ignoreLimitWarning": true
+}

Dockerfile

@@ -0,0 +1,6 @@
+FROM python:3.11-slim
+WORKDIR /app
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+COPY . .
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

README.md

@@ -1,23 +1,11 @@
 # analysis-engine
-be2
-python- FastAPI
-api: google safe browsing
 
-***
+## 다음 수정 : 
+이것도 금일(0503) 내로 진행합니다......
 
-## 현재 단계 : API, DB 조회 오류 수정, URL정규화+단축/https검사 구현
-#### 지난 단계: API, DB 조회 구현
+#### 1. server info조회 결과와 context 간 연결, 결과 json에 확실히 반영되도록 안정화
 
-조회 부분까지의 업로드를 다른 브랜치에 하려 했으나 실수로 main에 집어넣었습니다.
-오류 수정을 포함한 이후 단계부터는 develop, 또는 develop/기능 이름 브랜치에서 진행합니다.
+#### 2. api조회 비동기 반영 수정
 
-***
-
-### API, DB 조회 상세 설명
-+ 테스트를 위한 프로토타입 목적입니다. (최종 형태 X)
-  + 작동 구조만 맞추는 것에 가깝습니다.
-  + 더미 데이터로 테스트하며, 응답을 저장하지 않습니다.
-+ API는 google safe browsing만을 조회합니다.
-+ DB 조회 형식이 확정되지 않았습니다.
-  + case1> 요청-응답 : 분석 서버에서 메인 서버로 조회 요청 전송, 조회 결과를 응답받습니다. <현재
-  + case2> 직접 조회(pymysql) : 분석 서버에서 MySQL DB로 직접 질의를 통해 조회합니다.
+#### 3. feature 추출기 결합 + 학습한 모델 업로드
+*제발할수있었으면좋겠네요진짜*

analyzers/ai_analyze.py

@@ -1,4 +1,270 @@
-# ai를 이용하는 모든 부분을 해당 클래스에 선언.
-# userinfo, host, domain등.... 가장 많은 파츠를 보지 않나
-# 스코어링도 ai로 한다면 여기에 클래스 정의하게 됨
-# scoring engine을 별도 분리할 생각이 그다지 없긴 한데... 아무튼간에.
+# pre-trained ML 모델을 호출해 분석을 진행합니다.
+# 현재 연결 대상: CharCNN + FEB/XGB score-level fusion
+from core.context import AnalysisContext
+from ml.charcnn.predictor import predict_url as predict_charcnn_url
+from ml.xgb.predictor import predict_context as predict_xgb_context
+
+
+FUSION_METHOD = "weighted_average"
+XGB_WEIGHT = 0.65
+CHARCNN_WEIGHT = 0.35
+
+# 한 모델이 매우 강하게 위험으로 판단한 경우,
+# 가중 평균 점수에 최소 보정점을 더해 ML 결과에 반영한다.
+HIGH_CONFIDENCE_THRESHOLD = 85
+HIGH_CONFIDENCE_BONUS = 25
+
+
+# ML 입력 URL 결정 함수
+# 요구사항: redirect 추적 결과가 있으면 final_url을 우선 사용한다.
+def resolve_ml_target_url(context: AnalysisContext) -> str | None:
+    final_url = context.redirect.get("final_url")
+    if final_url:
+        return final_url
+
+    normalized_url = context.url.get("normalized_url")
+    if normalized_url:
+        return normalized_url
+
+    return context.request.get("original_url")
+
+
+def ai_analyze(context: AnalysisContext) -> dict:
+    target_url = resolve_ml_target_url(context)
+
+    charcnn_result = run_charcnn_inference(target_url)
+    charcnn_result = apply_charcnn_whitelist_score_cap(context, charcnn_result)
+
+    xgb_result = run_xgb_inference(context)
+    fusion_result = fuse_ml_scores(xgb_result, charcnn_result)
+
+    return build_ml_result(
+        score=fusion_result.get("score", 0),
+        threats=collect_ml_threats(xgb_result, charcnn_result, fusion_result),
+        charcnn=charcnn_result,
+        xgb=xgb_result,
+        fusion=fusion_result,
+    )
+
+
+def run_charcnn_inference(target_url: str | None) -> dict:
+    return predict_charcnn_url(target_url)
+
+
+def run_xgb_inference(context: AnalysisContext) -> dict:
+    return predict_xgb_context(context)
+
+
+def is_available(result: dict | None) -> bool:
+    return isinstance(result, dict) and result.get("available") is True
+
+
+def safe_score(result: dict | None) -> int | None:
+    if not isinstance(result, dict):
+        return None
+
+    if result.get("available") is not True:
+        return None
+
+    try:
+        return max(0, min(100, int(round(float(result.get("score", 0))))))
+    except (TypeError, ValueError):
+        return None
+
+
+def fuse_ml_scores(xgb: dict, charcnn: dict) -> dict:
+    xgb_score = safe_score(xgb)
+    charcnn_score = safe_score(charcnn)
+
+    contributors = []
+
+    if xgb_score is not None:
+        contributors.append("xgb")
+    if charcnn_score is not None:
+        contributors.append("charcnn")
+
+    if xgb_score is not None and charcnn_score is not None:
+        # 현재 CharCNN이 일부 정상 URL에도 매우 높은 점수를 내는 경향이 있어
+        # 기본 fusion은 XGB를 더 강하게 반영하는 가중 평균으로 유지한다.
+        base_score = round(xgb_score * XGB_WEIGHT + charcnn_score * CHARCNN_WEIGHT)
+        method = FUSION_METHOD
+        weights = {"xgb": XGB_WEIGHT, "charcnn": CHARCNN_WEIGHT}
+    elif xgb_score is not None:
+        base_score = xgb_score
+        method = "xgb_only"
+        weights = {"xgb": 1.0, "charcnn": 0.0}
+    elif charcnn_score is not None:
+        base_score = charcnn_score
+        method = "charcnn_only"
+        weights = {"xgb": 0.0, "charcnn": 1.0}
+    else:
+        base_score = 0
+        method = "unavailable"
+        weights = {"xgb": 0.0, "charcnn": 0.0}
+
+    bonus = HIGH_CONFIDENCE_BONUS if has_high_confidence_score(xgb_score, charcnn_score) else 0
+    score = max(0, min(100, int(base_score + bonus)))
+
+    return {
+        "available": bool(contributors),
+        "method": method,
+        "weights": weights,
+        "contributors": contributors,
+        "score": score,
+    }
+
+
+def has_high_confidence_score(*scores: int | None) -> bool:
+    for score in scores:
+        if score is not None and score >= HIGH_CONFIDENCE_THRESHOLD:
+            return True
+
+    return False
+
+
+def collect_ml_threats(xgb: dict, charcnn: dict, fusion: dict) -> list[str]:
+    # 위협 라벨은 fusion 점수와 별개로, 개별 모델이 반환한 threats를 그대로 반영한다.
+    # 단, CharCNN whitelist가 적용된 경우 apply_charcnn_whitelist_score_cap()에서
+    # charcnn.threats를 비우므로 top-level ml.threats에도 올라오지 않는다.
+    threats = []
+    for result in (xgb, charcnn):
+        if not isinstance(result, dict):
+            continue
+
+        for threat in result.get("threats", []) or []:
+            if threat not in threats:
+                threats.append(threat)
+
+    return threats
+
+
+def build_ml_result(
+    score: int = 0,
+    threats: list | None = None,
+    charcnn: dict | None = None,
+    xgb: dict | None = None,
+    fusion: dict | None = None,
+) -> dict:
+    if threats is None:
+        threats = []
+
+    return {
+        "score": score,
+        "threats": threats,
+        "charcnn": charcnn or {
+            "available": False,
+            "score": 0,
+            "label": "unknown",
+            "threats": [],
+        },
+        "xgb": xgb or {
+            "available": False,
+            "score": 0,
+            "label": "unknown",
+            "threats": [],
+        },
+        "fusion": fusion or {
+            "available": False,
+            "method": "unavailable",
+            "score": 0,
+            "contributors": [],
+        },
+    }
+
+# ============================================================
+# CharCNN whitelist score-cap post-processing
+# ============================================================
+# 목적:
+# - 유명 정상 도메인에서 CharCNN 오탐이 높게 나오는 경우만 완화한다.
+# - rule 기반 검사에서 flags가 있으면 whitelist를 적용하지 않는다.
+# - 다른 검사 항목, XGB, external, cert, redirect 결과는 건드리지 않는다.
+# - whitelist 적용 여부를 별도 라벨로 반환하지 않는다.
+
+WHITELIST_DOMAINS = {
+    "google.com", "google.co.kr", "naver.com", "daum.net", "yahoo.com", "bing.com",
+    "facebook.com", "instagram.com", "twitter.com", "linkedin.com", "youtube.com", "kakao.com",
+    "apple.com", "microsoft.com", "github.com", "stackoverflow.com", "cloudflare.com",
+    "amazon.com", "aws.com",
+}
+
+CHARCNN_WHITELIST_SCORE_CAP = 10
+
+
+def apply_charcnn_whitelist_score_cap(context: AnalysisContext, charcnn_result: dict | None) -> dict:
+    if not isinstance(charcnn_result, dict):
+        return {
+            "available": False,
+            "score": 0,
+            "label": "unknown",
+            "threats": [],
+        }
+
+    if not is_available(charcnn_result):
+        return charcnn_result
+
+    if has_rule_flags(context):
+        return charcnn_result
+
+    if not is_whitelisted_context_domain(context):
+        return charcnn_result
+
+    result = dict(charcnn_result)
+    capped_score = min(to_int_score(result.get("score", 0)), CHARCNN_WHITELIST_SCORE_CAP)
+
+    result["score"] = capped_score
+    result["threats"] = []
+
+    # CharCNN 세부 label을 그대로 malicious로 두면 하위 결과를 읽는 쪽에서 혼동될 수 있어
+    # score cap 이후에는 안전 쪽 label로 맞춘다. whitelist 관련 라벨은 별도로 추가하지 않는다.
+    if capped_score < 50:
+        result["label"] = "normal"
+
+    return result
+
+
+def has_rule_flags(context: AnalysisContext) -> bool:
+    rule_based = {}
+
+    if context is not None:
+        detection = getattr(context, "detection", {}) or {}
+        rule_based = detection.get("rule_based", {}) or {}
+
+    flags = rule_based.get("flags", [])
+    return bool(flags)
+
+
+def is_whitelisted_context_domain(context: AnalysisContext) -> bool:
+    hostname = get_context_hostname(context)
+
+    if not hostname:
+        return False
+
+    for allowed_domain in WHITELIST_DOMAINS:
+        allowed_domain = normalize_domain(allowed_domain)
+
+        if hostname == allowed_domain or hostname.endswith("." + allowed_domain):
+            return True
+
+    return False
+
+
+def get_context_hostname(context: AnalysisContext) -> str:
+    if context is None:
+        return ""
+
+    url_info = getattr(context, "url", {}) or {}
+    return normalize_domain(url_info.get("hostname"))
+
+
+def normalize_domain(domain: str | None) -> str:
+    if not domain:
+        return ""
+
+    return str(domain).strip().lower().strip(".")
+
+
+def to_int_score(value, default: int = 0) -> int:
+    try:
+        return max(0, min(100, int(round(float(value)))))
+    except (TypeError, ValueError):
+        return default