s0k0l/robotex
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8a464391a21cb3ea6ceb16244959949738e0a847
robotex/docs/AntiBot-Bypass-TZ.md
T

24 KiB
Raw Blame History

ТЗ: расширяемый модуль обхода антибот-проверок и капч

1. Контекст и цель

Сейчас в проекте обход капчи — это ручное действие: клиент сам должен прислать wait_for/click на нужный селектор и явное действие captcha с готовым captcha_type (src/engine/actions.py). Это работает, только если клиент заранее знает, какая защита стоит на сайте.

Цель — добавить отдельный слой, который САМ:

  1. определяет, что на странице сейчас показан какой-то антибот-барьер (и какой именно);
  2. проходит его по единому алгоритму, подходящему для целого класса барьеров;
  3. если барьер эскалировался до классической капчи с картинками — передаёт управление уже существующему модулю src/captcha (или его расширению).

Важно: это НЕ замена actions, а новый action (pass_challenge) + фоновая проверка, которую можно вызывать перед/после навигации и перед любым действием, если оно упёрлось в перехваченный клик.

Референс алгоритма (как это устроено в Scrapling, scrapling/engines/_browsers/_stealth.py::_cloudflare_solver + _base.py::_detect_cloudflare) — см. предыдущее обсуждение в чате. Ключевой вывод оттуда: сам "solve" — это не решение головоломки, а клик по чекбоксу с рандомизацией + поллинг, вся сложность — в детекте типа барьера и в том, чтобы окружение (фингерпринт) уже выглядело "человеческим" настолько, чтобы клик засчитался.

1.1. Главный сценарий отказа, который этот модуль обязан закрыть

В отличие от типичных open-source решений, у вас не стоит вопрос "решать капчу или нет" — сервис распознавания (2captcha) уже подключён и используется без колебаний (hcaptcha.py). Поэтому основной риск — не "не смогли решить", а:

  • API молча возвращает контент JS-challenge'а как будто это целевая страница. Сейчас в main.py:143 html = await page.inner_html('html') вызывается без какой-либо проверки, что это действительно страница, а не заглушка "Just a moment...". Если клиент прислал actions, ожидающие DOM, которого на challenge-странице никогда не будет, они просто упадут по таймауту с невнятной ошибкой — а если не прислал (например, забрать html целиком) — вернётся мусор с "status": "ok".
  • API виснет до общего timeout, потому что ждёт селектор, а страница вообще не та, и ни один action не расскажет, что происходит на самом деле.

Формула, которую должен обеспечить модуль: "открыли страницу -> если это barrier — осознали это явно (залогировали тип/провайдера) -> подождали или решили -> только после этого либо продолжили выполнять actions клиента, либо кинули понятную ошибку с указанием, на какой стадии застряли". Ни один ответ не должен уходить клиенту с пометкой "ok", если последнее состояние страницы — это всё ещё распознанный barrier (см. §7, "Финальный гейт перед ответом").

2. Классификация состояний

Любой антибот-барьер, который встречается боту при заходе на страницу, укладывается в одну из трёх стадий. Стадии одинаковы для всех провайдеров (Cloudflare, DataDome, Akamai, PerimeterX/Human, Imperva, AWS WAF Captcha, hCaptcha, reCAPTCHA) — отличаются только сигнатуры детекта, а не сам алгоритм прохождения.

Стадия Что видит пользователь Что делать Нужен ли внешний solver
A. Interstitial / "Please wait" Пустая страница-заглушка с спиннером, редиректом или JS-челленджем ("Checking your browser…", "Just a moment…", "DDoS protection by …") Просто ждать, пока страница сама не смёнится (поллинг DOM/URL), кликать не нужно Нет — только детект + ожидание
B. Checkbox-виджет Отдельная плашка/iframe с одним чекбоксом ("Verify you are human", Turnstile, invisible hCaptcha/reCAPTCHA checkbox) Клик по чекбоксу человекоподобным движением мыши, дождаться результата Обычно нет (иногда после клика эскалирует в стадию C)
C. Image/interactive challenge Сетка картинок, "выбери все со светофорами", audio-капча Нужно реальное распознавание — делегировать в 2captcha/аналог по sitekey, как уже сделано в hcaptcha.py Да

Ваше наблюдение верно: сама механика прохождения (A — просто подождать, B — кликнуть) не зависит от вендора. Расширяемость нужна только в детекторах: вы добавляете сигнатуру нового провайдера, а orchestrator и solver'ы переиспользуются.

