AI 安全监控：异常 API 调用模式识别与自动封禁

จุดเริ่มต้นของปัญหา

คืนหนึ่งในเดือนพฤษภาคม ระบบของผมเริ่มส่ง Alert ไม่หยุด "ConnectionError: timeout after 30000ms" แจ้งเตือนจากหลาย Endpoint พร้อมกัน พอเช็คดู Log พบว่า Token Usage พุ่งสูงผิดปกติ 2,847,000 Tokens ในเวลา 15 นาที ปรากฏว่ามี IP จากต่างประเทศกำลังพยายาม Brute-force API Key ของระบบ หลังจากวันนั้น ผมตัดสินใจสร้างระบบ Monitoring ที่ครอบคลุมและชาญฉลาดขึ้น บทความนี้จะสอนวิธีสร้างระบบตรวจจับ API ที่ผิดปกติและแบนอัตโนมัติโดยใช้ HolySheep AI ซึ่งมีอัตราเรทที่ประหยัดถึง 85% เมื่อเทียบกับผู้ให้บริการอื่น ราคาเริ่มต้นเพียง $0.42/ล้าน Tokens สำหรับ DeepSeek V3.2

ทำความเข้าใจรูปแบบการโจมตี API ที่พบบ่อย

ก่อนสร้างระบบป้องกัน ต้องเข้าใจภัยคุกคามก่อน การโจมตี API ส่วนใหญ่มีลักษณะเฉพาะ การเรียกที่ผิดปกติมักมีความถี่สูงผิดปกติ (Request มากกว่า 10 ครั้ง/วินาที) จาก IP เดียว, ใช้ Failed Authentication ติดต่อกันหลายครั้ง, เรียก Endpoint ที่ไม่เคยใช้มาก่อน, หรือส่ง Payload ขนาดใหญ่ผิดปกติ ระบบ Monitoring ที่ดีต้องเก็บข้อมูล Request แต่ละครั้งอย่างเป็นระบบ รวมถึง IP Address, Timestamp, Endpoint, Status Code, Response Time และ Token Usage

สร้างระบบตรวจจับแบบเรียลไทม์

โค้ดด้านล่างแสดงการสร้าง API Monitoring System พื้นฐานที่เชื่อมต่อกับ HolySheep AI API

import httpx
import time
from datetime import datetime, timedelta
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import asyncio

@dataclass
class APIRequest:
    ip_address: str
    timestamp: datetime
    endpoint: str
    status_code: int
    response_time_ms: float
    token_usage: int
    user_agent: str

@dataclass
class IPStats:
    request_count: int = 0
    failed_auth_count: int = 0
    total_tokens: int = 0
    avg_response_time: float = 0.0
    endpoints_called: List[str] = field(default_factory=list)
    last_request: datetime = field(default_factory=datetime.now)
    blocked: bool = False
    block_reason: Optional[str] = None
    block_until: Optional[datetime] = None

class APIMonitor:
    def __init__(self, holy_sheep_api_key: str):
        self.base_url = "https://api.holysheep.ai/v1"
        self.api_key = holy_sheep_api_key
        self.ip_stats: Dict[str, IPStats] = defaultdict(IPStats)
        self.alert_threshold = {
            "requests_per_minute": 60,
            "failed_auth": 5,
            "token_burst": 100000,
            "response_time_ms": 5000,
        }
        
    async def track_request(self, request: APIRequest):
        stats = self.ip_stats[request.ip_address]
        stats.request_count += 1
        stats.total_tokens += request.token_usage
        stats.last_request = request.timestamp
        stats.avg_response_time = (
            stats.avg_response_time * (stats.request_count - 1) + 
            request.response_time_ms
        ) / stats.request_count
        
        if request.endpoint not in stats.endpoints_called:
            stats.endpoints_called.append(request.endpoint)
            
        if request.status_code == 401:
            stats.failed_auth_count += 1
            
        await self._check_thresholds(request.ip_address)
        
    async def _check_thresholds(self, ip_address: str):
        stats = self.ip_stats[ip_address]
        
        if stats.blocked:
            if datetime.now() < stats.block_until:
                return
            else:
                stats.blocked = False
                
        recent_window = datetime.now() - timedelta(minutes=1)
        if stats.last_request >= recent_window:
            if stats.request_count >= self.alert_threshold["requests_per_minute"]:
                await self._block_ip(ip_address, "ส่ง Request เกิน 60 ครั้ง/นาที")
                
        if stats.failed_auth_count >= self.alert_threshold["failed_auth"]:
            await self._block_ip(ip_address, f"Failed Auth {stats.failed_auth_count} ครั้งติดต่อกัน")
            
        if stats.total_tokens >= self.alert_threshold["token_burst"]:
            await self._block_ip(ip_address, f"Token Usage ระเบิด {stats.total_tokens:,} tokens")

