From f2d4d441f43c0a0bd9a9391f03ade0c79e7fde4b Mon Sep 17 00:00:00 2001 From: protokey Date: Tue, 22 Sep 2026 22:41:14 +0400 Subject: [PATCH] =?UTF-8?q?feature:=20=D0=BF=D0=B5=D1=80=D0=B5=D0=B5=D0=B7?= =?UTF-8?q?=D0=B4=20=D0=BD=D0=B0=20=D1=85=D1=80=D0=BE=D0=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- requirements.txt | 1 - src/antibot/humanize.py | 127 ++++++++++++++++----- src/antibot/known_providers/cloudflare.py | 16 ++- src/engine/fingerprint.py | 67 +++++++++++ src/engine/geoip.py | 113 ++++++++++++++++++ src/main.py | 133 ++++++++++------------ 6 files changed, 355 insertions(+), 102 deletions(-) create mode 100644 src/engine/fingerprint.py create mode 100644 src/engine/geoip.py diff --git a/requirements.txt b/requirements.txt index 38370ba..f1d02a1 100644 --- a/requirements.txt +++ b/requirements.txt @@ -10,7 +10,6 @@ apify_fingerprint_datapoints==0.15.0 attrs==26.1.0 blessed==1.49.0 browserforge==1.2.4 -camoufox==0.5.6 certifi==2026.7.22 charset-normalizer==3.5.1 click==8.5.0 diff --git a/src/antibot/humanize.py b/src/antibot/humanize.py index 8b2591b..6af2d4b 100644 --- a/src/antibot/humanize.py +++ b/src/antibot/humanize.py @@ -1,34 +1,11 @@ """ человекоподобные движения мыши (безье+ускорение), тайминги кликов -Camoufox — это форк Firefox, заточенный под антидетект-автоматизацию (спуфинг fingerprint'а, WebGL, canvas, -шрифтов и т.д.), с Python-обёрткой поверх Playwright-протокола (Juggler). - -У него есть параметр humanize при запуске: - -``` -from camoufox.sync_api import Camoufox - -with Camoufox(humanize=True) as browser: - ... -``` - -Или с указанием максимального времени на движение: - -``` -Camoufox(humanize=1.5) # курсор доберётся до цели не более чем за 1.5 сек -``` - -Когда humanize включён, при вызовах вроде page.click() / locator.click() библиотека: - -- строит траекторию курсора от текущей позиции до цели не по прямой, а по кривой (используется подход, -похожий на ghost-cursor — Безье-путь с шумом), -- разбивает движение на промежуточные точки с нелинейной скоростью (ускорение/замедление к цели, как у реальной руки), -- добавляет небольшую случайную задержку перед самим кликом. - -Это реализовано на уровне их собственного форка драйвера, а не через голый Playwright API — обычный Playwright ( -page.click()) телепортирует курсор мгновенно и такого не делает. +Раньше это частично закрывал Camoufox (форк Firefox с движением курсора по кривой при humanize=True) — +после перехода на голый Playwright + Chromium это ушло: обычный page.click()/locator.click() телепортирует +курсор мгновенно. Функции ниже — единственный источник человекоподобности сейчас. """ +import math import random import string from typing import Optional @@ -36,6 +13,102 @@ from typing import Optional from playwright.async_api import TimeoutError from playwright.async_api import Page +# последняя известная позиция курсора на страницу — Playwright не даёт её прочитать из браузера, +# поэтому считаем сами, чтобы следующее движение начиналось не из (0, 0), а оттуда, где реально +# остановились в прошлый раз +_last_mouse_pos: dict[int, tuple[float, float]] = {} + + +def _ease_in_out(t: float) -> float: + """smoothstep: медленный старт, разгон к середине, торможение к цели — как у реальной руки.""" + return t * t * (3 - 2 * t) + + +def _cubic_bezier(p0, c1, c2, p1, t: float) -> tuple[float, float]: + mt = 1 - t + x = mt ** 3 * p0[0] + 3 * mt ** 2 * t * c1[0] + 3 * mt * t ** 2 * c2[0] + t ** 3 * p1[0] + y = mt ** 3 * p0[1] + 3 * mt ** 2 * t * c1[1] + 3 * mt * t ** 2 * c2[1] + t ** 3 * p1[1] + return x, y + + +def _control_points(x0, y0, x1, y1): + """Два контрольных узла, смещённых перпендикулярно линии движения — даёт дугу вместо прямой, + как у настоящего движения руки/трекпада, а не у синтетического teleport'а.""" + dx, dy = x1 - x0, y1 - y0 + dist = math.hypot(dx, dy) or 1.0 + nx, ny = -dy / dist, dx / dist + + curve = random.uniform(0.08, 0.28) * dist * random.choice((-1, 1)) + t1 = random.uniform(0.2, 0.4) + t2 = random.uniform(0.6, 0.8) + + c1 = (x0 + dx * t1 + nx * curve, y0 + dy * t1 + ny * curve) + c2 = (x0 + dx * t2 + nx * curve * random.uniform(0.5, 1.0), y0 + dy * t2 + ny * curve * random.uniform(0.5, 1.0)) + return c1, c2 + + +def _random_start_position(page: Page) -> tuple[float, float]: + vp = page.viewport_size or {"width": 1280, "height": 720} + return ( + random.uniform(vp["width"] * 0.3, vp["width"] * 0.7), + random.uniform(vp["height"] * 0.3, vp["height"] * 0.7), + ) + + +async def move_mouse_human( + page: Page, + x: float, + y: float, + *, + steps: Optional[int] = None, + min_duration: float = 0.25, + max_duration: float = 0.9, +) -> None: + """Двигает курсор к (x, y) по кривой Безье с ease-in-out таймингом и лёгким дрожанием на + промежуточных точках, вместо page.mouse.move(x, y), которое телепортирует курсор мгновенно. + + Начальная точка — последняя известная позиция курсора на этой странице (или случайная точка + у центра вьюпорта, если это первое движение в сессии).""" + start = _last_mouse_pos.get(id(page)) or _random_start_position(page) + c1, c2 = _control_points(start[0], start[1], x, y) + + dist = math.hypot(x - start[0], y - start[1]) + if steps is None: + steps = max(8, min(60, int(dist / 8))) + + duration = random.uniform(min_duration, max_duration) + step_delay_ms = (duration / steps) * 1000 + + for i in range(1, steps + 1): + t = _ease_in_out(i / steps) + px, py = _cubic_bezier(start, c1, c2, (x, y), t) + if i < steps: + px += random.uniform(-1.5, 1.5) + py += random.uniform(-1.5, 1.5) + print("Move:", px, py) + await page.mouse.move(px, py) + await page.wait_for_timeout(step_delay_ms * random.uniform(0.7, 1.3)) + + _last_mouse_pos[id(page)] = (x, y) + + +async def click_human( + page: Page, + x: float, + y: float, + *, + button: str = "left", + **move_kwargs, +) -> None: + """move_mouse_human + клик с человеческой задержкой между наведением и down/up, вместо + мгновенного page.mouse.click(x, y).""" + await move_mouse_human(page, x, y, **move_kwargs) + await page.wait_for_timeout(random.uniform(50, 180)) + await page.mouse.down(button=button) + await page.wait_for_timeout(random.uniform(40, 99)) + await page.mouse.up(button=button) + await move_mouse_human(page, x-50, y+50, **move_kwargs) + async def wait_for_networkidle(page: Page, timeout: Optional[int] = None): """Wait for the page to become idle (no network activity) even if there are never-ending requests.""" diff --git a/src/antibot/known_providers/cloudflare.py b/src/antibot/known_providers/cloudflare.py index 0f11c6e..5bf714f 100644 --- a/src/antibot/known_providers/cloudflare.py +++ b/src/antibot/known_providers/cloudflare.py @@ -68,7 +68,7 @@ from playwright.async_api import Page from ..base import Detector, ChallengeInfo, AntiBotProvider, Solver from ..base import Stage from ..registry import register -from ..humanize import wait_for_page_stability, wait_for_networkidle +from ..humanize import wait_for_page_stability, wait_for_networkidle, click_human log = logging.getLogger(__name__) log.setLevel(logging.DEBUG) @@ -140,13 +140,21 @@ __CF_PATTERN__ = re.compile(r"^https://challenges\.cloudflare\.com/cdn-cgi/chall class CloudflareSolver(Solver): async def solve(self, page: Page, info: ChallengeInfo): + iframe = page.frame(url=__CF_PATTERN__) + if iframe: + await page.wait_for_timeout(randint(4000, 7000)) # дать чекбоксу домонтироваться в closed shadow root + frame_el = await iframe.frame_element() + box = await frame_el.bounding_box() + else: + box = await page.query_selector("xpath=//input[@name='cf-turnstile-response']/../div").bounding_box() + if info.stage == Stage.CHECKBOX: # # Calculate the Captcha coordinates for any viewport - box = await page.locator("xpath=//input[@name='cf-turnstile-response']/../div").bounding_box() captcha_x, captcha_y = box["x"] + randint(26, 28), box["y"] + randint(25, 27) - # # TODO: не кликает! - await page.mouse.click(captcha_x, captcha_y, delay=randint(100, 200), button="left") + # await page.mouse.click(captcha_x, captcha_y, delay=randint(100, 200), button="left") + + await click_human(page, captcha_x, captcha_y) await wait_for_networkidle(page) diff --git a/src/engine/fingerprint.py b/src/engine/fingerprint.py new file mode 100644 index 0000000..5eeabe8 --- /dev/null +++ b/src/engine/fingerprint.py @@ -0,0 +1,67 @@ +""" +Генерация и persist согласованного browser fingerprint на профиль (user_data_dir). + +Используем browserforge — ту же библиотеку, которую под капотом использовал Camoufox — но +поверх голого Playwright Chromium: генерируем Fingerprint (navigator/screen/WebGL/canvas/шрифты/ +headers, всё согласовано друг с другом и с выбранным UA) и инжектим его через add_init_script. + +Важно: генерируем именно под browser=['chrome'] — раз движок у нас реальный Chromium, подделывать +navigator под Firefox/Safari было бы хуже, чем не подделывать вообще (несоответствие между +заявленным UA и реальными JS-фичами движка — сильный сигнал для антибота). +""" +import json +from pathlib import Path + +from browserforge.fingerprints import Fingerprint, FingerprintGenerator, Screen +from browserforge.fingerprints.generator import NavigatorFingerprint, ScreenFingerprint, VideoCard +from browserforge.injectors.utils import InjectFunction, only_injectable_headers + +FINGERPRINT_FILE = "fingerprint.json" + + +def _fingerprint_from_dict(data: dict) -> Fingerprint: + data = dict(data) + data["screen"] = ScreenFingerprint(**data["screen"]) + data["navigator"] = NavigatorFingerprint(**data["navigator"]) + if data.get("videoCard"): + data["videoCard"] = VideoCard(**data["videoCard"]) + return Fingerprint(**data) + + +def load_or_create_fingerprint(user_data_dir: Path, **generator_kwargs) -> Fingerprint: + """Держит fingerprint стабильным для профиля: генерируется один раз при первом запросе + этой сессии (user_data_dir) и переиспользуется во всех последующих — так же, как реальный + человек не меняет видеокарту/шрифты/UA между визитами на сайт.""" + fp_file = user_data_dir / FINGERPRINT_FILE + if fp_file.exists(): + return _fingerprint_from_dict(json.loads(fp_file.read_text())) + + generator_kwargs.setdefault("browser", ["chrome"]) + fingerprint = FingerprintGenerator().generate(**generator_kwargs) + fp_file.write_text(fingerprint.dumps()) + return fingerprint + + +def context_options_for(fingerprint: Fingerprint, **overrides) -> dict: + """Опции контекста (launch_persistent_context/new_context), согласованные с fingerprint'ом. + Headers сюда намеренно не кладём — их полный набор выставляет apply_fingerprint() через + set_extra_http_headers() уже после создания контекста, а он не мёржит, а заменяет, + так что выставлять их дважды бессмысленно.""" + return { + "user_agent": fingerprint.navigator.userAgent, + "viewport": { + "width": fingerprint.screen.width, + "height": fingerprint.screen.height, + **overrides.pop("viewport", {}), + }, + "device_scale_factor": fingerprint.screen.devicePixelRatio, + **overrides, + } + + +async def apply_fingerprint(context, fingerprint: Fingerprint, browser_name: str = "chromium"): + """Довешивает на уже созданный контекст заголовки и init-script с подделкой + navigator/screen/WebGL/codecs/battery. Вызывать сразу после launch_persistent_context/new_context, + до первого page.goto().""" + await context.set_extra_http_headers(only_injectable_headers(fingerprint.headers, browser_name)) + await context.add_init_script(InjectFunction(fingerprint)) diff --git a/src/engine/geoip.py b/src/engine/geoip.py new file mode 100644 index 0000000..9016fb2 --- /dev/null +++ b/src/engine/geoip.py @@ -0,0 +1,113 @@ +""" +Резолвинг геолокации/таймзоны исходящего IP через сам прокси (не локальный IP), с кэшем в файле. + +Раньше это делал Camoufox (geoip=True) под капотом. Смысл: если браузер репортит IP из +одной страны/города (через прокси), а navigator.timezone/geolocation — из другой, это +явное рассогласование, которое легко ловится антиботом. Поэтому таймзону/координаты нужно +брать по тому же IP, что увидит целевой сайт — то есть запрашивать через прокси, а не с +локальной машины. + +Кэш нужен по двум причинам: у ipinfo.io есть квота на бесплатные запросы, и результат для +одного и того же прокси (пока он не сменил exit-IP) не меняется от запроса к запросу — незачем +дёргать внешний сервис на каждый /v1/solve. +""" +import asyncio +import json +import logging +import time +from pathlib import Path + +import httpx +from playwright.async_api import ProxySettings + +log = logging.getLogger(__name__) + +IPINFO_URL = "https://ipinfo.io/json" +DEFAULT_CACHE_FILE = Path(__file__).resolve().parent.parent.parent / "extra" / "geoip_cache.json" +DEFAULT_TTL = 24 * 3600 # прокси может сменить exit-IP — не кэшируем навечно + +_cache_lock = asyncio.Lock() + + +def proxy_settings_to_url(proxy: ProxySettings) -> str: + server = proxy["server"] + username = proxy.get("username") + password = proxy.get("password") + if not username: + return server + scheme, _, rest = server.partition("://") + return f"{scheme}://{username}:{password}@{rest}" + + +def _load_cache(cache_file: Path) -> dict: + if not cache_file.exists(): + return {} + try: + return json.loads(cache_file.read_text()) + except (json.JSONDecodeError, OSError): + return {} + + +def _save_cache(cache_file: Path, cache: dict) -> None: + cache_file.parent.mkdir(parents=True, exist_ok=True) + cache_file.write_text(json.dumps(cache, indent=2, ensure_ascii=False)) + + +async def resolve_geo( + proxy: ProxySettings, + timeout: float = 10.0, + cache_file: Path = DEFAULT_CACHE_FILE, + ttl: float = DEFAULT_TTL, +) -> dict | None: + """Возвращает {"timezone_id", "geolocation": {"latitude", "longitude"}, "country", "ip"} + по IP, который виден снаружи через прокси, или None, если резолв не удался и в кэше + тоже ничего нет — в этом случае вызывающий код просто не переопределяет timezone/geolocation. + + Язык (locale) сюда намеренно не подмешиваем: Accept-Language — это предпочтение пользователя, + а не производная от геолокации IP, и его уже согласованно генерирует fingerprint (см. + engine/fingerprint.py). Подменять его на "en-" было бы отдельным, никак не + связанным с реальным поведением допущением.""" + key = proxy_settings_to_url(proxy) + + async with _cache_lock: + cache = _load_cache(cache_file) + entry = cache.get(key) + if entry and (time.time() - entry["ts"]) < ttl: + return entry["data"] + + try: + async with httpx.AsyncClient(proxy=key, timeout=timeout) as client: + resp = await client.get(IPINFO_URL) + resp.raise_for_status() + data = resp.json() + except Exception as e: + log.warning("не удалось резолвнуть geoip через прокси: %r", e) + # сеть/квота ipinfo могли отказать разово — лучше отдать протухший кэш, чем ничего, + # если прокси тот же самый (скорее всего и IP тот же) + if entry: + log.info("используем протухший geoip-кэш из-за сбоя запроса") + return entry["data"] + return None + + loc = data.get("loc") # "lat,lon" + geolocation = None + if loc and "," in loc: + lat_str, lon_str = loc.split(",", 1) + try: + geolocation = {"latitude": float(lat_str), "longitude": float(lon_str)} + except ValueError: + geolocation = None + + result = { + "ip": data.get("ip"), + "country": data.get("country"), # "RU", "US", ... + "timezone_id": data.get("timezone"), + "geolocation": geolocation, + } + + async with _cache_lock: + cache = _load_cache(cache_file) + cache[key] = {"ts": time.time(), "data": result} + _save_cache(cache_file, cache) + + return result diff --git a/src/main.py b/src/main.py index 257e451..c741062 100644 --- a/src/main.py +++ b/src/main.py @@ -6,17 +6,17 @@ from pathlib import Path from typing import Annotated from urllib.parse import urlsplit -from camoufox.async_api import AsyncCamoufox -from camoufox.fingerprints import generate_context_fingerprint -from camoufox.utils import determine_ua_os +from browserforge.fingerprints import Screen from fastapi import Cookie from fastapi import FastAPI from fastapi import Request from fastapi import Response from playwright.async_api import ProxySettings -from ua_parser import user_agent_parser +from playwright.async_api import async_playwright from engine import actions +from engine.fingerprint import apply_fingerprint, context_options_for, load_or_create_fingerprint +from engine.geoip import resolve_geo from engine.schemas import SolveRequest logger = logging.getLogger(__name__) @@ -24,8 +24,6 @@ logger.setLevel(logging.DEBUG) logger.addHandler(logging.StreamHandler()) BROWSER_SLOTS = 10 -IDENTITY_FILE = Path("identity.json") - def parse_proxy(proxy_url: str) -> ProxySettings: parts = urlsplit(proxy_url) @@ -38,22 +36,6 @@ def parse_proxy(proxy_url: str) -> ProxySettings: return settings -def load_or_create_identity(user_data_dir, os=None, ff_version=None, locale=None, time_zone=None, user_agent=None) -> dict: - identity_file = user_data_dir / IDENTITY_FILE - - if identity_file.exists(): - return json.loads(identity_file.read_text()) - - kwargs = {} - if user_agent: - os, ff_version = parse_os_and_ff(user_agent) - kwargs["config_overrides"] = {"navigator.UserAgent": user_agent} - - fp = generate_context_fingerprint(os=os, ff_version=ff_version, locale=locale, timezone=time_zone) - identity_file.write_text(json.dumps(fp)) - return fp - - app = FastAPI() @@ -62,17 +44,6 @@ async def root(): return {"message": "Hello World"} -_OS_MAP = {'win': 'windows', 'mac': 'macos', 'lin': 'linux'} - - -def parse_os_and_ff(user_agent) -> tuple[str, str]: - os_name = _OS_MAP[determine_ua_os(user_agent)] - parsed = user_agent_parser.ParseUserAgent(user_agent) - ff_version = int(parsed["major"]) if ( - parsed.get("family") == "Firefox" and parsed.get("major")) else 150 - return os_name, str(ff_version) - - @app.post("/v1/solve") async def solve( request: Request, @@ -106,52 +77,74 @@ async def solve( if not user_data_dir.exists(): user_data_dir.mkdir(parents=True) - config = load_or_create_identity(user_data_dir, user_agent=payload.user_agent) - ctx_opts = {**config["context_options"], "config": config["config"]} + fp_kwargs = {} + if payload.user_agent: + fp_kwargs["user_agent"] = payload.user_agent + fingerprint = load_or_create_fingerprint( + user_data_dir, + # только нижняя граница: реалистичный монитор должен быть не меньше запрошенного окна, + # а точный верхний потолок слишком часто не находится в датасете и тихо "ослабляется" + # browserforge (актуальный размер окна мы всё равно берём из payload.screen ниже, не отсюда) + screen=Screen(min_width=scr_size[0], min_height=scr_size[1]), + **fp_kwargs, + ) - async with AsyncCamoufox( - persistent_context=True, - user_data_dir=user_data_dir, - proxy=proxy, - window=scr_size, - i_know_what_im_doing=True, - geoip=True, - humanize=True, - main_world_eval=True, - **ctx_opts, - ) as camoufo: - await camoufo.add_init_script(config["init_script"]) - page = camoufo.pages[0] if camoufo.pages else await camoufo.new_page() - if payload.cookies: - await camoufo.add_cookies(payload.cookies) - page.set_default_timeout(30_000) - page.set_default_navigation_timeout(60_000) + ctx_kwargs = context_options_for( + fingerprint, + viewport={"width": scr_size[0], "height": scr_size[1]}, + ) - await page.goto(url=payload.url) + geo = await resolve_geo(proxy) + if geo: + if geo.get("timezone_id"): + ctx_kwargs["timezone_id"] = geo["timezone_id"] + if geo.get("geolocation"): + ctx_kwargs["geolocation"] = geo["geolocation"] + ctx_kwargs["permissions"] = ["geolocation"] - actions_result = await actions.execute( - page, payload.actions, payload.captcha, - antibot_name=payload.captcha_type, - success_locator=payload.captcha_success_locator, - detect_locator=payload.captcha_detect_locator, + async with async_playwright() as p: + context = await p.chromium.launch_persistent_context( + user_data_dir=str(user_data_dir), + proxy=proxy, + headless=False, + **ctx_kwargs, ) + try: + await apply_fingerprint(context, fingerprint) + page = context.pages[0] if context.pages else await context.new_page() + if payload.cookies: + await context.add_cookies(payload.cookies) + page.set_default_timeout(30_000) + page.set_default_navigation_timeout(60_000) - status = not "error" in actions_result[-1] - error = ["Something error", None][status] - cookies = await camoufo.cookies() - html = await page.inner_html('html') - # TODO: добавить как опцию. передачу скрина в base64. - # await page.screenshot( - # path=( - # f"/home/sokol/PycharmProjects/WebRoboApi/extra/screenshots/" - # f"{datetime.timestamp(datetime.now())}.png" - # ) - # ) + await page.goto(url=payload.url) + + actions_result = await actions.execute( + page, payload.actions, payload.captcha, + antibot_name=payload.captcha_type, + success_locator=payload.captcha_success_locator, + detect_locator=payload.captcha_detect_locator, + ) + + status = not "error" in actions_result[-1] + error = ["Something error", None][status] + cookies = await context.cookies() + html = await page.inner_html('html') + user_agent = await page.evaluate("() => navigator.userAgent") + # TODO: добавить как опцию. передачу скрина в base64. + # await page.screenshot( + # path=( + # f"/home/sokol/PycharmProjects/WebRoboApi/extra/screenshots/" + # f"{datetime.timestamp(datetime.now())}.png" + # ) + # ) + finally: + await context.close() BROWSER_SLOTS += 1 result = { "status": ["err", "ok"][status], - "user_agent": ctx_opts["user_agent"], + "user_agent": user_agent, "cookies": cookies, "actions": actions_result, }