Python curl_cffi 實作：Browser Impersonation、Session、Asyncio 與 WebSocket

python3 -m venv .venv
source .venv/bin/activate
python -m pip install 'curl_cffi==0.16.3'

from curl_cffi import get, CurlHttpVersion

response = get(
    "https://tls.browserleaks.com/json",
    impersonate="chrome124",
    timeout=15,
)
response.raise_for_status()
data = response.json()

print("HTTP version:", CurlHttpVersion(response.http_version).name)
print("JA3N:", data.get("ja3n_hash"))

python quick_start.py

Python 程式
    ↓
Session／AsyncSession／Low-level Curl API
    ↓
CFFI
    ↓
curl-impersonate fork／libcurl
    ↓
TLS、HTTP/1.1、HTTP/2、HTTP/3 與 socket

python -m pip install 'curl_cffi[cli]==0.16.3' 'requests==2.34.2'
python -m pip check

import platform
import curl_cffi
from curl_cffi import Curl

print(platform.python_version())
print(curl_cffi.__version__)
curl = Curl()
try:
    print(curl.version().decode())
finally:
    curl.close()

curl-cffi list

import json
import requests
from curl_cffi import Session, CurlHttpVersion, CurlOpt

URL = "https://tls.browserleaks.com/json"
UA = (
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
    "AppleWebKit/537.36 (KHTML, like Gecko) "
    "Chrome/124.0.0.0 Safari/537.36"
)
FIELDS = ("ja3_hash", "ja3n_hash", "ja4", "akamai_hash")


def show(name, response, version):
    response.raise_for_status()
    data = response.json()
    print(json.dumps({
        "client": name,
        "http_version": version,
        "fingerprints": {key: data.get(key) for key in FIELDS},
    }, ensure_ascii=False))


for name, headers in (("requests", {}), ("requests_ua", {"User-Agent": UA})):
    with requests.Session() as session:
        session.trust_env = False
        response = session.get(URL, headers=headers, timeout=(5, 15))
        version = {10: "1.0", 11: "1.1", 20: "2"}.get(response.raw.version)
        show(name, response, version)

for name, profile in (("curl_cffi", None), ("chrome124", "chrome124")):
    with Session(curl_options={CurlOpt.PROXY: ""}) as session:
        response = session.get(URL, impersonate=profile, timeout=15)
        show(name, response, CurlHttpVersion(response.http_version).name)

from curl_cffi import Session

with Session(impersonate="chrome124", timeout=15) as session:
    response = session.get(
        "https://httpbin.org/headers",
        headers={"Accept": "application/json"},
    )
    response.raise_for_status()
    assert response.json()["headers"]["Accept"] == "application/json"

import os
from curl_cffi import Session

BASE = os.environ.get("HTTP_ECHO_BASE", "https://httpbin.org")

with Session(impersonate="chrome124", timeout=15) as session:
    query = session.get(f"{BASE}/get", params={"page": 1, "tag": ["tls", "http2"]})
    query.raise_for_status()
    assert query.json()["args"]["tag"] == ["tls", "http2"]

    form = session.post(f"{BASE}/post", data={"name": "sample"})
    form.raise_for_status()
    assert form.json()["form"]["name"] == "sample"

    payload = session.post(f"{BASE}/post", json={"enabled": True, "count": 2})
    payload.raise_for_status()
    assert payload.json()["json"]["count"] == 2

    raw = session.post(
        f"{BASE}/post",
        content=b"sample body",
        headers={"Content-Type": "application/octet-stream"},
    )
    raw.raise_for_status()
    assert raw.json()["data"] == "sample body"

    for method in ("PUT", "PATCH", "DELETE"):
        response = session.request(method, f"{BASE}/anything", json={"sample": True})
        response.raise_for_status()
        assert response.json()["method"] == method

from curl_cffi import Session

with Session(timeout=15, allow_redirects=False) as session:
    response = session.get("https://httpbin.org/redirect/1")
    assert response.status_code == 302

    response = session.get(
        "https://httpbin.org/redirect/2",
        allow_redirects=True,
        max_redirects=3,
    )
    response.raise_for_status()
    assert [item.status_code for item in response.history] == [302, 302]

from curl_cffi import Session

with Session(impersonate="chrome124", timeout=15) as session:
    response = session.get("https://httpbin.org/cookies/set/article/demo")
    response.raise_for_status()
    response = session.get("https://httpbin.org/cookies")
    response.raise_for_status()
    assert response.json()["cookies"]["article"] == "demo"

    session.cookies.set("theme", "dark", domain="httpbin.org", path="/")
    assert session.cookies.get("theme", domain="httpbin.org", path="/") == "dark"
    session.cookies.delete("theme", domain="httpbin.org", path="/")

