[Chore] FastAPI 운영 환경 설정 및 Secret 분리

.dockerignore

@@ -14,4 +14,14 @@ tests
 requirements-dev.txt
 
 *.log
-.DS_Store
+.DS_Store
+
+secrets
+data_science/**
+!data_science/SMSModel/
+!data_science/SMSModel/phishing_model_artifact.pkl
+!data_science/SMSModel/phishing_vectorizer.pkl
+*.ipynb
+*.csv
+*.png
+*.log

.env.example

@@ -0,0 +1,13 @@
+ENV=local
+
+GEMINI_API_KEY=
+GEMINI_MODEL=gemini-flash-latest
+VIRUSTOTAL_API_KEY=
+GOOGLE_SAFE_BROWSING_API_KEY=
+MOCK_SECURITY_API=false
+
+RABBITMQ_URL=amqp://safefam:safefam-local@localhost:5672/
+RABBITMQ_CONSUMER_ENABLED=false
+
+NAIVE_BAYES_MODEL_PATH=data_science/SMSModel/phishing_model_artifact.pkl
+NAIVE_BAYES_VECTORIZER_PATH=data_science/SMSModel/phishing_vectorizer.pkl

.env.prod.example

@@ -0,0 +1,27 @@
+ENV=prod
+FASTAPI_IMAGE_TAG=
+
+GEMINI_API_KEY=
+GEMINI_MODEL=gemini-flash-latest
+VIRUSTOTAL_API_KEY=
+GOOGLE_SAFE_BROWSING_API_KEY=
+
+MOCK_SECURITY_API=false
+
+RABBITMQ_URL=
+RABBITMQ_ANALYSIS_EXCHANGE=safefam.analysis
+RABBITMQ_ANALYSIS_REQUEST_QUEUE=safefam.analysis.requested.q
+RABBITMQ_ANALYSIS_REQUEST_ROUTING_KEY=analysis.requested.v1
+RABBITMQ_ANALYSIS_COMPLETED_ROUTING_KEY=analysis.completed.v1
+RABBITMQ_ANALYSIS_PARTIAL_ROUTING_KEY=analysis.partial.v1
+RABBITMQ_ANALYSIS_FAILED_ROUTING_KEY=analysis.failed.v1
+RABBITMQ_ANALYSIS_DLQ=safefam.analysis.requested.dlq
+RABBITMQ_ANALYSIS_DLQ_ROUTING_KEY=analysis.requested.dead.v1
+RABBITMQ_PREFETCH_COUNT=1
+RABBITMQ_CONSUMER_ENABLED=true
+RABBITMQ_PUBLISH_TIMEOUT_SECONDS=5
+RABBITMQ_SHUTDOWN_TIMEOUT_SECONDS=30
+RABBITMQ_REQUEUE_BACKOFF_SECONDS=1
+
+NAIVE_BAYES_MODEL_PATH=/app/models/phishing_model_artifact.pkl
+NAIVE_BAYES_VECTORIZER_PATH=/app/models/phishing_vectorizer.pkl

.gitignore

@@ -4,4 +4,11 @@ __pycache__/
 .pytest_cache/
 .venv/
 .claude/settings.local.json
-.env
+.env
+.env.*
+!.env.example
+!.env.prod.example
+
+secrets/
+*.pem
+*.key

Dockerfile

@@ -1,26 +1,24 @@
 FROM python:3.11-slim
 
-WORKDIR /workspace
+WORKDIR /app
 
 ENV PYTHONDONTWRITEBYTECODE=1
 ENV PYTHONUNBUFFERED=1
+ENV NAIVE_BAYES_MODEL_PATH=/app/models/phishing_model_artifact.pkl
+ENV NAIVE_BAYES_VECTORIZER_PATH=/app/models/phishing_vectorizer.pkl
 
 COPY requirements.txt .
-
 RUN pip install --no-cache-dir -r requirements.txt
 
 COPY app app
+COPY data_science/SMSModel/phishing_model_artifact.pkl models/phishing_model_artifact.pkl
+COPY data_science/SMSModel/phishing_vectorizer.pkl models/phishing_vectorizer.pkl
 
-RUN useradd --create-home --shell /usr/sbin/nologin safefam
+RUN useradd --create-home --shell /usr/sbin/nologin safefam \
+    && chown -R safefam:safefam /app
 
 USER safefam
 
 EXPOSE 8000
 
-HEALTHCHECK \
-    --interval=30s \
-    --timeout=5s \
-    --retries=3 \
-    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/', timeout=3)"
-
 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

README.md

@@ -67,3 +67,48 @@ docker compose up --build
 서버가 켜지면 브라우저를 열고 아래 주소로 접속하여 정상 작동하는지 확인합니다.
 
 - **Swagger UI (API 문서)**: http://127.0.0.1:8000/docs
+
+## 운영 환경
+
+운영 환경에서는 `docker-compose.prod.yml`과 Git Commit SHA로 고정된 이미지를
+사용합니다. `--reload`, 소스 코드 바인드 마운트, 호스트 포트 공개는 사용하지
+않습니다.
+
+```bash
+cp .env.prod.example .env.runtime
+docker compose -f docker-compose.prod.yml --env-file .env.runtime up -d
+```
+
+`.env.runtime`의 실제 값은 저장소에 커밋하지 않습니다. EC2 IAM Role로 AWS
+Parameter Store의 `SecureString`을 조회하여 배포 시점에 생성합니다.
+
+### 운영 필수 Secret
+
+- `GEMINI_API_KEY`
+- `VIRUSTOTAL_API_KEY`
+- `GOOGLE_SAFE_BROWSING_API_KEY`
+- `RABBITMQ_URL`
+
+권장 Parameter Store 경로는 다음과 같습니다.
+
+```text
+/safefam/prod/ai/GEMINI_API_KEY
+/safefam/prod/ai/VIRUSTOTAL_API_KEY
+/safefam/prod/ai/GOOGLE_SAFE_BROWSING_API_KEY
+```
+
+운영에서 필수 Secret이 누락되거나 `MOCK_SECURITY_API=true`이면 애플리케이션은
+시작하지 않습니다.
+
+### 모델 파일
+
+운영 이미지에는 아래 두 개의 검증된 학습 산출물만 포함합니다.
+
+```text
+/app/models/phishing_model_artifact.pkl
+/app/models/phishing_vectorizer.pkl
+```
+
+컨테이너 시작 시 두 파일이 없으면 애플리케이션이 즉시 실패합니다. Pickle은
+임의 파일을 실행할 위험이 있으므로 저장소에서 관리하는 신뢰된 산출물만
+사용해야 합니다.

app/analysis/router.py

@@ -21,13 +21,19 @@ async def analyze_smishing(
     analysis_service: SmishingAnalysisService = Depends(get_analysis_service)
 ) -> SmishingAnalysisResponse:
 
-    logger.info(f"[Router] 통합 스미싱 분석 마스터 파이프라인 진입: {payload.text[:15]}...")
+    logger.info(
+        "[Router] 통합 스미싱 분석 요청 수신. text_length=%d",
+        len(payload.text),
+    )
 
     try:
         return await analysis_service.analyze_pipeline(payload.text)
-    except Exception as e:
-        logger.error(f"[Router] 스캔 처리 중 장애 발생: {str(e)}")
+    except Exception as exception:
+        logger.error(
+            "[Router] 스캔 처리 중 장애 발생. error_type=%s",
+            type(exception).__name__,
+        )
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"서버 내부 스캔 파이프라인 연산 중 오류: {str(e)}"
-        )
+            detail="서버 내부 스캔 파이프라인 연산 중 오류가 발생했습니다.",
+        ) from exception

app/analysis/service.py

@@ -101,8 +101,11 @@ async def url_track():
             # 로컬 규칙 기반 트랙: 금융기관 DB 대조 + 금융 키워드 + 계좌/카드번호 패턴 + 도메인 룰(.ru 등)
             try:
                 rule_result = self.rule_analyzer(text, traced_url)
-            except Exception:
-                logger.exception("[Analysis Service] 규칙 분석 중 오류 발생")
+            except Exception as exception:
+                logger.error(
+                    "[Analysis Service] 규칙 분석 중 오류 발생. error_type=%s",
+                    type(exception).__name__,
+                )
                 rule_result = {
                     "rule_score": 0,
                     "has_malicious_domain_pattern": False,
@@ -189,16 +192,19 @@ async def url_track():
                 url_analysis=real_url_analysis,
                 rule_analysis=rule_result
             )
-        except Exception as e:
-            logger.error(f"파이프라인 에러: {str(e)}")
+        except Exception as exception:
+            logger.error(
+                "파이프라인 오류. error_type=%s",
+                type(exception).__name__,
+            )
             # 파이프라인이 통째로 죽어 어떤 트랙도 실행되지 못한 경우, final_score=0/LOW를
             # 반환하면 "분석 실패"가 "안전 확인됨"으로 읽혀 fail-open이 된다 (텍스트 트랙
             # 양쪽 엔진이 동시에 실패한 경우를 막는 BOTH_ENGINES_UNAVAILABLE_FALLBACK_SCORE와
             # 같은 이유). status="ERROR"만 보고 걸러내지 않는 소비자를 위해 등급/점수 자체를
             # 최소 MEDIUM으로 강제한다.
             return SmishingAnalysisResponse(
                 status="ERROR",
-                message=str(e),
+                message="분석 파이프라인 처리 중 오류가 발생했습니다.",
                 final_score=RiskScoringEngine.PIPELINE_FAILURE_FALLBACK_SCORE,
                 risk_grade=RiskGrade.MEDIUM,
                 contribution_breakdown=ContributionBreakdown(llm=0, hybrid_url=0, rules=0),

app/analysis/text/gemini_analyzer.py

@@ -1,20 +1,17 @@
-import os
 import json
 import logging
 import httpx
-from dotenv import load_dotenv
 from app.analysis.risk_policy import determine_text_risk_grade
+from app.core.config import settings
 from app.infrastructure.gemini.client import GeminiClient
 
-load_dotenv()
-
 logger = logging.getLogger(__name__)
 
-GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
-GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-flash-latest")
+GEMINI_API_KEY = settings.GEMINI_API_KEY
+GEMINI_MODEL = settings.GEMINI_MODEL
 API_URL = f"https://generativelanguage.googleapis.com/v1beta/models/{GEMINI_MODEL}:generateContent"
 
-MOCK_ENABLED = os.getenv("MOCK_SECURITY_API", "False").lower() in ("true", "1", "t")
+MOCK_ENABLED = settings.MOCK_SECURITY_API
 
 # Gemini에게 구조화된 JSON 응답을 강제하기 위한 응답 스키마
 RESPONSE_SCHEMA = {
@@ -174,10 +171,16 @@ async def analyze_text_with_gemini(text: str) -> dict:
         logger.error("Gemini API 요청 타임아웃 발생")
         return _build_result(DEFAULT_ANALYSIS_RESULT, is_mock=False, error="Timeout")
 
-    except (KeyError, IndexError, json.JSONDecodeError) as e:
-        logger.error(f"Gemini 응답 파싱 실패: {str(e)}")
+    except (KeyError, IndexError, json.JSONDecodeError) as exception:
+        logger.error(
+            "Gemini 응답 파싱 실패. error_type=%s",
+            type(exception).__name__,
+        )
         return _build_result(DEFAULT_ANALYSIS_RESULT, is_mock=False, error="Parse Error")
 
-    except Exception as e:
-        logger.error(f"Gemini 연동 중 비정상 에러 발생: {str(e)}")
+    except Exception as exception:
+        logger.error(
+            "Gemini 연동 중 비정상 오류 발생. error_type=%s",
+            type(exception).__name__,
+        )
         return _build_result(DEFAULT_ANALYSIS_RESULT, is_mock=False, error="Unknown Error")

app/analysis/text/naive_bayes_analyzer.py

@@ -1,18 +1,13 @@
-import os
 import re
 import logging
-from pathlib import Path
 
 from app.analysis.risk_policy import determine_text_risk_grade
+from app.core.config import settings
 
 logger = logging.getLogger(__name__)
 
-# 프로젝트 루트 기준 사전 학습된 아티팩트 위치 (data_science/SMSModel/train_sms.py 산출물)
-_BASE_DIR = Path(__file__).resolve().parents[3]
-_DEFAULT_MODEL_DIR = _BASE_DIR / "data_science" / "SMSModel"
-
-MODEL_PATH = Path(os.getenv("NAIVE_BAYES_MODEL_PATH", str(_DEFAULT_MODEL_DIR / "phishing_model_artifact.pkl")))
-VECTORIZER_PATH = Path(os.getenv("NAIVE_BAYES_VECTORIZER_PATH", str(_DEFAULT_MODEL_DIR / "phishing_vectorizer.pkl")))
+MODEL_PATH = settings.NAIVE_BAYES_MODEL_PATH
+VECTORIZER_PATH = settings.NAIVE_BAYES_VECTORIZER_PATH
 
 # --- 전처리 정규식 : data_science/SMSModel/train_sms.py의 정규화/피처 추출 로직과 반드시 동일하게 유지 ---
 # (학습 시 벡터라이저가 본 입력 분포와 서빙 시 입력 분포가 어긋나면 모델이 무의미해짐)
@@ -92,9 +87,12 @@ def _load_artifacts() -> None:
         _classes = artifact["classes"]
         _vectorizer = joblib.load(VECTORIZER_PATH)
         logger.info(f"[NaiveBayes] 모델 로드 완료 (threshold={_threshold})")
-    except Exception as e:
-        _load_error = str(e)
-        logger.error(f"[NaiveBayes] 모델 로드 실패: {_load_error}")
+    except Exception as exception:
+        _load_error = type(exception).__name__
+        logger.error(
+            "[NaiveBayes] 모델 로드 실패. error_type=%s",
+            _load_error,
+        )
 
 
 def is_model_loaded() -> bool:
@@ -139,8 +137,11 @@ async def analyze_text_with_naive_bayes(text: str) -> dict:
                 "error_message": None
             }
         }
-    except Exception as e:
-        logger.error(f"[NaiveBayes] 추론 중 비정상 에러 발생: {str(e)}")
+    except Exception as exception:
+        logger.error(
+            "[NaiveBayes] 추론 중 비정상 오류 발생. error_type=%s",
+            type(exception).__name__,
+        )
         return {
             "engine": "naive_bayes",
             "is_available": False,

app/analysis/url/analyzer.py

@@ -1,8 +1,8 @@
 import logging
-import os
 from typing import ClassVar
 
 from app.analysis.ports import UrlSecurityProvider
+from app.core.config import settings
 from app.infrastructure.google_safe_browsing.client import (
     GoogleSafeBrowsingClient,
 )
@@ -12,10 +12,7 @@
 
 logger = logging.getLogger(__name__)
 
-MOCK_ENABLED = (
-    os.getenv("MOCK_SECURITY_API", "False").lower()
-    in ("true", "1", "t")
-)
+MOCK_ENABLED = settings.MOCK_SECURITY_API
 
 # Google Safe Browsing(1차)과 VirusTotal(2차 백업)을 제어하는 하이브리드 URL 분석 코어 엔진
 class HybridUrlAnalyzer:
@@ -50,7 +47,7 @@ def _is_unavailable(result: dict) -> bool:
     async def scan_url(self, traced_url: str) -> dict:
         # 쉘 환경변수에 따른 MOCK 모드 분기 로직 정상화
         if MOCK_ENABLED:
-            logger.info(f"[MOCK MODE] 하이브리드 URL 스캔 -> Target: {traced_url}")
+            logger.info("[MOCK MODE] 하이브리드 URL 스캔 시작")
             return {
                 "is_malicious": True,
                 "url_risk_score": 0.85,
@@ -237,9 +234,10 @@ async def _scan_gsb(
                 traced_url
             )
 
-        except Exception:
-            logger.exception(
-                "[Hybrid URL] GSB 호출 중 예외 발생"
+        except Exception as exception:
+            logger.error(
+                "[Hybrid URL] GSB 호출 중 예외 발생. error_type=%s",
+                type(exception).__name__,
             )
             return {
                 "is_malicious": False,
@@ -258,9 +256,10 @@ async def _scan_virustotal(
                 traced_url
             )
 
-        except Exception:
-            logger.exception(
-                "[Hybrid URL] VirusTotal 호출 중 예외 발생"
+        except Exception as exception:
+            logger.error(
+                "[Hybrid URL] VirusTotal 호출 중 예외 발생. error_type=%s",
+                type(exception).__name__,
             )
             return {
                 "is_malicious": False,