ブログに戻る

Yandex障害を生き延びたボット — ゲームバックエンドアーキテクチャもそうあるべき

公開日 2026年10月20日
Yandex障害を生き延びたボット — ゲームバックエンドアーキテクチャもそうあるべき AIを活用して生成

要点まとめ

Yandex障害で悪意あるボットが他プロバイダーへ移行した実態をCloudflare分析から解説し、IPベース検出の限界と行動フィンガープリント・分散レート制限・マルチプロバイダー構成によるゲームバックエンド保護策を紹介する、障害に強いアーキテクチャを今すぐ構築するためのゲーム開発者向け実践ガイド。

モニタリングダッシュボードに、トラフィックが一晩で40%減少したと表示されている。あなたは安堵のため息をつく — DDoSは止まったはずだ。しかし、2023年10月のYandex障害に関するCloudflareの分析は、この想定が壊滅的に間違っていることを証明した。悪意のあるボットはYandexのデータセンターが停止しても死ななかったのだ。彼らは数分以内にOVH、Hetzner、DigitalOceanへ移行した。攻撃者が動き続けている間、あなたの集計メトリクスは嘘をついていた。

もしゲームバックエンドが単一のクラウドプロバイダーでIPベースのボット検出に依存しているなら、インフラ障害が1回発生するだけで同じブラインドスポットに陥る。このランブックでは、Yandex障害がインフラ障害時のボットの挙動について何を明らかにしたか、そしてそれらを生き延びるゲームバックエンドをどう設計するかを解説する。

何が壊れるのか: インフラプロバイダーが停止したとき

大手インフラプロバイダーで障害が発生すると、4つの重要なシステムが同時に故障する — そしてそれらは、素朴なモニタリングでは改善に見える形で故障する。

認証レイヤーがオフラインになる。 プレイヤー認証が単一プロバイダーのIDサービスに依存している場合、プレイヤーはログインできない。Yandex障害の間、Yandexインフラに強く結合されたサービスは完全なサービス中断に見舞われた。プレイヤーはエラー画面を見ることになり、認証しないボットは他のエンドポイントへの攻撃を続けた。

ボット検出の基準フレームが失われる。 ほとんどのゲームバックエンドは、ASNやホスティングプロバイダーで悪意のあるIPを追跡している。それらのIPが停止すると、検出システムはそれらを「解決済み」とマークして監視をやめる。一方、同じボット群は別のプロバイダーの新しいIPに再出現する。

レート制限が無意味になる。 IPベースのレート制限は特定のアドレス範囲に紐づいている。ボットが移行すると、クリーンなレピュテーションスコアを持つ新しいIPを取得する。レートリミッターはそれらを新しい正規トラフィックとして扱う。

トラフィック集計が誤情報になる。 CloudflareのYandex分析によると、AS13238 (Yandex Search) のトラフィックはほぼ全て正規のクローラーで構成されており、そこでの低下は本物だった。しかしAS200350 (Yandex.Cloud) では悪意のある脆弱性スキャンとプロービングが減少した一方、大量の自動化トラフィックは単に移転しただけだった。99%の低下を示す集計トラフィックチャートは、あなたに間違ったストーリーを伝えた。

核心的な障害モード: 脅威が移動しただけなのに、モニタリングは状況が改善したと思い込む。

検出方法: ボット移行をリアルタイムで見つける

検出には、集計トラフィック量だけでなく、複数レイヤーでのモニタリングが必要だ。各レイヤーが移行パターンの異なる側面を捉える。

レイヤー1: トラフィックソース相関

毎秒の総リクエスト数を追跡する代わりに、自律システム番号 (ASN) 全体でのトラフィックソースの分布を監視する。あるプロバイダーのトラフィックが低下したとき、他のプロバイダーのトラフィックが比例して増加しているか確認する。ASN間の相関した低下・上昇パターンは、ボット移行のシグネチャである。

以下はこのパターンを識別するPythonディテクターだ:

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

@dataclass
class TrafficSnapshot:
    timestamp: float
    source_asn: str
    request_count: int
    unique_ips: int
    error_rate: float

