Upload 12 files

Dockerfile

@@ -30,7 +30,8 @@ RUN pip install --no-cache-dir \
     python-frontmatter \
     pyyaml \
     click \
-    aiohttp
+    aiohttp \
+    PyJWT
 
 # 애플리케이션 코드 복사
 COPY . .

pyproject.toml

@@ -26,6 +26,7 @@ uvicorn = "^0.27.0"
 aiohttp = "^3.9.0"
 sentence-transformers = "^2.2.0"
 torch = "^2.0.0"
+PyJWT = "^2.8.0"
 
 
 [tool.poetry.group.dev.dependencies]

src/auth/__init__.py

@@ -0,0 +1,23 @@
+"""
+User Authentication Module
+==========================
+
+Supabase Auth 기반 사용자 인증 및 세션 관리.
+
+지원 인증 방식:
+    - OAuth2 Authorization Code Flow (GPTs Actions용)
+    - Magic Link + 인증코드 (Claude Desktop/MCP 클라이언트용)
+
+주요 컴포넌트:
+    - OAuthProvider: GPTs용 OAuth2 엔드포인트 제공
+    - MagicLinkAuth: MCP 클라이언트용 이메일 인증
+    - UserSessionManager: 사용자 세션 토큰 관리 (MCP SessionManager와 별개)
+    - TokenValidator: JWT 토큰 검증
+"""
+
+from .config import SUPPORTED_CHAINS, AUTH_CODE_TTL_SECONDS
+
+__all__ = [
+    "SUPPORTED_CHAINS",
+    "AUTH_CODE_TTL_SECONDS",
+]

src/auth/config.py

@@ -0,0 +1,113 @@
+"""
+Authentication Configuration
+=============================
+
+인증 관련 환경변수 및 상수 정의.
+"""
+
+import os
+import logging
+
+logger = logging.getLogger("eodi.auth.config")
+
+# =============================================================================
+# 환경변수 로드
+# =============================================================================
+
+# Supabase 설정 (기존 - 재사용)
+SUPABASE_URL = os.getenv("SUPABASE_URL")
+SUPABASE_KEY = os.getenv("SUPABASE_KEY")
+
+# Supabase JWT Secret (토큰 검증용)
+# 위치: Supabase Dashboard > Settings > API > JWT Settings > JWT Secret
+SUPABASE_JWT_SECRET = os.getenv("SUPABASE_JWT_SECRET")
+
+# OAuth2 Provider 설정 (GPTs용)
+OAUTH_CLIENT_ID = os.getenv("OAUTH_CLIENT_ID", "eodi-gpts")
+OAUTH_CLIENT_SECRET = os.getenv("OAUTH_CLIENT_SECRET")
+
+# 서버 URL (콜백용)
+SERVER_BASE_URL = os.getenv(
+    "SERVER_BASE_URL",
+    "https://lovelymango-eodi-mcp.hf.space"
+)
+
+# Magic Link 리다이렉트 URL
+MAGIC_LINK_REDIRECT_URL = f"{SERVER_BASE_URL}/auth/callback"
+
+# =============================================================================
+# 상수
+# =============================================================================
+
+# 인증 코드 설정
+AUTH_CODE_TTL_SECONDS = 180  # 3분
+AUTH_CODE_LENGTH = 6
+AUTH_CODE_MAX_ATTEMPTS = 5  # 최대 시도 횟수 (brute force 방지)
+
+# OAuth State TTL
+OAUTH_STATE_TTL_SECONDS = 600  # 10분
+
+# 지원 체인 및 등급
+SUPPORTED_CHAINS = {
+    "HILTON": ["Member", "Silver", "Gold", "Diamond"],
+    "MARRIOTT": ["Member", "Silver", "Gold", "Platinum", "Titanium", "Ambassador"],
+    "IHG": ["Member", "Silver", "Gold", "Platinum", "Diamond"],
+    "ACCOR": ["Member", "Silver", "Gold", "Platinum", "Diamond"],
+    "HYATT": ["Member", "Discoverist", "Explorist", "Globalist"],
+}
+
+# 등급 정규화 매핑 (사용자 입력 → 정식 명칭)
+TIER_ALIASES = {
+    # 힐튼
+    "골드": "Gold",
+    "다이아": "Diamond",
+    "다이아몬드": "Diamond",
+    "실버": "Silver",
+    # 메리어트
+    "플래티넘": "Platinum",
+    "플래": "Platinum",
+    "티타늄": "Titanium",
+    "앰버서더": "Ambassador",
+    # IHG
+    "다이아": "Diamond",
+    # 하얏트
+    "글로벌리스트": "Globalist",
+    "익스플로리스트": "Explorist",
+}
+
+
+# =============================================================================
+# 환경변수 검증
+# =============================================================================
+
+def validate_config():
+    """필수 환경변수 검증. 서버 시작 시 호출."""
+    missing = []
+    
+    if not SUPABASE_URL:
+        missing.append("SUPABASE_URL")
+    if not SUPABASE_KEY:
+        missing.append("SUPABASE_KEY")
+    
+    # JWT Secret은 토큰 검증에만 필요, 없으면 경고만
+    if not SUPABASE_JWT_SECRET:
+        logger.warning(
+            "SUPABASE_JWT_SECRET이 설정되지 않았습니다. "
+            "서버 측 JWT 검증이 비활성화됩니다. "
+            "설정 위치: Supabase Dashboard > Settings > API > JWT Settings"
+        )
+    
+    # OAuth Secret은 GPTs 연동에만 필요, 없으면 경고만
+    if not OAUTH_CLIENT_SECRET:
+        logger.warning(
+            "OAUTH_CLIENT_SECRET이 설정되지 않았습니다. "
+            "GPTs OAuth 연동이 비활성화됩니다."
+        )
+    
+    if missing:
+        raise ValueError(
+            f"필수 환경변수가 누락되었습니다: {', '.join(missing)}. "
+            "HF Space: Settings > Repository secrets에서 설정하세요."
+        )
+    
+    logger.info("인증 설정 검증 완료")

src/auth/magic_link.py

