Реалізація захисту від API Scraping (bot detection)

Захист API від скрейпингу та виявлення ботів

API скрейпинг — систематичний збір даних з частотою вищою за нормальну взаємодію людини. Без захисту конкурент може завантажити весь каталог товарів за кілька годин, автоматично підбирати паролі або парсити базу контактів. Завдання — розрізнити бота від людини без шкоди для легітимних користувачів.

Шари захисту

Клієнт → WAF (репутація IP) → Rate Limiting → Bot Detection → API Logic
                                                     ↓
                              Fingerprint + Behavioral Analysis + CAPTCHA

Кожен шар фільтрує частину трафіку. Ідеальний захист — комбінація кількох методів, жоден з яких не ідеальний сам по собі.

Сигнали ботів та їх вага

Детектор на основі поведінкових ознак

import time
import statistics
from collections import defaultdict, deque

class BotDetector:
    def __init__(self, redis_client):
        self.r = redis_client
        self.window = 300  # 5-хвилинне вікно аналізу

    def analyze_request(self, request) -> dict:
        """Повертає score (0-100) та причини підозри"""
        score = 0
        reasons = []

        # 1. Заголовки браузера
        headers = request.headers
        ua = headers.get('User-Agent', '')

        bot_uas = ['python-requests', 'curl', 'wget', 'Go-http-client',
                   'Java/', 'okhttp', 'axios', 'node-fetch']
        for bot_ua in bot_uas:
            if bot_ua.lower() in ua.lower():
                score += 40
                reasons.append(f'bot_useragent:{bot_ua}')
                break

        if not ua:
            score += 40
            reasons.append('no_useragent')

        if not headers.get('Accept-Language'):
            score += 20
            reasons.append('no_accept_language')

        if not headers.get('Accept-Encoding'):
            score += 15
            reasons.append('no_accept_encoding')

        # 2. Аналіз часу запитів
        ip = request.remote_addr
        timing_score = self._analyze_timing(ip)
        if timing_score > 0:
            score += timing_score
            reasons.append(f'suspicious_timing:{timing_score}')

        # 3. Паттерн URL (послідовний обхід)
        path = request.path
        pattern_score = self._analyze_url_pattern(ip, path)
        if pattern_score > 0:
            score += pattern_score
            reasons.append(f'url_pattern:{pattern_score}')

        # 4. JA3 TLS fingerprint (через nginx змінну)
        ja3 = headers.get('X-JA3-Fingerprint')
        if ja3 and self._is_suspicious_ja3(ja3):
            score += 30
            reasons.append(f'suspicious_ja3:{ja3[:16]}')

        return {
            'score': min(score, 100),
            'is_bot': score >= 60,
            'reasons': reasons,
            'action': self._get_action(score)
        }

    def _analyze_timing(self, ip: str) -> int:
        """Аналіз інтервалів між запитами"""
        key = f"timing:{ip}"
        now = time.time()

        # Зберегти позначку часу
        self.r.lpush(key, now)
        self.r.ltrim(key, 0, 49)  # останні 50 запитів
        self.r.expire(key, self.window)

        timestamps = [float(t) for t in self.r.lrange(key, 0, -1)]
        if len(timestamps) < 5:
            return 0

        # Обчислити інтервали між запитами
        timestamps.sort()
        intervals = [timestamps[i+1] - timestamps[i]
                     for i in range(len(timestamps)-1)]

        if not intervals:
            return 0

        avg = statistics.mean(intervals)
        stdev = statistics.stdev(intervals) if len(intervals) > 1 else 0

        # Коефіцієнт варіації < 0.1 означає машинну точність
        cv = stdev / avg if avg > 0 else 0

        if cv < 0.05 and avg < 2.0:  # дуже регулярні, швидкі запити
            return 35
        if cv < 0.15 and avg < 1.0:  # регулярні, дуже швидкі
            return 25

        return 0