Mở Đầu: Khi API Key Bị Lộ - Cơn Ác Mộng Của Developer

Tối hôm đó, tôi nhận được notification từ hệ thống monitoring: 401 Unauthorized - Excessive API calls detected. Trong vòng 2 giờ, 50,000 request đã được thực hiện từ một địa chỉ IP ở Frankfurt, Đức. API Key của tôi đã bị commit lên GitHub public repository từ 3 ngày trước.

Thiệt hại: $847 tiền API credits bị đốt cháy chỉ trong 2 giờ. Kịch bản này hoàn toàn có thể phòng tránh nếu bạn có một hệ thống phát hiện rò rỉ API Key và cảnh báo bảo mật tự động.

Trong bài viết này, tôi sẽ chia sẻ cách xây dựng một hệ thống bảo mật API Key hoàn chỉnh sử dụng HolySheep AI - nền tảng với chi phí chỉ từ $0.42/MTok (tiết kiệm 85%+ so với OpenAI) và độ trễ dưới 50ms.

Tại Sao API Key Bị Rò Rỉ?

Kiến Trúc Hệ Thống Phát Hiện Rò Rỉ

+------------------+     +-------------------+     +------------------+
|   GitHub Webhook | --> |   Scanner Agent   | --> |  Alert System    |
+------------------+     +-------------------+     +------------------+
                                |
                                v
+------------------------+     +-------------------+     +--------------+
|   Log Monitoring       | --> |  Pattern Matching | --> | Auto-Revoke |
+------------------------+     +-------------------+     +--------------+
                                |
                                v
+------------------------+     +-------------------+
|   Usage Anomaly        | --> |  HolySheep API   |
|   Detection            |     |  (Monitor/Block)  |
+------------------------+     +-------------------+

Triển Khai Hệ Thống Bảo Mật Hoàn Chỉnh

1. Cài Đặt Môi Trường

pip install requests hashlib re json time datetime smtplib
git clone https://github.com/your-org/api-key-scanner.git
cd api-key-scanner

2. Cấu Hình API Key HolySheep

# config.py
import os


=== HOLYSHEEP API CONFIGURATION ===


Base URL: https://api.holysheep.ai/v1


Sign up: https://www.holysheep.ai/register


Pricing 2026: GPT-4.1 $8/MTok, Claude Sonnet 4.5 $15/MTok


DeepSeek V3.2 chỉ $0.42/MTok - tiết kiệm 85%+


HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"


=== SECURITY SETTINGS ===

ALERT_THRESHOLDS = {
    "unusual_ip_access": 10,        # Báo động nếu >10 request từ IP lạ
    "unusual_hour_access": True,    # Báo động nếu truy cập lúc 2-5h sáng
    "high_volume_burst": 100,       # Báo động nếu >100 request trong 5 phút
    "geolocation_change": True,     # Báo động nếu đổi quốc gia đột ngột
}


=== MONITORED KEYS (Encrypted Storage) ===

MONITORED_KEYS = []  # Sẽ được load từ secure storage


=== ALERT CONFIGURATION ===

ALERT_WEBHOOK = os.getenv("ALERT_WEBHOOK_URL")
SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK_URL")
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")


=== GITHUB INTEGRATION ===

GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
GITHUB_WEBHOOK_SECRET = os.getenv("GITHUB_WEBHOOK_SECRET")

3. Module Phát Hiện Rò Rỉ API Key

# api_key_scanner.py
import re
import hashlib
import requests
from datetime import datetime
from typing import Dict, List, Optional

