starred/Byparr
1
0
Fork 0
mirror of https://github.com/ThePhaseless/Byparr.git synced 2026-09-24 06:10:14 +01:00
Code Issues Packages Projects Releases 10 Wiki Activity

fix: click the challenge checkbox without tripping Cloudflare's detector

Browse Source 
Measuring the widget with page.evaluate() made Cloudflare reissue the
challenge every few seconds, so the interstitial never cleared. Locate the
container with locators instead and click it through the mouse, which leaves
its closed shadow root untouched.

Keep the solver lazy as well: ClickSolver.prepare() patches attachShadow,
which is what escalated a self-clearing challenge into a checkbox in the
first place.

A click can land while the widget is still self-verifying, so retry on a
cooldown for as long as the request budget lasts rather than stopping after
the first one, and confirm the marker is gone twice before reporting success
since it drops out between challenge rounds.

The bypass tests now pass on their own, so drop the xfail marks.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
ThePhaseless
2026-08-17 17:45:15 +02:00
co-authored by Claude Opus 5
parent 9067ca37df
commit 61db8d82ee
4 changed files with 106 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+1 -1
View File
@@ -51,7 +51,7 @@ RUN mkdir -p /home/byparr &&\
FROM app AS test
RUN \
uv sync --group test &&\
uv run pytest -rs --retries 3
uv run pytest -rs --retries 5
FROM app
ARG VERSION
src/endpoints.py
+81 -27
View File
@@ -1,22 +1,19 @@
import base64
import time
import warnings
from asyncio import wait_for
from asyncio import sleep
from contextlib import suppress
from http import HTTPStatus
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import RedirectResponse
from playwright.async_api import Error as PlaywrightError
from playwright.async_api import Page
from playwright.async_api import TimeoutError as PlaywrightTimeoutError
from playwright_captcha import CaptchaType
from playwright_captcha.solvers.click.cloudflare.utils.detection import (
detect_cloudflare_challenge,
)
from playwright_captcha.utils.exceptions import (
CaptchaDetectionError,
CaptchaSolvingError,
)
from src.models import (
HealthcheckResponse,
@@ -143,30 +140,87 @@ async def _navigate_and_solve(
return True, page_html, page_request, status
async def _solve_challenge(dep: BrowserDep, timer: TimeoutTimer) -> None:
"""Attempt to solve a detected Cloudflare interstitial challenge."""
logger.info("Challenge detected, attempting to solve...")
while timer.remaining() > 0:
with suppress(
TimeoutError,
PlaywrightTimeoutError,
CaptchaDetectionError,
CaptchaSolvingError,
):
await wait_for(
dep.solver.solve_captcha( # pyright: ignore[reportUnknownMemberType,reportUnknownArgumentType]
captcha_container=dep.page,
captcha_type=CaptchaType.CLOUDFLARE_INTERSTITIAL,
wait_checkbox_attempts=1,
wait_checkbox_delay=0.5,
),
timeout=min(15, timer.remaining()),
)
CHALLENGE_POLL_INTERVAL = 0.25
CHALLENGE_CLICK_SETTLE = 1.5
CHECKBOX_CLICK_COOLDOWN = 4
TOKEN_READ_TIMEOUT = 1000
CHECKBOX_INSET = 25
TURNSTILE_INPUT = 'input[name="cf-turnstile-response"]'
WIDGET_ANCESTOR_DEPTHS = (1, 2, 3, 4)
WIDGET_MIN_WIDTH = 40
WIDGET_MIN_HEIGHT = 20
if not await detect_cloudflare_challenge(dep.page, "interstitial"):
logger.debug("Challenge solved successfully.")
async def _challenge_widget_box(page: Page) -> dict[str, float] | None:
"""Measure the widget container with locators; running page scripts resets the challenge."""
for depth in WIDGET_ANCESTOR_DEPTHS:
widget = page.locator(f"{TURNSTILE_INPUT} >> xpath=ancestor::div[{depth}]")
with suppress(PlaywrightError, PlaywrightTimeoutError):
if await widget.count() == 0:
continue
box = await widget.first.bounding_box()
if (
box
and box["width"] > WIDGET_MIN_WIDTH
and box["height"] > WIDGET_MIN_HEIGHT
):
return box
return None
async def _click_challenge_checkbox(page: Page) -> bool:
"""Click the checkbox through its container, leaving its closed shadow root alone."""
box = await _challenge_widget_box(page)
if box is None:
return False
await page.mouse.move(box["x"] + CHECKBOX_INSET, box["y"] + box["height"] / 2)
await page.mouse.down()
await page.mouse.up()
return True
async def _checkbox_already_answered(page: Page) -> bool:
"""Report whether Turnstile has already filled in its response token."""
token = page.locator(TURNSTILE_INPUT)
with suppress(PlaywrightError, PlaywrightTimeoutError):
if await token.count() > 0:
return bool(await token.first.input_value(timeout=TOKEN_READ_TIMEOUT))
return False
async def _challenge_is_gone(page: Page) -> bool:
"""Confirm the interstitial is really gone and not just between navigations."""
if await detect_cloudflare_challenge(page, "interstitial"):
return False
await sleep(CHALLENGE_POLL_INTERVAL)
return not await detect_cloudflare_challenge(page, "interstitial")
async def _solve_challenge(dep: BrowserDep, timer: TimeoutTimer) -> None:
"""Wait out the interstitial, clicking its checkbox whenever one is offered."""
logger.info("Challenge detected, waiting for it to clear...")
clicks = 0
next_click = 0.0
while True:
if await _challenge_is_gone(dep.page):
logger.debug("Challenge cleared.")
if clicks:
await sleep(min(CHALLENGE_CLICK_SETTLE, timer.remaining()))
return
if time.perf_counter() >= next_click:
with suppress(PlaywrightError, PlaywrightTimeoutError):
if not await _checkbox_already_answered(
dep.page
) and await _click_challenge_checkbox(dep.page):
clicks += 1
next_click = time.perf_counter() + CHECKBOX_CLICK_COOLDOWN
logger.info("Clicked the challenge checkbox (attempt %d).", clicks)
if timer.remaining() <= 0:
break
await sleep(CHALLENGE_POLL_INTERVAL)
message = "Challenge still present when the request budget ran out"
raise TimeoutError(message)
src/utils.py
+5 -2
View File
@@ -106,10 +106,13 @@ async def get_browser(
browser = cast("Browser", browser_raw)
context = await browser.new_context()
page = await context.new_page()
async with ClickSolver(
solver = ClickSolver(
framework=FrameworkType.PLAYWRIGHT,
page=page,
max_attempts=MAX_ATTEMPTS,
attempt_delay=1,
) as solver:
)
try:
yield BrowserDepClass(page, solver, context)
finally:
await solver.cleanup()
tests/main_test.py
+19 -9
View File
@@ -23,18 +23,13 @@ from src.utils import BrowserDepClass
client = TestClient(app)
cloudflare_refuses = pytest.mark.xfail(
reason="Cloudflare rejects the interactive challenge click",
strict=False,
)
test_websites = [
pytest.param("https://ext.to/", marks=cloudflare_refuses),
"https://ext.to/",
# "https://www.ygg.re/",
pytest.param("https://extratorrent.st/", marks=cloudflare_refuses),
pytest.param("https://speed.cd/login", marks=cloudflare_refuses),
"https://extratorrent.st/",
"https://speed.cd/login",
'https://www.yggtorrent.top/engine/search?do=search&order=desc&sort=publish_date&name="UNESCAPED"+"DOUBLEQUOTES"&category=2145',
pytest.param("https://1337x.to/home/", marks=cloudflare_refuses),
"https://1337x.to/home/",
]
@@ -175,6 +170,8 @@ def fake_dep(
def locator(selector: str) -> MagicMock:
handle = MagicMock()
handle.count = AsyncMock(side_effect=lambda: count_for(selector))
handle.first.bounding_box = AsyncMock(return_value=None)
handle.first.input_value = AsyncMock(return_value="")
return handle
page.locator = MagicMock(side_effect=locator)
@@ -266,3 +263,16 @@ async def test_challenge_that_never_clears_returns_408():
)
assert exc.value.status_code == HTTPStatus.REQUEST_TIMEOUT
@pytest.mark.asyncio
async def test_marker_vanishing_mid_navigation_is_not_a_solved_challenge():
"""The marker drops out between challenge rounds; one clear read proves nothing."""
dep = fake_dep(challenged=True, marker_counts=[1, 0, 1])
with pytest.raises(HTTPException) as exc:
await read_item(
LinkRequest(url="https://example.test/login", max_timeout=2), dep
)
assert exc.value.status_code == HTTPStatus.REQUEST_TIMEOUT