ระบบเฝ้าระวังสถานะความปลอดภัย API สำหรับผู้เริ่มต้น

คุณเคยสงสัยไหมว่าแอปพลิเคชันที่คุณใช้อยู่มีความปลอดภัยเพียงพอหรือไม่ หรือมีใครพยายามเข้าถึงข้อมูลของคุณโดยไม่ได้รับอนุญาตหรือเปล่า บทความนี้จะพาคุณสร้างระบบเฝ้าระวังความปลอดภัย API ตั้งแต่เริ่มต้น แม้คุณไม่เคยมีประสบการณ์ด้านเทคนิคมาก่อนก็สามารถทำตามได้

API คืออะไร และทำไมต้องดูแลความปลอดภัย

API ย่อมาจาก Application Programming Interface เปรียบเสมือนประตูที่เชื่อมต่อระหว่างแอปพลิเคชันต่างๆ เมื่อคุณเปิดแอปธนาคาร แอปจะส่งคำขอไปยังเซิร์ฟเวอร์ผ่าน API เพื่อดึงข้อมูลบัญชีของคุณ ถ้าประตูนี้ไม่มีการล็อก ทุกคนก็สามารถเข้ามาดูข้อมูลส่วนตัวได้

ระบบเฝ้าระวังความปลอดภัยทำหน้าที่เหมือนกล้องวงจรปิดในอาคาร มันจะ:

• ตรวจจับความผิดปกติ เช่น มีคนพยายามเดารหัสผ่านหลายครั้ง
• แจ้งเตือนเมื่อมีการเข้าถึงข้อมูลผิดปกติ
• บันทึกประวัติการใช้งานเพื่อวิเคราะห์ย้อนหลัง
• ช่วยระบุต้นตอปัญหาเมื่อเกิดเหตุการณ์ไม่คาดคิด

เครื่องมือที่ต้องเตรียม

ก่อนเริ่มสร้างระบบ เราต้องเตรียมเครื่องมือดังนี้:

• บัญชี HolySheep AI — ผู้ให้บริการ API ราคาประหยัด อัตราแลกเปลี่ยน ¥1=$1 ประหยัดมากกว่า 85% สมัครที่นี่ รับเครดิตฟรีเมื่อลงทะเบียน รองรับ WeChat และ Alipay ความเร็วตอบสนองน้อยกว่า 50 มิลลิวินาที
• Python 3.8 ขึ้นไป — ภาษาโปรแกรมที่เข้าใจง่าย ดาวน์โหลดได้จาก python.org
• Text Editor — โปรแกรมเขียนโค้ด แนะนำ VS Code ฟรี

การติดตั้งและตั้งค่าเริ่มต้น

เปิด Terminal หรือ Command Prompt แล้วพิมพ์คำสั่งติดตั้งไลบรารีที่จำเป็น:

pip install requests python-dotenv flask flask-cors

สร้างโฟลเดอร์ใหม่ชื่อ api-security-monitor แล้วสร้างไฟล์ .env เพื่อเก็บความลับ:

HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
API_BASE_URL=https://api.holysheep.ai/v1
FLASK_PORT=5000
LOG_LEVEL=INFO

ระบบบันทึกและตรวจจับความผิดปกติ

สร้างไฟล์ security_monitor.py เพื่อเริ่มสร้างระบบเฝ้าระวัง:

import requests
import time
import json
from datetime import datetime
from collections import defaultdict
from dotenv import load_dotenv
import os

load_dotenv()

API_KEY = os.getenv("HOLYSHEEP_API_KEY")
BASE_URL = os.getenv("API_BASE_URL")

