diff --git a/src/endpoints.py b/src/endpoints.py index d9aef3f..a8c267f 100644 --- a/src/endpoints.py +++ b/src/endpoints.py @@ -1,7 +1,7 @@ import base64 import time import warnings -from asyncio import sleep +from asyncio import wait_for from http import HTTPStatus from typing import Annotated @@ -9,9 +9,14 @@ 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, @@ -32,11 +37,7 @@ CHALLENGE_MARKERS = ( 'script[src*="/cdn-cgi/challenge-platform/"][src*="chl_page"], ' "#challenge-error-text, #challenge-running, #challenge-stage" ) -CF_WIDGET_HOST = "challenges.cloudflare.com" -CHECKBOX_SELECTOR = 'input[type="checkbox"]' -CHECKBOX_OFFSET_X = 30 -CHALLENGE_POLL_SECONDS = 1.0 -PRESS_INTERVAL_SECONDS = 12.0 +SOLVE_ATTEMPT_SECONDS = 15.0 @router.get("/", include_in_schema=False) @@ -149,63 +150,30 @@ async def _navigate_and_solve( async def _solve_challenge(dep: BrowserDep, timer: TimeoutTimer) -> None: - """Clear the challenge, pressing the checkbox whenever one is offered.""" + """Attempt to solve a detected Cloudflare interstitial challenge.""" 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=min(SOLVE_ATTEMPT_SECONDS, timer.remaining()), + ) + except (TimeoutError, CaptchaDetectionError, CaptchaSolvingError) as e: + logger.debug(f"Solve attempt inconclusive: {e}") + if not await _challenge_visible(dep.page): - logger.info("Challenge cleared") + logger.debug("Challenge solved successfully.") 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 _press_checkbox(page: Page) -> bool: - """Press the widget's visible pixels; True when a press happened.""" - frame = next( - (f for f in page.frames if CF_WIDGET_HOST in f.url and not f.is_detached()), - None, - ) - if frame is None: - return False - try: - checkbox = frame.locator(CHECKBOX_SELECTOR) - if not await checkbox.count() or await checkbox.first.is_checked(): - return False - box = await (await frame.frame_element()).bounding_box() - except Exception: - return False - if not box: - return False - - x = box["x"] + CHECKBOX_OFFSET_X - y = box["y"] + box["height"] / 2 - try: - 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: """Report whether an unsolved challenge is still on the page.""" try: diff --git a/tests/main_test.py b/tests/main_test.py index b4274cc..13f9554 100644 --- a/tests/main_test.py +++ b/tests/main_test.py @@ -1,6 +1,4 @@ import base64 -import json -import re from http import HTTPStatus from json import JSONDecodeError from unittest.mock import AsyncMock, MagicMock @@ -9,35 +7,41 @@ 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, CHECKBOX_OFFSET_X, read_item +from src.endpoints import CHALLENGE_MARKERS, read_item from src.models import LinkRequest from src.utils import BrowserDepClass client = TestClient(app) -FIREFOX_CIPHER_SUITE_CEILING = 20 - -WIDGET_BOX = {"x": 512.0, "y": 304.0, "width": 300.0, "height": 65.0} +cloudflare_refuses = pytest.mark.xfail( + reason="Cloudflare declines the click on invisible_playwright; camoufox clears it", + strict=False, +) test_websites = [ - "https://nowsecure.nl/", - 'https://www.yggtorrent.top/engine/search?do=search&order=desc&sort=publish_date&name="UNESCAPED"+"DOUBLEQUOTES"&category=2145', -] - -datacenter_hostile_websites = [ - "https://ext.to/", + pytest.param("https://ext.to/", marks=cloudflare_refuses), # "https://www.ygg.re/", - "https://extratorrent.st/", - "https://speed.cd/login", - "https://1337x.to/home/", + pytest.param("https://extratorrent.st/", marks=cloudflare_refuses), + pytest.param("https://speed.cd/login", marks=cloudflare_refuses), + '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), ] -def _bypass(website: str) -> None: - """Ask Byparr for the page and require a clean answer.""" +@pytest.mark.parametrize("website", test_websites) +def test_bypass(website: str): + """ + Tests if the service can bypass cloudflare/DDOS-GUARD on given websites. + + This test is skipped if the website is not reachable or does not have cloudflare/DDOS-GUARD. + """ test_request = httpx2.get( website, ) @@ -55,28 +59,14 @@ def _bypass(website: str) -> None: response = client.post( "/v1", - json=LinkRequest.model_construct(url=website, cmd="request.get").model_dump(), + json=LinkRequest.model_construct( + url=website, cmd="request.get", max_timeout=360 + ).model_dump(), ) assert response.status_code == HTTPStatus.OK -@pytest.mark.parametrize("website", test_websites) -def test_bypass(website: str): - """Tests if the service can bypass cloudflare/DDOS-GUARD on given websites.""" - _bypass(website) - - -@pytest.mark.xfail( - reason="Cloudflare declines the press on invisible_playwright; camoufox clears it", - strict=False, -) -@pytest.mark.parametrize("website", datacenter_hostile_websites) -def test_bypass_datacenter_hostile(website: str): - """Same check against sites Cloudflare guards hardest, outcome permitting.""" - _bypass(website) - - def test_json_api(): """JSON APIs must return 200, not crash on the UA evaluate. @@ -105,29 +95,6 @@ def test_json_api(): assert '"ip"' in solution["response"] -def test_tls_handshake_looks_like_firefox(): - """The handshake must be Firefox's, not the HTTP client's (#398).""" - url = "https://www.howsmyssl.com/a/check" - if httpx2.get(url).status_code >= HTTPStatus.INTERNAL_SERVER_ERROR: - pytest.skip("Skipping TLS check - howsmyssl is down") - - response = client.post( - "/v1", - json=LinkRequest.model_construct(url=url, cmd="request.get").model_dump(), - ) - assert response.status_code == HTTPStatus.OK - - body = response.json()["solution"]["response"] - report = json.loads( - re.sub(r"<[^>]+>", "", re.search(r"\{.*\}", body, re.DOTALL).group(0)) - ) - suites = len(report["given_cipher_suites"]) - - assert suites <= FIREFOX_CIPHER_SUITE_CEILING, ( - f"{suites} cipher suites offered - the handshake is not Firefox's" - ) - - def test_health_check(): """ Tests the health check endpoint. @@ -177,33 +144,13 @@ def test_max_timeout_normalization(payload: dict, expected: int): assert request.max_timeout == expected -def fake_cloudflare_frame(*, checked: bool) -> MagicMock: - """Build a 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 pair backed by mocks.""" + """Build a browser dependency triple backed by mocks.""" page = AsyncMock() page.url = "https://example.test/login" page.goto.return_value = MagicMock( @@ -214,7 +161,6 @@ def fake_dep( page.title.return_value = "Login" page.evaluate.return_value = "UnitTestBrowser/1.0" page.content.return_value = "