На проверке Парсинг сайтов как мы зделали rtfox-browser — форк undetected-chromedriver с поддержкой SOCKS5

Как мы форкнули undetected-chromedriver и добавили SOCKS5, мультипроцессинг и модуль капчи​

Статья носит познавательный и развлекательный характер. Автор не призывает к нарушению правил сервисов.

Всем привет! Недавно мы столкнулись с темой парсинга. Начали искать инструменты — BeautifulSoup, Selenium, Playwright, Puppeteer. Сразу скажу: мы не конкуренты этим библиотекам, далее объясним почему.


Нам нужно было обойти CloudFlare, запускать несколько процессов одновременно, и чтобы на каждый процесс был свой SOCKS5 прокси с авторизацией.

---

Как мы нашли отправную точку​

Нам попалась библиотека undetected-chromedriver. Библиотека перестала поддерживаться разработчиком, но функционал сохранялся. При первом запуске сразу столкнулись с ошибкой — проблема была в патчинге ChromeDriver и его скачивании. После исправления запуск удался и капча была пройдена. В дальнейшем возникли проблемы с SOCKS5 и многопроцессорностью.


Тогда пришла мысль: почему бы не взять эту же библиотеку, прикрутить прокси и добавить многопроцессорность? Но чтобы это осуществить, нужно сначала понять как вообще всё работает.

---

Как устроена цепочка Chrome → ChromeDriver → Selenium​

Прежде чем рассказывать что мы сделали — разберём архитектуру изнутри.

Звено 1 — Chrome.exe​

Обычный браузер при запуске открывает WebSocket-сервер на каком-то порту и слушает команды DevTools Protocol (CDP). Через WebSocket можно отправить например:

{  "id": 1, "method": "Page.navigate",  "params": {"url": "https://google.com"}

Звено 2 — ChromeDriver​

ChromeDriver — это отдельная программа, которая работает как переводчик между вашим кодом (Selenium) и браузером Chrome. Вы пишете на Python driver.get("Google"), Selenium превращает это в HTTP-запрос и отправляет ChromeDriver. А ChromeDriver уже объясняет Chrome на его родном языке (DevTools Protocol) что нужно открыть страницу.


ChromeDriver одновременно является:

• HTTP-сервером (слушает порт, принимает команды от Selenium)
• WebSocket-клиентом (подключается к Chrome, отправляет CDP-команды)

Звено 3 — Selenium​

Selenium — это библиотека для Python которая позволяет программно управлять браузером. Selenium не управляет браузером напрямую — он отправляет команды ChromeDriver, а тот уже разбирается с Chrome.

Полная цепочка запуска​

Ваш код: driver = webdriver.Chrome()<br> ↓<br>Selenium запускает ChromeDriver.exe как отдельный процесс<br> ↓<br>ChromeDriver.exe запускает Chrome.exe с флагами автоматизации<br>

При запуске Chrome, ChromeDriver автоматически выставляет флаги автоматизации и сообщает что браузером управляет автоматическое ПО. Библиотека undetected-chromedriver как раз с этим и разбирается — она патчит ChromeDriver.exe, убирает эти флаги и патчит сигнатуру.

Полная цепочка запроса​

Ваш код: driver.get("Google")

↓Selenium → ChromeDriver: HTTP POST /session/abc/url        
↓ChromeDriver → Chrome: WebSocket команда Page.navigate        
↓Chrome → Google: HTTP GET /
↓Google → Chrome: HTML страница
↓Chrome → ChromeDriver: WebSocket ответ (готово)
↓ChromeDriver → Selenium: HTTP 200 OK
↓Selenium: возвращает управление вашему коду

---

Что мы реализовали​

1. Автоматическое определение платформы и версии Chrome​

Первым делом нам нужно было автоматически определять платформу (Windows, Linux, macOS) чтобы корректно получать версию установленного Chrome.


Пример для Linux:


google-chrome --version # Google Chrome 124.0.6367.91

Из этой строки извлекается major-версия (124).


Получение версии ChromeDriver:

• Для Chrome ≤ 114: обращается к старому CDN chromedriver.storage.googleapis.com
• Для Chrome ≥ 115: использует новый сервис chrome-for-testing, парсит JSON с актуальными версиями

Скачивание и распаковка:

1. Формирует URL для скачивания ZIP-архива
2. Скачивает во временную папку
3. Распаковывает архив
4. Ищет файл chromedriver (или .exe)
5. Перемещает его в папку проекта /drivers

Патч бинарника — что такое cdc-сигнатура?


В оригинальном ChromeDriver есть строка {window.cdc_adoqpoas...} — её используют сайты для обнаружения ботов. Патч делает следующее:

1. Ищет в бинарнике паттерн {window.cdc.*?;}
2. Заменяет его на заглушку {console.log("ok")}

Благодаря этому сайты не могут определить что управление идёт через Selenium.

---

2. Многопроцессорность​

С оригинальной библиотекой при запуске 10 процессов каждый запускал один и тот же chromedriver. Как мы уже знаем, Selenium запускает chromedriver на своём порту. Каждый новый воркер просто перезапускал chromedriver, создавал новый порт — и прошлый воркер падал.


Решение простое: мы не даём воркерам взаимодействовать с одним chromedriver. У нас есть эталонная версия (пропатченная). Каждый воркер получает только свою копию и работает с ней. При завершении копия удаляется.


Также реализована межпроцессорная блокировка:

• Если 100 воркеров одновременно вызовут ensure_driver(), они не будут скачивать драйвер 100 раз
• Первый процесс захватывает файловую блокировку (через fcntl на Linux/Mac или msvcrt на Windows)
• Остальные ждут пока первый не скачает и не пропатчит драйвер
• Затем все используют уже готовый файл

driver = uc.Chrome(worker_id="worker_1")

---

3. SOCKS5 прокси с авторизацией​

driver = uc.Chrome(proxy={"host": "1.2.3.4",
                          "port": 1080,
                          "user": "my_login",
                          "pass": "my_password"})

Chrome из коробки не умеет работать с SOCKS5 и авторизацией — только HTTP/HTTPS, и при этом требует заполнять данные во всплывающем окне. Вот как мы решили эту проблему.


Мы запускаем локальный SOCKS5-сервер с двумя потоками:

• Поток-слушатель — принимает трафик от Chrome на localhost
• Поток-отправитель — подключается к внешнему прокси и отправляет логин/пароль

Chrome получает флаг --proxy-server=127.0.0.1:56789 и шлёт туда весь трафик, думая что это обычный прокси без авторизации. А наш локальный прокси сам добавляет логин и пароль и форвардит запросы на внешний SOCKS5-сервер.


Полная картина:

Chrome получает флаг --proxy-server=127.0.0.1:56789          ↓
Chrome думает: "Это обычный прокси без авторизации"          ↓
Chrome отправляет HTTP-запрос на локальный порт 56789          ↓
Поток-слушатель принимает запрос          ↓
Поток-отправитель:  - подключается к внешнему SOCKS5 (1.2.3.4:1080)  - отправляет логин и пароль  - пересылает HTTP-запрос          ↓
Получает ответ от внешнего прокси          ↓
Возвращает ответ Chrome через поток-слушатель

---

4. Изолированное логирование через loguru​

Loguru из коробки поддерживает глобальный логер. Нам это не подходило — для прокси нужен свой изолированный логер с возможностью включать через флаг, для ChromeDriver свой, для пользователя свой.


Решение — создаём новый экземпляр логера и добавляем фильтры:

_lib_logger = Logger(
    core=Core(),
    exception=None,
    depth=0,
    record=False,
    lazy=False,
    colors=False,
    raw=False,
    capture=True,
    patchers=[],
    extra={},
)


_lib_logger.add(sys.stderr, level="ERROR")


def get_logger(debug: bool = False, debug_proxy: bool = False):

    _lib_logger.remove()

    _lib_logger.add(
        sys.stderr,
        level="DEBUG" if debug else "ERROR",
        format="<green>{time:HH:mm:ss}</green> | <level>{level: <8}</level> | "
               "<cyan>Chrome</cyan> - <level>{message}</level>",
        filter=lambda r: r["extra"].get("source") != "proxy"
    )

    _lib_logger.add(
        sys.stderr,
        level="DEBUG" if debug_proxy else "ERROR",
        format="<blue>{time:HH:mm:ss}</blue> | <level>{level: <8}</level> | "
               "<magenta>PROXY</magenta> - <level>{message}</level>",
        filter=lambda r: r["extra"].get("source") == "proxy"
    )
    return _lib_logger

В Chrome и прокси привязываем источник:

self.logger = self.logger_str.bind(source="chrome")

---

5. Модуль капчи с автозагрузкой солверов​

Мы реализовали автоматическую загрузку солверов из папки solvers/. Это значит: есть класс с логикой прохождения капчи — просто кладёшь .py файл в папку и метод автоматически появляется.


BaseCaptchaSolver — базовый класс:

from abc import ABC, abstractmethod


class BaseCaptchaSolver(ABC):
    name: str = None
    _service = None

    @property
    def driver(self):
        return self._service.driver

    @property
    def api_key(self):
        return self._service.api_key

    @abstractmethod
    def solve(self, **kwargs) -> bool:
        ...

Чтобы написать свой солвер достаточно:

1. Унаследоваться от BaseCaptchaSolver
2. Задать name
3. Реализовать метод solve()

Loader — автоматический поиск солверов:

import importlib.util
import inspect
from pathlib import Path
from loguru import logger


def load_solvers_from_dir(directory: Path, registry: dict):
    if not directory.exists():
        logger.debug(f"Solvers directory not found: {directory}")
        return

    for py_file in directory.glob("*.py"):
        if py_file.name.startswith("_"):
            continue
        try:
            spec = importlib.util.spec_from_file_location(py_file.stem, py_file)
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)

            for _, obj in inspect.getmembers(module, inspect.isclass):
                if (hasattr(obj, 'name')
                        and obj.name
                        and obj.__name__ != 'BaseCaptchaSolver'):
                    registry[obj.name] = obj
                    logger.debug(f"Captcha solver loaded: '{obj.name}' ← {py_file.name}")
        except Exception as e:
            logger.warning(f"Failed to load solver {py_file.name}: {e}")