class SecurityMonitor:
    def __init__(self):
        self.request_log = []
        self.failed_attempts = defaultdict(list)
        self.suspicious_ips = set()
        self.alert_threshold = 5
        self.time_window = 300

    def log_request(self, endpoint, ip, status, response_time):
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "endpoint": endpoint,
            "ip": ip,
            "status": status,
            "response_time_ms": response_time
        }
        self.request_log.append(log_entry)
        
        if status >= 400:
            self.failed_attempts[ip].append(time.time())
            self.check_failed_attempts(ip)
        
        if response_time > 1000:
            self.flag_suspicious(ip, "high_latency")
    
    def check_failed_attempts(self, ip):
        current_time = time.time()
        recent_failures = [
            t for t in self.failed_attempts[ip]
            if current_time - t < self.time_window
        ]
        self.failed_attempts[ip] = recent_failures
        
        if len(recent_failures) >= self.alert_threshold:
            self.suspicious_ips.add(ip)
            self.send_alert(ip, f"พบความพยายามเข้าถึงที่ล้มเหลว {len(recent_failures)} ครั้ง")
    
    def flag_suspicious(self, ip, reason):
        print(f"[{datetime.now().strftime('%H:%M:%S')}] ⚠️ พบพฤติกรรมน่าสงสัย: {ip} - {reason}")
    
    def send_alert(self, ip, message):
        print(f"[{datetime.now().strftime('%H:%M:%S')}] 🚨 แจ้งเตือน: {message}")
        self.analyze_with_ai(message)
    
    def analyze_with_ai(self, alert_message):
        headers = {
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": "gpt-4.1",
            "messages": [
                {
                    "role": "system",
                    "content": "คุณคือผู้เชี่ยวชาญด้านความปลอดภัย API วิเคราะห์เหตุการณ์ต่อไปนี้และเสนอแนะการแก้ไข"
                },
                {
                    "role": "user", 
                    "content": f"วิเคราะห์ข้อความแจ้งเตือนนี้: {alert_message}"
                }
            ],
            "temperature": 0.3
        }
        
        try:
            response = requests.post(
                f"{BASE_URL}/chat/completions",
                headers=headers,
                json=payload,
                timeout=10
            )
            
            if response.status_code == 200:
                result = response.json()
                ai_analysis = result["choices"][0]["message"]["content"]
                print(f"[AI Analysis] {ai_analysis}")
                return ai_analysis
            else:
                print(f"[Error] ไม่สามารถเชื่อมต่อ AI API: {response.status_code}")
                
        except Exception as e:
            print(f"[Error] {str(e)}")

if __name__ == "__main__":
    monitor = SecurityMonitor()
    
    print("=" * 50)
    print("ระบบเฝ้าระวังความปลอดภัย API เริ่มทำงาน")
    print("=" * 50)
    
    test_scenarios = [
        {"endpoint": "/api/user/login", "ip": "192.168.1.100", "status": 401, "response_time": 250},
        {"endpoint": "/api/user/login", "ip": "192.168.1.100", "status": 401, "response_time": 280},
        {"endpoint": "/api/user/login", "ip": "192.168.1.100", "status": 401, "response_time": 310},
        {"endpoint": "/api/user/login", "ip": "192.168.1.100", "status": 401, "response_time": 350},
        {"endpoint": "/api/user/login", "ip": "192.168.1.100", "status": 401, "response_time": 400},
        {"endpoint": "/api/user/login", "ip": "192.168.1.100", "status": 401, "response_time": 450},
        {"endpoint": "/api/data/search", "ip": "10.0.0.50", "status": 200, "response_time": 1500},
    ]
    
    for scenario in test_scenarios:
        monitor.log_request(
            scenario["endpoint"],
            scenario["ip"],
            scenario["status"],
            scenario["response_time"]
        )
        time.sleep(0.5)
    
    print("\n" + "=" * 50)
    print(f"IP ที่ติดธงน่าสงสัย: {monitor.suspicious_ips}")
    print(f"จำนวนบันทึกทั้งหมด: {len(monitor.request_log)}")
    print("=" * 50)

รันโปรแกรมด้วยคำสั่ง python security_monitor.py คุณจะเห็นผลลัพธ์การตรวจจับความผิดปกติแบบเรียลไทม์

สร้าง Dashboard แสดงสถานะความปลอดภัย

สร้างไฟล์ dashboard.py เพื่อแสดงผลแบบเว็บไซต์:

from flask import Flask, jsonify, request
from flask_cors import CORS
from security_monitor import SecurityMonitor
import threading
import time

app = Flask(__name__)
CORS(app)

monitor = SecurityMonitor()

@app.route("/api/log", methods=["POST"])
def add_log():
    data = request.json
    monitor.log_request(
        endpoint=data.get("endpoint", ""),
        ip=data.get("ip", "0.0.0.0"),
        status=data.get("status", 200),
        response_time=data.get("response_time", 0)
    )
    return jsonify({"success": True, "message": "บันทึกสำเร็จ"})

@app.route("/api/status")
def get_status():
    return jsonify({
        "total_requests": len(monitor.request_log),
        "suspicious_ips": list(monitor.suspicious_ips),
        "failed_attempts": {
            ip: len(times) 
            for ip, times in monitor.failed_attempts.items()
        }
    })

@app.route("/api/logs")
def get_logs():
    limit = request.args.get("limit", 100, type=int)
    return jsonify({
        "logs": monitor.request_log[-limit:],
        "count": len(monitor.request_log)
    })

def background_monitor():
    while True:
        time.sleep(60)
        print(f"[{time.strftime('%H:%M:%S')}] กำลังตรวจสอบสถานะ...")

if __name__ == "__main__":
    thread = threading.Thread(target=background_monitor, daemon=True)
    thread.start()
    
    print("เปิดเว็บ Dashboard ที่ http://localhost:5000")
    app.run(host="0.0.0.0", port=5000, debug=False)