class BotMigrationDetector:
    """Detects correlated traffic drops and rises across ASNs
    to identify bot farm migration during infrastructure failures."""

    def __init__(self, migration_threshold: float = 0.3):
        self.history: Dict[str, List[TrafficSnapshot]] = defaultdict(list)
        self.migration_threshold = migration_threshold
        self.window_seconds = 300  # 5-minute rolling window

    def record_snapshot(self, snapshot: TrafficSnapshot):
        self.history[snapshot.source_asn].append(snapshot)
        self._prune_old_data()

    def _prune_old_data(self):
        cutoff = time.time() - self.window_seconds
        for asn in self.history:
            self.history[asn] = [
                s for s in self.history[asn]
                if s.timestamp > cutoff
            ]

    def detect_migration(self) -> Optional[Dict]:
        """Returns migration report if suspicious pattern detected."""
        declining_asns = []
        rising_asns = []

        for asn, snapshots in self.history.items():
            if len(snapshots) < 2:
                continue

            mid = len(snapshots) // 2
            first_half = snapshots[:mid]
            second_half = snapshots[mid:]

            avg_first = sum(s.request_count for s in first_half) / len(first_half)
            avg_second = sum(s.request_count for s in second_half) / len(second_half)

            if avg_first > 0:
                change_ratio = (avg_second - avg_first) / avg_first

                if change_ratio < -self.migration_threshold:
                    declining_asns.append({
                        'asn': asn,
                        'drop_pct': abs(change_ratio) * 100,
                        'baseline': avg_first
                    })
                elif change_ratio > self.migration_threshold:
                    rising_asns.append({
                        'asn': asn,
                        'rise_pct': change_ratio * 100,
                        'current': avg_second
                    })

        if declining_asns and rising_asns:
            return {
                'type': 'bot_migration',
                'declining': declining_asns,
                'rising': rising_asns,
                'confidence': self._calculate_confidence(declining_asns, rising_asns)
            }
        return None

    def _calculate_confidence(self, declining, rising) -> float:
        """Higher confidence when drop and rise magnitudes match."""
        total_drop = sum(d['drop_pct'] for d in declining)
        total_rise = sum(r['rise_pct'] for r in rising)

        if total_drop == 0:
            return 0.0

        ratio = min(total_rise, total_drop) / max(total_rise, total_drop)
        return round(ratio * 100, 2)


# Usage: detect Yandex-style migration
detector = BotMigrationDetector(migration_threshold=0.25)

# Yandex traffic collapsing
detector.record_snapshot(TrafficSnapshot(
    timestamp=time.time() - 240,
    source_asn="AS13238", request_count=15000,
    unique_ips=8500, error_rate=0.02
))
detector.record_snapshot(TrafficSnapshot(
    timestamp=time.time() - 120,
    source_asn="AS13238", request_count=1200,  # 92% drop
    unique_ips=900, error_rate=0.45
))

# OVH traffic surging (bots migrated here)
detector.record_snapshot(TrafficSnapshot(
    timestamp=time.time() - 240,
    source_asn="AS16276", request_count=5000,
    unique_ips=3200, error_rate=0.01
))
detector.record_snapshot(TrafficSnapshot(
    timestamp=time.time() - 120,
    source_asn="AS16276", request_count=18500,  # 270% increase
    unique_ips=2800, error_rate=0.08
))

migration = detector.detect_migration()
if migration:
    print(f"Bot migration detected: {migration['confidence']}% confidence")
    for d in migration['declining']:
        print(f"  {d['asn']}: -{d['drop_pct']:.0f}%")
    for r in migration['rising']:
        print(f"  {r['asn']}: +{r['rise_pct']:.0f}%")

このディテクターは、複数のASNにわたる相関したトラフィックの低下と上昇を監視する。AS13238が92%低下し、AS16276が同時に270%上昇した場合、正規のトラフィック変化ではなくボット移行の可能性が高いとフラグを立てる。低下と上昇の規模が一致するとき、信頼度スコアが上昇する — これは協調的な移行の特徴である。

レイヤー2: 行動フィンガープリンティング

