A Cloudflare Turnstile solver API that charges only for solved tokens.

# Solve one Cloudflare Turnstile challenge with ZeroCaptcha and print its token.
#
# Needs Python 3.10 or later and requests (pip install requests).
# Put your page's details in main(), then run it with your API key in the
# environment:
#   ZEROCAPTCHA_KEY=zc_live_... python quickstart.py
# ZEROCAPTCHA_API, if set, points it at another API host.

import calendar
import json
import os
import sys
import threading
import time
import uuid
from email.utils import parsedate_to_datetime

import requests

API_URL = os.environ.get("ZEROCAPTCHA_API", "https://api.zerocaptcha.io")
API_KEY = os.environ.get("ZEROCAPTCHA_KEY", "")

POLL_SECONDS = 2  # between getTaskResult calls
REQUEST_SECONDS = 15  # the longest one HTTP request may take
MAX_REPLY = 1 << 20  # the most of a reply it reads, in bytes
UTC = "%Y-%m-%dT%H:%M:%SZ"  # how it writes times: UTC, to the second
# The longest the whole run may take:
DEADLINE_SECONDS = int(os.environ.get("ZEROCAPTCHA_DEADLINE_SECONDS", "180"))
# This task's Idempotency-Key: the UTC time it was made, then random. Run again
# with the same key and createTask returns the same task, so a lost reply never
# costs a second task. The API keeps a key for 24 hours from its first
# createTask, so it surely knows it until 24 hours after the time it starts with.
KEY_HOURS = 24
RESUMED_KEY = os.environ.get("ZEROCAPTCHA_INTENT_KEY", "")
INTENT_KEY = RESUMED_KEY or f"{time.strftime(UTC, time.gmtime())}-{uuid.uuid4()}"
try:
    KEY_EXPIRES = calendar.timegm(time.strptime(INTENT_KEY[:20], UTC)) + KEY_HOURS * 3600
except ValueError:
    KEY_EXPIRES = 0  # not a key this sample made
# The task's ID, once createTask has given it: from then on, it resumes the task.
task_id = os.environ.get("ZEROCAPTCHA_TASK_ID", "")
# The API's "try again later", like HTTP 429 and 5xx: call() sends the same
# request again, as long as the deadline allows.
RETRYABLE = ("ERROR_RATE_LIMIT", "ERROR_SERVICE_UNAVAILABLE",
             "ERROR_NO_SLOT_AVAILABLE", "ERROR_IDEMPOTENCY_KEY_IN_USE")