3. Архитектура модуля

Новый пакет src/antibot/ со структурой, идущей по аналогии с уже существующим паттерном в проекте (_actions: dict в actions.py, Captcha.__subclasses__() в captcha/base.py — то есть реестр через наследование/декоратор, без магии):

src/antibot/
    __init__.py          # экспорт orchestrator + регистрация встроенных детекторов
    base.py               # абстракции: ChallengeInfo, Detector, Solver
    registry.py           # реестр детекторов, порядок проверки
    stages/
        interstitial.py   # стадия A: универсальный wait-solver + сигнатуры провайдеров
        checkbox.py        # стадия B: универсальный click-solver + сигнатуры провайдеров
        interactive.py      # стадия C: мост к src/captcha (hcaptcha/recaptcha/...)
    humanize.py            # человекоподобные движения мыши (безье+ускорение), тайминги кликов
    orchestrator.py        # цикл: detect -> solve -> verify -> retry

3.1. Базовые абстракции (base.py)

from dataclasses import dataclass
from enum import Enum
from playwright.async_api import Page

class Stage(str, Enum):
    INTERSTITIAL = "interstitial"
    CHECKBOX = "checkbox"
    INTERACTIVE = "interactive"

@dataclass
class ChallengeInfo:
    stage: Stage
    provider: str          # "cloudflare_turnstile", "datadome", "hcaptcha", "recaptcha_v2", ...
    frame_locator: str | None = None   # css/regex для iframe, если применимо
    site_key: str | None = None        # для стадии C, чтобы передать в 2captcha
    meta: dict | None = None           # что-то специфичное для провайдера

class Detector:
    """Один детектор = одна сигнатура одного провайдера на одной стадии."""
    provider: str
    stage: Stage

    async def detect(self, page: Page) -> ChallengeInfo | None:
        raise NotImplementedError

class Solver:
    """Один solver на стадию (универсальный), НЕ на провайдера."""
    stage: Stage

    async def solve(self, page: Page, info: ChallengeInfo) -> bool:
        """Возвращает True, если стадия пройдена (или больше не требуется действий)."""
        raise NotImplementedError

Важно: Detector -> 1:1 с провайдером, Solver -> 1:1 со стадией. Не наоборот. Это и даёт расширяемость: новый провайдер = один новый класс детектора на несколько строк, без изменения логики клика/ожидания.

3.2. Реестр (registry.py)

_detectors: dict[Stage, list[Detector]] = {s: [] for s in Stage}

def register_detector(detector_cls):
    instance = detector_cls()
    _detectors[instance.stage].append(instance)
    return detector_cls

async def detect_any(page: Page, stage: Stage) -> ChallengeInfo | None:
    for detector in _detectors[stage]:
        info = await detector.detect(page)
        if info:
            return info
    return None

Порядок стадий при детекте всегда A -> B -> C (сначала проверяем "не висим ли мы ещё в заглушке", потом "нет ли чекбокса", потом "не показали ли уже картинки"), потому что провайдер может сразу показать любую из них в зависимости от risk score IP/фингерпринта.

4. Алгоритм по стадиям

4.1. Стадия A — Interstitial (stages/interstitial.py)

Детекторы (примеры сигнатур, добавлять новые — просто дописывать список):

  • Cloudflare: строка cType: 'non-interactive' / cType: 'managed' в HTML (как в оригинале), либо <title>Just a moment...</title>, либо <div id="challenge-running">.
  • DataDome: <meta name="datadome-cid">, geo.captcha-delivery.com в src скриптов, id="ddv1-captcha-container".
  • Akamai/PerimeterX/AWS WAF: аналогично — по характерным <script src>, <div id=...>, заголовку Server/X-* из response (это можно проверять и до детекта DOM — по HTTP-заголовкам ответа, что дешевле).
  • Generic fallback: если основной контент страницы (селектор, который клиент ожидает увидеть) отсутствует дольше N секунд, а document.readyState === 'complete' — считаем, что это неизвестный interstitial, и просто ждём с общим таймаутом.

Solver (один на всех):

async def solve_interstitial(page, info, timeout_ms=15000, poll_ms=500):
    elapsed = 0
    while elapsed < timeout_ms:
        if await detect_any(page, Stage.INTERSTITIAL) is None:
            return True
        await page.wait_for_timeout(poll_ms)
        elapsed += poll_ms
    return False

Никакого клика — это осознанно, вы это сами верно отметили. Тут вся работа делается заранее: приличный TLS/JS-фингерпринт (см. §5) должен довести risk score до значения, при котором сервер сам снимает заглушку через JS-редирект/cookie.

