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