protokeyandClaude Opus 5.5
08a6b52b9d
- Dockerfile: python 3.12, Google Chrome (patchright), Xvfb, tini как PID 1 против зомби-процессов - docker-compose: volume для extra/, shm_size 2gb, порт через API_PORT - requirements: добавлены playwright и camoufox - docs: Deployment.md, README и доки актуализированы под Chrome Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
259 lines
24 KiB
Markdown
259 lines
24 KiB
Markdown
# ТЗ: расширяемый модуль обхода антибот-проверок и капч
|
||
|
||
## 1. Контекст и цель
|
||
|
||
Сейчас в проекте обход капчи — это ручное действие: клиент сам должен прислать `wait_for`/`click` на нужный селектор и явное действие `captcha` с готовым `captcha_type` ([src/engine/actions.py](../src/engine/actions.py)). Это работает, только если клиент заранее знает, какая защита стоит на сайте.
|
||
|
||
Цель — добавить **отдельный слой**, который САМ:
|
||
|
||
1. определяет, что на странице сейчас показан какой-то антибот-барьер (и какой именно);
|
||
2. проходит его по единому алгоритму, подходящему для целого класса барьеров;
|
||
3. если барьер эскалировался до классической капчи с картинками — передаёт управление уже существующему модулю [src/captcha](../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](../src/captcha/hcaptcha.py)). Поэтому основной риск — не "не смогли решить", а:
|
||
|
||
- **API молча возвращает контент JS-challenge'а как будто это целевая страница.** Сейчас в [main.py:143](../src/main.py) `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](../src/captcha/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`)
|
||
|
||
```python
|
||
from dataclasses import dataclass
|
||
from enum import Enum
|
||
from patchright.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`)
|
||
|
||
```python
|
||
_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** (один на всех):
|
||
|
||
```python
|
||
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):
|
||
|
||
```python
|
||
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](../src/captcha/hcaptcha.py)) и отдаём в уже интегрированный 2captcha через существующий реестр `Captcha.__subclasses__()`. Единственное расширение — добавить `SolverReCaptcha(Captcha)` с `ctype = "recaptcha"` по образцу `SolverHCaptcha`, т.к. `TwoCaptcha` в зависимостях это уже умеет (`self._solver.recaptcha(...)`).
|
||
|
||
```python
|
||
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 — сейчас закрывается связкой Patchright + настоящий Google Chrome (`real_chrome=True`) в [engine/stealthy.py](../src/engine/stealthy.py), timezone/locale берутся из геолокации прокси ([api/geoip.py](../src/api/geoip.py)). Camoufox (Firefox) пока остаётся в зависимостях, но как движок не используется — проблемы были скорее с генератором отпечатков, чем с самим Camoufox.
|
||
|
||
## 6. Оркестратор (`orchestrator.py`)
|
||
|
||
```python
|
||
@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](../src/engine/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.
|