AI 安全监控：异常 API 调用パターン認識と自動ブロッキング実装ガイド

【結論】本ガイドでは、HolySheep AI の API を活用した AI 安全监控システムの構築方法を実践的に解説します。レート制限突破・異常リクエストパターン・不正アクセスをリアルタイムで検出し、自動的に API キーを無効化するシステムを提供，成本は公式比85%節約の¥1=$1で実装可能です。

HolySheep AI と主要APIサービスの比較

HolySheep AI は、中国本土の決済手段（WeChat Pay・Alipay）に対応しており、日本語対応サポートも提供するAsia-Pacific向けのAI APIゲートウェイです。登録すると無料クレジットが付与されるため、実際のプロジェクトでの動作検証而易く、低コストでセキュリティ監視システムを構築開始できます。

• Python 3.10以上
• Redis（リアルタイムカウンター用）
• SQLite（永続ログ用）
• HolySheep AI APIキー（ここから取得）

実装コード：基本监控システム

#!/usr/bin/env python3
"""
AI API 安全监控系统 - HolySheep AI 版
異常APIコールパターンのリアルタイム検出と自動ブロッキング
"""

import hashlib
import hmac
import time
import sqlite3
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from collections import defaultdict
import json
import threading
import redis

============================================================
設定セクション
============================================================

HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"

@dataclass
class SecurityConfig:
    """セキュリティ設定"""
    # レート制限（時間窓：秒）
    requests_per_minute: int = 60
    requests_per_hour: int = 1000
    
    # 異常検出閾値
    error_rate_threshold: float = 0.5  # 50%以上で異常判定
    avg_latency_multiplier: float = 3.0  # 平均レイテンシーの3倍以上
    suspicious_tokens_per_request: int = 100000  # 1リクエスト10万トークン以上
    
    # ブロック期間（秒）
    block_duration_short: int = 300  # 5分
    block_duration_long: int = 3600   # 1時間
    block_duration_permanent: int = 86400 * 30  # 30日


@dataclass
class APIRequest:
    """APIリクエスト記録"""
    request_id: str
    api_key: str
    endpoint: str
    timestamp: datetime
    tokens_used: int
    latency_ms: float
    status_code: int
    error_message: Optional[str] = None
    user_agent: str = ""
    ip_address: str = ""
    model: str = ""


@dataclass
class SecurityAlert:
    """セキュリティアラート"""
    alert_id: str
    alert_type: str  # "rate_limit", "anomaly", "block_detected"
    api_key: str
    severity: str  # "low", "medium", "high", "critical"
    description: str
    timestamp: datetime
    action_taken: str = ""


