DOCS/book/300_architecture/gateway_proxy_patterns.md

Gateway 프록시 패턴 아키텍처

작성일: 2025-08-21

작성자: Claude (51123 서버 관리자)

---

1. 개요

robeing-gateway (포트 8100)의 프록시 패턴 및 JWT 인증 처리 아키텍처 문서입니다. Gateway는 모든 API 요청의 진입점으로서 인증, 라우팅, UUID 변환을 담당합니다.

---

2. 핵심 역할

2.1 주요 기능

• JWT 토큰 검증: 내부적으로 토큰 유효성 검증
• Username → UUID 변환: JWT의 username을 UUID로 변환
• 요청 라우팅: 적절한 백엔드 서비스로 프록시
• 헤더 주입: X-User-Id, X-Username 헤더 추가
• 보안 게이트웨이: 인증되지 않은 요청 차단

2.3 Slack 웹훅 프록시 필수 규칙

• 원문 바디 보존: Slack Events/Interactive 프록시는 request.body() 원문 바이트를 그대로 전달한다.
• 재직렬화 금지: Slack 서명 검증 경로에서 json= 기반 재직렬화 전달을 금지한다.
• 권장 포워딩: client.post(..., content=raw_body, headers=...) 패턴을 기본으로 사용한다.
• 헤더 전달: X-Slack-Signature, X-Slack-Request-Timestamp, Content-Type를 함께 전달한다.
• 운영 검증: 배포 후 /gateway/slack/events의 200/403/500 분포와 upstream_status를 같은 시각 기준으로 확인한다.

2.2 서비스 위치

• 서버: 51123
• 포트: 8100
• 컨테이너: robeing-gateway

---

3. 인증 플로우

3.1 JWT 검증 프로세스

sequenceDiagram
    participant Client as 클라이언트
    participant Gateway as Gateway(8100)
    participant DB as PostgreSQL
    participant Service as 백엔드 서비스

    Client->>Gateway: API 요청 + JWT Token
    
    Gateway->>Gateway: JWT 서명 검증
    Note over Gateway: HS256 알고리즘
    
    alt 토큰 무효
        Gateway-->>Client: 401 Unauthorized
    else 토큰 유효
        Gateway->>Gateway: username 추출
        Note over Gateway: JWT payload에서
        
        Gateway->>DB: SELECT id FROM user WHERE username = ?
        DB-->>Gateway: UUID
        
        Gateway->>Service: 요청 전달
        Note over Service: Headers:<br/>X-User-Id: UUID<br/>X-Username: username
        
        Service-->>Gateway: 응답
        Gateway-->>Client: 응답 전달
    end

3.2 JWT 구조

{
  "header": {
    "alg": "HS256",
    "typ": "JWT"
  },
  "payload": {
    "username": "happybell80",
    "email": "goeun2dc@gmail.com",
    "exp": 1724356800,
    "iat": 1724270400
  }
}

---

4. UUID 변환 메커니즘

4.1 변환 로직

class GatewayProxy:
    async def convert_username_to_uuid(self, username: str) -> str:
        """Username을 UUID로 변환"""
        query = "SELECT id FROM user WHERE username = $1"
        result = await self.db.fetchone(query, username)
        
        if not result:
            raise UserNotFoundError(f"User {username} not found")
            
        return str(result['id'])
    
    async def process_request(self, request):
        # 1. JWT에서 username 추출
        token = request.headers.get('Authorization')
        payload = self.verify_jwt(token)
        username = payload['username']
        
        # 2. UUID 변환
        user_uuid = await self.convert_username_to_uuid(username)
        
        # 3. 헤더 추가
        request.headers['X-User-Id'] = user_uuid
        request.headers['X-Username'] = username
        
        # 4. 프록시
        return await self.proxy_to_service(request)

4.2 캐싱 전략

from functools import lru_cache
from datetime import datetime, timedelta

