البوتات نجت من انقطاع ياندكس — وبنية الواجهة الخلفية للعبتك يجب أن تنجو أيضًا

لوحة المراقبة لديك تُظهر انخفاض حركة المرور بنسبة 40% بين عشية وضحاها. تتنفس الصعداء — لا بد أن هجوم DDoS قد توقف. لكن تحليل Cloudflare لانقطاع ياندكس في أكتوبر 2023 أثبت أن هذا الافتراض خاطئ بشكل كارثي: البوتات الخبيثة لم تمت عندما أظلمت مراكز بيانات ياندكس. لقد هاجرت إلى OVH وHetzner وDigitalOcean في غضون دقائق. مقاييسك الإجمالية كذبت عليك بينما استمر المهاجمون في العمل.

إذا كانت الواجهة الخلفية للعبتك تعمل على مزود سحابي واحد مع كشف بوتات يعتمد على عنوان IP، فأنت على بُعد حدث بنية تحتية واحد من نفس النقطة العمياء. يشرح هذا الدليل ما يكشفه انقطاع ياندكس عن سلوك البوتات أثناء فشل البنية التحتية، وكيفية تصميم واجهات خلفية للألعاب تنجو من هذه الأعطال.

ما الذي يتعطل: عندما ينطفئ مزود البنية التحتية لديك

عندما يتعرض مزود بنية تحتية رئيسي لانقطاع، تفشل أربعة أنظمة حرجة في الوقت نفسه — وتفشل بطرق تبدو وكأنها تحسّن للمراقبة الساذجة.

طبقة المصادقة لديك تنقطع عن العمل. إذا كانت مصادقة اللاعبين تعتمد على خدمة هوية مزود واحد، فلن يتمكن اللاعبون من تسجيل الدخول. أثناء انقطاع ياندكس، شهدت الخدمات المرتبطة ارتباطًا وثيقًا ببنية ياندكس التحتية انقطاعًا كاملًا للخدمة. حصل اللاعبون على شاشات خطأ؛ بينما واصلت البوتات التي لا تُجري مصادقة ضرب نقاط النهاية في مكان آخر.

كشف البوتات لديك يفقد إطاره المرجعي. معظم الواجهات الخلفية للألعاب تتعقب عناوين IP الخبيثة حسب ASN أو مزود الاستضافة. عندما تختفي هذه العناوين، تعلن أنظمة الكشف أنها "تم حلها" وتتوقف عن البحث. في الوقت نفسه، يظهر أسطول البوتات نفسه على عناوين IP جديدة لدى مزود مختلف.

تحديد معدل الطلبات لديك يصبح عديم الفائدة. حدود المعدل المعتمدة على IP مرتبطة بنطاقات عناوين محددة. عندما تهاجر البوتات، تحصل على عناوين IP جديدة بسمعة نظيفة. يعاملها محدد المعدل لديك كحركة مرور شرعية جديدة.

إجماليات حركة المرور لديك تتحول إلى معلومات مضللة. أظهر تحليل Cloudflare لانقطاع ياندكس أن حركة مرور AS13238 (بحث ياندكس) كانت تتكون بالكامل تقريبًا من زاحفات شرعية — الانخفاض هناك كان حقيقيًا. لكن AS200350 (Yandex.Cloud) شهد انخفاضًا في فحص الثغرات الخبيثة والاستكشاف بينما انتقلت الأتمتة عالية الحجم ببساطة إلى مكان آخر. رسوم بيانية لإجمالي حركة المرور تُظهر انخفاضًا بنسبة 99% أخبرتك بالقصة الخاطئة.

نمط الفشل الأساسي: مراقبتك تعتقد أن الأمور تحسّنت بينما التهديد انتقل للتو.

كيفية الكشف: رصد هجرة البوتات في الوقت الفعلي

يتطلب الكشف مراقبة على مستويات متعددة، وليس فقط حجم حركة المرور الإجمالي. كل مستوى يلتقط جانبًا مختلفًا من نمط الهجرة.

المستوى 1: ارتباط مصدر حركة المرور

بدلاً من تتبع إجمالي الطلبات في الثانية، راقب توزيع مصادر حركة المرور عبر أرقام الأنظمة المستقلة (ASNs). عندما تنخفض حركة مرور أحد المزودين، تحقق مما إذا كانت حركة مرور المزودين الآخرين تزداد بشكل متناسب. نمط الانخفاض والارتفاع المترابط عبر ASNs هو بصمة هجرة البوتات.

إليك كاشف 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}%")

يراقب هذا الكاشف الانخفاضات والارتفاعات المترابطة في حركة المرور عبر ASNs متعددة. عندما ينخفض 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 تنجح، تكون قد حددت فشلًا جزئيًا في البنية التحتية — وليس هجرة بوتات. عندما تنجح جميع عمليات المصادقة ولكن حركة المرور غير الموثّقة ترتفع بشكل مفاجئ لدى مزود جديد، فهذه هجرة.

