Skip to content

feat: locale agnostic cloudflare captcha solver #412

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
subediparas5 wants to merge 1 commit into from
Closed

feat: locale agnostic cloudflare captcha solver #412

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 26 additions & 26 deletions scrapling/engines/_browsers/_stealth.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,20 +118,17 @@ def _cloudflare_solver(self, page: Page) -> None: # pragma: no cover
else:
log.info(f'The turnstile version discovered is "{challenge_type}"')
if challenge_type == "non-interactive":
while "<title>Just a moment...</title>" in (ResponseFactory._get_page_content(page)):
while self._detect_cloudflare(ResponseFactory._get_page_content(page)) == "non-interactive":
log.info("Waiting for Cloudflare wait page to disappear.")
page.wait_for_timeout(1000)
page.wait_for_load_state()
log.info("Cloudflare captcha is solved")
return None

else:
box_selector = "#cf_turnstile div, #cf-turnstile div, .turnstile>div>div"
box_locator = page.locator("#cf_turnstile div, #cf-turnstile div, .turnstile>div>div").last
if challenge_type != "embedded":
box_selector = ".main-content p+div>div>div"
while "Verifying you are human." in ResponseFactory._get_page_content(page):
# Waiting for the verify spinner to disappear, checking every 1s if it disappeared
page.wait_for_timeout(500)
box_locator = page.locator('input[name="cf-turnstile-response"]').locator("xpath=..").last

outer_box: Any = {}
iframe = page.frame(url=__CF_PATTERN__)
Expand All @@ -146,11 +143,14 @@ def _cloudflare_solver(self, page: Page) -> None: # pragma: no cover
outer_box = iframe.frame_element().bounding_box()

if not iframe or not outer_box:
if "<title>Just a moment...</title>" not in (ResponseFactory._get_page_content(page)):
if (
challenge_type == "embedded"
or self._detect_cloudflare(ResponseFactory._get_page_content(page)) is None
):
log.info("Cloudflare captcha is solved")
return None

outer_box = page.locator(box_selector).last.bounding_box()
outer_box = box_locator.bounding_box()

# Calculate the Captcha coordinates for any viewport
captcha_x, captcha_y = outer_box["x"] + randint(26, 28), outer_box["y"] + randint(25, 27)
Expand All @@ -161,20 +161,20 @@ def _cloudflare_solver(self, page: Page) -> None: # pragma: no cover

if challenge_type != "embedded":
attempts = 0
while "<title>Just a moment...</title>" in ResponseFactory._get_page_content(page):
while self._detect_cloudflare(ResponseFactory._get_page_content(page)) is not None:
# Wait for the page
if attempts >= 100:
log.info("Cloudflare page didn't disappear after 10s, continuing...")
break
page.wait_for_timeout(100)
attempts += 1

# page.locator(box_selector).last.wait_for(state="detached")
# page.locator(".zone-name-title").wait_for(state="hidden")

self._wait_for_page_stability(page, True, False)

if "<title>Just a moment...</title>" not in (ResponseFactory._get_page_content(page)):
if (
challenge_type == "embedded"
or self._detect_cloudflare(ResponseFactory._get_page_content(page)) is None
):
log.info("Cloudflare captcha is solved")
return None
else:
Expand Down Expand Up @@ -393,20 +393,17 @@ async def _cloudflare_solver(self, page: async_Page) -> None: # pragma: no cove
else:
log.info(f'The turnstile version discovered is "{challenge_type}"')
if challenge_type == "non-interactive":
while "<title>Just a moment...</title>" in (await ResponseFactory._get_async_page_content(page)):
while self._detect_cloudflare(await ResponseFactory._get_async_page_content(page)) == "non-interactive":
log.info("Waiting for Cloudflare wait page to disappear.")
await page.wait_for_timeout(1000)
await page.wait_for_load_state()
log.info("Cloudflare captcha is solved")
return None