class UserCache:
    def __init__(self, ttl_seconds=300):
        self.cache = {}
        self.ttl = timedelta(seconds=ttl_seconds)
    
    async def get_uuid(self, username: str) -> str:
        # 캐시 확인
        if username in self.cache:
            entry = self.cache[username]
            if datetime.now() < entry['expires']:
                return entry['uuid']
        
        # DB 조회
        uuid = await self.fetch_from_db(username)
        
        # 캐시 저장
        self.cache[username] = {
            'uuid': uuid,
            'expires': datetime.now() + self.ttl
        }
        
        return uuid

---

5. 라우팅 규칙

5.1 서비스 매핑

routes:
  # 인증 서비스
  - path: /api/auth/*
    service: auth-server
    host: localhost
    port: 9000
    auth_required: false
  
  # Gmail 아이템
  - path: /api/items/gmail/*
    service: robeing-monitor
    host: localhost
    port: 9024
    auth_required: true
  
  # 로빙 서비스 (51124 서버)
  - path: /api/robeing/*
    service: rb10508_micro
    host: 192.168.219.52
    port: 10508
    auth_required: true
  
  # 스킬 서비스
  - path: /api/skills/email/*
    service: skill-email
    host: 192.168.219.52
    port: 8501
    auth_required: true

5.2 동적 라우팅

class DynamicRouter:
    def __init__(self):
        self.routes = self.load_routes()
    
    async def route_request(self, path: str, headers: dict):
        # 경로 매칭
        for route in self.routes:
            if self.match_path(path, route['path']):
                # 인증 확인
                if route['auth_required'] and not headers.get('X-User-Id'):
                    raise AuthenticationError()
                
                # 프록시
                return await self.proxy(
                    host=route['host'],
                    port=route['port'],
                    path=path,
                    headers=headers
                )
        
        raise NotFoundError(f"No route for {path}")

---

6. 에러 처리

6.1 에러 응답 포맷

{
  "error": {
    "code": "AUTHENTICATION_FAILED",
    "message": "Invalid or expired token",
    "timestamp": "2025-08-21T10:30:00Z",
    "request_id": "req_123456"
  }
}

6.2 에러 코드

코드	HTTP Status	설명
INVALID_TOKEN	401	JWT 토큰 무효
TOKEN_EXPIRED	401	JWT 토큰 만료
USER_NOT_FOUND	404	Username에 해당하는 사용자 없음
SERVICE_UNAVAILABLE	503	백엔드 서비스 응답 없음
RATE_LIMIT_EXCEEDED	429	요청 제한 초과

---

7. 성능 최적화

7.1 Connection Pooling

class ConnectionPool:
    def __init__(self):
        self.pools = {}
    
    async def get_connection(self, service: str):
        if service not in self.pools:
            self.pools[service] = await aiohttp.ClientSession(
                connector=aiohttp.TCPConnector(
                    limit=100,
                    limit_per_host=30,
                    ttl_dns_cache=300
                )
            )
        return self.pools[service]

7.2 Request/Response 압축

@app.middleware("http")
async def compression_middleware(request: Request, call_next):
    response = await call_next(request)
    
    # gzip 압축 적용
    if "gzip" in request.headers.get("Accept-Encoding", ""):
        response.headers["Content-Encoding"] = "gzip"
        response.body = gzip.compress(response.body)
    
    return response

---

8. 모니터링

8.1 메트릭 수집

metrics = {
    "request_count": Counter("gateway_requests_total"),
    "request_duration": Histogram("gateway_request_duration_seconds"),
    "auth_failures": Counter("gateway_auth_failures_total"),
    "proxy_errors": Counter("gateway_proxy_errors_total")
}

@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    start_time = time.time()
    
    try:
        response = await call_next(request)
        metrics["request_count"].inc()
        return response
    except Exception as e:
        metrics["proxy_errors"].inc()
        raise
    finally:
        duration = time.time() - start_time
        metrics["request_duration"].observe(duration)

8.2 로그 포맷

{
  "timestamp": "2025-08-21T10:30:00Z",
  "level": "INFO",
  "service": "gateway",
  "request_id": "req_123456",
  "method": "POST",
  "path": "/api/robeing/chat",
  "username": "happybell80",
  "user_id": "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
  "backend_service": "rb10508_micro",
  "response_time_ms": 150,
  "status_code": 200
}

---

9. 보안 고려사항

9.1 Rate Limiting

from datetime import datetime, timedelta

class RateLimiter:
    def __init__(self, requests_per_minute=60):
        self.limits = {}
        self.max_requests = requests_per_minute
    
    async def check_limit(self, user_id: str) -> bool:
        now = datetime.now()
        minute_ago = now - timedelta(minutes=1)
        
        # 사용자별 요청 기록
        if user_id not in self.limits:
            self.limits[user_id] = []
        
        # 1분 이내 요청만 유지
        self.limits[user_id] = [
            ts for ts in self.limits[user_id] 
            if ts > minute_ago
        ]
        
        # 제한 확인
        if len(self.limits[user_id]) >= self.max_requests:
            return False
        
        self.limits[user_id].append(now)
        return True

9.2 CORS 설정

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.ro-being.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
    max_age=3600
)

---

10. 구현 예제

10.1 완전한 Gateway 클래스

from fastapi import FastAPI, Request, HTTPException
import httpx
import jwt
import asyncpg

class RobeingGateway:
    def __init__(self):
        self.app = FastAPI()
        self.db_pool = None
        self.http_client = httpx.AsyncClient()
        self.setup_routes()
    
    async def startup(self):
        self.db_pool = await asyncpg.create_pool(
            "postgresql://robeings:robeings@localhost/main_db"
        )
    
    def setup_routes(self):
        @self.app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
        async def proxy(request: Request, path: str):
            # JWT 검증
            token = request.headers.get("Authorization", "").replace("Bearer ", "")
            if not token and self.requires_auth(path):
                raise HTTPException(401, "Missing token")
            
            # Username → UUID 변환
            if token:
                payload = self.verify_jwt(token)
                username = payload['username']
                user_uuid = await self.get_user_uuid(username)
                
                # 헤더 추가
                headers = dict(request.headers)
                headers['X-User-Id'] = user_uuid
                headers['X-Username'] = username
            else:
                headers = dict(request.headers)
            
            # 백엔드 서비스로 프록시
            backend_url = self.get_backend_url(path)
            response = await self.http_client.request(
                method=request.method,
                url=backend_url,
                headers=headers,
                content=await request.body()
            )
            
            return response.json()
    
    def verify_jwt(self, token: str) -> dict:
        try:
            return jwt.decode(
                token, 
                "your-secret-key", 
                algorithms=["HS256"]
            )
        except jwt.InvalidTokenError:
            raise HTTPException(401, "Invalid token")
    
    async def get_user_uuid(self, username: str) -> str:
        async with self.db_pool.acquire() as conn:
            row = await conn.fetchrow(
                "SELECT id FROM user WHERE username = $1",
                username
            )
            if not row:
                raise HTTPException(404, f"User {username} not found")
            return str(row['id'])

---

11. 테스트 전략

11.1 단위 테스트

import pytest
from unittest.mock import Mock, patch

class TestGateway:
    @pytest.mark.asyncio
    async def test_username_to_uuid_conversion(self):
        gateway = RobeingGateway()
        
        # Mock DB response
        with patch.object(gateway, 'db_pool') as mock_pool:
            mock_pool.acquire.return_value.__aenter__.return_value.fetchrow.return_value = {
                'id': 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
            }
            
            uuid = await gateway.get_user_uuid('happybell80')
            assert uuid == 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
    
    def test_jwt_verification(self):
        gateway = RobeingGateway()
        
        # Valid token
        token = jwt.encode(
            {'username': 'test', 'exp': datetime.now() + timedelta(hours=1)},
            'your-secret-key',
            algorithm='HS256'
        )
        
        payload = gateway.verify_jwt(token)
        assert payload['username'] == 'test'

---

문서 끝