การควบคุมสิทธิ์การเข้าถึง AI API ในระดับ Production — บทเรียนจากโปรเจกต์จริง

ในฐานะ Senior AI Integration Engineer ที่ผ่านระบบหลายสิบโปรเจกต์ตั้งแต่ RAG ขนาดใหญ่ไปจนถึง Chatbot อีคอมเมิร์ซ ผมเจอปัญหาสิทธิ์การเข้าถึง (Permission Control) ซ้ำแล้วซ้ำเล่า — ทั้ง Token Leak, Budget Explosion และ Unauthorized Access จากภายในองค์กรเอง บทความนี้จะสอนวิธีสร้างระบบควบคุมสิทธิ์ AI API ที่ใช้งานได้จริงใน Production

ทำไมต้องควบคุมสิทธิ์อย่างเข้มงวด

จากประสบการณ์ที่ สมัครที่นี่ และใช้งาน HolySheep AI มาเกือบปี พบว่าการควบคุมสิทธิ์ที่ดีช่วยประหยัดค่าใช้จ่ายได้ถึง 40-60% เพราะ HolySheep มีอัตรา ¥1=$1 ซึ่งประหยัดกว่าผู้ให้บริการอื่นถึง 85%+ แต่ถ้าไม่ตั้ง Rate Limit และ Quota ให้เหมาะสม ค่าใช้จ่ายก็พุ่งได้ง่ายมาก

กรณีศึกษาที่ 1: ระบบ AI ลูกค้าสัมพันธ์อีคอมเมิร์ซ — รับมือ Traffic Spike

ร้านค้าออนไลน์แห่งหนึ่งเจอปัญหา Peak Season มีผู้ใช้พร้อมกัน 10,000+ คน แต่ระบบ AI Response หยุดทำงานเพราะไม่มีการจำกัดสิทธิ์ — บางทีมี User ที่ส่ง Request ซ้ำ 200 ครั้ง/วินาที จาก Bot หรือการ Refresh หน้าเกินไป

โครงสร้าง API Key หลายระดับ

สำหรับระบบ Production ที่ต้องการควบคุมสิทธิ์แบบละเอียด ควรแบ่ง API Key ตามหน้าที่การใช้งานดังนี้:

• Master Key — ใช้เฉพาะ Backend Admin สร้าง Key อื่นๆ เท่านั้น
• Service Key — ใช้สำหรับ Service-to-Service Communication
• User Key — ใช้ใน Client Side (มี Rate Limit ต่ำ)
• Read-only Key — สำหรับดึงข้อมูลอย่างเดียว ไม่สามารถสร้าง Resource ใหม่

Middleware ควบคุมสิทธิ์ด้วย Python

import time
from functools import wraps
from typing import Dict, List, Optional
from dataclasses import dataclass
from collections import defaultdict
import hashlib

@dataclass
class RateLimitConfig:
    requests_per_minute: int = 60
    requests_per_day: int = 10000
    max_tokens_per_request: int = 4096
    allowed_models: List[str] = None
    
    def __post_init__(self):
        if self.allowed_models is None:
            self.allowed_models = ["gpt-4.1", "claude-sonnet-4.5"]