@@ -0,0 +1,276 @@
+"""
+Magic Link Authentication
+=========================
+
+Claude Desktop 등 OAuth를 지원하지 않는 MCP 클라이언트용.
+
+흐름:
+1. 사용자가 이메일 입력 → user_request_auth Tool 호출
+2. Supabase Magic Link 발송
+3. 사용자가 이메일에서 링크 클릭 → /auth/callback 페이지
+4. 페이지에서 6자리 코드 표시
+5. 사용자가 GPT/Claude로 돌아와서 코드 입력 → user_verify_code Tool 호출
+6. 코드 검증 → 세션 토큰 반환
+"""
+
+import secrets
+import time
+import logging
+from typing import Dict, Any, Optional, Tuple
+from collections import defaultdict
+
+import httpx
+
+from .config import (
+    SUPABASE_URL,
+    SUPABASE_KEY,
+    MAGIC_LINK_REDIRECT_URL,
+    AUTH_CODE_TTL_SECONDS,
+    AUTH_CODE_LENGTH,
+    AUTH_CODE_MAX_ATTEMPTS,
+)
+
+logger = logging.getLogger("eodi.auth.magic_link")
+
+
+class AuthCodeStore:
+    """
+    인증 코드 임시 저장소.
+    
+    Magic Link 클릭 후 표시되는 6자리 코드 관리.
+    Rate limiting 및 brute force 방지 포함.
+    """
+    
+    def __init__(self, ttl_seconds: int = AUTH_CODE_TTL_SECONDS):
+        self.ttl = ttl_seconds
+        self._codes: Dict[str, Dict[str, Any]] = {}
+        self._attempts: Dict[str, int] = defaultdict(int)  # IP별 시도 횟수
+        self._last_cleanup = time.time()
+    
+    def _cleanup(self):
+        """만료된 코드 및 시도 횟수 정리"""
+        now = time.time()
+        if now - self._last_cleanup < 60:
+            return
+        
+        self._last_cleanup = now
+        
+        # 만료된 코드 삭제
+        expired = [
+            c for c, data in self._codes.items()
+            if now - data["created_at"] > self.ttl
+        ]
+        for c in expired:
+            del self._codes[c]
+        
+        # 오래된 시도 횟수 리셋
+        self._attempts.clear()
+    
+    def generate_code(
+        self, 
+        user_id: str, 
+        email: str, 
+        access_token: str, 
+        refresh_token: str
+    ) -> str:
+        """
+        6자리 인증 코드 생성.
+        
+        Args:
+            user_id: Supabase Auth UUID
+            email: 사용자 이메일
+            access_token: Supabase access_token
+            refresh_token: Supabase refresh_token
+            
+        Returns:
+            6자리 숫자 코드
+        """
+        self._cleanup()
+        
+        # 같은 user_id의 기존 코드 삭제 (중복 방지)
+        existing = [
+            c for c, data in self._codes.items()
+            if data["user_id"] == user_id
+        ]
+        for c in existing:
+            del self._codes[c]
+        
+        # 6자리 숫자 코드 생성
+        code = "".join([str(secrets.randbelow(10)) for _ in range(AUTH_CODE_LENGTH)])
+        
+        self._codes[code] = {
+            "user_id": user_id,
+            "email": email,
+            "access_token": access_token,
+            "refresh_token": refresh_token,
+            "created_at": time.time(),
+        }
+        
+        return code
+    
+    def verify_code(
+        self, 
+        code: str, 
+        client_ip: str = "unknown"
+    ) -> Tuple[bool, Optional[Dict[str, Any]], Optional[str]]:
+        """
+        코드 검증 및 소비.
+        
+        Args:
+            code: 6자리 인증 코드
+            client_ip: 클라이언트 IP (rate limiting용)
+            
+        Returns:
+            (유효여부, code_data, error_message)
+        """
+        self._cleanup()
+        
+        # Rate limiting 체크
+        if self._attempts[client_ip] >= AUTH_CODE_MAX_ATTEMPTS:
+            logger.warning(f"Rate limit exceeded for IP: {client_ip[:20]}...")
+            return False, None, f"너무 많은 시도입니다. {self.ttl}초 후 다시 시도해주세요."
+        
+        self._attempts[client_ip] += 1
+        
+        # 코드 정규화 (공백 제거 등)
+        code = code.strip().replace(" ", "").replace("-", "")
+        
+        if code not in self._codes:
+            return False, None, "코드가 일치하지 않습니다. 다시 확인해주세요."
+        
+        code_data = self._codes.pop(code)
+        
+        if time.time() - code_data["created_at"] > self.ttl:
+            return False, None, "코드가 만료되었습니다. 새 코드를 요청해주세요."
+        
+        # 성공 시 시도 횟수 리셋
+        self._attempts[client_ip] = 0
+        
+        return True, code_data, None
+
+
+class MagicLinkAuth:
+    """
+    Magic Link 인증 핸들러.
+    """
+    
+    def __init__(self):
+        self.code_store = AuthCodeStore()
+        self._http_client = None
+    
+    async def _get_client(self) -> httpx.AsyncClient:
+        """HTTP 클라이언트 (재사용)"""
+        if self._http_client is None:
+            self._http_client = httpx.AsyncClient(timeout=30)
+        return self._http_client
+    
+    async def send_magic_link(self, email: str) -> Tuple[bool, Optional[str]]:
+        """
+        Magic Link ���메일 발송.
+        
+        Args:
+            email: 사용자 이메일
+            
+        Returns:
+            (성공여부, error_message)
+        """
+        try:
+            client = await self._get_client()
+            
+            response = await client.post(
+                f"{SUPABASE_URL}/auth/v1/magiclink",
+                json={
+                    "email": email,
+                    "options": {
+                        "redirectTo": MAGIC_LINK_REDIRECT_URL
+                    }
+                },
+                headers={
+                    "apikey": SUPABASE_KEY,
+                    "Content-Type": "application/json"
+                }
+            )
+            
+            if response.status_code == 200:
+                logger.info(f"Magic Link 발송 성공: {email[:3]}***")
+                return True, None
+            else:
+                error_text = response.text[:200]
+                logger.error(f"Magic Link 발송 실패: {response.status_code} - {error_text}")
+                return False, f"이메일 발송에 실패했습니다: {error_text}"
+                
+        except Exception as e:
+            logger.error(f"Magic Link 발송 오류: {e}")
+            return False, f"이메일 발송 중 오류가 발생했습니다: {e}"
+    
+    def generate_auth_code(
+        self,
+        user_id: str,
+        email: str,
+        access_token: str,
+        refresh_token: str
+    ) -> str:
+        """
+        인증 성공 후 6자리 코드 생성.
+        
+        Magic Link 클릭 → /auth/callback에서 호출.
+        """
+        return self.code_store.generate_code(user_id, email, access_token, refresh_token)
+    
+    def verify_auth_code(
+        self, 
+        code: str, 
+        client_ip: str = "unknown"
+    ) -> Tuple[bool, Optional[str], Optional[str], Optional[str]]:
+        """
+        인증 코드 검증.
+        
+        Args:
+            code: 6자리 인증 코드
+            client_ip: 클라이언트 IP
+            
+        Returns:
+            (유효여부, session_token, email_masked, error_message)
+        """
+        is_valid, code_data, error = self.code_store.verify_code(code, client_ip)
+        
+        if not is_valid:
+            return False, None, None, error
+        
+        # 세션 생성
+        from .session_manager import get_user_session_manager
+        
+        session_manager = get_user_session_manager()
+        session_token = session_manager.create_session(
+            user_id=code_data["user_id"],
+            email=code_data["email"],
+            access_token=code_data["access_token"],
+            refresh_token=code_data["refresh_token"],
+            expires_in=3600
+        )
+        
+        # 이메일 마스킹
+        email = code_data["email"]
+        if "@" in email:
+            local, domain = email.split("@", 1)
+            if len(local) <= 2:
+                masked = local[0] + "*"
+            else:
+                masked = local[0] + "*" * (len(local) - 2) + local[-1]
+            email_masked = f"{masked}@{domain}"
+        else:
+            email_masked = email
+        
+        return True, session_token, email_masked, None
+
+
+# 전역 인스턴스 (Lazy loading)
+_magic_link_auth = None
+
+
+def get_magic_link_auth() -> MagicLinkAuth:
+    """MagicLinkAuth 싱글톤 인스턴스 반환"""
+    global _magic_link_auth
+    if _magic_link_auth is None:
+        _magic_link_auth = MagicLinkAuth()
+    return _magic_link_auth

src/auth/oauth_provider.py