def main():
    global task_id
    if not API_KEY:
        fail("Set ZEROCAPTCHA_KEY to your API key, zc_live_..., from the dashboard.")
    deadline = time.monotonic() + DEADLINE_SECONDS
    if not task_id:
        # createTask goes with the key only while one request still fits before
        # the API may forget it, and could make a second task.
        key_left = KEY_EXPIRES - time.time() - REQUEST_SECONDS
        if key_left <= 0:
            fail("createTask: the intent key is too old, or not from this sample: "
                 f"the API keeps a key for {KEY_HOURS} hours, then createTask could "
                 "start another task. Look for its task with GET "
                 f"/v1/tasks?idempotencyKey={INTENT_KEY}, or run without "
                 "ZEROCAPTCHA_INTENT_KEY to start a new one.")
        print(f"Creating the task with intent key {INTENT_KEY}, "
              f"valid until {time.strftime(UTC, time.gmtime(KEY_EXPIRES))}.",
              file=sys.stderr)
        task = call("createTask", {
            "task": {
                # Or "TurnstileTask", to solve through your own proxy, with "proxy" below.
                "type": "TurnstileTaskProxyless",
                "websiteURL": "https://example.com/login",  # the page with the widget
                "websiteKey": "0x4AAAAAAA...",  # the widget's data-sitekey
                # The widget's action and cData, which many sites check when they verify
                # the token: copy them from its data-action and data-cdata attributes, or
                # the action and cData options of turnstile.render(). Leave out any the
                # widget does not set.
                "metadata": {"action": "login", "cdata": "session-7f3a9c2e"},
                # "proxy": "http://user:pass@proxy.example.net:8080",  # TurnstileTask only
            },
            # Optional: where to POST the result when the task ends, instead of polling.
            # "callbackUrl": "https://hooks.example.com/zerocaptcha",
        }, min(deadline, time.monotonic() + key_left), {"Idempotency-Key": INTENT_KEY})
        if not isinstance(task.get("taskId"), str) or not task["taskId"]:
            unsure(f"createTask: unexpected reply: {json.dumps(task)[:200]}")
        task_id = task["taskId"]
    print(f"Waiting for task {task_id}.", file=sys.stderr)

    for _ in range(DEADLINE_SECONDS // POLL_SECONDS):
        if time.monotonic() + POLL_SECONDS > deadline:
            break
        time.sleep(POLL_SECONDS)
        if time.monotonic() >= deadline:  # the wait itself ran late
            break
        result = call("getTaskResult", {"taskId": task_id}, deadline)
        if result.get("status") == "processing":
            continue
        solution = result.get("solution")
        token = solution.get("token") if isinstance(solution, dict) else None
        if result.get("status") == "ready" and isinstance(token, str) and token:
            print(token)
            return
        unsure(f"getTaskResult: unexpected reply: {json.dumps(result)[:200]}")
    unsure(f"No token within {DEADLINE_SECONDS} seconds: "
           f"task {task_id} is still processing.")


# POSTs one call, with any extra headers, and returns its reply. It stops on
# an HTTP error, a reply that isn't JSON, and errorId 1, whose errorCode and
# errorDescription say what went wrong. A reply that only says to try again
# later is sent again after its Retry-After, or a pause that doubles each
# time, until the deadline; no request starts once the deadline has passed.
def call(method, body, deadline, headers=None):
    pause, tries = 1, 0
    failure = "no reply in time"  # what went wrong last, should time run out
    while True:
        tries += 1
        left = deadline - time.monotonic()
        if left <= 0:
            unsure(f"{method}: {failure}")
        try:
            status, asked, data = post(method, body, headers, min(REQUEST_SECONDS, left))
        except TimeoutError:
            unsure(f"{method}: no reply in time")
        except requests.RequestException as error:
            unsure(f"{method}: {error}")
        if len(data) > MAX_REPLY:
            unsure(f"{method}: the reply is too long")
        text = data.decode("utf-8", "replace")
        failure = f"HTTP {status}: {text[:200]}"
        if status == 200:
            try:
                reply = json.loads(text)
            except ValueError:
                unsure(f"{method}: the reply is not JSON: {text[:200]}")
            if not isinstance(reply, dict) or "errorId" not in reply:
                unsure(f"{method}: unexpected reply: {text[:200]}")
            if reply["errorId"] == 0:
                return reply
            code, description = reply.get("errorCode"), reply.get("errorDescription")
            failure = f"{code}: {description}"
            if code not in RETRYABLE:
                # The API's own "no": a failed task's reply says how it ended,
                # and a refused create made no task, unless an earlier
                # createTask with this key, in this run or one before, went
                # through unanswered. Any other refused poll leaves how the task
                # ended unknown.
                if "status" in reply or (method == "createTask" and tries == 1
                                         and not RESUMED_KEY):
                    fail(f"{method}: {failure}")
                unsure(f"{method}: {failure}")
        elif status != 429 and status < 500:
            unsure(f"{method}: {failure}")
        # Only "try again later" is left: wait as the reply asks, or pause.
        wait = retry_after(asked)
        if wait <= 0:
            wait = pause
        if time.monotonic() + wait >= deadline:
            unsure(f"{method}: {failure}")
        time.sleep(wait)
        pause = min(pause * 2, 16)


# POSTs once, and returns the reply's status, its Retry-After and up to
# MAX_REPLY + 1 bytes of its body. requests's own timeout limits only each
# wait for more bytes, so a reply that keeps trickling in could outlast any
# deadline: here a thread makes the request, and after `seconds` it is left
# behind with TimeoutError.
def post(method, body, headers, seconds):
    outcome = []

    def request():
        try:
            with requests.post(f"{API_URL}/{method}", json={"clientKey": API_KEY, **body},
                               headers=headers, timeout=seconds + 1,
                               stream=True) as response:
                data = b""
                for chunk in response.iter_content(64 * 1024):
                    data += chunk
                    if len(data) > MAX_REPLY:
                        break
                outcome.append((response.status_code,
                                response.headers.get("Retry-After", ""), data))
        except Exception as error:  # raised again below, in the caller's thread
            outcome.append(error)

    thread = threading.Thread(target=request, daemon=True)
    thread.start()
    thread.join(seconds)
    if not outcome:
        raise TimeoutError
    if isinstance(outcome[0], Exception):
        raise outcome[0]
    return outcome[0]


# The seconds a Retry-After asks to wait: its number of seconds, or the time
# until its HTTP date. 0 for anything else.
def retry_after(value):
    if value.isdecimal():
        return float(value)
    try:
        return parsedate_to_datetime(value).timestamp() - time.time()
    except (TypeError, ValueError):
        return 0


def fail(message):
    sys.exit(message)


# Stops when how the task ended is unknown: it may exist, and running again as
# the last line says picks it up instead of starting another. That is by its
# ID once createTask has given it, and before that by the intent key, while
# the API surely still keeps it.
def unsure(message):
    if task_id:
        fail(f"{message}\nRun again with ZEROCAPTCHA_TASK_ID={task_id} to keep "
             "waiting for this task instead of starting another.")
    fail(f"{message}\nRun again with ZEROCAPTCHA_INTENT_KEY={INTENT_KEY} before "
         f"{time.strftime(UTC, time.gmtime(KEY_EXPIRES))} to resume this task "
         "instead of starting another: the API keeps an intent key for "
         f"{KEY_HOURS} hours.")


if __name__ == "__main__":
    main()