diff --git a/Dockerfile b/Dockerfile index fc785ad..a98b2bc 100644 --- a/Dockerfile +++ b/Dockerfile @@ -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 diff --git a/src/endpoints.py b/src/endpoints.py index 4c5a449..81ad37e 100644 --- a/src/endpoints.py +++ b/src/endpoints.py @@ -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) diff --git a/src/utils.py b/src/utils.py index 54a7c98..baba58c 100644 --- a/src/utils.py +++ b/src/utils.py @@ -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() diff --git a/tests/main_test.py b/tests/main_test.py index 0d6bd57..6a1a1d3 100644 --- a/tests/main_test.py +++ b/tests/main_test.py @@ -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