@@ -0,0 +1,379 @@
+"""
+OAuth2 Provider for GPTs Actions
+================================
+
+Supabase Auth를 감싸는 OAuth2 Provider 레이어.
+GPTs Actions가 요구하는 /authorize, /token 엔드포인트 제공.
+"""
+
+import secrets
+import time
+import logging
+from typing import Dict, Any, Optional, Tuple
+from urllib.parse import urlencode
+
+import httpx
+
+from .config import (
+    SUPABASE_URL,
+    SUPABASE_KEY,
+    OAUTH_CLIENT_ID,
+    OAUTH_CLIENT_SECRET,
+    SERVER_BASE_URL,
+    OAUTH_STATE_TTL_SECONDS,
+)
+
+logger = logging.getLogger("eodi.auth.oauth")
+
+
+class OAuthStateManager:
+    """
+    OAuth state 파라미터 관리 (CSRF 방지).
+    """
+    
+    MAX_STATES = 1000
+    
+    def __init__(self, ttl_seconds: int = OAUTH_STATE_TTL_SECONDS):
+        self.ttl = ttl_seconds
+        self._states: Dict[str, Dict[str, Any]] = {}
+        self._last_cleanup = time.time()
+    
+    def _cleanup_expired(self):
+        """만료된 state 정리"""
+        now = time.time()
+        if now - self._last_cleanup < 60:
+            return
+        
+        self._last_cleanup = now
+        expired = [
+            s for s, data in self._states.items()
+            if now - data["created_at"] > self.ttl
+        ]
+        for s in expired:
+            del self._states[s]
+        
+        if expired:
+            logger.debug(f"만료된 state {len(expired)}개 정리됨")
+    
+    def create_state(self, redirect_uri: str, scope: str = "profile") -> str:
+        """
+        새 state 생성.
+        
+        Args:
+            redirect_uri: GPTs 콜백 URL
+            scope: 요청 스코프
+            
+        Returns:
+            state 문자열
+        """
+        self._cleanup_expired()
+        
+        # 메모리 보호
+        if len(self._states) >= self.MAX_STATES:
+            oldest = min(self._states.items(), key=lambda x: x[1]["created_at"])
+            del self._states[oldest[0]]
+        
+        state = secrets.token_urlsafe(32)
+        self._states[state] = {
+            "redirect_uri": redirect_uri,
+            "scope": scope,
+            "created_at": time.time(),
+        }
+        return state
+    
+    def validate_and_consume(self, state: str) -> Tuple[bool, Optional[Dict[str, Any]]]:
+        """
+        state 검증 및 소비 (일회용).
+        
+        Args:
+            state: 검증할 state
+            
+        Returns:
+            (유효여부, state_data)
+        """
+        self._cleanup_expired()
+        
+        if state not in self._states:
+            logger.warning(f"State not found: {state[:10]}...")
+            return False, None
+        
+        state_data = self._states.pop(state)
+        
+        if time.time() - state_data["created_at"] > self.ttl:
+            logger.warning(f"State expired: {state[:10]}...")
+            return False, None
+        
+        return True, state_data
+
+
+class AuthCodeStore:
+    """
+    OAuth Authorization Code 임시 저장소.
+    
+    Supabase 인증 완료 후 → GPTs에 전달할 code 관리.
+    """
+    
+    CODE_TTL_SECONDS = 60  # 1분 (매우 짧게)
+    MAX_CODES = 500
+    
+    def __init__(self):
+        self._codes: Dict[str, Dict[str, Any]] = {}
+        self._last_cleanup = time.time()
+    
+    def _cleanup(self):
+        now = time.time()
+        if now - self._last_cleanup < 30:
+            return
+        self._last_cleanup = now
+        
+        expired = [
+            c for c, data in self._codes.items()
+            if now - data["created_at"] > self.CODE_TTL_SECONDS
+        ]
+        for c in expired:
+            del self._codes[c]
+    
+    def create_code(
+        self,
+        user_id: str,
+        email: str,
+        access_token: str,
+        refresh_token: str,
+        redirect_uri: str
+    ) -> str:
+        """Authorization code 생성"""
+        self._cleanup()
+        
+        if len(self._codes) >= self.MAX_CODES:
+            oldest = min(self._codes.items(), key=lambda x: x[1]["created_at"])
+            del self._codes[oldest[0]]
+        
+        code = secrets.token_urlsafe(32)
+        self._codes[code] = {
+            "user_id": user_id,
+            "email": email,
+            "access_token": access_token,
+            "refresh_token": refresh_token,
+            "redirect_uri": redirect_uri,
+            "created_at": time.time(),
+        }
+        return code
+    
+    def consume_code(
+        self, 
+        code: str, 
+        redirect_uri: str
+    ) -> Tuple[bool, Optional[Dict[str, Any]]]:
+        """
+        Code 검증 및 소비.
+        
+        Returns:
+            (유효여부, code_data)
+        """
+        self._cleanup()
+        
+        if code not in self._codes:
+            return False, None
+        
+        code_data = self._codes.pop(code)
+        
+        # redirect_uri 검증
+        if code_data["redirect_uri"] != redirect_uri:
+            logger.warning("redirect_uri mismatch")
+            return False, None
+        
+        if time.time() - code_data["created_at"] > self.CODE_TTL_SECONDS:
+            return False, None
+        
+        return True, code_data
+
+
+class OAuthProvider:
+    """
+    OAuth2 Authorization Code Flow Provider.
+    
+    GPTs Actions 연동용.
+    
+    흐름:
+    1. GPTs → /oauth/authorize → Supabase 로그인 페이지
+    2. 사용자 로그인 → Supabase → /oauth/callback
+    3. /oauth/callback → GPTs callback URL로 code 전달
+    4. GPTs → /oauth/token → access_token 발급
+    """
+    
+    def __init__(self):
+        self.state_manager = OAuthStateManager()
+        self.code_store = AuthCodeStore()
+        self._http_client = None
+    
+    @property
+    def is_configured(self) -> bool:
+        """OAuth 설정 완료 여부"""
+        return bool(OAUTH_CLIENT_SECRET)
+    
+    async def _get_client(self) -> httpx.AsyncClient:
+        """HTTP 클라이언트 (재사용)"""
+        if self._http_client is None:
+            self._http_client = httpx.AsyncClient(timeout=30)
+        return self._http_client
+    
+    def get_authorization_url(
+        self,
+        redirect_uri: str,
+        scope: str = "profile"
+    ) -> Tuple[str, str]:
+        """
+        내부 로그인 페이지 URL 생성.
+        
+        Args:
+            redirect_uri: GPTs 콜백 URL
+            scope: 요청 스코프
+            
+        Returns:
+            (authorization_url, state)
+        """
+        state = self.state_manager.create_state(redirect_uri, scope)
+        
+        # 자체 로그인 페이지로 리다이렉트
+        params = {
+            "redirect_uri": redirect_uri,
+            "state": state,
+            "scope": scope,
+        }
+        
+        auth_url = f"{SERVER_BASE_URL}/auth/login?{urlencode(params)}"
+        return auth_url, state
+    
+    async def handle_supabase_callback(
+        self,
+        access_token: str,
+        refresh_token: str,
+        state: str
+    ) -> Tuple[bool, Optional[str], Optional[str], Optional[str]]:
+        """
+        Supabase 로그인 완료 후 처리.
+        
+        Args:
+            access_token: Supabase access_token
+            refresh_token: Supabase refresh_token
+            state: OAuth state
+            
+        Returns:
+            (성공여부, authorization_code, redirect_uri, error_message)
+        """
+        # state 검증
+        is_valid, state_data = self.state_manager.validate_and_consume(state)
+        if not is_valid:
+            return False, None, None, "Invalid or expired state"
+        
+        redirect_uri = state_data["redirect_uri"]
+        
+        # 토큰에서 사용자 정보 추출
+        try:
+            client = await self._get_client()
+            response = await client.get(
+                f"{SUPABASE_URL}/auth/v1/user",
+                headers={"Authorization": f"Bearer {access_token}"}
+            )
+            
+            if response.status_code != 200:
+                return False, None, None, "Failed to get user info"
+            
+            user_data = response.json()
+            user_id = user_data.get("id")
+            email = user_data.get("email")
+            
+        except Exception as e:
+            logger.error(f"User info 조회 실패: {e}")
+            return False, None, None, str(e)
+        
+        # Authorization code 생성
+        code = self.code_store.create_code(
+            user_id=user_id,
+            email=email,
+            access_token=access_token,
+            refresh_token=refresh_token,
+            redirect_uri=redirect_uri
+        )
+        
+        return True, code, redirect_uri, None
+    
+    async def exchange_code_for_token(
+        self,
+        code: str,
+        redirect_uri: str,
+        client_id: str,
+        client_secret: str
+    ) -> Tuple[bool, Optional[Dict[str, Any]], Optional[str]]:
+        """
+        Authorization code를 access_token으로 교환.
+        
+        Args:
+            code: Authorization code
+            redirect_uri: 원래 redirect_uri
+            client_id: OAuth 클라이언트 ID
+            client_secret: OAuth 클라이언트 Secret
+            
+        Returns:
+            (성공여부, token_response, error_message)
+        """
+        # 클라이언트 인증
+        if client_id != OAUTH_CLIENT_ID:
+            return False, None, "Invalid client_id"
+        if client_secret != OAUTH_CLIENT_SECRET:
+            return False, None, "Invalid client_secret"
+        
+        # code 검증 및 소비
+        is_valid, code_data = self.code_store.consume_code(code, redirect_uri)
+        if not is_valid:
+            return False, None, "Invalid or expired code"
+        
+        # 세션 생성
+        from .session_manager import get_user_session_manager
+        
+        session_manager = get_user_session_manager()
+        session_token = session_manager.create_session(
+            user_id=code_data["user_id"],
+            email=code_data["email"],
+            access_token=code_data["access_token"],
+            refresh_token=code_data["refresh_token"],
+            expires_in=3600
+        )
+        
+        # GPTs에 반환할 토큰 응답
+        # 주의: 여기서 반환하는 access_token은 우리의 session_token
+        token_response = {
+            "access_token": session_token,
+            "token_type": "Bearer",
+            "expires_in": 3600,
+            "refresh_token": secrets.token_urlsafe(32),  # 더미
+            "scope": "profile"
+        }
+        
+        return True, token_response, None
+    
+    async def refresh_token(
+        self,
+        refresh_token: str,
+        client_id: str,
+        client_secret: str
+    ) -> Tuple[bool, Optional[Dict[str, Any]], Optional[str]]:
+        """
+        Refresh token으로 새 access_token 발급.
+        
+        현재 구현에서는 미지원 (MVP).
+        """
+        # TODO: Supabase refresh token으로 새 토큰 발급 구현
+        return False, None, "Token refresh not implemented yet"
+
+
+# 전역 인스턴스 (Lazy loading)
+_oauth_provider = None
+
+
+def get_oauth_provider() -> OAuthProvider:
+    """OAuthProvider 싱글톤 인스턴스 반환"""
+    global _oauth_provider
+    if _oauth_provider is None:
+        _oauth_provider = OAuthProvider()
+    return _oauth_provider

src/auth/routes.py

