diff --git a/src/antibot/__init__.py b/src/antibot/__init__.py index c08363c..79690a8 100644 --- a/src/antibot/__init__.py +++ b/src/antibot/__init__.py @@ -1 +1,2 @@ from .known_providers import recaptcha +from .known_providers import cloudflare diff --git a/src/antibot/known_providers/cloudflare.py b/src/antibot/known_providers/cloudflare.py index cf0e9e3..60d6a44 100644 --- a/src/antibot/known_providers/cloudflare.py +++ b/src/antibot/known_providers/cloudflare.py @@ -1,20 +1,134 @@ +""" + + + + +//input[@name="cf-turnstile-response"] +//footer/footer-inner/footer-wrapper/div[@class="ray-id"] + +iFrame with checkbox + +
+
+
+ Значок zkillboard.com +

zkillboard.com

+
+

Выполнение проверки безопасности

+

Этот веб-сайт использует сервис безопасности для защиты от вредоносных ботов. Эта страница отображается, пока веб-сайт проверяет, что вы не бот.

+
+
+
+
+ +
+
+
+ + + +
+
+""" +import logging +import time + from playwright.async_api import Page -from ..base import Detector, ChallengeInfo +from ..base import Detector, ChallengeInfo, AntiBotProvider, Solver from ..base import Stage -from ..registry import register_detector +from ..registry import register + +log = logging.getLogger(__name__) +log.setLevel(logging.DEBUG) +log.addHandler(logging.StreamHandler()) -@register_detector class CloudflareDetector(Detector): - provider = 'cloudflare_turnstile' - stage = Stage.INTERSTITIAL - async def detect(self, page: Page) -> ChallengeInfo: - result = ChallengeInfo( - stage=self.stage, - provider=self.provider, - ) - # TODO: добавить функцию определения что страница с проверкой показана - raise NotImplementedError - return result + async def detect(self, page: Page, info: ChallengeInfo): + log.debug("CloudflareDetector run") + info.meta = {} + checkbox_selector = "иногда надо тыкнуть, сейчас этого нет" + marker_selector = info.detect_locator or "//input[@name='cf-turnstile-response']" + info.detect_locator = marker_selector + + log.debug(f"ищем {marker_selector} успех если есть {info.success_locator}") + if await page.locator(info.detect_locator).count() == 0: + log.debug("на странице нет маркера преграды") + if info.success_locator: + info.success = bool(await page.query_selector(info.success_locator)) + info.meta["reason"] = [ + "Нет маркера блокировки и нет маркера успеха :-(", + "Нет маркера блокировки зато есть маркер успеха :-)" + ][info.success] + else: + info.success = True + info.meta["reason"] = "нет маркера блокировки" + log.debug("Решение не требуется") + return + + info.stage = Stage.INTERSTITIAL + return + if await page.query_selector(checkbox_selector): + info.stage = Stage.CHECKBOX + else: + info.stage = Stage.INTERSTITIAL + + +class CloudflareSolver(Solver): + + async def solve(self, page: Page, info: ChallengeInfo): + if info.stage == Stage.CHECKBOX: + raise NotImplementedError + elif info.stage == Stage.INTERSTITIAL: + if info.success_locator: + try: + await page.wait_for_selector(info.success_locator) + except Exception: + return + else: + return + else: + deadline = time.time() + 30 + while time.time() < deadline: + if not await page.query_selector(info.detect_locator): + return + return + else: + raise NotImplementedError + + +@register +class ReCaptchaProvider(AntiBotProvider): + provider = 'cloudflare' + _solver_cls = CloudflareSolver + _detector_cls = CloudflareDetector