เปิดเบราว์เซอร์ไปที่ http://localhost:5000/api/status เพื่อดูสถานะความปลอดภัยแบบ JSON

การเชื่อมต่อ API วิเคราะห์ภัยคุกความอัจฉริยะ

เราสามารถใช้ AI จาก HolySheep AI ช่วยวิเคราะห์รูปแบบการโจมตีและเสนอวิธีป้องกัน โดยราคาของ GPT-4.1 อยู่ที่ $8 ต่อล้านโทเค็น หรือ Claude Sonnet 4.5 อยู่ที่ $15 ต่อล้านโทเค็น ซึ่งประหยัดมากเมื่อเทียบกับผู้ให้บริการอื่น

import requests

class ThreatAnalyzer:
    def __init__(self, api_key, base_url="https://api.holysheep.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url
    
    def analyze_attack_pattern(self, logs):
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        log_summary = self._summarize_logs(logs)
        
        payload = {
            "model": "gpt-4.1",
            "messages": [
                {
                    "role": "system",
                    "content": """คุณคือผู้เชี่ยวชาญด้านความปลอดภัยไซเบอร์ 
วิเคราะห์บันทึกกิจกรรม API และระบุ:
1. รูปแบบการโจมตีที่อาจเกิดขึ้น
2. IP ที่น่าสงสัย
3. ช่องโหว่ที่ควรแก้ไข
4. ข้อเสนอแนะการป้องกัน"""
                },
                {
                    "role": "user",
                    "content": f"วิเคราะห์บันทึกต่อไปนี้:\n{log_summary}"
                }
            ],
            "temperature": 0.5,
            "max_tokens": 1000
        }
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=headers,
            json=payload
        )
        
        if response.status_code == 200:
            return response.json()["choices"][0]["message"]["content"]
        return f"เกิดข้อผิดพลาด: {response.status_code}"
    
    def _summarize_logs(self, logs):
        summary = []
        for log in logs[-20:]:
            summary.append(
                f"{log.get('timestamp', '')} | "
                f"{log.get('ip', 'unknown')} | "
                f"{log.get('endpoint', '')} | "
                f"Status: {log.get('status', 0)}"
            )
        return "\n".join(summary)
    
    def generate_report(self, analysis):
        report_payload = {
            "model": "gpt-4.1",
            "messages": [
                {
                    "role": "system",
                    "content": "สร้างรายงานความปลอดภัยในรูปแบบที่เข้าใจง่าย มีสรุปปัญหาและขั้นตอนการแก้ไข"
                },
                {
                    "role": "user",
                    "content": f"จากผลวิเคราะห์ต่อไปนี้ สร้างรายงานภาษาไทย:\n{analysis}"
                }
            ]
        }
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=headers,
            json=report_payload
        )
        
        if response.status_code == 200:
            return response.json()["choices"][0]["message"]["content"]
        return None

if __name__ == "__main__":
    analyzer = ThreatAnalyzer(YOUR_HOLYSHEEP_API_KEY)
    
    sample_logs = [
        {"timestamp": "2024-01-15T10:00:00", "ip": "192.168.1.50", "endpoint": "/api/login", "status": 401},
        {"timestamp": "2024-01-15T10:00:05", "ip": "192.168.1.50", "endpoint": "/api/login", "status": 401},
        {"timestamp": "2024-01-15T10:00:10", "ip": "10.0.0.99", "endpoint": "/api/admin/users", "status": 403},
        {"timestamp": "2024-01-15T10:00:15", "ip": "192.168.1.50", "endpoint": "/api/login", "status": 401},
    ]
    
    print("กำลังวิเคราะห์รูปแบบการโจมตี...")
    analysis = analyzer.analyze_attack_pattern(sample_logs)
    print("\nผลวิเคราะห์:")
    print(analysis)
    
    print("\nกำลังสร้างรายงาน...")
    report = analyzer.generate_report(analysis)
    if report:
        print("\nรายงานความปลอดภัย:")
        print(report)

การทดสอบระบบเฝ้าระวัง

สร้างไฟล์ test_monitor.py เพื่อทดสอบการทำงาน:

import unittest
import sys
import os

sys.path.insert(0, os.path.dirname(__file__))

from security_monitor import SecurityMonitor