トラフィックの発生源だけでなく、その挙動も追跡する。プロバイダー間を移行するボットは、同一の行動シグネチャを保持している: 同じリクエスト間隔、ヘッダーパターン、セッション時間、エラーリトライ動作。IPレピュテーションは移行時にリセットされるが、行動はリセットされない。

各セッションの行動フィンガープリントを構築し、ハッシュ化する:

import hashlib
import statistics

def fingerprint_session(requests: list[dict]) -> str:
    """Generate a behavioral fingerprint from a session's request pattern."""
    intervals = []
    for i in range(1, len(requests)):
        delta = requests[i]['timestamp'] - requests[i-1]['timestamp']
        intervals.append(round(delta, 3))

    if len(intervals) < 2:
        return "insufficient_data"

    # Key behavioral features
    interval_mean = round(statistics.mean(intervals), 3)
    interval_stdev = round(statistics.stdev(intervals), 3)
    total_duration = round(requests[-1]['timestamp'] - requests[0]['timestamp'], 3)
    error_count = sum(1 for r in requests if r.get('status_code', 200) >= 400)
    retry_count = sum(1 for r in requests if r.get('is_retry', False))
    unique_paths = len(set(r['path'] for r in requests))
    user_agent_count = len(set(r.get('user_agent', '') for r in requests))

    fingerprint_data = (
        f"{interval_mean}|{interval_stdev}|{total_duration}|"
        f"{error_count}|{retry_count}|{unique_paths}|{user_agent_count}"
    )

    return hashlib.sha256(fingerprint_data.encode()).hexdigest()[:16]


# Example: same bot session detected at two different providers
session_provider_a = [
    {'timestamp': 1000.0, 'path': '/api/login', 'status_code': 200},
    {'timestamp': 1000.5, 'path': '/api/profile', 'status_code': 200},
    {'timestamp': 1001.0, 'path': '/api/match/join', 'status_code': 429},
    {'timestamp': 1001.5, 'path': '/api/match/join', 'status_code': 200, 'is_retry': True},
    {'timestamp': 1002.0, 'path': '/api/score', 'status_code': 200},
]

session_provider_b = [
    {'timestamp': 2000.0, 'path': '/api/login', 'status_code': 200},
    {'timestamp': 2000.5, 'path': '/api/profile', 'status_code': 200},
    {'timestamp': 2001.0, 'path': '/api/match/join', 'status_code': 429},
    {'timestamp': 2001.5, 'path': '/api/match/join', 'status_code': 200, 'is_retry': True},
    {'timestamp': 2002.0, 'path': '/api/score', 'status_code': 200},
]

fp_a = fingerprint_session(session_provider_a)
fp_b = fingerprint_session(session_provider_b)

print(f"Provider A fingerprint: {fp_a}")
print(f"Provider B fingerprint: {fp_b}")
print(f"Same bot: {fp_a == fp_b}")

元のプロバイダーが停止した直後に、同じ行動フィンガープリントが別のプロバイダーの新しいIPで出現するのを確認したら、それは偶然ではなく、確認された移行である。

レイヤー3: 認証状態モニタリング

認証済みリクエストと未認証リクエストの比率を経時的に追跡する。インフラ障害時には、正規のプレイヤーは認証に苦労し(比率が未認証側にシフトする)、一方認証しないボットは変わらずエンドポイントにヒットし続ける。

このレイヤーは、発生源プロバイダーに関係なく自動トラフィックから実際のプレイヤーを分離するため、特に価値が高い。horizOnのアカウントバウンドセッションを複数の認証方法(メール、Google、Apple、匿名)で使用している場合、認証方法ごとの成功率を監視できる。メール認証が失敗しGoogle認証が成功する場合、それは部分的なインフラ障害であり、ボット移行ではない。すべての認証が成功する一方で、新しいプロバイダーで未認証トラフィックが急増する場合、それは移行である。

修復方法: 効果的なフェイルオーバー戦略

検出だけでは、ゲームがリアルタイムで死ぬのを眺めているに過ぎない。今週実装できる3つのフェイルオーバーメカニズムを紹介する。

DNSベースのフェイルオーバー

