机器人挺过了Yandex宕机——你的游戏后端架构也应该如此

你的监控仪表盘显示流量一夜之间下降了40%。你松了一口气——DDoS攻击想必已经停止了。但Cloudflare对2023年10月Yandex宕机的分析证明，这个假设错得离谱：当Yandex数据中心宕机时，恶意机器人并没有消失。它们在几分钟内就迁移到了OVH、Hetzner和DigitalOcean。当攻击者继续运行时，你的聚合指标却欺骗了你。

如果你的游戏后端运行在单一的云提供商上，且依赖基于IP的机器人检测，那么一次基础设施故障事件就足以让你陷入同样的盲区。本运维手册将剖析Yandex宕机揭示了哪些基础设施故障期间的机器人行为，以及如何构建能够在其间存活的游戏后端架构。

故障剖析：当你的基础设施提供商停止服务

当大型基础设施提供商发生故障时，四个关键系统会同时失效——而且它们的失效方式在初级监控看来，反而像是情况变好了。

你的身份验证层离线。 如果你的玩家身份验证依赖单一提供商的身份服务，玩家将无法登录。在Yandex宕机期间，与Yandex基础设施紧密耦合的服务完全中断。玩家看到错误页面；而无需身份验证的机器人则继续在别处访问端点。

你的机器人检测失去参考系。 大多数游戏后端通过ASN或托管提供商追踪恶意IP。当这些IP失效时，你的检测系统会将其标记为“已解决”并停止追踪。与此同时，同一批机器人舰队以新的IP在另一家提供商重新出现。

你的速率限制形同虚设。 基于IP的速率限制与特定地址段绑定。当机器人迁移时，它们会获得全新且信誉干净的IP。你的速率限制器会将它们当作新的合法流量处理。

你的流量聚合数据变成误导信息。 Cloudflare的Yandex分析显示，AS13238（Yandex搜索）的流量几乎全部是合法爬虫——那里的下降是真实的。但AS200350（Yandex.Cloud）的恶意漏洞扫描和探测活动虽然减少了，高流量自动化程序却只是转移了位置。显示99%流量下降的聚合图表给你讲了一个错误的故事。

核心故障模式：威胁只是转移了，而你的监控却认为情况好转了。

检测方法：实时识别机器人迁移

检测需要在多个层面进行监控，而不仅仅是聚合流量规模。每个层面捕捉迁移模式的不同侧面。

第一层：流量来源关联

不要只是追踪每秒总请求数，而是要监控各自治系统编号（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%时，它会标记为可能的机器人迁移，而不是合法的流量变化。当下降与上升幅度相当时，置信度得分会增加——这正是协同迁移的标志。

第二层：行为指纹

不要只追踪流量来自哪里——还要追踪它的行为方式。在提供商之间迁移的机器人会保留完全相同的行为特征：相同的请求间隔、请求头模式、会话时长和错误重试行为。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上时，那就是确认的迁移——而非巧合。

第三层：认证状态监控

随时间追踪已认证与未认证请求的比例。在基础设施故障期间，合法玩家难以完成认证（比例向未认证倾斜），而从不认证的机器人则继续照常访问你的端点。

这一层特别有价值，因为它能将真实玩家与自动化流量区分开来，无论流量来自哪个提供商。如果你正在使用horizOn的账号绑定会话并结合多种认证方式（邮箱、Google、Apple、匿名），你可以监控每种方式的认证成功率。当邮箱认证失败但Google认证成功时，你识别出的是部分基础设施故障——而非机器人迁移。当所有认证都成功，但新提供商的未认证流量激增时，那才是迁移。

应对方法：有效的故障转移策略

只检测而不应对，等于眼睁睁看着游戏实时死去。以下是本周内即可实施的三种故障转移机制。

基于DNS的故障转移

为你的DNS配置较短的TTL和健康检查。当主区域发生故障时，DNS自动将流量路由到辅助区域。这无需任何应用改动。

推荐配置：

• TTL：60-120秒（更短意味着更快的故障转移，但DNS负载更高）
• 健康检查间隔：10-15秒
• 健康检查端点：/health返回200且响应时间<500ms
• 故障转移触发条件：连续3次健康检查失败

权衡：较短的TTL会使正常运行期间的DNS查询量增加10-20倍。对于一款拥有50,000日活跃玩家的游戏，这大约每月增加2-5美元的DNS成本——与它避免的停机时间相比，微不足道。

应用层熔断器

在你的游戏客户端中实现熔断器，以检测服务何时失败并自动路由到备用方案：

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")

注意标识符使用的是玩家ID，而非IP地址。迁移到新IP的机器人仍然携带相同的玩家指纹。即使在提供商级别的迁移之后，这种方法也能捕获它们。

预防方法：面向弹性的架构模式