Loader проходит по всем .py файлам в папке solvers/, находит классы-солверы и регистрирует их по имени. Например: в папке лежит my_solver.py с классом MySolver(name="my_captcha") — Loader найдёт его и добавит в реестр под ключом "my_captcha".


CaptchaService — главный класс:

from pathlib import Path
from loguru import logger
from .loader import load_solvers_from_dir
from .base import BaseCaptchaSolver


class CaptchaService:
    """
    Использование:
        captcha = CaptchaService(api_key="KEY", driver=driver)
        captcha.ebay_hcaptcha()
    """

    def __init__(self, api_key: str, driver, solvers_dir=None):
        self.api_key = api_key
        self.driver = driver
        self._solvers: dict[str, type[BaseCaptchaSolver]] = {}

        # Ищем solvers/ рядом с main.py (cwd)
        directory = Path(solvers_dir) if solvers_dir else Path.cwd() / "solvers"
        load_solvers_from_dir(directory, self._solvers)
        self._bind_methods()

    def _instantiate(self, solver_cls):
        instance = solver_cls()
        instance._service = self
        return instance

    def _bind_methods(self):
        for name, solver_cls in self._solvers.items():
            self._create_method(name, solver_cls)

    def _create_method(self, name: str, solver_cls):
        def method(**kwargs) -> bool:
            solver = self._instantiate(solver_cls)
            return solver.solve(**kwargs)

        method.__name__ = name
        method.__doc__ = solver_cls.__doc__ or f"Solve '{name}' captcha."
        setattr(self, name, method)

    def register(self, solver_cls):
        """Ручная регистрация солвера."""
        self._solvers[solver_cls.name] = solver_cls
        self._create_method(solver_cls.name, solver_cls)
        return self

    def available(self) -> list[str]:
        return list(self._solvers.keys())