短いTTLとヘルスチェックを設定してDNSを構成する。プライマリリージョンが障害を起こすと、DNSは自動的にセカンダリリージョンへトラフィックをルーティングする。アプリケーションの変更は不要だ。

推奨設定:

  • TTL: 60〜120秒(短いほどフェイルオーバーが速いが、DNS負荷が増加する)
  • ヘルスチェック間隔: 10〜15秒
  • ヘルスチェックエンドポイント: /health が200を返し、レスポンスタイム500ms未満
  • フェイルオーバートリガー: ヘルスチェックが3回連続で失敗

トレードオフ: 短いTTLは通常運用中のDNSクエリ量を10〜20倍に増加させる。デイリーアクティブプレイヤー50,000人のゲームの場合、DNSコストは月額およそ$2〜5の追加となる — 防ぐことのできるダウンタイムに比べれば無視できる額だ。

アプリケーションレベルのサーキットブレーカー

ゲームクライアントにサーキットブレーカーを実装し、サービス障害を検出して自動的に代替ルートへ振り向ける:

using System;
using System.Net.Http;
using System.Threading.Tasks;

public class ServiceCircuitBreaker
{
    private readonly string _primaryEndpoint;
    private readonly string[] _fallbackEndpoints;
    private readonly int _failureThreshold;
    private int _failureCount;
    private DateTime _lastFailure;
    private CircuitState _state = CircuitState.Closed;
    private int _currentFallbackIndex;
    private readonly object _lock = new();

    public enum CircuitState { Closed, Open, HalfOpen }

    public ServiceCircuitBreaker(
        string primary, string[] fallbacks, int failureThreshold = 5)
    {
        _primaryEndpoint = primary;
        _fallbackEndpoints = fallbacks;
        _failureThreshold = failureThreshold;
    }

    public async Task<HttpResponseMessage> SendRequest(HttpRequestMessage request)
    {
        lock (_lock)
        {
            if (_state == CircuitState.Open)
            {
                if (DateTime.UtcNow - _lastFailure > TimeSpan.FromSeconds(30))
                    _state = CircuitState.HalfOpen;
                else
                    return SendToFallback(request).GetAwaiter().GetResult();
            }
        }

        try
        {
            using var client = new HttpClient();
            client.Timeout = TimeSpan.FromSeconds(5);
            var response = await client.SendAsync(request);

            if (response.IsSuccessStatusCode)
            {
                lock (_lock) { _failureCount = 0; _state = CircuitState.Closed; }
                return response;
            }

            RecordFailure();
            return await SendToFallback(request);
        }
        catch (TaskCanceledException)
        {
            RecordFailure();
            return await SendToFallback(request);
        }
    }

    private void RecordFailure()
    {
        lock (_lock)
        {
            _failureCount++;
            _lastFailure = DateTime.UtcNow;
            if (_failureCount >= _failureThreshold)
                _state = CircuitState.Open;
        }
    }

    private async Task<HttpResponseMessage> SendToFallback(HttpRequestMessage request)
    {
        for (int i = 0; i < _fallbackEndpoints.Length; i++)
        {
            int idx = (_currentFallbackIndex + i) % _fallbackEndpoints.Length;
            try
            {
                var fallbackRequest = new HttpRequestMessage(
                    request.Method,
                    _fallbackEndpoints[idx] + request.RequestUri.PathAndQuery
                );
                using var client = new HttpClient { Timeout = TimeSpan.FromSeconds(5) };
                var response = await client.SendAsync(fallbackRequest);
                if (response.IsSuccessStatusCode)
                {
                    _currentFallbackIndex = idx;
                    return response;
                }
            }
            catch { continue; }
        }

        throw new HttpRequestException("All endpoints failed");
    }
}

// Usage in your game client
var breaker = new ServiceCircuitBreaker(
    primary: "https://api-primary.game.example.com",
    fallbacks: new[] {
        "https://api-secondary.game.example.com",
        "https://api-tertiary.game.example.com"
    },
    failureThreshold: 3
);

var request = new HttpRequestMessage(HttpMethod.Get, "/api/leaderboard/global");
var response = await breaker.SendRequest(request);

