Skip to content

KR_Sanic

Jump to bottom
somaz edited this page Sep 22, 2026 · 9 revisions

Python Sanic 개념 정리

1️⃣ Sanic 기초

Sanic은 Python 기반의 고성능 비동기 웹 프레임워크로, 빠른 HTTP 응답 속도를 제공하며 uvloop와 httptools를 활용하여 Node.js 및 Go와 비슷한 성능을 목표로 한다.

from sanic import Sanic
from sanic.response import json, text, html
from sanic.request import Request
from sanic import exceptions
import logging
import asyncio
from typing import Dict, Any, List, Optional, Union

# 애플리케이션 생성
app = Sanic("MyApplication")

# 기본 설정
app.config.FALLBACK_ERROR_FORMAT = "json"  # 오류 응답 형식
app.config.ACCESS_LOG = True  # 접근 로그 활성화
app.config.FORWARDED_SECRET = "YOUR_SECRET_KEY"  # 프록시 보안

# 로거 설정
logger = logging.getLogger('sanic')

# 기본 라우트 핸들러
@app.route("/", methods=["GET"])
async def index(request: Request) -> json:
    """
    기본 인덱스 페이지 핸들러
    
    Args:
        request: Sanic 요청 객체
        
    Returns:
        JSON 응답
    """
    return json({
        "message": "Welcome to Sanic API",
        "status": "success",
        "version": "1.0.0"
    })

# 경로 파라미터를 사용한 라우트
@app.route("/users/<user_id:int>", methods=["GET"])
async def get_user(request: Request, user_id: int) -> json:
    """
    사용자 정보 조회 핸들러
    
    Args:
        request: Sanic 요청 객체
        user_id: 사용자 ID
        
    Returns:
        사용자 정보 JSON 응답
    """
    # 데이터베이스 조회 로직 (예시)
    await asyncio.sleep(0.01)  # 비동기 작업 시뮬레이션
    
    # 사용자를 찾지 못한 경우 404 오류 발생
    if user_id > 1000:
        raise exceptions.NotFound(f"사용자 ID {user_id}를 찾을 수 없습니다")
    
    return json({
        "user_id": user_id,
        "name": f"User {user_id}",
        "email": f"user{user_id}@example.com",
        "created_at": "2023-01-01T00:00:00Z"
    })

# POST 요청 처리
@app.post("/users")
async def create_user(request: Request) -> json:
    """
    새 사용자 생성 핸들러
    
    Args:
        request: Sanic 요청 객체
        
    Returns:
        생성된 사용자 정보 JSON 응답
    """
    try:
        # 요청 본문에서 JSON 데이터 파싱
        data = request.json
        
        # 필수 필드 검증
        if not data or 'name' not in data or 'email' not in data:
            raise exceptions.BadRequest("이름과 이메일이 필요합니다")
            
        # 사용자 생성 로직 (예시)
        await asyncio.sleep(0.05)  # 비동기 작업 시뮬레이션
        
        # 생성된 사용자 정보 반환
        new_user = {
            "user_id": 1001,  # 예시 ID
            "name": data['name'],
            "email": data['email'],
            "created_at": "2023-01-01T00:00:00Z"
        }
        
        logger.info(f"새 사용자 생성: {new_user['user_id']}")
        return json({"status": "created", "user": new_user}, status=201)
        
    except Exception as e:
        logger.error(f"사용자 생성 오류: {str(e)}")
        return json({"error": str(e)}, status=500)

# 오류 핸들러
@app.exception(exceptions.NotFound)
async def not_found(request: Request, exception: exceptions.NotFound) -> json:
    """
    404 오류 핸들러
    
    Args:
        request: Sanic 요청 객체
        exception: 발생한 예외
        
    Returns:
        오류 메시지 JSON 응답
    """
    return json({
        "error": "not_found",
        "message": str(exception)
    }, status=404)

# 다양한 응답 타입
@app.route("/responses")
async def response_types(request: Request):
    """다양한 응답 타입 예제"""
    response_type = request.args.get('type', 'json')
    
    if response_type == 'text':
        return text("Plain text response")
    elif response_type == 'html':
        return html("<h1>HTML Response</h1>")
    else:
        return json({"type": "json", "message": "JSON Response"})

