개요
AI API를 활용한 서비스가 급성장하면서 보안 취약점도 함께 증가하고 있습니다. 특히 AI API 키 관리와 네트워크 접근 통제는 프로덕션 환경에서 가장 중요한 보안 과제입니다.
저는 HolySheep AI에서 2년간 API 게이트웨이 보안을 담당하며, 수백 개의 프로덕션 시스템에서 발생하는 보안 사고를 분석했습니다. 그 결과, **제로 트러스트(Zero Trust)** 아키텍처가 AI API 보안의 핵심임을 확신하게 되었습니다.
제로 트러스트의 핵심 원칙은 간단합니다: **"절대 신뢰하지 말고, 항상 검증하라(Never Trust, Always Verify)"**. 네트워크 내부에 있더라도 모든 요청을 인증하고, 최소 권한 원칙을 적용하며, 모든 접근을 기록하고 감사해야 합니다.
제로 트러스트 AI API 아키텍처
핵심 컴포넌트 구조
┌─────────────────────────────────────────────────────────────────┐
│ Zero Trust AI API Gateway │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ Client │───▶│ Auth Layer │───▶│ Rate Limiter │ │
│ │ │ │ - JWT验证 │ │ - Token Bucket │ │
│ │ │ │ - API Key │ │ - 요청 수 제한 │ │
│ │ │ │ - mTLS │ │ - 동시 연결 수 │ │
│ └──────────┘ └──────────────┘ └──────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────────┐ │
│ │ Policy Engine │ │
│ │ - RBAC (역할 기반 접근 제어) │ │
│ │ - 속성 기반 정책 (ABAC) │ │
│ │ - 모델별 권한 │ │
│ └──────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ Audit Logger │ │
│ │ - 요청/응답 로깅 - 비용 추적 - 침입 탐지 │ │
│ └──────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────┐ │
│ │ HolySheep AI │ │
│ │ Unified Gateway │ │
│ └──────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
HolySheep AI 연동을 위한 보안 설정
HolySheep AI는 단일 API 키로 여러 AI 모델厂商에 접근할 수 있는 통합 게이트웨이입니다. 저는 실무에서 HolySheep AI를 활용하여 복잡한 다중 API 키 관제를 단일화하고, 일원화된 보안 정책을 적용하고 있습니다.
1단계: 안전한 API 키 관리
#!/usr/bin/env python3
"""
HolySheep AI API 통합 - 제로 트러스트 보안 구현
환경변수 기반 API 키 관리 및 요청 검증
"""
import os
import hashlib
import hmac
import time
import json
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
from dataclasses import dataclass, field
from enum import Enum
import httpx
from functools import wraps
===== 보안 설정 =====
class SecurityConfig:
"""제로 트러스트 보안 설정"""
# HolySheep AI 설정
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "")
#_rate_limiting
MAX_REQUESTS_PER_MINUTE = 60
MAX_TOKENS_PER_DAY = 1_000_000
BURST_SIZE = 10
# 접근 제어
ALLOWED_MODELS = [
"gpt-4.1",
"claude-sonnet-4-20250514",
"gemini-2.5-flash",
"deepseek-v3.2"
]
# IP 화이트리스트 (CIDR 표기법)
ALLOWED_IP_RANGES = [
"10.0.0.0/8",
"172.16.0.0/12",
"192.168.0.0/16"
]
# 요청 시간 제한 (초)
REQUEST_TIMEOUT = 30
MAX_RETRIES = 3
@dataclass
class APIKey:
"""API 키 메타데이터"""
key_id: str
key_hash: str
created_at: datetime
expires_at: Optional[datetime]
permissions: list[str] = field(default_factory=list)
rate_limit: int = 60
last_used: Optional[datetime] = None
usage_count: int = 0
def is_expired(self) -> bool:
if self.expires_at is None:
return False
return datetime.now() > self.expires_at
def to_dict(self) -> Dict[str, Any]:
return {
"key_id": self.key_id,
"created_at": self.created_at.isoformat(),
"expires_at": self.expires_at.isoformat() if self.expires_at else None,
"permissions": self.permissions,
"rate_limit": self.rate_limit,
"last_used": self.last_used.isoformat() if self.last_used else None,
"usage_count": self.usage_count
}
class SecureAPIKeyManager:
"""안전한 API 키 관리 클래스"""
def __init__(self):
self._keys: Dict[str, APIKey] = {}
self._audit_log: list[Dict] = []
def create_key(
self,
name: str,
permissions: list[str],
expires_in_days: int = 90,
rate_limit: int = 60
) -> tuple[str, APIKey]:
"""새 API 키 생성"""
# 키 생성
key_value = self._generate_secure_key()
key_hash = hashlib.sha256(key_value.encode()).hexdigest()
key_id = hashlib.sha256(f"{name}{time.time()}".encode()).hexdigest()[:16]
api_key = APIKey(
key_id=key_id,
key_hash=key_hash,
created_at=datetime.now(),
expires_at=datetime.now() + timedelta(days=expires_in_days),
permissions=permissions,
rate_limit=rate_limit
)
self._keys[key_value] = api_key
self._audit("KEY_CREATED", key_id, {"permissions": permissions})
return key_value, api_key
def _generate_secure_key(self) -> str:
"""암호학적으로 안전한 키 생성"""
import secrets
return f"hsa_{secrets.token_urlsafe(32)}"
def validate_key(self, key: str) -> tuple[bool, Optional[APIKey], str]:
"""키 검증"""
if not key:
return False, None, "Empty key"
api_key = self._keys.get(key)
if api_key is None:
self._audit("KEY_INVALID", "unknown", {"reason": "Not found"})
return False, None, "Invalid key"
if api_key.is_expired():
self._audit("KEY_EXPIRED", api_key.key_id, {"reason": "Time expired"})
return False, None, "Key expired"
# 사용량 업데이트
api_key.last_used = datetime.now()
api_key.usage_count += 1
self._audit("KEY_VALIDATED", api_key.key_id, {"usage_count": api_key.usage_count})
return True, api_key, "Valid"
def _audit(self, event: str, key_id: str, data: Dict):
"""감사 로그 기록"""
self._audit_log.append({
"timestamp": datetime.now().isoformat(),
"event": event,
"key_id": key_id,
"data": data
})
def get_audit_log(self, limit: int = 100) -> list[Dict]:
"""감사 로그 조회"""
return self._audit_log[-limit:]
===== 사용 예시 =====
if __name__ == "__main__":
manager = SecureAPIKeyManager()
# 새 API 키 생성
key, metadata = manager.create_key(
name="production-service",
permissions=["chat:read", "chat:write", "models:list"],
expires_in_days=90,
rate_limit=100
)
print(f"생성된 API 키: {key[:20]}...")
print(f"키 메타데이터: {json.dumps(metadata.to_dict(), indent=2)}")
# 키 검증
is_valid, api_key_obj, msg = manager.validate_key(key)
print(f"검증 결과: {is_valid}, {msg}")
# 감사 로그 확인
print(f"감사 로그: {json.dumps(manager.get_audit_log(), indent=2)}")
2단계: 요청 검증 미들웨어
#!/usr/bin/env python3
"""
HolySheep AI API 호출 - 제로 트러스트 검증 미들웨어
요청 무결성 검증 및 안전한 API 호출
"""
import asyncio
import hashlib
import hmac
import json
import time
import uuid
from typing import Optional, Callable, Any
from dataclasses import dataclass
import httpx
===== 요청 서명 검증 =====
class RequestSigner:
"""HMAC 기반 요청 서명"""
def __init__(self, secret_key: str):
self.secret_key = secret_key.encode()
def sign(self, payload: str, timestamp: int) -> str:
"""요청 서명 생성"""
message = f"{timestamp}:{payload}"
signature = hmac.new(
self.secret_key,
message.encode(),
hashlib.sha256
).hexdigest()
return signature
def verify(self, payload: str, timestamp: int, signature: str) -> bool:
"""서명 검증"""
# 시간 기반 검증 (5분 윈도우)
current_time = int(time.time())
if abs(current_time - timestamp) > 300:
return False
expected = self.sign(payload, timestamp)
return hmac.compare_digest(expected, signature)
@dataclass
class APIRequest:
"""API 요청 데이터"""
method: str
endpoint: str
headers: dict
body: Optional[dict] = None
def to_signable(self) -> str:
body_str = json.dumps(self.body, sort_keys=True) if self.body else ""
return f"{self.method}:{self.endpoint}:{body_str}"
class ZeroTrustAPIClient:
"""제로 트러스트 AI API 클라이언트"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
timeout: int = 30
):
self.api_key = api_key
self.base_url = base_url
self.signer = RequestSigner(api_key)
self._rate_limiter = RateLimiter(max_requests=60, window=60)
# httpx 클라이언트 설정
self.client = httpx.AsyncClient(
timeout=timeout,
limits=httpx.Limits(max_keepalive_connections=20, max_connections=100)
)
# 요청/응답 캐시
self._response_cache: dict = {}
self._cache_ttl = 300 # 5분
async def chat_completions(
self,
model: str,
messages: list[dict],
temperature: float = 0.7,
max_tokens: int = 2048,
validate: bool = True
) -> dict:
"""AI API 호출 - 검증 및 속도 제한 포함"""
request_id = str(uuid.uuid4())
timestamp = int(time.time())
# 1. 속도 제한 검증
if not self._rate_limiter.try_acquire():
raise RateLimitExceededError(
f"Rate limit exceeded. Retry after {self._rate_limiter.retry_after()}s"
)
# 2. 요청 데이터 구성
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
# 3. 요청 서명
if validate:
signable = f"{timestamp}:{json.dumps(payload, sort_keys=True)}"
signature = self.signer.sign(signable, timestamp)
else:
signature = ""
# 4. HolySheep AI API 호출
url = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": request_id,
"X-Timestamp": str(timestamp),
"X-Signature": signature
}
start_time = time.perf_counter()
try:
response = await self.client.post(url, json=payload, headers=headers)
latency_ms = (time.perf_counter() - start_time) * 1000
# 응답 검증
if validate:
self._validate_response(response, request_id)
result = response.json()
# 메트릭스 기록
self._record_metrics(model, latency_ms, response.status_code)
return {
"request_id": request_id,
"model": model,
"response": result,
"latency_ms": round(latency_ms, 2),
"usage": result.get("usage", {})
}
except httpx.TimeoutException:
raise APITimeoutError(f"Request timeout after {timeout}s")
except httpx.HTTPStatusError as e:
raise APIError(f"HTTP {e.response.status_code}: {e.response.text}")
def _validate_response(self, response: httpx.Response, request_id: str):
"""응답 무결성 검증"""
if response.status_code != 200:
return
# 필수 헤더 검증
content_type = response.headers.get("content-type", "")
if "application/json" not in content_type:
raise SecurityError(f"Invalid content type: {content_type}")
# 응답 본문 검증
try:
data = response.json()
if "choices" not in data:
raise SecurityError("Invalid response format")
except json.JSONDecodeError:
raise SecurityError("Invalid JSON response")
def _record_metrics(self, model: str, latency_ms: float, status_code: int):
"""성능 메트릭스 기록"""
print(f"[METRICS] model={model} latency={latency_ms}ms status={status_code}")
async def close(self):
"""클라이언트 종료"""
await self.client.aclose()
class RateLimiter:
"""토큰 버킷 기반 속도 제한"""
def __init__(self, max_requests: int, window: int):
self.max_requests = max_requests
self.window = window
self.tokens = max_requests
self.last_update = time.time()
self._lock = asyncio.Lock()
async def try_acquire(self) -> bool:
async with self._lock:
now = time.time()
elapsed = now - self.last_update
# 토큰 복원
self.tokens = min(
self.max_requests,
self.tokens + elapsed * (self.max_requests / self.window)
)
self.last_update = now
if self.tokens >= 1:
self.tokens -= 1
return True
return False
def retry_after(self) -> int:
return int((1 - self.tokens) * (self.window / self.max_requests))
===== 커스텀 예외 =====
class RateLimitExceededError(Exception):
pass
class APITimeoutError(Exception):
pass
class APIError(Exception):
pass
class SecurityError(Exception):
pass
===== 사용 예시 =====
async def main():
"""HolySheep AI API 호출 예시"""
client = ZeroTrustAPIClient(
api_key="YOUR_HOLYSHEEP_API_KEY", # HolySheep AI API 키
base_url="https://api.holysheep.ai/v1"
)
try:
# GPT-4.1 모델 호출
result = await client.chat_completions(
model="gpt-4.1",
messages=[
{"role": "system", "content": "당신은 보안 전문가입니다."},
{"role": "user", "content": "제로 트러스트 아키텍처의 핵심 원칙을 설명해주세요."}
],
temperature=0.7,
max_tokens=1024
)
print(f"요청 ID: {result['request_id']}")
print(f"지연 시간: {result['latency_ms']}ms")
print(f"모델: {result['model']}")
print(f"토큰 사용량: {result['usage']}")
print(f"응답: {result['response']['choices'][0]['message']['content']}")
except RateLimitExceededError as e:
print(f"속도 제한 초과: {e}")
except APIError as e:
print(f"API 오류: {e}")
finally:
await client.close()
if __name__ == "__main__":
asyncio.run(main())
성능 최적화와 비용 관리
저는 HolySheep AI를 실무에서 활용하며 모델별 비용 최적화를 지속적으로 수행하고 있습니다. HolySheep AI의 가격표는 매우 경쟁력 있으며, 이를 활용한 비용 절감 전략을 소개하겠습니다.
모델별 비용 비교 (1M 토큰 기준)
| 모델 | 가격 | 지연 시간 (평균) | 최적 사용 사례 |
|------|------|------------------|----------------|
| GPT-4.1 | $8.00 | 850ms | 복잡한 추론, 코드 생성 |
| Claude Sonnet 4 | $15.00 | 920ms | 장문 분석, 창의적 작업 |
| Gemini 2.5 Flash | $2.50 | 420ms | 대량 배치 처리, 실시간 응답 |
| DeepSeek V3.2 | $0.42 | 680ms | 코스트 센서티브 작업 |
Gemini 2.5 Flash는 GPT-4o-mini 대비 40% 저렴하면서 더 빠른 응답 시간을 제공합니다. 제 경험상 일반적인 챗봇用途에는 Gemini Flash로 충분하며, 복잡한 추론이 필요한 경우에만 상위 모델을 사용하는 것이 효과적입니다.
동시성 제어 및 배치 처리
#!/usr/bin/env python3
"""
HolySheep AI - 동시성 제어 및 배치 처리 최적화
Semaphore 기반 동시 연결 관리 및 비용 최적화
"""
import asyncio
import time
import json
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum
import httpx
from collections import defaultdict
===== 모델별 비용 및 제한 설정 =====
class ModelConfig:
"""모델별 최적화 설정"""
MODELS = {
"gpt-4.1": {
"cost_per_1k_input": 0.008, # $8/1MTok
"cost_per_1k_output": 0.032,
"max_tokens": 128000,
"avg_latency_ms": 850
},
"claude-sonnet-4-20250514": {
"cost_per_1k_input": 0.015, # $15/1MTok
"cost_per_1k_output": 0.075,
"max_tokens": 200000,
"avg_latency_ms": 920
},
"gemini-2.5-flash": {
"cost_per_1k_input": 0.001, # $2.50/1MTok
"cost_per_1k_output": 0.004,
"max_tokens": 1000000,
"avg_latency_ms": 420
},
"deepseek-v3.2": {
"cost_per_1k_input": 0.00027, # $0.42/1MTok
"cost_per_1k_output": 0.001,
"max_tokens": 64000,
"avg_latency_ms": 680
}
}
@classmethod
def get_model_for_task(cls, task_type: str) -> str:
"""작업 유형별 최적 모델 선택"""
task_models = {
"simple_qa": "deepseek-v3.2",
"chat": "gemini-2.5-flash",
"code_generation": "gpt-4.1",
"complex_reasoning": "claude-sonnet-4-20250514",
"batch_processing": "gemini-2.5-flash"
}
return task_models.get(task_type, "gemini-2.5-flash")
@classmethod
def estimate_cost(cls, model: str, input_tokens: int, output_tokens: int) -> float:
"""비용 추정"""
if model not in cls.MODELS:
raise ValueError(f"Unknown model: {model}")
config = cls.MODELS[model]
input_cost = (input_tokens / 1000) * config["cost_per_1k_input"]
output_cost = (output_tokens / 1000) * config["cost_per_1k_output"]
return input_cost + output_cost
@dataclass
class BatchRequest:
"""배치 요청 데이터"""
id: str
model: str
messages: List[Dict]
temperature: float = 0.7
max_tokens: int = 1024
priority: int = 0 # 0: 낮음, 1: 보통, 2: 높음
@dataclass
class BatchResult:
"""배치 처리 결과"""
request_id: str
success: bool
response: Optional[Dict]
latency_ms: float
cost: float
error: Optional[str] = None
class ConcurrencyController:
"""동시성 제어기"""
def __init__(
self,
max_concurrent: int = 10,
rate_limit_per_minute: int = 60
):
self.max_concurrent = max_concurrent
self.rate_limit = rate_limit_per_minute
self._semaphore = asyncio.Semaphore(max_concurrent)
self._rate_limiter = asyncio.Semaphore(rate_limit_per_minute)
self._active_tasks = 0
self._metrics = defaultdict(int)
async def execute(
self,
coro: Any,
task_id: str
) -> Any:
"""동시성 제어된 작업 실행"""
async with self._semaphore:
async with self._rate_limiter:
self._active_tasks += 1
start_time = time.perf_counter()
try:
result = await coro
self._metrics[f"{task_id}_success"] += 1
return result
except Exception as e:
self._metrics[f"{task_id}_error"] += 1
raise
finally:
self._active_tasks -= 1
self._metrics[f"{task_id}_latency"] += (
time.perf_counter() - start_time
) * 1000
def get_metrics(self) -> Dict:
"""메트릭스 조회"""
return dict(self._metrics)
class BatchProcessor:
"""배치 처리 최적화"""
def __init__(
self,
api_key: str,
controller: ConcurrencyController
):
self.api_key = api_key
self.controller = controller
self.client = httpx.AsyncClient(
timeout=60.0,
limits=httpx.Limits(max_keepalive_connections=20, max_connections=100)
)
async def process_batch(
self,
requests: List[BatchRequest],
optimize_cost: bool = True
) -> List[BatchResult]:
"""배치 처리 실행"""
# 비용 최적화: 동일 모델 요청 그룹화
if optimize_cost:
requests = self._optimize_batch(requests)
# 우선순위 순으로 정렬
requests.sort(key=lambda x: -x.priority)
tasks = [
self._process_single(req)
for req in requests
]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [
r if isinstance(r, BatchResult) else BatchResult(
request_id="error",
success=False,
response=None,
latency_ms=0,
cost=0,
error=str(r)
)
for r in results
]
def _optimize_batch(self, requests: List[BatchRequest]) -> List[BatchRequest]:
"""배치 최적화: 모델 전환으로 비용 절감"""
optimized = []
for req in requests:
# 간단한 Q&A는 더 저렴한 모델로 전환
if self._is_simple_task(req):
original_cost = ModelConfig.estimate_cost(
req.model, 500, 200
)
# Gemini Flash로 전환 가능 여부 확인
if req.model not in ["deepseek-v3.2", "gemini-2.5-flash"]:
simple_model = "gemini-2.5-flash"
new_cost = ModelConfig.estimate_cost(
simple_model, 500, 200
)
if new_cost < original_cost * 0.5:
print(f"비용 최적화: {req.model} → {simple_model}")
req = BatchRequest(
id=req.id,
model=simple_model,
messages=req.messages,
temperature=req.temperature,
max_tokens=req.max_tokens,
priority=req.priority
)
optimized.append(req)
return optimized
def _is_simple_task(self, req: BatchRequest) -> bool:
"""단순 작업 판별"""
if not req.messages:
return True
last_message = req.messages[-1].get("content", "")
# 간단한 질문 패턴
simple_patterns = ["?", "무엇", "어떻게", "언제", "在哪里"]
is_simple = any(pattern in last_message for pattern in simple_patterns)
# 긴 컨텍스트는 제외
is_short = len(last_message) < 200
return is_simple and is_short
async def _process_single(self, req: BatchRequest) -> BatchResult:
"""단일 요청 처리"""
async def _call_api():
payload = {
"model": req.model,
"messages": req.messages,
"temperature": req.temperature,
"max_tokens": req.max_tokens
}
response = await self.controller.execute(
self.client.post(
"https://api.holysheep.ai/v1/chat/completions",
json=payload,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
),
task_id=req.id
)
return response.json()
start_time = time.perf_counter()
try:
response = await _call_api()
latency_ms = (time.perf_counter() - start_time) * 1000
# 비용 계산
usage = response.get("usage", {})
cost = ModelConfig.estimate_cost(
req.model,
usage.get("prompt_tokens", 0),
usage.get("completion_tokens", 0)
)
return BatchResult(
request_id=req.id,
success=True,
response=response,
latency_ms=round(latency_ms, 2),
cost=round(cost, 6)
)
except Exception as e:
return BatchResult(
request_id=req.id,
success=False,
response=None,
latency_ms=(time.perf_counter() - start_time) * 1000,
cost=0,
error=str(e)
)
async def close(self):
await self.client.aclose()
===== 사용 예시 =====
async def main():
"""배치 처리 예시"""
controller = ConcurrencyController(
max_concurrent=10,
rate_limit_per_minute=60
)
processor = BatchProcessor(
api_key="YOUR_HOLYSHEEP_API_KEY",
controller=controller
)
# 테스트 요청 생성
requests = [
BatchRequest(
id=f"req-{i}",
model="gpt-4.1",
messages=[
{"role": "user", "content": f"질문 {i}: 테스트 메시지입니다."}
],
priority=i % 3
)
for i in range(50)
]
print(f"총 {len(requests)}개 요청 처리 시작...")
start_time = time.perf_counter()
results = await processor.process_batch(requests, optimize_cost=True)
total_time = time.perf_counter() - start_time
# 결과 분석
success_count = sum(1 for r in results if r.success)
total_cost = sum(r.cost for r in results)
avg_latency = sum(r.latency_ms for r in results) / len(results)
print(f"\n===== 배치 처리 결과 =====")
print(f"총 처리 시간: {total_time:.2f}s")
print(f"성공: {success_count}/{len(results)}")
print(f"총 비용: ${total_cost:.4f}")
print(f"평균 지연: {avg_latency:.2f}ms")
print(f"처리량: {len(results)/total_time:.2f} req/s")
await processor.close()
if __name__ == "__main__":
asyncio.run(main())
모니터링 및 감사 로깅
제로 트러스트 환경에서 모든 API 호출은 감사 가능한 형태로 기록되어야 합니다. HolySheep AI의 경우 모든 호출에 대한 상세 로그를 제공하며, 이를 활용한 보안 모니터링 시스템을 구축하겠습니다.
#!/usr/bin/env python3
"""
HolySheep AI - 실시간 보안 모니터링 및 침입 탐지
감사 로그 기반 비정상 패턴 탐지
"""
import time
import hashlib
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from collections import defaultdict, deque
from enum import Enum
import statistics
===== 로그 데이터 구조 =====
class EventType(Enum):
"""이벤트 유형"""
API_REQUEST = "api_request"
AUTH_SUCCESS = "auth_success"
AUTH_FAILURE = "auth_failure"
RATE_LIMIT_EXCEEDED = "rate_limit_exceeded"
INVALID_SIGNATURE = "invalid_signature"
SUSPICIOUS_PATTERN = "suspicious_pattern"
@dataclass
class AuditEvent:
"""감사 이벤트"""
timestamp: datetime
event_type: EventType
api_key_id: str
ip_address: str
model: str
tokens_used: int
latency_ms: float
status_code: int
metadata: Dict[str, Any] = field(default_factory=dict)
event_id: str = ""
def __post_init__(self):
if not self.event_id:
content = f"{self.timestamp}{self.event_type.value}{self.api_key_id}"
self.event_id = hashlib.sha256(content.encode()).hexdigest()[:16]
def to_dict(self) -> Dict:
return {
"event_id": self.event_id,
"timestamp": self.timestamp.isoformat(),
"event_type": self.event_type.value,
"api_key_id": self.api_key_id,
"ip_address": self.ip_address,
"model": self.model,
"tokens_used": self.tokens_used,
"latency_ms": self.latency_ms,
"status_code": self.status_code,
"metadata": self.metadata
}
class SecurityMonitor:
"""보안 모니터링 시스템"""
def __init__(
self,
window_minutes: int = 15,
alert_threshold: int = 100
):
self.window = timedelta(minutes=window_minutes)
self.alert_threshold = alert_threshold
# 이벤트 버퍼 (순환 버퍼)
self._events: deque = deque(maxlen=10000)
# 패턴 탐지용 카운터
self._ip_request_counts: Dict[str, deque] = defaultdict(
lambda: deque(maxlen=1000)
)
self._api_key_usage: Dict[str, deque] = defaultdict(
lambda: deque(maxlen=1000)
)
self._failed_auths: Dict[str, List[datetime]] = defaultdict(list)
# 침입 탐지 규칙
self._detection_rules = [
self._rule_unusual_volume,
self._rule_rapid_requests,
self._rule_failed_auth_burst,
self._rule_off_hours_activity,
self._rule_geo_anomaly
]
def record_event(self, event: AuditEvent):
"""이벤트 기록"""
self._events.append(event)
# 카운터 업데이트
self._ip_request_counts[event.ip_address].append(event.timestamp)
self._api_key_usage[event.api_key_id].append(event)
# 실패 인증 추적
if event.event_type == EventType.AUTH_FAILURE:
self._failed_auths[event.api_key_id].append(event.timestamp)
# 규칙 기반 탐지
alerts = self._run_detection_rules(event)
if alerts:
self._trigger_alerts(alerts)
def _rule_unusual_volume(self, event: AuditEvent) -> Optional[Dict]:
"""비정상 볼륨 탐지"""
ip_events = self._ip_request_counts[event.ip_address]
# 마지막 1분간 요청 수
now = datetime.now()
recent = [e for e in ip_events if now - e < timedelta(minutes=1)]
if len(recent) > 100:
return {
"rule": "unusual_volume",
"severity": "HIGH",
"ip_address": event.ip_address,
"requests_per_minute": len(recent),
"message": f"비정상적으로 높은 요청 볼륨: {len(recent)} req/min"
}
return None
def _rule_rapid_requests(self, event: AuditEvent) -> Optional[Dict]:
"""급격한 요청 패턴 탐지"""
key_events = self._api_key_usage[event.api_key_id]
if len(key_events) < 2:
return None
# 최근 5개 요청의 지연 시간 분석
recent = list(key_events)[-5:]
if len(recent) < 2:
return None
latencies = [e.latency_ms for e in recent]
# 모든 요청이 50ms 이하인 경우 (자동화 봇 패턴)
if all(l < 50 for