diff --git a/src/endpoints.py b/src/endpoints.py index 3d10c49..0a8a3e7 100644 --- a/src/endpoints.py +++ b/src/endpoints.py @@ -1,7 +1,7 @@ import base64 import time import warnings -from asyncio import sleep, wait_for +from asyncio import sleep from http import HTTPStatus from typing import Annotated @@ -9,14 +9,9 @@ from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import RedirectResponse 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, @@ -46,10 +41,18 @@ CHALLENGE_MARKERS = ( "#challenge-error-text, #challenge-running, #challenge-stage" ) -# How long to let Cloudflare verify a click before trying again, and how often -# to look. Verification took 5-15s in testing. -CHALLENGE_SETTLE_SECONDS = 20.0 +# The widget lives in an iframe served from here; the checkbox is an invisible +# input inside it, so it is pressed by position rather than by locator. 30px in +# from the widget's left edge is the middle of the box Cloudflare draws. +CF_WIDGET_HOST = "challenges.cloudflare.com" +CHECKBOX_SELECTOR = 'input[type="checkbox"]' +CHECKBOX_OFFSET_X = 30 + +# How often to look, and how long to leave a press alone before trying again. +# Verification takes 5-15s, and pressing over the top of it just restarts the +# cycle. CHALLENGE_POLL_SECONDS = 1.0 +PRESS_INTERVAL_SECONDS = 12.0 @router.get("/", include_in_schema=False) @@ -165,59 +168,108 @@ async def _solve_challenge(dep: BrowserDep, timer: TimeoutTimer) -> None: """ Attempt to solve a detected Cloudflare interstitial challenge. - The solver's own verdict is not trusted. It judges its click by waiting for - networkidle, which returns the moment the network happens to be quiet -- - measured at 9ms after the click -- and then reports "challenge still - present". Cloudflare needs seconds: it replaces the checkbox with - "verifying you are human" before the page turns over. So try, then wait for - the challenge itself to go away, and only give up once the budget is gone. + Handles both shapes Cloudflare serves: the non-interactive challenge, which + clears itself given a few seconds, and the interactive one, which needs the + checkbox pressed. Both are covered by the same loop -- watch for the + challenge markup to disappear, and press whenever a checkbox is on offer. - That bounds what used to be unbounded. With max_attempts at sys.maxsize the - solver retried an unreachable widget ~1300 times per request and the caller - waited out the whole max_timeout for a 408 it was always going to get. + playwright-captcha's solver is deliberately not used here. It clicks the + checkbox input directly, and that input is invisible, so the click reports + success while `checked` never flips. It also judges the result by waiting + for networkidle, which returned 9ms after the click while Cloudflare was + still verifying, so it reported failure on challenges that were about to + pass. """ logger.info("Challenge detected, attempting to solve...") + last_press = -PRESS_INTERVAL_SECONDS while timer.remaining() > 0: - try: - 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=timer.remaining(), - ) - except (CaptchaDetectionError, CaptchaSolvingError) as e: - # Both mean "not solved yet", not "cannot be solved": the widget is - # mid-verification and momentarily absent, or the click has landed - # and the verdict was taken too early. - logger.debug(f"Solver attempt inconclusive: {e}") - - if await _challenge_cleared(dep, timer): + if not await _challenge_visible(dep.page): logger.info("Challenge cleared") return + elapsed = timer.duration - timer.remaining() + if elapsed - last_press >= PRESS_INTERVAL_SECONDS and await _press_checkbox( + dep.page + ): + last_press = elapsed + + await sleep(CHALLENGE_POLL_SECONDS) + message = "Challenge still present when the request budget ran out" raise TimeoutError(message) -async def _challenge_cleared(dep: BrowserDep, timer: TimeoutTimer) -> bool: - """ - Wait out Cloudflare's post-click verification, up to CHALLENGE_SETTLE_SECONDS. +def _cloudflare_frame(page: Page) -> object | None: + """Find the turnstile widget's frame; None while Cloudflare is between states.""" + for frame in page.frames: + if CF_WIDGET_HOST in frame.url and not frame.is_detached(): + return frame + return None - Polls rather than waiting for a load event: Cloudflare swaps the widget for - a spinner in place and only navigates once it is satisfied, so there is no - single event to await. + +async def _press_point(page: Page) -> tuple[float, float] | None: """ - deadline = min(CHALLENGE_SETTLE_SECONDS, timer.remaining()) - waited = 0.0 - while waited < deadline: - if not await _challenge_visible(dep.page): - return True - await sleep(CHALLENGE_POLL_SECONDS) - waited += CHALLENGE_POLL_SECONDS - return not await _challenge_visible(dep.page) + Where to press, or None when there is nothing to press right now. + + Cloudflare cycles between "checking if you are human", where the widget + frame holds no input at all, and the state where the checkbox is offered, + so the input has to exist before reaching for the mouse. A box that is + already checked is skipped too: a press has landed and is being verified, + and pressing over the top restarts that verification -- which is how ext.to + and speed.cd stayed on "performing security verification" for a full 300s + budget while being pressed a dozen times. + + The point is measured from the iframe rather than the input, because the + input is invisible: it sits under a styled overlay, so Playwright reports a + successful click on it while `checked` never flips. That is why + playwright-captcha's own click has never solved one of these. + """ + frame = _cloudflare_frame(page) + if frame is None: + return None + try: + checkbox = frame.locator(CHECKBOX_SELECTOR) + if not await checkbox.count() or await checkbox.first.is_checked(): + return None + element = await frame.frame_element() + box = await element.bounding_box() + except Exception: + # The widget is mid-swap; try again on the next poll. + return None + if not box: + return None + return box["x"] + CHECKBOX_OFFSET_X, box["y"] + box["height"] / 2 + + +async def _press_checkbox(page: Page) -> bool: + """ + Press the widget's visible pixels. True when a press actually happened. + + Measured against extratorrent.st and ext.to on a residential connection, + this clears the challenge and returns a cf_clearance cookie, where clicking + the input never did. + """ + point = await _press_point(page) + if point is None: + return False + x, y = point + try: + # Approach before pressing: a cursor that teleports onto the target is + # itself a signal. + await page.mouse.move(x - 180, y - 120) + await sleep(0.3) + await page.mouse.move(x - 45, y - 20, steps=18) + await sleep(0.2) + await page.mouse.move(x, y, steps=10) + await sleep(0.35) + await page.mouse.down() + await sleep(0.08) + await page.mouse.up() + except Exception as exc: + logger.debug(f"Checkbox press failed: {exc}") + return False + logger.info(f"Pressed the Cloudflare checkbox at ({x:.0f}, {y:.0f})") + return True async def _challenge_visible(page: Page) -> bool: diff --git a/tests/main_test.py b/tests/main_test.py index 8b43de5..9c1209b 100644 --- a/tests/main_test.py +++ b/tests/main_test.py @@ -9,14 +9,10 @@ import httpx2 import pytest from fastapi import HTTPException from playwright.async_api import TimeoutError as PlaywrightTimeoutError -from playwright_captcha.utils.exceptions import ( - CaptchaDetectionError, - CaptchaSolvingError, -) from starlette.testclient import TestClient from main import app -from src.endpoints import CHALLENGE_MARKERS, read_item +from src.endpoints import CHALLENGE_MARKERS, CHECKBOX_OFFSET_X, read_item from src.models import LinkRequest from src.utils import BrowserDepClass @@ -26,6 +22,9 @@ client = TestClient(app) # 52. A small margin absorbs Firefox version drift without letting 52 through. FIREFOX_CIPHER_SUITE_CEILING = 20 +# The turnstile iframe's geometry, as measured on a real challenge page. +WIDGET_BOX = {"x": 512.0, "y": 304.0, "width": 300.0, "height": 65.0} + # Sites Byparr clears from any network, datacenter ranges included. These carry # the hard assertion: if the bypass breaks, one of these goes red. test_websites = [ @@ -203,18 +202,40 @@ def test_max_timeout_normalization(payload: dict, expected: int): assert request.max_timeout == expected +def fake_cloudflare_frame(*, checked: bool) -> MagicMock: + """Build a turnstile widget frame offering one checkbox in the given state.""" + frame = MagicMock() + frame.url = ( + "https://challenges.cloudflare.com/cdn-cgi/challenge-platform/h/b/turnstile" + ) + frame.is_detached = MagicMock(return_value=False) + + checkbox = MagicMock() + checkbox.count = AsyncMock(return_value=1) + checkbox.first.is_checked = AsyncMock(return_value=checked) + frame.locator = MagicMock(return_value=checkbox) + + element = AsyncMock() + element.bounding_box.return_value = WIDGET_BOX + frame.frame_element = AsyncMock(return_value=element) + return frame + + def fake_dep( *, fail_states: set[str] | None = None, challenged: bool = False, marker_counts: list[int] | None = None, + checkbox: str | None = None, ) -> BrowserDepClass: """ Build a browser dependency triple backed by mocks. `challenged` makes the detector report a Cloudflare challenge. - `marker_counts` drives the "is it still up?" check that runs after each - solve attempt: one entry per look, the last one repeating forever. + `marker_counts` drives the "is it still up?" check that runs on each poll: + one entry per look, the last one repeating forever. + `checkbox` puts a widget frame on the page with the box "checked" or + "unchecked"; without it the page carries no widget at all. """ page = AsyncMock() page.url = "https://example.test/login" @@ -250,6 +271,11 @@ def fake_dep( raise PlaywrightTimeoutError(message) page.wait_for_load_state.side_effect = wait_for_load_state + page.frames = ( + [] + if checkbox is None + else [fake_cloudflare_frame(checked=checkbox == "checked")] + ) context = AsyncMock() context.cookies.return_value = [] @@ -268,7 +294,7 @@ async def test_networkidle_timeout_after_domcontentloaded_returns_content(): assert response.status == "ok" assert response.solution.status == HTTPStatus.OK assert response.solution.response == "Login" - dep.solver.solve_captcha.assert_not_called() + dep.page.mouse.down.assert_not_called() @pytest.mark.asyncio @@ -284,19 +310,16 @@ async def test_domcontentloaded_timeout_returns_408(): @pytest.mark.asyncio -async def test_challenge_that_clears_after_the_click_succeeds(): +async def test_challenge_that_clears_is_reported_as_success(): """ - The solver's own "challenge still present" verdict must not end the request. + A challenge is over when its markup goes away, not when a solver says so. - It judges its click by waiting for networkidle, which returns as soon as the - network happens to be quiet -- 9ms after the click, in practice -- while - Cloudflare is still showing "verifying you are human". Byparr has to wait - for the challenge markup itself to go away. + playwright-captcha judged its own click by waiting for networkidle, which + returns as soon as the network happens to be quiet -- 9ms after the click, + in practice -- while Cloudflare is still showing "verifying you are human", + and then reported failure on challenges that were about to pass. """ dep = fake_dep(challenged=True, marker_counts=[1, 0]) - dep.solver.solve_captcha.side_effect = CaptchaSolvingError( - "challenge still present or expected content not detected" - ) response = await read_item( LinkRequest(url="https://example.test/login", max_timeout=5), dep @@ -306,13 +329,46 @@ async def test_challenge_that_clears_after_the_click_succeeds(): assert response.solution.status == HTTPStatus.OK +@pytest.mark.asyncio +async def test_unchecked_box_is_pressed_on_the_widgets_visible_pixels(): + """ + The press must land on the widget, not on the input. + + The input is invisible -- it sits under a styled overlay -- so a click on it + reports success while `checked` never flips. Pressing the pixels Cloudflare + actually draws is what clears the challenge. + """ + dep = fake_dep(challenged=True, marker_counts=[1, 1, 0], checkbox="unchecked") + + await read_item(LinkRequest(url="https://example.test/login", max_timeout=5), dep) + + dep.page.mouse.down.assert_called() + assert dep.page.mouse.move.call_args.args[:2] == ( + WIDGET_BOX["x"] + CHECKBOX_OFFSET_X, + WIDGET_BOX["y"] + WIDGET_BOX["height"] / 2, + ) + + +@pytest.mark.asyncio +async def test_checked_box_is_left_alone_while_cloudflare_verifies(): + """ + Pressing a box that is already checked restarts Cloudflare's verification. + + ext.to and speed.cd sat on "performing security verification" for a full + 300s budget while being pressed a dozen times, never getting far enough + into the check to finish it. + """ + dep = fake_dep(challenged=True, marker_counts=[1, 1, 0], checkbox="checked") + + await read_item(LinkRequest(url="https://example.test/login", max_timeout=5), dep) + + dep.page.mouse.down.assert_not_called() + + @pytest.mark.asyncio async def test_challenge_that_never_clears_returns_408(): """A challenge still up when the budget runs out is a timeout, not a 500.""" dep = fake_dep(challenged=True, marker_counts=[1]) - dep.solver.solve_captcha.side_effect = CaptchaDetectionError( - "Cloudflare iframes not found" - ) with pytest.raises(HTTPException) as exc: await read_item(