# 애플리케이션 실행
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, debug=True, auto_reload=True)

✅ 특징:

  • 비동기 처리 기반의 고성능 HTTP 서비스
  • 타입 힌팅으로 코드 가독성 및 IDE 지원 향상
  • 다양한 경로 파라미터 타입 지원(int, string, uuid, float, path)
  • 다양한 응답 형식(JSON, HTML, 텍스트, 파일 등)
  • 내장된 예외 처리 메커니즘
  • 자동 리로드 기능으로 개발 생산성 향상
  • HTTP/2 및 웹소켓 지원
  • 미들웨어와 리스너 기반의 확장성
  • uvloop 활용으로 표준 asyncio보다 2배 이상 빠른 성능
  • 구조화된 로깅 시스템

2️⃣ 미들웨어와 리스너

Sanic은 요청 처리 전후에 코드를 실행할 수 있는 미들웨어와 서버 생명주기 이벤트를 처리하는 리스너를 제공하여 애플리케이션을 유연하게 확장할 수 있게 한다.

from sanic import Sanic, Request, response
from sanic.response import json
import uuid
import time
import logging
import asyncio
import aioredis
from datetime import datetime
from typing import Dict, Any, Optional, Callable

app = Sanic("MiddlewareApp")
logger = logging.getLogger('sanic.middleware')

# 요청 미들웨어 정의
@app.middleware('request')
async def add_request_id(request: Request):
    """
    각 요청에 고유 ID 추가
    
    Args:
        request: Sanic 요청 객체
    """
    request_id = str(uuid.uuid4())
    request.ctx.request_id = request_id
    logger.info(f"Request started: {request_id}")
    
    # 요청 시작 시간 기록
    request.ctx.start_time = time.time()

# 다른 요청 미들웨어 - 인증 검사
@app.middleware('request')
async def check_auth(request: Request):
    """
    인증 헤더 검사 미들웨어
    
    Args:
        request: Sanic 요청 객체
    """
    # 인증이 필요 없는 경로는 건너뜀
    if request.path in ['/login', '/register', '/health']:
        return
    
    auth_header = request.headers.get('Authorization')
    if not auth_header or not auth_header.startswith('Bearer '):
        # 이 라인은 미들웨어에서 응답을 반환하므로 요청 처리가 중단됨
        return json({'error': 'unauthorized', 'message': '인증이 필요합니다'}, status=401)
    
    # 실제 구현에서는 토큰 검증 로직 추가
    token = auth_header.split(' ')[1]
    # 예: validate_token(token)
    
    # 인증된 사용자 정보를 요청 컨텍스트에 저장
    request.ctx.user = {'id': 1, 'role': 'user'}  # 예시 데이터

# 응답 미들웨어 정의
@app.middleware('response')
async def add_response_headers(request: Request, response):
    """
    응답 헤더 추가 미들웨어
    
    Args:
        request: Sanic 요청 객체
        response: Sanic 응답 객체
    """
    # 요청 ID 헤더 추가
    if hasattr(request.ctx, 'request_id'):
        response.headers['X-Request-ID'] = request.ctx.request_id
    
    # 응답 시간 기록 및 헤더 추가
    if hasattr(request.ctx, 'start_time'):
        duration = time.time() - request.ctx.start_time
        response.headers['X-Response-Time'] = f"{duration:.6f}s"
        logger.info(f"Request {request.ctx.request_id} completed in {duration:.6f}s")
    
    # 보안 헤더 추가
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-Frame-Options'] = 'DENY'
    response.headers['X-XSS-Protection'] = '1; mode=block'

