diff --git a/AGENTS.md b/AGENTS.md index adcdb92..0c8c3d2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -3,7 +3,7 @@ ## Project overview - FastAPI service that mimics FlareSolverr-style API for bypassing anti-bot pages using invisible_playwright. -- Entry point: main app in main.py; routes and request flow defined in src/endpoints.py and src/models.py. +- Entry point: main app in main.py; routes and request flow in src/endpoints.py, challenge handling in src/challenge.py, response bodies in src/content.py, models in src/models.py. - Browser lifecycle is owned by get_browser() in src/utils.py, which yields a page and context for each request. ## Architecture and data flow diff --git a/src/challenge.py b/src/challenge.py new file mode 100644 index 0000000..984eff2 --- /dev/null +++ b/src/challenge.py @@ -0,0 +1,118 @@ +import time +from asyncio import sleep +from contextlib import suppress + +from playwright.async_api import Error as PlaywrightError +from playwright.async_api import FloatRect, Page +from playwright.async_api import TimeoutError as PlaywrightTimeoutError +from playwright_captcha.solvers.click.cloudflare.utils.detection import ( + CF_INTERSTITIAL_INDICATORS_SELECTORS, + detect_cloudflare_challenge, +) + +from src.utils import TimeoutTimer, logger + +__all__ = [ + "CF_INTERSTITIAL_INDICATORS_SELECTORS", + "challenge_present", + "solve_challenge", +] + +POLL_INTERVAL = 0.25 +CLICK_SETTLE = 1.5 +CLICK_COOLDOWN = 4 +PROBE_INTERVAL = 0.5 +TOKEN_READ_TIMEOUT = 1000 +BOX_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 +WIDGET_MAX_HEIGHT = 120 + + +async def challenge_present(page: Page) -> bool: + """Report whether the Cloudflare interstitial is up.""" + return await detect_cloudflare_challenge(page, "interstitial") + + +async def 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 + await widget.first.scroll_into_view_if_needed(timeout=BOX_READ_TIMEOUT) + box = await widget.first.bounding_box(timeout=BOX_READ_TIMEOUT) + if ( + box + and box["width"] > WIDGET_MIN_WIDTH + and WIDGET_MIN_HEIGHT < box["height"] < WIDGET_MAX_HEIGHT + ): + return box + return None + + +async def click_checkbox(page: Page) -> bool: + """Click the checkbox through its container, leaving its closed shadow root alone.""" + box = await widget_box(page) + if box is None: + return False + await page.mouse.move(box["x"] + CHECKBOX_INSET, box["y"] + box["height"] / 2) + try: + await page.mouse.down() + finally: + 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 challenge_present(page): + return False + await sleep(POLL_INTERVAL) + return not await challenge_present(page) + + +async def solve_challenge(page: Page, 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(page): + logger.debug("Challenge cleared.") + if clicks: + await sleep(min(CLICK_SETTLE, timer.remaining())) + return + + if time.perf_counter() >= next_click: + landed = False + with suppress(PlaywrightError, PlaywrightTimeoutError): + landed = not await checkbox_already_answered( + page + ) and await click_checkbox(page) + if landed: + clicks += 1 + logger.info("Clicked the challenge checkbox (attempt %d).", clicks) + next_click = time.perf_counter() + ( + CLICK_COOLDOWN if landed else PROBE_INTERVAL + ) + + if timer.remaining() <= 0: + break + await sleep(POLL_INTERVAL) + + message = "Challenge still present when the request budget ran out" + raise TimeoutError(message) diff --git a/src/content.py b/src/content.py new file mode 100644 index 0000000..d5bc4cf --- /dev/null +++ b/src/content.py @@ -0,0 +1,42 @@ +import base64 + +from playwright.async_api import Page + +from src.models import LinkRequest +from src.utils import logger + + +async def build_response_content( + page: Page, + request: LinkRequest, + page_request: object, + *, + challenge_detected: bool, + page_html: str | None, +) -> tuple[str, str]: + """Build (content_type, response_content) from the settled page.""" + if request.return_only_cookies: + return "text/html", "" + + if page_request and page_request.headers.get("content-type", "").startswith( + "application/pdf" + ): + return await fetch_pdf_content(page) + + response_content = ( + page_html + if page_html is not None and not challenge_detected + else await page.content() + ) + return "text/html", response_content + + +async def fetch_pdf_content(page: Page) -> tuple[str, str]: + """Fetch raw PDF bytes as base64, falling back to viewer HTML on failure.""" + try: + fetch_response = await page.request.fetch(page.url) + response_content = base64.b64encode(await fetch_response.body()).decode("ascii") + except Exception: + logger.exception("Failed to fetch PDF bytes, falling back to viewer HTML") + return "text/html", await page.content() + return "application/pdf", response_content diff --git a/src/endpoints.py b/src/endpoints.py index 5468b87..894afbb 100644 --- a/src/endpoints.py +++ b/src/endpoints.py @@ -1,27 +1,27 @@ -import base64 import time import warnings -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 FloatRect, Page from playwright.async_api import TimeoutError as PlaywrightTimeoutError -from playwright_captcha.solvers.click.cloudflare.utils.detection import ( - detect_cloudflare_challenge, -) +from src.challenge import challenge_present, solve_challenge +from src.content import build_response_content from src.models import ( HealthcheckResponse, LinkRequest, LinkResponse, Solution, ) -from src.utils import BrowserDepClass, TimeoutTimer, get_browser, logger +from src.utils import ( + BrowserDepClass, + TimeoutTimer, + get_browser, + logger, + remaining_ms, +) warnings.filterwarnings("ignore", category=SyntaxWarning) @@ -65,7 +65,7 @@ async def read_item(request: LinkRequest, dep: BrowserDep) -> LinkResponse: await setup_routes(request, dep) try: - challenge_detected, page_html, page_request, status = await _navigate_and_solve( + challenge_detected, page_html, page_request = await _navigate_and_solve( dep, request, timer ) except (TimeoutError, PlaywrightTimeoutError) as e: @@ -77,7 +77,7 @@ async def read_item(request: LinkRequest, dep: BrowserDep) -> LinkResponse: cookies = await dep.context.cookies() content_type, response_content = await build_response_content( - dep, + dep.page, request, page_request, challenge_detected=challenge_detected, @@ -93,7 +93,7 @@ async def read_item(request: LinkRequest, dep: BrowserDep) -> LinkResponse: solution=Solution( user_agent=user_agent, url=dep.page.url, - status=status, + status=HTTPStatus.OK, cookies=cookies, headers=page_request.headers if page_request else {}, response=response_content, @@ -120,170 +120,29 @@ async def _navigate_and_solve( dep: BrowserDep, request: LinkRequest, timer: TimeoutTimer, -) -> tuple[bool, str | None, object, HTTPStatus]: +) -> tuple[bool, str | None, object]: """Navigate to the URL, then solve a challenge or wait for network idle.""" page_html: str | None = None - page_request = await dep.page.goto(request.url, timeout=_remaining_ms(timer)) - status = page_request.status if page_request else HTTPStatus.OK + page_request = await dep.page.goto(request.url, timeout=remaining_ms(timer)) await dep.page.wait_for_load_state( - state="domcontentloaded", timeout=_remaining_ms(timer) + state="domcontentloaded", timeout=remaining_ms(timer) ) - if not await detect_cloudflare_challenge(dep.page, "interstitial"): + if not await challenge_present(dep.page): page_html = await dep.page.content() await _wait_for_networkidle(dep, timer) - return False, page_html, page_request, status + return False, page_html, page_request - await _solve_challenge(dep, timer) + await solve_challenge(dep.page, timer) await _wait_for_networkidle(dep, timer) - status = HTTPStatus.OK - return True, page_html, page_request, status - - -MIN_WAIT_MS = 1.0 - - -def _remaining_ms(timer: TimeoutTimer) -> float: - """Milliseconds left, never 0 - Playwright reads that as no timeout at all.""" - return max(MIN_WAIT_MS, timer.remaining() * 1000) - - -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) -WIDGET_MIN_WIDTH = 40 -WIDGET_MIN_HEIGHT = 20 -WIDGET_MAX_HEIGHT = 120 - - -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 - await widget.first.scroll_into_view_if_needed(timeout=BOX_READ_TIMEOUT) - box = await widget.first.bounding_box(timeout=BOX_READ_TIMEOUT) - if ( - box - and box["width"] > WIDGET_MIN_WIDTH - and WIDGET_MIN_HEIGHT < box["height"] < WIDGET_MAX_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) - try: - await page.mouse.down() - finally: - 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: - landed = False - with suppress(PlaywrightError, PlaywrightTimeoutError): - landed = not await _checkbox_already_answered( - dep.page - ) 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 - await sleep(CHALLENGE_POLL_INTERVAL) - - message = "Challenge still present when the request budget ran out" - raise TimeoutError(message) + return True, page_html, page_request async def _wait_for_networkidle(dep: BrowserDep, timer: TimeoutTimer) -> None: """Wait for network idle, tolerating post-DOM-load stalls.""" try: - await dep.page.wait_for_load_state("networkidle", timeout=_remaining_ms(timer)) + await dep.page.wait_for_load_state("networkidle", timeout=remaining_ms(timer)) except PlaywrightTimeoutError: logger.info( "networkidle timed out after domcontentloaded; continuing with loaded page" ) - - -async def build_response_content( - dep: BrowserDep, - request: LinkRequest, - page_request: object, - *, - challenge_detected: bool, - page_html: str | None, -) -> tuple[str, str]: - """Build (content_type, response_content) from the settled page.""" - if request.return_only_cookies: - return "text/html", "" - - if page_request and page_request.headers.get("content-type", "").startswith( - "application/pdf" - ): - return await _fetch_pdf_content(dep) - - response_content = ( - page_html - if page_html is not None and not challenge_detected - else await dep.page.content() - ) - return "text/html", response_content - - -async def _fetch_pdf_content(dep: BrowserDep) -> tuple[str, str]: - """Fetch raw PDF bytes as base64, falling back to viewer HTML on failure.""" - try: - fetch_response = await dep.page.request.fetch(dep.page.url) - response_content = base64.b64encode(await fetch_response.body()).decode("ascii") - except Exception: - logger.exception("Failed to fetch PDF bytes, falling back to viewer HTML") - return "text/html", await dep.page.content() - return "application/pdf", response_content diff --git a/src/utils.py b/src/utils.py index aca7d52..454323a 100644 --- a/src/utils.py +++ b/src/utils.py @@ -39,6 +39,14 @@ class TimeoutTimer(BaseModel): return max(0, self.duration - (time.perf_counter() - self.start_time)) +MIN_WAIT_MS = 1.0 + + +def remaining_ms(timer: TimeoutTimer) -> float: + """Milliseconds left, never 0 - Playwright reads that as no timeout at all.""" + return max(MIN_WAIT_MS, timer.remaining() * 1000) + + class BrowserDepClass(NamedTuple): page: Page context: BrowserContext diff --git a/tests/main_test.py b/tests/main_test.py index 553305f..02fd602 100644 --- a/tests/main_test.py +++ b/tests/main_test.py @@ -7,15 +7,13 @@ import httpx2 import pytest from fastapi import HTTPException from playwright.async_api import TimeoutError as PlaywrightTimeoutError -from playwright_captcha.solvers.click.cloudflare.utils.detection import ( - CF_INTERSTITIAL_INDICATORS_SELECTORS, -) from starlette.testclient import TestClient from main import app -from src.endpoints import _remaining_ms, read_item +from src.challenge import CF_INTERSTITIAL_INDICATORS_SELECTORS +from src.endpoints import read_item from src.models import LinkRequest -from src.utils import BrowserDepClass, TimeoutTimer +from src.utils import BrowserDepClass, TimeoutTimer, remaining_ms client = TestClient(app) @@ -218,12 +216,27 @@ async def test_domcontentloaded_timeout_returns_408(): assert exc.value.status_code == HTTPStatus.REQUEST_TIMEOUT +@pytest.mark.asyncio +async def test_status_is_always_ok_like_flaresolverr(): + """FlareSolverr hardcodes 200 because Selenium cannot report the real code.""" + dep = fake_dep() + dep.page.goto.return_value = MagicMock( + status=HTTPStatus.FORBIDDEN, + headers={"content-type": "text/html"}, + request=MagicMock(headers={"user-agent": "UnitTestBrowser/1.0"}), + ) + + response = await read_item(LinkRequest(url="https://example.test/login"), dep) + + assert response.solution.status == HTTPStatus.OK + + def test_exhausted_budget_never_disables_playwright_timeouts(): """Playwright reads timeout=0 as no timeout at all, so the floor must hold.""" spent = TimeoutTimer(duration=0) assert spent.remaining() == 0 - assert _remaining_ms(spent) > 0 + assert remaining_ms(spent) > 0 @pytest.mark.asyncio