class TestSecurityMonitor(unittest.TestCase):
    def setUp(self):
        self.monitor = SecurityMonitor()
    
    def test_log_normal_request(self):
        self.monitor.log_request("/api/test", "127.0.0.1", 200, 100)
        self.assertEqual(len(self.monitor.request_log), 1)
        self.assertEqual(self.monitor.request_log[0]["status"], 200)
    
    def test_detect_failed_attempts(self):
        for i in range(6):
            self.monitor.log_request("/api/login", "192.168.1.1", 401, 200)
        
        self.assertIn("192.168.1.1", self.monitor.suspicious_ips)
    
    def test_high_latency_detection(self):
        self.monitor.log_request("/api/slow", "10.0.0.1", 200, 1500)
        self.assertEqual(len(self.monitor.request_log), 1)
    
    def test_multiple_ips_tracking(self):
        self.monitor.log_request("/api/test", "192.168.1.1", 401, 100)
        self.monitor.log_request("/api/test", "192.168.1.2", 401, 100)
        self.monitor.log_request("/api/test", "192.168.1.1", 401, 100)
        
        self.assertEqual(len(self.monitor.failed_attempts["192.168.1.1"]), 2)
        self.assertEqual(len(self.monitor.failed_attempts["192.168.1.2"]), 1)

if __name__ == "__main__":
    print("กำลังทดสอบระบบเฝ้าระวังความปลอดภัย...")
    print("=" * 50)
    unittest.main(verbosity=2)

รันการทดสอบด้วย python -m pytest test_monitor.py -v หรือ python test_monitor.py

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

กรณีที่ 1: ได้รับข้อผิดพลาด 401 Unauthorized

สาเหตุ: API Key ไม่ถูกต้องหรือหมดอายุ

# วิธีแก้ไข: ตรวจสอบ API Key ในไฟล์ .env
แนะนำให้ตรวจสอบว่าไม่มีช่องว่างหรืออักขระพิเศษต่อท้าย

ตรวจสอบว่าคีย์ถูกโหลดอย่างถูกต้อง
from dotenv import load_dotenv
load_dotenv()
print(f"API Key loaded: {os.getenv('HOLYSHEEP_API_KEY')[:10]}...")

ถ้าใช้งานไม่ได้ ให้สร้างคีย์ใหม่ที่ HolySheep AI Dashboard

กรรมที่ 2: Flask แสดงข้อผิดพลาด "Port already in use"

สาเหตุ: มีโปรแกรมอื่นใช้งานพอร์ต 5000 อยู่

# วิธีแก้ไข: เปลี่ยนพอร์ตในการรัน

วิธีที่ 1: เปลี่ยนพอร์ตในโค้ด
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5001, debug=False)

วิธีที่ 2: ใช้ environment variable
app.run(
    host="0.0.0.0", 
    port=int(os.getenv("FLASK_PORT", 5001)), 
    debug=False
)

วิธีที่ 3: หากใช้ Windows ปิดโปรแกรมที่ใช้พอร์ตดังกล่าว
netstat -ano | findstr :5000
taskkill /PID [process_id] /F

กรณีที่ 3: การตอบสนองของ API ช้ามากหรือค้าง

สาเหตุ: เครือข่ายมีปัญหาหรือ API timeout

# วิธีแก้ไข: เพิ่ม timeout และ error handling

def call_api_with_retry(url, headers, payload, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.post(
                url, 
                headers=headers, 
                json=payload, 
                timeout=30
            )
            return response
        except requests.exceptions.Timeout:
            print(f"ครั้งที่ {attempt + 1}: หมดเวลา ลองใหม่...")
            time.sleep(2 ** attempt)
        except requests.exceptions.ConnectionError as e:
            print(f"ไม่สามารถเชื่อมต่อ: {e}")
            break
    
    return None

ใช้งาน
response = call_api_with_retry(
    f"{BASE_URL}/chat/completions",
    headers,
    payload
)

if response:
    print(f"สถานะ: {response.status_code}")
else:
    print("ไม่สามารถเชื่อมต่อ API หลังจากลองหลายครั้ง")

กรณีที่ 4: ไม่สามารถติดตั้งไลบรารีได้

สาเหตุ: pip หรือ Python environment มีปัญหา

# วิธีแก้ไข: อัปเกรด pip และใช้ virtual environment

ขั้นตอนที่ 1: อัปเกรด pip
python -m pip install --upgrade pip

ขั้นตอนที่ 2: สร้าง virtual environment
python -m venv venv

ขั้นตอนที่ 3: เปิดใช้งาน environment
Windows:
venv\Scripts\activate
macOS/Linux:
source venv/bin/activate

ขั้นตอนที่ 4: ติดตั้งไลบรารีอีกครั้ง
pip install requests python-dotenv flask flask-cors

ตรวจสอบการติดตั้ง
pip list

สรุป

ในบทความนี้เราได้เรียนรู้การสร้างระบบเฝ้าระวังความปลอดภัย API ตั้งแต่พื้นฐาน ครอบคลุมการบันทึกกิจกรรม การตรวจจับความผิดปกติ การแจ้งเตือน และการวิเคราะห์ด้วย AI ระบบนี้สามารถขยายเพิ่มเติมได้ตามความต้องการ เช่น การเชื่อมต่อฐานข้อมูล การส่งอีเมลแจ้งเตือน หรือการ