Cloudflare WAF in Python

This async example uses one sticky proxy and one browser identity to encounter the challenge, solve it, and make the first request after the solve.

python -m pip install wreq

asyncio is included with Python 3.11 and later, so it does not need to be installed separately.

cloudflare_waf.pypython

import asyncio
import os
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit

from wreq import Client, Cookie, Emulation, Proxy

API_URL = "https://api.uncaptcha.io/v1/task/execute"
USER_AGENT = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
    "AppleWebKit/537.36 (KHTML, like Gecko) "
    "Chrome/143.0.0.0 Safari/537.36"
)


def add_query_parameter(url: str, name: str, value: str) -> str:
    parts = urlsplit(url)
    query = dict(parse_qsl(parts.query, keep_blank_values=True))
    query[name] = value
    return urlunsplit(parts._replace(query=urlencode(query)))


async def solve_waf(
    *,
    url: str,
    proxy: str,
    challenge_html: str,
) -> dict:
    async with Client(timeout=35) as api_client:
        response = await api_client.post(
            API_URL,
            headers={"X-Api-Key": os.environ["UNCAPTCHA_API_KEY"]},
            json={
                "task_type": "waf",
                "task_data": {
                    "url": url,
                    "proxy": proxy,
                    "user_agent": USER_AGENT,
                    "html": challenge_html,
                },
            },
        )
        payload = await response.json()

    if not response.status.is_success() or not payload.get("success"):
        raise RuntimeError(payload.get("message", "unknown uncaptcha.io error"))
    return payload["data"]["solution"]


async def main() -> None:
    target = "https://example.com/protected"
    proxy = os.environ["SOLVE_PROXY"]

    async with Client(
        emulation=Emulation.Chrome143,
        user_agent=USER_AGENT,
        proxies=[Proxy.all(proxy)],
        cookie_store=True,
        timeout=30,
    ) as session:
        challenged = await session.get(target)
        if challenged.headers.get("Cf-Mitigated") != "challenge":
            raise RuntimeError("The response was not a Cloudflare challenge")

        solution = await solve_waf(
            url=target,
            proxy=proxy,
            challenge_html=await challenged.text(),
        )

        session.cookie_jar.add(
            Cookie("cf_clearance", solution["clearance"]),
            target,
        )
        if solution.get("cf_bm"):
            session.cookie_jar.add(
                Cookie("__cf_bm", solution["cf_bm"]),
                target,
            )

        referer = target
        if solution.get("cf_rt"):
            referer = add_query_parameter(
                target, "__cf_chl_tk", solution["cf_rt"]
            )

        response = await session.post(
            target,
            headers={**solution["headers"], "Referer": referer},
            form=solution.get("attributes", {}),
        )
        response.raise_for_status()
        print(response.status, response.url)


if __name__ == "__main__":
    asyncio.run(main())

If you do not need to control the follow-up request, use wafauto and decode response.base64_body instead.