@@ -0,0 +1,568 @@
+"""
+Authentication HTTP Routes
+==========================
+
+OAuth2 및 Magic Link 인증 엔드포인트.
+server_streamable.py의 routes에 확장됩니다.
+"""
+
+import logging
+from urllib.parse import urlencode
+
+from starlette.routing import Route
+from starlette.requests import Request
+from starlette.responses import JSONResponse, RedirectResponse, HTMLResponse
+
+from .oauth_provider import get_oauth_provider
+from .magic_link import get_magic_link_auth
+from .config import SERVER_BASE_URL, SUPABASE_URL
+
+logger = logging.getLogger("eodi.auth.routes")
+
+
+# =============================================================================
+# OAuth2 엔드포인트 (GPTs용)
+# =============================================================================
+
+async def oauth_authorize(request: Request) -> RedirectResponse:
+    """
+    GET /oauth/authorize
+    
+    GPTs OAuth 시작점. 로그인 페이지로 리다이렉트.
+    """
+    oauth_provider = get_oauth_provider()
+    
+    if not oauth_provider.is_configured:
+        return JSONResponse({
+            "error": "OAuth not configured",
+            "message": "OAUTH_CLIENT_SECRET 환경변수가 설정되지 않았습니다."
+        }, status_code=503)
+    
+    redirect_uri = request.query_params.get("redirect_uri")
+    scope = request.query_params.get("scope", "profile")
+    gpts_state = request.query_params.get("state", "")
+    
+    if not redirect_uri:
+        return JSONResponse({"error": "redirect_uri is required"}, status_code=400)
+    
+    # 내부 state 생성 (redirect_uri 저장용)
+    auth_url, internal_state = oauth_provider.get_authorization_url(redirect_uri, scope)
+    
+    # GPTs state도 함께 전달
+    if gpts_state:
+        auth_url += f"&gpts_state={gpts_state}"
+    
+    logger.info("OAuth authorize: redirect to login page")
+    return RedirectResponse(auth_url, status_code=302)
+
+
+async def oauth_callback(request: Request):
+    """
+    GET /oauth/callback
+    
+    (내부 엔드포인트 - 직접 사용되지 않음)
+    """
+    return JSONResponse({"message": "Use /auth/callback instead"})
+
+
+async def oauth_token(request: Request) -> JSONResponse:
+    """
+    POST /oauth/token
+    
+    Authorization code → Access token 교환.
+    """
+    oauth_provider = get_oauth_provider()
+    
+    if not oauth_provider.is_configured:
+        return JSONResponse({"error": "OAuth not configured"}, status_code=503)
+    
+    # form-urlencoded 또는 JSON 바디 처리
+    content_type = request.headers.get("content-type", "")
+    
+    if "application/x-www-form-urlencoded" in content_type:
+        form = await request.form()
+        grant_type = form.get("grant_type")
+        code = form.get("code")
+        redirect_uri = form.get("redirect_uri")
+        client_id = form.get("client_id")
+        client_secret = form.get("client_secret")
+        refresh_token = form.get("refresh_token")
+    else:
+        try:
+            body = await request.json()
+            grant_type = body.get("grant_type")
+            code = body.get("code")
+            redirect_uri = body.get("redirect_uri")
+            client_id = body.get("client_id")
+            client_secret = body.get("client_secret")
+            refresh_token = body.get("refresh_token")
+        except Exception:
+            return JSONResponse({"error": "invalid_request"}, status_code=400)
+    
+    if grant_type == "authorization_code":
+        if not code or not redirect_uri:
+            return JSONResponse({"error": "invalid_request"}, status_code=400)
+        
+        success, token_response, error = await oauth_provider.exchange_code_for_token(
+            code=code,
+            redirect_uri=redirect_uri,
+            client_id=client_id,
+            client_secret=client_secret
+        )
+        
+        if success:
+            return JSONResponse(token_response)
+        else:
+            return JSONResponse(
+                {"error": "invalid_grant", "error_description": error}, 
+                status_code=400
+            )
+    
+    elif grant_type == "refresh_token":
+        success, token_response, error = await oauth_provider.refresh_token(
+            refresh_token=refresh_token,
+            client_id=client_id,
+            client_secret=client_secret
+        )
+        
+        if success:
+            return JSONResponse(token_response)
+        else:
+            return JSONResponse(
+                {"error": "invalid_grant", "error_description": error}, 
+                status_code=400
+            )
+    
+    else:
+        return JSONResponse({"error": "unsupported_grant_type"}, status_code=400)
+
+
+# =============================================================================
+# Magic Link 엔드포인트 (MCP 클라이언트용)
+# =============================================================================
+
+async def auth_login_page(request: Request) -> HTMLResponse:
+    """
+    GET /auth/login
+    
+    로그인 페이지. 이메일 입력 → Magic Link 발��.
+    """
+    redirect_uri = request.query_params.get("redirect_uri", "")
+    state = request.query_params.get("state", "")
+    gpts_state = request.query_params.get("gpts_state", "")
+    
+    html = f"""
+    <!DOCTYPE html>
+    <html>
+    <head>
+        <title>Eodi 로그인</title>
+        <meta charset="utf-8">
+        <meta name="viewport" content="width=device-width, initial-scale=1">
+        <style>
+            body {{ 
+                font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
+                max-width: 400px; 
+                margin: 50px auto; 
+                padding: 20px;
+                background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
+                min-height: 100vh;
+                box-sizing: border-box;
+            }}
+            .container {{
+                background: white;
+                padding: 40px 30px;
+                border-radius: 16px;
+                box-shadow: 0 10px 40px rgba(0,0,0,0.2);
+            }}
+            h1 {{ 
+                color: #333; 
+                font-size: 28px; 
+                margin-bottom: 8px;
+                text-align: center;
+            }}
+            .subtitle {{
+                color: #666;
+                text-align: center;
+                margin-bottom: 30px;
+            }}
+            input[type="email"] {{
+                width: 100%;
+                padding: 14px;
+                border: 2px solid #e0e0e0;
+                border-radius: 10px;
+                font-size: 16px;
+                margin-bottom: 16px;
+                box-sizing: border-box;
+                transition: border-color 0.2s;
+            }}
+            input[type="email"]:focus {{
+                outline: none;
+                border-color: #667eea;
+            }}
+            button {{
+                width: 100%;
+                padding: 14px;
+                background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
+                color: white;
+                border: none;
+                border-radius: 10px;
+                font-size: 16px;
+                font-weight: 600;
+                cursor: pointer;
+                transition: transform 0.2s, box-shadow 0.2s;
+            }}
+            button:hover {{ 
+                transform: translateY(-2px);
+                box-shadow: 0 5px 20px rgba(102, 126, 234, 0.4);
+            }}
+            button:disabled {{
+                background: #ccc;
+                transform: none;
+                box-shadow: none;
+            }}
+            .message {{ margin-top: 16px; text-align: center; }}
+            .success {{ color: #10b981; }}
+            .error {{ color: #ef4444; }}
+            .info {{ 
+                color: #6b7280; 
+                font-size: 13px; 
+                margin-top: 20px;
+                text-align: center;
+            }}
+        </style>
+    </head>
+    <body>
+        <div class="container">
+            <h1>🏨 Eodi</h1>
+            <p class="subtitle">여행 혜택의 모든 것</p>
+            
+            <form id="loginForm">
+                <input type="hidden" name="redirect_uri" value="{redirect_uri}">
+                <input type="hidden" name="state" value="{state}">
+                <input type="hidden" name="gpts_state" value="{gpts_state}">
+                <input type="email" name="email" placeholder="이메일 주소" required>
+                <button type="submit" id="submitBtn">로그인 링크 받기</button>
+            </form>
+            
+            <div id="message"></div>
+            <p class="info">로그인 링크는 10분간 유효합니다.</p>
+        </div>
+        
+        <script>
+            document.getElementById('loginForm').addEventListener('submit', async (e) => {{
+                e.preventDefault();
+                const form = e.target;
+                const email = form.email.value;
+                const button = document.getElementById('submitBtn');
+                const messageDiv = document.getElementById('message');
+                
+                button.disabled = true;
+                button.textContent = '전송 중...';
+                messageDiv.innerHTML = '';
+                
+                try {{
+                    const response = await fetch('/auth/send-link', {{
+                        method: 'POST',
+                        headers: {{'Content-Type': 'application/json'}},
+                        body: JSON.stringify({{
+                            email: email,
+                            redirect_uri: form.redirect_uri.value,
+                            state: form.state.value,
+                            gpts_state: form.gpts_state.value
+                        }})
+                    }});
+                    
+                    const data = await response.json();
+                    
+                    if (data.success) {{
+                        messageDiv.innerHTML = '<p class="message success">✅ 이메일을 확인해주세요!</p>';
+                        button.textContent = '전송 완료';
+                    }} else {{
+                        messageDiv.innerHTML = '<p class="message error">❌ ' + data.error + '</p>';
+                        button.disabled = false;
+                        button.textContent = '로그인 링크 받기';
+                    }}
+                }} catch (err) {{
+                    messageDiv.innerHTML = '<p class="message error">❌ 오류가 발생했습니다.</p>';
+                    button.disabled = false;
+                    button.textContent = '로그인 링크 받기';
+                }}
+            }});
+        </script>
+    </body>
+    </html>
+    """
+    return HTMLResponse(html)
+
+
+async def auth_send_link(request: Request) -> JSONResponse:
+    """
+    POST /auth/send-link
+    
+    Magic Link 발송 API.
+    """
+    magic_link_auth = get_magic_link_auth()
+    
+    try:
+        body = await request.json()
+        email = body.get("email")
+        
+        if not email:
+            return JSONResponse(
+                {"success": False, "error": "이메일을 입력해주세요."}, 
+                status_code=400
+            )
+        
+        success, error = await magic_link_auth.send_magic_link(email)
+        
+        if success:
+            return JSONResponse({"success": True})
+        else:
+            return JSONResponse({"success": False, "error": error}, status_code=400)
+            
+    except Exception as e:
+        logger.error(f"Magic Link 발송 오류: {e}")
+        return JSONResponse({"success": False, "error": str(e)}, status_code=500)
+
+
+async def auth_callback_page(request: Request) -> HTMLResponse:
+    """
+    GET /auth/callback
+    
+    Magic Link 클릭 후 랜딩 페이지.
+    Supabase가 URL fragment에 토큰을 포함.
+    """
+    html = """
+    <!DOCTYPE html>
+    <html>
+    <head>
+        <title>Eodi 인증 완료</title>
+        <meta charset="utf-8">
+        <meta name="viewport" content="width=device-width, initial-scale=1">
+        <style>
+            body { 
+                font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
+                max-width: 400px; 
+                margin: 50px auto; 
+                padding: 20px;
+                background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
+                min-height: 100vh;
+                box-sizing: border-box;
+            }
+            .container {
+                background: white;
+                padding: 40px 30px;
+                border-radius: 16px;
+                box-shadow: 0 10px 40px rgba(0,0,0,0.2);
+                text-align: center;
+            }
+            h1 { color: #333; font-size: 28px; margin-bottom: 10px; }
+            h2 { color: #333; font-size: 22px; margin-bottom: 10px; }
+            .code {
+                font-size: 42px;
+                font-weight: bold;
+                letter-spacing: 10px;
+                color: #667eea;
+                margin: 24px 0;
+                padding: 20px;
+                background: linear-gradient(135deg, #f0f4ff 0%, #e8ecff 100%);
+                border-radius: 12px;
+            }
+            .instruction {
+                color: #4b5563;
+                margin: 16px 0;
+                line-height: 1.6;
+            }
+            .timer {
+                color: #9ca3af;
+                font-size: 14px;
+                margin-top: 16px;
+            }
+            .error { color: #ef4444; }
+            .loading { color: #6b7280; }
+        </style>
+    </head>
+    <body>
+        <div class="container">
+            <h1>🏨 Eodi</h1>
+            <div id="content">
+                <p class="loading">인증 처리 중...</p>
+            </div>
+        </div>
+        
+        <script>
+            async function processAuth() {
+                const contentDiv = document.getElementById('content');
+                
+                // URL fragment에서 토큰 추출
+                const hash = window.location.hash.substring(1);
+                const params = new URLSearchParams(hash);
+                
+                const accessToken = params.get('access_token');
+                const refreshToken = params.get('refresh_token') || '';
+                
+                if (!accessToken) {
+                    contentDiv.innerHTML = `
+                        <h2 class="error">❌ 인증 실패</h2>
+                        <p class="instruction">링크가 만료되었거나 잘못되었습니다.<br>다시 시도해주세요.</p>
+                    `;
+                    return;
+                }
+                
+                try {
+                    const response = await fetch('/auth/process-token', {
+                        method: 'POST',
+                        headers: {'Content-Type': 'application/json'},
+                        body: JSON.stringify({
+                            access_token: accessToken,
+                            refresh_token: refreshToken
+                        })
+                    });
+                    
+                    const data = await response.json();
+                    
+                    if (data.success) {
+                        contentDiv.innerHTML = `
+                            <h2>✅ 인증 완료!</h2>
+                            <p class="instruction">아래 코드를 입력해주세요</p>
+                            <div class="code">${data.code}</div>
+                            <p class="instruction">GPT 또는 Claude로 돌아가서<br>이 코드를 입력하세요.</p>
+                            <p class="timer">⏱️ 코드는 3분간 유효합니다</p>
+                        `;
+                    } else {
+                        contentDiv.innerHTML = `
+                            <h2 class="error">❌ 오류 발생</h2>
+                            <p class="instruction">${data.error}</p>
+                        `;
+                    }
+                } catch (err) {
+                    contentDiv.innerHTML = `
+                        <h2 class="error">❌ 오류 발생</h2>
+                        <p class="instruction">잠시 후 다시 시도해주세요.</p>
+                    `;
+                }
+            }
+            
+            processAuth();
+        </script>
+    </body>
+    </html>
+    """
+    return HTMLResponse(html)
+
+
+async def auth_process_token(request: Request) -> JSONResponse:
+    """
+    POST /auth/process-token
+    
+    Supabase 토큰으로 인증 코드 생성.
+    """
+    magic_link_auth = get_magic_link_auth()
+    
+    try:
+        body = await request.json()
+        access_token = body.get("access_token")
+        refresh_token = body.get("refresh_token", "")
+        
+        if not access_token:
+            return JSONResponse(
+                {"success": False, "error": "토큰이 없습니다."}, 
+                status_code=400
+            )
+        
+        # 토큰에서 사용자 정보 추출
+        import httpx
+        
+        async with httpx.AsyncClient(timeout=30) as client:
+            response = await client.get(
+                f"{SUPABASE_URL}/auth/v1/user",
+                headers={"Authorization": f"Bearer {access_token}"}
+            )
+            
+            if response.status_code != 200:
+                return JSONResponse(
+                    {"success": False, "error": "사용자 정보를 가져올 수 없습니다."}, 
+                    status_code=400
+                )
+            
+            user_data = response.json()
+            user_id = user_data.get("id")
+            email = user_data.get("email")
+        
+        # 인증 코드 생성
+        code = magic_link_auth.generate_auth_code(
+            user_id=user_id,
+            email=email,
+            access_token=access_token,
+            refresh_token=refresh_token
+        )
+        
+        return JSONResponse({"success": True, "code": code})
+        
+    except Exception as e:
+        logger.error(f"토큰 처리 오류: {e}")
+        return JSONResponse({"success": False, "error": str(e)}, status_code=500)
+
+
+async def auth_verify_code_api(request: Request) -> JSONResponse:
+    """
+    POST /auth/verify-code
+    
+    인증 코드 검증 API (REST용).
+    """
+    magic_link_auth = get_magic_link_auth()
+    
+    try:
+        body = await request.json()
+        code = body.get("code")
+        
+        if not code:
+            return JSONResponse(
+                {"success": False, "error": "코드를 입력해주세요."}, 
+                status_code=400
+            )
+        
+        # 클라이언트 IP
+        client_ip = request.headers.get(
+            "X-Forwarded-For",
+            request.client.host if request.client else "unknown"
+        )
+        if "," in client_ip:
+            client_ip = client_ip.split(",")[0].strip()
+        
+        success, session_token, email_masked, error = magic_link_auth.verify_auth_code(
+            code, client_ip
+        )
+        
+        if success:
+            return JSONResponse({
+                "success": True,
+                "session_token": session_token,
+                "email_masked": email_masked,
+                "message": f"인증 완료! {email_masked}"
+            })
+        else:
+            return JSONResponse({"success": False, "error": error}, status_code=400)
+            
+    except Exception as e:
+        logger.error(f"코드 검증 오류: {e}")
+        return JSONResponse({"success": False, "error": str(e)}, status_code=500)
+
+
+# =============================================================================
+# 라우트 목록
+# =============================================================================
+
+auth_routes = [
+    # OAuth2 (GPTs용)
+    Route("/oauth/authorize", oauth_authorize, methods=["GET"]),
+    Route("/oauth/callback", oauth_callback, methods=["GET"]),
+    Route("/oauth/token", oauth_token, methods=["POST"]),
+    
+    # Magic Link (MCP 클라이언트용)
+    Route("/auth/login", auth_login_page, methods=["GET"]),
+    Route("/auth/send-link", auth_send_link, methods=["POST"]),
+    Route("/auth/callback", auth_callback_page, methods=["GET"]),
+    Route("/auth/process-token", auth_process_token, methods=["POST"]),
+    Route("/auth/verify-code", auth_verify_code_api, methods=["POST"]),
+]