monitor = APIMonitor("YOUR_HOLYSHEEP_API_KEY")

ระบบแบนอัตโนมัติพร้อม Webhook แจ้งเตือน

เมื่อตรวจพบพฤติกรรมผิดปกติ ระบบจะแบน IP และส่งแจ้งเตือนไปยัง Webhook ทันที

import hashlib
import json

class AutoBlockSystem:
    def __init__(self, webhook_url: str = None):
        self.blocked_ips: Dict[str, dict] = {}
        self.webhook_url = webhook_url
        self.block_duration = timedelta(minutes=30)
        
    async def _block_ip(self, ip_address: str, reason: str):
        block_data = {
            "ip_address": ip_address,
            "reason": reason,
            "blocked_at": datetime.now().isoformat(),
            "block_until": (datetime.now() + self.block_duration).isoformat(),
            "auto_unblock": True,
            "severity": self._calculate_severity(reason)
        }
        
        self.blocked_ips[ip_address] = block_data
        
        if self.webhook_url:
            await self._send_alert(block_data)
            
        print(f"🚫 บล็อก IP {ip_address} เนื่องจาก: {reason}")
        
    def _calculate_severity(self, reason: str) -> str:
        if "Failed Auth" in reason:
            return "HIGH"
        elif "Token Usage" in reason:
            return "CRITICAL"
        return "MEDIUM"
        
    async def _send_alert(self, block_data: dict):
        async with httpx.AsyncClient() as client:
            await client.post(
                self.webhook_url,
                json={
                    "event": "ip_blocked",
                    "data": block_data,
                    "signature": self._generate_signature(block_data)
                }
            )
            
    def _generate_signature(self, data: dict) -> str:
        message = json.dumps(data, sort_keys=True)
        return hashlib.sha256(
            message.encode() + self.api_key.encode()
        ).hexdigest()
        
    def is_blocked(self, ip_address: str) -> bool:
        if ip_address not in self.blocked_ips:
            return False
            
        block_info = self.blocked_ips[ip_address]
        if datetime.now() >= datetime.fromisoformat(block_info["block_until"]):
            del self.blocked_ips[ip_address]
            return False
            
        return True

block_system = AutoBlockSystem(webhook_url="https://your-webhook-endpoint.com/alert")

สร้าง Middleware สำหรับ FastAPI

โค้ดด้านล่างแสดงการใช้งานร่วมกับ FastAPI เพื่อป้องกันทุก Request

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
import uvicorn

app = FastAPI(title="Secure AI API with HolySheep")

class SecurityMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, monitor: APIMonitor, blocker: AutoBlockSystem):
        super().__init__(app)
        self.monitor = monitor
        self.blocker = blocker
        
    async def dispatch(self, request: Request, call_next):
        client_ip = request.client.host
        user_agent = request.headers.get("user-agent", "unknown")
        
        if self.blocker.is_blocked(client_ip):
            return JSONResponse(
                status_code=403,
                content={"error": "IP ถูกบล็อกชั่วคราว", 
                        "retry_after": "30 นาที"}
            )
            
        start_time = time.time()
        
        response = await call_next(request)
        
        response_time = (time.time() - start_time) * 1000
        
        api_request = APIRequest(
            ip_address=client_ip,
            timestamp=datetime.now(),
            endpoint=str(request.url.path),
            status_code=response.status_code,
            response_time_ms=response_time,
            token_usage=int(response.headers.get("X-Token-Usage", 0)),
            user_agent=user_agent
        )
        
        await self.monitor.track_request(api_request)
        
        return response

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    response = await forward_to_holysheep(request)
    return response