class APIKeyLeakDetector:
    """
    Hệ thống phát hiện rò rỉ API Key
    Sử dụng HolySheep AI API để phân tích và giám sát
    """
    
    # Patterns phổ biến của API keys
    API_KEY_PATTERNS = {
        "holysheep": r"sk-holysheep-[a-zA-Z0-9]{32,}",
        "openai": r"sk-[a-zA-Z0-9]{48}",
        "anthropic": r"sk-ant-[a-zA-Z0-9]{48}",
        "google": r"AIza[0-9A-Za-z\\-_]{35}",
        "aws": r"AKIA[0-9A-Z]{16}",
    }
    
    def __init__(self, holysheep_api_key: str):
        self.holysheep_api_key = holysheep_api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.known_leaks = set()
        
    def scan_text(self, text: str) -> Dict[str, any]:
        """
        Quét text để tìm API keys
        """
        findings = []
        
        for key_type, pattern in self.API_KEY_PATTERNS.items():
            matches = re.finditer(pattern, text)
            for match in matches:
                key_hash = self._hash_key(match.group())
                if key_hash not in self.known_leaks:
                    findings.append({
                        "type": key_type,
                        "key": match.group()[:20] + "***",  # Masked
                        "hash": key_hash,
                        "position": match.start(),
                        "timestamp": datetime.now().isoformat()
                    })
                    self.known_leaks.add(key_hash)
        
        return {
            "found": len(findings) > 0,
            "count": len(findings),
            "findings": findings
        }
    
    def scan_github_repo(self, repo_url: str, access_token: str) -> Dict:
        """
        Quét GitHub repository cho API keys
        """
        # Extract owner/repo from URL
        parts = repo_url.replace("https://github.com/", "").split("/")
        owner, repo = parts[0], parts[1]
        
        headers = {
            "Authorization": f"token {access_token}",
            "Accept": "application/vnd.github.v3+json"
        }
        
        # Search for secrets in code
        search_query = f"repo:{owner}/{repo} " + " OR ".join([
            "sk-", "api_key", "apikey", "api-key", "secret",
            "password", "token", "bearer"
        ])
        
        response = requests.get(
            "https://api.github.com/search/code",
            headers=headers,
            params={"q": search_query, "per_page": 100}
        )
        
        if response.status_code == 200:
            return self._analyze_github_results(response.json())
        else:
            return {"error": f"GitHub API error: {response.status_code}"}
    
    def _analyze_github_results(self, results: Dict) -> Dict:
        """
        Phân tích kết quả từ GitHub API
        """
        suspicious_files = []
        
        for item in results.get("items", []):
            # Kiểm tra file name
            filename = item.get("name", "").lower()
            
            if any(suspicious in filename for suspicious in 
                   ["env", "config", "secret", "credential", ".json"]):
                suspicious_files.append({
                    "file": item.get("name"),
                    "path": item.get("path"),
                    "url": item.get("html_url")
                })
        
        return {
            "total_matches": results.get("total_count", 0),
            "suspicious_files": suspicious_files,
            "risk_level": "HIGH" if len(suspicious_files) > 0 else "LOW"
        }
    
    def check_key_validity(self, key: str, provider: str = "holysheep") -> bool:
        """
        Kiểm tra xem API key còn valid không
        """
        if provider == "holysheep":
            headers = {
                "Authorization": f"Bearer {key}",
                "Content-Type": "application/json"
            }
            
            try:
                response = requests.get(
                    f"{self.base_url}/models",
                    headers=headers,
                    timeout=5
                )
                return response.status_code == 200
            except requests.exceptions.RequestException:
                return False
        
        return False
    
    def _hash_key(self, key: str) -> str:
        """Tạo hash của API key để so sánh"""
        return hashlib.sha256(key.encode()).hexdigest()[:16]



=== HOLYSHEEP USAGE ANALYZER ===

class HolySheepUsageAnalyzer:
    """
    Phân tích usage patterns với HolySheep AI API
    HolySheep Pricing 2026: DeepSeek V3.2 $0.42/MTok
    """
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.baseline_usage = {}
        
    def get_usage_stats(self) -> Dict:
        """
        Lấy thống kê sử dụng từ HolySheep
        """
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        try:
            # Get account info
            response = requests.get(
                f"{self.base_url}/usage",
                headers=headers,
                timeout=10
            )
            
            if response.status_code == 200:
                data = response.json()
                return self._analyze_usage_pattern(data)
            else:
                return {"error": f"Status {response.status_code}"}
                
        except requests.exceptions.Timeout:
            return {"error": "Request timeout - có thể bị rate limit"}
        except requests.exceptions.ConnectionError:
            return {"error": "ConnectionError - kiểm tra network"}
    
    def _analyze_usage_pattern(self, data: Dict) -> Dict:
        """
        Phân tích pattern sử dụng
        """
        current_usage = data.get("total_usage", 0)
        
        # So sánh với baseline
        if self.baseline_usage:
            change_percent = (
                (current_usage - self.baseline_usage["total"]) 
                / self.baseline_usage["total"] * 100
            )
        else:
            change_percent = 0
        
        return {
            "total_tokens": current_usage,
            "daily_usage": data.get("daily_usage", []),
            "cost_estimate": self._estimate_cost(current_usage),
            "change_from_baseline": change_percent,
            "anomaly_detected": change_percent > 200  # Báo động nếu tăng >200%
        }
    
    def _estimate_cost(self, tokens: int) -> float:
        """
        Ước tính chi phí với HolySheep Pricing
        GPT-4.1: $8/MTok, DeepSeek V3.2: $0.42/MTok
        """
        m_tokens = tokens / 1_000_000
        avg_cost_per_mtok = 2.50  # Trung bình các model
        
        return round(m_tokens * avg_cost_per_mtok, 4)

