บทนำ
ในระบบ microservices สมัยใหม่ API Gateway ทำหน้าที่เป็นจุดเชื่อมต่อกลางที่รับ request ทั้งหมดเข้ามา การวิเคราะห์ log จาก API Gateway อย่างมีประสิทธิภาพจึงเป็นสิ่งจำเป็นอย่างยิ่งสำหรับการรักษาความปลอดภัยและประสิทธิภาพการทำงาน บทความนี้จะอธิบายเทคนิคขั้นสูงในการวิเคราะห์ log เพื่อระบุ abnormal traffic พร้อมโค้ด production-ready ที่สามารถนำไปใช้งานได้จริง
สถาปัตยกรรมระบบ Log Analysis
ระบบ API Gateway Log Analysis ที่ดีควรประกอบด้วยองค์ประกอบหลักดังนี้ ส่วนแรกคือ Log Collector ที่ทำหน้าที่รวบรวม log จาก API Gateway หลายตัว ส่วนที่สองคือ Log Parser ที่ parse ข้อมูลดิบให้เป็น structured data ส่วนที่สามคือ Anomaly Detector ที่วิเคราะห์ patterns เพื่อระบุความผิดปกติ และส่วนสุดท้ายคือ Alert Manager ที่จัดการส่งการแจ้งเตือนเมื่อพบสิ่งผิดปกติ
การออกแบบ Log Schema
ก่อนที่จะเริ่มวิเคราะห์ เราต้องกำหนดโครงสร้างของ log ให้เป็นมาตรฐานเดียวกัน โดย log แต่ละรายการควรประกอบด้วยฟิลด์ที่จำเป็นสำหรับการวิเคราะห์ ซึ่งรวมถึง timestamp ที่บันทึกเวลาละเอียดถึงมิลลิวินาที, client_ip ที่ระบุที่อยู่ IP ของผู้ใช้งาน, request_path ที่เก็บ path ของ API ที่ถูกเรียก, http_method ที่ระบุ GET POST PUT DELETE และอื่นๆ, status_code ที่เก็บ HTTP status ของ response, response_time_ms ที่วัดเวลาตอบสนองเป็นมิลิวินาที, request_size_bytes ที่วัดขนาดของ request, user_agent ที่เก็บข้อมูล browser หรือ client, และ trace_id สำหรับการ track request ผ่านระบบ
import json
import re
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
@dataclass
class APILogEntry:
"""Structured log entry สำหรับ API Gateway"""
timestamp: datetime
client_ip: str
request_path: str
http_method: str
status_code: int
response_time_ms: float
request_size_bytes: int
user_agent: str
trace_id: str
service_name: str
api_key_id: Optional[str] = None
error_message: Optional[str] = None
class APILogParser:
"""Parser สำหรับ API Gateway log ในรูปแบบ JSON"""
# Pattern สำหรับ Apache/Nginx combined log format
LOG_PATTERN = re.compile(
r'(?P<ip>[\d\.]+) - (?P<user>[\w\-]+) '
r'\[(?P<timestamp>[^\]]+)\] '
r'"(?P<method>\w+) (?P<path>[^\s]+)[^"]*" '
r'(?P<status>\d+) (?P<size>\d+) '
r'"(?P<referer>[^"]*)" "(?P<ua>[^"]*)"'
)
# JSON field mapping
JSON_FIELDS = {
'timestamp': ['timestamp', 'time', '@timestamp', 'ts'],
'client_ip': ['client_ip', 'remote_addr', 'ip', 'clientIp'],
'request_path': ['request_path', 'path', 'uri', 'requestUri'],
'http_method': ['http_method', 'method', 'verb', 'httpMethod'],
'status_code': ['status_code', 'status', 'statusCode'],
'response_time_ms': ['response_time_ms', 'response_time', 'latency', 'duration'],
'request_size_bytes': ['request_size_bytes', 'request_size', 'size'],
'user_agent': ['user_agent', 'user_agent', 'ua'],
'trace_id': ['trace_id', 'traceId', 'request_id', 'requestId'],
'service_name': ['service_name', 'service', 'app'],
}
def parse_json_line(self, line: str) -> Optional[APILogEntry]:
"""Parse log line ในรูปแบบ JSON"""
try:
data = json.loads(line)
return self._extract_fields(data)
except json.JSONDecodeError:
return None
def _extract_fields(self, data: dict) -> APILogEntry:
"""Extract fields จาก dict โดยใช้ field mapping"""
def find_value(key_list):
for key in key_list:
if key in data:
return data[key]
return None
timestamp_str = find_value(self.JSON_FIELDS['timestamp'])
if isinstance(timestamp_str, str):
timestamp = datetime.fromisoformat(timestamp_str.replace('Z', '+00:00'))
else:
timestamp = datetime.now()
return APILogEntry(
timestamp=timestamp,
client_ip=str(find_value(self.JSON_FIELDS['client_ip']) or '0.0.0.0'),
request_path=str(find_value(self.JSON_FIELDS['request_path']) or '/'),
http_method=str(find_value(self.JSON_FIELDS['http_method']) or 'GET'),
status_code=int(find_value(self.JSON_FIELDS['status_code']) or 0),
response_time_ms=float(find_value(self.JSON_FIELDS['response_time_ms']) or 0),
request_size_bytes=int(find_value(self.JSON_FIELDS['request_size_bytes']) or 0),
user_agent=str(find_value(self.JSON_FIELDS['user_agent']) or ''),
trace_id=str(find_value(self.JSON_FIELDS['trace_id']) or ''),
service_name=str(find_value(self.JSON_FIELDS['service_name']) or 'unknown'),
)
ตัวอย่างการใช้งาน
sample_log = json.dumps({
"timestamp": "2024-01-15T10:30:45.123Z",
"client_ip": "192.168.1.100",
"request_path": "/api/v1/users",
"method": "GET",
"status": 200,
"response_time": 45.5,
"traceId": "abc123xyz"
})
parser = APILogParser()
entry = parser.parse_json_line(sample_log)
print(f"Parsed: {entry.request_path} from {entry.client_ip} in {entry.response_time_ms}ms")อัลกอริทึมการตรวจจับ Anomaly
การตรวจจับการรับส่งข้อมูลผิดปกติต้องใช้หลายเทคนิคร่วมกัน โดยแต่ละเทคนิคจะจับ pattern ที่แตกต่างกัน เทคนิคแรกคือ Rate-based Detection ที่ตรวจจับ request rate ที่ผิดปกติ เช่น brute force attack หรือ scraping เทคนิคที่สองคือ Statistical Anomaly Detection ที่ใช้ statistical methods เช่น Z-score หรือ IQR เพื่อหา outliers ใน response time หรือ request size เทคนิคที่สามคือ Pattern-based Detection ที่ตรวจจับ known attack patterns เช่น SQL injection, path traversal หรือ XSS และเทคนิคสุดท้ายคือ ML-based Detection ที่ใช้ machine learning เพื่อจำแนก traffic ที่น่าสงสัย
from collections import defaultdict, deque
from dataclasses import dataclass
from enum import Enum
from typing import Dict, List, Tuple
import math
class ThreatLevel(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class AnomalyResult:
threat_level: ThreatLevel
anomaly_type: str
description: str
evidence: dict
recommended_action: str
class AnomalyDetector:
"""ระบบตรวจจับ anomaly แบบ multi-layer"""
def __init__(
self,
rate_threshold: int = 100, # requests per minute
error_rate_threshold: float = 0.1, # 10%
p99_latency_threshold_ms: float = 500,
sliding_window_size: int = 60 # seconds
):
self.rate_threshold = rate_threshold
self.error_rate_threshold = error_rate_threshold
self.p99_latency_threshold = p99_latency_threshold_ms
# In-memory storage สำหรับ real-time analysis
self.ip_request_times: Dict[str, deque] = defaultdict(lambda: deque(maxlen=1000))
self.ip_errors: Dict[str, deque] = defaultdict(lambda: deque(maxlen=100))
self.ip_latencies: Dict[str, deque] = defaultdict(lambda: deque(maxlen=500))
self.endpoint_stats: Dict[str, dict] = defaultdict(lambda: {
'total': 0, 'errors': 0, 'latencies': deque(maxlen=1000)
})
# Known attack patterns
self.attack_patterns = [
(r'(\bunion\b.*\bselect\b|\bor\b1\s*=\s*1)', 'SQL_INJECTION'),
(r'(\.\./|\.\.\\|%2e%2e)', 'PATH_TRAVERSAL'),
(r'( List[AnomalyResult]:
"""วิเคราะห์ log entry เดียว"""
results = []
# Layer 1: Rate-based detection
rate_result = self._check_rate_limit(log_entry)
if rate_result:
results.append(rate_result)
# Layer 2: Error rate detection
error_result = self._check_error_rate(log_entry)
if error_result:
results.append(error_result)
# Layer 3: Latency anomaly detection
latency_result = self._check_latency_anomaly(log_entry)
if latency_result:
results.append(latency_result)
# Layer 4: Pattern-based detection
pattern_result = self._check_attack_patterns(log_entry)
if pattern_result:
results.append(pattern_result)
# Layer 5: Geo/IP reputation (simplified)
geo_result = self._check_suspicious_ip(log_entry)
if geo_result:
results.append(geo_result)
return results
def _check_rate_limit(self, entry: APILogEntry) -> Optional[AnomalyResult]:
"""ตรวจจับ request rate ที่สูงผิดปกติ"""
self.ip_request_times[entry.client_ip].append(entry.timestamp)
# นับ requests ในช่วง 1 นาทีที่ผ่านมา
current_time = entry.timestamp
recent_requests = sum(
1 for t in self.ip_request_times[entry.client_ip]
if (current_time - t).total_seconds() < 60
)
if recent_requests > self.rate_threshold:
return AnomalyResult(
threat_level=ThreatLevel.HIGH,
anomaly_type="HIGH_REQUEST_RATE",
description=f"IP {entry.client_ip} มี request rate {recent_requests} ครั้ง/นาที",
evidence={'recent_requests': recent_requests, 'threshold': self.rate_threshold},
recommended_action="block_ip"
)
return None
def _check_error_rate(self, entry: APILogEntry) -> Optional[AnomalyResult]:
"""ตรวจจับ error rate ที่สูง"""
self.ip_errors[entry.client_ip].append(entry.status_code >= 400)
total = len(self.ip_errors[entry.client_ip])
errors = sum(1 for e in self.ip_errors[entry.client_ip] if e)
error_rate = errors / total if total > 0 else 0
if total >= 20 and error_rate > self.error_rate_threshold:
return AnomalyResult(
threat_level=ThreatLevel.MEDIUM,
anomaly_type="HIGH_ERROR_RATE",
description=f"IP {entry.client_ip} มี error rate {error_rate:.1%}",
evidence={'error_rate': error_rate, 'total_requests': total},
recommended_action="verify_human"
)
return None
def _check_latency_anomaly(self, entry: APILogEntry) -> Optional[AnomalyResult]:
"""ตรวจจับ latency ที่ผิดปกติใช้ IQR method"""
self.ip_latencies[entry.client_ip].append(entry.response_time_ms)
latencies = list(self.ip_latencies[entry.client_ip])
if len(latencies) < 20:
return None
latencies.sort()
q1_idx = len(latencies) // 4
q3_idx = 3 * len(latencies) // 4
q1, q3 = latencies[q1_idx], latencies[q3_idx]
iqr = q3 - q1
upper_bound = q3 + 1.5 * iqr
if entry.response_time_ms > upper_bound and entry.response_time_ms > self.p99_latency_threshold:
return AnomalyResult(
threat_level=ThreatLevel.LOW,
anomaly_type="LATENCY_ANOMALY",
description=f"Response time {entry.response_time_ms:.1f}ms สูงผิดปกติ",
evidence={'latency': entry.response_time_ms, 'upper_bound': upper_bound},
recommended_action="investigate"
)
return None
def _check_attack_patterns(self, entry: APILogEntry) -> Optional[AnomalyResult]:
"""ตรวจจับ known attack patterns"""
combined = f"{entry.request_path} {entry.user_agent}"
for pattern, attack_type in self.attack_patterns:
import re
if re.search(pattern, combined, re.IGNORECASE):
return AnomalyResult(
threat_level=ThreatLevel.CRITICAL,
anomaly_type=attack_type,
description=f"ตรวจพบ {attack_type} pattern",
evidence={'matched_pattern': pattern, 'path': entry.request_path},
recommended_action="block_immediately"
)
return None
def _check_suspicious_ip(self, entry: APILogEntry) -> Optional[AnomalyResult]:
"""ตรวจจับ IP ที่น่าสงสัย (simplified)"""
# ตรวจจับ private IP ที่ปลอมตัว
if entry.client_ip.startswith(('10.', '172.16.', '192.168.')):
# ในบางกรณีอาจเป็น internal attack
pass
# ตรวจจับ IP ที่เข้าถึง endpoints ที่ไม่มีอยู่จำนวนมาก
if entry.status_code == 404:
self.endpoint_stats[entry.request_path]['not_found_count'] = \
self.endpoint_stats[entry.request_path].get('not_found_count', 0) + 1
if self.endpoint_stats[entry.request_path]['not_found_count'] > 10:
return AnomalyResult(
threat_level=ThreatLevel.MEDIUM,
anomaly_type="PROBING_ATTACK",
description=f"IP {entry.client_ip} กำลัง probe endpoints",
evidence={'path': entry.request_path, '404_count': self.endpoint_stats[entry.request_path]['not_found_count']},
recommended_action="rate_limit"
)
return None
ทดสอบการทำงาน
detector = AnomalyDetector(rate_threshold=10) # threshold ต่ำสำหรับทดสอบ
สร้าง log entries ทดสอบ
test_logs = [
APILogEntry(
timestamp=datetime.now(),
client_ip="192.168.1.100",
request_path="/api/users",
http_method="GET",
status_code=200,
response_time_ms=50.0,
request_size_bytes=100,
user_agent="Mozilla/5.0",
trace_id="test123",
service_name="user-service"
) for _ in range(15)
]
ทดสอบ rate limit detection
print("=== Rate Limit Detection Test ===")
for log in test_logs:
results = detector.analyze(log)
if results:
for r in results:
print(f"[{r.threat_level.value.upper()}] {r.description}") การใช้ AI สำหรับการวิเคราะห์ Log ขั้นสูง
สำหรับการวิเคราะห์ที่ซับซ้อนมากขึ้น เราสามารถใช้ AI API จาก HolySheep AI เพื่อจำแนก threats และสร้าง insights จาก log data ซึ่งมีประสิทธิภาพสูงและราคาประหยัดกว่าบริการอื่นถึง 85% โดยรองรับ WeChat และ Alipay สำหรับการชำระเงิน
import aiohttp
import asyncio
from typing import List, Dict, Any
class HolySheepLogAnalyzer:
"""ใช้ HolySheep AI API สำหรับ advanced log analysis"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=aiohttp.ClientTimeout(total=30)
)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def analyze_logs_batch(
self,
logs: List[Dict[str, Any]],
analysis_type: str = "security"
) -> Dict[str, Any]:
"""
วิเคราะห์ batch ของ logs โดยใช้ AI
analysis_type: 'security', 'performance', 'business'
"""
prompt = self._build_analysis_prompt(logs, analysis_type)
async with self.session.post(
f"{self.BASE_URL}/chat/completions",
json={
"model": "gpt-4.1", # $8/MTok - ประหยัดที่สุดสำหรับ log analysis
"messages": [
{
"role": "system",
"content": """คุณเป็นผู้เชี่ยวชาญด้าน API Security Analysis
วิเคราะห์ log entries และระบุ:
1. potential threats หรือ attacks
2. พฤติกรรมผิดปกติของ users
3. ปัญหาด้าน performance
4. คำแนะนำสำหรับการป้องกัน
ตอบกลับเป็น JSON format ที่มีโครงสร้างชัดเจน"""
},
{
"role": "user",
"content": prompt
}
],
"temperature": 0.1, # Low temperature สำหรับ analysis ที่แม่นยำ
"max_tokens": 2000
}
) as response:
if response.status != 200:
error_text = await response.text()
raise Exception(f"API Error: {response.status} - {error_text}")
result = await response.json()
return self._parse_ai_response(result)
def _build_analysis_prompt(
self,
logs: List[Dict[str, Any]],
analysis_type: str
) -> str:
"""สร้าง prompt สำหรับ AI analysis"""
# จำกัดจำนวน logs เพื่อไม่ให้ token สูงเกินไป
recent_logs = logs[-50:] if len(logs) > 50 else logs
logs_json = "\n".join([
json.dumps(log, ensure_ascii=False) for log in recent_logs
])
prompts = {
"security": f"""วิเคราะห์ API Gateway logs ต่อไปนี้เพื่อหา:
- Brute force attempts
- SQL Injection หรือ XSS attempts
- Suspicious IP addresses
- Unusual access patterns
Logs:
{logs_json}
ตอบเป็น JSON พร้อม fields: threats[], suspicious_ips[], recommendations[]""",
"performance": f"""วิเคราะห์ API Gateway logs ต่อไปนี้เพื่อหา:
- Slow endpoints
- High latency patterns
- Error hotspots
- Capacity issues
Logs:
{logs_json}
ตอบเป็น JSON พร้อม fields: slow_endpoints[], latency_issues[], recommendations[]""",
"business": f"""วิเคราะห์ API Gateway logs ต่อไปนี้เพื่อหา:
- Popular endpoints
- User behavior patterns
- Peak usage times
- Business insights
Logs:
{logs_json}
ตอบเป็น JSON พร้อม fields: insights[], metrics{}, recommendations[]"""
}
return prompts.get(analysis_type, prompts["security"])
def _parse_ai_response(self, response: Dict) -> Dict[str, Any]:
"""Parse AI response เป็น structured data"""
try:
content = response['choices'][0]['message']['content']
# ลอง extract JSON จาก response
import re
json_match = re.search(r'``json\s*(.*?)\s*``', content, re.DOTALL)
if json_match:
return json.loads(json_match.group(1))
# ถ้าไม่มี code block ลอง parse ทั้งหมด
return json.loads(content)
except (KeyError, json.JSONDecodeError) as e:
return {
"error": f"Failed to parse AI response: {str(e)}",
"raw_content": response.get('choices', [{}])[0].get('message', {}).get('content', '')
}
async def get_security_summary(self, logs: List[Dict[str, Any]]) -> str:
"""สร้าง security summary สำหรับ dashboard"""
analysis = await self.analyze_logs_batch(logs, "security")
summary_parts = []
if 'threats' in analysis and analysis['threats']:
summary_parts.append(f"พบ {len(analysis['threats'])} threats:")
for threat in analysis['threats'][:5]:
summary_parts.append(f" - {threat}")
if 'suspicious_ips' in analysis and analysis['suspicious_ips']:
summary_parts.append(f"\nIP ที่น