From 426aae431011b9b4f535302f8f2c3401f89ee2cb Mon Sep 17 00:00:00 2001 From: ThePhaseless Date: Tue, 18 Aug 2026 03:30:49 +0200 Subject: [PATCH] fix: address review findings on the challenge solver nowsecure.nl carries no interstitial, so the turnstile detector only widened the entry condition without ever being able to satisfy the exit one, leaving a turnstile-only page reported as solved on the first poll. Detect on the interstitial alone. Restore the user-agent guard that went missing: an absent header made Solution reject None and turned into an unhandled 500, where it used to degrade to an empty string. Fall back to "" directly rather than reaching for evaluate(), which CDP refuses. Bound the widget measurement, which inherited Playwright's 30s default at each of four depths and so could run far past the request budget, throttle the probe when no click lands so the scan no longer repeats every tick, and release the mouse button through a finally so a failed press cannot leave it held down. Cover the click path: the fixture pinned bounding_box to None, so nothing exercised the code this branch exists to add. Co-Authored-By: Claude Opus 5 --- src/endpoints.py | 37 ++++++++++++++++++++++--------------- tests/main_test.py | 42 ++++++++++++++++++++++++++++-------------- 2 files changed, 50 insertions(+), 29 deletions(-) diff --git a/src/endpoints.py b/src/endpoints.py index 81ad37e..6ba329d 100644 --- a/src/endpoints.py +++ b/src/endpoints.py @@ -9,7 +9,7 @@ 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 FloatRect, Page from playwright.async_api import TimeoutError as PlaywrightTimeoutError from playwright_captcha.solvers.click.cloudflare.utils.detection import ( detect_cloudflare_challenge, @@ -84,7 +84,9 @@ async def read_item(request: LinkRequest, dep: BrowserDep) -> LinkResponse: page_html=page_html, ) - user_agent = page_request.request.headers.get("user-agent") if page_request else "" + user_agent = ( + page_request.request.headers.get("user-agent") or "" if page_request else "" + ) return LinkResponse( message="Success", @@ -127,10 +129,7 @@ async def _navigate_and_solve( state="domcontentloaded", timeout=timer.remaining() * 1000 ) - challenge_active = await detect_cloudflare_challenge( - dep.page, "interstitial" - ) or await detect_cloudflare_challenge(dep.page, "turnstile") - if not challenge_active: + if not await detect_cloudflare_challenge(dep.page, "interstitial"): page_html = await dep.page.content() await _wait_for_networkidle(dep, timer) return False, page_html, page_request, status @@ -144,6 +143,8 @@ CHALLENGE_POLL_INTERVAL = 0.25 CHALLENGE_CLICK_SETTLE = 1.5 CHECKBOX_CLICK_COOLDOWN = 4 TOKEN_READ_TIMEOUT = 1000 +BOX_READ_TIMEOUT = 1000 +CHECKBOX_PROBE_INTERVAL = 0.5 CHECKBOX_INSET = 25 TURNSTILE_INPUT = 'input[name="cf-turnstile-response"]' WIDGET_ANCESTOR_DEPTHS = (1, 2, 3, 4) @@ -151,14 +152,14 @@ WIDGET_MIN_WIDTH = 40 WIDGET_MIN_HEIGHT = 20 -async def _challenge_widget_box(page: Page) -> dict[str, float] | None: +async def _challenge_widget_box(page: Page) -> FloatRect | 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() + box = await widget.first.bounding_box(timeout=BOX_READ_TIMEOUT) if ( box and box["width"] > WIDGET_MIN_WIDTH @@ -174,8 +175,10 @@ async def _click_challenge_checkbox(page: Page) -> bool: 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() + try: + await page.mouse.down() + finally: + await page.mouse.up() return True @@ -209,13 +212,17 @@ async def _solve_challenge(dep: BrowserDep, timer: TimeoutTimer) -> None: return if time.perf_counter() >= next_click: + landed = False with suppress(PlaywrightError, PlaywrightTimeoutError): - if not await _checkbox_already_answered( + landed = 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) + ) and await _click_challenge_checkbox(dep.page) + if landed: + clicks += 1 + logger.info("Clicked the challenge checkbox (attempt %d).", clicks) + next_click = time.perf_counter() + ( + CHECKBOX_CLICK_COOLDOWN if landed else CHECKBOX_PROBE_INTERVAL + ) if timer.remaining() <= 0: break diff --git a/tests/main_test.py b/tests/main_test.py index f2915ca..d8218a9 100644 --- a/tests/main_test.py +++ b/tests/main_test.py @@ -54,7 +54,7 @@ def test_bypass(website: str): response = client.post( "/v1", json=LinkRequest.model_construct( - url=website, cmd="request.get", max_timeout=360 + url=website, cmd="request.get", max_timeout=60 ).model_dump(), ) @@ -146,6 +146,8 @@ def fake_dep( fail_states: set[str] | None = None, challenged: bool = False, marker_counts: list[int] | None = None, + widget_box: dict[str, float] | None = None, + user_agent: str | None = "UnitTestBrowser/1.0", ) -> BrowserDepClass: """Build a browser dependency pair backed by mocks.""" page = AsyncMock() @@ -153,7 +155,7 @@ def fake_dep( page.goto.return_value = MagicMock( status=HTTPStatus.OK, headers={"content-type": "text/html"}, - request=MagicMock(headers={"user-agent": "UnitTestBrowser/1.0"}), + request=MagicMock(headers={"user-agent": user_agent} if user_agent else {}), ) page.title.return_value = "Login" page.evaluate.return_value = "UnitTestBrowser/1.0" @@ -169,7 +171,7 @@ 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.bounding_box = AsyncMock(return_value=widget_box) handle.first.input_value = AsyncMock(return_value="") return handle @@ -215,22 +217,34 @@ async def test_domcontentloaded_timeout_returns_408(): @pytest.mark.asyncio -async def test_user_agent_survives_csp_blocked_evaluate(): - """UA comes from request headers when page CSP blocks evaluate (#394). - - No CSP configuration (header, meta tag, or internal viewer document) may - turn /v1 into a 500. - """ - dep = fake_dep() - dep.page.evaluate.side_effect = Exception("call to eval() blocked by CSP") - +async def test_missing_user_agent_header_is_not_a_500(): + """A request without a user-agent header degrades to empty, never a 500 (#394).""" response = await read_item( LinkRequest(url="https://example.test/login"), - dep, + fake_dep(user_agent=None), ) assert response.status == "ok" - assert response.solution.user_agent == "UnitTestBrowser/1.0" + assert response.solution.user_agent == "" + + +@pytest.mark.asyncio +async def test_checkbox_is_clicked_while_the_challenge_is_up(): + """A measurable widget gets a humanised press, not a raw synthetic click.""" + dep = fake_dep( + challenged=True, + marker_counts=[1, 1, 0], + widget_box={"x": 100.0, "y": 200.0, "width": 300.0, "height": 60.0}, + ) + + response = await read_item( + LinkRequest(url="https://example.test/login", max_timeout=5), dep + ) + + assert response.solution.status == HTTPStatus.OK + dep.page.mouse.move.assert_awaited_once_with(125.0, 230.0) + dep.page.mouse.down.assert_awaited_once() + dep.page.mouse.up.assert_awaited_once() @pytest.mark.asyncio