from pathlib import Path
from tempfile import TemporaryDirectory
from curl_cffi import CurlMime, Session

with TemporaryDirectory() as folder:
    path = Path(folder) / "sample.txt"
    path.write_text("sample file", encoding="utf-8")
    multipart = CurlMime()
    try:
        multipart.addpart(
            name="attachment", filename="sample.txt",
            content_type="text/plain", local_path=str(path),
        )
        multipart.addpart(
            name="metadata", filename="metadata.json",
            content_type="application/json", data=b'{"sample":true}',
        )
        with Session(timeout=15) as session:
            response = session.post(
                "https://httpbin.org/post",
                data={"label": "article"}, multipart=multipart,
            )
            response.raise_for_status()
            assert response.json()["files"]["attachment"] == "sample file"
    finally:
        multipart.close()

import os
from curl_cffi import Session

with Session(proxy=os.environ["HTTP_PROXY_URL"], timeout=15) as session:
    response = session.get("https://httpbin.org/get")
    response.raise_for_status()
    print(response.status_code)

import os
from curl_cffi import Session

with Session(
    verify=os.environ["TEST_CA_FILE"],
    cert=(os.environ["TEST_CLIENT_CERT"], os.environ["TEST_CLIENT_KEY"]),
    timeout=15,
) as session:
    response = session.get(os.environ["TEST_MTLS_URL"])
    response.raise_for_status()
    print(response.status_code)

from curl_cffi import Session, CurlHttpVersion, CurlOpt

with Session(curl_options={CurlOpt.PROXY: ""}, timeout=12) as session:
    response = session.get(
        "https://fp.impersonate.pro/api/http3",
        impersonate="chrome150",
        http_version=CurlHttpVersion.V3ONLY,
    )
    response.raise_for_status()
    assert response.http_version == CurlHttpVersion.V3
    data = response.json()
    print({"status": response.status_code, "version": "HTTP/3", "fields": list(data)})

import asyncio
from curl_cffi import AsyncSession
from curl_cffi.requests.exceptions import RequestException


async def fetch_many(urls, concurrency=3):
    if concurrency < 1:
        raise ValueError("concurrency must be positive")
    results = []
    async with AsyncSession(
        impersonate="chrome124", max_clients=concurrency, timeout=10
    ) as session:
        async def fetch(index, url):
            try:
                response = await session.get(url)
                response.raise_for_status()
                return {"index": index, "status": response.status_code, "error": None}
            except RequestException as error:
                return {"index": index, "status": None, "error": type(error).__name__}

        for start in range(0, len(urls), concurrency):
            batch = urls[start:start + concurrency]
            results.extend(await asyncio.gather(*(
                fetch(start + offset, url) for offset, url in enumerate(batch)
            )))
    return results

from curl_cffi import Session, CurlOpt

with Session(curl_options={CurlOpt.TIMEOUT_MS: 15_000}) as session:
    with session.stream("GET", "https://httpbin.org/bytes/4096", timeout=10) as response:
        response.raise_for_status()
        received = 0
        for chunk in response.iter_content():
            received += len(chunk)
        assert received == 4096

from pathlib import Path
from tempfile import NamedTemporaryFile
from curl_cffi import Session


class DownloadTooLarge(Exception):
    pass


def download(url: str, output: Path, max_bytes: int) -> int:
    if max_bytes <= 0:
        raise ValueError("max_bytes must be positive")
    partial = None
    received = 0
    try:
        with NamedTemporaryFile(
            mode="wb", dir=output.parent, prefix=output.name + ".", delete=False
        ) as destination:
            partial = Path(destination.name)

            def on_chunk(chunk: bytes) -> int:
                nonlocal received
                if received + len(chunk) > max_bytes:
                    raise DownloadTooLarge("response exceeds size limit")
                written = destination.write(chunk)
                if written != len(chunk):
                    raise OSError("incomplete file write")
                received += written
                return written

            with Session(timeout=30, allow_redirects=False) as session:
                response = session.get(url, content_callback=on_chunk)
                response.raise_for_status()
                if not 200 <= response.status_code < 300:
                    raise ValueError("unexpected HTTP status")

        partial.replace(output)
        return received
    finally:
        if partial is not None:
            partial.unlink(missing_ok=True)

