Skip to content

Repository files navigation

pycronet

pycronet 是面向 Windows x64 的 Python HTTP/WebSocket 客户端。Python 通过 ctypes 调用小型 C++ Shim,Shim 再调用 Chromium Native Cronet;Python 不直接管理 Cronet 回调指针和原生对象生命周期。

当前包版本为 0.3.0,底层目标是 Chromium 145.0.7632.162。Chromium 145 之后官方不再提供 Native Cronet 接口;升级内核时必须同步维护补丁、ABI 和构建产物。

特性

  • Requests 风格的 Session、AsyncSession 和模块级请求函数。
  • 完整响应与流式响应,支持取消、超时、重定向和 Cookie。
  • HTTP、HTTPS、HTTP/HTTPS 代理、SOCKS5、SOCKS5H。
  • 每个 Session 独立的 TLS profile、证书校验策略和 CookieStore。
  • Cronet WebSocket:WSS、子协议、Origin、自定义请求头、文本和二进制消息。
  • 文件上传和下载;不需要 CFFI、Rust、Cargo、PyO3、nanobind 或额外网络库。

安装

运行时需要 Windows x64、Python 3.9 至 3.14 和 Microsoft Visual C++ 运行库。安装 wheel 后,包内应包含:

pycronet/
  _native_bin/
    pycronet_shim.dll
    cronet.145.0.7632.162.dll

加载器只接受带完整版本号的 Cronet DLL。如果报 DLL 缺失,请确认这两个文件都在包内的 _native_bin 目录,且名称未被修改。

py -3.12 -m pip install -e .

从 Chromium 产物构建

构建 Shim 需要 Visual Studio 2022 C++、Windows SDK、CMake,以及已经构建好的 Cronet DLL 和导入库。Chromium 源码可以放在项目目录旁边或其他位置,通过环境变量传入:

$env:CHROMIUM_ROOT = "../chromium"
$env:CRONET_ROOT = "../chromium/src/out/Cronet145"
cmake -S native_shim -B build/native-shim -G "Visual Studio 17 2022" -A x64 -DCHROMIUM_ROOT=$env:CHROMIUM_ROOT -DCRONET_ROOT=$env:CRONET_ROOT
cmake --build build/native-shim --config Release

CRONET_ROOT 必须包含 cronet.145.0.7632.162.dll 和 cronet.145.0.7632.162.dll.lib。构建会把 Shim 和 Cronet DLL 复制到 pycronet/_native_bin/;.lib 只在编译时使用,.pdb 只用于调试,不应放入 wheel。也可以运行 py -3.12 setup.py build_py,它读取相同的环境变量。

基本请求

重复请求应复用 Session,以复用 Engine、连接池、Cookie 和 TLS 配置:

import pycronet

with pycronet.Session(timeout=15.0, chrometls="chrome_150") as session:
    response = session.get("https://example.com/")
    response.raise_for_status()
    print(response.status_code, response.text)

timeout 单位是秒,可在单次请求覆盖。支持 get、post、put、delete、patch、head、options,参数包括 params、headers、cookies、data 和 json:

with pycronet.Session(headers={"User-Agent": "my-client/1.0"}) as session:
    response = session.post(
        "https://example.com/api",
        params={"page": 1},
        json={"name": "test"},
        headers={"X-Request-ID": "demo"},
    )

模块级 pycronet.getpycronet.post 等函数适合一次性同步请求;每次调用都会创建并关闭临时 Session:

response = pycronet.get("https://example.com/api", params={"page": 1}, timeout=10.0)
print(response.status_code, response.json())

异步的一次性请求使用 pycronet.async_getpycronet.async_post 等函数,必须在异步函数中等待:

async def fetch_once():
    response = await pycronet.async_get("https://example.com/api", timeout=10.0)
    return response.json()

print(asyncio.run(fetch_once()))

需要连续请求、连接复用、共享 Cookie 或共享 TLS 配置时,应创建显式 Session。异步代码也可以使用 AsyncSession:

import asyncio
import pycronet

async def main():
    async with pycronet.AsyncSession(chrometls="chrome_150") as session:
        response = await session.get("https://example.com/")
        print(response.status_code, response.text)