このパターンは、あるプロバイダーが停止したときのカスケード障害を防ぐ。サーバーフォールバックの設計の分析で説明したように、自動フェイルオーバーは30秒の一時的な問題と4時間の障害を分ける違いである。

分散レート制限

トラフィックが新しいリージョンにフェイルオーバーするとき、レート制限もそこで機能する必要がある。各サーバーにローカルなIPベースのレート制限は、フェイルオーバー時に同じボットがすべてのエンドポイントでクリーンな状態を得るため、無意味になる。

全リージョンからアクセス可能な共有ストアを使用して分散レート制限を実装する:

import redis
import time

class DistributedRateLimiter:
    """Sliding window rate limiter backed by Redis.
    Works identically across all deployment regions."""

    def __init__(self, redis_url: str, max_requests: int = 100,
                 window_seconds: int = 60):
        self.redis = redis.from_url(redis_url)
        self.max_requests = max_requests
        self.window = window_seconds

    def is_allowed(self, identifier: str) -> tuple[bool, int]:
        """Returns (allowed, remaining_requests)."""
        key = f"ratelimit:{identifier}"
        now = time.time()
        window_start = now - self.window

        pipe = self.redis.pipeline()
        pipe.zremrangebyscore(key, 0, window_start)  # Remove expired entries
        pipe.zadd(key, {f"{now}:{id(identifier)}": now})  # Add current request
        pipe.zcard(key)  # Count requests in window
        pipe.expire(key, self.window)
        results = pipe.execute()

        request_count = results[2]
        remaining = max(0, self.max_requests - request_count)

        if request_count > self.max_requests:
            # Remove the request we just added — it's over limit
            pipe.zremrangebyscore(key, now, now)
            pipe.execute()
            return False, 0

        return True, remaining


# Usage: same limiter works from any region
limiter = DistributedRateLimiter(
    redis_url="redis://shared-redis.internal:6379",
    max_requests=60,   # 1 request per second average
    window_seconds=60
)

# Identify by player ID, not IP — survives migration
allowed, remaining = limiter.is_allowed("player:8f3a2b1c")

識別子がIPアドレスではなくプレイヤーIDを使用していることに注目してほしい。新しいIPに移行したボットも同じプレイヤーフィンガープリントを保持している。このアプローチは、プロバイダーレベルの移行後もボットを捕捉する。

予防方法: レジリエンスのためのアーキテクチャパターン

修復は反応的だ。予防とは、インフラ障害が発生することを前提としたアーキテクチャを構築することだ — なぜなら発生するからだ。

パターン1: 重要サービスのマルチプロバイダー展開

認証、マッチメイキング、ボット検出を少なくとも2つのクラウドプロバイダーに展開する。これはスタック全体を2回実行するという意味ではない — 障害を生き延びなければならないサービスをプロバイダー非依存にするという意味だ。

コストの現実: 重要サービスのマルチプロバイダー展開は、通常インフラ請求額に30〜50%追加される。10,000人の同時プレイヤーを抱えるゲームバックエンドの場合、月額およそ$800〜1,500の追加だ。同じゲームの2時間の障害コスト — IAP収益の損失、プレイヤーの離脱、サポートチケットを考慮すると — 平均$5,000〜12,000になる。

最小限のマルチプロバイダー構成:

  • 認証: プライマリ + セカンダリプロバイダー
  • プレイヤーデータ / セーブ状態: プロバイダー間でレプリケーション
  • マッチメイキング: 少なくとも1つのフォールバックリージョン
  • リーダーボードと分析: 短時間の障害は許容可能(キューに入れてリプレイ)

パターン2: ステートレスでプロバイダー非依存のボット検出

トラフィックの発生源に関係なく、同じように機能するボット検出を設計する。Yandex障害は、IPベースの検出が根本的に脆弱であることを証明した — ボットは移行した瞬間に新しいIPを取得する。

インフラ変更を生き延びる行動特徴を中心に検出を構築する:

from dataclasses import dataclass
from typing import List