4.2. Стадия B — Checkbox (stages/checkbox.py)

Детекторы:

  • Cloudflare Turnstile: iframe[src*="challenges.cloudflare.com/cdn-cgi/challenge-platform"] или script[src*="challenges.cloudflare.com/turnstile/v"] (embedded-вариант).
  • hCaptcha standalone checkbox: iframe[src*="hcaptcha.com"][src*="frame=checkbox"] / div[data-hcaptcha-widget-id] без активного challenge-фрейма.
  • reCAPTCHA v2 checkbox: iframe[src*="recaptcha"][title*="reCAPTCHA"], .g-recaptcha.

Solver (один на всех, важна человекоподобность — см. §5.2):

async def solve_checkbox(page, info, max_attempts=3):
    for attempt in range(max_attempts):
        frame = await locate_frame(page, info.frame_locator)
        box = await bounding_box_of_checkbox(frame)
        if box is None:
            if await detect_any(page, Stage.CHECKBOX) is None:
                return True  # само пропало / уже пройдено
            await page.wait_for_timeout(500)
            continue

        await humanize.move_and_click(page, box)   # см. humanize.py
        await wait_for_network_idle(page)

        # после клика — либо чисто, либо эскалация в картинки (стадия C)
        if await detect_any(page, Stage.CHECKBOX) is None:
            return True
        if await detect_any(page, Stage.INTERACTIVE):
            return False  # передаём выше, пусть оркестратор переключит стадию
    return False

Это то же самое, что у Scrapling (случайный оффсет внутри чекбокса + случайная задержка клика + поллинг до 100 итераций по 100мс), только вынесено в общий для всех провайдеров кусок.

4.3. Стадия C — Interactive (картинки) (stages/interactive.py)

Тут новых алгоритмов не изобретаем — берём sitekey (детектор его достаёт из DOM, как уже делает hcaptcha.py:32) и отдаём в уже интегрированный 2captcha через существующий реестр Captcha.__subclasses__(). Единственное расширение — добавить SolverReCaptcha(Captcha) с ctype = "recaptcha" по образцу SolverHCaptcha, т.к. TwoCaptcha в зависимостях это уже умеет (self._solver.recaptcha(...)).

async def solve_interactive(page, info, captcha_key: str):
    solver = load_captcha(captcha_key, info.provider)  # "hcaptcha" | "recaptcha"
    await solver.solve(info.frame_locator or info.meta["container_selector"], page)
    return True

5. Слой "похожести на человека" (humanize.py)

Это то, что раньше бесплатно давал Camoufox, а Patchright/чистый Playwright — нет. Нужно реализовать самим:

  1. Траектория мыши — не прыжок в точку, а несколько промежуточных точек по кривой Безье от текущей позиции курсора до цели, с нелинейным (ease-in-out) распределением скорости и небольшим шумом (±1-2px) на каждом шаге. page.mouse.move(x, y, steps=N) даёт линейную интерполяцию — этого мало, нужно звать mouse.move несколько раз вручную по расчётным точкам.
  2. Ускорение/замедление клика — задержка mouse.down() -> mouse.up() рандомная (100-250мс), плюс случайный сдвиг координаты клика внутри bounding box (не в центр, как это делает и сам Scrapling — randint(26,28)).
  3. Общая стелс-часть (не про капчу, но без неё стадия A вообще не пройдёт): подмена navigator.webdriver, canvas noise, консистентные timezone/locale/UA — то, что в вашем проекте уже частично закрыто Camoufox-фингерпринтом в main.py (generate_context_fingerprint). Это нужно сохранить как есть — не переизобретать, Camoufox с этим справляется хорошо, вопрос отказа от него у Scrapling был про производительность/стабильность, а не про качество стелса.

6. Оркестратор (orchestrator.py)

@dataclass
class ChallengeOutcome:
    cleared: bool
    last_stage: Stage | None = None   # на чём застряли, если cleared=False
    last_provider: str | None = None
    reason: str | None = None         # "no captcha_key" / "timeout" / ...

