บทนำ: ปัญหาที่เกิดขึ้นจริง

ในช่วงเดือนมีนาคม 2026 ทีมพัฒนาของเราเผชิญกับเหตุการณ์วิกฤตที่ส่งผลกระทบต่อระบบ API ทั้งหมด การเรียกใช้ API ที่ควรจะใช้เวลาตอบสนองประมาณ 45-50 มิลลิวินาที กลับเกิดข้อผิดพลาดแบบ ConnectionError: timeout after 30s อย่างต่อเนื่อง หลังจากตรวจสอบอย่างละเอียด พบว่าเป็นการโจมตีแบบ Brute Force ที่พยายามเข้าถึง API Key ผ่านการสุ่มทดลองหลายพันครั้งต่อนาที ทำให้เกิดภาระเกินขีดจำกัดและระบบปฏิเสธการเข้าถึงทั้งหมด จากเหตุการณ์นี้ เราได้พัฒนาระบบ API Security Audit ที่สามารถตรวจจับรูปแบบการเรียกใช้ที่ผิดปกติและส่งการแจ้งเตือนได้อย่างทันท่วงที ซึ่งจะอธิบายรายละเอียดทั้งหมดในบทความนี้

หลักการทำงานของ API Security Audit

ระบบตรวจสอบความปลอดภัย API ที่มีประสิทธิภาพต้องมีความสามารถในการวิเคราะห์พฤติกรรมการใช้งานแบบเรียลไทม์ โดยอาศัยหลักการต่อไปนี้:

การติดตั้งและใช้งานระบบตรวจสอบ

ก่อนอื่นเราต้องติดตั้งไลบรารีที่จำเป็นสำหรับการพัฒนาระบบ API Security Audit โดยใช้คำสั่ง pip install สำหรับไลบรารีที่เกี่ยวข้อง จากนั้นสร้างคลาส APIAudit เพื่อจัดการกับการตรวจสอบความปลอดภัยทั้งหมด ระบบจะใช้งาน Redis สำหรับการเก็บข้อมูลการใช้งานแบบเรียลไทม์ และใช้โครงสร้างข้อมูลแบบ Sorted Set ในการติดตามจำนวนคำขอต่อช่วงเวลา 
import time
import hashlib
import redis
from datetime import datetime, timedelta
from collections import defaultdict
from typing import Dict, List, Optional, Tuple
import json
import logging