# 커스텀 미들웨어 함수 생성
def rate_limit_middleware(limit: int = 10, window: int = 60):
    """
    속도 제한 미들웨어 팩토리
    
    Args:
        limit: 시간 창 내 최대 요청 수
        window: 시간 창 (초)
        
    Returns:
        미들웨어 함수
    """
    request_counts = {}
    
    @app.middleware('request')
    async def rate_limit(request: Request):
        # 클라이언트 식별
        client_ip = request.client_ip
        current_time = int(time.time())
        
        # 현재 클라이언트의 요청 기록 가져오기
        if client_ip not in request_counts:
            request_counts[client_ip] = []
        
        # 시간 창 외의 요청 제거
        request_counts[client_ip] = [timestamp for timestamp in request_counts[client_ip] 
                               if timestamp > current_time - window]
        
        # 요청 수 확인
        if len(request_counts[client_ip]) >= limit:
            return json({
                'error': 'rate_limit_exceeded',
                'message': f'요청 한도 초과, {window}초 후에 다시 시도하세요'
            }, status=429)
        
        # 현재 요청 시간 기록
        request_counts[client_ip].append(current_time)
    
    return rate_limit

# 서버 시작 전 리스너
@app.listener('before_server_start')
async def setup_db(app: Sanic, loop):
    """
    서버 시작 전 데이터베이스 연결 설정
    
    Args:
        app: Sanic 앱 인스턴스
        loop: 이벤트 루프
    """
    logger.info("데이터베이스 연결 설정 중...")
    # 예시: PostgreSQL 연결 설정
    # import asyncpg
    # app.ctx.db_pool = await asyncpg.create_pool(
    #     user='user', password='password',
    #     database='database', host='localhost'
    # )
    
    # 예시: Redis 연결 설정
    try:
        app.ctx.redis = await aioredis.from_url(
            "redis://localhost", encoding="utf-8", decode_responses=True
        )
        logger.info("Redis 연결 성공")
    except Exception as e:
        logger.error(f"Redis 연결 실패: {e}")
    
    logger.info("서버 시작 준비 완료")

# 서버 시작 후 리스너
@app.listener('after_server_start')
async def notify_server_started(app: Sanic, loop):
    """
    서버 시작 알림
    
    Args:
        app: Sanic 앱 인스턴스
        loop: 이벤트 루프
    """
    logger.info(f"서버가 시작됨! http://{app.config.HOST}:{app.config.PORT}")
    
    # 필요한 경우 서버 시작 알림 또는 초기 작업 수행
    # 예: 캐시 워밍업, 초기 데이터 로드 등
    async def warmup_cache():
        logger.info("캐시 워밍업 중...")
        await asyncio.sleep(1)
        # 캐시 워밍업 로직
        logger.info("캐시 워밍업 완료")
    
    # 백그라운드 작업으로 실행
    asyncio.create_task(warmup_cache())

# 서버 중지 전 리스너
@app.listener('before_server_stop')
async def notify_server_stopping(app: Sanic, loop):
    """
    서버 중지 알림
    
    Args:
        app: Sanic 앱 인스턴스
        loop: 이벤트 루프
    """
    logger.info("서버 중지 중...")
    # 진행 중인 작업 완료 또는 정리

# 서버 중지 후 리스너
@app.listener('after_server_stop')
async def close_db(app: Sanic, loop):
    """
    서버 중지 후 리소스 정리
    
    Args:
        app: Sanic 앱 인스턴스
        loop: 이벤트 루프
    """
    logger.info("리소스 정리 중...")
    
    # Redis 연결 종료
    if hasattr(app.ctx, 'redis'):
        await app.ctx.redis.close()
        logger.info("Redis 연결 종료됨")
    
    # 데이터베이스 풀 종료
    if hasattr(app.ctx, 'db_pool'):
        await app.ctx.db_pool.close()
        logger.info("데이터베이스 연결 풀 종료됨")
    
    logger.info("모든 리소스가 정리됨")

# 리스너를 함수로 등록하는 방법
async def setup_something(app: Sanic, loop):
    logger.info("추가 설정 실행 중...")
    app.ctx.start_time = datetime.now()

# 나중에 리스너 등록
app.register_listener(setup_something, 'before_server_start')

# 헬스 체크 경로
@app.route('/health')
async def health_check(request: Request):
    """서버 상태 확인 엔드포인트"""
    uptime = None
    if hasattr(app.ctx, 'start_time'):
        uptime = (datetime.now() - app.ctx.start_time).total_seconds()
    
    return json({
        'status': 'healthy',
        'version': '1.0.0',
        'uptime_seconds': uptime
    })

