Handle captchas in Selenium flows

python -m pip install selenium requests

export BYPASS_API_KEY="your-api-key"

$env:BYPASS_API_KEY = "your-api-key"

contact_selenium.py
bypass_client.py
page_helpers.py

import os
import time

import requests

API_URL = "https://bypass.so"
CLIENT_KEY = os.environ.get("BYPASS_API_KEY", "")


class BypassError(Exception):
    def __init__(self, code, description="", task_id=None):
        super().__init__(f"{code}: {description}")
        self.code = code
        self.description = description
        self.task_id = task_id


def _read(response):
    try:
        data = response.json()
    except ValueError:
        raise BypassError("HTTP_ERROR", f"HTTP {response.status_code} without a JSON body")
    if data.get("errorId"):
        raise BypassError(
            data.get("errorCode", "ERROR_UNKNOWN"),
            data.get("errorDescription", ""),
            data.get("taskId"),
        )
    return data


def solve(task_type, params, timeout=120, poll_every=2.0):
    if not CLIENT_KEY:
        raise BypassError("ERROR_KEY_INVALID", "set the BYPASS_API_KEY environment variable")
    body = {"clientKey": CLIENT_KEY, "type": task_type, **params}
    data = _read(requests.post(f"{API_URL}/createTask", json=body, timeout=30))
    task_id = data["taskId"]
    deadline = time.monotonic() + timeout
    while data.get("status") != "ready":
        if time.monotonic() > deadline:
            raise TimeoutError(f"task {task_id} was not ready after {timeout}s")
        time.sleep(poll_every)
        data = _read(requests.get(
            f"{API_URL}/getTaskResult/{task_id}",
            params={"clientKey": CLIENT_KEY},
            timeout=30,
        ))
    solution = dict(data.get("solution") or {})
    solution["taskId"] = task_id
    return solution


def token_of(solution):
    return solution.get("gRecaptchaResponse") or solution.get("token") or solution.get("text")


def balance():
    headers = {"Authorization": f"Bearer {CLIENT_KEY}"}
    return _read(requests.get(f"{API_URL}/balance", headers=headers, timeout=30))


def report_bad(task_id):
    headers = {"Authorization": f"Bearer {CLIENT_KEY}"}
    return _read(requests.post(
        f"{API_URL}/reportBad",
        params={"task_id": task_id},
        headers=headers,
        timeout=30,
    ))

FIELDS = {
    "RecaptchaV2Task": "g-recaptcha-response",
    "RecaptchaV3Task": "g-recaptcha-response",
    "HCaptchaTask": "h-captcha-response",
    "TurnstileTask": "cf-turnstile-response",
}

PROBE_JS = r"""
function () {
  const kinds = [
    ["TurnstileTask", ".cf-turnstile", "iframe[src*='challenges.cloudflare.com']"],
    ["HCaptchaTask", ".h-captcha", "iframe[src*='hcaptcha.com']"],
    ["RecaptchaV2Task", ".g-recaptcha", "iframe[src*='/recaptcha/']"],
  ];
  for (const [type, box, frame] of kinds) {
    const host = document.querySelector(box + "[data-sitekey]");
    if (host) return { type, sitekey: host.dataset.sitekey, callback: host.dataset.callback || null };
    const iframe = document.querySelector(frame);
    if (iframe) {
      const url = new URL(iframe.src, location.href);
      const hash = new URLSearchParams(url.hash.slice(1));
      const sitekey = url.searchParams.get("k") || url.searchParams.get("sitekey")
        || hash.get("sitekey") || (url.pathname.match(/0x4[\w-]+/) || [])[0];
      if (sitekey) return { type, sitekey, callback: null };
    }
  }
  const any = document.querySelector("[data-sitekey]");
  if (!any) return null;
  const key = any.dataset.sitekey;
  const type = key.startsWith("0x4") ? "TurnstileTask"
    : /^[0-9a-f-]{36}$/i.test(key) ? "HCaptchaTask" : "RecaptchaV2Task";
  return { type, sitekey: key, callback: any.dataset.callback || null };
}
"""