class APIKeyManager:
    def __init__(self):
        self.keys: Dict[str, dict] = {}
        self.usage_tracker: Dict[str, list] = defaultdict(list)
        self.api_endpoint = "https://api.holysheep.ai/v1"
        
    def create_key(self, key_name: str, config: RateLimitConfig) -> str:
        """สร้าง API Key ใหม่พร้อมกำหนดสิทธิ์"""
        key_id = hashlib.sha256(
            f"{key_name}_{time.time()}_{config.requests_per_minute}".encode()
        ).hexdigest()[:32]
        
        self.keys[key_id] = {
            "name": key_name,
            "config": config,
            "created_at": time.time(),
            "is_active": True,
            "total_usage": 0
        }
        return key_id
    
    def verify_key(self, api_key: str) -> Optional[RateLimitConfig]:
        """ตรวจสอบความถูกต้องและสิทธิ์ของ API Key"""
        if api_key not in self.keys:
            return None
            
        key_info = self.keys[api_key]
        if not key_info["is_active"]:
            return None
            
        return key_info["config"]
    
    def check_rate_limit(self, api_key: str, request_tokens: int) -> tuple[bool, str]:
        """ตรวจสอบ Rate Limit และ Quota"""
        if api_key not in self.keys:
            return False, "Invalid API Key"
            
        config = self.keys[api_key]["config"]
        current_time = time.time()
        
        # ตรวจสอบ Rate Limit รายนาที
        recent_requests = [
            t for t in self.usage_tracker[api_key] 
            if current_time - t < 60
        ]
        if len(recent_requests) >= config.requests_per_minute:
            return False, f"Rate limit exceeded: {config.requests_per_minute}/min"
        
        # ตรวจสอบ Token Limit ราย Request
        if request_tokens > config.max_tokens_per_request:
            return False, f"Token limit exceeded: max {config.max_tokens_per_request}"
            
        # บันทึกการใช้งาน
        self.usage_tracker[api_key].append(current_time)
        self.keys[api_key]["total_usage"] += request_tokens
        
        return True, "OK"
    
    def revoke_key(self, api_key: str) -> bool:
        """ยกเลิก API Key"""
        if api_key in self.keys:
            self.keys[api_key]["is_active"] = False
            return True
        return False

ตัวอย่างการใช้งาน
manager = APIKeyManager()

สร้าง Key สำหรับ User ทั่วไป
user_config = RateLimitConfig(
    requests_per_minute=30,
    requests_per_day=5000,
    max_tokens_per_request=2048,
    allowed_models=["gpt-4.1"]
)
user_key = manager.create_key("customer_user", user_config)

สร้าง Key สำหรับ Admin
admin_config = RateLimitConfig(
    requests_per_minute=300,
    requests_per_day=100000,
    max_tokens_per_request=32768,
    allowed_models=["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash"]
)
admin_key = manager.create_key("admin_service", admin_config)

print(f"User Key: {user_key}")
print(f"Admin Key: {admin_key}")

Proxy Server สำหรับ AI API พร้อม Caching และ Quota

import requests
import redis
import json
import hashlib
from fastapi import FastAPI, HTTPException, Header, Request
from fastapi.responses import JSONResponse
from typing import Optional
import time

app = FastAPI(title="AI Proxy with Permission Control")

Redis สำหรับเก็บ Cache และ Quota
redis_client = redis.Redis(host='localhost', port=6379, db=0)

API Key Manager
api_manager = APIKeyManager()

def get_user_quota(key_id: str) -> dict:
    """ดึงข้อมูล Quota ปัจจุบัน"""
    quota_key = f"quota:{key_id}"
    quota_data = redis_client.get(quota_key)
    
    if quota_data:
        return json.loads(quota_data)
    
    return {
        "daily_used": 0,
        "monthly_used": 0,
        "last_reset": time.time()
    }

def update_user_quota(key_id: str, tokens_used: int):
    """อัปเดต Quota หลังใช้งาน"""
    quota_key = f"quota:{key_id}"
    quota = get_user_quota(key_id)
    
    quota["daily_used"] += tokens_used
    quota["monthly_used"] += tokens_used
    
    # Reset รายวันถ้าเกิน 24 ชั่วโมง
    if time.time() - quota["last_reset"] > 86400:
        quota["daily_used"] = tokens_used
        quota["last_reset"] = time.time()
    
    redis_client.setex(quota_key, 86400, json.dumps(quota))

@app.post("/v1/chat/completions")
async def chat_completions(
    request: Request,
    authorization: Optional[str] = Header(None)
):
    # ตรวจสอบ Authorization Header
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing or invalid authorization")
    
    api_key = authorization.replace("Bearer ", "")
    config = api_manager.verify_key(api_key)
    
    if not config:
        raise HTTPException(status_code=401, detail="Invalid or inactive API key")
    
    # ดึง Request Body
    body = await request.json()
    
    # ตรวจสอบ Model ที่อนุญาต
    model = body.get("model", "")
    if model not in config.allowed_models:
        raise HTTPException(
            status_code=403, 
            detail=f"Model '{model}' not allowed. Allowed: {config.allowed_models}"
        )
    
    # คำนวณ Token ที่จะใช้
    estimated_tokens = estimate_tokens(body)
    
    # ตรวจสอบ Rate Limit
    allowed, message = api_manager.check_rate_limit(api_key, estimated_tokens)
    if not allowed:
        raise HTTPException(status_code=429, detail=message)
    
    # ตรวจสอบ Quota
    quota = get_user_quota(api_key)
    daily_limit = config.requests_per_day * config.max_tokens_per_request
    if quota["daily_used"] + estimated_tokens > daily_limit:
        raise HTTPException(
            status_code=429, 
            detail="Daily quota exceeded"
        )
    
    # ส่ง Request ไปยัง HolySheep API
    try:
        response = requests.post(
            f"https://api.holysheep.ai/v1/chat/completions",
            headers={
                "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
                "Content-Type": "application/json"
            },
            json=body,
            timeout=30
        )
        
        # อัปเดต Quota
        actual_tokens = response.json().get("usage", {}).get("total_tokens", estimated_tokens)
        update_user_quota(api_key, actual_tokens)
        
        return response.json()
        
    except requests.exceptions.Timeout:
        raise HTTPException(status_code=504, detail="AI API timeout")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Internal error: {str(e)}")

def estimate_tokens(body: dict) -> int:
    """ประมาณการ Token จาก Request"""
    messages = body.get("messages", [])
    total_chars = sum(len(str(m.get("content", ""))) for m in messages)
    return total_chars // 4 + 100  # Rough estimation

@app.get("/quota")
async def get_quota(authorization: Optional[str] = Header(None)):
    """ดึงข้อมูล Quota ปัจจุบัน"""
    if not authorization:
        raise HTTPException(status_code=401)
    
    api_key = authorization.replace("Bearer ", "")
    config = api_manager.verify_key(api_key)
    
    if not config:
        raise HTTPException(status_code=401)
    
    quota = get_user_quota(api_key)
    daily_limit = config.requests_per_day * config.max_tokens_per_request
    
    return {
        "daily_used": quota["daily_used"],
        "daily_limit": daily_limit,
        "daily_remaining": max(0, daily_limit - quota["daily_used"]),
        "monthly_used": quota["monthly_used"]
    }

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

กรณีศึกษาที่ 2: RAG System ขององค์กร — แยกสิทธิ์ตามแผนก

องค์กรขนาดใหญ่ที่มีข้อมูลลูกค้า ข้อมูลการเงิน และเอกสารภายในหลายระดับ ต้องการให้แผนกต่างๆ เข้าถึง AI ได้เฉพาะเอกสารที่เกี่ยวข้อง ระบบ RAG ที่ดีต้องมี:

• Document Level Permission — กำหนดว่า Key นี้สามารถ Query เอกสารกลุ่มไหนได้บ้าง
• Embedding Model Restriction — บางแผนกต้องใช้ Model เฉพาะทาง
• Audit Trail — บันทึกว่าใครถามอะไร เอกสารอะไรถูกดึงขึ้นมา

from enum import Enum
from typing import Set, Dict
import logging

class Department(Enum):
    SALES = "sales"
    FINANCE = "finance"
    HR = "hr"
    IT = "it"
    EXECUTIVE = "executive"

class DocumentPermission:
    """ระบบจัดการสิทธิ์เอกสาร"""
    
    # กำหนดว่าแต่ละแผนกเข้าถึงเอกสารประเภทไหนได้
    DEPARTMENT_DOCUMENT_MAP = {
        Department.SALES: {"customer_data", "product_info", "sales_reports", "marketing"},
        Department.FINANCE: {"financial_reports", "invoices", "budgets", "tax_docs"},
        Department.HR: {"employee_records", "policies", "payroll", "recruitment"},
        Department.IT: {"technical_docs", "security_logs", "system_configs", "code_repos"},
        Department.EXECUTIVE: {"*"}  # เข้าถึงได้ทุกอย่าง
    }
    
    @classmethod
    def get_allowed_documents(cls, department: Department) -> Set[str]:
        return cls.DEPPARTMENT_DOCUMENT_MAP.get(department, set())
    
    @classmethod
    def can_access_document(cls, department: Department, doc_type: str) -> bool:
        allowed = cls.get_allowed_documents(department)
        if "*" in allowed:
            return True
        return doc_type in allowed

class RAGPermissionMiddleware:
    """Middleware สำหรับ RAG System"""
    
    def __init__(self):
        self.audit_logger = logging.getLogger("rag_audit")
        self.audit_logger.setLevel(logging.INFO)
        
        # ตั้งค่า File Handler สำหรับ Audit Log
        handler = logging.FileHandler("/var/log/rag_audit.log")
        handler.setFormatter(
            logging.Formatter("%(asctime)s - %(user_id)s - %(action)s - %(details)s")
        )
        self.audit_logger.addHandler(handler)
        
    def check_rag_permission(
        self, 
        api_key: str, 
        query: str, 
        collection: str
    ) -> tuple[bool, str]:
        """ตรวจสอบสิทธิ์การ Query RAG"""
        
        # ดึงข้อมูล Key และ Department
        key_info = self._get_key_info(api_key)
        if not key_info:
            return False, "Invalid API key"
            
        department = key_info.get("department")
        
        # ตรวจสอบสิทธิ์เอกสาร
        if not DocumentPermission.can_access_document(department, collection):
            self._log_audit(
                user_id=key_info.get("user_id"),
                action="ACCESS_DENIED",
                details=f"Tried to access {collection}"
            )
            return False, f"Department {department.value} cannot access {collection}"
        
        # ตรวจสอบ Query ไม่ให้ถามเกี่ยวกับเอกสารที่ไม่มีสิทธิ์
        sensitive_keywords = self._detect_sensitive_query(query)
        if sensitive_keywords:
            for keyword in sensitive_keywords:
                # ตรวจว่ามีสิทธิ์เข้าถึงเอกสารประเภทนั้นหรือไม่
                if not DocumentPermission.can_access_document(department, keyword):
                    return False, f"Query contains restricted topic: {keyword}"
        
        self._log_audit(
            user_id=key_info.get("user_id"),
            action="QUERY_APPROVED",
            details=f"Query on {collection}: {query[:100]}"
        )
        
        return True, "Approved"
    
    def _detect_sensitive_query(self, query: str) -> list:
        """ตรวจจับคำที่อาจเกี่ยวกับข้อมูลที่ต้องมีสิทธิ์พิเศษ"""
        query_lower = query.lower()
        sensitive = []
        
        keywords_map = {
            "employee": ["employee_records", "hr", "payroll"],
            "salary": ["employee_records", "hr", "payroll"],
            "revenue": ["financial_reports", "finance"],
            "budget": ["budgets", "financial_reports", "finance"],
            "customer": ["customer_data", "sales"],
            "strategy": ["executive", "strategic_plans"]
        }
        
        for keyword, doc_types in keywords_map.items():
            if keyword in query_lower:
                sensitive.extend(doc_types)
        
        return list(set(sensitive))
    
    def _get_key_info(self, api_key: str) -> dict:
        """ดึงข้อมูล Key (ใน Production ควรดึงจาก Database)"""
        # ตัวอย่าง: ดึงจาก Redis หรือ Database
        return {
            "user_id": "user_123",
            "department": Department.SALES,
            "tier": "premium"
        }
    
    def _log_audit(self, user_id: str, action: str, details: str):
        """บันทึก Audit Log"""
        self.audit_logger.info(
            f"user_id={user_id} action={action} details={details}"
        )

ตัวอย่างการใช้งาน
rag_middleware = RAGPermissionMiddleware()

User จาก Sales พยายามถามเกี่ยวกับ Revenue Report
allowed, message = rag_middleware.check_rag_permission(
    api_key="sales_key_xxx",
    query="What was our Q3 revenue?",
    collection="financial_reports"
)
print(f"Result: {allowed}, Message: {message}")

User จาก Sales พยายามถามเกี่ยวกับ Employee Salary
allowed, message = rag_middleware.check_rag_permission(
    api_key="sales_key_xxx",
    query="What is John Doe's salary?",
    collection="employee_records"
)
print(f"Result: {allowed}, Message: {message}")

ราคาและการเลือก Model ตาม Use Case

สำหรับการควบคุมสิทธิ์ที่ดี ควรกำหนดด้วยว่า Key แต่ละประเภทใช้ Model ใดได้บ้าง ตามราคา 2026 ของ HolySheep AI:

• DeepSeek V3.2 @ $0.42/MTok — สำหรับ Task ทั่วไป, Summary, Classification
• Gemini 2.5 Flash @ $2.50/MTok — สำหรับ Real-time Chat, งานที่ต้องการ Speed
• GPT-4.1 @ $8/MTok — สำหรับ Complex Reasoning, Code Generation
• Claude Sonnet 4.5 @ $15/MTok — สำหรับ Long-context Analysis, Creative Writing

ด้วยอัตรา ¥1=$1 ของ HolySheep ที่ประหยัดกว่า 85% รวมถึงรองรับ WeChat/Alipay ทำให้การชำระเงินสะดวกมาก ระบบใช้งานจริงมี Latency เพียง <50ms ทำให้ User Experience ลื่นไหล

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

กรณีที่ 1: "401 Unauthorized" แม้ใส่ Key ถูกต้อง

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

# ❌ วิธีผิด - ใช้ Header ผิด format
response = requests.post(
    "https://api.holysheep.ai/v1/chat/completions",
    headers={"Authorization": user_key},  # ผิด! ต้องมี Bearer
    json=data
)

✅ วิธีถูก - ต้องมี Bearer prefix
response = requests.post(
    "https://api.holysheep.ai/v1/chat/completions",
    headers={
        "Authorization": f"Bearer {user_key}",
        "Content-Type": "application/json"
    },
    json=data
)

ตรวจสอบ Response Status
if response.status_code == 401:
    error_detail = response.json()
    print(f"Error: {error_detail.get('error', {}).get('message', 'Unknown')}")
    # อาจเป็นเพราะ: key หมดอายุ, ถูก revoke, หรือ formatผิด

กรณีที่ 2: "429 Rate Limit Exceeded" แม้ไม่ได้ส่ง Request เยอะ

สาเหตุ: ไม่ได้ Reset Usage Tracker หรือ Rate Limit Config ต่ำเกินไป หรือมี Background Job ที่ซ่อนอยู่

# ✅ วิธีแก้: ตรวจสอบและ Reset Usage อย่างเหมาะสม

class KeyManager:
    def __init__(self):
        self.usage: Dict[str, List[float]] = defaultdict(list)
        
    def cleanup_old_usage(self, key_id: str, window_seconds: int = 60):
        """ลบ Usage ที่เก่ากว่า window"""
        current_time = time.time()
        self.usage[key_id] = [
            t for t in self.usage[key_id] 
            if current_time - t < window_seconds
        ]
    
    def get_remaining_requests(self, key_id: str, limit: int) -> int:
        """ดึงจำนวน Request ที่เหลือ"""
        self.cleanup_old_usage(key_id)
        used = len(self.usage[key_id])
        return max(0, limit - used)
    
    def check_and_increment(self, key_id: str, limit: int) -> bool:
        """ตรวจสอบและเพิ่ม Usage"""
        self.cleanup_old_usage(key_id)
        
        if len(self.usage[key_id]) >= limit:
            return False  # Rate limit exceeded
            
        self.usage[key_id].append(time.time())
        return True

ใช้งาน
manager = KeyManager()

ตรวจสอบก่อนส่ง Request
remaining = manager.get_remaining_requests("user_key_123", limit=60)
print(f"Remaining requests this minute: {remaining}")

if not manager.check_and_increment("user_key_123", limit=60):
    raise HTTPException(429, "Rate limit exceeded, please wait")

กรณีที่ 3: "403 Forbidden" สำหรับ Model ที่ควรจะใช้ได้

สาเหตุ: ไม่ได้อัปเดต allowed_models หรือ Model name ไม่ตรงกับที่กำหนด

# ✅ วิธีแก้: ตรวจสอบ Model name อย่างละเอียด

VALID_MODELS = {
    # Model name ที่ HolySheep API รองรับ
    "gpt-4.1": {"display": "GPT-4.1", "price_per_mtok": 8.0},
    "claude-sonnet-4.5": {"display": "Claude Sonnet 4.5", "price_per_mtok": 15.0},
    "gemini-2.5-flash": {"display": "Gemini 2.5 Flash", "price_per_mtok": 2.50},
    "deepseek-v3.2": {"display": "DeepSeek V3.2", "price_per_mtok": 0.42}
}

def validate_model_request(api_key: str, requested_model: str) -> dict:
    """ตรวจสอบ Model request"""
    
    # ตรวจสอบว่า Model มีอยู่จริงในระบบ
    if requested_model not in VALID_MODELS:
        available = ", ".join(VALID_MODELS.keys())
        raise HTTPException(
            400, 
            f"Invalid model '{requested_model}'. Available: {available}"
        )
    
    # ตรวจสอบว่า Key มีสิทธิ์ใช้ Model นี้
    key_config = get_key_config(api_key)
    if key_config and requested_model not in key_config.get("allowed_models", []):
        allowed = key_config.get("allowed_models", [])
        raise HTTPException(
            403,
            f"API key cannot access {requested_model}. "
            f"Allowed models: {allowed}"
        )
    
    return {
        "model": requested_model,
        "display": VALID_MODELS[requested_model]["display"],
        "estimated_cost": calculate_cost(requested_model)
    }

ตัวอย่างการใช้งาน
try:
    model_info = validate_model_request("user_key_xxx", "gpt-4.1")
    print(f"Using {model_info['display']}, est. cost: ${model_info['estimated_cost']}")
except HTTPException as e:
    print(f"Error: {e.detail}")

สรุป

การควบคุมสิทธิ์ AI API ที่ดีไม่ใช่แค่เรื่องความปลอดภัย แต่ยังเป็นเรื่องของการจัดการต้นทุนและประสิทธิภาพ ด้วยระบบ Multi-tier API Key, Rate Limiting, Quota Management และ Department-based Permission ที่อธิบายไป คุณจะสามารถ:

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

  • 📚 บทช่วยสอน AI API
  • 💰 ดูราคา
  • 📖 เอกสารสำหรับนักพัฒนา
  • 🚀 สมัครฟรี

  บทความที่เกี่ยวข้อง

  • AI API ส่งข้อมูลยังไง? คู่มือ JSON สำหรับมือใหม่ ฉบับเข้าใจง
  • API เข้ารหัสข้อมูลยี่ห้อไหนดีที่สุด? เปรียบเทียบความเร็วจริง
  • AI API Peak QPS: คู่มือฉบับสมบูรณ์เพื่อรองรับโหลดระดับ Produ