4. Module Cảnh Báo Bảo Mật Tự Động

# alert_system.py
import requests
import json
from datetime import datetime
from typing import Dict, List
from enum import Enum

class AlertLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class SecurityAlertSystem:
    """
    Hệ thống cảnh báo bảo mật đa kênh
    Hỗ trợ: Email, Slack, Telegram, Webhook
    """
    
    def __init__(self):
        self.alert_history = []
        
    def send_alert(
        self,
        title: str,
        message: str,
        level: AlertLevel,
        channels: List[str],
        metadata: Dict = None
    ):
        """
        Gửi cảnh báo qua nhiều kênh
        """
        alert = {
            "title": title,
            "message": message,
            "level": level.value,
            "timestamp": datetime.now().isoformat(),
            "metadata": metadata or {}
        }
        
        self.alert_history.append(alert)
        
        # Gửi qua các kênh
        if "slack" in channels:
            self._send_slack(alert)
        if "telegram" in channels:
            self._send_telegram(alert)
        if "email" in channels:
            self._send_email(alert)
        if "webhook" in channels:
            self._send_webhook(alert)
            
    def _format_slack_message(self, alert: Dict) -> Dict:
        """Format message cho Slack"""
        level_emoji = {
            "low": ":large_yellow_circle:",
            "medium": ":large_orange_circle:",
            "high": ":red_circle:",
            "critical": ":rotating_light:"
        }
        
        return {
            "blocks": [
                {
                    "type": "header",
                    "text": {
                        "type": "plain_text",
                        "text": f"{level_emoji.get(alert['level'], ':warning:')} {alert['title']}"
                    }
                },
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": f"*{alert['message']}*\n\n⏰ {alert['timestamp']}"
                    }
                },
                {
                    "type": "context",
                    "elements": [
                        {
                            "type": "mrkdwn",
                            "text": f"📊 Level: *{alert['level'].upper()}*"
                        }
                    ]
                }
            ]
        }
    
    def _send_slack(self, alert: Dict):
        """Gửi qua Slack webhook"""
        webhook_url = "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"
        
        try:
            response = requests.post(
                webhook_url,
                json=self._format_slack_message(alert),
                timeout=10
            )
            return response.status_code == 200
        except requests.exceptions.RequestException as e:
            print(f"Slack send failed: {e}")
            return False
    
    def _send_telegram(self, alert: Dict):
        """Gửi qua Telegram bot"""
        bot_token = "YOUR_TELEGRAM_BOT_TOKEN"
        chat_id = "YOUR_CHAT_ID"
        
        level_prefix = {
            "low": "⚠️ LOW",
            "medium": "🔶 MEDIUM", 
            "high": "🔴 HIGH",
            "critical": "🚨 CRITICAL"
        }
        
        message = f"""
{level_prefix.get(alert['level'], '⚠️')} *{alert['title']}*

{alert['message']}

🕐 {alert['timestamp']}
"""
        
        try:
            response = requests.post(
                f"https://api.telegram.org/bot{bot_token}/sendMessage",
                json={
                    "chat_id": chat_id,
                    "text": message,
                    "parse_mode": "Markdown"
                },
                timeout=10
            )
            return response.status_code == 200
        except requests.exceptions.RequestException as e:
            print(f"Telegram send failed: {e}")
            return False
    
    def _send_email(self, alert: Dict):
        """Gửi qua Email"""
        import smtplib
        from email.mime.text import MIMEText
        from email.mime.multipart import MIMEMultipart
        
        sender = "[email protected]"
        receivers = ["[email protected]"]
        
        msg = MIMEMultipart('alternative')
        msg['Subject'] = f"[{alert['level'].upper()}] {alert['title']}"
        msg['From'] = sender
        msg['To'] = ", ".join(receivers)
        
        html = f"""
        <h2 style="color: {'red' if alert['level'] in ['high', 'critical'] else 'orange'}">
            {alert['title']}
        </h2>
        <p>{alert['message']}</p>
        <p><strong>Timestamp:</strong> {alert['timestamp']}</p>
        <p><strong>Level:</strong> {alert['level'].upper()}</p>
        """
        
        msg.attach(MIMEText(html, 'html'))
        
        try:
            # Gửi email (cần SMTP server config)
            return True
        except Exception as e:
            print(f"Email send failed: {e}")
            return False
    
    def _send_webhook(self, alert: Dict):
        """Gửi qua generic webhook"""
        webhook_url = "https://your-webhook-endpoint.com/security"
        
        try:
            response = requests.post(
                webhook_url,
                json=alert,
                headers={"Content-Type": "application/json"},
                timeout=10
            )
            return response.status_code in [200, 201, 202]
        except requests.exceptions.RequestException as e:
            print(f"Webhook send failed: {e}")
            return False