Как использовать:

rom rtfox_browser.captcha import CaptchaService
captcha = CaptchaService(    api_key="YOUR_2CAPTCHA_KEY",    driver=driver
)
# Посмотреть доступные солверы
print(captcha.available())
# ['ebay_hcaptcha', 'aws_image']
# Решить капчу
captcha.ebay_hcaptcha()

Свой солвер — просто бросить файл в папку:

from rtfox_browser.captcha import BaseCaptchaSolver


class MySolver(BaseCaptchaSolver):  
  name = "my_captcha"  
  def solve(self, **kwargs) -> bool:      
    # self.driver — браузер      
    # self.api_key — ключ API      
    return True

---

Итог​

Три основные проблемы решены:

• Обход CloudFlare из коробки
• SOCKS5 прокси с авторизацией
• Многопроцессорность с изоляцией воркеров

Плюс добавили изолированное логирование и модуль капчи с плагин-архитектурой.


Библиотека доступна на GitHub и PyPI:


pip install rtfox-browser

• GitHub: https://github.com/rtf-labs-studio/rtfox-browser
• PyPI: Client Challenge
• Telegram: https://t.me/rtf_labs_studio
• YouTube: RTF Labs Studio

---

Установка и использование​

Установка​

pip install rtfox-browser

Требования: Python >= 3.9

Базовый запуск​

import rtfox_browser as uc
driver = uc.Chrome()
driver.get("https://example.com")
driver.quit()

Запуск с прокси​

import rtfox_browser as uc
driver = uc.Chrome(proxy={"host": "1.2.3.4",
                          "port": 1080,
                          "user": "my_login",
                          "pass": "my_password"})
driver.get("https://example.com")
driver.quit()

Запуск нескольких процессов​

rom multiprocessing import Pool
import rtfox_browser as uc

def run_worker(worker_id):
  driver = uc.Chrome(worker_id=worker_id,proxy={"host": "1.2.3.4",
                                                "port": 1080,
                                                "user": "my_login",
                                                "pass": "my_password"})  
  driver.get("https://example.com")  
  driver.quit()
 
if __name__ == "__main__":  
  with Pool(4) as pool:      
  pool.map(run_worker, ["w1", "w2", "w3", "w4"])

Решение капчи​

import rtfox_browser as uc
from rtfox_browser.captcha import CaptchaService
driver = uc.Chrome()
driver.get("https://example.com")
captcha = CaptchaService(api_key="YOUR_2CAPTCHA_KEY",
                         driver=driver
)
# Посмотреть доступные солверы
print(captcha.available())
# ['ebay_hcaptcha', 'aws_image']
# Решить капчу
captcha.ebay_hcaptcha()

Обработка ошибок прокси​

from rtfox_browser.exceptions import ProxyError, ProxyAuthError, ProxyInvalidAddressError

try:
    driver = uc.Chrome(proxy={...})
except ProxyInvalidAddressError:
    print("Invalid proxy address")
except ProxyAuthError:
    print("Wrong credentials")
except ProxyError:
    print("Proxy connection failed")

Включение отладочных логов​

import rtfox_browser as uc
# Логи ChromeDriver
driver = uc.Chrome(log_debug=True)
# Логи прокси
driver = uc.Chrome(log_debug_proxy=True)
# Все логи
driver = uc.Chrome(log_debug=True, log_debug_proxy=True)