应对是被动的。预防意味着构建一种假设基础设施故障必然发生的架构——因为它确实会发生。

模式一：关键服务的多提供商部署

至少在两个云提供商之间部署你的认证、matchmaking和机器人检测服务。这并不意味着将整个技术栈运行两遍——而是指必须在故障中存活的关键服务要与提供商解耦。

成本现实：多提供商部署关键服务通常会使基础设施账单增加30-50%。对于支持10,000名并发玩家的游戏后端，这大约每月增加800-1,500美元。而同一款游戏遭遇2小时宕机的成本——计入IAP收入损失、玩家流失和支持工单——平均为5,000-12,000美元。

最低可行的多提供商方案：

• 认证：主提供商 + 备用提供商
• 玩家数据/存档状态：跨提供商复制
• Matchmaking：至少一个备用区域
• 排行榜和分析：可以容忍短暂中断（排队并重放）

模式二：无状态、与提供商无关的机器人检测

设计你的机器人检测，使其无论流量来自何处都以相同方式工作。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时，它的行为指纹保持不变——分类器依然能捕获它。

模式三：将认证与基础设施解耦

你的认证系统应该能够在任何单一基础设施故障中存活。使用多个认证提供商，这样玩家就有访问其账号的备用选项。

实现带优雅降级机制的认证：

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身份基础设施紧密耦合的服务完全无法认证用户。玩家被锁在门外；而不需要认证的机器人却继续运行。多提供商认证策略意味着，即使你的主认证提供商宕机，你的玩家也能继续游戏。

模式四：跨提供商的全局可观测性

你的监控系统需要同时覆盖所有提供商。将监控拆分为不同的关注点：

流量分布监控： 追踪每个ASN、每个区域、每个提供商的请求量。当分布状况在5分钟窗口内偏移超过20%时发出告警。

认证健康监控： 追踪每种认证方式和每个提供商的认证成功率。当任何提供商的成功率降至95%以下时发出告警。

机器人检测监控： 追踪每个提供商被标记会话与总会话的比例。当一个提供商的该比例降至基线以下，而另一个提供商同时上升时发出告警——这就是迁移。

跨提供商关联： 这是关键环节。单个提供商的指标可能看起来健康，而整个系统却在遭受攻击。构建并排展示所有提供商的仪表盘，让你一眼就能发现迁移模式。

最佳实践：构建抗机器人攻击游戏后端架构的5个步骤

1. 监控迁移，而不仅仅是流量规模。 建立跨提供商流量关联，检测一个地方流量下降而另一个地方上升的情况。在Yandex宕机期间，仅靠聚合指标会让你误判——AS200350的流量确实下降了，但恶意活动在OVH和Hetzner仍在继续。

2. 在每一个外部依赖上实现熔断器。 你的游戏客户端、排行榜服务和matchmaking逻辑都需要自动故障转移。熔断器模式可以防止某个提供商宕机时出现级联故障。每月在staging环境中通过关闭端点来测试它们。

3. 使用行为检测，而非IP检测。 行为分析（请求间隔、重试率、请求头模式）无论机器人托管在哪里都有效。基于IP的检测在机器人迁移到具有干净IP信誉的新提供商的那一刻就会失效。上面提到的行为分类器可以捕获在99%流量下降中存活的机器人。

4. 设计支持优雅降级的认证。 使用horizOn的账号绑定会话并支持多种认证方式（邮箱、Google、Apple、匿名）。当一个认证提供商不可用时，玩家可以使用另一种方式来维持对游戏状态的访问。这避免了最坏的情况：玩家被锁在门外，而机器人继续运行。

5. 每季度测试一次故障转移。 在staging环境中模拟提供商故障。验证DNS故障转移是否在TTL内触发、熔断器是否正确打开、速率限制在备用区域是否生效。记录从故障检测到完全恢复的时间。未经测试的故障转移不是故障转移——那只是希望。

结论

Yandex宕机证明，恶意行为者的基础设施弹性比大多数游戏后端更强。当Yandex数据中心宕机时，机器人并没有消失——它们在几分钟内迁移到其他提供商并继续运行。你的聚合流量仪表盘显示情况有所改善。而现实却变得更糟。

首先将本文中的行为机器人检测器部署到你的staging环境。运行一周以建立基线数据，然后通过屏蔽一个ASN的流量来模拟提供商故障，观察迁移检测器是否触发。你发现的模式将准确揭示当前架构的脆弱之处。

对于正在构建能够承受基础设施故障的认证和玩家数据系统的游戏工作室，horizOn提供了带版本感知云存档的账号绑定会话——这些服务确保即使在意外发生时，玩家也能保留游戏进度。免费试用，或浏览API文档，了解账号绑定会话如何与你现有的后端集成。

---

来源：Yandex宕机：数据显示恶意机器人并未消失，而是转移了阵地