INJECT_JS = r"""
function ({ field, token, callback }) {
  let targets = [...document.querySelectorAll('[name="' + field + '"]')];
  if (field === "h-captcha-response") {
    targets.push(...document.querySelectorAll('[name="g-recaptcha-response"]'));
  }
  if (!targets.length) {
    const input = document.createElement("input");
    input.type = "hidden";
    input.name = field;
    (document.querySelector("form") || document.body).appendChild(input);
    targets = [input];
  }
  for (const el of targets) {
    el.value = token;
    el.dispatchEvent(new Event("input", { bubbles: true }));
    el.dispatchEvent(new Event("change", { bubbles: true }));
  }
  const fn = callback && callback.split(".").reduce((o, k) => (o ? o[k] : undefined), window);
  if (typeof fn === "function") fn(token);
  return targets.length;
}
"""

widget = WebDriverWait(driver, 20).until(
    lambda d: d.execute_script("return (" + PROBE_JS + ")();")
)
print(widget)

solution = solve(widget["type"], {
    "websiteURL": driver.current_url,
    "websiteKey": widget["sitekey"],
})
token = token_of(solution)

field = solution.get("tokenField") or FIELDS[widget["type"]]
driver.execute_script(
    "return (" + INJECT_JS + ")(arguments[0]);",
    {"field": field, "token": token, "callback": widget["callback"]},
)
driver.find_element(By.CSS_SELECTOR, 'form [type="submit"]').click()

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait

from bypass_client import solve, token_of
from page_helpers import FIELDS, INJECT_JS, PROBE_JS

PAGE_URL = "https://example.com/contact"
PROBE = "return (" + PROBE_JS + ")();"
INJECT = "return (" + INJECT_JS + ")(arguments[0]);"


def main():
    driver = webdriver.Chrome()
    try:
        driver.get(PAGE_URL)
        wait = WebDriverWait(driver, 20)
        widget = wait.until(lambda d: d.execute_script(PROBE))
        print("found", widget["type"], widget["sitekey"])

        driver.find_element(By.NAME, "name").send_keys("Ada")
        driver.find_element(By.NAME, "email").send_keys("[email protected]")
        driver.find_element(By.NAME, "message").send_keys("Hello from my test run")

        solution = solve(widget["type"], {
            "websiteURL": driver.current_url,
            "websiteKey": widget["sitekey"],
        })
        field = solution.get("tokenField") or FIELDS[widget["type"]]
        driver.execute_script(INJECT, {
            "field": field,
            "token": token_of(solution),
            "callback": widget["callback"],
        })

        driver.find_element(By.CSS_SELECTOR, 'form [type="submit"]').click()
        wait.until(EC.url_changes(PAGE_URL))
        print("submitted, now at", driver.current_url)
    finally:
        driver.quit()


if __name__ == "__main__":
    main()

python contact_selenium.py

found TurnstileTask 0x4AAAAAAA-SITE-KEY
submitted, now at https://example.com/contact/thanks

from selenium.webdriver.common.by import By

from bypass_client import solve

image = driver.find_element(By.CSS_SELECTOR, "img.captcha")
solution = solve("ImageToTextTask", {"imageBase64": image.screenshot_as_base64})

box = driver.find_element(By.NAME, "captcha")
box.clear()
box.send_keys(solution["text"])
driver.find_element(By.CSS_SELECTOR, 'form [type="submit"]').click()

site_key = driver.execute_script("""
  const s = document.querySelector("script[src*='recaptcha'][src*='render=']");
  return s ? new URL(s.src).searchParams.get("render") : null;
""")
solution = solve("RecaptchaV3Task", {
    "websiteURL": driver.current_url,
    "websiteKey": site_key,
    "action": "submit",
})