=== AUTO-REVOKE SYSTEM ===

class AutoRevokeSystem:
    """
    Hệ thống tự động vô hiệu hóa API key khi phát hiện rò rỉ
    """
    
    def __init__(self, holysheep_api_key: str):
        self.api_key = holysheep_api_key
        self.base_url = "https://api.holysheep.ai/v1"
        
    def revoke_key(self, key_to_revoke: str, reason: str) -> Dict:
        """
        Vô hiệu hóa API key qua HolySheep Admin API
        """
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "action": "revoke",
            "key_hash": hashlib.sha256(key_to_revoke.encode()).hexdigest()[:16],
            "reason": reason,
            "timestamp": datetime.now().isoformat()
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/admin/keys/revoke",
                headers=headers,
                json=payload,
                timeout=10
            )
            
            return {
                "success": response.status_code == 200,
                "response": response.json() if response.status_code == 200 else {"error": response.text}
            }
        except requests.exceptions.RequestException as e:
            return {"success": False, "error": str(e)}
    
    def rotate_key(self, old_key: str) -> Dict:
        """
        Tạo API key mới và vô hiệu hóa key cũ
        """
        # 1. Generate new key
        new_key = f"sk-holysheep-{secrets.token_urlsafe(32)}"
        
        # 2. Revoke old key
        revoke_result = self.revoke_key(old_key, "Key rotation - security update")
        
        if revoke_result["success"]:
            return {
                "success": True,
                "new_key": new_key,
                "old_key_revoked": True
            }
        
        return {"success": False, "error": "Failed to revoke old key"}

5. Main Orchestrator - Ghép Nối Toàn Bộ Hệ Thống

# main.py
#!/usr/bin/env python3
"""
API Key Security Scanner - Main Orchestrator
============================================
Hệ thống giám sát API Key toàn diện
Author: HolySheep AI Security Team
"""

import os
import time
import schedule
import hashlib
import requests
from dotenv import load_dotenv
from api_key_scanner import APIKeyLeakDetector, HolySheepUsageAnalyzer
from alert_system import SecurityAlertSystem, AutoRevokeSystem, AlertLevel

load_dotenv()


=== CONFIGURATION ===

HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")  # Key từ https://www.holysheep.ai/register
MONITORED_KEYS = os.getenv("MONITORED_API_KEYS", "").split(",")


=== INITIALIZE SYSTEMS ===

scanner = APIKeyLeakDetector(HOLYSHEEP_API_KEY)
usage_analyzer = HolySheepUsageAnalyzer(HOLYSHEEP_API_KEY)
alert_system = SecurityAlertSystem()
auto_revoke = AutoRevokeSystem(HOLYSHEEP_API_KEY)