@app.get("/health")
async def health_check():
    return {"status": "healthy", "blocked_ips": len(blocker.blocked_ips)}

app.add_middleware(
    SecurityMiddleware,
    monitor=monitor,
    blocker=block_system
)

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

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. ConnectionError: timeout after 30000ms

ข้อผิดพลาดนี้เกิดจากการเรียก API ซ้ำเร็วเกินไปจน Server ปฏิเสธ วิธีแก้คือเพิ่ม Retry Logic พร้อม Exponential Backoff และตั้งค่า Timeout ให้เหมาะสม

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=2, min=1, max=10)
)
async def call_holysheep_api(messages: list):
    async with httpx.AsyncClient(timeout=45.0) as client:
        response = await client.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={
                "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
                "Content-Type": "application/json"
            },
            json={
                "model": "gpt-4o",
                "messages": messages,
                "max_tokens": 2000
            }
        )
        
        if response.status_code == 429:
            raise Exception("Rate limit exceeded - รอสักครู่")
            
        response.raise_for_status()
        return response.json()

2. 401 Unauthorized - Invalid API Key

ข้อผิดพลาดนี้เกิดจาก API Key ไม่ถูกต้องหรือหมดอายุ ตรวจสอบว่าใช้ Key จาก HolySheep เท่านั้นและไม่ได้เก็บในโค้ดโดยตรง ให้ใช้ Environment Variable แทน

import os

def get_api_key() -> str:
    api_key = os.environ.get("HOLYSHEEP_API_KEY")
    
    if not api_key:
        raise ValueError(
            "ไม่พบ HOLYSHEEP_API_KEY ใน Environment Variable\n"
            "กรุณาตั้งค่าด้วย: export HOLYSHEEP_API_KEY='your-key-here'"
        )
        
    if not api_key.startswith("hs_"):
        raise ValueError(
            "API Key ไม่ถูกต้อง - HolySheep Key ต้องขึ้นต้นด้วย 'hs_' "
            "ลงทะเบียนได้ที่: https://www.holysheep.ai/register"
        )
        
    return api_key

HOLYSHEEP_API_KEY = get_api_key()

3. Rate Limit Exceeded - เกินโควต้า

ข้อผิดพลาด 429 เกิดจากการส่ง Request เกิน Rate Limit ของแผนที่ใช้อยู่ วิธีแก้คือตรวจสอบ Usage Dashboard และเพิ่ม Rate Limiter ในโค้ด

import time
from collections import deque

class RateLimiter:
    def __init__(self, max_requests: int = 60, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = deque()
        
    def is_allowed(self) -> bool:
        now = time.time()
        
        while self.requests and self.requests[0] <= now - self.window_seconds:
            self.requests.popleft()
            
        if len(self.requests) < self.max_requests:
            self.requests.append(now)
            return True
            
        return False
        
    def wait_time(self) -> float:
        if not self.requests:
            return 0.0
            
        oldest_request = self.requests[0]
        time_passed = time.time() - oldest_request
        
        if time_passed >= self.window_seconds:
            return 0.0
            
        return self.window_seconds - time_passed

rate_limiter = RateLimiter(max_requests=60, window_seconds=60)

async def limited_api_call(messages: list):
    if not rate_limiter.is_allowed():
        wait = rate_limiter.wait_time()
        print(f"รอ {wait:.1f} วินาที ก่อนส่ง Request ถัดไป...")
        await asyncio.sleep(wait)
        
    return await call_holysheep_api(messages)

ประสบการณ์จริง: ลดค่าใช้จ่าย 85% ด้วย HolySheep AI

หลังจากติดตั้งระบบ Monitoring นี้และเปลี่ยนมาใช้ HolySheep AI ผลลัพธ์ที่ได้น่าประทับใจมาก ก่อนหน้านี้ใช้ OpenAI และ Anthropic โดยตรงเดือนละปร