@dataclass
class SessionFeatures:
    request_count: int
    duration_seconds: float
    interval_variance: float      # Low variance = bot-like regularity
    error_retry_rate: float       # Bots retry aggressively; humans quit
    unique_endpoints: int         # Bots hit predictable endpoint patterns
    header_consistency: float     # Bots reuse identical headers

class BehavioralClassifier:
    """Bot classifier based on behavioral features, not IP addresses.
    Survives provider migration because behavior doesn't change."""

    # Weights calibrated from production bot traffic analysis
    WEIGHTS = {
        'interval_variance': 0.30,   # Most discriminating feature
        'error_retry_rate': 0.25,
        'header_consistency': 0.20,
        'endpoint_diversity': 0.15,
        'session_duration': 0.10,
    }

    def classify(self, features: SessionFeatures) -> float:
        """Returns bot probability from 0.0 (human) to 1.0 (bot)."""
        score = 0.0

        # Low interval variance → bot (humans are irregular)
        if features.interval_variance < 0.1:
            score += self.WEIGHTS['interval_variance']
        elif features.interval_variance < 0.3:
            score += self.WEIGHTS['interval_variance'] * 0.5

        # High retry rate → bot (humans give up after 1-2 failures)
        if features.error_retry_rate > 0.8:
            score += self.WEIGHTS['error_retry_rate']
        elif features.error_retry_rate > 0.5:
            score += self.WEIGHTS['error_retry_rate'] * 0.5

        # Perfect header consistency → bot (humans have browser variations)
        if features.header_consistency > 0.95:
            score += self.WEIGHTS['header_consistency']

        # Very few unique endpoints → bot (humans explore)
        if features.unique_endpoints <= 3 and features.request_count > 20:
            score += self.WEIGHTS['endpoint_diversity']

        # Suspiciously precise session duration (e.g., exactly 300s)
        if abs(features.duration_seconds % 60) < 2:
            score += self.WEIGHTS['session_duration']

        return min(score, 1.0)


# Usage
classifier = BehavioralClassifier()

bot_session = SessionFeatures(
    request_count=150, duration_seconds=300.0,
    interval_variance=0.02, error_retry_rate=0.95,
    unique_endpoints=2, header_consistency=0.99
)

human_session = SessionFeatures(
    request_count=45, duration_seconds=1847.3,
    interval_variance=0.67, error_retry_rate=0.15,
    unique_endpoints=12, header_consistency=0.73
)

print(f"Bot score: {classifier.classify(bot_session):.2f}")   # ~0.85
print(f"Human score: {classifier.classify(human_session):.2f}")  # ~0.10

この分類器はリクエストがどのASNから来るかを気にしない。ボットがYandex.CloudからOVHに移行しても、その行動フィンガープリントは変わらない — そして分類器は依然として捕捉する。

パターン3: 認証をインフラから切り離す

認証システムは、単一のインフラ障害を生き延びるべきだ。複数の認証プロバイダーを使用して、プレイヤーがアカウントにアクセスするためのフォールバックオプションを提供する。

グレースフルデグラデーションを備えた認証を実装する:

class MultiProviderAuth:
    """Authentication with automatic provider fallback."""

    def __init__(self, providers: list):
        # Order by reliability — primary first
        self.providers = providers

    async def authenticate(self, credentials: dict) -> dict:
        last_error = None

        for provider in self.providers:
            try:
                result = await provider.authenticate(credentials)
                return {
                    'success': True,
                    'provider': provider.name,
                    'player_id': result['player_id'],
                    'session_token': result['session_token']
                }
            except Exception as e:
                last_error = e
                continue

        return {
            'success': False,
            'error': 'All authentication providers failed',
            'last_error': str(last_error)
        }

このパターンが重要なのは、Yandex障害の間、YandexのIDインフラに強く結合されたサービスはユーザーをまったく認証できなかったからだ。プレイヤーはロックアウトされ、認証しないボットは動き続けた。マルチプロバイダー認証戦略は、プライマリ認証プロバイダーが停止してもプレイヤーがプレイを続けられることを意味する。

パターン4: プロバイダーを横断するオブザーバビリティ

モニタリングはすべてのプロバイダーを同時に見渡せる必要がある。モニタリングを個別の関心事に分割する:

トラフィック分布モニタリング: ASNごと、リージョンごと、プロバイダーごとのリクエスト量を追跡する。5分間のウィンドウで分布が20%以上シフトしたらアラートを出す。

認証ヘルスモニタリング: 認証方法ごと、プロバイダーごとの認証成功率を追跡する。いずれかのプロバイダーの成功率が95%を下回ったらアラートを出す。

ボット検出モニタリング: プロバイダーごとのフラグ付きセッションと総セッションの比率を追跡する。あるプロバイダーで比率がベースラインを下回り、別のプロバイダーで上昇したらアラートを出す — それは移行だ。

クロスプロバイダー相関: これが重要なピースだ。個々のプロバイダーメトリクスは健全に見えても、システム全体が攻撃下にあることがある。全プロバイダーを並べて表示するダッシュボードを構築し、移行パターンを一目で発見できるようにする。

ベストプラクティス: ボット耐性のあるゲームバックエンドアーキテクチャへの5つのステップ

  1. 量だけでなく移行を監視する。 ある場所でトラフィックが低下し別の場所で上昇するのを検出するクロスプロバイダートラフィック相関を設定する。集計メトリクスだけではYandex障害時に騙されていただろう — AS200350のトラフィックは低下したが、悪意のある活動はOVHとHetznerで継続した。

  2. すべての外部依存関係にサーキットブレーカーを実装する。 ゲームクライアント、リーダーボードサービス、マッチメイキングロジックのすべてに自動フェイルオーバーが必要だ。サーキットブレーカーパターンは、あるプロバイダーが停止したときのカスケード障害を防ぐ。ステージングでエンドポイントを停止させて毎月テストする。

  3. IP検出ではなく行動検出を使用する。 行動分析(リクエスト間隔、リトライ率、ヘッダーパターン)は、ボットがどこでホストされていても機能する。IPベースの検出は、ボットがクリーンなIPレピュテーションを持つ新しいプロバイダーに移行した瞬間に失敗する。上記の行動分類器は、99%のトラフィック低下を生き延びたボットを捕捉する。

  4. グレースフルデグラデーションのために認証を設計する。 horizOnのアカウントバウンドセッションを複数の認証方法(メール、Google、Apple、匿名)で使用する。1つの認証プロバイダーが利用できない場合、プレイヤーは別の方法でゲーム状態へのアクセスを維持できる。これにより最悪のシナリオ — プレイヤーがロックアウトされ、ボットが動き続ける — を防ぐ。

  5. フェイルオーバーを四半期ごとにテストする。 ステージングでプロバイダー障害をシミュレートする。DNSフェイルオーバーがTTL内にトリガーされ、サーキットブレーカーが正しく開き、フォールバックリージョンでレート制限が機能することを確認する。障害検出から完全復旧までの時間を記録する。テストされていないフェイルオーバーはフェイルオーバーではない — それは希望だ。

結論

Yandex障害は、悪意のある攻撃者がほとんどのゲームバックエンドよりも優れたインフラレジリエンスを持っていることを証明した。Yandexのデータセンターが停止してもボットは死ななかった — 彼らは数分以内に他のプロバイダーへ移行し、動き続けた。あなたの集計トラフィックダッシュボードは改善を示した。現実は悪化していた。

まず、この記事の行動ボットディテクターをステージング環境にデプロイすることから始めよう。1週間実行してベースラインを確立し、次に1つのASNからのトラフィックをブロックしてプロバイダー障害をシミュレートし、移行ディテクターが作動するか観察する。発見したパターンは、現在のアーキテクチャのどこに脆弱性があるかを正確に明らかにする。

インフラ障害を生き延びる認証とプレイヤーデータシステムを構築しているゲームスタジオ向けに、horizOnはリビジョン対応クラウドセーブを備えたアカウントバウンドセッションを提供している — 予期しないことが発生してもプレイヤーが進行状況を失わないように設計されたサービスだ。無料で試すか、APIドキュメントを参照して、アカウントバウンドセッションが既存のバックエンドとどのように統合されるかを確認してほしい。


出典: Yandex障害: データが示す、悪意あるボットは停止せず移動した