def check_github_leaks():
    """
    Kiểm tra rò rỉ trên GitHub mỗi 30 phút
    """
    print("[*] Checking GitHub for leaked keys...")
    
    # Danh sách repositories cần monitor
    repos_to_check = [
        "https://github.com/your-org/frontend-app",
        "https://github.com/your-org/backend-service",
    ]
    
    for repo in repos_to_check:
        result = scanner.scan_github_repo(repo, os.getenv("GITHUB_TOKEN"))
        
        if result.get("risk_level") == "HIGH":
            alert_system.send_alert(
                title=f"⚠️ GitHub Leak Detected in {repo.split('/')[-1]}",
                message=f"Tìm thấy {result['total_matches']} file nghi ngờ chứa API key",
                level=AlertLevel.CRITICAL,
                channels=["slack", "telegram", "email"],
                metadata={"repo": repo, "files": result["suspicious_files"]}
            )

def check_usage_anomaly():
    """
    Kiểm tra anomaly trong usage mỗi 15 phút
    """
    print("[*] Checking usage patterns...")
    
    stats = usage_analyzer.get_usage_stats()
    
    if stats.get("anomaly_detected"):
        estimated_cost = stats.get("cost_estimate", 0)
        
        alert_system.send_alert(
            title=f"🚨 Usage Anomaly Detected - ${estimated_cost}",
            message=f"Usage tăng {stats['change_from_baseline']:.1f}% so với baseline. "
                    f"Có thể API key đã bị compromise.",
            level=AlertLevel.HIGH,
            channels=["slack", "telegram", "email", "webhook"],
            metadata=stats
        )
        
        # Tự động revoke nếu mức tăng > 500%
        if stats['change_from_baseline'] > 500:
            print("[!] Auto-revoking compromised keys...")
            for key in MONITORED_KEYS:
                auto_revoke.revoke_key(
                    key, 
                    f"Auto-revoke: Usage anomaly detected ({stats['change_from_baseline']:.1f}% increase)"
                )

def monitor_realtime():
    """
    Monitor realtime sử dụng HolySheep Webhook
    HolySheep cung cấp <50ms latency monitoring
    """
    print("[*] Monitoring real-time API calls...")
    
    # Gọi HolySheep API để verify key status
    headers = {
        "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
        "Content-Type": "application/json"
    }
    
    try:
        response = requests.get(
            "https://api.holysheep.ai/v1/models",
            headers=headers,
            timeout=5
        )
        
        if response.status_code == 401:
            alert_system.send_alert(
                title="🔴 API Key Authentication Failed",
                message="HolySheep API trả về 401 Unauthorized. "
                        "API key có thể đã bị revoke hoặc expired.",
                level=AlertLevel.CRITICAL,
                channels=["slack", "telegram", "email"]
            )
        elif response.status_code == 429:
            alert_system.send_alert(
                title="🟡 Rate Limit Warning",
                message="HolySheep API rate limit exceeded. "
                        "Kiểm tra usage limits tại dashboard.",
                level=AlertLevel.MEDIUM,
                channels=["slack"]
            )
            
    except requests.exceptions.Timeout:
        alert_system.send_alert(
            title="⚠️ Connection Timeout",
            message="Không thể kết nối đến HolySheep API. "
                    "Kiểm tra network hoặc API status.",
            level=AlertLevel.MEDIUM,
            channels=["slack", "telegram"]
        )
    except requests.exceptions.ConnectionError as e:
        alert_system.send_alert(
            title="🔴 Connection Error",
            message=f"ConnectionError: {str(e)}",
            level=AlertLevel.HIGH,
            channels=["slack", "telegram", "email"]
        )


=== SCHEDULE JOBS ===

schedule.every(30).minutes.do(check_github_leaks)
schedule.every(15).minutes.do(check_usage_anomaly)
schedule.every(5).minutes.do(monitor_realtime)

def main():
    """
    Main entry point
    """
    print("=" * 50)
    print("🔒 API Key Security Scanner Started")
    print("=" * 50)
    print(f"[*] HolySheep API: https://api.holysheep.ai/v1")
    print(f"[*] Pricing: GPT-4.1 $8/MTok, DeepSeek $0.42/MTok")
    print(f"[*] Monitoring {len(MONITORED_KEYS)} API keys")
    print("=" * 50)
    
    while True:
        schedule.run_pending()
        time.sleep(1)

if __name__ == "__main__":
    main()