class DatabaseManager:
    """データベース管理"""
    
    def __init__(self, db_path: str = "ai_security.db"):
        self.db_path = db_path
        self._init_database()
    
    def _init_database(self):
        """データベーステーブルを初期化"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # APIリクエストログテーブル
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS api_request_logs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                request_id TEXT NOT NULL,
                api_key_hash TEXT NOT NULL,
                endpoint TEXT NOT NULL,
                timestamp TEXT NOT NULL,
                tokens_used INTEGER DEFAULT 0,
                latency_ms REAL DEFAULT 0,
                status_code INTEGER DEFAULT 200,
                error_message TEXT,
                user_agent TEXT,
                ip_address TEXT,
                model TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        
        # ブロックリストテーブル
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS blocked_keys (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                api_key_hash TEXT NOT NULL UNIQUE,
                blocked_at TEXT NOT NULL,
                expires_at TEXT,
                reason TEXT,
                severity TEXT,
                is_permanent INTEGER DEFAULT 0
            )
        """)
        
        # セキュリティアラートテーブル
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS security_alerts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                alert_id TEXT NOT NULL UNIQUE,
                alert_type TEXT NOT NULL,
                api_key_hash TEXT NOT NULL,
                severity TEXT NOT NULL,
                description TEXT,
                timestamp TEXT NOT NULL,
                action_taken TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        
        # インデックス作成
        cursor.execute("CREATE INDEX IF NOT EXISTS idx_logs_api_key ON api_request_logs(api_key_hash)")
        cursor.execute("CREATE INDEX IF NOT EXISTS idx_logs_timestamp ON api_request_logs(timestamp)")
        cursor.execute("CREATE INDEX IF NOT EXISTS idx_blocked_keys ON blocked_keys(api_key_hash)")
        
        conn.commit()
        conn.close()
    
    def log_request(self, request: APIRequest):
        """リクエストをログに記録"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        api_key_hash = hashlib.sha256(request.api_key.encode()).hexdigest()
        
        cursor.execute("""
            INSERT INTO api_request_logs 
            (request_id, api_key_hash, endpoint, timestamp, tokens_used, 
             latency_ms, status_code, error_message, user_agent, ip_address, model)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            request.request_id,
            api_key_hash,
            request.endpoint,
            request.timestamp.isoformat(),
            request.tokens_used,
            request.latency_ms,
            request.status_code,
            request.error_message,
            request.user_agent,
            request.ip_address,
            request.model
        ))
        
        conn.commit()
        conn.close()
    
    def block_api_key(self, api_key: str, reason: str, 
                      duration: int, severity: str, is_permanent: bool = False):
        """APIキーをブロック"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        api_key_hash = hashlib.sha256(api_key.encode()).hexdigest()
        blocked_at = datetime.now()
        expires_at = None if is_permanent else (blocked_at + timedelta(seconds=duration))
        
        cursor.execute("""
            INSERT OR REPLACE INTO blocked_keys 
            (api_key_hash, blocked_at, expires_at, reason, severity, is_permanent)
            VALUES (?, ?, ?, ?, ?, ?)
        """, (
            api_key_hash,
            blocked_at.isoformat(),
            expires_at.isoformat() if expires_at else None,
            reason,
            severity,
            1 if is_permanent else 0
        ))
        
        conn.commit()
        conn.close()
    
    def is_blocked(self, api_key: str) -> Tuple[bool, Optional[str]]:
        """APIキーがブロックされているかチェック"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        api_key_hash = hashlib.sha256(api_key.encode()).hexdigest()
        
        cursor.execute("""
            SELECT blocked_at, expires_at, reason, is_permanent
            FROM blocked_keys
            WHERE api_key_hash = ?
        """, (api_key_hash,))
        
        result = cursor.fetchone()
        conn.close()
        
        if not result:
            return False, None
        
        blocked_at = datetime.fromisoformat(result[0])
        expires_at = datetime.fromisoformat(result[2]) if result[1] else None
        is_permanent = bool(result[3])
        
        if is_permanent:
            return True, result[2]  # 恒久ブロック
        
        if expires_at and datetime.now() > expires_at:
            return False, None  # 期限切れ
        
        return True, result[2]  # アクティブなブロック


