Ngày 11 tháng 11 năm 2025 — ngày hội sale lớn nhất năm của thương mại điện tử Việt Nam. Đội ngũ kỹ thuật của tôi vừa triển khai hệ thống chatbot AI hỗ trợ khách hàng cho một sàn thương mại điện tử lớn. Hệ thống xử lý 50.000+ yêu cầu mỗi phút, tích hợp mô hình GPT-4.1 thông qua HolySheep AI. Chỉ 2 giờ sau khi ra mắt, đội bảo mật phát hiện một địa chỉ IP lạ đang cố gắng truy cập API endpoint với API key đã bị rò rỉ từ một repository GitHub công khai. Nếu không có kiến trúc Zero Trust, kẻ tấn công đã có thể tiêu tốn hết ngân sách $5.000/ngày của doanh nghiệp trong vòng 10 phút.
Bài viết này chia sẻ kinh nghiệm thực chiến của tôi khi thiết kế hệ thống bảo mật API AI theo nguyên tắc Zero Trust, giúp bạn tránh những rủi ro mà chính team tôi đã gặp phải.
1. Tại Sao Kiến Trúc Zero Trust Quan Trọng Cho API AI?
Trong kiến trúc truyền thống, chúng ta thường tin tưởng các request từ internal network. Nhưng với hệ thống AI API, điều này hoàn toàn không đúng:
- API key là "chìa khóa" duy nhất — Một key bị lộ có thể truy cập toàn bộ tài nguyên
- Chi phí gọi API rất cao — GPT-4.1 có giá $8/MTok, nghĩa là $1 có thể bị "cuốn bay" chỉ trong 125.000 token
- Tấn công bot tự động — Kẻ xấu sử dụng script tự động để khai thác API bị lộ
- Multi-tenant environment — Nhiều ứng dụng chia sẻ cùng API provider
Với HolySheep AI, tỷ giá chỉ ¥1=$1 (tiết kiệm 85%+ so với các provider khác), nhưng nếu không bảo mật đúng cách, chi phí vẫn có thể tăng vọt ngoài tầm kiểm soát.
2. Nguyên Tắc Cốt Lõi Của Zero Trust API
2.1. Never Trust, Always Verify (Không Tin Tưởng, Luôn Xác Minh)
Mọi request đến API đều phải được xác minh, bất kể nguồn gốc. Điều này bao gồm:
- Xác minh API key còn hạn và có quyền
- Kiểm tra IP whitelist/blacklist
- Rate limiting theo từng client
- Logging và monitoring thời gian thực
2.2. Least Privilege Access (Quyền Truy Cập Tối Thiểu)
Mỗi service chỉ nên có quyền truy cập đúng những gì cần thiết. Đừng dùng admin key cho mọi thứ.
2.3. Assume Breach (Giả Định Hệ Thống Đã Bị Xâm Nhập)
Thiết kế hệ thống như thể kẻ tấn công đã ở bên trong. Điều này giúp chuẩn bị kế hoạch phản ứng nhanh.
3. Triển Khai Zero Trust Với HolySheep AI API
3.1. Cấu Hình API Client An Toàn
"""
HolySheep AI API Client với Zero Trust Security Layer
Phiên bản: 2.0.0
Author: HolySheep AI Team
"""
import os
import time
import hashlib
import hmac
import asyncio
from typing import Optional, Dict, Any
from dataclasses import dataclass
from datetime import datetime, timedelta
import httpx
from ratelimit import limits, sleep_and_retry
@dataclass
class APIKeyConfig:
"""Cấu hình API Key với các thuộc tính bảo mật"""
key: str
secret: str
allowed_ips: list[str]
rate_limit_per_minute: int
max_budget_usd: float
current_spend: float = 0.0
created_at: datetime = None
def __post_init__(self):
if self.created_at is None:
self.created_at = datetime.now()
def is_ip_allowed(self, ip: str) -> bool:
"""Kiểm tra IP có trong whitelist không"""
if "*" in self.allowed_ips:
return True
return ip in self.allowed_ips
def check_budget(self, estimated_cost: float) -> bool:
"""Kiểm tra ngân sách còn đủ không"""
return (self.current_spend + estimated_cost) <= self.max_budget_usd
def add_spend(self, amount: float) -> None:
"""Cập nhật chi phí đã sử dụng"""
self.current_spend += amount
class HolySheepZeroTrustClient:
"""
Zero Trust Client cho HolySheep AI API
- Tự động rate limiting
- Kiểm tra ngân sách trước mỗi request
- IP whitelist validation
- Request signing với HMAC
- Retry với exponential backoff
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(
self,
api_key: str,
api_secret: str,
allowed_ips: list[str],
rate_limit: int = 60,
max_budget: float = 100.0
):
self.config = APIKeyConfig(
key=api_key,
secret=api_secret,
allowed_ips=allowed_ips,
rate_limit_per_minute=rate_limit,
max_budget_usd=max_budget
)
self._request_count = 0
self._last_reset = time.time()
self._lock = asyncio.Lock()
def _generate_signature(self, timestamp: int, method: str, path: str, body: str = "") -> str:
"""Tạo HMAC signature để xác minh request integrity"""
message = f"{timestamp}{method}{path}{body}"
signature = hmac.new(
self.config.secret.encode(),
message.encode(),
hashlib.sha256
).hexdigest()
return signature
def _get_client_ip(self, request) -> str:
"""Trích xuất IP thực từ request, xử lý proxy"""
forwarded = request.headers.get("X-Forwarded-For")
if forwarded:
return forwarded.split(",")[0].strip()
return request.client.host
def _reset_rate_counter(self) -> None:
"""Reset counter mỗi phút"""
current_time = time.time()
if current_time - self._last_reset >= 60:
self._request_count = 0
self._last_reset = current_time
async def _check_rate_limit(self, client_ip: str) -> bool:
"""Kiểm tra và cập nhật rate limit"""
async with self._lock:
self._reset_rate_counter()
if self._request_count >= self.config.rate_limit_per_minute:
return False
self._request_count += 1
return True
async def chat_completion(
self,
messages: list[Dict],
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 2048,
client_ip: Optional[str] = None
) -> Dict[str, Any]:
"""
Gọi Chat Completion API với Zero Trust checks
Args:
messages: Danh sách message theo format OpenAI
model: Model ID (gpt-4.1, claude-sonnet-4.5, deepseek-v3.2)
temperature: Độ ngẫu nhiên (0-2)
max_tokens: Số token tối đa trả về
client_ip: IP của client (tự động detect nếu không truyền)
Returns:
Response dict từ API
Raises:
PermissionError: IP không được phép
BudgetExceededError: Vượt ngân sách
RateLimitError: Quá rate limit
"""
# Estimate cost (rough calculation)
input_tokens = sum(len(m.get("content", "").split()) * 1.3 for m in messages)
estimated_tokens = input_tokens + max_tokens
pricing = {
"gpt-4.1": 8.0, # $8/MTok
"claude-sonnet-4.5": 15.0, # $15/MTok
"gemini-2.5-flash": 2.50, # $2.50/MTok
"deepseek-v3.2": 0.42 # $0.42/MTok
}
rate = pricing.get(model, 8.0)
estimated_cost = (estimated_tokens / 1_000_000) * rate
# === ZERO TRUST CHECKS ===
# 1. Budget check
if not self.config.check_budget(estimated_cost):
raise BudgetExceededError(
f"Estimated cost ${estimated_cost:.4f} exceeds budget. "
f"Current spend: ${self.config.current_spend:.2f}"
)
# 2. IP whitelist check
if not self.config.is_ip_allowed(client_ip or "*"):
raise PermissionError(f"IP {client_ip} not in whitelist")
# 3. Rate limit check
if not await self._check_rate_limit(client_ip):
raise RateLimitError(
f"Rate limit exceeded: {self.config.rate_limit_per_minute} req/min"
)
# === MAKE REQUEST ===
timestamp = int(time.time())
headers = {
"Authorization": f"Bearer {self.config.key}",
"Content-Type": "application/json",
"X-Timestamp": str(timestamp),
"X-Signature": self._generate_signature(
timestamp, "POST", "/chat/completions", str(messages)
),
"X-Client-IP": client_ip or "unknown"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
try:
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.BASE_URL}/chat/completions",
json=payload,
headers=headers
)
response.raise_for_status()
result = response.json()
# Update actual spend
actual_tokens = result.get("usage", {}).get("total_tokens", estimated_tokens)
actual_cost = (actual_tokens / 1_000_000) * rate
self.config.add_spend(actual_cost)
return result
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
raise RateLimitError("HolySheep API rate limit exceeded")
elif e.response.status_code == 401:
raise PermissionError("Invalid API key")
raise APIError(f"API error: {e.response.status_code}")
class BudgetExceededError(Exception):
"""Raised when API spend exceeds budget limit"""
pass
class RateLimitError(Exception):
"""Raised when rate limit is exceeded"""
pass
class APIError(Exception):
"""Raised for general API errors"""
pass3.2. Middleware Bảo Mật Cho FastAPI
"""
FastAPI Middleware cho Zero Trust API Security
Triển khai rate limiting, IP filtering, và request signing
"""
from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.security import APIKeyHeader
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from typing import Optional, Callable
import time
import logging
from collections import defaultdict
from dataclasses import dataclass
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
=== CONFIGURATION ===
ALLOWED_IPS = {
"app-server-1": ["10.0.1.100", "10.0.1.101"],
"app-server-2": ["10.0.2.100"],
"development": ["*"] # Chỉ enable trong dev environment
}
RATE_LIMITS = {
"default": 100, # requests per minute
"authenticated": 500,
"premium": 1000
}
=== CUSTOM EXCEPTIONS ===
class SecurityException(Exception):
"""Base exception cho security errors"""
def __init__(self, message: str, ip: str = None, path: str = None):
self.message = message
self.ip = ip
self.path = path
self.timestamp = time.time()
super().__init__(self.message)
# Log ngay lập tức để detect attacks
logger.warning(
f"SECURITY_EVENT | type={self.__class__.__name__} | "
f"ip={ip} | path={path} | ts={self.timestamp}"
)
class IPBlockedError(SecurityException):
"""IP không được phép truy cập"""
pass
class RateLimitExceededError(SecurityException):
"""Vượt quá rate limit"""
pass
class InvalidSignatureError(SecurityException):
"""Signature không hợp lệ"""
pass
@dataclass
class RateLimitEntry:
"""Theo dõi rate limit cho mỗi IP"""
count: int
window_start: float
blocked_until: Optional[float] = None
class ZeroTrustMiddleware(BaseHTTPMiddleware):
"""
Middleware kiểm tra Zero Trust cho mọi request
1. Extract và validate client IP
2. Check IP whitelist
3. Apply rate limiting
4. Validate request signature
5. Log security events
"""
def __init__(self, app, allowed_ips: dict, rate_limits: dict):
super().__init__(app)
self.allowed_ips = allowed_ips
self.rate_limits = rate_limits
self.rate_tracker: dict[str, RateLimitEntry] = defaultdict(
lambda: RateLimitEntry(0, time.time())
)
self._cleanup_interval = 300 # Cleanup every 5 minutes
self._last_cleanup = time.time()
def _get_client_ip(self, request: Request) -> str:
"""Extract real client IP, handle proxies"""
# Check X-Forwarded-For header
forwarded = request.headers.get("x-forwarded-for")
if forwarded:
# Lấy IP đầu tiên (client thật sự)
return forwarded.split(",")[0].strip()
# Check X-Real-IP header
real_ip = request.headers.get("x-real-ip")
if real_ip:
return real_ip.strip()
# Fallback to direct connection IP
if request.client:
return request.client.host
return "unknown"
def _is_ip_allowed(self, ip: str, endpoint_prefix: str = "") -> bool:
"""Check if IP is in any allowed list"""
# Check development wildcard first
dev_ips = self.allowed_ips.get("development", [])
if "*" in dev_ips:
return True
# Check all endpoint-specific lists
for endpoint_ips in self.allowed_ips.values():
if ip in endpoint_ips:
return True
return False
def _check_rate_limit(self, ip: str) -> tuple[bool, int]:
"""
Kiểm tra rate limit
Returns: (is_allowed, current_count)
"""
now = time.time()
entry = self.rate_tracker[ip]
# Reset window if expired (1 minute window)
if now - entry.window_start >= 60:
entry.count = 0
entry.window_start = now
# Check if currently blocked
if entry.blocked_until and now < entry.blocked_until:
return False, entry.count
# Check if would exceed limit
if entry.count >= self.rate_limits["default"]:
# Block for 5 minutes on violation
entry.blocked_until = now + 300
return False, entry.count
# Increment counter
entry.count += 1
return True, entry.count
def _validate_signature(self, request: Request) -> bool:
"""Validate HMAC signature cho sensitive endpoints"""
# Skip for GET requests
if request.method == "GET":
return True
signature = request.headers.get("x-signature")
timestamp = request.headers.get("x-timestamp")
if not signature or not timestamp:
# Log warning but don't block (backward compatibility)
logger.warning(f"Missing signature headers from {self._get_client_ip(request)}")
return True
# Check timestamp is within 5 minutes
try:
ts = int(timestamp)
if abs(time.time() - ts) > 300:
return False
except ValueError:
return False
return True
def _cleanup_old_entries(self) -> None:
"""Dọc dữ liệu rate limit cũ"""
now = time.time()
if now - self._last_cleanup < self._cleanup_interval:
return
to_remove = []
for ip, entry in self.rate_tracker.items():
# Remove entries older than 10 minutes
if now - entry.window_start > 600:
to_remove.append(ip)
for ip in to_remove:
del self.rate_tracker[ip]
self._last_cleanup = now
async def dispatch(self, request: Request, call_next: Callable):
client_ip = self._get_client_ip(request)
path = request.url.path
# === SECURITY CHECKS ===
# 1. IP Whitelist Check
if not self._is_ip_allowed(client_ip):
logger.error(f"BLOCKED_IP | {client_ip} | {path}")
raise IPBlockedError(
"Access denied: IP not in whitelist",
ip=client_ip,
path=path
)
# 2. Rate Limit Check
allowed, count = self._check_rate_limit(client_ip)
if not allowed:
logger.error(f"RATE_LIMIT_EXCEEDED | {client_ip} | count={count}")
raise RateLimitExceededError(
f"Rate limit exceeded. Try again in 5 minutes.",
ip=client_ip,
path=path
)
# 3. Signature Validation (for POST/PUT/PATCH)
if request.method in ["POST", "PUT", "PATCH"]:
if not self._validate_signature(request):
logger.error(f"INVALID_SIGNATURE | {client_ip} | {path}")
raise InvalidSignatureError(
"Invalid request signature",
ip=client_ip,
path=path
)
# === PROCESS REQUEST ===
response = await call_next(request)
# Add security headers
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["Strict-Transport-Security"] = "max-age=31536000"
response.headers["X-RateLimit-Remaining"] = str(
self.rate_limits["default"] - count
)
return response
=== FASTAPI APP EXAMPLE ===
app = FastAPI(title="Zero Trust AI API", version="2.0.0")
Add Zero Trust Middleware
app.add_middleware(
ZeroTrustMiddleware,
allowed_ips=ALLOWED_IPS,
rate_limits=RATE_LIMITS
)
Exception handlers
@app.exception_handler(IPBlockedError)
async def ip_blocked_handler(request: Request, exc: IPBlockedError):
return {"error": exc.message, "type": "IP_BLOCKED"}, 403
@app.exception_handler(RateLimitExceededError)
async def rate_limit_handler(request: Request, exc: RateLimitExceededError):
return {"error": exc.message, "type": "RATE_LIMIT_EXCEEDED"}, 429
@app.exception_handler(InvalidSignatureError)
async def signature_handler(request: Request, exc: InvalidSignatureError):
return {"error": exc.message, "type": "INVALID_SIGNATURE"}, 4013.3. API Key Rotation Và Quản Lý Tài Khoản
/**
* API Key Management System với Zero Trust
* - Tự động rotation
* - Phân quyền theo service
* - Monitoring chi phí theo thời gian thực
*/
// === TYPES ===
interface APIKey {
id: string;
name: string;
prefix: string; // First 8 chars for identification
hash: string; // Hashed version for storage
permissions: Permission[];
allowedIps: string[];
rateLimit: number;
budgetLimit: number;
currentSpend: number;
createdAt: Date;
expiresAt: Date;
lastUsed: Date;
isActive: boolean;
}
type Permission =
| 'chat:complete'
| 'chat:stream'
| 'embeddings:create'
| 'models:list'
| 'admin:keys';
interface CostAlert {
threshold: number; // Percentage (e.g., 50, 75, 90, 100)
notified: boolean;
notifiedAt?: Date;
}
// === API KEY MANAGER CLASS ===
class ZeroTrustKeyManager {
private keys: Map = new Map();
private costAlerts: Map = new Map();
private holySheepBaseUrl = 'https://api.holysheep.ai/v1';
// Pricing reference (USD per million tokens)
private readonly PRICING: Record = {
'gpt-4.1': 8.00,
'claude-sonnet-4.5': 15.00,
'gemini-2.5-flash': 2.50,
'deepseek-v3.2': 0.42,
};
/**
* Tạo API Key mới với cấu hình bảo mật
*/
async createKey(
name: string,
permissions: Permission[],
options: {
allowedIps?: string[];
rateLimit?: number;
budgetLimit?: number;
expiresInDays?: number;
} = {}
): Promise<{ key: string; keyId: string }> {
// Generate secure random key
const key = sk-hs-${this.generateSecureToken(48)};
const keyId = key_${this.generateSecureToken(16)};
const apiKey: APIKey = {
id: keyId,
name,
prefix: key.substring(0, 12),
hash: await this.hashKey(key),
permissions,
allowedIps: options.allowedIps || ['*'],
rateLimit: options.rateLimit || 100,
budgetLimit: options.budgetLimit || 100,
currentSpend: 0,
createdAt: new Date(),
expiresAt: new Date(Date.now() + (options.expiresInDays || 90) * 24 * 60 * 60 * 1000),
lastUsed: new Date(),
isActive: true,
};
this.keys.set(keyId, apiKey);
// Setup cost alerts
this.costAlerts.set(keyId, [
{ threshold: 50, notified: false },
{ threshold: 75, notified: false },
{ threshold: 90, notified: false },
{ threshold: 100, notified: false },
]);
console.log(✅ Created API Key: ${name} (${keyId}));
console.log( Permissions: ${permissions.join(', ')});
console.log( Budget Limit: $${options.budgetLimit || 100});
return { key, keyId };
}
/**
* Validate request với Zero Trust checks
*/
async validateRequest(request: {
apiKey: string;
ip: string;
endpoint: string;
estimatedTokens: number;
model: string;
}): Promise<{ valid: boolean; error?: string; key?: APIKey }> {
const { apiKey, ip, endpoint, estimatedTokens, model } = request;
// 1. Extract key ID from key
const keyId = await this.findKeyId(apiKey);
if (!keyId) {
return { valid: false, error: 'Invalid API key format' };
}
const key = this.keys.get(keyId);
if (!key) {
return { valid: false, error: 'API key not found' };
}
// 2. Check if key is active
if (!key.isActive) {
return { valid: false, error: 'API key has been deactivated' };
}
// 3. Check expiration
if (new Date() > key.expiresAt) {
return { valid: false, error: 'API key has expired' };
}
// 4. Verify key hash
const hashMatch = await this.verifyKeyHash(apiKey, key.hash);
if (!hashMatch) {
return { valid: false, error: 'API key verification failed' };
}
// 5. Check IP whitelist
if (!this.isIpAllowed(ip, key.allowedIps)) {
console.warn(🚫 IP ${ip} blocked for key ${keyId});
return { valid: false, error: 'IP not in whitelist' };
}
// 6. Check permissions
const requiredPermission = this.getRequiredPermission(endpoint);
if (!key.permissions.includes(requiredPermission) &&
!key.permissions.includes('admin:keys')) {
return { valid: false, error: Missing permission: ${requiredPermission} };
}
// 7. Check budget
const estimatedCost = (estimatedTokens / 1_000_000) * (this.PRICING[model] || 8);
if (key.currentSpend + estimatedCost > key.budgetLimit) {
console.warn(💰 Budget exceeded for ${keyId}: $${key.currentSpend} + $${estimatedCost} > $${key.budgetLimit});
return { valid: false, error: 'Budget limit exceeded' };
}
// Update last used
key.lastUsed = new Date();
return { valid: true, key };
}
/**
* Cập nhật chi phí và kiểm tra alerts
*/
async recordUsage(keyId: string, tokens: number, model: string): Promise {
const key = this.keys.get(keyId);
if (!key) return;
const cost = (tokens / 1_000_000) * (this.PRICING[model] || 8);
key.currentSpend += cost;
// Check cost alerts
const alerts = this.costAlerts.get(keyId) || [];
for (const alert of alerts) {
const percentage = (key.currentSpend / key.budgetLimit) * 100;
if (percentage >= alert.threshold && !alert.notified) {
await this.sendCostAlert(key, alert.threshold);
alert.notified = true;
alert.notifiedAt = new Date();
}
}
// Auto-disable at 100%
if (key.currentSpend >= key.budgetLimit) {
await this.disableKey(keyId, 'Budget limit reached');
}
}
/**
* Tự động rotation key
*/
async rotateKey(oldKeyId: string): Promise<{ newKey: string; newKeyId: string }> {
const oldKey = this.keys.get(oldKeyId);
if (!oldKey) {
throw new Error('Key not found');
}
// Create new key with same permissions
const newKeyResult = await this.createKey(
${oldKey.name} (rotated),
oldKey.permissions,
{
allowedIps: oldKey.allowedIps,
rateLimit: oldKey.rateLimit,
budgetLimit: oldKey.budgetLimit,
expiresInDays: 90,
}
);
// Deactivate old key after grace period (24 hours)
setTimeout(() => {
if (this.keys.has(oldKeyId)) {
this.disableKey(oldKeyId, 'Rotation completed');
}
}, 24 * 60 * 60 * 1000);
console.log(🔄 Key rotated: ${oldKeyId} → ${newKeyResult.newKeyId});
return newKeyResult;
}
/**
* Lấy báo cáo chi phí chi tiết
*/
getCostReport(keyId?: string): any {
if (keyId) {
const key = this.keys.get(keyId);
if (!key) return null;
return {
keyId,
name: key.name,
currentSpend: $${key.currentSpend.toFixed(2)},
budgetLimit: $${key.budgetLimit.toFixed(2)},
usagePercent: ${((key.currentSpend / key.budgetLimit) * 100).toFixed(1)}%,
remaining: $${(key.budgetLimit - key.currentSpend).toFixed(2)},
lastUsed: key.lastUsed.toISOString(),
isActive: key.isActive,
};
}
// All keys report
const report = {
totalKeys: this.keys.size,
activeKeys: Array.from(this.keys.values()).filter(k => k.isActive).length,
totalSpend: 0,
totalBudget: 0,
keys: [] as any[],
};
for (const key of this.keys.values()) {
report.totalSpend += key.currentSpend;
report.totalBudget += key.budgetLimit;
report.keys.push(this.getCostReport(key.id));
}
return report;
}
// === HELPER METHODS ===
private generateSecureToken(length: number): string {
const chars = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789';
const array = new Uint8Array(length);
crypto.getRandomValues(array);
return Array.from(array, byte => chars[byte % chars.length]).join('');
}
private async hashKey(key: string): Promise {
const encoder = new TextEncoder();
const data = encoder.encode(key + 'salt_holysheep_2024');
const hashBuffer = await crypto.subtle.digest('SHA-256', data);
return Array.from(new Uint8Array(hashBuffer))
.map(b => b.toString(16).padStart(2, '0'))
.join('');
}
private async verifyKeyHash(key: string, storedHash: string): Promise {
const hash = await this.hashKey(key);
return hash === storedHash;
}
private isIpAllowed(ip: string, allowedIps: string[]): boolean {
if (allowedIps.includes('*')) return true;
return allowedIps.includes(ip);
}
private getRequiredPermission(endpoint: string): Permission {
if (endpoint.includes('/chat/completions')) return 'chat:complete';
if (endpoint.includes('/chat/streaming')) return 'chat:stream';
if (endpoint.includes('/embeddings')) return 'embeddings:create';
if (endpoint.includes('/models')) return 'models:list';
return 'chat:complete';
}
private async findKeyId(key: string): Promise {
const prefix = key.substring(0, 12);
for (const [keyId, keyData] of this.keys.entries()) {
if (keyData.prefix === prefix) {
return keyId;
}
}
return null;
}
private async sendCostAlert(key: APIKey, threshold: number): Promise {
console.log(🚨 COST ALERT: ${key.name} reached ${threshold}% of budget ($${key.currentSpend.toFixed(2)} / $${key.budgetLimit}));
// Implement notification here (email, Slack, etc.)
}
private async disableKey(keyId: string, reason: string): Promise {
const key = this.keys.get(keyId);
if (key) {
key.isActive = false;
console.log(❌ Key disabled: ${keyId} - ${reason});
}
}
}
// === USAGE EXAMPLE ===
async function main() {
const manager = new ZeroTrustKeyManager