asyncio.run(main())

响应、错误和流式读取

Response 提供 status_code、headers、url、content、text、json() 和 cookies。raise_for_status() 在 4xx/5xx 时抛出 HTTPStatusError;网络、超时和 Cronet 错误抛出 RequestError。

大响应使用 stream(),通过有界事件队列逐块消费:

with pycronet.Session() as session:
    with session.stream("GET", "https://example.com/large.bin") as response:
        response.raise_for_status()
        with open("large.bin", "wb") as output:
            for chunk in response.iter_content(64 * 1024):
                output.write(chunk)

异步流式读取使用 aiter_content() 或 aiter_lines()。流关闭后会释放底层请求,建议始终使用 with/async with。

Cookie 和代理

Session 默认维护 Python CookieJar。需要让 Chromium 在自动重定向等原生流程中携带 Cookie 时,开启内存 CookieStore:

with pycronet.Session(enable_cookie_store=True) as session:
    session.get("https://example.com/login")
    response = session.get("https://example.com/account")

proxies 支持 http://、https://、socks5:// 和 socks5h://:

with pycronet.Session(proxies="http://127.0.0.1:8080") as session:
    response = session.get("https://example.com/")

with pycronet.Session(proxies="socks5://user:password@127.0.0.1:1080") as session:
    response = session.get("https://example.com/")

用户名和密码只用于代理认证,不会写入日志、错误文本或公开代理字符串;每个凭据最多 255 个 UTF-8 字节。非法 URI、未知 scheme 或超长凭据会在创建 Session 时报告错误。不同 Session 不共享 Cookie、代理或 TLS 配置。

TLS 指纹 profile

内置 profile 位于 pycronet/tls_profiles.json,常用名称有 chrome_144、chrome_150 和测试 profile:

response = pycronet.get("https://example.com/", chrometls="chrome_150", verify=False)

verify=False 仅关闭该 Engine 的证书校验,适合本地自签名测试,不建议用于生产。cipher、曲线和签名算法名称会在 Python 层转换为 Chromium experimental options 使用的十六进制 ID;0x 开头的值会规范化,未知名称会跳过,全部无效时回退 Chromium 默认值。TLS_GREASE 不转换为固定数字,而由 TLS 栈按连接生成。

可运行时注册 profile:

pycronet.add_tls_profile("local_test", {
    "cipher_suites": ["TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256", "0xC014"],
    "tls_curves": ["X25519", "P-256"],
    "tls_extensions": [],
    "signature_algorithms": ["rsa_pss_rsae_sha256"],
})
with pycronet.Session(chrometls="local_test") as session:
    response = session.get("https://example.com/")

profile 不能突破 Chromium 145/BoringSSL 的实际能力。TLS 1.3 cipher 顺序、部分扩展顺序和签名算法可用性可能与其他版本不同。

WebSocket

通过 Session 创建 WebSocket,推荐使用 async with:

import pycronet

async with pycronet.AsyncSession(chrometls="chrome_150") as session:
    async with session.websocket(
        "wss://echo.websocket.events",
        subprotocols=["chat"],
        origin="https://example.com",
        headers={"X-Client": "pycronet"},
    ) as ws:
        await ws.send_text("hello")
        print(await ws.recv())
        await ws.send_binary(b"data")

subprotocols 是子协议候选列表,会放入 Sec-WebSocket-Protocol 请求头。服务端从中选择一个��在握手响应中确认;没有支持的协议时可能拒绝连接。它不是 Content-Type,也不会改变单条消息的文本或二进制类型。

send_text() 发送 UTF-8 文本,send_binary() 发送 bytes,recv() 返回 str 或 bytes。正常关闭会抛出 ConnectionClosed;网络错误抛出 WebSocketError。

需要回调风格时使用 WebSocketApp。run_forever() 默认阻塞当前线程,传入 blocking=False 可在后台线程运行。回调参数依次是 WebSocketApp、消息、关闭码和原因,或异常���象:

def on_open(ws):
    ws.send("hello")

