79 lines
2.7 KiB
Python
79 lines
2.7 KiB
Python
import logging
|
|
import os
|
|
|
|
from dotenv import load_dotenv
|
|
from fastapi import HTTPException, Security
|
|
from fastapi.security import APIKeyHeader
|
|
from services.api_key_service import validate_api_key
|
|
from snowflake import SnowflakeGenerator
|
|
|
|
logger = logging.getLogger(__name__)
|
|
load_dotenv()
|
|
|
|
# .env 파일에서 관리자 API 키를 로드
|
|
ADMIN_API_KEY = os.getenv("ADMIN_API_KEY")
|
|
|
|
# 헤더 설정
|
|
api_key_header = APIKeyHeader(
|
|
name="X-API-KEY", auto_error=False, description="Client-specific API Key"
|
|
)
|
|
admin_api_key_header = APIKeyHeader(
|
|
name="X-Admin-KEY", auto_error=False, description="Key for administrative tasks"
|
|
)
|
|
|
|
|
|
def get_api_key(api_key: str = Security(api_key_header)):
|
|
"""요청 헤더의 X-API-KEY가 유효한지 Redis를 통해 검증합니다."""
|
|
if not validate_api_key(api_key):
|
|
logger.warning(f"유효하지 않은 API 키로 접근 시도: {api_key}")
|
|
raise HTTPException(status_code=401, detail="Invalid or missing API Key")
|
|
return api_key
|
|
|
|
|
|
def get_admin_key(admin_key: str = Security(admin_api_key_header)):
|
|
"""관리자용 API 키를 검증합니다."""
|
|
if not ADMIN_API_KEY:
|
|
logger.error(
|
|
"ADMIN_API_KEY가 서버에 설정되지 않았습니다. 관리자 API를 사용할 수 없습니다."
|
|
)
|
|
raise HTTPException(status_code=500, detail="Server configuration error")
|
|
|
|
if not admin_key or admin_key != ADMIN_API_KEY:
|
|
logger.warning("유효하지 않은 관리자 키로 관리 API 접근 시도.")
|
|
raise HTTPException(status_code=403, detail="Not authorized for this operation")
|
|
return admin_key
|
|
|
|
|
|
class APIKeyLoader:
|
|
@staticmethod
|
|
def load_gemini_key() -> str:
|
|
key = os.getenv("GEMINI_API_KEY")
|
|
if not key:
|
|
logger.error("GEMINI_API_KEY 환경 변수가 설정되지 않았습니다.")
|
|
raise ValueError("GEMINI_API_KEY 환경 변수가 설정되지 않았습니다.")
|
|
return key
|
|
|
|
@staticmethod
|
|
def load_claude_key() -> str:
|
|
key = os.getenv("ANTHROPIC_API_KEY")
|
|
if not key:
|
|
logger.error("ANTHROPIC_API_KEY 환경 변수가 설정되지 않았습니다.")
|
|
raise ValueError("ANTHROPIC_API_KEY 환경 변수가 설정되지 않았습니다.")
|
|
return key
|
|
|
|
@staticmethod
|
|
def load_gpt_key() -> str:
|
|
key = os.getenv("OPENAI_API_KEY")
|
|
if not key:
|
|
logger.error("OPENAI_API_KEY 환경 변수가 설정되지 않았습니다.")
|
|
raise ValueError("OPENAI_API_KEY 환경 변수가 설정되지 않았습니다.")
|
|
return key
|
|
|
|
|
|
def create_key(node: int = 1) -> str:
|
|
"""
|
|
Snowflake 알고리즘 기반 고유 키 생성기 (request_id용)
|
|
"""
|
|
generator = SnowflakeGenerator(node)
|
|
return str(next(generator))
|