✅ 특징:

  • 요청/응답 처리를 위한 강력한 미들웨어 체인
  • 서버 생명주기 관리를 위한 리스너 시스템
  • 미들웨어를 활용한 공통 기능 구현 (인증, 로깅, 속도 제한 등)
  • 요청/응답 컨텍스트로 데이터 공유
  • 비동기 데이터베이스 연결 관리
  • 백그라운드 작업 실행 기능
  • 애플리케이션 상태 및 리소스 관리
  • 요청별 고유 ID 및 추적 기능
  • 보안 관련 헤더 자동 추가
  • 요청 처리 성능 측정 및 모니터링
  • 우아한 종료 및 리소스 정리 지원

3️⃣ 블루프린트와 라우팅

블루프린트는 Sanic 애플리케이션을 모듈화하고 구조화하는 강력한 방법으로, 관련 기능을 논리적으로 그룹화하고 코드 재사용성을 높일 수 있다.

from sanic import Sanic, Blueprint, Request, HTTPResponse
from sanic.response import json, text, html, file, redirect
from sanic import exceptions
import logging
from typing import Dict, Any, List, Optional, Tuple

# 메인 애플리케이션 인스턴스 생성
app = Sanic("BlueprintApp")
logger = logging.getLogger('sanic.blueprints')

# 기본 API 블루프린트 생성
api_v1 = Blueprint('api_v1', url_prefix='/api/v1')
api_v2 = Blueprint('api_v2', url_prefix='/api/v2')

# 관리자 영역 블루프린트 생성
admin = Blueprint('admin', url_prefix='/admin')

# 사용자 블루프린트 (v1)
users_v1 = Blueprint('users_v1', url_prefix='/users')
api_v1.blueprint(users_v1)  # api_v1에 사용자 블루프린트 중첩

# 사용자 블루프린트 (v2)
users_v2 = Blueprint('users_v2', url_prefix='/users')
api_v2.blueprint(users_v2)  # api_v2에 사용자 블루프린트 중첩

# 인증 블루프린트
auth = Blueprint('auth', url_prefix='/auth')

#################################################
# API v1 라우트 정의
#################################################

@api_v1.route('/')
async def api_index(request: Request) -> HTTPResponse:
    """API v1 인덱스 페이지"""
    return json({
        'version': 'v1',
        'endpoints': ['/users', '/items', '/auth']
    })

@users_v1.route('/')
async def list_users_v1(request: Request) -> HTTPResponse:
    """사용자 목록 조회 (v1)"""
    users = [
        {'id': 1, 'name': 'User 1'},
        {'id': 2, 'name': 'User 2'}
    ]
    return json({'users': users})

@users_v1.route('/<user_id:int>')
async def get_user_v1(request: Request, user_id: int) -> HTTPResponse:
    """단일 사용자 조회 (v1)"""
    return json({'id': user_id, 'name': f'User {user_id}'})

@users_v1.post('/')
async def create_user_v1(request: Request) -> HTTPResponse:
    """사용자 생성 (v1)"""
    user_data = request.json
    # 사용자 생성 로직
    return json({'status': 'created', 'user': user_data}, status=201)

# 아이템 라우트 - 모든 HTTP 메서드 처리
@api_v1.route('/items/<item_id:int>', methods=['GET', 'PUT', 'DELETE'])
async def handle_item_v1(request: Request, item_id: int) -> HTTPResponse:
    """아이템 처리 (v1) - 여러 HTTP 메서드 처리"""
    if request.method == 'GET':
        return json({'item_id': item_id, 'name': f'Item {item_id}'})
    elif request.method == 'PUT':
        item_data = request.json
        # 아이템 업데이트 로직
        return json({'status': 'updated', 'item': item_data})
    elif request.method == 'DELETE':
        # 아이템 삭제 로직
        return json({'status': 'deleted'})

#################################################
# API v2 라우트 정의 (새 기능 추가)
#################################################

@api_v2.route('/')
async def api_index_v2(request: Request) -> HTTPResponse:
    """API v2 인덱스 페이지"""
    return json({
        'version': 'v2',
        'endpoints': ['/users', '/items', '/auth', '/stats']
    })