import asyncio
import os
from curl_cffi import AsyncSession


async def upload():
    async def chunks():
        yield b"first\n"
        await asyncio.sleep(0)
        yield b"second\n"

    base = os.environ.get("HTTP_ECHO_BASE", "https://httpbin.org")
    async with AsyncSession(timeout=15) as session:
        response = await session.post(
            f"{base}/post",
            content=chunks(),
            headers={"Content-Type": "application/octet-stream"},
        )
        response.raise_for_status()
        assert response.json()["data"] == "first\nsecond\n"


asyncio.run(upload())

import asyncio
import os
from curl_cffi import AsyncSession, CurlWsFlag


async def exchange(url):
    async with AsyncSession(impersonate="chrome150") as session:
        async with session.ws_connect(
            url,
            timeout=10,
            recv_queue_size=32,
            send_queue_size=32,
            max_message_size=1024 * 1024,
        ) as ws:
            await ws.send("hello", flags=CurlWsFlag.TEXT, timeout=5)
            assert await ws.recv_str(timeout=5) == "hello"
            await ws.send_bytes(b"sample")
            payload, flags = await ws.recv(timeout=5)
            assert payload == b"sample"
            assert flags & CurlWsFlag.BINARY
            await ws.send_json({"sample": True})
            assert await ws.recv_json(timeout=5) == {"sample": True}


asyncio.run(asyncio.wait_for(exchange(os.environ["TEST_WS_URL"]), timeout=20))

import json
import os
from dataclasses import asdict
from pathlib import Path
from curl_cffi import Fingerprint

source = Path(os.environ["FINGERPRINT_SOURCE"])
target = Path(os.environ["FINGERPRINT_OUTPUT"])
fingerprint = Fingerprint(**json.loads(source.read_text(encoding="utf-8")))
fingerprint.headers["Accept-Language"] = "zh-TW,zh;q=0.9"
with target.open("x", encoding="utf-8") as destination:
    json.dump(asdict(fingerprint), destination, ensure_ascii=False, indent=2)

from io import BytesIO
from curl_cffi import Curl, CurlInfo, CurlOpt, CurlHttpVersion

body = BytesIO()
headers = BytesIO()
curl = Curl()
try:
    curl.setopt(CurlOpt.URL, "https://httpbin.org/get")
    curl.setopt(CurlOpt.HTTPHEADER, [b"Accept: application/json"])
    curl.setopt(CurlOpt.WRITEDATA, body)
    curl.setopt(CurlOpt.HEADERFUNCTION, headers.write)
    curl.setopt(CurlOpt.TIMEOUT_MS, 15_000)
    curl.perform()
    print({
        "status": curl.getinfo(CurlInfo.RESPONSE_CODE),
        "http_version": CurlHttpVersion(curl.getinfo(CurlInfo.HTTP_VERSION)).name,
        "dns_seconds": curl.getinfo(CurlInfo.NAMELOOKUP_TIME),
        "connect_seconds": curl.getinfo(CurlInfo.CONNECT_TIME),
        "tls_seconds": curl.getinfo(CurlInfo.APPCONNECT_TIME),
        "total_seconds": curl.getinfo(CurlInfo.TOTAL_TIME),
    })
finally:
    curl.close()

from io import BytesIO
from curl_cffi import Curl, CurlInfo, CurlOpt

payload = b"upload via READFUNCTION"
source = BytesIO(payload)
body = BytesIO()
curl = Curl()
try:
    curl.setopt(CurlOpt.URL, "https://httpbin.org/put")
    curl.setopt(CurlOpt.UPLOAD, 1)
    curl.setopt(CurlOpt.INFILESIZE_LARGE, len(payload))
    curl.setopt(CurlOpt.READFUNCTION, source.read)
    curl.setopt(CurlOpt.WRITEDATA, body)
    curl.setopt(CurlOpt.TIMEOUT_MS, 15_000)
    curl.perform()
    assert curl.getinfo(CurlInfo.RESPONSE_CODE) == 200
finally:
    curl.close()

curl-cffi list --json
curl-cffi get https://tls.browserleaks.com/json --impersonate chrome124

git clone https://github.com/hsunAlfred/http-client-fingerprint-examples.git
cd http-client-fingerprint-examples/python-curl-cffi
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
python diagnostic.py \
  --url https://tls.browserleaks.com/json \
  --profile chrome124 \
  --concurrency 2 \
  --output fingerprint-results.jsonl

python -m pip check
python -m unittest -v test_diagnostic.py