else:
box_selector = "#cf_turnstile div, #cf-turnstile div, .turnstile>div>div"
box_locator = page.locator("#cf_turnstile div, #cf-turnstile div, .turnstile>div>div").last
if challenge_type != "embedded":
box_selector = ".main-content p+div>div>div"
while "Verifying you are human." in (await ResponseFactory._get_async_page_content(page)):
# Waiting for the verify spinner to disappear, checking every 1s if it disappeared
await page.wait_for_timeout(500)
box_locator = page.locator('input[name="cf-turnstile-response"]').locator("xpath=..").last

outer_box: Any = {}
iframe = page.frame(url=__CF_PATTERN__)
Expand All @@ -421,11 +418,14 @@ async def _cloudflare_solver(self, page: async_Page) -> None: # pragma: no cove
outer_box = await (await iframe.frame_element()).bounding_box()

if not iframe or not outer_box:
if "<title>Just a moment...</title>" not in (await ResponseFactory._get_async_page_content(page)):
if (
challenge_type == "embedded"
or self._detect_cloudflare(await ResponseFactory._get_async_page_content(page)) is None
):
log.info("Cloudflare captcha is solved")
return None

outer_box = await page.locator(box_selector).last.bounding_box()
outer_box = await box_locator.bounding_box()

# Calculate the Captcha coordinates for any viewport
captcha_x, captcha_y = outer_box["x"] + randint(26, 28), outer_box["y"] + randint(25, 27)
Expand All @@ -436,20 +436,20 @@ async def _cloudflare_solver(self, page: async_Page) -> None: # pragma: no cove

if challenge_type != "embedded":
attempts = 0
while "<title>Just a moment...</title>" in (await ResponseFactory._get_async_page_content(page)):
while self._detect_cloudflare(await ResponseFactory._get_async_page_content(page)) is not None:
# Wait for the page
if attempts >= 100:
log.info("Cloudflare page didn't disappear after 10s, continuing...")
break
await page.wait_for_timeout(100)
attempts += 1

# await page.locator(box_selector).last.wait_for(state="detached")
# await page.locator(".zone-name-title").wait_for(state="hidden")

await self._wait_for_page_stability(page, True, False)

if "<title>Just a moment...</title>" not in (await ResponseFactory._get_async_page_content(page)):
if (
challenge_type == "embedded"
or self._detect_cloudflare(await ResponseFactory._get_async_page_content(page)) is None
):
log.info("Cloudflare captcha is solved")
return None
else:
Expand Down
46 changes: 46 additions & 0 deletions tests/fetchers/async/test_cloudflare_solver.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock

import pytest

from scrapling.engines._browsers._stealth import AsyncStealthySession
from scrapling.engines.toolbelt.convertor import ResponseFactory


@pytest.mark.asyncio
async def test_solver_clicks_localized_cloudflare_challenge(monkeypatch):
solved = False

async def content(_page):
if solved:
return "<html><title>La Redoute</title></html>"
return (
"<html><title>Un instant…</title>"
'<input name="cf-turnstile-response">'
"<script>cType: 'interactive'</script></html>"
)

async def click(*args, **kwargs):
nonlocal solved
solved = True

monkeypatch.setattr(ResponseFactory, "_get_async_page_content", AsyncMock(side_effect=content))
monkeypatch.setattr(AsyncStealthySession, "_wait_for_networkidle", AsyncMock())
monkeypatch.setattr(AsyncStealthySession, "_wait_for_page_stability", AsyncMock())

box = MagicMock()
box.bounding_box = AsyncMock(return_value={"x": 100, "y": 200})
response_input = MagicMock()
response_input.locator.return_value = MagicMock(last=box)
page = MagicMock()
page.frame.return_value = None
page.locator.return_value = response_input
page.mouse = SimpleNamespace(click=AsyncMock(side_effect=click))

await object.__new__(AsyncStealthySession)._cloudflare_solver(page)

page.locator.assert_any_call('input[name="cf-turnstile-response"]')
response_input.locator.assert_called_once_with("xpath=..")
x, y = page.mouse.click.await_args.args
assert 126 <= x <= 128
assert 225 <= y <= 227