@users_v2.route('/')
async def list_users_v2(request: Request) -> HTTPResponse:
    """사용자 목록 조회 (v2) - 향상된 기능"""
    # 페이지네이션 지원
    page = int(request.args.get('page', 1))
    per_page = int(request.args.get('per_page', 10))
    
    # 샘플 데이터
    users = [
        {'id': i, 'name': f'User {i}', 'email': f'user{i}@example.com'}
        for i in range((page-1)*per_page + 1, page*per_page + 1)
    ]
    
    return json({
        'users': users,
        'pagination': {
            'page': page,
            'per_page': per_page,
            'total': 100,  # 샘플 총 사용자 수
            'pages': 10
        }
    })

@users_v2.route('/<user_id:int>')
async def get_user_v2(request: Request, user_id: int) -> HTTPResponse:
    """단일 사용자 조회 (v2) - 향상된 정보"""
    # 확장된 사용자 정보
    return json({
        'id': user_id,
        'name': f'User {user_id}',
        'email': f'user{user_id}@example.com',
        'profile': {
            'bio': 'User bio text',
            'avatar_url': f'https://example.com/avatars/{user_id}.jpg'
        },
        'created_at': '2023-01-01T00:00:00Z'
    })

@api_v2.route('/stats')
async def get_stats(request: Request) -> HTTPResponse:
    """통계 API (v2에만 있는 새 기능)"""
    return json({
        'users_count': 100,
        'items_count': 500,
        'active_users': 42
    })

#################################################
# 인증 블루프린트
#################################################

@auth.route('/login', methods=['POST'])
async def login(request: Request) -> HTTPResponse:
    """사용자 로그인 처리"""
    data = request.json
    username = data.get('username')
    password = data.get('password')
    
    # 로그인 로직 (예시)
    if username == 'admin' and password == 'password':
        return json({
            'token': 'sample-jwt-token',
            'user_id': 1,
            'expires_in': 3600
        })
    else:
        return json({'error': 'invalid_credentials'}, status=401)

@auth.route('/logout', methods=['POST'])
async def logout(request: Request) -> HTTPResponse:
    """사용자 로그아웃 처리"""
    # 로그아웃 로직
    return json({'status': 'logged_out'})

#################################################
# 관리자 블루프린트
#################################################

@admin.route('/')
async def admin_index(request: Request) -> HTTPResponse:
    """관리자 대시보드 인덱스"""
    return html('<h1>Admin Dashboard</h1>')

@admin.route('/users')
async def admin_users(request: Request) -> HTTPResponse:
    """관리자용 사용자 관리 페이지"""
    return html('<h1>User Management</h1>')

#################################################
# 정적 파일 및 특수 라우트
#################################################

# 정적 파일 제공 (app 레벨)
app.static('/static', './static')
app.static('/favicon.ico', './static/favicon.ico')

# 리디렉션 라우트
@app.route('/old-path')
async def old_path_redirect(request: Request) -> HTTPResponse:
    """이전 경로에서 새 경로로 리디렉션"""
    return redirect('/api/v2')

# 파일 다운로드 라우트
@app.route('/download/<filename:path>')
async def download_file(request: Request, filename: str) -> HTTPResponse:
    """파일 다운로드 처리"""
    return await file(f'./downloads/{filename}')

# 에러 처리 블루프린트 수준
@api_v1.exception(exceptions.NotFound)
async def api_v1_not_found(request: Request, exception: exceptions.NotFound) -> HTTPResponse:
    """API v1 404 오류 핸들러"""
    return json({
        'error': 'not_found',
        'message': str(exception),
        'api_version': 'v1'
    }, status=404)

# URL 파라미터 타입 예제
@app.route('/params/<name:str>/<value:float>/<uuid:uuid>')
async def params_example(request: Request, name: str, value: float, uuid: str) -> HTTPResponse:
    """다양한 URL 파라미터 타입 예제"""
    return json({
        'name': name,
        'value': value,
        'uuid': uuid
    })

# 블루프린트 미들웨어 예제
@api_v1.middleware('request')
async def api_v1_middleware(request: Request):
    """API v1에만 적용되는 미들웨어"""
    request.ctx.api_version = 'v1'
    logger.info(f"API v1 요청: {request.path}")

# 블루프린트를 애플리케이션에 등록
app.blueprint(api_v1)
app.blueprint(api_v2)
app.blueprint(admin)
app.blueprint(auth)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, debug=True)

