AWS WAF CAPTCHA solver API

{
  "clientKey": "sur_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
  "task": {
    "type": "AwsCaptchaCookie",
    "websiteURL": "https://example.com/protected",
    "userAgent": "YOUR_SUPPORTED_USER_AGENT",
    "proxy": "http://user:password@proxy.example.com:8080",
    "cookies": "session=YOUR_SESSION"
  }
}

{
  "errorId": 0,
  "status": "ready",
  "solution": {
    "aws-waf-token": "<aws-waf-token-value>"
  }
}

# aws-captcha-python.py - solve AWS WAF CAPTCHA and request the protected page with the
# aws-waf-token cookie.
# Install: pip install requests
# Run: python aws-captcha-python.py

import time

import requests


# Your SurSolver API key.
CLIENT_KEY = "sur_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
# The protected page that answers with the challenge.
TARGET_URL = "https://example.com/page"
# The proxy for the solve and the replay: http://user:pass@host:port or socks5://host:port.
PROXY_URL = "http://username:password@proxy.example.com:8080"
# Existing session cookies as a Cookie header value (name=value; name2=value2).
SESSION_COOKIES = None
# The User-Agent the solve runs with; send the result with this same user agent.
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/155.0.0.0 Safari/537.36"
# Text that appears only in a response the site accepted.
SUCCESS_TEXT = "PASTE_ACCEPTED_MARKER"


API = "https://api.sursolver.com"
POLL_ATTEMPTS = 60


def call(path, body):
    # API errors are HTTP 200 with errorId 1; anything else is a transport failure.
    response = requests.post(f"{API}{path}", json=body, timeout=30)
    response.raise_for_status()
    data = response.json()
    if data["errorId"] != 0:
        raise RuntimeError(f'{data["errorCode"]}: {data["errorDescription"]}')
    return data


def solve(task):
    created = call("/createTask", {"clientKey": CLIENT_KEY, "task": task})
    for _ in range(POLL_ATTEMPTS):
        result = call("/getTaskResult", {"clientKey": CLIENT_KEY, "taskId": created["taskId"]})
        if result["status"] == "ready":
            return result["solution"]
        time.sleep(3)
    raise TimeoutError(f'task {created["taskId"]} was not ready in {POLL_ATTEMPTS} polls')


def without_empty(fields):
    # Optional task fields are sent only when they have a value.
    return {name: value for name, value in fields.items() if value not in (None, "")}


PROXIES = {"http": PROXY_URL, "https": PROXY_URL}


def main():
    solution = solve({
        "type": "AwsCaptchaCookie",
        "websiteURL": TARGET_URL,
        "userAgent": USER_AGENT,
        **without_empty({
            "cookies": SESSION_COOKIES,
        }),
        "proxy": PROXY_URL,
    })

    # Replay with the user agent the solve ran with, through the same proxy.
    headers = {"User-Agent": USER_AGENT}
    cookies = dict(part.strip().split("=", 1) for part in (SESSION_COOKIES or "").split(";")
                   if "=" in part)
    cookies["aws-waf-token"] = solution["aws-waf-token"]
    cookies = {
        **cookies,
        **solution.get("cookies", {}),
        "aws-waf-token": solution["aws-waf-token"],
    }
    response = requests.get(
        TARGET_URL,
        headers=headers,
        cookies=cookies,
        proxies=PROXIES,
        timeout=30,
    )
    if not response.ok or SUCCESS_TEXT not in response.text:
        raise SystemExit(f"The site did not accept the result (HTTP {response.status_code}).")
    print("The site accepted the result.")


if __name__ == "__main__":
    main()