src/auth/session_manager.py

@@ -0,0 +1,232 @@
+"""
+User Session Manager
+====================
+
+사용자 인증 세션 관리.
+
+주의: 이 모듈은 MCP 프로토콜 세션(server_streamable.py의 SessionManager)과 별개입니다.
+- MCP SessionManager: MCP 프로토콜 통신용 세션 ID 관리
+- UserSessionManager: 사용자 인증 상태 및 JWT 토큰 관리
+"""
+
+import secrets
+import time
+import logging
+from typing import Dict, Any, Optional
+from dataclasses import dataclass, field
+
+logger = logging.getLogger("eodi.auth.session")
+
+
+@dataclass
+class UserSession:
+    """사용자 인증 세션 정보"""
+    user_id: str
+    email: str
+    access_token: str  # Supabase access_token
+    refresh_token: str  # Supabase refresh_token
+    expires_at: float  # Unix timestamp
+    created_at: float = field(default_factory=time.time)
+    
+    @property
+    def is_expired(self) -> bool:
+        """토큰 만료 여부 (30초 버퍼)"""
+        return time.time() > (self.expires_at - 30)
+    
+    @property
+    def email_masked(self) -> str:
+        """이메일 마스킹 (u***@example.com)"""
+        if not self.email or "@" not in self.email:
+            return self.email or "unknown"
+        
+        local, domain = self.email.split("@", 1)
+        if len(local) <= 2:
+            masked_local = local[0] + "*"
+        else:
+            masked_local = local[0] + "*" * (len(local) - 2) + local[-1]
+        return f"{masked_local}@{domain}"
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """API 응답용 딕셔너리 (민감정보 제외)"""
+        return {
+            "user_id": self.user_id,
+            "email_masked": self.email_masked,
+            "is_expired": self.is_expired,
+            "created_at": self.created_at,
+        }
+
+
+class UserSessionManager:
+    """
+    사용자 인증 세션 관리자.
+    
+    MCP Tool 호출에서 사용자를 식별하기 위한 세션 토큰을 관리합니다.
+    
+    흐름:
+    1. 사용자가 OAuth/Magic Link로 인증
+    2. Supabase 토큰을 받아 세션 생성
+    3. 세션 토큰을 LLM에게 반환
+    4. 이후 Tool 호출 시 세션 토큰을 파라미터로 전달
+    5. 세션 토큰으로 user_id 조회 → DB 접근
+    
+    주의: 메모리 기반 저장이므로 서버 재시작 시 세션 소실됩니다.
+    """
+    
+    SESSION_TTL_SECONDS = 3600  # 1시간
+    MAX_SESSIONS = 1000  # 메모리 보호
+    
+    def __init__(self):
+        self._sessions: Dict[str, UserSession] = {}
+        self._user_to_session: Dict[str, str] = {}  # user_id -> session_token (최신)
+        self._last_cleanup = time.time()
+    
+    def _cleanup_expired(self):
+        """만료된 세션 정리"""
+        now = time.time()
+        if now - self._last_cleanup < 300:  # 5분마다
+            return
+        
+        self._last_cleanup = now
+        expired = [
+            token for token, session in self._sessions.items()
+            if session.is_expired
+        ]
+        
+        for token in expired:
+            session = self._sessions.pop(token, None)
+            if session:
+                # user_to_session 매핑도 정리
+                if self._user_to_session.get(session.user_id) == token:
+                    del self._user_to_session[session.user_id]
+        
+        if expired:
+            logger.debug(f"만료된 세션 {len(expired)}개 정리됨")
+    
+    def create_session(
+        self,
+        user_id: str,
+        email: str,
+        access_token: str,
+        refresh_token: str,
+        expires_in: int = 3600
+    ) -> str:
+        """
+        새 사용자 세션 생성.
+        
+        Args:
+            user_id: Supabase Auth UUID
+            email: 사용자 이메일
+            access_token: Supabase access_token
+            refresh_token: Supabase refresh_token
+            expires_in: 토큰 만료까지 초
+            
+        Returns:
+            session_token (LLM이 후속 호출에서 사용)
+        """
+        self._cleanup_expired()
+        
+        # 메모리 보호
+        if len(self._sessions) >= self.MAX_SESSIONS:
+            # 가장 오래된 세션 삭제
+            oldest_token = min(
+                self._sessions.keys(),
+                key=lambda t: self._sessions[t].created_at
+            )
+            old_session = self._sessions.pop(oldest_token)
+            if self._user_to_session.get(old_session.user_id) == oldest_token:
+                del self._user_to_session[old_session.user_id]
+            logger.warning("세션 한도 초과, 가장 오래된 세션 삭제")
+        
+        # 기존 세션이 있으면 갱신
+        existing_token = self._user_to_session.get(user_id)
+        if existing_token and existing_token in self._sessions:
+            session = self._sessions[existing_token]
+            session.access_token = access_token
+            session.refresh_token = refresh_token
+            session.expires_at = time.time() + expires_in
+            logger.info(f"세션 갱신: {session.email_masked}")
+            return existing_token
+        
+        # 새 세션 생성
+        session_token = secrets.token_urlsafe(32)
+        session = UserSession(
+            user_id=user_id,
+            email=email,
+            access_token=access_token,
+            refresh_token=refresh_token,
+            expires_at=time.time() + expires_in
+        )
+        
+        self._sessions[session_token] = session
+        self._user_to_session[user_id] = session_token
+        
+        logger.info(f"새 세션 생성: {session.email_masked}")
+        return session_token
+    
+    def get_session(self, session_token: str) -> Optional[UserSession]:
+        """
+        세션 토큰으로 세션 조회.
+        
+        Args:
+            session_token: 세션 토큰
+            
+        Returns:
+            UserSession 또는 None (만료/미존재)
+        """
+        self._cleanup_expired()
+        
+        session = self._sessions.get(session_token)
+        if session and not session.is_expired:
+            return session
+        return None
+    
+    def get_user_id(self, session_token: str) -> Optional[str]:
+        """
+        세션 토큰에서 user_id 추출.
+        
+        Args:
+            session_token: 세션 토큰
+            
+        Returns:
+            user_id 또는 None
+        """
+        session = self.get_session(session_token)
+        return session.user_id if session else None
+    
+    def invalidate_session(self, session_token: str) -> bool:
+        """
+        세션 무효화 (로그아웃).
+        
+        Args:
+            session_token: 세션 토큰
+            
+        Returns:
+            성공 여부
+        """
+        session = self._sessions.pop(session_token, None)
+        if session:
+            if self._user_to_session.get(session.user_id) == session_token:
+                del self._user_to_session[session.user_id]
+            logger.info(f"세션 무효화: {session.email_masked}")
+            return True
+        return False
+    
+    def get_stats(self) -> Dict[str, Any]:
+        """세션 통계"""
+        self._cleanup_expired()
+        return {
+            "total_sessions": len(self._sessions),
+            "unique_users": len(self._user_to_session),
+        }
+
+
+# 전역 인스턴스 (Lazy loading)
+_user_session_manager = None
+
+
+def get_user_session_manager() -> UserSessionManager:
+    """UserSessionManager 싱글톤 인스턴스 반환"""
+    global _user_session_manager
+    if _user_session_manager is None:
+        _user_session_manager = UserSessionManager()
+    return _user_session_manager