✅ 특징:

  • 기능별 모듈화로 코드 구조화
  • URL 접두사로 API 경로 조직화
  • 중첩 블루프린트로 복잡한 라우팅 계층 구성
  • HTTP 메서드별 라우트 처리 (GET, POST, PUT, DELETE 등)
  • API 버전 관리를 위한 효과적인 구조
  • 블루프린트별 미들웨어 정의 가능
  • 블루프린트별 예외 처리 기능
  • 다양한 응답 타입 지원 (JSON, HTML, 파일, 리디렉션)
  • 정적 파일 서빙 기능
  • URL 파라미터의 타입 검증 및 변환
  • 요청 컨텍스트로 데이터 저장 및 공유
  • RESTful API 설계 패턴 지원

4️⃣ 인증과 보안

Sanic에서는 미들웨어, 데코레이터, 그리고 JWT와 같은 도구를 활용하여 견고한 인증 시스템과 보안 메커니즘을 구현할 수 있다.

from sanic import Sanic, Blueprint, Request, HTTPResponse
from sanic.response import json
from sanic.exceptions import Unauthorized, Forbidden
from functools import wraps
import jwt
import time
import os
import logging
import uuid
import hashlib
import secrets
from typing import Dict, Any, Callable, Optional

# 애플리케이션 설정
app = Sanic("SecureApp")
app.config.SECRET_KEY = os.environ.get('SECRET_KEY', 'your-secret-key-here')  # 실제론 환경 변수 사용

# 로거 설정
logger = logging.getLogger('sanic.security')

# 사용자 임시 데이터 (실제로는 데이터베이스에서 관리)
USERS = {
    1: {"id": 1, "username": "admin", "password_hash": hashlib.sha256("admin_password".encode()).hexdigest(), "role": "admin"},
    2: {"id": 2, "username": "user", "password_hash": hashlib.sha256("user_password".encode()).hexdigest(), "role": "user"}
}

# 차단된 토큰 추적 (Redis 등으로 대체 가능)
BLACKLISTED_TOKENS = set()

####################################################
# 토큰 생성 및 관리 함수
####################################################

def generate_token(user_id: int, role: str = "user", expiry: int = 3600) -> str:
    """
    JWT 토큰 생성
    
    Args:
        user_id: 사용자 ID
        role: 사용자 역할
        expiry: 만료 시간(초)
        
    Returns:
        str: 생성된 JWT 토큰
    """
    payload = {
        'sub': user_id,
        'role': role,
        'iat': int(time.time()),
        'exp': int(time.time()) + expiry,
        'jti': str(uuid.uuid4())
    }
    
    token = jwt.encode(
        payload,
        app.config.SECRET_KEY,
        algorithm='HS256'
    )
    
    logger.info(f"사용자 ID {user_id}의 토큰 생성됨")
    return token

def verify_token(token: str) -> Dict[str, Any]:
    """
    JWT 토큰 검증
    
    Args:
        token: JWT 토큰
        
    Returns:
        Dict: 토큰 페이로드
        
    Raises:
        jwt.InvalidTokenError: 유효하지 않은 토큰
        jwt.ExpiredSignatureError: 만료된 토큰
    """
    # 차단 목록 확인
    if token in BLACKLISTED_TOKENS:
        raise jwt.InvalidTokenError("Blacklisted token")
    
    # 토큰 검증
    payload = jwt.decode(
        token,
        app.config.SECRET_KEY,
        algorithms=['HS256']
    )
    
    return payload

####################################################
# 인증 데코레이터
####################################################

