# 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()