class RateLimitManager:
    """レート制限管理（Redis使用）"""
    
    def __init__(self, redis_host: str = "localhost", redis_port: int = 6379):
        try:
            self.redis = redis.Redis(
                host=redis_host,
                port=redis_port,
                decode_responses=True
            )
            self.redis.ping()
            self.use_redis = True
        except (redis.ConnectionError, redis.RedisError):
            # Redis接続失敗時はオンメモリに戻る
            self.use_redis = False
            self.counters: Dict[str, List[Tuple[float, int]]] = defaultdict(list)
            self.lock = threading.Lock()
    
    def check_rate_limit(self, api_key: str, config: SecurityConfig) -> Tuple[bool, str]:
        """
        レート制限をチェック
        戻り値: (許可フラグ, 理由)
        """
        now = time.time()
        minute_key = f"ratelimit:{api_key}:minute"
        hour_key = f"ratelimit:{api_key}:hour"
        
        if self.use_redis:
            return self._check_redis_rate_limit(
                now, minute_key, hour_key, 
                config.requests_per_minute, config.requests_per_hour
            )
        else:
            return self._check_memory_rate_limit(
                api_key, now, config.requests_per_minute, config.requests_per_hour
            )
    
    def _check_redis_rate_limit(self, now: float, minute_key: str, 
                                 hour_key: str, 
                                 rpm_limit: int, 
                                 rph_limit: int) -> Tuple[bool, str]:
        pipe = self.redis.pipeline()
        
        # 分間のクリーンアップとカウント
        pipe.zremrangebyscore(minute_key, 0, now - 60)
        pipe.zadd(minute_key, {str(now): now})
        pipe.zcard(minute_key)
        
        # 毎時のクリーンアップとカウント
        pipe.zremrangebyscore(hour_key, 0, now - 3600)
        pipe.zadd(hour_key, {str(now): now})
        pipe.zcard(hour_key)
        
        # TTL設定
        pipe.expire(minute_key, 120)
        pipe.expire(hour_key, 3700)
        
        results = pipe.execute()
        minute_count = results[2]
        hour_count = results[5]
        
        if minute_count > rpm_limit:
            return False, f"1分間リクエスト数上限超過: {minute_count}/{rpm_limit}"
        
        if hour_count > rph_limit:
            return False, f"1時間リクエスト数上限超過: {hour_count}/{rph_limit}"
        
        return True, "OK"
    
    def _check_memory_rate_limit(self, api_key: str, now: float,
                                  rpm_limit: int, rph_limit: int) -> Tuple[bool, str]:
        with self.lock:
            # 古いエントリを削除（分間）
            self.counters[api_key] = [
                (ts, cnt) for ts, cnt in self.counters[api_key]
                if now - ts < 3600  # 1時間以内のもののみ保持
            ]
            
            minute_count = sum(
                cnt for ts, cnt in self.counters[api_key]
                if now - ts < 60
            )
            
            hour_count = sum(
                cnt for ts, cnt in self.counters[api_key]
                if now - ts < 3600
            )
            
            if minute_count >= rpm_limit:
                return False, f"1分間リクエスト数上限超過: {minute_count}/{rpm_limit}"
            
            if hour_count >= rph_limit:
                return False, f"1時間リクエスト数上限超過: {hour_count}/{rph_limit}"
            
            # カウントを加算
            self.counters[api_key].append((now, 1))
            
            return True, "OK"


class AnomalyDetector:
    """異常検出エンジン"""
    
    def __init__(self, db_manager: DatabaseManager, config: SecurityConfig):
        self.db = db_manager
        self.config = config
    
    def analyze_pattern(self, api_key: str, recent_minutes: int = 15) -> List[SecurityAlert]:
        """
        最近のリクエストパターンを分析して異常を検出
        """
        alerts = []
        now = datetime.now()
        since = now - timedelta(minutes=recent_minutes)
        
        conn = sqlite3.connect(self.db.db_path)
        cursor = conn.cursor()
        
        api_key_hash = hashlib.sha256(api_key.encode()).hexdigest()
        
        # 過去N分間の全リクエストを取得
        cursor.execute("""
            SELECT timestamp, tokens_used, latency_ms, status_code, error_message, model
            FROM api_request_logs
            WHERE api_key_hash = ? AND timestamp >= ?
            ORDER BY timestamp DESC
        """, (api_key_hash, since.isoformat()))
        
        requests = cursor.fetchall()
        conn.close()
        
        if len(requests) < 5:
            return alerts  # データ不足
        
        # レイテンシ分析
        latencies = [r[2] for r in requests if r[2] > 0]
        if latencies:
            avg_latency = sum(latencies) / len(latencies)
            max_latency = max(latencies)
            
            if max_latency > avg_latency * self.config.avg_latency_multiplier:
                alerts.append(SecurityAlert(
                    alert_id=f"alert_{int(time.time())}_{api_key_hash[:8]}",
                    alert_type="anomaly",
                    api_key=api_key,
                    severity="medium",
                    description=f"異常レイテンシ検出: 最大{max_latency:.0f}ms (平均{avg_latency:.0f}msの{self.config.avg_latency_multiplier}倍以上)",
                    timestamp=now,
                    action_taken="監視強化"
                ))
        
        # エラー率分析
        error_count = sum(1 for r in requests if r[3] >= 400)
        error_rate = error_count / len(requests)
        
        if error_rate > self.config.error_rate_threshold:
            severity = "high" if error_rate > 0.7 else "medium"
            alerts.append(SecurityAlert(
                alert_id=f"alert_{int(time.time())}_{api_key_hash[:8]}",
                alert_type="anomaly",
                api_key=api_key,
                severity=severity,
                description=f"高エラー率検出: {error_rate*100:.1f}% ({error_count}/{len(requests)}件が4xx/5xx)",
                timestamp=now,
                action_taken="リクエスト検証中"
            ))
        
        # トークン使用量異常検出
        tokens_list = [r[1] for r in requests if r[1] > 0]
        if tokens_list:
            max_tokens = max(tokens_list)
            avg_tokens = sum(tokens_list) / len(tokens_list)
            
            if max_tokens > self.config.suspicious_tokens_per_request:
                alerts.append(SecurityAlert(
                    alert_id=f"alert_{int(time.time())}_{api_key_hash[:8]}",
                    alert_type="anomaly",
                    api_key=api_key,
                    severity="high",
                    description=f"異常トークン使用量: 最大{max_tokens:,}トークン (平均{avg_tokens:,.0f})",
                    timestamp=now,
                    action_taken="詳細调查中"
                ))
        
        # 短時間集中リクエスト検出
        timestamps = [datetime.fromisoformat(r[0]) for r in requests]
        if len(timestamps) >= 3:
            time_diffs = [(timestamps[i] - timestamps[i+1]).total_seconds() 
                          for i in range(len(timestamps)-1)]
            avg_diff = sum(time_diffs) / len(time_diffs)
            
            if avg_diff < 0.1:  # 100ms以下の平均間隔
                alerts.append(SecurityAlert(
                    alert_id=f"alert_{int(time.time())}_{api_key_hash[:8]}",
                    alert_type="rate_limit",
                    api_key=api_key,
                    severity="high",
                    description=f"短時間集中リクエスト: 平均{avg_diff*1000:.1f}ms間隔で{len(requests)}件のリクエスト",
                    timestamp=now,
                    action_taken="レート制限強化"
                ))
        
        return alerts