def protected(roles: Optional[list] = None) -> Callable:
    """
    보호된 경로를 위한 데코레이터
    
    Args:
        roles: 허용된 역할 목록 (None이면 모든 인증된 사용자 허용)
        
    Returns:
        Callable: 데코레이터 함수
    """
    def decorator(f):
        @wraps(f)
        async def decorated_function(request: Request, *args, **kwargs):
            # 인증 헤더 확인
            auth_header = request.headers.get('Authorization')
            if not auth_header:
                logger.warning(f"인증 헤더 누락: {request.path}")
                raise Unauthorized("인증 토큰이 필요합니다")
            
            # 헤더 형식 확인
            parts = auth_header.split()
            if parts[0].lower() != 'bearer' or len(parts) != 2:
                logger.warning(f"잘못된 인증 헤더 형식: {auth_header}")
                raise Unauthorized("잘못된 인증 헤더 형식")
            
            token = parts[1]
            
            try:
                # 토큰 검증
                payload = verify_token(token)
                
                # 사용자 정보 요청 컨텍스트에 저장
                request.ctx.user_id = payload['sub']
                request.ctx.role = payload['role']
                
                # 역할 기반 액세스 제어 확인
                if roles and payload['role'] not in roles:
                    logger.warning(f"권한 부족 - 사용자 {request.ctx.user_id}, 역할: {request.ctx.role}, 필요: {roles}")
                    raise Forbidden("이 작업을 수행할 권한이 없습니다")
                
                # 원래 핸들러 호출
                return await f(request, *args, **kwargs)
                
            except jwt.ExpiredSignatureError:
                logger.warning(f"만료된 토큰: {token[:10]}...")
                raise Unauthorized("토큰이 만료되었습니다")
                
            except jwt.InvalidTokenError as e:
                logger.warning(f"유효하지 않은 토큰: {token[:10]}..., 오류: {str(e)}")
                raise Unauthorized("유효하지 않은 토큰입니다")
                
            except Exception as e:
                logger.error(f"인증 중 오류: {str(e)}")
                raise Unauthorized("인증 처리 중 오류가 발생했습니다")
        
        return decorated_function
    return decorator

####################################################
# 보안 헤더 미들웨어
####################################################

@app.middleware('response')
async def add_security_headers(request: Request, response):
    """보안 관련 헤더 추가"""
    # XSS 방지
    response.headers['X-XSS-Protection'] = '1; mode=block'
    # 클릭재킹 방지
    response.headers['X-Frame-Options'] = 'DENY'
    # MIME 스니핑 방지
    response.headers['X-Content-Type-Options'] = 'nosniff'
    # 콘텐츠 보안 정책 설정
    response.headers['Content-Security-Policy'] = "default-src 'self'"
    # HSTS 설정 (HTTPS 사용 시)
    response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'
    # 캐시 제어 (인증이 필요한 페이지)
    if hasattr(request.ctx, 'user_id'):
        response.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate, max-age=0'
        response.headers['Pragma'] = 'no-cache'

####################################################
# CSRF 보호 미들웨어
####################################################

CSRF_TOKENS = {}  # 실제 구현에서는 Redis 등 사용

def generate_csrf_token() -> str:
    """CSRF 토큰 생성"""
    return secrets.token_hex(16)

@app.middleware('request')
async def csrf_protection(request: Request):
    """CSRF 보호 미들웨어"""
    # CSRF 보호가 필요 없는 메서드 또는 경로 제외
    safe_methods = ['GET', 'HEAD', 'OPTIONS']
    bypass_paths = ['/api/token', '/login']
    
    if request.method in safe_methods or request.path in bypass_paths:
        return
    
    # CSRF 토큰 확인
    request_csrf = request.headers.get('X-CSRF-Token')
    session_id = request.cookies.get('session_id')
    
    if not session_id or not request_csrf or CSRF_TOKENS.get(session_id) != request_csrf:
        logger.warning(f"CSRF 토큰 불일치: {request.path}")
        raise Forbidden("유효하지 않은 CSRF 토큰")

####################################################
# 사용자 인증 라우트
####################################################