async def pass_challenges(page, *, captcha_key: str | None, timeout_ms=45000) -> ChallengeOutcome:
    stages = [Stage.INTERSTITIAL, Stage.CHECKBOX, Stage.INTERACTIVE]
    deadline = time.monotonic() + timeout_ms / 1000
    solvers = {
        Stage.INTERSTITIAL: solve_interstitial,
        Stage.CHECKBOX: solve_checkbox,
        Stage.INTERACTIVE: partial(solve_interactive, captcha_key=captcha_key),
    }

    while time.monotonic() < deadline:
        for stage in stages:
            info = await detect_any(page, stage)
            if info is None:
                continue
            if stage is Stage.INTERACTIVE and not captcha_key:
                return ChallengeOutcome(False, stage, info.provider, "no captcha_key provided")
            log.info("Challenge detected: stage=%s provider=%s", stage, info.provider)
            ok = await solvers[stage](page, info)
            if not ok and stage == Stage.CHECKBOX:
                continue  # пусть цикл заново найдёт INTERACTIVE
            break
        else:
            return ChallengeOutcome(True)  # ни одна стадия не сработала -> барьеров не осталось
    # вышли по таймауту — фиксируем, на чём именно, для диагностики клиенту
    stuck = await detect_any(page, Stage.INTERSTITIAL) or await detect_any(page, Stage.CHECKBOX) or await detect_any(page, Stage.INTERACTIVE)
    return ChallengeOutcome(False, stuck.stage if stuck else None, stuck.provider if stuck else None, "timeout")

Возврат структурированного ChallengeOutcome, а не голого bool, — намеренно: клиенту важно не просто "не получилось", а на чём именно (interstitial завис / checkbox не кликается / нет ключа для картинок), чтобы отличать "сайт нам недоступен" от "забыли передать captcha_key".

7. Интеграция в текущий пайплайн

  • Новый action pass_challenge в actions.py, регистрируется через @register как остальные — принимает опциональные timeout, использует уже переданный captcha_key из ctx (он уже прокидывается в execute()).
  • В main.py вызов pass_challenges(page, ...) идёт сразу после page.goto(), безусловно (не по флагу) — это закрывает случай, когда антибот встаёт стеной ДО того, как клиент вообще может прислать wait_for/click на нужный селектор (он же не знает заранее html чужого сайта). Флагом можно вынести только timeout_ms для этой стадии.
  • captcha_selector/captcha_type в SolveRequest — при использовании оркестратора становятся необязательными (детектор сам найдёт sitekey), но их стоит оставить как явный override на случай нестандартной вёрстки.

7.1. Финальный гейт перед ответом (закрывает §1.1)

Мало прогнать pass_challenges один раз до actions — сайт может подсунуть challenge и повторно (например, между двумя действиями клиента, после навигации внутри page_action). Поэтому:

  1. До actions.execute(...): вызвать pass_challenges; если outcome.cleared is False — сразу вернуть {"status": "err", "error": f"blocked by {outcome.last_provider} at stage {outcome.last_stage}"}, не выполняя список actions вообще (они гарантированно упадут на несуществующих селекторах и дадут менее внятную ошибку).
  2. После actions.execute(...), перед page.inner_html('html'): ещё раз дёрнуть быстрый детект (detect_any по всем трём стадиям, без solve, timeout ~0) — если что-то вылезло уже в процессе выполнения actions (SPA-редирект, скрытый iframe подгрузился с опозданием), это тоже должно перевести status в "err", а не в "ok" с challenge-страницей внутри html.
  3. Оба места используют один и тот же pass_challenges/detect_any, так что провайдеров и сигнатуры поддерживаем в одном месте — из §8.

8. Как добавлять нового провайдера (для будущих вас)

  1. Определить стадию по которой он срабатывает (обычно A, иногда сразу B).
  2. Написать класс Detector с сигнатурой (селектор/строка в HTML/паттерн URL iframe/спец. заголовок ответа).
  3. Задекорировать @register_detector.
  4. Ничего больше — solver уже общий для стадии.

9. Открытые вопросы / решить перед реализацией

  • Нужно ли ловить сигнатуры по HTTP-заголовкам ответа (page.on("response")), а не только по DOM — это быстрее и надёжнее для Akamai/PerimeterX, которые иногда отдают спец. статус-коды (403/405) с телом-заглушкой ещё до полной отрисовки.
  • Лимит retry и timeout по каждой стадии — сейчас в набросках взяты значения по аналогии со Scrapling (3 попытки чекбокса, ~15-60с на interstitial), нужно откалибровать под реальные сайты.
  • Что делать, если стадия C (картинки) сработала, а captcha_key не передан клиентом — падать с понятной ошибкой "challenge escalated to image captcha, but no captcha_key provided", а не тихо повисать в timeout.
Reference in new issue View Git Blame Copy Permalink