كيفية المعالجة: استراتيجيات تجاوز الفشل التي تعمل

الكشف دون معالجة هو مجرد مشاهدة لعبتك تموت في الوقت الفعلي. إليك ثلاث آليات لتجاوز الفشل يمكنك تنفيذها هذا الأسبوع.

تجاوز الفشل المعتمد على DNS

اضبط DNS الخاص بك مع فترات TTL قصيرة وفحوصات صحة. عندما تفشل منطقتك الأساسية، يوجّه DNS حركة المرور تلقائيًا إلى المناطق الثانوية. لا يتطلب هذا أي تغييرات في التطبيق.

الإعدادات الموصى بها:

• TTL: 60-120 ثانية (الأقصر يعني تجاوز فشل أسرع، لكن حملاً أكبر على DNS)
• فاصل فحص الصحة: 10-15 ثانية
• نقطة نهاية فحص الصحة: /health تُرجع 200 مع زمن استجابة أقل من 500ms
• مشغّل تجاوز الفشل: 3 إخفاقات متتالية في فحص الصحة

المقايضة: فترات TTL القصيرة تزيد حجم استعلامات DNS بمقدار 10-20x أثناء التشغيل العادي. للعبة بها 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")

لاحظ أن المعرّف يستخدم معرف اللاعب، وليس عنوان IP. البوتات التي تهاجر إلى عناوين IP جديدة لا تزال تحمل نفس بصمة اللاعب. هذا النهج يلتقطها حتى بعد الهجرة على مستوى المزود.

كيفية الوقاية: أنماط معمارية للمرونة

المعالجة رد فعلية. الوقاية تعني بناء بنية تفترض أن فشل البنية التحتية سيحدث — لأنه سيحدث.

النمط 1: نشر متعدد المزودين للخدمات الحرجة

انشر المصادقة، ومطابقة اللاعبين، وكشف البوتات عبر مزودي سحابة على الأقل. هذا لا يعني تشغيل مجموعتك التقنية بالكامل مرتين — بل يعني أن الخدمات التي يجب أن تنجو من الانقطاع تكون مستقلة عن المزود.

واقع التكلفة: النشر متعدد المزودين للخدمات الحرجة يضيف عادةً 30-50% إلى فاتورة البنية التحتية لديك. لواجهة خلفية للعبة تخدم 10,000 لاعب متزامن، يبلغ ذلك حوالي 800-1,500 دولار إضافية شهريًا. تكلفة انقطاع لمدة ساعتين لنفس اللعبة — مع احتساب خسارة إيرادات المشتريات داخل التطبيق، وفقدان اللاعبين، وتذاكر الدعم — يبلغ متوسطها 5,000-12,000 دولار.

الحد الأدنى القابل للتطبيق للنشر متعدد المزودين:

• المصادقة: مزود أساسي + ثانوي
• بيانات اللاعب / حالة الحفظ: منسوخة عبر المزودين
• مطابقة اللاعبين: منطقة احتياطية واحدة على الأقل
• لوحات الصدارة والتحليلات: يمكنها تحمل انقطاعات قصيرة (طابور وإعادة تشغيل)

النمط 2: كشف بوتات عديم الحالة ومستقل عن المزود

صمّم كشف البوتات لديك ليعمل بشكل متطابق بغض النظر عن مصدر حركة المرور. أثبت انقطاع ياندكس أن الكشف المعتمد على 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)
        }

هذا النمط مهم لأنه أثناء انقطاع ياندكس، لم تتمكن الخدمات المرتبطة ارتباطًا وثيقًا ببنية هوية ياندكس من مصادقة المستخدمين إطلاقًا. تم إغلاق الوصول أمام اللاعبين؛ بينما استمرت البوتات التي لا تُجري مصادقة في العمل. استراتيجية مصادقة متعددة المزودين تعني أن لاعبيك يستمرون في اللعب حتى عندما يكون مزود المصادقة الأساسي لديك معطلاً.

النمط 4: مراقبة تمتد عبر المزودين

تحتاج مراقبتك إلى الرؤية عبر جميع المزودين في الوقت نفسه. قسّم المراقبة إلى اهتمامات منفصلة:

مراقبة توزيع حركة المرور: تتبع حجم الطلبات لكل ASN، ولكل منطقة، ولكل مزود. أطلق تنبيهًا عندما يتغير التوزيع بأكثر من 20% في نافذة مدتها 5 دقائق.

مراقبة صحة المصادقة: تتبع معدلات نجاح المصادقة لكل طريقة ولكل مزود. أطلق تنبيهًا عندما ينخفض معدل نجاح أي مزود عن 95%.