def on_message(ws, message, is_text):
    print("received:", message, "text=" + str(is_text))

def on_close(ws, code, reason, was_clean):
    print("closed:", code, reason, "clean=" + str(was_clean))

def on_error(ws, error, net_error):
    print("error:", error, "net_error=" + str(net_error))

with pycronet.Session() as session:
    app = pycronet.WebSocketApp(
        session, "wss://echo.websocket.events",
        on_open=on_open, on_message=on_message,
        on_close=on_close, on_error=on_error,
        subprotocols=["chat"],
    )
    app.run_forever()

文件上传和下载

with pycronet.Session() as session:
    session.upload_file("https://example.com/upload", "image.png",
                        field_name="file", additional_fields={"tag": "demo"})
    result = session.download_file("https://example.com/archive.zip",
                                   "downloads/archive.zip")
    print(result["size"])

一次性同步函数 pycronet.upload_filepycronet.download_file 会自动创建临时 Session:

uploaded = pycronet.upload_file(
    "https://example.com/upload", "image.png",
    field_name="file", additional_fields={"tag": "demo"}, timeout=30.0,
)
downloaded = pycronet.download_file(
    "https://example.com/archive.zip", "downloads/archive.zip", timeout=60.0,
)
print(uploaded.status_code, downloaded["file_path"], downloaded["size"])

异步函数 pycronet.async_upload_filepycronet.async_download_file 需要使用 await

async def transfer():
    uploaded = await pycronet.async_upload_file(
        "https://example.com/upload", "image.png", field_name="file"
    )
    downloaded = await pycronet.async_download_file(
        "https://example.com/archive.zip", "downloads/archive.zip"
    )
    return uploaded.status_code, downloaded["file_path"]

测试和性能评测

py -3.12 -m pip install -e .[test]
py -3.12 -m pytest -q
py -3.12 -m pip install tox
tox

tox 配置覆盖 Python 3.9 至 3.14,未安装的解释器会跳过。基准脚本在独立子进程中比较 pycronet、requests、httpx、aiohttp 和 curl_cffi:

py -3.12 benchmarks/run_all.py
py -3.12 benchmarks/run_all.py --quick

结果写入 benchmarks/results/,按相同模式、payload、并发条件分组,再按 RPS 降序和 p50 延迟升序排列。该基准主要衡量本地 HTTP 桥接和连接复用开销,不能替代公网 HTTPS、代理、上传或 WebSocket 专项测试。

目录结构

pycronet/
  __init__.py             公共导出
  api.py                  模块级请求函数
  _session.py             Session 和 AsyncSession
  _transport.py           同步/异步传输层
  response.py             Response 和 StreamResponse
  cookies.py              CookieJar
  profiles.py             TLS profile 注册与序列化
  tls_ids.py              TLS 名称到十六进制 ID 的映射
  websocket.py            WebSocket 和 WebSocketApp
  _native/                ctypes 绑定和事件分发
  _native_bin/            运行时 DLL
native_shim/               C++17 Shim 工程
patch_chromium_145_0_7632_162/
                            Chromium 145 补丁和构建说明
tests/                     Python 测试
benchmarks/                性能脚本和结果
tls_fingerprint_server/    本地 TLS 指纹测试服务

带下划线的模块是内部实现细节,不建议业务代码直接导入。

已知限制

  • 仅支持 Windows x64,不支持 ARM64、Linux 或 macOS。
  • 不支持 free-threaded Python。
  • Shim 和 Cronet DLL 必须使用同一版本和架构。
  • TLS profile 不能突破 Chromium 145/BoringSSL 的能力。
  • verify=False 会降低证书安全性,只应在受控测试环境使用。
  • 升级 Chromium 后必须重新验证 ABI、DLL 和 profile。

参考和致谢

本项目参考了 cronet-cloakcyCronetpython-cronet。感谢这些项目及其贡献者提供的实践经验。

About

基于cronet,完整模拟谷歌浏览器https/wss请求协议指纹,可自定义tls套件,设置代理

Topics

Resources

Stars

0 stars

Watchers

0 watching

Forks

Releases

Packages

Contributors

Languages