ตั้งค่า logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class APIAudit:
    """ระบบตรวจสอบความปลอดภัย API"""
    
    def __init__(
        self,
        redis_host: str = "localhost",
        redis_port: int = 6379,
        redis_db: int = 0,
        rate_limit: int = 100,  # จำนวนคำขอต่อนาที
        window_size: int = 60   # ขนาดหน้าต่างเวลา (วินาที)
    ):
        self.redis_client = redis.Redis(
            host=redis_host,
            port=redis_port,
            db=redis_db,
            decode_responses=True
        )
        self.rate_limit = rate_limit
        self.window_size = window_size
        self.anomaly_threshold = 3  # จำนวนครั้งที่ผิดปกติก่อนแจ้งเตือน
        self.alert_callbacks = []
        
    def register_alert_callback(self, callback):
        """ลงทะเบียนฟังก์ชันสำหรับการแจ้งเตือน"""
        self.alert_callbacks.append(callback)
    
    def _get_redis_key(self, api_key: str, endpoint: str) -> str:
        """สร้าง Redis key สำหรับการติดตาม"""
        key_hash = hashlib.md5(api_key.encode()).hexdigest()[:8]
        return f"api_audit:{key_hash}:{endpoint}"
    
    def _record_request(
        self,
        api_key: str,
        endpoint: str,
        ip_address: str,
        timestamp: Optional[float] = None
    ) -> Dict:
        """บันทึกคำขอและคืนค่าสถานะปัจจุบัน"""
        if timestamp is None:
            timestamp = time.time()
        
        redis_key = self._get_redis_key(api_key, endpoint)
        current_window = int(timestamp // self.window_size) * self.window_size
        
        # เพิ่มคำขอใน sorted set
        request_id = f"{current_window}:{timestamp}:{ip_address}"
        self.redis_client.zadd(
            redis_key,
            {request_id: timestamp}
        )
        
        # ลบคำขอเก่าที่เกินหน้าต่างเวลา
        cutoff_time = timestamp - (self.window_size * 2)
        self.redis_client.zremrangebyscore(redis_key, '-inf', cutoff_time)
        
        # เก็บข้อมูล IP
        ip_key = f"{redis_key}:ips"
        self.redis_client.hincrby(ip_key, ip_address, 1)
        self.redis_client.expire(ip_key, self.window_size * 3)
        
        return {
            'timestamp': timestamp,
            'window': current_window,
            'ip': ip_address
        }
    
    def check_rate_limit(
        self,
        api_key: str,
        endpoint: str,
        ip_address: str
    ) -> Tuple[bool, Dict]:
        """ตรวจสอบอัตราการจำกัดคำขอ"""
        current_time = time.time()
        redis_key = self._get_redis_key(api_key, endpoint)
        current_window = int(current_time // self.window_size) * self.window_size
        
        # นับคำขอในหน้าต่างปัจจุบัน
        request_count = self.redis_client.zcount(
            redis_key,
            current_window,
            current_window + self.window_size
        )
        
        is_allowed = request_count < self.rate_limit
        remaining = max(0, self.rate_limit - request_count)
        
        return is_allowed, {
            'current_count': request_count,
            'limit': self.rate_limit,
            'remaining': remaining,
            'reset_in': self.window_size - (current_time - current_window),
            'is_allowed': is_allowed
        }
    
    def detect_anomalies(
        self,
        api_key: str,
        endpoint: str,
        ip_address: str,
        payload_size: int = 0
    ) -> List[Dict]:
        """ตรวจจับความผิดปกติในรูปแบบการใช้งาน"""
        anomalies = []
        redis_key = self._get_redis_key(api_key, endpoint)
        current_time = time.time()
        
        # ตรวจสอบการเปลี่ยนแปลง IP อย่างรุนแรง
        ip_key = f"{redis_key}:ips"
        ip_distribution = self.redis_client.hgetall(ip_key)
        
        if len(ip_distribution) > 5:
            anomalies.append({
                'type': 'MULTIPLE_IPS',
                'severity': 'HIGH',
                'message': f'พบ {len(ip_distribution)} IP ที่แตกต่างกันในช่วงเวลาสั้น',
                'details': ip_distribution
            })
        
        # ตรวจสอบความถี่ผิดปกติ
        recent_requests = self.redis_client.zcount(
            redis_key,
            current_time - 300,
            current_time
        )
        
        if recent_requests > self.rate_limit * 4:
            anomalies.append({
                'type': 'HIGH_FREQUENCY',
                'severity': 'CRITICAL',
                'message': f'ความถี่การใช้งานสูงผิดปกติ: {recent_requests} คำขอใน 5 นาที',
                'details': {'requests': recent_requests, 'threshold': self.rate_limit * 4}
            })
        
        # ตรวจสอบขนาด payload
        if payload_size > 1000000:  # เกิน 1MB
            anomalies.append({
                'type': 'LARGE_PAYLOAD',
                'severity': 'MEDIUM',
                'message': f'ขนาด Payload ผิดปกติ: {payload_size} bytes',
                'details': {'size': payload_size}
            })
        
        # ตรวจสอบการใช้งานในช่วงเวลาผิดปกติ
        hour = datetime.fromtimestamp(current_time).hour
        if hour < 6 and recent_requests > 10:
            anomalies.append({
                'type': 'OFF_HOURS_USAGE',
                'severity': 'MEDIUM',
                'message': f'การใช้งานในช่วงเวลาผิดปกติ: {hour}:00 น.',
                'details': {'hour': hour, 'requests': recent_requests}
            })
        
        return anomalies

การสร้างระบบแจ้งเตือนแบบเรียลไทม์

หลังจากมีระบบตรวจจับความผิดปกติแล้ว ขั้นตอนถัดไปคือการสร้างระบบแจ้งเตือนที่สามารถส่งข้อความไปยังช่องทางต่างๆ ได้ทันที ซึ่งรวมถึงการแจ้งเตือนผ่านอีเมล การส่งข้อความไปยัง Discord Webhook และการบันทึกเหตุการณ์ลงในไฟล์ล็อก ระบบจะใช้โครงสร้าง Strategy Pattern เพื่อให้สามารถเพิ่มช่องทางการแจ้งเตือนใหม่ได้โดยไม่ต้องแก้ไขโค้ดหลัก 
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import aiohttp
import asyncio


@dataclass
class Alert:
    """โครงสร้างข้อมูลการแจ้งเตือน"""
    severity: str  # LOW, MEDIUM, HIGH, CRITICAL
    alert_type: str
    message: str
    api_key_prefix: str  # แสดงเฉพาะ 8 ตัวอักษรแรกของ API Key
    ip_address: str
    timestamp: float
    details: Dict[str, Any]
    
    def to_dict(self) -> Dict:
        return {
            'severity': self.severity,
            'alert_type': self.alert_type,
            'message': self.message,
            'api_key': f"{self.api_key_prefix}...",
            'ip': self.ip_address,
            'timestamp': datetime.fromtimestamp(self.timestamp).isoformat(),
            'details': self.details
        }


class AlertStrategy(ABC):
    """Abstract base class สำหรับกลยุทธ์การแจ้งเตือน"""
    
    @abstractmethod
    async def send(self, alert: Alert) -> bool:
        pass
    
    @abstractmethod
    def get_channel_name(self) -> str:
        pass


class EmailAlertStrategy(AlertStrategy):
    """การแจ้งเตือนผ่านอีเมล"""
    
    def __init__(
        self,
        smtp_host: str,
        smtp_port: int,
        sender_email: str,
        sender_password: str,
        recipient_emails: List[str]
    ):
        self.smtp_host = smtp_host
        self.smtp_port = smtp_port
        self.sender_email = sender_email
        self.sender_password = sender_password
        self.recipient_emails = recipient_emails
    
    async def send(self, alert: Alert) -> bool:
        try:
            msg = MIMEMultipart('alternative')
            msg['Subject'] = f"[{alert.severity}] API Security Alert - {alert.alert_type}"
            msg['From'] = self.sender_email
            msg['To'] = ', '.join(self.recipient_emails)
            
            html_content = f"""
            <html>
            <body style="font-family: Arial, sans-serif;">
                <h2 style="color: {'red' if alert.severity == 'CRITICAL' else 'orange'};">
                    การแจ้งเตือนความปลอดภัย API
                </h2>
                <table style="border-collapse: collapse; width: 100%;">
                    <tr>
                        <td style="padding: 8px; border: 1px solid #ddd; font-weight: bold;">ระดับความรุนแรง</td>
                        <td style="padding: 8px; border: 1px solid #ddd;">{alert.severity}</td>
                    </tr>
                    <tr>
                        <td style="padding: 8px; border: 1px solid #ddd; font-weight: bold;">ประเภทการแจ้งเตือน</td>
                        <td style="padding: 8px; border: 1px solid #ddd;">{alert.alert_type}</td>
                    </tr>
                    <tr>
                        <td style="padding: 8px; border: 1px solid #ddd; font-weight: bold;">ข้อความ</td>
                        <td style="padding: 8px; border: 1px solid #ddd;">{alert.message}</td>
                    </tr>
                    <tr>
                        <td style="padding: 8px; border: 1px solid #ddd; font-weight: bold;">IP Address</td>
                        <td style="padding: 8px; border: 1px solid #ddd;">{alert.ip_address}</td>
                    </tr>
                    <tr>
                        <td style="padding: 8px; border: 1px solid #ddd; font-weight: bold;">เวลา</td>
                        <td style="padding: 8px; border: 1px solid #ddd;">{datetime.fromtimestamp(alert.timestamp)}</td>
                    </tr>
                </table>
                <p>รายละเอียดเพิ่มเติม:</p>
                <pre>{json.dumps(alert.details, indent=2)}</pre>
            </body>
            </html>
            """
            
            msg.attach(MIMEText(html_content, 'html'))
            
            with smtplib.SMTP(self.smtp_host, self.smtp_port) as server:
                server.starttls()
                server.login(self.sender_email, self.sender_password)
                server.send_message(msg)
            
            logger.info(f"Email alert sent successfully for {alert.alert_type}")
            return True
            
        except Exception as e:
            logger.error(f"Failed to send email alert: {e}")
            return False
    
    def get_channel_name(self) -> str:
        return "Email"


class DiscordWebhookStrategy(AlertStrategy):
    """การแจ้งเตือนผ่าน Discord Webhook"""
    
    def __init__(self, webhook_url: str, mention_role_id: Optional[str] = None):
        self.webhook_url = webhook_url
        self.mention_role_id = mention_role_id
    
    async def send(self, alert: Alert) -> bool:
        try:
            color_map = {
                'LOW': 0x00FF00,
                'MEDIUM': 0xFFA500,
                'HIGH': 0xFF4500,
                'CRITICAL': 0xFF0000
            }
            
            payload = {
                "embeds": [{
                    "title": f"🚨 {alert.alert_type}",
                    "description": alert.message,
                    "color": color_map.get(alert.severity, 0xFF0000),
                    "fields": [
                        {
                            "name": "🔔 ระดับความรุนแรง",
                            "value": alert.severity,
                            "inline": True
                        },
                        {
                            "name": "🌐 IP Address",
                            "value": alert.ip_address,
                            "inline": True
                        },
                        {
                            "name": "🔑 API Key",
                            "value": f"``{alert.api_key_prefix}...``",
                            "inline": False
                        }
                    ],
                    "timestamp": datetime.fromtimestamp(alert.timestamp).isoformat(),
                    "footer": {
                        "text": "API Security Audit System"
                    }
                }]
            }
            
            if self.mention_role_id:
                payload["content"] = f"<@>{self.mention_role_id}"
            
            async with aiohttp.ClientSession() as session:
                await session.post(self.webhook_url, json=payload)
            
            logger.info(f"Discord alert sent successfully for {alert.alert_type}")
            return True
            
        except Exception as e:
            logger.error(f"Failed to send Discord alert: {e}")
            return False
    
    def get_channel_name(self) -> str:
        return "Discord"


class AlertManager:
    """ตัวจัดการการแจ้งเตือนหลายช่องทาง"""
    
    def __init__(self):
        self.strategies: List[AlertStrategy] = []
        self.alert_history: List[Alert] = []
        self.max_history = 1000
        self.cooldown_period = 300  # 5 นาที ก่อนส่งการแจ้งเตือนซ้ำ
    
    def add_strategy(self, strategy: AlertStrategy):
        """เพิ่มช่องทางการแจ้งเตือน"""
        self.strategies.append(strategy)
        logger.info(f"Added alert strategy: {strategy.get_channel_name()}")
    
    async def send_alert(self, alert: Alert) -> Dict[str, bool]:
        """ส่งการแจ้งเตือนไปยังทุกช่องทาง"""
        # ตรวจสอบ cooldown
        recent_similar = [
            a for a in self.alert_history
            if a.alert_type == alert.alert_type
            and alert.timestamp - a.timestamp < self.cooldown_period
        ]
        
        if recent_similar:
            logger.info(f"Alert {alert.alert_type} suppressed due to cooldown")
            return {strategy.get_channel_name(): False for strategy in self.strategies}
        
        # ส่งการแจ้งเตือนไปยังทุกช่องทาง
        results = {}
        tasks = [strategy.send(alert) for strategy in self.strategies]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        for strategy, response in zip(self.strategies, responses):
            channel_name = strategy.get_channel_name()
            if isinstance(response, Exception):
                logger.error(f"Failed to send to {channel_name}: {response}")
                results[channel_name] = False
            else:
                results[channel_name] = response
        
        # บันทึกประวัติ
        self.alert_history.append(alert)
        if len(self.alert_history) > self.max_history:
            self.alert_history = self.alert_history[-self.max_history:]
        
        return results

การผสานรวมกับ HolySheep AI API

ในการใช้งานจริง เราจะผสานรวมระบบตรวจสอบความปลอดภัยเข้ากับการเรียกใช้ API ของ HolySheep AI ซึ่งเป็นแพลตฟอร์ม AI API ที่มีราคาประหยัดกว่า 85% เมื่อเทียบกับบริการอื่น โดยมีความเร็วในการตอบสนองน้อยกว่า 50 มิลลิวินาที และรองรับการชำระเงินผ่าน WeChat และ Alipay สำหรับนักพัฒนาที่ต้องการทดลองใช้งาน สามารถสมัครที่นี่เพื่อรับเครดิตฟรีเมื่อลงทะเบียน 
import httpx
from typing import Optional, Dict, Any, List
import json


class HolySheepAPIClient:
    """Client สำหรับ HolySheep AI API พร้อมระบบตรวจสอบความปลอดภัย"""
    
    def __init__(
        self,
        api_key: str,
        audit_system: Optional[APIAudit] = None,
        alert_manager: Optional[AlertManager] = None
    ):
        self.base_url = "https://api.holysheep.ai/v1"
        self.api_key = api_key
        self.audit_system = audit_system
        self.alert_manager = alert_manager
        self.client = httpx.AsyncClient(timeout=60.0)
        
        # เก็บเฉพาะ 8 ตัวอักษรแรกของ API Key สำหรับการแจ้งเตือน
        self.api_key_prefix = api_key[:8] if len(api_key) >= 8 else api_key
    
    def _get_client_ip(self, request_ip: Optional[str] = None) -> str:
        """ดึง IP Address ของไคลเอนต์"""
        return request_ip or "unknown"
    
    async def chat_completions(
        self,
        messages: List[Dict[str, str]],
        model: str = "gpt-4.1",
        temperature: float = 0.7,
        max_tokens: int = 1000,
        request_ip: Optional[str] = None
    ) -> Dict[str, Any]:
        """เรียกใช้ Chat Completions API พร้อมตรวจสอบความปลอดภัย"""
        
        ip_address = self._get_client_ip(request_ip)
        endpoint = "chat/completions"
        
        # ตรวจสอบความปลอดภัยก่อนเรียกใช้ API
        if self.audit_system:
            # บันทึกคำขอ
            self.audit_system._record_request(
                self.api_key, endpoint, ip_address
            )
            
            # ตรวจสอบ Rate Limit
            is_allowed, rate_info = self.audit_system.check_rate_limit(
                self.api_key, endpoint, ip_address
            )
            
            if not is_allowed:
                # ส่งการแจ้งเตือนเมื่อเกิน Rate Limit
                if self.alert_manager:
                    alert = Alert(
                        severity='HIGH',
                        alert_type='RATE_LIMIT_EXCEEDED',
                        message=f'API Key {self.api_key_prefix}... เกินขีดจำกัดการใช้งาน',
                        api_key_prefix=self.api_key_prefix,
                        ip_address=ip_address,
                        timestamp=time.time(),
                        details=rate_info
                    )
                    await self.alert_manager.send_alert(alert)
                
                return {
                    'error': True,
                    'code': 'RATE_LIMIT_EXCEEDED',
                    'message': f'เกินขีดจำกัด {rate_info["limit"]} คำขอต่อนาที',
                    'retry_after': rate_info['reset_in']
                }
            
            # ตรวจจับความผิดปกติ
            payload_size = len(json.dumps(messages).encode('utf-8'))
            anomalies = self.audit_system.detect_anomalies(
                self.api_key, endpoint, ip_address, payload_size
            )
            
            for anomaly in anomalies:
                if self.alert_manager:
                    alert = Alert(
                        severity=anomaly['severity'],
                        alert_type=anomaly['type'],
                        message=anomaly['message'],
                        api_key_prefix=self.api_key_prefix,
                        ip_address=ip_address,
                        timestamp=time.time(),
                        details=anomaly['details']
                    )
                    await self.alert_manager.send_alert(alert)
        
        # เรียกใช้ HolySheep AI API
        headers = {
            'Authorization': f'Bearer {self.api_key}',
            'Content-Type': 'application/json'
        }
        
        payload = {
            'model': model,
            'messages': messages,
            'temperature': temperature,
            'max_tokens': max_tokens
        }
        
        try:
            response = await self.client.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload
            )
            
            if response.status_code == 401:
                if self.alert_manager:
                    alert = Alert(
                        severity='CRITICAL',
                        alert_type='UNAUTHORIZED_ACCESS',
                        message=f'ความพยายามเข้าถึง API ด้วย Key ที่ไม่ถูกต้อง',
                        api_key_prefix=self.api_key_prefix,
                        ip_address=ip_address,
                        timestamp=

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง