ในระบบ AI Agent ที่ใช้งานจริงใน production environment การจัดการ tool permissions เป็นหัวใจสำคัญที่หลายทีมมองข้าม แต่กลับกลายเป็นจุดอ่อนด้าน security และความเสถียรของระบบในที่สุด ในฐานะวิศวกรที่ดูแล AI infrastructure มาหลายปี ผมเคยเจอกับปัญหา tool ที่ไม่ได้รับอนุญาตถูกเรียกใช้, API ที่ล่มแล้วไม่มี fallback, และ audit log ที่ไม่ครบถ้วนจนไม่สามารถตรวจสอบปัญหาได้ บทความนี้จะพาคุณสร้าง HolySheep MCP Tools Permission Governance ที่ครบวงจร พร้อม production-ready code ที่ผมใช้งานจริงในทีม
ทำไม Tool Permission Governance ถึงสำคัญใน Production
เมื่อคุณ deploy AI agent ที่เชื่อมต่อกับ MCP tools หลายตัว ทุก tool เหล่านั้นคือ potential attack surface ที่ hacker สามารถใช้เพื่อเข้าถึงข้อมูลภายในองค์กร หรือทำ resource exhaustion attack ผมเคยเจอกรณีที่ agent ถูกใช้งานเกินขีดจำกัดเพราะไม่มี rate limiting และ cost พุ่งไปถึง $10,000 ภายในวันเดียว การ implement permission governance ที่ดีจึงไม่ใช่ทางเลือก แต่เป็น ความจำเป็นทางธุรกิจ
HolySheep AI มาพร้อมกับ built-in permission system ที่รองรับ unified authentication, audit logging และ model fallback ทำให้คุณประหยัดเวลา development 2-3 เดือน และมี latency ต่ำกว่า 50ms พร้อมราคาที่ประหยัดกว่า 85%+
สถาปัตยกรรม Unified Authentication System
สถาปัตยกรรมที่เราจะสร้างประกอบด้วย 4 ชั้นหลัก:
- Authentication Layer - ตรวจสอบ API key และ JWT tokens
- Permission Store - Centralized storage สำหรับ tool permissions ที่ใช้ Redis หรือ PostgreSQL
- Audit Logger - บันทึกทุก tool invocation อย่าง asynchronous
- Model Gateway - จัดการ model routing และ fallback chain
import asyncio
import hashlib
import hmac
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, List, Optional, Set
from collections import defaultdict
import aiofiles
import json
class PermissionLevel(Enum):
DENY = 0
READ = 1
WRITE = 2
EXECUTE = 3
ADMIN = 4
@dataclass
class ToolPermission:
tool_name: str
permission_level: PermissionLevel
allowed_users: Set[str] = field(default_factory=set)
allowed_roles: Set[str] = field(default_factory=set)
rate_limit_per_minute: int = 60
quota_per_day: int = 10000
@dataclass
class AuditEntry:
timestamp: float
user_id: str
tool_name: str
action: str
status: str
latency_ms: float
tokens_used: int = 0
ip_address: str = ""
error_message: str = ""
class UnifiedAuthGateway:
"""HolySheep MCP Unified Authentication Gateway"""
def __init__(self, base_url: str = "https://api.holysheep.ai/v1"):
self.base_url = base_url
self.permissions: Dict[str, ToolPermission] = {}
self.user_roles: Dict[str, Set[str]] = defaultdict(set)
self.usage_counters: Dict[str, List[float]] = defaultdict(list)
self.audit_queue: asyncio.Queue = asyncio.Queue()
self._setup_default_permissions()
def _setup_default_permissions(self):
"""Initialize default tool permissions"""
default_tools = [
ToolPermission("file_read", PermissionLevel.READ, rate_limit_per_minute=100),
ToolPermission("file_write", PermissionLevel.WRITE, rate_limit_per_minute=50),
ToolPermission("database_query", PermissionLevel.EXECUTE, rate_limit_per_minute=30),
ToolPermission("web_search", PermissionLevel.READ, rate_limit_per_minute=200),
ToolPermission("code_execute", PermissionLevel.EXECUTE, rate_limit_per_minute=10),
ToolPermission("admin_panel", PermissionLevel.ADMIN, rate_limit_per_minute=5),
]
for tool in default_tools:
self.permissions[tool.tool_name] = tool
def _verify_api_key(self, api_key: str) -> Optional[str]:
"""Verify HolySheep API key and return user_id"""
if not api_key or not api_key.startswith("hs_"):
return None
# Simulate key verification - in production, call HolySheep API
key_hash = hashlib.sha256(api_key.encode()).hexdigest()[:16]
return f"user_{key_hash}"
def _check_permission(self, user_id: str, tool_name: str) -> bool:
"""Check if user has permission to use tool"""
if tool_name not in self.permissions:
return False
perm = self.permissions[tool_name]
# Check user directly
if user_id in perm.allowed_users:
return True
# Check user roles
user_roles = self.user_roles.get(user_id, set())
if user_roles.intersection(perm.allowed_roles):
return True
return False
def _check_rate_limit(self, user_id: str, tool_name: str) -> bool:
"""Check if user is within rate limits"""
current_time = time.time()
minute_ago = current_time - 60
# Clean old entries
self.usage_counters[user_id] = [
t for t in self.usage_counters[user_id] if t > minute_ago
]
current_count = len(self.usage_counters[user_id])
limit = self.permissions.get(tool_name, ToolPermission(tool_name, PermissionLevel.DENY)).rate_limit_per_minute
return current_count < limit
async def authorize_tool_call(
self,
api_key: str,
tool_name: str,
ip_address: str = ""
) -> tuple[bool, Optional[str]]:
"""Main authorization method - returns (authorized, error_message)"""
start_time = time.time()
user_id = self._verify_api_key(api_key)
if not user_id:
await self._log_audit(AuditEntry(
timestamp=start_time,
user_id="unknown",
tool_name=tool_name,
action="authorize",
status="denied",
latency_ms=(time.time() - start_time) * 1000,
ip_address=ip_address,
error_message="Invalid API key"
))
return False, "Invalid API key"
if not self._check_permission(user_id, tool_name):
await self._log_audit(AuditEntry(
timestamp=start_time,
user_id=user_id,
tool_name=tool_name,
action="authorize",
status="denied",
latency_ms=(time.time() - start_time) * 1000,
ip_address=ip_address,
error_message="Permission denied"
))
return False, f"User {user_id} does not have permission to use {tool_name}"
if not self._check_rate_limit(user_id, tool_name):
await self._log_audit(AuditEntry(
timestamp=start_time,
user_id=user_id,
tool_name=tool_name,
action="authorize",
status="rate_limited",
latency_ms=(time.time() - start_time) * 1000,
ip_address=ip_address,
error_message="Rate limit exceeded"
))
return False, "Rate limit exceeded"
# Record usage
self.usage_counters[user_id].append(time.time())
return True, None
async def _log_audit(self, entry: AuditEntry):
"""Async audit logging to prevent blocking"""
await self.audit_queue.put(entry)
async def process_audit_log(self):
"""Background worker to write audit logs"""
while True:
entry = await self.audit_queue.get()
try:
async with aiofiles.open("audit_log.jsonl", mode="a") as f:
await f.write(json.dumps(asdict(entry)) + "\n")
except Exception as e:
print(f"Audit log write failed: {e}")
finally:
self.audit_queue.task_done()
Initialize gateway
gateway = UnifiedAuthGateway()Tool Call Audit System แบบ Real-time
การ audit ไม่ใช่แค่การบันทึก log แต่ต้องสามารถ detect anomaly และ alert ได้แบบ real-time ผมออกแบบระบบที่ใช้ sliding window algorithm สำหรับ anomaly detection และส่ง alert ผ่าน webhook เมื่อพบ pattern ที่ผิดปกติ
import asyncio
from datetime import datetime, timedelta
from collections import defaultdict
from dataclasses import dataclass
from typing import Callable, Dict, List, Optional
import aiohttp
@dataclass
class AnomalyAlert:
alert_type: str
user_id: str
tool_name: str
threshold: float
actual_value: float
timestamp: datetime
severity: str # LOW, MEDIUM, HIGH, CRITICAL
class AuditMonitor:
"""Real-time audit monitoring with anomaly detection"""
def __init__(self, webhook_url: Optional[str] = None):
self.webhook_url = webhook_url
self.anomaly_thresholds = {
"rapid_calls": {"threshold": 50, "window_seconds": 60, "severity": "HIGH"},
"high_token_usage": {"threshold": 100000, "window_seconds": 3600, "severity": "MEDIUM"},
"failed_auth_rate": {"threshold": 0.5, "window_seconds": 300, "severity": "CRITICAL"},
"unusual_tool_access": {"threshold": 10, "window_seconds": 3600, "severity": "MEDIUM"},
}
self.call_history: Dict[str, List[float]] = defaultdict(list)
self.token_usage: Dict[str, List[tuple[float, int]]] = defaultdict(list)
self.auth_failures: Dict[str, List[float]] = defaultdict(list)
self.alert_callbacks: List[Callable] = []
def record_tool_call(self, user_id: str, tool_name: str, timestamp: float):
"""Record a tool call for monitoring"""
key = f"{user_id}:{tool_name}"
self.call_history[key].append(timestamp)
self._clean_old_entries(self.call_history[key])
def record_token_usage(self, user_id: str, timestamp: float, tokens: int):
"""Record token usage for cost monitoring"""
self.token_usage[user_id].append((timestamp, tokens))
self._clean_old_entries([t[0] for t in self.token_usage[user_id]])
def record_auth_failure(self, identifier: str, timestamp: float):
"""Record authentication failure"""
self.auth_failures[identifier].append(timestamp)
self._clean_old_entries(self.auth_failures[identifier])
def _clean_old_entries(self, entries: List[float], window_seconds: int = 3600):
"""Remove entries older than window"""
cutoff = time.time() - window_seconds
while entries and entries[0] < cutoff:
entries.pop(0)
async def check_anomalies(self, user_id: str) -> List[AnomalyAlert]:
"""Check for anomalies and return alerts"""
alerts = []
current_time = time.time()
# Check rapid calls
for key, timestamps in self.call_history.items():
if not key.startswith(f"{user_id}:"):
continue
recent_calls = [t for t in timestamps if current_time - t < 60]
threshold = self.anomaly_thresholds["rapid_calls"]["threshold"]
if len(recent_calls) > threshold:
alerts.append(AnomalyAlert(
alert_type="rapid_calls",
user_id=user_id,
tool_name=key.split(":")[1],
threshold=threshold,
actual_value=len(recent_calls),
timestamp=datetime.now(),
severity=self.anomaly_thresholds["rapid_calls"]["severity"]
))
# Check failed auth rate
recent_failures = [t for t in self.auth_failures.get(user_id, [])
if current_time - t < 300]
total_attempts = len(recent_failures) + 1 # Assume at least 1 success
failure_rate = len(recent_failures) / total_attempts
if failure_rate > self.anomaly_thresholds["failed_auth_rate"]["threshold"]:
alerts.append(AnomalyAlert(
alert_type="failed_auth_rate",
user_id=user_id,
tool_name="N/A",
threshold=self.anomaly_thresholds["failed_auth_rate"]["threshold"],
actual_value=failure_rate,
timestamp=datetime.now(),
severity=self.anomaly_thresholds["failed_auth_rate"]["severity"]
))
# Send alerts
for alert in alerts:
await self._send_alert(alert)
for callback in self.alert_callbacks:
await callback(alert)
return alerts
async def _send_alert(self, alert: AnomalyAlert):
"""Send alert via webhook"""
if not self.webhook_url:
return
try:
async with aiohttp.ClientSession() as session:
await session.post(self.webhook_url, json={
"alert_type": alert.alert_type,
"user_id": alert.user_id,
"tool_name": alert.tool_name,
"severity": alert.severity,
"threshold": alert.threshold,
"actual_value": alert.actual_value,
"timestamp": alert.timestamp.isoformat(),
})
except Exception as e:
print(f"Failed to send alert: {e}")
def add_alert_callback(self, callback: Callable):
"""Add callback for alert handling"""
self.alert_callbacks.append(callback)
def get_user_stats(self, user_id: str) -> Dict:
"""Get usage statistics for a user"""
current_time = time.time()
hour_ago = current_time - 3600
tool_calls = {
key.split(":")[1]: len([t for t in timestamps if t > hour_ago])
for key, timestamps in self.call_history.items()
if key.startswith(f"{user_id}:")
}
total_tokens = sum(
tokens for _, tokens in self.token_usage.get(user_id, [])
if