@app.route('/api/token', methods=['POST'])
async def get_token(request: Request) -> HTTPResponse:
    """
    사용자 인증 및 토큰 발급
    
    Request:
        JSON: {username, password}
        
    Response:
        JSON: {token, expires_in}
    """
    data = request.json
    username = data.get('username')
    password = data.get('password')
    
    if not username or not password:
        return json({'error': 'missing_fields', 'message': '사용자명과 비밀번호가 필요합니다'}, status=400)
    
    # 비밀번호 해싱
    hashed_password = hashlib.sha256(password.encode()).hexdigest()
    
    # 사용자 확인 (실제로는 데이터베이스 조회)
    user = None
    for u in USERS.values():
        if u['username'] == username and u['password_hash'] == hashed_password:
            user = u
            break
    
    if user:
        # 토큰 생성
        token = generate_token(user['id'], user['role'])
        
        # 세션 설정 및 CSRF 토큰 생성
        session_id = str(uuid.uuid4())
        csrf_token = generate_csrf_token()
        CSRF_TOKENS[session_id] = csrf_token
        
        response = json({
            'token': token,
            'csrf_token': csrf_token,
            'expires_in': 3600,
            'user_id': user['id'],
            'role': user['role']
        })
        
        # 세션 쿠키 설정
        response.cookies['session_id'] = session_id
        response.cookies['session_id']['httponly'] = True
        response.cookies['session_id']['secure'] = True  # 프로덕션 환경 HTTPS
        response.cookies['session_id']['samesite'] = 'Lax'
        
        return response
    else:
        logger.warning(f"로그인 실패: {username}")
        return json({'error': 'invalid_credentials', 'message': '잘못된 사용자명 또는 비밀번호'}, status=401)

@app.route('/api/logout', methods=['POST'])
@protected()
async def logout(request: Request) -> HTTPResponse:
    """
    로그아웃 처리
    - 현재 토큰을 무효화
    """
    # 현재 토큰 추출
    token = request.headers.get('Authorization').split()[1]
    
    # 토큰 차단 목록에 추가
    BLACKLISTED_TOKENS.add(token)
    
    # 세션 ID가 있는 경우 CSRF 토큰 제거
    session_id = request.cookies.get('session_id')
    if session_id and session_id in CSRF_TOKENS:
        del CSRF_TOKENS[session_id]
    
    # 쿠키 클리어 응답
    response = json({'status': 'logged_out'})
    response.cookies['session_id'] = ''
    response.cookies['session_id']['expires'] = 0
    
    return response

####################################################
# 보호된 라우트 예제
####################################################

@app.route('/api/profile')
@protected()
async def get_profile(request: Request) -> HTTPResponse:
    """
    사용자 프로필 정보 조회 (모든 인증된 사용자)
    """
    user_id = request.ctx.user_id
    user = USERS.get(user_id)
    
    if not user:
        return json({'error': 'user_not_found'}, status=404)
    
    # 민감한 정보 제외
    return json({
        'id': user['id'],
        'username': user['username'],
        'role': user['role']
    })

@app.route('/api/admin/users')
@protected(['admin'])
async def list_users(request: Request) -> HTTPResponse:
    """
    모든 사용자 목록 조회 (관리자만)
    """
    # 사용자 목록 반환 (비밀번호 해시 제외)
    users_list = [
        {'id': u['id'], 'username': u['username'], 'role': u['role']}
        for u in USERS.values()
    ]
    return json({'users': users_list})

# 기본 오류 핸들러
@app.exception(Unauthorized)
async def unauthorized(request: Request, exception: Unauthorized) -> HTTPResponse:
    return json({
        'error': 'unauthorized',
        'message': str(exception)
    }, status=401)

@app.exception(Forbidden)
async def forbidden(request: Request, exception: Forbidden) -> HTTPResponse:
    return json({
        'error': 'forbidden',
        'message': str(exception)
    }, status=403)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, debug=True)

✅ 특징:

  • JWT 기반 인증 시스템 구현
  • 역할 기반 접근 제어(RBAC) 구현
  • 데코레이터 패턴으로 경로 보호
  • CSRF 토큰 기반 방어 메커니즘
  • 보안 강화 HTTP 헤더 추가
  • 토큰 차단 목록 지원
  • 세션 관리 및 안전한 쿠키 사용
  • 비밀번호 해시 처리
  • 사용자 인증 및 로그아웃 워크플로
  • 상세한 보안 로깅
  • 오류 처리 및 표준화된 응답
  • 비동기 인증 처리
  • 환경 설정 분리 (비밀키 관리)

주요 팁

✅ 모범 사례:

  • 비동기 코드 최적화
  • 에러 처리 구현
  • 미들웨어 활용
  • 캐싱 전략
  • 데이터베이스 최적화
  • 보안 설정
  • 로깅 구현
  • 테스트 작성

Clone this wiki locally