class HolySheepAPIClient:
    """HolySheep AI API クライアント"""
    
    def __init__(self, api_key: str, db_manager: DatabaseManager,
                 rate_limiter: RateLimitManager, 
                 anomaly_detector: AnomalyDetector,
                 config: SecurityConfig = None):
        self.api_key = api_key
        self.db = db_manager
        self.rate_limiter = rate_limiter
        self.anomaly_detector = anomaly_detector
        self.config = config or SecurityConfig()
        self.session = None
    
    def _init_session(self):
        """HTTPセッションを初期化"""
        import urllib.request
        import urllib.error
        self.session = urllib.request
        self._urlopen = self.session.urlopen
    
    def _make_request(self, endpoint: str, data: dict, 
                      timeout: int = 30) -> Tuple[dict, float]:
        """
        APIリクエストを実行し、リクエストをログに記録
        戻り値: (レスポンス辞書, レイテンシms)
        """
        import urllib.request
        import urllib.error
        import json
        
        if not self.session:
            self._init_session()
        
        url = f"{HOLYSHEEP_BASE_URL}/{endpoint}"
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "User-Agent": "HolySheep-SecurityMonitor/1.0"
        }
        
        body = json.dumps(data).encode("utf-8")
        req = urllib.request.Request(url, data=body, headers=headers, method="POST")
        
        start_time = time.time()
        
        try:
            with self.session.urlopen(req, timeout=timeout) as response:
                latency_ms = (time.time() - start_time) * 1000
                response_data = json.loads(response.read().decode("utf-8"))
                
                return {
                    "success": True,
                    "data": response_data,
                    "status_code": response.status
                }, latency_ms
                
        except urllib.error.HTTPError as e:
            latency_ms = (time.time() - start_time) * 1000
            error_body = e.read().decode("utf-8") if e.fp else ""
            
            return {
                "success": False,
                "error": str(e),
                "status_code": e.code,
                "error_body": error_body
            }, latency_ms
            
        except urllib.error.URLError as e:
            latency_ms = (time.time() - start_time) * 1000
関連リソース
📚 AI API 記事一覧
💰 料金を見る
📖 開発者ドキュメント
🚀 無料登録
関連記事
Claude Function Calling の制限と最佳実践：パラメータ数・ネスト階層の深掘り
HolySheep AI APIアクセスにおけるゼロトラストセキュリティの実装方法