src/auth/token_validator.py

@@ -0,0 +1,142 @@
+"""
+JWT Token Validator
+==================
+
+Supabase Auth가 발급한 JWT 토큰 검증.
+"""
+
+import logging
+from typing import Dict, Any, Optional, Tuple
+from datetime import datetime, timezone
+
+try:
+    import jwt
+except ImportError:
+    jwt = None  # PyJWT 미설치 시 graceful 처리
+
+from .config import SUPABASE_JWT_SECRET
+
+logger = logging.getLogger("eodi.auth.token_validator")
+
+
+class TokenValidator:
+    """
+    JWT 토큰 검증기.
+    
+    Supabase Auth가 발급한 access_token의 서명 및 만료 검증.
+    서버 측에서 토큰 유효성을 확인할 때 사용.
+    """
+    
+    def __init__(self, jwt_secret: str = None):
+        """
+        Args:
+            jwt_secret: Supabase JWT Secret (없으면 환경변수에서 로드)
+        """
+        self.jwt_secret = jwt_secret or SUPABASE_JWT_SECRET
+        self._enabled = bool(self.jwt_secret) and jwt is not None
+        
+        if jwt is None:
+            logger.warning("PyJWT가 설치되지 않았습니다. pip install PyJWT")
+        elif not self._enabled:
+            logger.warning("JWT Secret 미설정 - 토큰 검증이 비활성화됩니다")
+    
+    @property
+    def is_enabled(self) -> bool:
+        """토큰 검증 활성화 여부"""
+        return self._enabled
+    
+    def validate(self, token: str) -> Tuple[bool, Optional[Dict[str, Any]], Optional[str]]:
+        """
+        JWT 토큰 검증.
+        
+        Args:
+            token: JWT access_token
+            
+        Returns:
+            (유효여부, payload, error_message)
+        """
+        if not self._enabled:
+            # JWT Secret 미설정 시 검증 스킵 (개발 환경용)
+            logger.warning("JWT 검증 스킵 (Secret 미설정 또는 PyJWT 미설치)")
+            return True, {"sub": "unknown"}, None
+        
+        try:
+            payload = jwt.decode(
+                token,
+                self.jwt_secret,
+                algorithms=["HS256"],
+                options={"verify_aud": False}  # Supabase는 aud 클레임이 다양함
+            )
+            return True, payload, None
+            
+        except jwt.ExpiredSignatureError:
+            return False, None, "토큰이 만료되었습니다"
+        except jwt.InvalidTokenError as e:
+            logger.warning(f"JWT 검증 실패: {e}")
+            return False, None, f"유효하지 않은 토큰입니다: {e}"
+    
+    def extract_user_id(self, token: str) -> Optional[str]:
+        """
+        토큰에서 user_id(sub 클레임) 추출.
+        
+        Args:
+            token: JWT access_token
+            
+        Returns:
+            user_id 또는 None
+        """
+        is_valid, payload, _ = self.validate(token)
+        if is_valid and payload:
+            return payload.get("sub")
+        return None
+    
+    def is_expired(self, token: str) -> bool:
+        """
+        토큰 만료 여부 확인.
+        
+        Args:
+            token: JWT access_token
+            
+        Returns:
+            만료 여부
+        """
+        is_valid, _, error = self.validate(token)
+        return not is_valid and error and "만료" in error
+    
+    def get_expiry_time(self, token: str) -> Optional[datetime]:
+        """
+        토큰 만료 시간 추출.
+        
+        Args:
+            token: JWT access_token
+            
+        Returns:
+            만료 시간 또는 None
+        """
+        if jwt is None:
+            return None
+            
+        try:
+            # 검증 없이 페이로드만 디코딩
+            payload = jwt.decode(
+                token,
+                options={"verify_signature": False}
+            )
+            exp = payload.get("exp")
+            if exp:
+                return datetime.fromtimestamp(exp, tz=timezone.utc)
+            return None
+        except Exception:
+            return None
+
+
+# 전역 인스턴스 (Lazy loading)
+_token_validator = None
+
+
+def get_token_validator() -> TokenValidator:
+    """TokenValidator 싱글톤 인스턴스 반환"""
+    global _token_validator
+    if _token_validator is None:
+        _token_validator = TokenValidator()
+    return _token_validator

src/auth/tool_handlers.py