Tích Hợp GitHub Actions - CI/CD Pipeline

# .github/workflows/api-key-security.yml
name: API Key Security Scan

on:
  push:
    branches: [ main, develop ]
  pull_request:
    branches: [ main ]

jobs:
  security-scan:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.9'
    
    - name: Install dependencies
      run: |
        pip install requests PyYAML
    
    - name: Run API Key Scanner
      env:
        HOLYSHEEP_API_KEY: ${{ secrets.HOLYSHEEP_API_KEY }}
        GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
      run: python main.py --scan-only
    
    - name: Check for exposed keys
      run: |
        # Scan for common patterns
        git log --oneline -1
        git diff HEAD~1 --name-only | while read file; do
          if grep -E "(sk-|api_key|apikey)" "$file" 2>/dev/null; then
            echo "::error::Potential API key found in $file"
            exit 1
          fi
        done
    
    - name: Slack Notification
      if: failure()
      uses: slackapi/slack-github-action@v1
      with:
        payload: |
          {
            "text": "🚨 Security Alert: Possible API key exposure detected",
            "blocks": [{
              "type": "section",
              "text": {
                "type": "mrkdwn",
                "text": "*API Key Security Scan Failed*\nRepository: ${{ github.repository }}\nCommit: ${{ github.sha }}"
              }
            }]
          }
      env:
        SLACK_WEBHOOK_TYPE: INCOMING_WEBHOOK
        SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}

Tích Hợp Webhook Với HolySheep Dashboard

# holySheep_webhook_server.py
"""
HolySheep Webhook Server - Nhận event từ HolySheep API
Endpoint: https://your-domain.com/webhook/holySheep
"""

from flask import Flask, request, jsonify
import hmac
import hashlib
import json
from datetime import datetime

app = Flask(__name__)

WEBHOOK_SECRET = "your-webhook-secret-from-holySheep"

@app.route('/webhook/holySheep', methods=['POST'])
def handle_holySheep_webhook():
    """
    Xử lý webhook từ HolySheep
    Events: usage_alert, key_compromised, rate_limit_exceeded
    """
    # Verify signature
    signature = request.headers.get('X-HolySheep-Signature')
    payload = request.get_data()
    
    expected_sig = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    if not hmac.compare_digest(signature, expected_sig):
        return jsonify({"error": "Invalid signature"}), 401
    
    event = request.json
    
    # Process events
    event_type = event.get('event')
    
    if event_type == 'usage_alert':
        handle_usage_alert(event)
    elif event_type == 'key_compromised':
        handle_key_compromised(event)
    elif event_type == 'rate_limit_exceeded':
        handle_rate_limit(event)
    
    return jsonify({"status": "processed"}), 200

def handle_usage_alert(event):
    """
    Xử lý cảnh báo usage
    HolySheep gửi alert khi usage vượt ngưỡng
    """
    data = event.get('data', {})
    current_usage = data.get('usage', 0)
    threshold = data.get('threshold', 0)
    percentage = (current_usage / threshold) * 100 if threshold else 0
    
    # Log alert
    print(f"[ALERT] Usage at {percentage:.1f}% of threshold")
    
    # Trigger notification
    alert_system.send_alert(
        title=f"📊 HolySheep Usage Alert: {percentage:.1f}%",
        message=f"Usage đã đạt {percentage:.1f}% của ngưỡng cho phép. "
                f"Current: {current_usage} tokens, Threshold: {threshold}",
        level=AlertLevel.HIGH if percentage > 80 else AlertLevel.MEDIUM,
        channels=["slack", "telegram"]
    )

def handle_key_compromised(event):
    """
    HolySheep thông báo key có thể đã bị compromise
    """
    data = event.get('data', {})
    ip_address = data.get('ip', 'Unknown')
    location = data.get('location', 'Unknown')
    timestamp = data.get('timestamp')
    
    print(f"[CRITICAL] Key compromised from IP: {ip_address}, Location: {location}")
    
    # Immediately revoke
    auto_revoke.revoke_key(
        data.get('key_hash'),
        f"Compromise detected from {location} at {timestamp}"
    )
    
    # Send critical alert
    alert_system.send_alert(
        title="🚨 CRITICAL: API Key Comprom

Tài nguyên liên quan
Bài viết liên quan