مراقبة كشف البوتات: تتبع نسبة الجلسات المُعلَّمة إلى إجمالي الجلسات لكل مزود. أطلق تنبيهًا عندما تنخفض النسبة عن خط الأساس لدى مزود واحد بينما ترتفع لدى مزود آخر — فهذه هجرة.

الارتباط عبر المزودين: هذه هي القطعة الحرجة. يمكن أن تبدو مقاييس المزود الفردي سليمة بينما النظام بأكمله تحت الهجوم. أنشئ لوحات معلومات تعرض جميع المزودين جنبًا إلى جنب بحيث يمكنك رصد أنماط الهجرة في لمحة.

أفضل الممارسات: 5 خطوات لبنية واجهة خلفية للعبة مقاومة للبوتات

1. راقب الهجرة، وليس الحجم فقط. أعدّ ارتباط حركة مرور عبر المزودين لاكتشاف انخفاض حركة المرور في مكان وارتفاعها في مكان آخر. المقاييس الإجمالية وحدها كانت ستخدعك أثناء انقطاع ياندكس — انخفضت حركة مرور AS200350، لكن النشاط الخبيث استمر لدى OVH وHetzner.

2. نفّذ قواطع دارة على كل اعتماد خارجي. عميل اللعبة، وخدمة لوحة الصدارة، ومنطق مطابقة اللاعبين جميعها تحتاج إلى تجاوز فشل تلقائي. نمط قاطع الدارة يمنع الفشل المتتالي عندما يتعطل أحد المزودين. اختبرها شهريًا بتعطيل نقاط النهاية في بيئة الاختبار.

3. استخدم الكشف السلوكي، وليس كشف IP. التحليل السلوكي (فترات الطلبات، ومعدلات إعادة المحاولة، وأنماط الترويسات) يعمل بغض النظر عن مكان استضافة البوتات. الكشف المعتمد على IP يفشل في اللحظة التي تهاجر فيها البوتات إلى مزود جديد بسمعة IP نظيفة. المصنّف السلوكي أعلاه يلتقط البوتات التي تنجو من انخفاضات حركة المرور بنسبة 99%.

4. صمّم المصادقة لتدهور سلس. استخدم جلسات horizOn المرتبطة بالحساب مع طرق مصادقة متعددة (البريد الإلكتروني، Google، Apple، مجهول). عندما يكون مزود مصادقة واحد غير متاح، يمكن للاعبين استخدام طريقة أخرى للحفاظ على الوصول إلى حالة لعبتهم. يمنع هذا السيناريو الأسوأ: حرمان اللاعبين من الدخول بينما تستمر البوتات في العمل.

5. اختبر تجاوز الفشل لديك كل ثلاثة أشهر. حاكِ انقطاعات المزودين في بيئة الاختبار. تحقق من أن تجاوز فشل DNS يعمل ضمن TTL الخاص بك، وأن قواطع الدارة تفتح بشكل صحيح، وأن تحديد معدل الطلبات يعمل في المناطق الاحتياطية. وثّق الوقت من اكتشاف الفشل إلى الاسترداد الكامل. تجاوز الفشل غير المختبر ليس تجاوز فشل — بل أمل.

الخلاصة

أثبت انقطاع ياندكس أن الجهات الخبيثة لديها مرونة بنية تحتية أفضل من معظم الواجهات الخلفية للألعاب. البوتات لم تمت عندما أظلمت مراكز بيانات ياندكس — لقد هاجرت إلى مزودين آخرين في غضون دقائق واستمرت في العمل. لوحة معلومات حركة المرور الإجمالية لديك أظهرت تحسنًا. الواقع ازداد سوءًا.

ابدأ بنشر كاشف البوتات السلوكي من هذا المقال في بيئة الاختبار لديك. شغّله لمدة أسبوع لتحديد خطوط الأساس، ثم حاكِ انقطاع مزود عن طريق حظر حركة المرور من ASN واحد ولاحظ ما إذا كان كاشف الهجرة يعمل. الأنماط التي تكتشفها ستكشف بالضبط أين نقاط الضعف في بنيتك الحالية.

لاستوديوهات الألعاب التي تبني أنظمة مصادقة وبيانات لاعبين تنجو من فشل البنية التحتية، يوفّر horizOn جلسات مرتبطة بالحساب مع حفظ سحابي يدرك المراجعات — خدمات مصممة بحيث يحتفظ لاعبوك بتقدمهم حتى عند حدوث غير المتوقع. جرّبه مجانًا أو استكشف وثائق API لترى كيف تتكامل الجلسات المرتبطة بالحساب مع واجهتك الخلفية الحالية.

---

المصدر: انقطاع ياندكس: البيانات تُظهر أن البوتات الخبيثة تحركت بدلاً من أن تختفي