@@ -0,0 +1,343 @@
+"""
+User Tool Handlers
+==================
+
+user_* MCP Tool의 실제 실행 로직.
+server_streamable.py의 execute_tool_async에서 호출됩니다.
+"""
+
+import logging
+from typing import Dict, Any
+
+from .session_manager import get_user_session_manager
+from .magic_link import get_magic_link_auth
+from .config import SUPPORTED_CHAINS, TIER_ALIASES, SERVER_BASE_URL
+
+logger = logging.getLogger("eodi.auth.tool_handlers")
+
+
+async def execute_user_tool(name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    User Tool 실행 라우터.
+    
+    Args:
+        name: Tool 이름 (user_*)
+        arguments: Tool 인자
+        
+    Returns:
+        Tool 실행 결과
+    """
+    if name == "user_get_profile":
+        return await handle_get_profile(arguments)
+    elif name == "user_update_membership":
+        return await handle_update_membership(arguments)
+    elif name == "user_delete_membership":
+        return await handle_delete_membership(arguments)
+    elif name == "user_request_auth":
+        return await handle_request_auth(arguments)
+    elif name == "user_verify_code":
+        return await handle_verify_code(arguments)
+    else:
+        return {"success": False, "error": f"Unknown user tool: {name}"}
+
+
+async def handle_get_profile(arguments: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    user_get_profile Tool 핸들러.
+    
+    세션 토큰이 있으면 프로필 반환, 없으면 인증 URL 반환.
+    """
+    session_token = arguments.get("session_token")
+    user_session_manager = get_user_session_manager()
+    
+    if not session_token:
+        # 인증되지 않음 → 인증 URL 안내
+        return {
+            "success": True,
+            "connected": False,
+            "auth_url": f"{SERVER_BASE_URL}/auth/login",
+            "message": "Eodi에 연결되지 않았습니다. auth_url에서 로그인하거나, user_request_auth로 이메일 인증을 시작하세요."
+        }
+    
+    # 세션 검증
+    session = user_session_manager.get_session(session_token)
+    if not session:
+        return {
+            "success": True,
+            "connected": False,
+            "auth_url": f"{SERVER_BASE_URL}/auth/login",
+            "message": "세션이 만료되었습니다. 다시 로그인해주세요."
+        }
+    
+    # 멤버십 조회
+    try:
+        from src.db.supabase_adapter import SupabaseAdapter
+        adapter = SupabaseAdapter()
+        
+        memberships = adapter.get_user_memberships(session.user_id)
+        profile = adapter.get_user_profile(session.user_id)
+    except Exception as e:
+        logger.error(f"프로필 조회 오류: {e}")
+        memberships = []
+        profile = None
+    
+    return {
+        "success": True,
+        "connected": True,
+        "user_id": session.user_id,
+        "profile": {
+            "email_masked": session.email_masked,
+            "preferred_airports": profile.get("preferred_airports", []) if profile else [],
+            "created_at": profile.get("created_at") if profile else None,
+        },
+        "memberships": [
+            {
+                "chain": m["chain"],
+                "tier": m["tier"],
+                "expires_at": m.get("expires_at"),
+            }
+            for m in memberships
+        ],
+        "memberships_count": len(memberships)
+    }
+
+
+async def handle_update_membership(arguments: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    user_update_membership Tool 핸들러.
+    """
+    session_token = arguments.get("session_token")
+    chain = arguments.get("chain", "").upper()
+    tier = arguments.get("tier", "")
+    user_session_manager = get_user_session_manager()
+    
+    # 인증 확인
+    if not session_token:
+        return {
+            "success": False,
+            "error": "session_token이 필요합니다. 먼저 user_get_profile을 호출하여 인증 상태를 확인하세요."
+        }
+    
+    session = user_session_manager.get_session(session_token)
+    if not session:
+        return {
+            "success": False,
+            "error": "세션이 만료되었습니다. 다시 로그인해주세요.",
+            "auth_url": f"{SERVER_BASE_URL}/auth/login"
+        }
+    
+    # 체인 유효성
+    if chain not in SUPPORTED_CHAINS:
+        return {
+            "success": False,
+            "error": f"지원하지 않는 체인입니다: {chain}",
+            "supported_chains": list(SUPPORTED_CHAINS.keys())
+        }
+    
+    # 등급 정규화
+    tier_normalized = TIER_ALIASES.get(tier, tier)
+    
+    # 등급 유효성
+    valid_tiers = SUPPORTED_CHAINS[chain]
+    tier_lower_map = {t.lower(): t for t in valid_tiers}
+    
+    if tier_normalized.lower() not in tier_lower_map:
+        return {
+            "success": False,
+            "error": f"{chain}의 유효하지 않은 등급입니다: {tier}",
+            "valid_tiers": valid_tiers
+        }
+    
+    tier_final = tier_lower_map[tier_normalized.lower()]
+    
+    # 멤버십 저장
+    try:
+        from src.db.supabase_adapter import SupabaseAdapter
+        adapter = SupabaseAdapter()
+        
+        # 기존 멤버십 확인
+        existing = adapter.get_user_memberships(session.user_id)
+        is_update = any(m["chain"] == chain for m in existing)
+        
+        result = adapter.upsert_membership(session.user_id, chain, tier_final)
+        
+        if result:
+            action = "updated" if is_update else "created"
+            chain_names = {
+                "HILTON": "힐튼",
+                "MARRIOTT": "메리어트",  
+                "IHG": "IHG",
+                "ACCOR": "아코르",
+                "HYATT": "하얏트",
+            }
+            chain_name = chain_names.get(chain, chain)
+            
+            return {
+                "success": True,
+                "action": action,
+                "membership": {
+                    "chain": chain,
+                    "tier": tier_final,
+                },
+                "message": f"{chain_name} {tier_final} 등급이 {'수정' if is_update else '등록'}되었습니다."
+            }
+        else:
+            return {
+                "success": False,
+                "error": "멤버십 저장에 실패했습니다. 잠시 후 다시 시도해주세요."
+            }
+    except Exception as e:
+        logger.error(f"멤버십 저장 오류: {e}")
+        return {
+            "success": False,
+            "error": f"멤버십 저장 중 오류가 발생했습니다: {e}"
+        }
+
+
+async def handle_delete_membership(arguments: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    user_delete_membership Tool 핸들러.
+    """
+    session_token = arguments.get("session_token")
+    chain = arguments.get("chain", "").upper()
+    user_session_manager = get_user_session_manager()
+    
+    # 인증 확인
+    if not session_token:
+        return {
+            "success": False,
+            "error": "session_token이 필요합니다."
+        }
+    
+    session = user_session_manager.get_session(session_token)
+    if not session:
+        return {
+            "success": False,
+            "error": "세션이 만료되었습니다. 다시 로그인해주세요."
+        }
+    
+    # 체인 유효성
+    if chain not in SUPPORTED_CHAINS:
+        return {
+            "success": False,
+            "error": f"지원하지 않는 체인입니다: {chain}"
+        }
+    
+    # 멤버십 삭제
+    try:
+        from src.db.supabase_adapter import SupabaseAdapter
+        adapter = SupabaseAdapter()
+        
+        if adapter.delete_membership(session.user_id, chain):
+            chain_names = {
+                "HILTON": "힐튼",
+                "MARRIOTT": "메리어트",
+                "IHG": "IHG",
+                "ACCOR": "아코르",
+                "HYATT": "하얏트",
+            }
+            chain_name = chain_names.get(chain, chain)
+            
+            return {
+                "success": True,
+                "message": f"{chain_name} 멤버십이 삭제되었습니다."
+            }
+        else:
+            return {
+                "success": False,
+                "error": "멤버십 삭제에 실패했습니다."
+            }
+    except Exception as e:
+        logger.error(f"멤버십 삭제 오류: {e}")
+        return {
+            "success": False,
+            "error": f"멤버십 삭제 중 오류가 발생했습니다: {e}"
+        }
+
+
+async def handle_request_auth(arguments: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    user_request_auth Tool 핸들러.
+    
+    이메일로 Magic Link 발송.
+    """
+    email = arguments.get("email", "").strip().lower()
+    magic_link_auth = get_magic_link_auth()
+    
+    if not email:
+        return {
+            "success": False,
+            "error": "이메일 주소를 입력해주세요."
+        }
+    
+    # 간단한 이메일 형식 검증
+    if "@" not in email or "." not in email:
+        return {
+            "success": False,
+            "error": "올바른 이메일 형식이 아닙니다."
+        }
+    
+    success, error = await magic_link_auth.send_magic_link(email)
+    
+    if success:
+        # 이메일 마스킹
+        local, domain = email.split("@", 1)
+        if len(local) <= 2:
+            masked = local[0] + "*"
+        else:
+            masked = local[0] + "*" * (len(local) - 2) + local[-1]
+        email_masked = f"{masked}@{domain}"
+        
+        return {
+            "success": True,
+            "message": f"{email_masked}으로 로그인 링크를 보냈습니다.\n\n📧 이메일에서 링크를 클릭한 후,\n화면에 표시되는 6자리 코드를 user_verify_code로 입력해주세요.",
+            "next_step": "user_verify_code"
+        }
+    else:
+        return {
+            "success": False,
+            "error": error or "이메일 발송에 실패했습니다."
+        }
+
+
+async def handle_verify_code(arguments: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    user_verify_code Tool 핸들러.
+    
+    6자리 인증 코드 검증.
+    """
+    code = arguments.get("code", "").strip()
+    magic_link_auth = get_magic_link_auth()
+    
+    if not code:
+        return {
+            "success": False,
+            "error": "인증 코드를 입력해주세요."
+        }
+    
+    # 코드 길이 검증
+    code_clean = code.replace(" ", "").replace("-", "")
+    if len(code_clean) != 6 or not code_clean.isdigit():
+        return {
+            "success": False,
+            "error": "6자리 숫자 코드를 입력해주세요."
+        }
+    
+    # client_ip는 MCP Tool에서 직접 접근 불가 → 기본값 사용
+    success, session_token, email_masked, error = magic_link_auth.verify_auth_code(
+        code_clean, "mcp_client"
+    )
+    
+    if success:
+        return {
+            "success": True,
+            "connected": True,
+            "session_token": session_token,
+            "email_masked": email_masked,
+            "message": f"✅ 인증 완료! {email_masked}\n\n이제 멤버십을 등록해볼까요?\n어떤 호텔 체인 멤버십을 가지고 계세요?"
+        }
+    else:
+        return {
+            "success": False,
+            "error": error
+        }

src/db/supabase_adapter.py

@@ -303,6 +303,158 @@ class SupabaseAdapter:
             }
         except Exception as e:
             return {"error": str(e)}
+    
+    # =========================================================================
+    # 사용자 프로필 (User Gate)
+    # =========================================================================
+    
+    def get_user_profile(self, user_id: str) -> Optional[Dict[str, Any]]:
+        """
+        사용자 프로필 조회.
+        
+        Args:
+            user_id: Supabase Auth UUID
+            
+        Returns:
+            프로필 정보 또는 None
+        """
+        try:
+            result = self.client.table("user_profiles")\
+                .select("*")\
+                .eq("user_id", user_id)\
+                .execute()
+            
+            return result.data[0] if result.data else None
+        except Exception as e:
+            print(f"⚠️ 프로필 조회 실패 ({user_id}): {e}")
+            return None
+    
+    def upsert_user_profile(
+        self, 
+        user_id: str, 
+        preferred_airports: List[str] = None,
+        display_name: str = None
+    ) -> Optional[Dict[str, Any]]:
+        """
+        사용자 프로필 생성/수정.
+        
+        Args:
+            user_id: Supabase Auth UUID
+            preferred_airports: 선호 공항 코드 리스트
+            display_name: 표시 이름
+            
+        Returns:
+            저장된 프로필 또는 None
+        """
+        try:
+            data = {"user_id": user_id}
+            if preferred_airports is not None:
+                data["preferred_airports"] = preferred_airports
+            if display_name is not None:
+                data["display_name"] = display_name
+            
+            result = self.client.table("user_profiles")\
+                .upsert(data, on_conflict="user_id")\
+                .execute()
+            
+            return result.data[0] if result.data else None
+        except Exception as e:
+            print(f"⚠️ 프로필 저장 실패 ({user_id}): {e}")
+            return None
+    
+    # =========================================================================
+    # 멤버십 (User Gate)
+    # =========================================================================
+    
+    def get_user_memberships(self, user_id: str) -> List[Dict[str, Any]]:
+        """
+        사용자의 모든 멤버십 조회.
+        
+        Args:
+            user_id: Supabase Auth UUID
+            
+        Returns:
+            멤버십 리스트
+        """
+        try:
+            result = self.client.table("user_memberships")\
+                .select("*")\
+                .eq("user_id", user_id)\
+                .execute()
+            
+            return result.data or []
+        except Exception as e:
+            print(f"⚠️ 멤버십 조회 실패 ({user_id}): {e}")
+            return []
+    
+    def upsert_membership(
+        self,
+        user_id: str,
+        chain: str,
+        tier: str,
+        expires_at: str = None
+    ) -> Optional[Dict[str, Any]]:
+        """
+        멤버십 등록/수정.
+        
+        Args:
+            user_id: Supabase Auth UUID
+            chain: 호텔 체인 코드 (HILTON, MARRIOTT 등)
+            tier: 등급 (Gold, Platinum 등)
+            expires_at: 만료일 (선택, YYYY-MM-DD)
+            
+        Returns:
+            저장된 멤버십 또는 None
+        """
+        try:
+            data = {
+                "user_id": user_id,
+                "chain": chain.upper(),
+                "tier": tier
+            }
+            if expires_at:
+                data["expires_at"] = expires_at
+            
+            # user_id + chain 복합키로 upsert
+            # Supabase는 on_conflict에 복합키를 직접 지원하지 않으므로
+            # 먼저 삭제 후 삽입하는 방식 사용
+            self.client.table("user_memberships")\
+                .delete()\
+                .eq("user_id", user_id)\
+                .eq("chain", chain.upper())\
+                .execute()
+            
+            result = self.client.table("user_memberships")\
+                .insert(data)\
+                .execute()
+            
+            return result.data[0] if result.data else None
+        except Exception as e:
+            print(f"⚠️ 멤버십 저장 실패 ({user_id}, {chain}): {e}")
+            return None
+    
+    def delete_membership(self, user_id: str, chain: str) -> bool:
+        """
+        멤버십 삭제.
+        
+        Args:
+            user_id: Supabase Auth UUID
+            chain: 호텔 체인 코드
+            
+        Returns:
+            삭제 성공 여부
+        """
+        try:
+            self.client.table("user_memberships")\
+                .delete()\
+                .eq("user_id", user_id)\
+                .eq("chain", chain.upper())\
+                .execute()
+            
+            return True
+        except Exception as e:
+            print(f"⚠️ 멤버십 삭제 실패 ({user_id}, {chain}): {e}")
+            return False
 
 
 # =============================================================================

src/mcp/server_streamable.py

@@ -405,6 +405,100 @@ TOOLS = [
             },
             "required": []
         }
+    },
+    # ============================================================
+    # User Tools (User Gate - 사용자 인증/멤버십)
+    # ============================================================
+    {
+        "name": "user_get_profile",
+        "description": (
+            "현재 사용자의 프로필과 멤버십 정보를 조회합니다. "
+            "연결되지 않은 경우 auth_url을 반환합니다."
+        ),
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "session_token": {
+                    "type": "string",
+                    "description": "인증 세션 토큰 (이전 인증에서 받은 값)"
+                }
+            },
+            "required": []
+        }
+    },
+    {
+        "name": "user_update_membership",
+        "description": "호텔 체인 멤버십을 등록하거나 수정합니다.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "chain": {
+                    "type": "string",
+                    "enum": ["HILTON", "MARRIOTT", "IHG", "ACCOR", "HYATT"],
+                    "description": "호텔 체인 코드"
+                },
+                "tier": {
+                    "type": "string",
+                    "description": "멤버십 등급 (예: Gold, Platinum, Diamond)"
+                },
+                "session_token": {
+                    "type": "string",
+                    "description": "인증 세션 토큰"
+                }
+            },
+            "required": ["chain", "tier", "session_token"]
+        }
+    },
+    {
+        "name": "user_delete_membership",
+        "description": "호텔 체인 멤버십을 삭제합니다.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "chain": {
+                    "type": "string",
+                    "enum": ["HILTON", "MARRIOTT", "IHG", "ACCOR", "HYATT"],
+                    "description": "삭제할 호텔 체인 코드"
+                },
+                "session_token": {
+                    "type": "string",
+                    "description": "인증 세션 토큰"
+                }
+            },
+            "required": ["chain", "session_token"]
+        }
+    },
+    {
+        "name": "user_request_auth",
+        "description": (
+            "Magic Link 인증을 요청합니다. "
+            "이메일로 로그인 링크가 발송됩니다."
+        ),
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "email": {
+                    "type": "string",
+                    "format": "email",
+                    "description": "인증에 사용할 이메일 주소"
+                }
+            },
+            "required": ["email"]
+        }
+    },
+    {
+        "name": "user_verify_code",
+        "description": "Magic Link 인증 후 받은 6자리 코드를 검증합니다.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "code": {
+                    "type": "string",
+                    "description": "6자리 인증 코드"
+                }
+            },
+            "required": ["code"]
+        }
     }
 ]
 
@@ -696,8 +790,13 @@ async def execute_tool_async(name: str, arguments: Dict[str, Any]) -> Dict[str,
         return await convert_currency(**arguments)
     elif name == "get_travel_info":
         return await get_travel_info(**arguments)
-    else:
-        return {"success": False, "error": f"Unknown tool: {name}"}
+    
+    # User 도구 (User Gate)
+    if name.startswith("user_"):
+        from src.auth.tool_handlers import execute_user_tool
+        return await execute_user_tool(name, arguments)
+    
+    return {"success": False, "error": f"Unknown tool: {name}"}
 
 
 # ============================================================
@@ -1408,6 +1507,14 @@ routes = [
     Route("/messages/", legacy_messages_endpoint, methods=["POST", "OPTIONS"]),
 ]
 
+# Auth routes 확장 (User Gate)
+try:
+    from src.auth.routes import auth_routes
+    routes.extend(auth_routes)
+    logger.info(f"Auth routes loaded: {len(auth_routes)} endpoints")
+except ImportError as e:
+    logger.warning(f"Auth routes not loaded: {e}")
+
 middleware = [
     Middleware(
         CORSMiddleware,