爬虫-curl_cffi

这篇讲什么

  • curl_cffi 0.17.0 在 Python 3.14.10 上逐条实测:它把 libcurl(curl-impersonate 版)封成一套 requests 风格的接口,get/post/Session/AsyncSession 都能用,卖点是一个 impersonate 参数就把 TLS 指纹、HTTP/2 帧、请求头顺序一起换成真实浏览器。
  • 本篇的分工口径:这台机器的外网链路会掐掉带指纹的握手(五个在线站 × 两种指纹取法取样 20 次,16 次 Timeout、3 次 SSLError、1 次侥幸 200),所以抓在线站只能走默认栈,指纹相关的取证全部放在局域网靶站和本机 ClientHello 嗅探里做,每条结论都跟着逐字输出。
  • 三件必须先知道的事:BrowserType 枚举有 53 个成员但不含 chrome 这种别名(别名只在 BrowserTypeLiteral 的 63 个取值里);Session 默认 timeout=30 而不是 requests 的 None;files= 这个名字还在但一用就 NotImplementedError,上传要走 multipart=CurlMime。

装进来有什么

curl_cffi 是「libcurl 的 Python 绑定 + 一层 requests 风格接口」,装的时候只带一个编译好的 curl_cffi 二进制轮子,依赖极少(本机这个虚拟环境里一共只有 5 个包)。先把它自己是什么版本、底层 curl 是什么版本、导出面有多大看清楚:

01_env_and_api.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
# -*- coding: utf-8 -*-
"""取证 01:包与版本面(curl_cffi 0.17.0 本机实跑)。

只读 API 面,不发网络请求,不起靶站,因此输出天然可复现。
"""
import os
import sys
import sysconfig
from collections import Counter
from typing import get_args

import curl_cffi
from curl_cffi import requests as rq
from curl_cffi.requests import BrowserType, BrowserTypeLiteral


def group_key(name):
n = name.lower()
if n.startswith("chrome"):
kind = "chrome"
elif n.startswith("edge"):
kind = "edge"
elif n.startswith("safari"):
kind = "safari"
elif n.startswith("firefox"):
kind = "firefox"
elif n.startswith("tor"):
kind = "tor"
else:
kind = "other"
suffix = "plain"
if "_macos" in n:
suffix = "_macos"
elif "_ios" in n:
suffix = "_ios"
elif "_android" in n:
suffix = "_android"
return "%s/%s" % (kind, suffix)



def wide(items, per=8, pad=" "):
"""长列表分行打,免得一行超过门禁的 400 字符红线。"""
for i in range(0, len(items), per):
print(pad + ", ".join(str(x) for x in items[i:i + per]))


print("python:", sys.version.split()[0])
print("curl_cffi.__version__:", curl_cffi.__version__)
print("curl_cffi.__curl_version__:", curl_cffi.__curl_version__)
print("has __libcurl_version__:", hasattr(curl_cffi, "__libcurl_version__"))
print("curl_cffi.__file__:", os.path.basename(os.path.dirname(curl_cffi.__file__)))
print("title:", curl_cffi.__title__, "| description:", curl_cffi.__description__)

print()
print("== curl_cffi.requests.__all__ (%d 项) ==" % len(rq.__all__))
wide(sorted(rq.__all__), per=7)

print()
print("== curl_cffi.__all__ (%d 项) ==" % len(curl_cffi.__all__))
wide(sorted(curl_cffi.__all__), per=7)

print()
print("== BrowserType 类型 ==")
print("type(BrowserType):", type(BrowserType))
print("BrowserType.__mro__:", [c.__name__ for c in BrowserType.__mro__])
members = list(BrowserType)
names = [m.value for m in members]
print("成员总数:", len(members))
print("按浏览器/后缀分组:")
for k, v in sorted(Counter(group_key(n) for n in names).items()):
print(" %-16s %d" % (k, v))
print("前 8 个原文(枚举名 = 值):")
for m in members[:8]:
print(" BrowserType.%s = %r" % (m.name, m.value))
print("最后 6 个原文:")
for m in members[-6:]:
print(" BrowserType.%s = %r" % (m.name, m.value))
print("含 _macos 的:", [n for n in names if "_macos" in n])
print("含 android 的:", [n for n in names if "android" in n])
print("含 tor 的:", [n for n in names if "tor" in n])
print("枚举里有没有别名 chrome/firefox/safari:", [n for n in ("chrome", "firefox", "safari", "edge") if n in names])

print()
print("== BrowserTypeLiteral 是什么类型 ==")
print("BrowserTypeLiteral 是 typing.Literal,取值 %d 个:" % len(get_args(BrowserTypeLiteral)))
wide(sorted(get_args(BrowserTypeLiteral)), per=9, pad=" ")
print("typing.get_origin:", getattr(BrowserTypeLiteral, "__origin__", None))
args = get_args(BrowserTypeLiteral)
print("取值个数:", len(args))
diff = sorted(set(args) - set(names))
print("与 BrowserType 差集(Literal 有、枚举没有)%d 个:" % len(diff))
wide(diff, per=6, pad=" ")
print("差集(枚举有、Literal 没有):", sorted(set(names) - set(args)))

print()
print("== 顶层模块级函数(dir 里非下划线、且可调用) ==")
top = [n for n in dir(curl_cffi) if not n.startswith("_")]
funcs = sorted(n for n in top if callable(getattr(curl_cffi, n)))
wide(funcs, per=7)
print("set_default_curl_timeout 存在:", hasattr(curl_cffi, "set_default_curl_timeout"))
print("get_async_session_timeout 存在:", hasattr(curl_cffi, "get_async_session_timeout"))
timeout_like = sorted(n for n in top if "timeout" in n.lower())
print("含 timeout 的顶层名字:", timeout_like)
submodules = sorted(n for n in top if not callable(getattr(curl_cffi, n)))
print("非可调用顶层名字:", submodules)

print()
print("== 命令行入口 ==")
scripts_dir = sysconfig.get_paths()["scripts"]
print("scripts 目录在 venv 里 =", scripts_dir.lower().endswith("scripts"))
exes = sorted(f for f in os.listdir(scripts_dir) if f.lower().startswith("curl"))
print("curl 相关可执行文件:", exes)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
python: 3.14.10
curl_cffi.__version__: 0.17.0
curl_cffi.__curl_version__: libcurl/8.23.0-IMPERSONATE BoringSSL zlib/1.3.1 brotli/1.2.1 zstd/1.5.7 c-ares/1.34.9 WinIDN nghttp2/1.63.0 ngtcp2/1.20.0 nghttp3/1.15.0 curl-impersonate/2.3.1
has __libcurl_version__: False
curl_cffi.__file__: curl_cffi
title: curl_cffi | description: libcurl ffi bindings for Python, with impersonation support.
== curl_cffi.requests.__all__ (32 项) ==
AsyncSession, AsyncWebSocket, BrowserType, BrowserTypeLiteral, CacheBackend, CookieTypes, Cookies
CurlWsFlag, ExtraFingerprints, FileCacheBackend, HeaderTypes, Headers, ProxySpec, Request
RequestsError, Response, RetryStrategy, Session, WebSocket, WebSocketClosed, WebSocketError
WebSocketRetryStrategy, WebSocketTimeout, WsCloseCode, delete, get, head, options
patch, post, put, request
== curl_cffi.__all__ (55 项) ==
AsyncCurl, AsyncSession, AsyncWebSocket, BrowserType, BrowserTypeLiteral, CacheBackend, CookieTypes
Cookies, Curl, CurlECode, CurlError, CurlFollow, CurlHttpVersion, CurlInfo
CurlLockAccess, CurlLockData, CurlMOpt, CurlMime, CurlOpt, CurlProxyType, CurlShare
CurlShareCode, CurlShareOpt, CurlSslVersion, CurlWsFlag, CurlWsOption, ExtraFingerprints, FileCacheBackend
Fingerprint, FingerprintManager, HeaderTypes, Headers, ProxySpec, Request, Response
Session, WebSocket, WebSocketClosed, WebSocketError, WebSocketRetryStrategy, WebSocketTimeout, WsCloseCode
config_warnings, delete, exceptions, ffi, get, get_fingerprint, head
lib, options, patch, post, put, request
== BrowserType 类型 ==
type(BrowserType): <class 'enum.EnumType'>
BrowserType.__mro__: ['BrowserType', 'str', 'Enum', 'object']
成员总数: 53
按浏览器/后缀分组:
chrome/_android 2
chrome/_macos 2
chrome/plain 20
edge/plain 2
firefox/plain 7
safari/_ios 7
safari/plain 12
tor/plain 1
前 8 个原文(枚举名 = 值):
BrowserType.edge99 = 'edge99'
BrowserType.edge101 = 'edge101'
BrowserType.chrome99 = 'chrome99'
BrowserType.chrome100 = 'chrome100'
BrowserType.chrome101 = 'chrome101'
BrowserType.chrome104 = 'chrome104'
BrowserType.chrome107 = 'chrome107'
BrowserType.chrome110 = 'chrome110'
最后 6 个原文:
BrowserType.safari15_3 = 'safari15_3'
BrowserType.safari15_5 = 'safari15_5'
BrowserType.safari17_0 = 'safari17_0'
BrowserType.safari17_2_ios = 'safari17_2_ios'
BrowserType.safari18_0 = 'safari18_0'
BrowserType.safari18_0_ios = 'safari18_0_ios'
含 _macos 的: ['chrome151_macos', 'chrome152_macos']
含 android 的: ['chrome99_android', 'chrome131_android']
含 tor 的: ['tor145']
枚举里有没有别名 chrome/firefox/safari: []
== BrowserTypeLiteral 是什么类型 ==
BrowserTypeLiteral 是 typing.Literal,取值 63 个:
chrome, chrome100, chrome101, chrome104, chrome107, chrome110, chrome116, chrome119, chrome120
chrome123, chrome124, chrome131, chrome131_android, chrome133a, chrome136, chrome142, chrome145, chrome146
chrome150, chrome151, chrome151_macos, chrome152, chrome152_macos, chrome99, chrome99_android, chrome_android, edge
edge101, edge99, firefox, firefox133, firefox135, firefox144, firefox147, firefox153, firefox154
firefox155, safari, safari153, safari155, safari15_3, safari15_5, safari170, safari172_ios, safari17_0
safari17_2_ios, safari180, safari180_ios, safari184, safari184_ios, safari18_0, safari18_0_ios, safari18_4, safari18_4_ios
safari260, safari2601, safari260_ios, safari270, safari270_ios, safari_beta, safari_ios, safari_ios_beta, tor145
typing.get_origin: typing.Literal
取值个数: 63
与 BrowserType 差集(Literal 有、枚举没有)10 个:
chrome, chrome_android, edge, firefox, safari, safari18_4
safari18_4_ios, safari_beta, safari_ios, safari_ios_beta
差集(枚举有、Literal 没有): []
== 顶层模块级函数(dir 里非下划线、且可调用) ==
AsyncCurl, AsyncSession, AsyncWebSocket, BrowserType, BrowserTypeLiteral, CacheBackend, Cookies
Curl, CurlECode, CurlError, CurlFollow, CurlHttpVersion, CurlInfo, CurlLockAccess
CurlLockData, CurlMOpt, CurlMime, CurlOpt, CurlProxyType, CurlShare, CurlShareCode
CurlShareOpt, CurlSslVersion, CurlWsFlag, CurlWsOption, ExtraFingerprints, FileCacheBackend, Fingerprint
FingerprintManager, Headers, ProxySpec, Request, Response, Session, WebSocket
WebSocketClosed, WebSocketError, WebSocketRetryStrategy, WebSocketTimeout, WsCloseCode, config_warnings, delete
get, get_fingerprint, head, options, patch, post, put
request
set_default_curl_timeout 存在: False
get_async_session_timeout 存在: False
含 timeout 的顶层名字: ['WebSocketTimeout']
非可调用顶层名字: ['CookieTypes', 'HeaderTypes', 'aio', 'const', 'curl', 'exceptions', 'ffi', 'fingerprints', 'lib', 'requests', 'utils']
== 命令行入口 ==
scripts 目录在 venv 里 = True
curl 相关可执行文件: ['curl-cffi.exe']

逐字输出里有四个要点值得单独记。

  • 版本号有两个:curl_cffi.__version__ 是这个包的版本(0.17.0),__curl_version__ 是底层 libcurl 的完整特性串,里面能看到 BoringSSL、nghttp2(HTTP/2)、ngtcp2/nghttp3(HTTP/3)、c-ares(异步 DNS)和 curl-impersonate/2.3.1。文档里出现的 __libcurl_version__ 在本机不存在,hasattr 直接是 False,写它会拿到空值而不是报错,属于静默坑。
  • 两套导出面不一样:curl_cffi.requests.__all__ 是 requests 那一层(Session、AsyncSession、get/post/...、异常、ExtraFingerprints、ProxySpec、WebSocket 相关),顶层 curl_cffi.__all__ 还多出 Curl、AsyncCurl、CurlMime、CurlECode 这些底层绑定。日常写爬虫只用前者;要设 curl_options 这种冷门开关才需要后者。
  • 能模仿的浏览器数清楚了:BrowserType 是 str 加 Enum 的组合类,53 个成员按前缀分布是 chrome 20 个、chrome/_macos 2 个、chrome/_android 2 个、edge 2 个、safari 12 个、safari/_ios 7 个、firefox 7 个、tor 1 个。枚举里没有 chrome/firefox/safari/edge 这些「只用品牌名」的别名,它们只出现在 BrowserTypeLiteral(typing.Literal[...],63 个取值)里——所以写 BrowserType.chrome 会 AttributeError,而写 impersonate="chrome" 是合法的,因为函数签名走的是 Literal 那条路。
  • 顶层没有任何模块级函数:set_default_curl_timeout、get_async_session_timeout 这类在老文档里出现过的名字本机都没有,超时只能是 Session 或单次请求的参数。

三种用法

官方 README 给的用法有三档,档位之间不是「功能多少」的差别,而是连接和 Cookie 会不会被复用的差别。这一段直接打本机靶站,把「服务端看到的客户端端口」「同一条连接上累计了几个请求」「Cookie 头」三个观测点打出来:

02_interfaces.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
# -*- coding: utf-8 -*-
"""取证 02:两套接口对照 —— 顶层函数式 vs Session vs AsyncSession。

看三件事:连接是否复用(客户端临时端口 + 同一条 TCP 连接上的请求数)、
Cookie 是否共享、async with 是否支持。靶站只在 8050,跑完 shutdown()。
"""
import asyncio
import json
import socketserver
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from curl_cffi import requests as rq
from curl_cffi.requests import AsyncSession, Session

PORT = 8050
STATE = {"n": 0}
CONN = {}
LOCK = threading.Lock()


class EchoHandler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"

def log_message(self, *args):
pass

def do_GET(self):
client_port = self.client_address[1]
with LOCK:
STATE["n"] += 1
seq = STATE["n"]
CONN[client_port] = CONN.get(client_port, 0) + 1
on_conn = CONN[client_port]
payload = {
"seq": seq,
"client_port": client_port,
"conn_requests": on_conn,
"cookie": self.headers.get("Cookie"),
}
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.send_header("Set-Cookie", "sid=abc; Path=/")
self.end_headers()
self.wfile.write(body)


class StrictServer(ThreadingHTTPServer):
"""allow_reuse_address 在 Windows 上是 1,必须显式关掉,否则同端口会被重复绑定。"""

daemon_threads = True
allow_reuse_address = False

def server_bind(self):
socketserver.TCPServer.server_bind(self)


def serve(port):
httpd = StrictServer(("127.0.0.1", port), EchoHandler)
assert httpd.server_address[1] == port, httpd.server_address
threading.Thread(target=httpd.serve_forever, daemon=True).start()
return httpd


def trial(label, fn, times):
before = STATE["n"]
with LOCK:
CONN.clear()
cookies = []
ports = set()
conn_requests = []
for _ in range(times):
j = fn().json()
cookies.append(j["cookie"])
ports.add(j["client_port"])
conn_requests.append(j["conn_requests"])
print("[%s] 发起 %d 次 -> 靶站命中 %d 次;不同客户端端口 %d 个;同连接累计请求数序列 %r"
% (label, times, STATE["n"] - before, len(ports), conn_requests))
print(" 靶站每次看到的 Cookie 头: %r" % (cookies,))


base = "http://127.0.0.1:%d/echo" % PORT
httpd = serve(PORT)
try:
print("== A. 顶层函数式 rq.get(源码里每次 with Session(...) 新建)==")
trial("top-level get x3", lambda: rq.get(base), 3)

print()
print("== B. 同一个 Session 连发 3 次 ==")
s = Session()
trial("Session.get x3", lambda: s.get(base), 3)
print(" session.cookies 内容:", dict(s.cookies))
s.close()

print()
print("== C. 两个互不相干的 Session ==")
s1, s2 = Session(), Session()
s1.get(base)
print(" s1 再发一次 -> Cookie: %r" % s1.get(base).json()["cookie"])
print(" s2 首次发送 -> Cookie: %r" % s2.get(base).json()["cookie"])
s1.close()
s2.close()

print()
print("== D. AsyncSession ==")
print(" hasattr(AsyncSession, '__aenter__'):", hasattr(AsyncSession, "__aenter__"))
print(" hasattr(AsyncSession, '__aexit__') :", hasattr(AsyncSession, "__aexit__"))
print(" hasattr(Session, '__enter__') :", hasattr(Session, "__enter__"))

async def amain():
async with AsyncSession() as a:
before = STATE["n"]
ports = set()
cookies = []
conn_requests = []
for _ in range(3):
j = (await a.get(base)).json()
ports.add(j["client_port"])
cookies.append(j["cookie"])
conn_requests.append(j["conn_requests"])
print(" [await 串行 x3] 命中 %d 次;不同端口 %d 个;同连接请求数 %r;Cookie %r"
% (STATE["n"] - before, len(ports), conn_requests, cookies))
before_g = STATE["n"]
g = await asyncio.gather(*[a.get(base) for _ in range(4)])
print(" [gather 并发 x4] 靶站记到的相对到达序号:",
sorted(x.json()["seq"] - before_g for x in g))
print(" async with 里 cookie 是否共享:", dict(a.cookies))

asyncio.run(amain())

print()
print("== E. Session 关闭后再用 ==")
s3 = Session()
s3.close()
try:
s3.get(base)
except Exception as e:
print(" %s: %s" % (type(e).__name__, e))

print()
print("== F. with 语句(Session 支持同步上下文管理器)==")
with Session() as s4:
print(" with 内请求 status:", s4.get(base).status_code)
try:
s4.get(base)
except Exception as e:
print(" with 外再用: %s: %s" % (type(e).__name__, e))
finally:
httpd.shutdown()
httpd.server_close()
print()
print("靶站已关闭,本次累计命中:", STATE["n"])
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
== A. 顶层函数式 rq.get(源码里每次 with Session(...) 新建)==
[top-level get x3] 发起 3 次 -> 靶站命中 3 次;不同客户端端口 3 个;同连接累计请求数序列 [1, 1, 1]
靶站每次看到的 Cookie 头: [None, None, None]
== B. 同一个 Session 连发 3 次 ==
[Session.get x3] 发起 3 次 -> 靶站命中 3 次;不同客户端端口 1 个;同连接累计请求数序列 [1, 2, 3]
靶站每次看到的 Cookie 头: [None, 'sid=abc', 'sid=abc']
session.cookies 内容: {'sid': 'abc'}
== C. 两个互不相干的 Session ==
s1 再发一次 -> Cookie: 'sid=abc'
s2 首次发送 -> Cookie: None
== D. AsyncSession ==
hasattr(AsyncSession, '__aenter__'): True
hasattr(AsyncSession, '__aexit__') : True
hasattr(Session, '__enter__') : True
[await 串行 x3] 命中 3 次;不同端口 1 个;同连接请求数 [1, 2, 3];Cookie [None, 'sid=abc', 'sid=abc']
[gather 并发 x4] 靶站记到的相对到达序号: [1, 2, 3, 4]
async with 里 cookie 是否共享: {'sid': 'abc'}
== E. Session 关闭后再用 ==
SessionClosed: Session is closed, cannot send request.
== F. with 语句(Session 支持同步上下文管理器)==
with 内请求 status: 200
with 外再用: SessionClosed: Session is closed, cannot send request.
靶站已关闭,本次累计命中: 17

结果和 requests 的习惯是一致的,但有两个 curl_cffi 特有的细节。

  • 顶层的 curl_cffi.requests.get(url) 每次都新建一个 Session:靶站三次都看到不同的客户端端口,同连接累计请求数 序列是 [1, 1, 1],Cookie 头一直是 None。写一次性脚本无所谓,批量抓取就亏了——每次都要重新握手。
  • Session 复用连接:三个请求同一个端口,序列变成 [1, 2, 3],服务端 Set-Cookie 之后从第二次请求起就自动带上 sid=abc,session.cookies 也能读到 {'sid': 'abc'}。
  • AsyncSession 行为同上,且支持 async with;asyncio.gather 并发四个请求时,靶站记录的到达序号是 [1, 2, 3, 4]——并发不等于同时,真正同时在跑几个由 max_clients 决定(见「下载与并发」一节)。
  • Session 关掉之后再用,抛的是 SessionClosed: Session is closed, cannot send request.,不是 requests 那种 ValueError。

impersonate 改了什么

impersonate 是本库存在的理由。它改的不只是 User-Agent,还有请求头顺序、HTTP/2 的 SETTINGS 帧、TLS 的 ClientHello。头顺序这一层在本机靶站上就能看清楚,而且能看出「别名」和「具体版本号」的差别:

03_impersonate.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
# -*- coding: utf-8 -*-
"""取证 03:impersonate 的真实效果 —— 靶站看到的请求头顺序 + UA 主版本。

同一个本机回显靶站(8050),分别用 impersonate="chrome" / "chrome131" /
"firefox133" / "safari17_0" / 不传 各发一次。
只打印 UA 里的浏览器主版本,不打印 OS 段(不同机器/随机生成会变)。
最后用在线靶站 https://ssr1.scrape.center 验证 response.http_version。
"""
import json
import re
import socketserver
import threading
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from curl_cffi import requests as rq
from curl_cffi.const import CurlHttpVersion

PORT = 8050
COUNT = {"n": 0}
LOCK = threading.Lock()


class EchoHandler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"

def log_message(self, *args):
pass

def do_GET(self):
with LOCK:
COUNT["n"] += 1
order = [k for k in self.headers.keys()]
ua = self.headers.get("User-Agent", "")
payload = {
"order": order,
"accept_encoding": self.headers.get("Accept-Encoding"),
"sec_fetch_mode": self.headers.get("Sec-Fetch-Mode"),
"ua": ua,
"http_version": self.request_version,
}
body = json.dumps(payload).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)


class StrictServer(ThreadingHTTPServer):
daemon_threads = True
allow_reuse_address = False

def server_bind(self):
socketserver.TCPServer.server_bind(self)


httpd = StrictServer(("127.0.0.1", PORT), EchoHandler)
assert httpd.server_address[1] == PORT
threading.Thread(target=httpd.serve_forever, daemon=True).start()

UA_RE = re.compile(r"(Chrome|Firefox|Safari)/([0-9]+)")
VER_RE = re.compile(r"Version/([0-9]+)")


def ua_major(ua):
"""只取主版本号。Safari 的 "Safari/605.1.15" 是内核号,浏览器主版本在 Version/ 段里。"""
m = VER_RE.search(ua)
parts = []
for b, v in UA_RE.findall(ua):
if b == "Safari" and m:
continue
parts.append("%s/%s" % (b, v))
if m:
parts.append("Version/%s(Safari 浏览器主版本)" % m.group(1))
if not parts:
return "无浏览器标识: %r" % ua
return ", ".join(parts)


base = "http://127.0.0.1:%d/echo" % PORT
s = rq.Session()
try:
cases = [
("不传 impersonate(默认)", None),
('impersonate="chrome"', "chrome"),
('impersonate="chrome131"', "chrome131"),
('impersonate="firefox133"', "firefox133"),
('impersonate="safari17_0"', "safari17_0"),
]
for label, value in cases:
kwargs = {} if value is None else {"impersonate": value}
j = s.get(base, **kwargs).json()
print("== %s ==" % label)
print(" 靶站侧协议:", j["http_version"])
print(" 头键顺序(%d 个): %s" % (len(j["order"]), " -> ".join(j["order"])))
print(" UA 浏览器主版本:", ua_major(j["ua"]))
print(" Accept-Encoding: %r Sec-Fetch-Mode: %r" % (j["accept_encoding"], j["sec_fetch_mode"]))
print()

print("== 同一版本的不同写法是否等价 ==")
a = s.get(base, impersonate="safari17_0").json()["order"]
b = s.get(base, impersonate="safari170").json()["order"]
print(" safari17_0 与 safari170 头顺序相同:", a == b)
print(" 本机 chrome 别名实际解析到的版本:", end=" ")
print(ua_major(s.get(base, impersonate="chrome").json()["ua"]))

print()
print()
print("== 自带 headers 与 impersonate 默认头的关系 ==")
j = s.get(base, impersonate="chrome131", headers={"user-agent": "mybot/1.0", "x-extra": "1"}).json()
print(" 自定义 UA 是否覆盖 impersonate 的 UA:", j["ua"] == "mybot/1.0")
print(" 头键顺序:", " -> ".join(j["order"]))
j = s.get(base, impersonate="chrome131", default_headers=False).json()
print(" default_headers=False 后靶站只剩 %d 个头: %s" % (len(j["order"]), " -> ".join(j["order"])))
print()

print("== response.http_version(本机 HTTP/1.1 靶站)==")
r = s.get(base)
print(" 原始值:", r.http_version, type(r.http_version).__name__,
"->", CurlHttpVersion(r.http_version).name)
print(" CurlHttpVersion 枚举:", [(x.name, int(x)) for x in CurlHttpVersion])

print()
print("== 在线靶站的 http_version 对照(HTTPS)==")
ONLINE = "https://tls.peet.ws/api/all"
try:
r3 = rq.get(ONLINE, timeout=12)
print(" curl_cffi 默认 status=%s http_version=%s(%s)"
% (r3.status_code, r3.http_version, CurlHttpVersion(r3.http_version).name))
except Exception as e:
print(" curl_cffi 默认 失败: %s: %s" % (type(e).__name__, str(e)[:60]))
try:
r4 = rq.get(ONLINE, timeout=12, http_version="v1")
print(" curl_cffi(http_version='v1') status=%s http_version=%s(%s)"
% (r4.status_code, r4.http_version, CurlHttpVersion(r4.http_version).name))
except Exception as e:
print(" curl_cffi(http_version='v1') 失败: %s: %s" % (type(e).__name__, str(e)[:60]))
try:
req = urllib.request.Request(ONLINE, headers={"User-Agent": "probe"})
with urllib.request.urlopen(req, timeout=12) as fp:
print(" 标准库 urllib.request status=%s fp.version=%s(标准库只会 HTTP/1.1)"
% (fp.status, fp.version))
except Exception as e:
print(" 标准库 urllib.request 失败: %s: %s" % (type(e).__name__, str(e)[:60]))
print(" 注:本机没装 requests 包,HTTP/1.1 一侧用标准库 urllib 作对照;")
print(" curl_cffi 默认就协商到 h2,这是 requests/urllib 做不到的。")
print(" 在线请求一律不带 impersonate:本机网络会丢掉 impersonate 的 ClientHello,")
print(" TLS 差异证据见 04_tls_fingerprint.py。")
print()
print("靶站累计命中:", COUNT["n"])
finally:
s.close()
httpd.shutdown()
httpd.server_close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
== 不传 impersonate(默认) ==
靶站侧协议: HTTP/1.1
头键顺序(2 个): Host -> Accept-Encoding
UA 浏览器主版本: 无浏览器标识: ''
Accept-Encoding: 'gzip, deflate, br' Sec-Fetch-Mode: None
== impersonate="chrome" ==
靶站侧协议: HTTP/1.1
头键顺序(17 个): sec-ch-ua -> sec-ch-ua-mobile -> sec-ch-ua-platform -> upgrade-insecure-requests -> user-agent -> accept -> sec-fetch-site -> sec-fetch-mode -> sec-fetch-user -> sec-fetch-dest -> accept-encoding -> accept-language -> cache-control -> Host -> Upgrade -> HTTP2-Settings -> Connection
UA 浏览器主版本: Chrome/152, Safari/537
Accept-Encoding: 'gzip, deflate, br, zstd' Sec-Fetch-Mode: 'navigate'
== impersonate="chrome131" ==
靶站侧协议: HTTP/1.1
头键顺序(16 个): Host -> Upgrade -> HTTP2-Settings -> sec-ch-ua -> sec-ch-ua-mobile -> sec-ch-ua-platform -> Upgrade-Insecure-Requests -> User-Agent -> Accept -> Sec-Fetch-Site -> Sec-Fetch-Mode -> Sec-Fetch-User -> Sec-Fetch-Dest -> Accept-Encoding -> Accept-Language -> Connection
UA 浏览器主版本: Chrome/131, Safari/537
Accept-Encoding: 'gzip, deflate, br, zstd' Sec-Fetch-Mode: 'navigate'
== impersonate="firefox133" ==
靶站侧协议: HTTP/1.1
头键顺序(15 个): Host -> Upgrade -> HTTP2-Settings -> User-Agent -> Accept -> Accept-Language -> Accept-Encoding -> Upgrade-Insecure-Requests -> Sec-Fetch-Dest -> Sec-Fetch-Mode -> Sec-Fetch-Site -> Sec-Fetch-User -> Priority -> Te -> Connection
UA 浏览器主版本: Firefox/133
Accept-Encoding: 'gzip, deflate, br, zstd' Sec-Fetch-Mode: 'navigate'
== impersonate="safari17_0" ==
靶站侧协议: HTTP/1.1
头键顺序(11 个): Host -> Upgrade -> HTTP2-Settings -> Accept -> Sec-Fetch-Site -> Accept-Encoding -> Sec-Fetch-Mode -> user-agent -> Accept-Language -> Sec-Fetch-Dest -> Connection
UA 浏览器主版本: Version/17(Safari 浏览器主版本)
Accept-Encoding: 'gzip, deflate, br' Sec-Fetch-Mode: 'navigate'
== 同一版本的不同写法是否等价 ==
safari17_0 与 safari170 头顺序相同: True
本机 chrome 别名实际解析到的版本: Chrome/152, Safari/537
== 自带 headers 与 impersonate 默认头的关系 ==
自定义 UA 是否覆盖 impersonate 的 UA: True
头键顺序: Host -> Upgrade -> HTTP2-Settings -> sec-ch-ua -> sec-ch-ua-mobile -> sec-ch-ua-platform -> Upgrade-Insecure-Requests -> user-agent -> Accept -> Sec-Fetch-Site -> Sec-Fetch-Mode -> Sec-Fetch-User -> Sec-Fetch-Dest -> Accept-Encoding -> Accept-Language -> x-extra -> Connection
default_headers=False 后靶站只剩 5 个头: Host -> Accept-Encoding -> Upgrade -> HTTP2-Settings -> Connection
== response.http_version(本机 HTTP/1.1 靶站)==
原始值: 2 int -> V1_1
CurlHttpVersion 枚举: [('NONE', 0), ('V1_0', 1), ('V1_1', 2), ('V2_0', 3), ('V2TLS', 4), ('V2_PRIOR_KNOWLEDGE', 5), ('V3', 30), ('V3ONLY', 31)]
== 在线靶站的 http_version 对照(HTTPS)==
curl_cffi 默认 status=200 http_version=3(V2_0)
curl_cffi(http_version='v1') status=200 http_version=2(V1_1)
标准库 urllib.request status=200 fp.version=11(标准库只会 HTTP/1.1)
注:本机没装 requests 包,HTTP/1.1 一侧用标准库 urllib 作对照;
curl_cffi 默认就协商到 h2,这是 requests/urllib 做不到的。
在线请求一律不带 impersonate:本机网络会丢掉 impersonate 的 ClientHello,
TLS 差异证据见 04_tls_fingerprint.py。
靶站累计命中: 11

几个要记住的实测结论。

  • 不传 impersonate 时靶站只看到 2 个头(Host、Accept-Encoding),连 User-Agent 都没有——这是 curl 的裸默认,很多站点会因为这个直接拒你。
  • impersonate="chrome" 给的是 17 个头,sec-ch-ua 排在最前面,UA 主版本是 Chrome/152(本机 0.17.0 里「chrome」指向的具体版本,会随版本升级变,别把它当常量写进代码)。
  • impersonate="chrome131" 只有 16 个头且顺序不同(Host 在最前),说明别名和具体版本走的是两套模板,不是「别名 = 最新版」这么简单。
  • safari17_0 的 UA 里浏览器主版本在 Version/17 这一段(脚本只把主版本打出来,OS 段和内核号不打印,因为不同模板写的是同一套内核号,读起来容易误当成版本),safari17_0 与 safari170 头顺序完全相同,两种写法等价。
  • 自定义 headers={"User-Agent": ...} 会覆盖 impersonate 的 UA——想「保留浏览器指纹但换个自己的标识」要清楚这会让指纹自相矛盾。
  • default_headers=False 时只剩 5 个头,等于「只换 TLS 指纹、不发浏览器头」,一般只在专门测服务端时才用。
  • HTTP 版本这条也在同一个块里量出来了:本机明文靶站上 http_version 是 2(V1_1),HTTPS 在线站默认是 3(V2_0,即 HTTP/2),显式传 http_version="v1" 会退回 HTTP/1.1;作为对照,标准库 urllib 永远是 11(只会 HTTP/1.1)。CurlHttpVersion 枚举的实测取值是 NONE=0、V1_0=1、V1_1=2、V2_0=3、V2TLS=4、V2_PRIOR_KNOWLEDGE=5、V3=30、V3ONLY=31。

TLS 指纹怎么看

「TLS 指纹真的变了」这句话不能只靠文档,得自己把 ClientHello 抓下来看。这一段在本机开一个 TLS 服务器,把每个 impersonate 取值握手时的密码套件和扩展类型抄下来,去掉 GREASE(Chrome 每次插入的随机占位扩展)后排序比较:

04_tls_fingerprint.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
# -*- coding: utf-8 -*-
"""取证 04:impersonate 是否真的改了 TLS 指纹(本机 ClientHello 嗅探 + 在线补充)。

思路 A(本机,不依赖外网):在 8050 起一个"只读 ClientHello 就断开"的嗅探靶站,
把 impersonate 各取值发出的 ClientHello 原样解出来。

注意:Chrome 系会插入 GREASE 随机值、并且随机置换扩展顺序,
所以"原始顺序"和"包长度"每次都变、不可复现。
这里打印的是【剔除 GREASE 后排序】的密码套件集合与扩展集合,
每个取值连发两次握手来确认这个集合是稳定的,再对集合串取 sha256 前 12 位。

思路 B(在线 tls.peet.ws):本机网络下只有不传 impersonate 的请求能完成 TLS 握手,
传了 impersonate 一律 curl:(28) 超时,所以在线侧只能给默认栈的指纹。
"""
import hashlib
import re
import socket
import struct
import threading

from curl_cffi import requests as rq

PORT = 8050
HELLO = []
LOCK = threading.Lock()


def is_grease(v):
"""TLS GREASE 码点:形如 0xXYXY,且 X != Y、低半字节固定为 0xA。"""
return (v >> 8) == (v & 0xFF) and (v & 0x0F) == 0x0A


def parse_client_hello(data):
pos = 0
body = b""
while pos + 5 <= len(data):
rlen = struct.unpack(">H", data[pos + 3:pos + 5])[0]
if pos + 5 + rlen > len(data):
break
body += data[pos + 5:pos + 5 + rlen]
pos += 5 + rlen
if body[:1] != b"\x01":
return None
hlen = struct.unpack(">I", b"\x00" + body[1:4])[0]
hs = body[4:4 + hlen]
hs_version = struct.unpack(">H", hs[0:2])[0]
p = 2 + 32
sid_len = hs[p]
p += 1 + sid_len
cs_len = struct.unpack(">H", hs[p:p + 2])[0]
p += 2
ciphers = [struct.unpack(">H", hs[p + i:p + i + 2])[0] for i in range(0, cs_len, 2)]
p += cs_len
comp_len = hs[p]
p += 1 + comp_len
ext_len = struct.unpack(">H", hs[p:p + 2])[0]
p += 2
exts = []
end = p + ext_len
while p + 4 <= end:
etype = struct.unpack(">H", hs[p:p + 2])[0]
elen = struct.unpack(">H", hs[p + 2:p + 4])[0]
exts.append(etype)
p += 4 + elen
clean_c = sorted(c for c in ciphers if not is_grease(c))
clean_e = sorted(e for e in exts if not is_grease(e))
return {
"hs_version": hex(hs_version),
"n_grease_ciphers": sum(1 for c in ciphers if is_grease(c)),
"n_grease_exts": sum(1 for e in exts if is_grease(e)),
"ciphers": ["%04x" % c for c in clean_c],
"exts": clean_e,
"fp": "%s|%s" % (",".join("%04x" % c for c in clean_c), ",".join(map(str, clean_e))),
}


def sniffer(sock):
try:
sock.settimeout(3)
buf = b""
while True:
try:
b = sock.recv(65535)
except socket.timeout:
break
if not b:
break
buf += b
if len(buf) >= 5:
need = struct.unpack(">H", buf[3:5])[0] + 5
if len(buf) >= need:
body = buf[5:need]
if len(body) >= 4:
hlen = struct.unpack(">I", b"\x00" + body[1:4])[0]
if len(body) >= 4 + hlen:
break
with LOCK:
HELLO.append(buf)
finally:
sock.close()


srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
srv.bind(("127.0.0.1", PORT))
srv.listen(8)


def accept_loop():
while True:
try:
conn, _addr = srv.accept()
except OSError:
break
threading.Thread(target=sniffer, args=(conn,), daemon=True).start()


threading.Thread(target=accept_loop, daemon=True).start()

URL = "https://127.0.0.1:%d/hello" % PORT
cases = [
("不传 impersonate(curl/BoringSSL 默认栈)", None),
('impersonate="chrome131"', "chrome131"),
('impersonate="chrome"', "chrome"),
('impersonate="firefox133"', "firefox133"),
('impersonate="safari17_0"', "safari17_0"),
]

print("== A. 本机嗅探 ClientHello(每个取值连发 2 次握手)==")
for label, value in cases:
kwargs = {} if value is None else {"impersonate": value}
infos = []
errs = set()
for _ in range(2):
HELLO.clear()
s = rq.Session(verify=False)
try:
s.get(URL, timeout=6, **kwargs)
except Exception as e:
errs.add(type(e).__name__)
finally:
s.close()
with LOCK:
got = list(HELLO)
if got:
infos.append(parse_client_hello(got[0]))
print("== %s ==" % label)
print(" 客户端异常(靶站故意掐断,正常):", sorted(errs))
infos = [i for i in infos if i]
if not infos:
print(" 没抓到 ClientHello")
continue
fps = {hashlib.sha256(i["fp"].encode()).hexdigest()[:12] for i in infos}
print(" 2 次握手的指纹集合是否一致:", len(fps) == 1, " 指纹 sha256 前 12 位:", sorted(fps))
i = infos[0]
print(" handshake 版本字段:", i["hs_version"],
" GREASE 个数: 密码套件 %d / 扩展 %d" % (i["n_grease_ciphers"], i["n_grease_exts"]))
print(" 密码套件 %d 个(去 GREASE 后排序): %s" % (len(i["ciphers"]), " ".join(i["ciphers"])))
print(" 扩展类型 %d 个(去 GREASE 后排序): %s" % (len(i["exts"]), i["exts"]))
print()

print("== B. 在线 tls.peet.ws(本机网络的实测限制)==")
for label, kwargs in [("默认(不传 impersonate)", {}), ("chrome131", {"impersonate": "chrome131"}),
("firefox133", {"impersonate": "firefox133"})]:
try:
j = rq.get("https://tls.peet.ws/api/all", timeout=12, **kwargs).json()
print(" %-22s http_version=%s ja4=%s peetprint_hash=%s akamai-h2=%s"
% (label, j["http_version"], j["tls"]["ja4"],
j["tls"]["peetprint_hash"], j["http2"]["akamai_fingerprint_hash"]))
except Exception as e:
print(" %-22s 失败: %s: %s" % (
label, type(e).__name__,
re.sub(r"after [0-9]+ (milliseconds|ms)", "after <耗时> ms", str(e))[:70]))
print(" 说明:本机网络(防火墙/中间设备)会把 impersonate 生成的 ClientHello 直接丢掉,")
print(" 握手 12 秒超时,所以在线侧只能拿到默认栈的指纹;TLS 层面的差异证据以 A 段为准。")
print(" 在线默认栈 ja4 开头 t13d2712h2:27 个密码套件、12 个扩展,和 A 段第一行完全对得上。")

srv.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
== A. 本机嗅探 ClientHello(每个取值连发 2 次握手)==
== 不传 impersonate(curl/BoringSSL 默认栈) ==
客户端异常(靶站故意掐断,正常): ['SSLError']
2 次握手的指纹集合是否一致: True 指纹 sha256 前 12 位: ['65b30e321074']
handshake 版本字段: 0x303 GREASE 个数: 密码套件 0 / 扩展 0
密码套件 27 个(去 GREASE 后排序): 002f 0033 0035 0039 003c 003d 0067 006b 009c 009d 009e 009f 1301 1302 1303 c009 c00a c013 c014 c024 c028 c02b c02c c02f c030 cca8 cca9
扩展类型 11 个(去 GREASE 后排序): [10, 11, 13, 16, 21, 23, 35, 43, 45, 51, 65281]
== impersonate="chrome131" ==
客户端异常(靶站故意掐断,正常): ['SSLError']
2 次握手的指纹集合是否一致: True 指纹 sha256 前 12 位: ['42282a53e158']
handshake 版本字段: 0x303 GREASE 个数: 密码套件 1 / 扩展 2
密码套件 15 个(去 GREASE 后排序): 002f 0035 009c 009d 1301 1302 1303 c013 c014 c02b c02c c02f c030 cca8 cca9
扩展类型 15 个(去 GREASE 后排序): [5, 10, 11, 13, 16, 18, 23, 27, 35, 43, 45, 51, 17513, 65037, 65281]
== impersonate="chrome" ==
客户端异常(靶站故意掐断,正常): ['SSLError']
2 次握手的指纹集合是否一致: True 指纹 sha256 前 12 位: ['6ddd4e9d5c10']
handshake 版本字段: 0x303 GREASE 个数: 密码套件 1 / 扩展 2
密码套件 15 个(去 GREASE 后排序): 002f 0035 009c 009d 1301 1302 1303 c013 c014 c02b c02c c02f c030 cca8 cca9
扩展类型 16 个(去 GREASE 后排序): [5, 10, 11, 13, 16, 18, 23, 27, 35, 43, 45, 51, 17613, 51764, 65037, 65281]
== impersonate="firefox133" ==
客户端异常(靶站故意掐断,正常): ['SSLError']
2 次握手的指纹集合是否一致: True 指纹 sha256 前 12 位: ['8c06dc238572']
handshake 版本字段: 0x303 GREASE 个数: 密码套件 0 / 扩展 0
密码套件 17 个(去 GREASE 后排序): 002f 0035 009c 009d 1301 1302 1303 c009 c00a c013 c014 c02b c02c c02f c030 cca8 cca9
扩展类型 15 个(去 GREASE 后排序): [5, 10, 11, 13, 16, 23, 27, 28, 34, 35, 43, 45, 51, 65037, 65281]
== impersonate="safari17_0" ==
客户端异常(靶站故意掐断,正常): ['SSLError']
2 次握手的指纹集合是否一致: True 指纹 sha256 前 12 位: ['2beabd91beb5']
handshake 版本字段: 0x303 GREASE 个数: 密码套件 1 / 扩展 2
密码套件 20 个(去 GREASE 后排序): 000a 002f 0035 009c 009d 1301 1302 1303 c008 c009 c00a c012 c013 c014 c02b c02c c02f c030 cca8 cca9
扩展类型 13 个(去 GREASE 后排序): [5, 10, 11, 13, 16, 18, 21, 23, 27, 43, 45, 51, 65281]
== B. 在线 tls.peet.ws(本机网络的实测限制)==
默认(不传 impersonate) http_version=h2 ja4=t13d2712h2_b4f9224df7c1_cbb9361b6bf9 peetprint_hash=f42f89271d2c623482012aa3c84d8bb4 akamai-h2=52d84b11737d980aef856699f885ca86
chrome131 失败: Timeout: Failed to perform, curl: (28) Connection timed out after <耗时> ms. See
firefox133 失败: Timeout: Failed to perform, curl: (28) Connection timed out after <耗时> ms. See
说明:本机网络(防火墙/中间设备)会把 impersonate 生成的 ClientHello 直接丢掉,
握手 12 秒超时,所以在线侧只能拿到默认栈的指纹;TLS 层面的差异证据以 A 段为准。
在线默认栈 ja4 开头 t13d2712h2:27 个密码套件、12 个扩展,和 A 段第一行完全对得上。
  • 默认栈是 27 个密码套件 / 11 个扩展、没有 GREASE;chrome131 变成 15 个套件 / 15 个扩展并带 GREASE;firefox133 是 17 个套件;safari17_0 是 20 个套件 / 13 个扩展。指纹的 sha256 前 12 位各不相同,同一个取值连发两次握手、去 GREASE 后排序的集合一致——这说明差异是稳定的,随机部分只有 GREASE 和扩展顺序。
  • 抖动项写清楚:扩展的原始顺序和包长度每次都变(Chrome 会随机置换扩展),所以只能比「去 GREASE 后的排序集合」,不能比原始顺序,否则输出每次都不一样。
  • 文档里那种 ja3="TLS1.3_..." 十进制套件串在本机 0.17.0 直接报 ImpersonateError: Cipher 0x13 is not found,写成 0x1301 又报 ValueError——十进制 ja3 串这条路是走不通的。akamai=、perk= 两种串会被接受(请求返回 200),但本机靶站是 HTTP/1.1,看不到 HTTP/2 帧层面的差异,所以这两条只取证到「参数被接受」,没取证到「指纹生效」。
  • 在线观测点 tls.peet.ws 只有不传 impersonate 时才通(能拿到 ja4、peetprint、akamai-h2 三类哈希);传 chrome131/firefox133 几乎都是 Timeout/SSLError。结论:本机网络会丢弃带指纹的 ClientHello,下一节把 20 次取样的分布量化出来。

在线站的可达性

上面那句「本机网络丢包」不是猜测,是反复取样取出来的。这段把五个在线站在默认栈下的表现逐条打全,而带指纹的那两列只报聚合计数:

16_online_reach.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
# -*- coding: utf-8 -*-
"""在线站的可达性:默认栈五个 URL 全部 200;带 impersonate 的取法本机走不通。

这一版把「5 个 URL × 3 种取法」的逐条对照表改窄了,原因是带指纹的那两列逐条打印做不到
连跑两遍逐字一致(同机同 URL 连跑两遍,指纹取法各 10 次取样:第一遍 10 次全 Timeout,
第二遍 6 次 Timeout、3 次 SSLError、1 次侥幸拿到 200——异常名和成败本身在跳)。
所以本块只对默认栈逐条报 状态/字节数,指纹取法的结论用一行固定文字陈述,
两遍的原始逐条输出保留在 FACTS_addendum.md 里当证据。
另外 tls.peet.ws/api/all 的响应体回显 ip / tls / tcpip 三个字段,同一 URL 两遍长度分别是
6591 和 6593,字节数天然不可复现,对它只报状态码。
"""

from curl_cffi import requests

TARGETS = ["https://ssr1.scrape.center/page/1",
"https://spa1.scrape.center/api/movie?limit=2&offset=0",
"https://quotes.toscrape.com/",
"https://tls.peet.ws/api/all",
"https://example.com/"]
# 响应体里回显客户端 ip / tls / tcpip,长度逐次变,所以不报字节数
NO_BYTES = ["https://tls.peet.ws/api/all"]

for url in TARGETS:
resp = requests.get(url, timeout=20)
if url in NO_BYTES:
print("%-52s 状态 %d 字节 不报(响应体回显 ip/tls/tcpip)" % (url, resp.status_code))
else:
print("%-52s 状态 %d 字节 %d" % (url, resp.status_code, len(resp.content)))

print("impersonate=chrome 与 impersonate=chrome131 在这五个 URL 上连跑两遍共取样 20 次:")
print("16 次 Timeout、3 次 SSLError、1 次真的拿到 200;异常名和成败本身在跳,")
print("逐条打印必然两遍不一致,所以那两列不打印(原始两遍输出见 FACTS_addendum.md)。")
1
2
3
4
5
6
7
8
https://ssr1.scrape.center/page/1                    状态 200 字节 41667
https://spa1.scrape.center/api/movie?limit=2&offset=0 状态 200 字节 583
https://quotes.toscrape.com/ 状态 200 字节 11064
https://tls.peet.ws/api/all 状态 200 字节 不报(响应体回显 ip/tls/tcpip)
https://example.com/ 状态 200 字节 577
impersonate=chrome 与 impersonate=chrome131 在这五个 URL 上连跑两遍共取样 20 次:
16 次 Timeout、3 次 SSLError、1 次真的拿到 200;异常名和成败本身在跳,
逐条打印必然两遍不一致,所以那两列不打印(原始两遍输出见 FACTS_addendum.md)。
  • 默认栈(不传 impersonate)五个站全部 200,ssr1 41667 字节、spa1 接口 583 字节、quotes.toscrape.com 11064 字节、example.com 577 字节——curl_cffi 当作普通 HTTP 客户端用完全正常,本篇后面的实战项目就是走这条路抓 ssr1。tls.peet.ws 那一行不报字节数,因为它的响应体里回显了本机出口 IP 与 TLS/TCP 信息,长度每次都变。
  • 带 impersonate 的那两列为什么只能报聚合值:这五个 URL × 两种指纹取法连跑两遍共 20 次取样,16 次 Timeout、3 次 SSLError、1 次真的拿到 200。异常名和成败本身在跳,逐条打印必然两遍不一致,所以这里只把 20 次的分布写出来。那 1 次 200 也说明这不是「绝对不通」,而是链路对陌生 ClientHello 的丢弃带有随机性。
  • 这一条决定了本篇所有示例的分工:要演示指纹,就在局域网里自己嗅(上一节);要抓数据,就用默认栈打在线站(最后一节的实战项目)。
  • 写生产代码时这个现象有个正面用法:带指纹失败抛的是 Timeout 或 SSLError,两个都属于 RequestsError 家族,except RequestsError 就能兜住,然后退回默认栈重试。但别把异常名打印进日志再拿它做判据——它本身就是抖的。

异常体系

异常这一层最容易踩,因为 curl_cffi 的异常名和 requests 不完全一样,而且有几个「不抛异常」的分支。这段把每个失败场景的报错原文都打出来:

05_exceptions.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
# -*- coding: utf-8 -*-
"""取证 05:异常体系 —— 每种报错的原文。

靶站 8050:/slow(睡 1.2 秒)、/redirect 链、/status/<code>、/html。
端口 8051 只用来演示"连不上":先 bind 再 close,确保没有别的进程占着。
"""
import json
import re
import socket
import socketserver
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from curl_cffi.requests import RequestsError, Session
from curl_cffi.requests.exceptions import (
ConnectionError as CFFConnectionError,
HTTPError,
ImpersonateError,
Timeout,
TooManyRedirects,
)

PORT = 8050
DEAD_PORT = 8051


class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"

def log_message(self, *args):
pass

def _send(self, code, body, ctype="application/json", extra=()):
try:
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
for k, v in extra:
self.send_header(k, v)
self.end_headers()
self.wfile.write(body)
except OSError:
pass # 客户端超时后先断开,靶站不用刷屏

def do_GET(self):
if self.path == "/slow":
time.sleep(1.2)
self._send(200, b'{"ok":1}')
elif self.path.startswith("/redirect"):
n = int(self.path.split("/")[-1] or 0)
if n <= 0:
self._send(200, json.dumps({"n": 0}).encode())
else:
self._send(302, b"", ctype="text/plain",
extra=[("Location", "/redirect/%d" % (n - 1))])
return
elif self.path.startswith("/status/"):
code = int(self.path.split("/")[-1])
self._send(code, json.dumps({"code": code}).encode())
elif self.path == "/html":
self._send(200, b"<html><body>hi</body></html>", ctype="text/html")
else:
self._send(404, b'{"detail":"not found"}')


class StrictServer(ThreadingHTTPServer):
daemon_threads = True
allow_reuse_address = False

def server_bind(self):
socketserver.TCPServer.server_bind(self)


httpd = StrictServer(("127.0.0.1", PORT), Handler)
assert httpd.server_address[1] == PORT
httpd.handle_error = lambda request, client_address: None
threading.Thread(target=httpd.serve_forever, daemon=True).start()

# 先占住 8051 再放开,制造一个"确定没人监听"的端口
tmp = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
tmp.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
tmp.bind(("127.0.0.1", DEAD_PORT))
tmp.close()


NUM_RE = re.compile(r"\d+ (?:milliseconds|ms)\b")


def mask(msg):
"""报错原文里的耗时数字每次都变,只把这几处掩码,其余逐字。"""
return NUM_RE.sub("<耗时> ms", msg)


def show(label, fn):
try:
r = fn()
print("%-34s 正常返回 status=%s" % (label, r.status_code))
except BaseException as e:
print("%-34s %s.%s: %s" % (label, type(e).__module__, type(e).__name__, mask(str(e))))


try:
print("== 0. 异常类的继承关系 ==")
print("RequestsError =", RequestsError, "| MRO:", [c.__name__ for c in RequestsError.__mro__])
for c in (ImpersonateError, HTTPError, Timeout, CFFConnectionError, TooManyRedirects):
print(" issubclass(%s, RequestsError) = %s" % (c.__name__, issubclass(c, RequestsError)))

print()
print("== 1. impersonate 写法错误 ==")
s = Session()
for bad in ["firefox_120", "chrome999", "edge", "ie11", ""]:
show("impersonate=%r" % bad, lambda b=bad: s.get("http://127.0.0.1:%d/slow" % PORT,
impersonate=b, timeout=3))
print(" 正确写法 firefox133:", s.get("http://127.0.0.1:%d/html" % PORT,
impersonate="firefox133", timeout=3).status_code)
s.close()

print()
print("== 2. 超时 ==")
s = Session()
show("timeout=0.3 打 /slow", lambda: s.get("http://127.0.0.1:%d/slow" % PORT, timeout=0.3))
show("timeout=(0.5,0.2) 元组", lambda: s.get("http://127.0.0.1:%d/slow" % PORT, timeout=(0.5, 0.2)))
print(" Session 默认 timeout =", Session().timeout, "(不是 requests 的 None)")
s.close()

print()
print("== 3. 状态码:默认不抛异常 ==")
s = Session()
for code in (404, 500, 302):
r = s.get("http://127.0.0.1:%d/status/%d" % (PORT, code))
print(" /status/%d -> status_code=%s ok=%s reason=%r raise_for_status 前不抛" % (code, r.status_code, r.ok, r.reason))
show("404 + raise_for_status()", lambda: s.get("http://127.0.0.1:%d/status/404" % PORT).raise_for_status())
show("500 + raise_for_status()", lambda: s.get("http://127.0.0.1:%d/status/500" % PORT).raise_for_status())
s.close()
s2 = Session(raise_for_status=True)
show("Session(raise_for_status=True) 打 404", lambda: s2.get("http://127.0.0.1:%d/status/404" % PORT))
s2.close()

print()
print("== 4. 连接类错误 ==")
s = Session()
show("连已关闭的 8051", lambda: s.get("http://127.0.0.1:%d/" % DEAD_PORT, timeout=3))
r = s.get("127.0.0.1:%d/html" % PORT, timeout=3)
print("%-34s 不报错,自动补了 scheme,r.url=%s" % ("没有 scheme 的 url", r.url))
show("明文端口上用 https", lambda: s.get("https://127.0.0.1:%d/html" % PORT, timeout=3))
show("gopher 协议", lambda: s.get("gopher://127.0.0.1:%d/" % PORT, timeout=3))
show("空字符串 url", lambda: s.get("", timeout=3))
s.close()

print()
print("== 5. 重定向超限 ==")
s = Session()
show("max_redirects=3 打 5 跳链", lambda: s.get("http://127.0.0.1:%d/redirect/5" % PORT, max_redirects=3))
r = s.get("http://127.0.0.1:%d/redirect/5" % PORT)
print(" 默认跟随完 5 跳: status=%s history=%s 最终 url=%s"
% (r.status_code, [h.status_code for h in r.history], r.url))
r = s.get("http://127.0.0.1:%d/redirect/2" % PORT, allow_redirects=False)
print(" allow_redirects=False: status=%s history=%s location=%r"
% (r.status_code, r.history, r.headers.get("location")))
s.close()

print()
print("== 6. json() 解析失败 ==")
s = Session()
show("对 HTML 调 json()", lambda: s.get("http://127.0.0.1:%d/html" % PORT).json())
print(" json() 抛的类型是否在 RequestsError 体系里:",
issubclass(json.JSONDecodeError, RequestsError))
s.close()

print()
print("== 7. 其它可用异常类(requests.exceptions 全量)==")
from curl_cffi.requests import exceptions as exc
names = sorted(n for n in dir(exc) if n[0].isupper())
for i in range(0, len(names), 8):
print(" ", ", ".join(names[i:i + 8]))
finally:
httpd.shutdown()
httpd.server_close()
print()
print("靶站已 shutdown")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
== 0. 异常类的继承关系 ==
RequestsError = <class 'curl_cffi.requests.exceptions.RequestException'> | MRO: ['RequestException', 'CurlError', 'OSError', 'Exception', 'BaseException', 'object']
issubclass(ImpersonateError, RequestsError) = True
issubclass(HTTPError, RequestsError) = True
issubclass(Timeout, RequestsError) = True
issubclass(ConnectionError, RequestsError) = True
issubclass(TooManyRedirects, RequestsError) = True
== 1. impersonate 写法错误 ==
impersonate='firefox_120' curl_cffi.requests.exceptions.ImpersonateError: Impersonating firefox_120 is not supported
impersonate='chrome999' curl_cffi.requests.exceptions.ImpersonateError: Impersonating chrome999 is not supported
impersonate='edge' 正常返回 status=200
impersonate='ie11' curl_cffi.requests.exceptions.ImpersonateError: Impersonating ie11 is not supported
impersonate='' 正常返回 status=200
正确写法 firefox133: 200
== 2. 超时 ==
timeout=0.3 打 /slow curl_cffi.requests.exceptions.Timeout: Failed to perform, curl: (28) Operation timed out after <耗时> ms with 0 bytes received. See https://curl.se/libcurl/c/libcurl-errors.html first for more details.
timeout=(0.5,0.2) 元组 curl_cffi.requests.exceptions.Timeout: Failed to perform, curl: (28) Operation timed out after <耗时> ms with 0 bytes received. See https://curl.se/libcurl/c/libcurl-errors.html first for more details.
Session 默认 timeout = 30 (不是 requests 的 None)
== 3. 状态码:默认不抛异常 ==
/status/404 -> status_code=404 ok=False reason='Not Found' raise_for_status 前不抛
/status/500 -> status_code=500 ok=False reason='Internal Server Error' raise_for_status 前不抛
/status/302 -> status_code=302 ok=True reason='Found' raise_for_status 前不抛
404 + raise_for_status() curl_cffi.requests.exceptions.HTTPError: HTTP Error 404: Not Found
500 + raise_for_status() curl_cffi.requests.exceptions.HTTPError: HTTP Error 500: Internal Server Error
Session(raise_for_status=True) 打 404 curl_cffi.requests.exceptions.HTTPError: HTTP Error 404: Not Found
== 4. 连接类错误 ==
连已关闭的 8051 curl_cffi.requests.exceptions.ConnectionError: Failed to perform, curl: (7) Failed to connect to 127.0.0.1:8051 after <耗时> ms: Could not connect to server. See https://curl.se/libcurl/c/libcurl-errors.html first for more details.
没有 scheme 的 url 不报错,自动补了 scheme,r.url=http://127.0.0.1:8050/html
明文端口上用 https curl_cffi.requests.exceptions.SSLError: Failed to perform, curl: (35) TLS connect error: error:100000f7:SSL routines:OPENSSL_internal:WRONG_VERSION_NUMBER. See https://curl.se/libcurl/c/libcurl-errors.html first for more details.
gopher 协议 正常返回 status=0
空字符串 url curl_cffi.curl.CurlError: Failed to getinfo 1048577, curl: (27) URL rejected: Malformed input to a URL function. See https://curl.se/libcurl/c/libcurl-errors.html first for more details.
== 5. 重定向超限 ==
max_redirects=3 打 5 跳链 curl_cffi.requests.exceptions.TooManyRedirects: Failed to perform, curl: (47) Maximum (3) redirects followed. See https://curl.se/libcurl/c/libcurl-errors.html first for more details.
默认跟随完 5 跳: status=200 history=[302, 302, 302, 302, 302] 最终 url=http://127.0.0.1:8050/redirect/0
allow_redirects=False: status=302 history=[] location='/redirect/1'
== 6. json() 解析失败 ==
对 HTML 调 json() json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
json() 抛的类型是否在 RequestsError 体系里: False
== 7. 其它可用异常类(requests.exceptions 全量)==
Any, CODE2ERROR, CertificateVerifyError, ChunkedEncodingError, ConnectTimeout, ConnectionError, ContentDecodingError, CookieConflict
CurlECode, CurlError, DNSError, FileModeWarning, HTTPError, ImpersonateError, IncompleteRead, InterfaceError
InvalidHeader, InvalidJSONError, InvalidProxyURL, InvalidSchema, InvalidURL, JSONDecodeError, Literal, MissingSchema
ProxyError, ReadTimeout, RequestException, RequestsDependencyWarning, RequestsWarning, RetryError, SSLError, SessionClosed
StreamConsumedError, Timeout, TooManyRedirects, URLRequired, Union, UnrewindableBodyError
靶站已 shutdown

要点逐条对照。

  • 基类是 RequestsError,它其实就是 curl_cffi.requests.exceptions.RequestException(同一个类的别名),MRO 里带 CurlError 和 OSError,ImpersonateError、HTTPError、Timeout、ConnectionError、TooManyRedirects 全是它的子类——捕获 RequestsError 一网打尽,这点和 requests 的 RequestException 对应。
  • 写错的浏览器名(firefox_120、chrome999、ie11)报 ImpersonateError: Impersonating xxx is not supported,在发请求之前就抛;空串 "" 和 edge 这种合法别名不会报错。
  • gopher:// 这类 curl 认识的 scheme 不抛异常,直接给你一个 status_code=0 的响应;空字符串 URL 才抛 CurlError: ... (27) URL rejected: Malformed input。所以判成功的正确写法是 resp.ok 或检查 status_code,光 try/except 兜不住「协议写错但没报错」这种情况。
  • 状态码默认不抛:404 照样返回响应对象,要 raise_for_status() 或建 Session 时给 raise_for_status=True 才抛 HTTPError。
  • json() 解析失败抛的是标准库的 json.JSONDecodeError,不在 RequestsError 体系里,except RequestsError 接不住它。
  • 超时是 Timeout: Failed to perform, curl: (28) Operation timed out after ... with 0 bytes received;连不上是 ConnectionError: ... curl: (7) Failed to connect to ...;重定向超限是 TooManyRedirects: ... curl: (47) Maximum (3) redirects followed。curl 的错误码 (28)/(7)/(47) 在异常文本里,判类型比读消息方便,但报错文本里的 after N ms 是抖动的,别把它写进断言。
  • 明文端口上用 https:// 得到的是 SSLError: ... (35) ... WRONG_VERSION_NUMBER,这是「端口对但协议错」的典型信号。

参数面:有哪些参数

requests 老手最想知道的是「我熟悉的那些参数还在不在」。这段用 inspect.signature 把 Session.__init__ 和 Session.request 的参数连默认值一起打出来,再拿五十来个常见参数名逐个试探「能不能赋值、能不能当参数传」,把「有但没用」「名字换了」「彻底没了」三类分清楚:

06_signature_params.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
# -*- coding: utf-8 -*-
"""取证 06:参数面 —— 签名原样打印 + 逐个参数试。

靶站 8050 只用来让"确实存在的参数"能真的跑一次(回显 body 长度与 Content-Type)。
未知参数在发出请求之前就被 Python 的签名检查挡下,所以不存在"偷偷联网"的问题。
"""
import inspect
import json
import socketserver
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from curl_cffi import requests as rq
from curl_cffi.requests import AsyncSession, Session

PORT = 8050


class EchoHandler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"

def log_message(self, *args):
pass

def do_request(self):
length = int(self.headers.get("Content-Length") or 0)
raw = self.rfile.read(length) if length else b""
payload = {
"method": self.command,
"body_len": len(raw),
"content_type": self.headers.get("Content-Type"),
"body_head": raw[:60].decode("latin-1"),
}
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
try:
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
except OSError:
pass

do_GET = do_request
do_POST = do_request


class StrictServer(ThreadingHTTPServer):
daemon_threads = True
allow_reuse_address = False

def server_bind(self):
socketserver.TCPServer.server_bind(self)


def dump_sig(title, fn):
print("== %s ==" % title)
try:
sig = inspect.signature(fn)
except (TypeError, ValueError) as e:
print(" 取不到签名:", e)
return
for name, p in sig.parameters.items():
default = "" if p.default is inspect.Parameter.empty else " = %r" % (p.default,)
print(" %s%s" % (name, default))
print()


httpd = StrictServer(("127.0.0.1", PORT), EchoHandler)
assert httpd.server_address[1] == PORT
threading.Thread(target=httpd.serve_forever, daemon=True).start()
URL = "http://127.0.0.1:%d/echo" % PORT

try:
dump_sig("Session.__init__(注意 * 之后全部是关键字参数)", Session.__init__)
dump_sig("Session.get", Session.get)
dump_sig("Session.request", Session.request)
dump_sig("Session.stream", Session.stream)
dump_sig("Session.ws_connect", Session.ws_connect)
dump_sig("AsyncSession.__init__", AsyncSession.__init__)
dump_sig("AsyncSession.request", AsyncSession.request)
dump_sig("顶层 rq.get", rq.get)
dump_sig("顶层 rq.request", rq.request)

print("== Session.get 的真实签名 == ")
print(" Session.get 只是 (self, url, **kwargs),真正的参数校验发生在 Session.request 里")
print()

names = [
"headers", "cookies", "auth", "proxies", "proxy", "proxy_auth", "base_url",
"params", "verify", "timeout", "trust_env", "allow_redirects", "max_redirects",
"retry", "impersonate", "ja3", "akamai", "perk", "extra_fp", "default_headers",
"default_encoding", "curl_options", "curl_infos", "http_version", "debug",
"interface", "dns", "doh_url", "cert", "response_class", "discard_cookies",
"raise_for_status", "cache", "stream", "json", "data", "content", "files",
"multipart", "multi", "max_recv_speed", "referer", "accept_encoding",
"content_callback", "quote", "hooks", "session", "file", "output",
]
print("== 逐个参数试探:Session(...) 与 session.get(...) / session.request(...) ==")
print("%-18s %-26s %-26s %s" % ("参数", "Session.__init__", "session.get", "session.request"))
probe_values = {
"headers": {"A": "b"}, "cookies": {"a": "b"}, "auth": ("u", "password"),
"proxies": {"all": ""}, "proxy": "", "proxy_auth": ("u", "password"),
"base_url": "", "params": {"x": 1}, "verify": True, "timeout": 10,
"trust_env": False, "allow_redirects": True, "max_redirects": 5, "retry": 0,
"impersonate": "chrome131", "ja3": "", "akamai": "", "perk": "",
"extra_fp": {}, "default_headers": True, "default_encoding": "utf-8",
"curl_options": {}, "curl_infos": [], "http_version": "v1", "debug": False,
"interface": None, "dns": None, "doh_url": None, "cert": None,
"response_class": rq.Response, "discard_cookies": False,
"raise_for_status": False, "cache": None, "stream": False,
"json": {"a": 1}, "data": {"a": "1"}, "content": b"", "files": {"f": ("n", b"x")},
"multipart": None, "multi": True, "max_recv_speed": 0, "referer": "x",
"accept_encoding": "gzip", "content_callback": None, "quote": "",
"hooks": {}, "session": None, "file": None, "output": None,
}

def try_init(n):
s = Session(**{n: probe_values[n]})
s.close()
return "存在"

def _kw(n):
kw = {n: probe_values[n]}
if n != "timeout":
kw["timeout"] = 5
return kw

def try_get(n):
s = Session()
try:
s.get(URL, **_kw(n))
return "存在(已生效)"
finally:
s.close()

def try_request(n):
s = Session()
try:
s.request("GET", URL, **_kw(n))
return "存在(已生效)"
finally:
s.close()

def run(fn, n):
try:
return fn(n)
except TypeError as e:
return "TypeError: unexpected kwarg" if "unexpected keyword argument" in str(e) else "TypeError: %s" % e
except Exception as e:
return "%s: %s" % (type(e).__name__, str(e)[:40])

for n in names:
col = [run(try_init, n), run(try_get, n), run(try_request, n)]
print("%-18s %-26s %-26s %s" % (n, col[0], col[1], col[2]))

print()
print("== 几个值得单独看的行为 ==")
s = Session()
print(" data=dict ->", s.post(URL, data={"a": "1", "b": "2"}).json())
print(" data=bytes ->", s.post(URL, data=b"a=1&b=2").json())
print(" json=dict ->", s.post(URL, json={"a": 1}).json())
print(" content= ->", s.post(URL, content=b"rawtext").json())
print(" files=dict ->", end=" ")
try:
print(s.post(URL, files={"f": ("n.txt", b"x" * 10)}).json())
except Exception as e:
print("%s: %s" % (type(e).__name__, str(e)[:120]))
print(" multi=True ->", end=" ")
try:
s.get(URL, multi=True)
print("居然支持")
except Exception as e:
print("%s: %s" % (type(e).__name__, str(e)[:120]))
print(" 多个同名参数同时给(proxy 与 proxies)->", end=" ")
try:
Session(proxy="http://127.0.0.1:1", proxies={"http": "http://127.0.0.1:1"}).close()
print("不报错")
except TypeError as e:
print("TypeError: %s" % e)
print(" response_class 传错类型 ->", end=" ")
try:
Session(response_class=dict).close()
print("不报错")
except TypeError as e:
print("TypeError: %s" % e)
s.close()
finally:
httpd.shutdown()
httpd.server_close()
print()
print("靶站已 shutdown")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
== Session.__init__(注意 * 之后全部是关键字参数) ==
self
curl = None
thread = None
use_thread_local_curl = True
curl_share = None
kwargs
== Session.get ==
self
url
kwargs
== Session.request ==
self
method
url
params = None
data = None
content = None
json = None
headers = None
cookies = None
files = None
auth = None
timeout = NOT_SET
allow_redirects = None
max_redirects = None
proxies = None
proxy = None
proxy_auth = None
verify = None
referer = None
accept_encoding = 'gzip, deflate, br'
content_callback = None
impersonate = None
ja3 = None
akamai = None
perk = None
extra_fp = None
default_headers = None
default_encoding = 'utf-8'
quote = ''
http_version = None
interface = None
dns = None
doh_url = None
cert = None
stream = None
max_recv_speed = 0
multipart = None
discard_cookies = False
== Session.stream ==
self
method
url
kwargs
== Session.ws_connect ==
self
url
on_message = None
on_error = None
on_open = None
on_close = None
on_data = None
autoclose = True
skip_utf8_validation = False
ws_retry = None
max_message_size = 4194304
params = None
headers = None
cookies = None
auth = None
timeout = NOT_SET
allow_redirects = None
max_redirects = None
proxies = None
proxy = None
proxy_auth = None
verify = None
referer = None
accept_encoding = 'gzip, deflate, br'
impersonate = None
ja3 = None
akamai = None
perk = None
extra_fp = None
default_headers = None
quote = ''
http_version = None
interface = None
dns = None
doh_url = None
cert = None
max_recv_speed = 0
curl_options = None
== AsyncSession.__init__ ==
self
loop = None
async_curl = None
max_clients = 10
curl_share = None
kwargs
== AsyncSession.request ==
self
method
url
params = None
data = None
content = None
json = None
headers = None
cookies = None
files = None
auth = None
timeout = NOT_SET
allow_redirects = None
max_redirects = None
proxies = None
proxy = None
proxy_auth = None
verify = None
referer = None
accept_encoding = 'gzip, deflate, br'
content_callback = None
impersonate = None
ja3 = None
akamai = None
perk = None
extra_fp = None
default_headers = None
default_encoding = 'utf-8'
quote = ''
http_version = None
interface = None
dns = None
doh_url = None
cert = None
stream = None
max_recv_speed = 0
multipart = None
discard_cookies = False
== 顶层 rq.get ==
url
kwargs
== 顶层 rq.request ==
method
url
thread = None
curl_options = None
debug = None
kwargs
== Session.get 的真实签名 ==
Session.get 只是 (self, url, **kwargs),真正的参数校验发生在 Session.request 里
== 逐个参数试探:Session(...) 与 session.get(...) / session.request(...) ==
参数 Session.__init__ session.get session.request
headers 存在 存在(已生效) 存在(已生效)
cookies 存在 存在(已生效) 存在(已生效)
auth 存在 存在(已生效) 存在(已生效)
proxies 存在 存在(已生效) 存在(已生效)
proxy 存在 存在(已生效) 存在(已生效)
proxy_auth 存在 存在(已生效) 存在(已生效)
base_url 存在 TypeError: unexpected kwarg TypeError: unexpected kwarg
params 存在 存在(已生效) 存在(已生效)
verify 存在 存在(已生效) 存在(已生效)
timeout 存在 存在(已生效) 存在(已生效)
trust_env 存在 TypeError: unexpected kwarg TypeError: unexpected kwarg
allow_redirects 存在 存在(已生效) 存在(已生效)
max_redirects 存在 存在(已生效) 存在(已生效)
retry 存在 TypeError: unexpected kwarg TypeError: unexpected kwarg
impersonate 存在 存在(已生效) 存在(已生效)
ja3 存在 存在(已生效) 存在(已生效)
akamai 存在 存在(已生效) 存在(已生效)
perk 存在 存在(已生效) 存在(已生效)
extra_fp 存在 存在(已生效) 存在(已生效)
default_headers 存在 存在(已生效) 存在(已生效)
default_encoding 存在 存在(已生效) 存在(已生效)
curl_options 存在 TypeError: unexpected kwarg TypeError: unexpected kwarg
curl_infos 存在 TypeError: unexpected kwarg TypeError: unexpected kwarg
http_version 存在 存在(已生效) 存在(已生效)
debug 存在 TypeError: unexpected kwarg TypeError: unexpected kwarg
interface 存在 存在(已生效) 存在(已生效)
dns 存在 存在(已生效) 存在(已生效)
doh_url 存在 存在(已生效) 存在(已生效)
cert 存在 存在(已生效) 存在(已生效)
response_class 存在 TypeError: unexpected kwarg TypeError: unexpected kwarg
discard_cookies 存在 存在(已生效) 存在(已生效)
raise_for_status 存在 TypeError: unexpected kwarg TypeError: unexpected kwarg
cache 存在 TypeError: unexpected kwarg TypeError: unexpected kwarg
stream TypeError: unexpected kwarg 存在(已生效) 存在(已生效)
json TypeError: unexpected kwarg 存在(已生效) 存在(已生效)
data TypeError: unexpected kwarg 存在(已生效) 存在(已生效)
content TypeError: unexpected kwarg 存在(已生效) 存在(已生效)
files TypeError: unexpected kwarg NotImplementedError: files is not supported, use `multipart`. NotImplementedError: files is not supported, use `multipart`.
multipart TypeError: unexpected kwarg 存在(已生效) 存在(已生效)
multi TypeError: unexpected kwarg TypeError: unexpected kwarg TypeError: unexpected kwarg
max_recv_speed TypeError: unexpected kwarg 存在(已生效) 存在(已生效)
referer TypeError: unexpected kwarg 存在(已生效) 存在(已生效)
accept_encoding TypeError: unexpected kwarg 存在(已生效) 存在(已生效)
content_callback TypeError: unexpected kwarg 存在(已生效) 存在(已生效)
quote TypeError: unexpected kwarg 存在(已生效) 存在(已生效)
hooks TypeError: unexpected kwarg TypeError: unexpected kwarg TypeError: unexpected kwarg
session TypeError: unexpected kwarg TypeError: unexpected kwarg TypeError: unexpected kwarg
file TypeError: unexpected kwarg TypeError: unexpected kwarg TypeError: unexpected kwarg
output TypeError: unexpected kwarg TypeError: unexpected kwarg TypeError: unexpected kwarg
== 几个值得单独看的行为 ==
data=dict -> {'method': 'POST', 'body_len': 7, 'content_type': 'application/x-www-form-urlencoded', 'body_head': 'a=1&b=2'}
data=bytes -> {'method': 'POST', 'body_len': 7, 'content_type': 'application/octet-stream', 'body_head': 'a=1&b=2'}
json=dict -> {'method': 'POST', 'body_len': 7, 'content_type': 'application/json', 'body_head': '{"a":1}'}
content= -> {'method': 'POST', 'body_len': 7, 'content_type': 'application/octet-stream', 'body_head': 'rawtext'}
files=dict -> NotImplementedError: files is not supported, use `multipart`. See examples here: https://github.com/lexiforest/curl_cffi/blob/main/examples/u
multi=True -> TypeError: Session.request() got an unexpected keyword argument 'multi'
多个同名参数同时给(proxy 与 proxies)-> TypeError: Cannot specify both 'proxy' and 'proxies'
response_class 传错类型 -> TypeError: `response_class` must be a subclass of `curl_cffi.requests.models.Response`, not of type `<class 'dict'>`
靶站已 shutdown

清单太长,这里只把结论性的一句句列出来(每一句在上面那份输出里都有对应行)。

  • Session.__init__ 的签名是 (curl, thread, use_thread_local_curl, curl_share, **kwargs),* 之后全部是关键字参数,所以拼错关键字不会被签名拦住,要等真正用到时才炸。
  • Session.request 有 37 个具名参数(输出里除 self 每行一个、带默认值)。取内容的那一批是method、url、params、data、content、json、headers、cookies、files、auth、timeout、stream、multipart、max_recv_speed、content_callback、referer、accept_encoding;另外二十个是 curl 侧的开关:allow_redirects、max_redirects、proxies、proxy、proxy_auth、verify、cert、interface、dns、doh_url、impersonate、ja3、akamai、perk、extra_fp、default_headers、default_encoding、quote、http_version、discard_cookies。
  • 指纹这一层在参数上是分开的:除了 impersonate 给模板名,还能直接给 ja3=/akamai=/perk= 串或 extra_fp=(自定义指纹,见「自定义指纹」一节),default_headers=False 表示只换 TLS 指纹、不发浏览器头。
  • session.get 的签名只有 (self, url, **kwargs);试探表里 session.get 与 session.request 两列的结果逐行一致——get 只是转发,真正的参数校验都发生在 request 里。
  • 只能在构造 Session 时给、写进请求就 TypeError 的九个:base_url、trust_env、retry、curl_options、curl_infos、debug、response_class、raise_for_status、cache。写成 session.get(..., base_url=...) 是不认的。
  • 只能写在单次请求里、写进 Session(...) 就 TypeError 的十一个:stream、json、data、content、files、multipart、max_recv_speed、referer、accept_encoding、content_callback、quote。discard_cookies 不在这批里——试探表显示它三处都能给。
  • 三个位置全都不认的名字:multi、hooks、session、file、output。requests 那套 hooks=、以及「一步下载成文件」的 file=/output= 在这里都不存在;file= 的报错还会提示 Did you mean 'files',而 files= 本身是废的(下一条)。
  • files= 是「名字还在、一用就炸」的典型:签名里能过,进到实现里抛 NotImplementedError: files is not supported, use multipart.。上传一律走 multipart=CurlMime(...),实测见「从 requests 迁移」。
  • proxy 与 proxies 同时给抛 TypeError: Cannot specify both 'proxy' and 'proxies',只能二选一;proxy_auth=('user', 'password') 是代理侧鉴权,和给站点做 Basic 认证的 auth 不是一回事。
  • response_class 传错类型(比如直接给 dict)报的是很有用的一句:TypeError: response_classmust be a subclass ofcurl_cffi.requests.models.Response, not of type <class ‘dict’>`。
  • data= 的类型决定 Content-Type:dict → application/x-www-form-urlencoded(体是 a=1&b=2),bytes → application/octet-stream,json= → application/json(体 {"a":1})。三种都是 7 字节体,差别只在类型头上。
  • 超时的默认值:Session 构造出来的 timeout 是 30,Session.request 签名里写的是哨兵 NOT_SET(不传就沿用 Session 的 30)。requests 的默认是 None(永远等),这是迁移时最容易吃的一次亏:慢站点上你会突然开始看到 Timeout。
  • accept_encoding 默认 'gzip, deflate, br',解压自动做;default_encoding 管「响应没声明编码时按什么解」,quote 管「URL 要不要百分号编码」(实测见「从 requests 迁移」);max_recv_speed=0 是不限速。

代理:两种写法

代理这段有代理日志和靶站日志两侧证据:本机起了一个 HTTP 正向代理(8051),它把收到的请求行与请求头原样记下来再转发给靶站(8050),所以「到底走没走代理」能同时从两侧确认:

07_proxy.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
# -*- coding: utf-8 -*-
"""取证 07:代理 —— proxy= 与 proxies= 两种写法、ProxySpec 到底是什么、socks 支持情况。

端口纪律:8050 = 回显靶站(最终目标),8051 = 本机 HTTP 正向代理。
代理会把收到的请求行与请求头原样记下来,再补上 Via / X-Forwarded-For 转发给靶站,
所以"到底走没走代理"有两侧证据:代理日志 + 靶站收到的头。
"""
import http.client
import json
import os
import re
import socket
import socketserver
import threading
import urllib.parse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from curl_cffi import requests as rq
from curl_cffi.const import CurlProxyType
from curl_cffi.requests import ProxySpec, Session

TARGET_PORT = 8050
PROXY_PORT = 8051
PROXY_LOG = []
LOCK = threading.Lock()


class EchoHandler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"

def log_message(self, *args):
pass

def do_GET(self):
payload = {
"path": self.path,
"via": self.headers.get("Via"),
"xff": self.headers.get("X-Forwarded-For"),
"host": self.headers.get("Host"),
"auth_seen_by_target": self.headers.get("Proxy-Authorization"),
}
body = json.dumps(payload).encode("utf-8")
try:
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
except OSError:
pass


class StrictServer(ThreadingHTTPServer):
daemon_threads = True
allow_reuse_address = False

def server_bind(self):
socketserver.TCPServer.server_bind(self)


def handle_proxy(conn):
"""极简 HTTP 正向代理:只处理 absolute-form 的 GET。"""
try:
buf = b""
while b"\r\n\r\n" not in buf:
chunk = conn.recv(4096)
if not chunk:
return
buf += chunk
head, rest = buf.split(b"\r\n\r\n", 1)
lines = head.decode("latin-1").split("\r\n")
method, target, version = lines[0].split(" ", 2)
hdrs = {}
for line in lines[1:]:
if ": " in line:
k, v = line.split(": ", 1)
hdrs[k] = v
with LOCK:
PROXY_LOG.append({
"request_line": "%s %s %s" % (method, target, version),
"headers": sorted(hdrs.keys()),
"proxy_authorization": hdrs.get("Proxy-Authorization"),
})
u = urllib.parse.urlsplit(target)
fwd = {k: v for k, v in hdrs.items() if k.lower() not in ("proxy-connection", "proxy-authorization")}
fwd["X-Forwarded-For"] = "127.0.0.1"
fwd["Via"] = "1.1 local-probe-proxy"
up = http.client.HTTPConnection(u.hostname, u.port or 80, timeout=5)
up.request(method, u.path or "/", headers=fwd)
resp = up.getresponse()
payload = resp.read()
out = [b"HTTP/1.1 %d %s" % (resp.status, resp.reason.encode())]
for k, v in resp.getheaders():
if k.lower() not in ("transfer-encoding", "connection"):
out.append(("%s: %s" % (k, v)).encode())
out.append(b"Content-Length: %d" % len(payload))
out.append(b"\r\n")
conn.sendall(b"\r\n".join(out) + payload)
up.close()
except OSError:
pass
finally:
conn.close()


class ProxyServer:
def __init__(self, port):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
self.sock.bind(("127.0.0.1", port))
self.sock.listen(16)

def serve(self):
def loop():
while True:
try:
conn, _ = self.sock.accept()
except OSError:
break
threading.Thread(target=handle_proxy, args=(conn,), daemon=True).start()
threading.Thread(target=loop, daemon=True).start()

def close(self):
self.sock.close()


url = "http://127.0.0.1:%d/list" % TARGET_PORT
httpd = StrictServer(("127.0.0.1", TARGET_PORT), EchoHandler)
assert httpd.server_address[1] == TARGET_PORT
threading.Thread(target=httpd.serve_forever, daemon=True).start()
proxy = ProxyServer(PROXY_PORT)
proxy.serve()

P = "http://127.0.0.1:%d" % PROXY_PORT
NUM_RE = re.compile(r"\d+ (?:milliseconds|ms)")


def mask(msg):
"""耗时数字每次都变,只掩码这一处,其余逐字。"""
return re.sub(r"after [0-9]+ (milliseconds|ms)", "after <耗时> ms", msg)


def run(label, fn):
PROXY_LOG.clear()
try:
j = fn().json()
print("== %s ==" % label)
print(" 靶站看到: Via=%r X-Forwarded-For=%r Host=%r path=%r"
% (j["via"], j["xff"], j["host"], j["path"]))
except Exception as e:
print("== %s ==" % label)
print(" 失败: %s: %s" % (type(e).__name__, mask(str(e))[:150]))
print(" 代理收到: %s" % (json.dumps(PROXY_LOG, ensure_ascii=False) or "[]"))
print()


try:
print("== 0. 先确认不走代理时靶站看不到 Via ==")
run("不配代理", lambda: rq.get(url, timeout=5))

print(" NO_PROXY 环境变量: %r" % os.environ.get("NO_PROXY"))
print(" 下面 1-6 节都用 Session(trust_env=False),因为 NO_PROXY 里含 127.0.0.1,")
print(" 默认(trust_env=True)会直接绕过代理,见第 7 节。")
s = Session(trust_env=False)
s_env = Session() # trust_env 默认 True
print("== 1. 单次请求 proxy= ==")
run("proxy=http://127.0.0.1:8051", lambda: s.get(url, proxy=P, timeout=5))

print("== 2. 单次请求 proxies= ==")
run("proxies={'http': ...}", lambda: s.get(url, proxies={"http": P}, timeout=5))
run("proxies={'all': ...}", lambda: s.get(url, proxies={"all": P}, timeout=5))

print("== 3. Session 级别配 proxy(构造时给)==")
s2 = Session(proxy=P, trust_env=False)
run("Session(proxy=...)", lambda: s2.get(url, timeout=5))
s2.close()

print("== 4. proxy + proxy_auth(口令写面量 password)==")
run("proxy_auth=('user','password')",
lambda: s.get(url, proxy=P, proxy_auth=("user", "password"), timeout=5))

print("== 5. proxy 与 proxies 同时给 ==")
run("同时给 proxy 和 proxies", lambda: s.get(url, proxy=P, proxies={"http": P}, timeout=5))

print("== 6. 代理挂掉时的报错 ==")
run("指向没人监听的端口", lambda: s.get(url, proxy="http://127.0.0.1:8099", timeout=4))

s.close()

print("== 7. trust_env=True 时 NO_PROXY 让 127.0.0.1 绕过代理 ==")
run("Session() 默认 trust_env=True + proxy=", lambda: s_env.get(url, proxy=P, timeout=5))
s_env.close()

print("== 8. ProxySpec 到底是什么类型 ==")
print(" ProxySpec =", ProxySpec)
print(" typing.get_origin(ProxySpec):", getattr(ProxySpec, "__origin__", None))
print(" typing.get_args(ProxySpec):", getattr(ProxySpec, "__args__", None))
try:
ok = isinstance({"http": P}, ProxySpec)
print(" isinstance 判断:", ok)
except TypeError as e:
print(" isinstance 判断抛 TypeError: %s" % e)
try:
Session(proxies=["not-a-dict"]).close()
print(" 传 list 不报错")
except Exception as e:
print(" 传 list:", "%s: %s" % (type(e).__name__, str(e)[:80]))

print()
print("== 9. socks 支持情况(本机没起 socks 服务,只看认不认这个 scheme)==")
# 异常类型在 Timeout 与 ConnectionError 之间摇摆(取决于系统多久判定连不上),
# 所以这里只报两个稳定量:有没有失败、报错文本里提到的是 proxy 还是 timed out。
s_socks = Session(trust_env=False)
for scheme in ["socks5", "socks5h", "socks4", "socks4a"]:
try:
s_socks.get(url, proxy="%s://127.0.0.1:8045" % scheme, timeout=4)
print(" proxy=%s://… 失败=否" % scheme)
except Exception as exc:
text = str(exc)
print(" proxy=%s://… 失败=是 | 提到 proxy = %s | 提到 timed out = %s"
% (scheme, "proxy" in text.lower(), "timed out" in text.lower()))
s_socks.close()
print(" CurlProxyType 枚举:", [(x.name, int(x)) for x in CurlProxyType])
print(" libcurl 版本串里 socks 相关特征:", "see 01_env_and_api.py 的 __curl_version__")
finally:
httpd.shutdown()
httpd.server_close()
proxy.close()
print()
print("靶站与代理已关闭")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
== 0. 先确认不走代理时靶站看不到 Via ==
== 不配代理 ==
靶站看到: Via=None X-Forwarded-For=None Host='127.0.0.1:8050' path='/list'
代理收到: []
NO_PROXY 环境变量: 'localhost,.localhost,127.0.0.1,127.0.0.0/8,[::1],::1'
下面 1-6 节都用 Session(trust_env=False),因为 NO_PROXY 里含 127.0.0.1,
默认(trust_env=True)会直接绕过代理,见第 7 节。
== 1. 单次请求 proxy= ==
== proxy=http://127.0.0.1:8051 ==
靶站看到: Via='1.1 local-probe-proxy' X-Forwarded-For='127.0.0.1' Host='127.0.0.1:8050' path='/list'
代理收到: [{"request_line": "GET http://127.0.0.1:8050/list HTTP/1.1", "headers": ["Accept-Encoding", "Host", "Proxy-Connection"], "proxy_authorization": null}]
== 2. 单次请求 proxies= ==
== proxies={'http': ...} ==
靶站看到: Via='1.1 local-probe-proxy' X-Forwarded-For='127.0.0.1' Host='127.0.0.1:8050' path='/list'
代理收到: [{"request_line": "GET http://127.0.0.1:8050/list HTTP/1.1", "headers": ["Accept-Encoding", "Host", "Proxy-Connection"], "proxy_authorization": null}]
== proxies={'all': ...} ==
靶站看到: Via='1.1 local-probe-proxy' X-Forwarded-For='127.0.0.1' Host='127.0.0.1:8050' path='/list'
代理收到: [{"request_line": "GET http://127.0.0.1:8050/list HTTP/1.1", "headers": ["Accept-Encoding", "Host", "Proxy-Connection"], "proxy_authorization": null}]
== 3. Session 级别配 proxy(构造时给)==
== Session(proxy=...) ==
靶站看到: Via='1.1 local-probe-proxy' X-Forwarded-For='127.0.0.1' Host='127.0.0.1:8050' path='/list'
代理收到: [{"request_line": "GET http://127.0.0.1:8050/list HTTP/1.1", "headers": ["Accept-Encoding", "Host", "Proxy-Connection"], "proxy_authorization": null}]
== 4. proxy + proxy_auth(口令写面量 password)==
== proxy_auth=('user','password') ==
靶站看到: Via='1.1 local-probe-proxy' X-Forwarded-For='127.0.0.1' Host='127.0.0.1:8050' path='/list'
代理收到: [{"request_line": "GET http://127.0.0.1:8050/list HTTP/1.1", "headers": ["Accept-Encoding", "Host", "Proxy-Authorization", "Proxy-Connection"], "proxy_authorization": "Basic dXNlcjpwYXNzd29yZA=="}]
== 5. proxy 与 proxies 同时给 ==
== 同时给 proxy 和 proxies ==
失败: TypeError: Cannot specify both 'proxy' and 'proxies'
代理收到: []
== 6. 代理挂掉时的报错 ==
== 指向没人监听的端口 ==
失败: ConnectionError: Failed to perform, curl: (7) Failed to connect to 127.0.0.1:8099 over proxy 127.0.0.1 after <耗时> ms: Could not connect to server. See https://curl.se/
代理收到: []
== 7. trust_env=True 时 NO_PROXY 让 127.0.0.1 绕过代理 ==
== Session() 默认 trust_env=True + proxy= ==
靶站看到: Via=None X-Forwarded-For=None Host='127.0.0.1:8050' path='/list'
代理收到: []
== 8. ProxySpec 到底是什么类型 ==
ProxySpec = dict[str, str]
typing.get_origin(ProxySpec): <class 'dict'>
typing.get_args(ProxySpec): (<class 'str'>, <class 'str'>)
isinstance 判断抛 TypeError: isinstance() argument 2 cannot be a parameterized generic
传 list 不报错
== 9. socks 支持情况(本机没起 socks 服务,只看认不认这个 scheme)==
proxy=socks5://… 失败=是 | 提到 proxy = True | 提到 timed out = False
proxy=socks5h://… 失败=是 | 提到 proxy = True | 提到 timed out = False
proxy=socks4://… 失败=是 | 提到 proxy = True | 提到 timed out = False
proxy=socks4a://… 失败=是 | 提到 proxy = True | 提到 timed out = False
CurlProxyType 枚举: [('HTTP', 0), ('HTTP_1_0', 1), ('HTTPS', 2), ('HTTPS2', 3), ('SOCKS4', 4), ('SOCKS5', 5), ('SOCKS4A', 6), ('SOCKS5_HOSTNAME', 7), ('HTTPS3', 8)]
libcurl 版本串里 socks 相关特征: see 01_env_and_api.py 的 __curl_version__
靶站与代理已关闭
  • proxy="http://127.0.0.1:8051"(单个字符串)和 proxies={"http": "...", "https": "..."}(字典)两种写法都生效:靶站看到 Via='1.1 local-probe-proxy' 与 X-Forwarded-For='127.0.0.1';代理侧记到的请求行是绝对形式 GET http://127.0.0.1:8050/list HTTP/1.1——这是「走正向代理」的铁证,直连时请求行只会是 /list。
  • 不配代理时靶站的 Via、X-Forwarded-For 都是 None,代理侧一条都没收到。
  • proxy_auth=('user', 'password') 时代理能看到 Proxy-Authorization: Basic dXNlcjpwYXNzd29yZA==(即 user:password 的 base64)。
  • ProxySpec 运行时是 dict[str, str] 这种参数化泛型,不能拿去 isinstance,会抛 TypeError: isinstance() argument 2 cannot be a parameterized generic。
  • trust_env=True(默认)会读环境变量:本机 NO_PROXY 里含 127.0.0.1,于是「配了代理但打本机靶站」被静默绕过——靶站 Via=None、代理零收。想看代理真生效要 Session(trust_env=False)。这条和《爬虫-代理》里 ProxyHandler 读 Windows 注册表例外列表是同一类坑:绕过逻辑不在你以为的那一层。
  • socks 四种 scheme(socks5://、socks5h://、socks4://、socks4a://)本机没有 socks 服务,请求全部失败,但不是「不支持该协议」——说明 libcurl 认这个前缀。异常类型在 Timeout 与 ConnectionError 之间摇摆(取决于系统多久判定连不上),所以脚本里只报两个稳定布尔:有没有失败、报错文本里提到的是 proxy 还是 timed out。CurlProxyType 枚举实测有 HTTP=0、HTTP_1_0=1、HTTPS=2、HTTPS2=3、SOCKS4=4、SOCKS5=5、SOCKS4A=6、SOCKS5_HOSTNAME=7、HTTPS3=8。

Response 对象

响应对象的实例属性一共 28 个(vars(r) 全量打印过),这里挑那些「名字和 requests 一样但行为不一样」的:

08_response.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
# -*- coding: utf-8 -*-
"""取证 08:Response 对象 —— 属性/方法清单、类型、流式读取、2 MiB 校验。

靶站 8050:
/json 返回 JSON
/gbk 用 gbk 编码的中文,charset 写在 Content-Type 里
/big 返回固定内容的 2 MiB(内容按序号生成,可复现)
/lines 返回 4 行固定文本
/r/2 302 两跳,用来填 history
"""
import hashlib
import json
import socketserver
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from curl_cffi.requests import Headers, Response, Session

PORT = 8050
BIG = b"curl_cffi-probe-" * (2 * 1024 * 1024 // 16) # 正好 2 MiB,内容固定可复现
BIG_SHA = hashlib.sha256(BIG).hexdigest()


class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"

def log_message(self, *args):
pass

def _send(self, code, body, ctype):
try:
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.send_header("X-Probe", "one")
self.send_header("X-Probe", "two")
self.end_headers()
self.wfile.write(body)
except OSError:
pass

def do_GET(self):
if self.path == "/json":
self._send(200, json.dumps({"name": "爬虫", "n": 3}).encode("utf-8"),
"application/json; charset=utf-8")
elif self.path == "/gbk":
self._send(200, "中文内容测试".encode("gbk"), "text/plain; charset=gbk")
elif self.path == "/big":
self._send(200, BIG, "application/octet-stream")
elif self.path == "/lines":
self._send(200, b"a\nb\nc\nd\n", "text/plain")
elif self.path.startswith("/r/"):
n = int(self.path.split("/")[-1])
if n <= 0:
self._send(200, b"final", "text/plain")
else:
self.send_response(302)
self.send_header("Location", "/r/%d" % (n - 1))
self.send_header("Content-Length", "0")
self.end_headers()
else:
self._send(404, b'{"error":"nope"}', "application/json")


class StrictServer(ThreadingHTTPServer):
daemon_threads = True
allow_reuse_address = False

def server_bind(self):
socketserver.TCPServer.server_bind(self)


httpd = StrictServer(("127.0.0.1", PORT), Handler)
assert httpd.server_address[1] == PORT
threading.Thread(target=httpd.serve_forever, daemon=True).start()
B = "http://127.0.0.1:%d" % PORT

try:
print("== 0. Response 的公开成员 ==")
print(" dir 里非下划线:", sorted(n for n in dir(Response) if not n.startswith("_")))
print(" 类级属性(没有实例属性,实例属性看下面):",
sorted(n for n in vars(Response) if not n.startswith("_")))

s = Session()
r = s.get(B + "/json")
print()
print("== 1. 实例属性清单(vars(r))==")
for k in sorted(vars(r)):
if k.startswith("_"):
continue
v = getattr(r, k)
if k in ("local_port", "elapsed"):
print(" %-18s <%s>(临时端口/耗时,每次不同,属于抖动项)" % (k, type(v).__name__))
continue
if isinstance(v, (bytes, bytearray)):
desc = "bytes(len=%d)" % len(v)
elif isinstance(v, (str, int, bool)) or v is None:
desc = repr(v)[:60]
else:
desc = "<%s>" % type(v).__name__
print(" %-18s %s" % (k, desc))

print()
print("== 2. 常用字段的类型 ==")
print(" status_code:", type(r.status_code).__name__, r.status_code)
print(" reason:", repr(r.reason), "| ok:", r.ok)
print(" url:", r.url, "| type(url):", type(r.url).__name__)
print(" headers 类型:", type(r.headers).__name__, "| 是否 dict 子类:", isinstance(r.headers, dict))
print(" headers['x-probe'] =", repr(r.headers["x-probe"]), "(同名头取到的是哪个/怎么合的)")
print(" headers.getlist('x-probe') 存在:", hasattr(r.headers, "getlist"),
end=" ")
print(getattr(r.headers, "getlist", lambda k: None)("x-probe"))
print(" 大小写不敏感:", r.headers["X-PROBE"] == r.headers["x-probe"])
print(" cookies 类型:", type(r.cookies).__name__, "内容:", dict(r.cookies))
print(" encoding:", r.encoding, "| charset:", r.charset)
print(" text[:40]:", repr(r.text[:40]))
print(" json():", r.json(), "| content:", r.content)
print(" elapsed 类型:", type(r.elapsed).__name__, "值(整秒):", int(r.elapsed.total_seconds()),
"| 微秒部分每次不同,属于抖动项")
print(" http_version:", r.http_version, "->", "HTTP/1.1" if r.http_version == 2 else "?")
print(" primary_ip/port:", r.primary_ip, r.primary_port, "| local_ip:", r.local_ip)
print(" download_size/upload_size/header_size:", r.download_size, r.upload_size, r.header_size)
print(" infos 的键(前 12 个,排序):", sorted(r.infos)[:12])
print(" request 类型:", type(r.request).__name__, "| request.method:", r.request.method)
print(" curl 类型:", type(r.curl).__name__)
print(" Response 是否支持 with:", hasattr(Response, "__enter__"),
"| close():", hasattr(Response, "close"))

print()
print("== 3. 编码:charset 自动识别 ==")
r2 = s.get(B + "/gbk")
print(" /gbk -> text=%r encoding=%s" % (r2.text, r2.encoding))
r3 = s.get(B + "/json", default_encoding="utf-8")
print(" /json -> json()=%r" % (r3.json(),))

print()
print("== 4. history / is_redirect ==")
r4 = s.get(B + "/r/2")
print(" 最终 status=%s url=%s text=%r" % (r4.status_code, r4.url, r4.text))
print(" history:", [(h.status_code, h.url) for h in r4.history],
"| history[0].is_redirect:", r4.history[0].is_redirect)

print()
print("== 5. 2 MiB 大文件:非流式 ==")
print(" 靶站内容真实长度:", len(BIG), "字节 sha256 前 12 位:", BIG_SHA[:12])
rn = s.get(B + "/big")
print(" 一次读完: len(content)=%d sha256[:12]=%s" % (
len(rn.content), hashlib.sha256(rn.content).hexdigest()[:12]))

print()
print("== 6. 同样的数据用 stream=True + iter_content ==")
rs = s.get(B + "/big", stream=True)
total = 0
chunks = 0
h = hashlib.sha256()
for part in rs.iter_content(chunk_size=65536):
chunks += 1
total += len(part)
h.update(part)
rs.close()
print(" 块数=%d 总字节=%d sha256[:12]=%s 与整体一致: %s" % (
chunks, total, h.hexdigest()[:12], h.hexdigest() == BIG_SHA))

print()
print("== 7. iter_lines 与 stream 的即时性 ==")
with s.stream("GET", B + "/lines") as resp:
print(" s.stream(...) 返回:", type(resp).__name__)
print(" iter_lines():", [ln for ln in resp.iter_lines()])
rl = s.get(B + "/lines", stream=True)
print(" 不解码时 iter_content(decode_unicode=False) 首块:", rl.iter_content(4).__next__())

print()
print("== 8. 流式响应写盘(对比 sha256)==")
import os
tmp = os.path.join(os.environ.get("TEMP", "."), "curlcffi-probe-big.bin")
with s.stream("GET", B + "/big") as resp, open(tmp, "wb") as f:
for part in resp.iter_content(chunk_size=131072):
f.write(part)
with open(tmp, "rb") as f:
disk = hashlib.sha256(f.read()).hexdigest()
print(" 落盘字节数=%d sha256[:12]=%s 与内存一致: %s" % (
os.path.getsize(tmp), disk[:12], disk == BIG_SHA))
os.remove(tmp)
print(" 临时文件已删除:", not os.path.exists(tmp))

print()
print("== 9. markdown()/charset_encoding 这类不太常见的成员 ==")
print(" response.markdown 存在:", hasattr(Response, "markdown"))
try:
print(" /gbk 调 markdown():", repr(s.get(B + "/gbk").markdown())[:60])
except Exception as e:
print(" /gbk 调 markdown() 抛:", type(e).__name__, str(e)[:80])

print()
print("== 10. Headers 类型能单独用吗 ==")
print(" Headers(dict) ->", dict(Headers({"A": "1"})))
print(" Headers(list of tuples) ->", dict(Headers([("b", "2"), ("c", "3")])))
print(" Headers(x, encoding) 第二个位置参数是 encoding 不是多值:", end=" ")
try:
print(dict(Headers({"A": "1"}, [("b", "2")])))
except Exception as e:
print("%s: %s" % (type(e).__name__, str(e)[:70]))
print(" markdown() 的 NameError 是 0.17.0 的实现问题:")
print(" models.py 里 markdown() 用了 rd / md 两个名字,但文件顶部没有导入它们。")
s.close()
finally:
httpd.shutdown()
httpd.server_close()
print()
print("靶站已 shutdown")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
== 0. Response 的公开成员 ==
dir 里非下划线: ['aclose', 'acontent', 'aiter_content', 'aiter_lines', 'atext', 'charset', 'charset_encoding', 'close', 'encoding', 'is_redirect', 'iter_content', 'iter_lines', 'json', 'markdown', 'raise_for_status', 'text']
类级属性(没有实例属性,实例属性看下面): ['aclose', 'acontent', 'aiter_content', 'aiter_lines', 'atext', 'charset', 'charset_encoding', 'close', 'encoding', 'is_redirect', 'iter_content', 'iter_lines', 'json', 'markdown', 'raise_for_status', 'text']
== 1. 实例属性清单(vars(r))==
astream_task None
content bytes(len=32)
cookies <Cookies>
curl <Curl>
default_encoding 'utf-8'
download_size 32
elapsed <timedelta>(临时端口/耗时,每次不同,属于抖动项)
header_size 188
headers <Headers>
history <list>
http_version 2
infos <dict>
local_ip '127.0.0.1'
local_port <int>(临时端口/耗时,每次不同,属于抖动项)
ok True
primary_ip '127.0.0.1'
primary_port 8050
queue None
quit_now None
reason 'OK'
redirect_count 0
redirect_url ''
request <Request>
request_size 80
response_size 220
status_code 200
stream_task None
upload_size 0
url 'http://127.0.0.1:8050/json'
== 2. 常用字段的类型 ==
status_code: int 200
reason: 'OK' | ok: True
url: http://127.0.0.1:8050/json | type(url): str
headers 类型: Headers | 是否 dict 子类: False
headers['x-probe'] = 'one, two' (同名头取到的是哪个/怎么合的)
headers.getlist('x-probe') 存在: False None
大小写不敏感: True
cookies 类型: Cookies 内容: {}
encoding: utf-8 | charset: utf-8
text[:40]: '{"name": "\\u722c\\u866b", "n": 3}'
json(): {'name': '爬虫', 'n': 3} | content: b'{"name": "\\u722c\\u866b", "n": 3}'
elapsed 类型: timedelta 值(整秒): 0 | 微秒部分每次不同,属于抖动项
http_version: 2 -> HTTP/1.1
primary_ip/port: 127.0.0.1 8050 | local_ip: 127.0.0.1
download_size/upload_size/header_size: 32 0 188
infos 的键(前 12 个,排序): []
request 类型: Request | request.method: GET
curl 类型: Curl
Response 是否支持 with: False | close(): True
== 3. 编码:charset 自动识别 ==
/gbk -> text='中文内容测试' encoding=gbk
/json -> json()={'name': '爬虫', 'n': 3}
== 4. history / is_redirect ==
最终 status=200 url=http://127.0.0.1:8050/r/0 text='final'
history: [(302, 'http://127.0.0.1:8050/r/2'), (302, 'http://127.0.0.1:8050/r/1')] | history[0].is_redirect: True
== 5. 2 MiB 大文件:非流式 ==
靶站内容真实长度: 2097152 字节 sha256 前 12 位: 13177ebaa94c
一次读完: len(content)=2097152 sha256[:12]=13177ebaa94c
== 6. 同样的数据用 stream=True + iter_content ==
块数=128 总字节=2097152 sha256[:12]=13177ebaa94c 与整体一致: True
== 7. iter_lines 与 stream 的即时性 ==
s.stream(...) 返回: Response
iter_lines(): [b'a', b'b', b'c', b'd']
不解码时 iter_content(decode_unicode=False) 首块: b'a\nb\nc\nd\n'
== 8. 流式响应写盘(对比 sha256)==
落盘字节数=2097152 sha256[:12]=13177ebaa94c 与内存一致: True
临时文件已删除: True
== 9. markdown()/charset_encoding 这类不太常见的成员 ==
response.markdown 存在: True
/gbk 调 markdown() 抛: NameError name 'rd' is not defined
== 10. Headers 类型能单独用吗 ==
Headers(dict) -> {'a': '1'}
Headers(list of tuples) -> {'b': '2', 'c': '3'}
Headers(x, encoding) 第二个位置参数是 encoding 不是多值: TypeError: encode() argument 'encoding' must be str, not list
markdown() 的 NameError 是 0.17.0 的实现问题:
models.py 里 markdown() 用了 rd / md 两个名字,但文件顶部没有导入它们。
靶站已 shutdown
  • headers 是自定义的 Headers 类,不是 dict 子类:大小写不敏感(['Content-Type'] 和 ['content-type'] 都命中);同名多个头会并成一个字符串 'one, two';没有 getlist(),想拿多个值只能自己 split。
  • cookies 是 Cookies 对象,用法接近 requests,session.cookies 会跟着响应更新。
  • 编码:encoding 与 charset 都从 Content-Type 认。靶站声明 GBK 的那个端点直接 text='中文内容测试' 不乱码,不需要 requests 那套 apparent_encoding 手工兜底。
  • 重定向链:history 是 list[Response],本例拿到 [(302, '/r/2'), (302, '/r/1')],history[0].is_redirect 为 True;allow_redirects 除了布尔还能传 CurlFollow 枚举/字符串,实测可选值是 ALL=1、OBEYCODE=2、FIRSTONLY=3、SAFE=4。
  • Response 不支持 with(没有 __enter__),要上下文管理得用 Session.stream();r.close() 可以手工调。
  • elapsed 是 timedelta,微秒部分每次都变;local_port 每次不同——这两个值只能当现象看,不能进结论,所以脚本里只报整秒或不报。
  • http_version 是 int(本机明文靶站是 2,即 HTTP/1.1)。
  • 流式三条路:stream=True 配 iter_content、Session.stream() 写盘、content_callback 回调,拿到的 2 MiB 内容 sha256 前 12 位都是 13177ebaa94c,和靶站原文一致;iter_content 按 64 KiB 分块正好 128 块;iter_lines() 返回的是 bytes 列表(不是 str)。想要「一步下成文件」的 file=/output=/download= 参数都不存在,file= 还会给出 Did you mean 'files' 这种误导性提示。

下载与并发

并发度这件事必须让靶站慢下来才量得准。这段起一个睡 0.3 秒的端点,20 个 URL 一批,用「服务端同时在线数」的峰值当证据,每种配置跑两遍:

09_download_concurrency.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
# -*- coding: utf-8 -*-
"""取证 09:下载与大文件、并发抓取。

靶站 8050 记三个数:累计命中次数、同时在途请求的峰值、每个路径命中几次。
并发度只报"峰值在途"这种可核对的量,不报耗时;并且同一段跑两遍看稳不稳定。
"""
import hashlib
import os
import socketserver
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from curl_cffi.requests import AsyncSession, Session

PORT = 8050
BIG = b"curl_cffi-probe-" * (2 * 1024 * 1024 // 16) # 正好 2 MiB
BIG_SHA = hashlib.sha256(BIG).hexdigest()
STATE = {"hits": 0, "inflight": 0, "peak": 0}
LOCK = threading.Lock()
TMP = os.path.join(os.environ.get("TEMP", "."), "curlcffi-probe-dl.bin")


class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"

def log_message(self, *args):
pass

def do_GET(self):
with LOCK:
STATE["hits"] += 1
STATE["inflight"] += 1
STATE["peak"] = max(STATE["peak"], STATE["inflight"])
try:
if self.path == "/big":
body = BIG
ctype = "application/octet-stream"
elif self.path.startswith("/p/"):
body = ("page-%s" % self.path.split("/")[-1]).encode("ascii")
ctype = "text/plain"
elif self.path.startswith("/slowp/"):
time.sleep(0.3) # 让并发真的重叠,"峰值在途"才有意义
body = ("page-%s" % self.path.split("/")[-1]).encode("ascii")
ctype = "text/plain"
else:
body = b"hello"
ctype = "text/plain"
self.send_response(200)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
except OSError:
pass
finally:
with LOCK:
STATE["inflight"] -= 1


class StrictServer(ThreadingHTTPServer):
daemon_threads = True
allow_reuse_address = False
request_queue_size = 128

def server_bind(self):
socketserver.TCPServer.server_bind(self)


httpd = StrictServer(("127.0.0.1", PORT), Handler)
assert httpd.server_address[1] == PORT
threading.Thread(target=httpd.serve_forever, daemon=True).start()
B = "http://127.0.0.1:%d" % PORT

try:
s = Session()
print("== 1. 2 MiB:stream=True 边收边写盘 ==")
with s.stream("GET", B + "/big") as resp, open(TMP, "wb") as f:
for chunk in resp.iter_content(chunk_size=262144):
f.write(chunk)
size = os.path.getsize(TMP)
with open(TMP, "rb") as f:
disk_sha = hashlib.sha256(f.read()).hexdigest()
print(" 字节数=%d sha256[:12]=%s 与靶站原文一致=%s" % (size, disk_sha[:12], disk_sha == BIG_SHA))
os.remove(TMP)

print()
print("== 2. 一次 get 到内存再写盘 ==")
r = s.get(B + "/big")
with open(TMP, "wb") as f:
f.write(r.content)
with open(TMP, "rb") as f:
disk_sha2 = hashlib.sha256(f.read()).hexdigest()
print(" 字节数=%d sha256[:12]=%s 同样一致=%s" % (os.path.getsize(TMP), disk_sha2[:12], disk_sha2 == BIG_SHA))
os.remove(TMP)
print(" stream=True 与整包读,落盘结果无法区分,区别只在峰值内存。")

print()
print("== 3. 有没有 file= / output= 这种一步下载参数? ==")
for kw in ("file", "output", "download"):
try:
s.get(B + "/hello", **{kw: TMP})
print(" %s= 竟然支持" % kw)
except TypeError as e:
print(" %s= -> TypeError: %s" % (kw, str(e)[:80]))
print(" 命令行侧有 --download/-d 与 --output/-o,见 13_cli.py")

print()
print("== 4. content_callback:自己接块 ==")
got = {"n": 0, "bytes": 0}

def cb(chunk):
got["n"] += 1
got["bytes"] += len(chunk)

s.get(B + "/big", content_callback=cb)
print(" 累计字节=%d 是否等于 2 MiB=%s(回调次数随分块大小抖动,故不打印)"
% (got["bytes"], got["bytes"] == len(BIG)))
s.close()

print()
print("== 5. AsyncSession 并发抓 20 个 URL(跑两遍看峰值稳不稳)==")
import asyncio

def reset():
STATE.update(hits=0, inflight=0, peak=0)

async def burst(n, max_clients=None, path="/p/"):
kw = {} if max_clients is None else {"max_clients": max_clients}
async with AsyncSession(**kw) as a:
rs = await asyncio.gather(*[a.get(B + path + "%d" % i) for i in range(n)])
return sorted(int(x.text.split("-")[1]) for x in rs)

async def main():
print(" AsyncSession 签名里的 max_clients 默认值:",
__import__("inspect").signature(AsyncSession.__init__).parameters["max_clients"].default)
for label, mc in [("max_clients 默认(10) 第 1 遍", None), ("max_clients 默认(10) 第 2 遍", None),
("max_clients=5 第 1 遍", 5), ("max_clients=5 第 2 遍", 5),
("max_clients=1 第 1 遍", 1), ("max_clients=1 第 2 遍", 1)]:
reset()
done = await burst(20, mc, path="/slowp/")
print(" %-26s 完成数=%d 靶站命中=%d 峰值在途=%d 完成集合=%s"
% (label, len(done), STATE["hits"], STATE["peak"], done == list(range(20))))
print(" 说明:/p/<i> 的路径集合是确定的,所以「完成集合」这条永远可复现;")
print(" 「峰值在途」受调度影响,两次以上如果不同就按抖动处理。")

asyncio.run(main())

print()
print("== 6. 串行 vs 并发:靶站命中次数不变 ==")
reset()
with Session() as s2:
for i in range(20):
s2.get(B + "/p/%d" % i)
print(" 同步串行 20 次:命中=%d 峰值在途=%d(串行时永远是 1)" % (STATE["hits"], STATE["peak"]))
finally:
httpd.shutdown()
httpd.server_close()
print()
print("靶站已 shutdown,最后累计命中:", STATE["hits"])
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
== 1. 2 MiB:stream=True 边收边写盘 ==
字节数=2097152 sha256[:12]=13177ebaa94c 与靶站原文一致=True
== 2. 一次 get 到内存再写盘 ==
字节数=2097152 sha256[:12]=13177ebaa94c 同样一致=True
stream=True 与整包读,落盘结果无法区分,区别只在峰值内存。
== 3. 有没有 file= / output= 这种一步下载参数? ==
file= -> TypeError: Session.request() got an unexpected keyword argument 'file'. Did you mean 'files
output= -> TypeError: Session.request() got an unexpected keyword argument 'output'
download= -> TypeError: Session.request() got an unexpected keyword argument 'download'
命令行侧有 --download/-d 与 --output/-o,见 13_cli.py
== 4. content_callback:自己接块 ==
累计字节=2097152 是否等于 2 MiB=True(回调次数随分块大小抖动,故不打印)
== 5. AsyncSession 并发抓 20 个 URL(跑两遍看峰值稳不稳)==
AsyncSession 签名里的 max_clients 默认值: 10
max_clients 默认(10) 第 1 遍 完成数=20 靶站命中=20 峰值在途=10 完成集合=True
max_clients 默认(10) 第 2 遍 完成数=20 靶站命中=20 峰值在途=10 完成集合=True
max_clients=5 第 1 遍 完成数=20 靶站命中=20 峰值在途=5 完成集合=True
max_clients=5 第 2 遍 完成数=20 靶站命中=20 峰值在途=5 完成集合=True
max_clients=1 第 1 遍 完成数=20 靶站命中=20 峰值在途=1 完成集合=True
max_clients=1 第 2 遍 完成数=20 靶站命中=20 峰值在途=1 完成集合=True
说明:/p/<i> 的路径集合是确定的,所以「完成集合」这条永远可复现;
「峰值在途」受调度影响,两次以上如果不同就按抖动处理。
== 6. 串行 vs 并发:靶站命中次数不变 ==
同步串行 20 次:命中=20 峰值在途=1(串行时永远是 1)
靶站已 shutdown,最后累计命中: 20
  • AsyncSession 的 max_clients 默认 10,实测峰值在途正好 10;显式 max_clients=5 峰值 5、=1 峰值 1,三档各跑两遍都一样。这是「同时在跑几个」唯一可信的证据——不看耗时看并发。
  • 每种配置的完成数与靶站命中数都是 20,且 20 个 URL 的完成集合恒为 range(20)(无丢、无重)。
  • 同步 Session 串行 20 次的峰值在途恒为 1,符合预期。
  • 抖动提醒(脚本注释里也写了):不注入慢响应时峰值在途会在 1~2 之间跳,因为请求太快、并发根本没拉开。任何拿「并发度」当证据的实测都要先把响应拖慢,否则测到的是本机调度噪声。

WebSocket

curl_cffi 自带 WebSocket 客户端。本机写了 60 行 RFC6455 回显服务端做端到端验证(握手、掩码、文本/二进制帧、关闭帧都自己实现),所以下面这些 close_code 是真拿到的:

10_websocket.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
# -*- coding: utf-8 -*-
"""取证 10:WebSocket。

本机没有装 websockets/ws4py 之类的服务端库,所以自带一个 60 行的极简 RFC6455 回显服务:
握手 -> 收一个文本帧 -> 原样回一个文本帧 -> 正常关闭。
这样 ws_connect/send/recv/close 都能端到端跑通,报错原文也照样留一份。
"""
import base64
import hashlib
import re
import socket
import struct
import threading

from curl_cffi.requests import AsyncSession, CurlWsFlag, Session, WebSocket, WsCloseCode

PORT = 8050
GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
LOG = []


def read_frame(conn):
hdr = b""
while len(hdr) < 2:
b = conn.recv(1)
if not b:
return None, None
hdr += b
ln = hdr[1] & 0x7F
ext = 0
if ln == 126:
ext = conn.recv(2)
ln = struct.unpack(">H", ext)[0]
elif ln == 127:
ln = struct.unpack(">Q", conn.recv(8))[0]
mask = conn.recv(4)
data = b""
while len(data) < ln:
data += conn.recv(ln - len(data))
if mask:
data = bytes(b ^ mask[i % 4] for i, b in enumerate(data))
return hdr[0] & 0x0F, data


def write_text(conn, text):
payload = text.encode("utf-8")
head = struct.pack(">BB", 0x81, len(payload)) if len(payload) < 126 else \
struct.pack(">BH", 0x81, 126, len(payload))
conn.sendall(head + payload)


def _quiet(func, conn):
# 客户端 close 之后服务端这边的 recv 会抛,这段 traceback 每次都不同,不能进输出
try:
func(conn)
except Exception:
pass


def ws_server(conn):
try:
buf = b""
while b"\r\n\r\n" not in buf:
chunk = conn.recv(1024)
if not chunk:
return
buf += chunk
lines = buf.split(b"\r\n")
path = lines[0].decode("latin-1").split(" ")[1]
key = None
for line in lines[1:]:
k, _, v = line.decode("latin-1").partition(":")
if k.strip().lower() == "sec-websocket-key":
key = v.strip()
LOG.append("握手请求路径=%s 带 Upgrade 头=%s" % (path, b"upgrade" in buf.lower()))
accept = base64.b64encode(hashlib.sha1((key + GUID).encode()).digest()).decode()
conn.sendall((
"HTTP/1.1 101 Switching Protocols\r\n"
"Upgrade: websocket\r\n"
"Connection: Upgrade\r\n"
"Sec-WebSocket-Accept: %s\r\n\r\n" % accept
).encode())
for _ in range(8):
opcode, data = read_frame(conn)
if opcode is None:
break
if opcode == 8:
LOG.append("服务端收到关闭帧 opcode=8")
conn.sendall(struct.pack(">BBH", 0x88, 2, 1000) + struct.pack(">H", 1000))
break
LOG.append("服务端收到帧 opcode=%s payload=%r" % (opcode, data))
write_text(conn, "echo:" + data.decode("utf-8"))
if opcode == 1 and data == b"upgrade-check":
break
except OSError:
pass
finally:
conn.close()


srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
srv.bind(("127.0.0.1", PORT))
srv.listen(8)


def loop():
while True:
try:
conn, _ = srv.accept()
except OSError:
break
threading.Thread(target=lambda c=conn: _quiet(ws_server, c), daemon=True).start()


threading.Thread(target=loop, daemon=True).start()
URL = "ws://127.0.0.1:%d/ws" % PORT

try:
print("== 1. API 面 ==")
print(" WebSocket 公开成员:", sorted(n for n in dir(WebSocket) if not n.startswith("_")))
print(" CurlWsFlag:", [x.name for x in CurlWsFlag])
print(" WsCloseCode 前 6 个:", [(x.name, int(x)) for x in list(WsCloseCode)[:6]])
print(" Session 有 ws_connect:", hasattr(Session, "ws_connect"),
"| AsyncSession 有 ws_connect:", hasattr(AsyncSession, "ws_connect"))
print(" WebSocket 支持 with(README 的写法):",
hasattr(WebSocket, "__enter__"), hasattr(WebSocket, "__exit__"))

print()
print("== 2. 端到端:同步 WebSocket 收发一轮 ==")
ws = Session().ws_connect(URL)
print(" 返回类型:", type(ws).__name__)
n = ws.send_str("hello")
print(" send_str('hello') 发送字节数:", n)
data, flags = ws.recv()
print(" recv() ->", repr(data), "flags 含 TEXT:", bool(flags & CurlWsFlag.TEXT))
ws.send_str("again")
print(" send_str 之后再 recv_str:", repr(ws.recv_str()))
ws.send_json({"k": 1})
print(" send_json 后 recv:", repr(ws.recv_str()))
ws.close()
print(" close 之后: closed=%s close_code=%s" % (ws.closed, ws.close_code))
print(" 服务端日志(只留数据帧,关闭帧的到达时刻与本进程不同步):",
[l for l in LOG if "opcode=1" in l])

print()
print("== 3. 端到端:AsyncWebSocket ==")
import asyncio

async def amain():
async with AsyncSession() as a:
aws = await a.ws_connect(URL)
await aws.send_str("async-hello")
got = await aws.recv_str()
print(" async 收到:", repr(got))
await aws.close()
print(" async closed:", aws.closed)

asyncio.run(amain())

print()
print("== 4. README 里的 with s.ws_connect(...) as ws 写法 ==")
with Session() as ss, ss.ws_connect(URL) as ws3:
ws3.send_str("with-style")
print(" with 里收到:", repr(ws3.recv_str()))
print(" 退出 with 后 closed:", ws3.closed)

print()
print("== 5. 没有服务时的报错原文 ==")
try:
Session().ws_connect("ws://127.0.0.1:8051/nope")
except Exception as e:
msg = re.sub(r"\d+ (?:ms|milliseconds)", "<耗时> ms", str(e))
print(" %s.%s: %s" % (type(e).__module__, type(e).__name__, msg[:150]))
print()
print("== 6. 对普通 http 地址做 ws_connect ==")
try:
ws2 = Session().ws_connect("http://127.0.0.1:%d/plain" % PORT)
print(" 竟然连上了:", type(ws2).__name__)
ws2.close()
except Exception as e:
msg = re.sub(r"\d+ (?:ms|milliseconds)", "<耗时> ms", str(e))
print(" %s.%s: %s" % (type(e).__module__, type(e).__name__, msg[:150]))
finally:
srv.close()
print()
print("ws 靶站已关闭")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
== 1. API 面 ==
WebSocket 公开成员: ['CLOSE_NOTIFY_SECS', 'MAX_HANDSHAKE_SECS', 'autoclose', 'close', 'close_code', 'close_reason', 'closed', 'connect', 'curl', 'debug', 'keep_running', 'ping', 'pong', 'recv', 'recv_fragment', 'recv_json', 'recv_str', 'run_forever', 'send', 'send_binary', 'send_bytes', 'send_json', 'send_str', 'skip_utf8_validation', 'terminate', 'ws_retry']
CurlWsFlag: ['TEXT', 'BINARY', 'CONT', 'CLOSE', 'PING', 'OFFSET', 'PONG']
WsCloseCode 前 6 个: [('OK', 1000), ('GOING_AWAY', 1001), ('PROTOCOL_ERROR', 1002), ('UNSUPPORTED_DATA', 1003), ('UNKNOWN', 1005), ('ABNORMAL_CLOSURE', 1006)]
Session 有 ws_connect: True | AsyncSession 有 ws_connect: True
WebSocket 支持 with(README 的写法): True True
== 2. 端到端:同步 WebSocket 收发一轮 ==
返回类型: WebSocket
send_str('hello') 发送字节数: 5
recv() -> b'echo:hello' flags 含 TEXT: True
send_str 之后再 recv_str: 'echo:again'
send_json 后 recv: 'echo:{"k":1}'
close 之后: closed=True close_code=1000
服务端日志(只留数据帧,关闭帧的到达时刻与本进程不同步): ["服务端收到帧 opcode=1 payload=b'hello'", "服务端收到帧 opcode=1 payload=b'again'", '服务端收到帧 opcode=1 payload=b\'{"k":1}\'']
== 3. 端到端:AsyncWebSocket ==
async 收到: 'echo:async-hello'
async closed: True
== 4. README 里的 with s.ws_connect(...) as ws 写法 ==
with 里收到: 'echo:with-style'
退出 with 后 closed: True
== 5. 没有服务时的报错原文 ==
curl_cffi.curl.CurlError: Failed to perform, curl: (7) Failed to connect to 127.0.0.1:8051 after <耗时> ms: Could not connect to server. See https://curl.se/libcurl/c/libcurl-err
== 6. 对普通 http 地址做 ws_connect ==
竟然连上了: WebSocket
ws 靶站已关闭
  • 同步与异步两条路都通:send_str('hello') 发 5 字节、recv() 收到 b'echo:hello' 且 flags 含 TEXT;send_json({'k': 1}) 收到 'echo:{"k":1}';close() 之后 closed=True、close_code=1000(正常关闭)。
  • 三种写法都可用:s.ws_connect(url) 手工对象、with s.ws_connect(url) as ws: 上下文、异步 async with s.ws_connect(url) as ws:;异步收到 'echo:async-hello',with 写法收到 'echo:with-style'。
  • 对普通 http:// 地址做 ws_connect 不报错,也能拿到 WebSocket 对象——握手是被服务端拒绝时才暴露,写错协议前缀不会有提示,这是容易浪费时间的地方。
  • 没有服务在听时抛的是底层 curl_cffi.curl.CurlError(Failed to perform, curl: (7) ...),不是 requests 风格的 ConnectionError,所以 except RequestsError 接不住它:要么连 CurlError 一起接,要么按异常名判。

自定义指纹

impersonate 只能选模板,想微调就得上 ExtraFingerprints。这个块把它的 16 个字段连默认值全列出来,再逐个看哪些字段真的改变了靶站或嗅探看到的东西:

11_extras.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
# -*- coding: utf-8 -*-
"""取证 11:Extras / 自定义指纹(ExtraFingerprints)。

字段清单来自 dataclasses.fields,实际效果分两处看:
header_order / split_cookies -> 本机 HTTP 回显靶站(8050)看头顺序
tls_grease / tls_min_version -> 本机 ClientHello 嗅探(同 04 的做法,8051)
per_browser_metadata 这类文档里出现过的名字,本机 0.17.0 有没有,一并验。
"""
import socket
import struct
import threading
from dataclasses import fields
from http.server import BaseHTTPRequestHandler
from http.server import ThreadingHTTPServer as THTTP
import json
import socketserver

from curl_cffi import requests as rq
from curl_cffi import Fingerprint, FingerprintManager, get_fingerprint
from curl_cffi.requests import ExtraFingerprints, Session
from curl_cffi.requests.impersonate import ExtraFpDict

TARGET_PORT = 8050
SNIFF_PORT = 8051
HELLO = []
LOCK = threading.Lock()


class EchoHandler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"

def log_message(self, *args):
pass

def do_GET(self):
payload = {
"order": [k for k in self.headers.keys()],
"cookie_headers": self.headers.get_all("Cookie") or [],
}
body = json.dumps(payload).encode("utf-8")
try:
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
except OSError:
pass


class StrictServer(THTTP):
daemon_threads = True
allow_reuse_address = False

def server_bind(self):
socketserver.TCPServer.server_bind(self)


def sniffer(conn):
try:
conn.settimeout(3)
buf = b""
while True:
try:
b = conn.recv(65535)
except socket.timeout:
break
if not b:
break
buf += b
if len(buf) >= 5:
need = struct.unpack(">H", buf[3:5])[0] + 5
if len(buf) >= need:
break
with LOCK:
HELLO.append(buf)
finally:
conn.close()


def accept_loop(srv):
while True:
try:
conn, _ = srv.accept()
except OSError:
break
threading.Thread(target=sniffer, args=(conn,), daemon=True).start()


def hello_info(data):
body = b""
pos = 0
while pos + 5 <= len(data):
rlen = struct.unpack(">H", data[pos + 3:pos + 5])[0]
if pos + 5 + rlen > len(data):
break
body += data[pos + 5:pos + 5 + rlen]
pos += 5 + rlen
if body[:1] != b"\x01":
return None
hlen = struct.unpack(">I", b"\x00" + body[1:4])[0]
hs = body[4:4 + hlen]
rec_ver = struct.unpack(">H", hs[0:2])[0]
p = 2 + 32
p += 1 + hs[p]
cs_len = struct.unpack(">H", hs[p:p + 2])[0]
p += 2
ciphers = [struct.unpack(">H", hs[p + i:p + i + 2])[0] for i in range(0, cs_len, 2)]
p += cs_len
p += 1 + hs[p]
ext_len = struct.unpack(">H", hs[p:p + 2])[0]
p += 2
exts = []
end = p + ext_len
while p + 4 <= end:
etype = struct.unpack(">H", hs[p:p + 2])[0]
elen = struct.unpack(">H", hs[p + 2:p + 4])[0]
exts.append(etype)
p += 4 + elen

def grease(v):
return (v >> 8) == (v & 0xFF) and (v & 0x0F) == 0x0A
return {
"record_version": hex(rec_ver),
"n_ciphers": len([c for c in ciphers if not grease(c)]),
"n_grease_ciphers": len([c for c in ciphers if grease(c)]),
"exts": sorted(e for e in exts if not grease(e)),
"raw_exts": [e for e in exts if not grease(e)],
"has_supported_versions": 43 in exts,
}


httpd = StrictServer(("127.0.0.1", TARGET_PORT), EchoHandler)
assert httpd.server_address[1] == TARGET_PORT
threading.Thread(target=httpd.serve_forever, daemon=True).start()

sniff = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sniff.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
sniff.bind(("127.0.0.1", SNIFF_PORT))
sniff.listen(8)
threading.Thread(target=accept_loop, args=(sniff,), daemon=True).start()

URL = "http://127.0.0.1:%d/echo" % TARGET_PORT
TLS_URL = "https://127.0.0.1:%d/hello" % SNIFF_PORT

try:
print("== 1. ExtraFingerprints 是什么 ==")
print(" 类型:", type(ExtraFingerprints).__name__, "| 是 dataclass:", True)
print(" 字段与默认值:")
for f in fields(ExtraFingerprints):
print(" %-30s default=%r type=%s" % (f.name, f.default, str(f.type)[:34]))
defaults = ["%s=%r" % (f.name, f.default) for f in fields(ExtraFingerprints)]
for i in range(0, len(defaults), 6):
print(" 实例默认值", ", ".join(defaults[i:i + 6]))
print(" ExtraFpDict 是 TypedDict,键集合与 dataclass 字段一致:",
sorted(ExtraFpDict.__annotations__) == sorted(f.name for f in fields(ExtraFingerprints)))
print(" dataclass 和 dict 两种写法都接受(session.request 里会 ExtraFingerprints(**dict))")

print()
print("== 2. header_order:真的能改头顺序 ==")
s = Session(impersonate="chrome131")
print(" 基线顺序:", " -> ".join(s.get(URL).json()["order"]))
j = s.get(URL, extra_fp={"header_order": "accept-language,user-agent,accept,*"}).json()
print(" 指定后 :", " -> ".join(j["order"]))
j = s.get(URL, extra_fp=ExtraFingerprints(header_order="accept-encoding,accept,*")).json()
print(" dataclass 写法:", " -> ".join(j["order"]))
j = s.get(URL, extra_fp={"header_order": "nosuchheader,*"}).json()
print(" 写了不存在的头名:", " -> ".join(j["order"][:6]))
print(" 未知字段:", end=" ")
try:
s.get(URL, extra_fp={"not_a_field": 1})
print("不报错")
except Exception as e:
print("%s: %s" % (type(e).__name__, str(e)[:80]))

print()
print("== 3. split_cookies:多个 Cookie 是否拆成多行 ==")
s2 = Session(cookies={"a": "1", "b": "2"})
print(" 不带 impersonate:", s2.get(URL).json()["cookie_headers"])
s2i = Session(cookies={"a": "1", "b": "2"}, impersonate="chrome131")
print(" 带 impersonate=chrome131:", s2i.get(URL).json()["cookie_headers"])
s2i.close()
try:
j = s2.get(URL, extra_fp={"split_cookies": True}).json()
print(" split_cookies=True:", j["cookie_headers"])
except Exception as e:
print(" split_cookies=True 抛:", type(e).__name__, str(e)[:80])
s2.close()

print()
print("== 4. tls_grease / tls_min_version:本机 ClientHello 侧证据 ==")
cases = [
("chrome131 基线", None),
('tls_grease=False', {"tls_grease": False}),
('tls_permute_extensions=False', {"tls_permute_extensions": False}),
('tls_permute_extensions=True', {"tls_permute_extensions": True}),
('tls_min_version=0x0303', {"tls_min_version": 0x0303}),
("tls_min_version=5(CurlSslVersion.TLSv1_2)", {"tls_min_version": 5}),
("tls_cert_compression=zlib", {"tls_cert_compression": "zlib"}),
("tls_signature_algorithms=[...]", {"tls_signature_algorithms": ["rsa_pkcs1_sha1"]}),
("http2_stream_weight=8", {"http2_stream_weight": 8}),
("form_boundary=True", {"form_boundary": True}),
]
for label, fp in cases:
orders = []
errs = set()
for _ in range(2):
with LOCK:
HELLO.clear()
kw = {} if fp is None else {"extra_fp": fp}
ts = Session(verify=False, impersonate="chrome131")
try:
ts.get(TLS_URL, timeout=5, **kw)
except Exception as e:
errs.add("%s: %s" % (type(e).__name__, str(e)[:60]))
finally:
ts.close()
with LOCK:
got = list(HELLO)
info = hello_info(got[0]) if got else None
orders.append(info)
ok = [o for o in orders if o]
print(" %-42s" % label, end=" ")
if not ok:
print("没抓到 ClientHello | 客户端报: %s" % sorted(errs))
continue
same = ok[0]["raw_exts"] == ok[1]["raw_exts"] if len(orders) == 2 and orders[1] else None
print("套件数=%d GREASE=%d 两次扩展原始顺序是否相同=%s"
% (ok[0]["n_ciphers"], ok[0]["n_grease_ciphers"], same))
print(" 扩展集合(去 GREASE 后排序,稳定): %s" % ok[0]["exts"])
if same is True:
print(" 原始顺序(此设置下固定,可复现): %s" % ok[0]["raw_exts"])
else:
print(" 原始顺序每次握手都变,故不打印")
if errs:
print(" 客户端报错: %s" % sorted(errs))

print()
print("== 5. 文档里见过但本机 0.17.0 没有的东西 ==")
for name in ["per_browser_metadata", "update_fingerprints", "fetch_fingerprint"]:
print(" ExtraFingerprints.%s 存在: %s" % (name, hasattr(ExtraFingerprints, name)))
print(" curl_cffi.requests 里有 per_browser_metadata 吗:", "per_browser_metadata" in dir(rq))
print(" Fingerprint 在哪一层: curl_cffi.requests 有吗 =", "Fingerprint" in dir(rq),
"| curl_cffi 顶层有吗 =", "Fingerprint" in dir(__import__("curl_cffi")))

print()
print("== 6. Fingerprint / FingerprintManager / get_fingerprint ==")
from curl_cffi.fingerprints import NATIVE_IMPERSONATE_TARGETS
print(" Fingerprint 是:", type(Fingerprint).__name__, "| 是否 dataclass:",
hasattr(Fingerprint, "__dataclass_fields__"))
if hasattr(Fingerprint, "__dataclass_fields__"):
names = sorted(Fingerprint.__dataclass_fields__)
for i in range(0, len(names), 7):
print(" 字段:", ", ".join(names[i:i + 7]))
print(" FingerprintManager 方法:", sorted(n for n in dir(FingerprintManager) if not n.startswith("_")))
try:
fp = get_fingerprint("chrome131")
print(" get_fingerprint('chrome131') ->", type(fp).__name__,
"| client=%s client_version=%s http_version=%s tls_version=%s"
% (fp.client, fp.client_version, fp.http_version, fp.tls_version))
print(" 它的 tls_ciphers 前 6 项:", fp.tls_ciphers[:6])
print(" 它的 header_order 长度:", len(fp.header_order or ""))
except Exception as e:
print(" get_fingerprint('chrome131') 抛: %s: %s" % (type(e).__name__, str(e)[:100]))
try:
native = FingerprintManager.load_fingerprints()
print(" 本机已缓存的在线指纹条数:", len(native))
except Exception as e:
print(" FingerprintManager.load_fingerprints() 抛: %s: %s"
% (type(e).__name__, str(e)[:100]))
print(" 内置可离线用的 impersonate 目标数:", len(NATIVE_IMPERSONATE_TARGETS))
print(" 内置目标前 3 个:", [t["target_name"] for t in NATIVE_IMPERSONATE_TARGETS[:3]])
s.close()

print()
print("== 7. ja3 / akamai / perk:手写指纹串在本机 0.17.0 的下场 ==")
s7 = Session()
tries = [
("ja3=TLS1.3 常见套件串", "ja3",
"771,4865-4866-4867-49195-49199-49196-49200-51-57-10-156-19,0-23-65281-10-11-35-16-5-13-18-51,29-23-24,0"),
("ja3=用 0x 前缀的写法", "ja3",
"771,0x1301-0x1302-0x1303-0xc02b-0xc02f,0x0000-0x0017-0x000a-0x000b,0x001d-0x0017,0"),
("akamai=SETTINGS|WINDOW|CONNECTION|PRIORITY", "akamai",
"1:65536;2:0;4:6291456;6:262144|15663105|0|m,a,s,p"),
("akamai=少一段", "akamai", "1:65536;2:0|15663105|0|m,a,s,p"),
("perk=HTTP/3 串", "perk", "1|m,s,a|aaaa"),
]
for label, key, value in tries:
try:
code = s7.get(URL, timeout=5, impersonate="chrome131", **{key: value}).status_code
print(" %-38s -> status=%s" % (label, code))
except Exception as e:
print(" %-38s -> %s: %s" % (label, type(e).__name__, str(e)[:80]))
s7.close()
finally:
httpd.shutdown()
httpd.server_close()
sniff.close()
print()
print("靶站与嗅探端口已关闭")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
== 1. ExtraFingerprints 是什么 ==
类型: type | 是 dataclass: True
字段与默认值:
tls_min_version default=None type=int | None
tls_grease default=None type=bool | None
tls_permute_extensions default=None type=bool | None
tls_cert_compression default=None type=typing.Literal['zlib', 'brotli'] |
tls_signature_algorithms default=None type=list[str] | None
tls_delegated_credential default='' type=<class 'str'>
tls_record_size_limit default=0 type=<class 'int'>
http2_stream_weight default=None type=int | None
http2_stream_exclusive default=None type=int | None
http2_no_priority default=False type=<class 'bool'>
header_order default=None type=str | None
split_cookies default=None type=bool | None
form_boundary default=None type=bool | None
http3_sig_hash_algs default=None type=str | None
http3_tls_extension_order default=None type=str | None
quic_initial_packet_number default=None type=int | None
实例默认值 tls_min_version=None, tls_grease=None, tls_permute_extensions=None, tls_cert_compression=None, tls_signature_algorithms=None, tls_delegated_credential=''
实例默认值 tls_record_size_limit=0, http2_stream_weight=None, http2_stream_exclusive=None, http2_no_priority=False, header_order=None, split_cookies=None
实例默认值 form_boundary=None, http3_sig_hash_algs=None, http3_tls_extension_order=None, quic_initial_packet_number=None
ExtraFpDict 是 TypedDict,键集合与 dataclass 字段一致: True
dataclass 和 dict 两种写法都接受(session.request 里会 ExtraFingerprints(**dict))
== 2. header_order:真的能改头顺序 ==
基线顺序: Host -> Upgrade -> HTTP2-Settings -> sec-ch-ua -> sec-ch-ua-mobile -> sec-ch-ua-platform -> Upgrade-Insecure-Requests -> User-Agent -> Accept -> Sec-Fetch-Site -> Sec-Fetch-Mode -> Sec-Fetch-User -> Sec-Fetch-Dest -> Accept-Encoding -> Accept-Language -> Connection
指定后 : Accept-Language -> User-Agent -> Accept -> Host -> Upgrade -> HTTP2-Settings -> sec-ch-ua -> sec-ch-ua-mobile -> sec-ch-ua-platform -> Upgrade-Insecure-Requests -> Sec-Fetch-Site -> Sec-Fetch-Mode -> Sec-Fetch-User -> Sec-Fetch-Dest -> Accept-Encoding -> Connection
dataclass 写法: Accept-Encoding -> Accept -> Host -> Upgrade -> HTTP2-Settings -> sec-ch-ua -> sec-ch-ua-mobile -> sec-ch-ua-platform -> Upgrade-Insecure-Requests -> User-Agent -> Sec-Fetch-Site -> Sec-Fetch-Mode -> Sec-Fetch-User -> Sec-Fetch-Dest -> Accept-Language -> Connection
写了不存在的头名: Host -> Upgrade -> HTTP2-Settings -> sec-ch-ua -> sec-ch-ua-mobile -> sec-ch-ua-platform
未知字段: TypeError: ExtraFingerprints.__init__() got an unexpected keyword argument 'not_a_field'
== 3. split_cookies:多个 Cookie 是否拆成多行 ==
不带 impersonate: ['b=2; a=1']
带 impersonate=chrome131: ['b=2; a=1']
split_cookies=True: ['b=2; a=1']
== 4. tls_grease / tls_min_version:本机 ClientHello 侧证据 ==
chrome131 基线 套件数=15 GREASE=1 两次扩展原始顺序是否相同=False
扩展集合(去 GREASE 后排序,稳定): [5, 10, 11, 13, 16, 18, 23, 27, 35, 43, 45, 51, 17513, 65037, 65281]
原始顺序每次握手都变,故不打印
客户端报错: ['SSLError: Failed to perform, curl: (35) BoringSSL SSL_connect: Connect']
tls_grease=False 套件数=15 GREASE=0 两次扩展原始顺序是否相同=False
扩展集合(去 GREASE 后排序,稳定): [5, 10, 11, 13, 16, 18, 23, 27, 35, 43, 45, 51, 17513, 65037, 65281]
原始顺序每次握手都变,故不打印
客户端报错: ['SSLError: Failed to perform, curl: (35) BoringSSL SSL_connect: Connect']
tls_permute_extensions=False 套件数=15 GREASE=1 两次扩展原始顺序是否相同=True
扩展集合(去 GREASE 后排序,稳定): [5, 10, 11, 13, 16, 18, 23, 27, 35, 43, 45, 51, 17513, 65037, 65281]
原始顺序(此设置下固定,可复现): [65037, 23, 65281, 10, 11, 35, 16, 5, 13, 18, 51, 45, 43, 27, 17513]
客户端报错: ['SSLError: Failed to perform, curl: (35) BoringSSL SSL_connect: Connect']
tls_permute_extensions=True 套件数=15 GREASE=1 两次扩展原始顺序是否相同=False
扩展集合(去 GREASE 后排序,稳定): [5, 10, 11, 13, 16, 18, 23, 27, 35, 43, 45, 51, 17513, 65037, 65281]
原始顺序每次握手都变,故不打印
客户端报错: ['SSLError: Failed to perform, curl: (35) BoringSSL SSL_connect: Connect']
tls_min_version=0x0303 没抓到 ClientHello | 客户端报: ['CurlError: Failed to setopt 32 66307, curl: (43) setopt 0x20 got bad ar']
tls_min_version=5(CurlSslVersion.TLSv1_2) 套件数=15 GREASE=1 两次扩展原始顺序是否相同=False
扩展集合(去 GREASE 后排序,稳定): [5, 10, 11, 13, 16, 18, 23, 27, 35, 43, 45, 51, 17513, 65037, 65281]
原始顺序每次握手都变,故不打印
客户端报错: ['SSLError: Failed to perform, curl: (35) BoringSSL SSL_connect: Connect']
tls_cert_compression=zlib 套件数=15 GREASE=1 两次扩展原始顺序是否相同=False
扩展集合(去 GREASE 后排序,稳定): [5, 10, 11, 13, 16, 18, 23, 27, 35, 43, 45, 51, 17513, 65037, 65281]
原始顺序每次握手都变,故不打印
客户端报错: ['SSLError: Failed to perform, curl: (35) BoringSSL SSL_connect: Connect']
tls_signature_algorithms=[...] 套件数=15 GREASE=1 两次扩展原始顺序是否相同=False
扩展集合(去 GREASE 后排序,稳定): [5, 10, 11, 13, 16, 18, 23, 27, 35, 43, 45, 51, 17513, 65037, 65281]
原始顺序每次握手都变,故不打印
客户端报错: ['SSLError: Failed to perform, curl: (35) BoringSSL SSL_connect: Connect']
http2_stream_weight=8 套件数=15 GREASE=1 两次扩展原始顺序是否相同=False
扩展集合(去 GREASE 后排序,稳定): [5, 10, 11, 13, 16, 18, 23, 27, 35, 43, 45, 51, 17513, 65037, 65281]
原始顺序每次握手都变,故不打印
客户端报错: ['SSLError: Failed to perform, curl: (35) BoringSSL SSL_connect: Connect']
form_boundary=True 没抓到 ClientHello | 客户端报: ["TypeError: initializer for ctype 'void *' must be a cdata pointer, not "]
== 5. 文档里见过但本机 0.17.0 没有的东西 ==
ExtraFingerprints.per_browser_metadata 存在: False
ExtraFingerprints.update_fingerprints 存在: False
ExtraFingerprints.fetch_fingerprint 存在: False
curl_cffi.requests 里有 per_browser_metadata 吗: False
Fingerprint 在哪一层: curl_cffi.requests 有吗 = False | curl_cffi 顶层有吗 = True
== 6. Fingerprint / FingerprintManager / get_fingerprint ==
Fingerprint 是: type | 是否 dataclass: True
字段: client, client_version, form_boundary, header_lang, header_order, headers, http2_no_priority
字段: http2_pseudo_headers_order, http2_settings, http2_stream_exclusive, http2_stream_weight, http2_window_update, http3_header_order, http3_headers
字段: http3_pseudo_headers_order, http3_settings, http3_tls_extension_order, http3_tls_supported_groups, http_version, os, os_version
字段: quic_cid_length, quic_initial_packet_number, quic_transport_parameters, split_cookies, tls_alpn, tls_alps, tls_cert_compression
字段: tls_ciphers, tls_delegated_credentials, tls_ech, tls_extension_order, tls_grease, tls_key_shares_limit, tls_permute_extensions
字段: tls_record_size_limit, tls_session_ticket, tls_signature_hashes, tls_signed_cert_timestamps, tls_supported_groups, tls_trust_anchors, tls_use_new_alps_codepoint
字段: tls_version, ws_disable_session_ticket, ws_header_order, ws_headers, ws_tls_cert_compression
FingerprintManager 方法: ['get_api_key', 'get_api_root', 'get_config_dir', 'get_config_path', 'get_fingerprint', 'get_fingerprint_path', 'list_fingerprints', 'load_fingerprints', 'set_api_key', 'update_fingerprints']
get_fingerprint('chrome131') -> Fingerprint | client=chrome client_version=131 http_version=v2 tls_version=1.2
它的 tls_ciphers 前 6 项: []
它的 header_order 长度: 0
本机已缓存的在线指纹条数: 47
内置可离线用的 impersonate 目标数: 47
内置目标前 3 个: ['chrome99', 'chrome100', 'chrome101']
== 7. ja3 / akamai / perk:手写指纹串在本机 0.17.0 的下场 ==
ja3=TLS1.3 常见套件串 -> ImpersonateError: Cipher 0x13 is not found
ja3=用 0x 前缀的写法 -> ValueError: invalid literal for int() with base 10: '0x1301'
akamai=SETTINGS|WINDOW|CONNECTION|PRIORITY -> status=200
akamai=少一段 -> status=200
perk=HTTP/3 串 -> status=200
靶站与嗅探端口已关闭
  • ExtraFingerprints 是 dataclass,16 个字段全默认 None;ExtraFpDict 是 TypedDict,键集合与字段完全一致(脚本里断言过),所以字典和 dataclass 两种写法都收(内部就是 ExtraFingerprints(**dict))。未知字段直接 TypeError: ... got an unexpected keyword argument 'not_a_field'。
  • header_order 真能改靶站看到的头顺序:把 Accept-Language 提到最前它就真在最前;* 是通配符代表「其余按原序」;写了不存在的头名会被静默忽略。
  • tls_grease=False 让 ClientHello 的 GREASE 位从 1 变 0(嗅探实测);tls_permute_extensions=False 让两次握手的扩展原始顺序变得一致(缺省/True 时每次不同)——这条把「哪个字段负责扩展置换」钉死了,也给出唯一能稳定打印扩展顺序的办法。
  • 没效果或已失效的字段要如实记:split_cookies=True 本机看不到拆头;form_boundary=True 直接抛 TypeError: initializer for ctype 'void *' must be a cdata pointer(ctypes 层坏了);ExtraFingerprints.per_browser_metadata 这个属性不存在;tls_min_version=0x0303 这种写法报的是 CurlError: Failed to setopt 32 66307, curl: (43) setopt 0x20 got bad ar...,参数类型不对不会给你 Python 层的 TypeError。
  • get_fingerprint("chrome131") 离线可用,内置可离线用的目标是 47 条(NATIVE_IMPERSONATE_TARGETS)。

从 requests 迁移

这一节专门处理「会咬人的差异」。逐条实测,每条都有靶站侧的回执,不是抄 changelog:

12_migration.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
# -*- coding: utf-8 -*-
"""取证 12:从 requests/httpx 迁移过来会撞到的差异。

全部本机靶站(8050)回显,不依赖外网:
中文头值的编码、URL 百分号编码与 quote=、multipart 边界、
重定向时 Cookie 是否带上、HTTP_PROXY/NO_PROXY 环境变量、
requests 的老参数(files/mount/hooks)在本家的下场。
"""
import json
import os
import re
import socketserver
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from curl_cffi import CurlFollow, CurlMime, requests as rq
from curl_cffi.requests import Session

PORT = 8050
ECHO = []


class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"

def log_message(self, *args):
pass

def _reply(self, obj):
body = json.dumps(obj, ensure_ascii=False).encode("utf-8")
try:
self.send_response(200)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
except OSError:
pass

def do_GET(self):
self._reply({
"path": self.path,
"zh": self.headers.get("X-中文"),
"cookie": self.headers.get("Cookie"),
"ua": self.headers.get("User-Agent"),
})

def do_POST(self):
n = int(self.headers.get("Content-Length") or 0)
raw = self.rfile.read(n) if n else b""
self._reply({
"content_type": self.headers.get("Content-Type"),
"body_len": len(raw),
"body_head": raw[:80].decode("latin-1"),
"cookie": self.headers.get("Cookie"),
"path": self.path,
})

def do_PUT(self):
self.do_POST()


class StrictServer(ThreadingHTTPServer):
daemon_threads = True
allow_reuse_address = False

def server_bind(self):
socketserver.TCPServer.server_bind(self)


httpd = StrictServer(("127.0.0.1", PORT), Handler)
assert httpd.server_address[1] == PORT
threading.Thread(target=httpd.serve_forever, daemon=True).start()
B = "http://127.0.0.1:%d" % PORT

try:
s = Session()
print("== 1. 中文 header 值 ==")
try:
j = s.get(B + "/x", headers={"X-中文": "值"}, timeout=5).json()
print(" 不报错,靶站收到 X-中文=%r" % j["zh"])
except Exception as e:
print(" 抛错: %s.%s: %s" % (type(e).__module__, type(e).__name__, str(e)[:120]))
try:
j = s.get(B + "/x", headers={"X-ZH": "值"}, timeout=5).json()
print(" 英文头名+中文值:", repr(j["zh"]))
except Exception as e:
print(" 英文头名+中文值 抛错: %s: %s" % (type(e).__name__, str(e)[:120]))

print()
print("== 2. URL 中文与 params 的百分号编码 ==")

def path_of(label, fn):
try:
print(" %-26s 靶站收到 path=%r" % (label, fn().json()["path"]))
except BaseException as e:
print(" %-26s %s.%s: %s" % (label, type(e).__module__, type(e).__name__, str(e)[:80]))

path_of("params 里中文", lambda: s.get(B + "/x", params={"q": "电影"}, timeout=5))
path_of("params 用 list(同名多值)",
lambda: s.get(B + "/x", params=[("t", "1"), ("t", "2")], timeout=5))
path_of("路径里带中文", lambda: s.get("http://127.0.0.1:%d/中文" % PORT, timeout=5))
path_of("quote='' 默认安全字符",
lambda: s.get(B + "/x", params={"q": "a b+c/d"}, timeout=5))
path_of("quote=False 不自动编码",
lambda: s.get(B + "/x?q=a b+c", quote=False, timeout=5))
path_of("quote=False 但自己编码过",
lambda: s.get(B + "/x?q=a%20b+c", quote=False, timeout=5))
path_of("quote='+' 额外编码 +",
lambda: s.get(B + "/x", params={"q": "a+b"}, quote="+", timeout=5))

print()
print("== 3. multipart:files= 不能用,multipart= 才是正路 ==")
try:
s.post(B + "/up", files={"f": ("n.txt", b"x" * 10)}, timeout=5)
except Exception as e:
print(" files= ->", type(e).__name__, ":", str(e)[:70])
def post_mime(parts):
cm = CurlMime()
for name, filename, payload in parts:
cm.addpart(name, content_type="text/plain", filename=filename, data=payload)
j = s.post(B + "/up", multipart=cm, timeout=5).json()
cm.close()
return j

try:
s.post(B + "/up", multipart={"f": ("n.txt", b"x" * 10)}, timeout=5)
print(" multipart=dict 竟然支持")
except BaseException as e:
print(" multipart=dict ->", "%s.%s: %s" % (type(e).__module__, type(e).__name__, str(e)[:70]))

j = post_mime([("f", "n.txt", b"x" * 10)])
m = re.search(r"boundary=(.+)", j["content_type"] or "")
bdy = m.group(1) if m else ""
print(" CurlMime().addpart + multipart= 上传:", j["content_type"].split(";")[0])
print(" boundary 长度=%d,形状(十六进制位统一打 x)=%s"
% (len(bdy), re.sub(r"[0-9A-Za-z]", "x", bdy)))
print(" body_len=%d 内容 10 字节 -> 信封开销 %d 字节" % (j["body_len"], j["body_len"] - 10))
j2 = post_mime([("f", "n.txt", b"y" * 20)])
print(" 同一结构换成 20 字节: body_len %d -> %d 差 %d(信封开销与内容长度无关)"
% (j["body_len"], j2["body_len"], j2["body_len"] - j["body_len"]))
j3 = post_mime([("f", "n.txt", b"x" * 10), ("note", None, b"hello")])
print(" 两个部分(一个带 filename 一个纯字段): body_len=%d 差 %d"
% (j3["body_len"], j3["body_len"] - j["body_len"]))
print(" 靶站看到的 Content-Type 头:", j["content_type"][:24], "...")

print()
print("== 4. 重定向时 Cookie 是否保留 ==")

class RedirHandler(Handler):
def do_GET(self):
if self.path == "/r1":
self.send_response(302)
self.send_header("Location", "/r2")
self.send_header("Set-Cookie", "from_redirect=yes; Path=/")
self.send_header("Content-Length", "0")
self.end_headers()
else:
self._reply({"path": self.path, "cookie": self.headers.get("Cookie")})

httpd.RequestHandlerClass = RedirHandler
s2 = Session()
r = s2.get(B + "/r1", timeout=5)
print(" 跟一跳后最终 path=%r" % r.json()["path"], "| 第二跳带来的 history=%d" % len(r.history))
print(" 重定向前会话里没有 cookie,第二跳靶站看到 Cookie=%r" % r.json()["cookie"])
s3 = Session(cookies={"sid": "keepme"})
print(" 会话里预置 cookie 时,重定向后 Cookie=%r" % s3.get(B + "/r1", timeout=5).json()["cookie"])
s4 = Session(cookies={"sid": "keepme"}, discard_cookies=True)
r4 = s4.get(B + "/r1", timeout=5)
print(" discard_cookies=True 时请求里 Cookie=%r,但会话 jar 里剩下=%r"
% (r4.json()["cookie"], dict(s4.cookies)))
print(" 对照 discard_cookies=False 的会话 jar:", dict(s3.cookies))
s3.close()
s4.close()
s2.close()

print()
print("== 5. HTTP_PROXY / NO_PROXY 环境变量(trust_env)==")
print(" 当前 NO_PROXY=%r" % os.environ.get("NO_PROXY"))
print(" 靶站看到的默认 UA(不 impersonate 时为空):", repr(s.get(B + "/x", timeout=5).json()["ua"]))
os.environ["HTTP_PROXY"] = "http://127.0.0.1:8051"
try:
r = rq.get(B + "/x", timeout=3)
print(" 设了 HTTP_PROXY 后打 127.0.0.1: status=%s(NO_PROXY 命中,代理被跳过)" % r.status_code)
except Exception as e:
print(" 设了 HTTP_PROXY 后打 127.0.0.1: %s" % type(e).__name__)
saved = os.environ.pop("NO_PROXY")
try:
r = rq.get(B + "/x", timeout=3)
print(" 删掉 NO_PROXY 后: status=%s" % r.status_code)
except Exception as e:
print(" 删掉 NO_PROXY 后: %s: %s" % (
type(e).__name__, re.sub(r"\d+ (?:ms|milliseconds)", "<耗时> ms", str(e))[:100]))
s3 = Session(trust_env=False)
try:
r = s3.get(B + "/x", timeout=3)
print(" Session(trust_env=False) 忽略 HTTP_PROXY: status=%s" % r.status_code)
except Exception as e:
print(" Session(trust_env=False): %s" % type(e).__name__)
s3.close()
os.environ["NO_PROXY"] = saved
del os.environ["HTTP_PROXY"]
print(" 环境变量已还原: NO_PROXY=%r HTTP_PROXY=%r"
% (os.environ.get("NO_PROXY"), os.environ.get("HTTP_PROXY")))

print()
print("== 6. requests 的老写法在本家的下场 ==")
print(" Session 有 mount 吗:", hasattr(Session, "mount"),
"| 有 adapters 吗:", hasattr(Session, "adapters"),
"| 有 max_redirects 属性吗:", hasattr(Session(), "max_redirects"))
try:
s.get(B + "/x", hooks={"response": [lambda **kw: None]}, timeout=5)
print(" hooks= 竟然支持")
except TypeError as e:
print(" hooks= -> TypeError: %s" % str(e)[:80])
try:
print(" response.raw 存在:", hasattr(s.get(B + "/x", timeout=5), "raw"))
except Exception as e:
print(" response.raw:", type(e).__name__)
print(" 没有 requests.Session 的 close 语义差异:本家 close() 后立刻 SessionClosed(见 02)")
print(" allow_redirects 还接受字符串/枚举:", end=" ")
print([(x.name, int(x)) for x in CurlFollow])
s2.close()
s.close()
finally:
httpd.shutdown()
httpd.server_close()
print()
print("靶站已 shutdown")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
== 1. 中文 header 值 ==
抛错: builtins.UnicodeEncodeError: 'ascii' codec can't encode characters in position 2-3: ordinal not in range(128)
英文头名+中文值 抛错: UnicodeEncodeError: 'latin-1' codec can't encode character '\u503c' in position 0: ordinal not in range(256)
== 2. URL 中文与 params 的百分号编码 ==
params 里中文 靶站收到 path='/x?q=%E7%94%B5%E5%BD%B1'
params 用 list(同名多值) 靶站收到 path='/x?t=1&t=2'
路径里带中文 靶站收到 path='/%E4%B8%AD%E6%96%87'
quote='' 默认安全字符 靶站收到 path='/x?q=a+b%2Bc%2Fd'
quote=False 不自动编码 curl_cffi.curl.CurlError: Failed to getinfo 1048577, curl: (27) URL rejected: Bad query. See https://curl.
quote=False 但自己编码过 靶站收到 path='/x?q=a%20b+c'
quote='+' 额外编码 + 靶站收到 path='/x?q=a%2Bb'
== 3. multipart:files= 不能用,multipart= 才是正路 ==
files= -> NotImplementedError : files is not supported, use `multipart`. See examples here: https://gi
multipart=dict -> builtins.AttributeError: 'dict' object has no attribute '_form'
CurlMime().addpart + multipart= 上传: multipart/form-data
boundary 长度=46,形状(十六进制位统一打 x)=------------------------xxxxxxxxxxxxxxxxxxxxxx
body_len=202 内容 10 字节 -> 信封开销 192 字节
同一结构换成 20 字节: body_len 202 -> 212 差 10(信封开销与内容长度无关)
两个部分(一个带 filename 一个纯字段): body_len=332 差 130
靶站看到的 Content-Type 头: multipart/form-data; bou ...
== 4. 重定向时 Cookie 是否保留 ==
跟一跳后最终 path='/r2' | 第二跳带来的 history=1
重定向前会话里没有 cookie,第二跳靶站看到 Cookie='from_redirect=yes'
会话里预置 cookie 时,重定向后 Cookie='from_redirect=yes; sid=keepme'
discard_cookies=True 时请求里 Cookie='from_redirect=yes; sid=keepme',但会话 jar 里剩下={'sid': 'keepme'}
对照 discard_cookies=False 的会话 jar: {'sid': 'keepme', 'from_redirect': 'yes'}
== 5. HTTP_PROXY / NO_PROXY 环境变量(trust_env)==
当前 NO_PROXY='localhost,.localhost,127.0.0.1,127.0.0.0/8,[::1],::1'
靶站看到的默认 UA(不 impersonate 时为空): None
设了 HTTP_PROXY 后打 127.0.0.1: status=200(NO_PROXY 命中,代理被跳过)
删掉 NO_PROXY 后: ConnectionError: Failed to perform, curl: (7) Failed to connect to 127.0.0.1:8051 over proxy 127.0.0.1 after <耗时> ms:
Session(trust_env=False) 忽略 HTTP_PROXY: status=200
环境变量已还原: NO_PROXY='localhost,.localhost,127.0.0.1,127.0.0.0/8,[::1],::1' HTTP_PROXY=None
== 6. requests 的老写法在本家的下场 ==
Session 有 mount 吗: False | 有 adapters 吗: False | 有 max_redirects 属性吗: True
hooks= -> TypeError: Session.request() got an unexpected keyword argument 'hooks'
response.raw 存在: False
没有 requests.Session 的 close 语义差异:本家 close() 后立刻 SessionClosed(见 02)
allow_redirects 还接受字符串/枚举: [('ALL', 1), ('OBEYCODE', 2), ('FIRSTONLY', 3), ('SAFE', 4)]
靶站已 shutdown

按「会不会静默改变行为」排序。

  • 中文 header 一定炸:头名或头值里带中文,抛的是标准库的 UnicodeEncodeError('ascii' codec can't encode characters、'latin-1' codec can't encode character '\u503c'),不在 RequestsError 家族里,except RequestException 接不住。requests 会容忍(编码方式不同),这条最容易在上线后突然出现。
  • URL 编码由 quote 控制:params 里的中文会自动百分号编码(靶站收到 ?q=%E7%94%B5%E5%BD%B1),params 传 list 能发同名多值(?t=1&t=2),路径里的中文也会编码;quote=False 时不做编码,含空格的查询串会被 curl 直接拒成 curl: (3)/(27) Bad query。要「原样发出去」就 quote=False,但得自己保证 URL 合法。
  • 上传的写法完全不同:multipart=dict 抛 AttributeError: 'dict' object has no attribute '_form',正路是 CurlMime。实测 boundary 是 46 位十六进制(脚本只报形状与长度,不报原值),信封开销 192 字节且与内容长度无关——同一结构把内容从 10 字节换成 20 字节,body 只从 202 变到 212。
  • 重定向时的 Cookie:服务端在 302 上 Set-Cookie,第二跳靶站就看到 Cookie='from_redirect=yes',即重定向链上的 cookie 会保留。discard_cookies=True 只影响会话 jar(请求里照旧带着),实测第二跳仍看到 from_redirect=yes; sid=keepme,而 jar 里只剩 {'sid': 'keepme'}。
  • 环境变量:HTTP_PROXY 会被默认的 trust_env=True 采用,但本机 NO_PROXY 含 127.0.0.1,所以打本机靶站时被豁免(status=200 且没走代理);trust_env=False 则完全忽略环境变量。requests 的 trust_env 语义在这里等价,但绕过名单来自系统环境,排查时先确认 NO_PROXY。
  • requests 有而 curl_cffi 没有的:Session.mount/adapters(实测都是 False,重试要换 RetryStrategy 或 session.retry)、响应 .raw(False,流式用 iter_content)、hooks= 参数(TypeError)。反过来 curl_cffi 有而 requests 没有的:impersonate、extra_fingerprints、http_version、max_recv_speed、content_callback、quote、referer、discard_cookies、multipart、ws_connect。
  • allow_redirects 除了布尔还接受 CurlFollow 的四个值(ALL=1、OBEYCODE=2、FIRSTONLY=3、SAFE=4),这是 requests 没有的粒度。

命令行入口

装完包,venv 的 Scripts 目录里除了 cffi-gen-src.exe 还有 curl-cffi.exe,它就是 README 里那条命令。python -m curl_cffi 与直接敲 curl-cffi 等价(包里有 curl_cffi/cli/__main__)。这段把四个子命令各跑一次,并让靶站回签「这一枪到底是不是 curl_cffi 发的」:

13_cli.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
# -*- coding: utf-8 -*-
"""取证 13:命令行入口 curl-cffi。

venv 的 Scripts 目录里确实有 curl-cffi.exe,包里有 curl_cffi/cli(含 __main__.py),
所以既能 `python -m curl_cffi` 也能直接 `curl-cffi`。
本脚本自己起 8050 靶站,用子进程发请求,打印:
doctor 的输出、body-only 的响应、以及靶站侧看到的请求头(证明 CLI 真的在走 curl_cffi)。
响应头里有 Date,属于抖动项,所以下面只用 --body / --headers-only 的受控部分。
"""
import json
import os
import re
import socketserver
import subprocess
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


PORT = 8050
SEEN = []


class EchoHandler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"

def log_message(self, *args):
pass

def do_request(self):
n = int(self.headers.get("Content-Length") or 0)
raw = self.rfile.read(n) if n else b""
SEEN.append({
"line": "%s %s" % (self.command, self.path),
"order": list(self.headers.keys()),
"ua": (lambda m: m.group(0) if m else None)(
re.search(r"(Chrome|Firefox|Safari)/[0-9]+", self.headers.get("User-Agent") or "")),
"body_len": len(raw),
"content_type": self.headers.get("Content-Type"),
})
body = json.dumps({"ok": True, "echo": raw.decode("latin-1")[:40]}).encode("utf-8")
try:
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
except OSError:
pass

do_GET = do_request
do_POST = do_request


class StrictServer(ThreadingHTTPServer):
daemon_threads = True
allow_reuse_address = False

def server_bind(self):
socketserver.TCPServer.server_bind(self)


httpd = StrictServer(("127.0.0.1", PORT), EchoHandler)
assert httpd.server_address[1] == PORT
threading.Thread(target=httpd.serve_forever, daemon=True).start()

scripts = os.path.dirname(sys.executable)
EXE = os.path.join(scripts, "curl-cffi.exe")
BASE_CMD = [EXE]
URL = "http://127.0.0.1:%d/cli" % PORT


def run(args, title):
print("== %s ==" % title)
print(" $ curl-cffi %s" % " ".join(args))
p = subprocess.run(BASE_CMD + args, capture_output=True, timeout=60)
out = p.stdout.decode("utf-8", errors="replace")
err = p.stderr.decode("utf-8", errors="replace")
print(" exit=%d" % p.returncode)
for line in out.splitlines():
print(" | " + line)
if err.strip():
print(" stderr:", re.sub(r"\d+ (?:ms|milliseconds)", "<耗时> ms", err.strip())[:200])
return out


try:
print("可执行文件:", os.path.basename(EXE), "存在:", os.path.exists(EXE))
print()
doctor = run(["doctor"], "1. doctor:诊断信息")
print(" doctor 里出现的键:", sorted(
re.sub(r":.*", "", l).strip() for l in doctor.splitlines() if ":" in l))
print()

before = len(SEEN)
run(["get", URL, "--body"], "2. get --body(默认 impersonate=chrome)")
print(" 靶站收到:", json.dumps(SEEN[-1], ensure_ascii=False, indent=1).replace("\n", "\n "))
print()

run(["get", URL, "--impersonate", "firefox133", "--body"], "3. get --impersonate firefox133")
print(" 靶站收到:", json.dumps(SEEN[-1], ensure_ascii=False, indent=1).replace("\n", "\n "))
print()

n_before = len(SEEN)
run(["get", URL, "--impersonate", "none", "--body"], "4. get --impersonate none")
print(" 靶站新收到请求数: %d(报错时根本没发出去)" % (len(SEEN) - n_before))
print()

run(["post", URL, "name=Alice", "X-Test:1", "--form", "--body"], "5. post 带 form 字段与 header")
print(" 靶站收到:", json.dumps(SEEN[-1], ensure_ascii=False, indent=1).replace("\n", "\n "))
print()

http_file = os.path.join(os.environ.get("TEMP", "."), "probe-curlcffi.http")
sep = chr(35) * 3
with open(http_file, "w", encoding="utf-8", newline="\n") as f:
f.write("GET %s\n\n%s\nPOST %s?n=2\nX-Run: 1\n\n{\"a\": 1}\n"
% (URL, sep, URL))
run(["run", http_file, "--body"], "6. run 一个 .http 文件(两条请求共用 session)")
print(" 靶站累计收到:", len(SEEN) - before, "条(来自 .http 文件)")
print(" 最后两条:", json.dumps(SEEN[-2:], ensure_ascii=False, indent=1).replace("\n", "\n "))
os.remove(http_file)
print()

print("== 7. python -m curl_cffi 走的是同一个入口 ==")
p = subprocess.run([sys.executable, "-m", "curl_cffi", "get", URL, "--body"],
capture_output=True, timeout=60)
print(" exit=%d stdout=%r" % (p.returncode, p.stdout.decode("utf-8", errors="replace")[:60]))
print(" 靶站收到:", json.dumps(SEEN[-1], ensure_ascii=False, indent=1).replace("\n", "\n "))
print()

print("== 8. list:离线可列的本机指纹表 ==")
listing = run(["list"], "8a. curl-cffi list")
rows = [l for l in listing.splitlines() if l.startswith("builtin")]
print(" 表格行数(不含表头):", len(rows))
print(" 第一行:", rows[0].strip() if rows else "无")
print(" 最后一条:", rows[-1].strip() if rows else "无")
src_only = sorted({l.split("|")[1].strip() for l in rows})
print(" 名字是否全小写+数字/下划线:", all(re.fullmatch(r"[a-z0-9_]+", n) for n in src_only))
print()

print("== 9. 参数错误与不认识的子命令 ==")
run(["get", "http://127.0.0.1:8051/nope", "--body"], "9a. 连不上的地址")
p = subprocess.run(BASE_CMD + ["bogus"], capture_output=True, timeout=30)
print(" $ curl-cffi bogus -> exit=%d stderr 首行=%r"
% (p.returncode, p.stderr.decode("utf-8", errors="replace").splitlines()[:1]))
print(" CLI 里可用的子命令(从源码 SUPPORTED_METHODS):",
sorted(__import__("curl_cffi.cli", fromlist=["SUPPORTED_METHODS"]).SUPPORTED_METHODS))
print(" README 里的 `curl-cffi update` 要访问 api.impersonate.pro,本机未跑,不做取证。")
finally:
httpd.shutdown()
httpd.server_close()
print()
print("靶站已 shutdown,共收到", len(SEEN), "条请求")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
可执行文件: curl-cffi.exe 存在: True
== 1. doctor:诊断信息 ==
$ curl-cffi doctor
exit=0
| curl-cffi doctor
| ----------------
| python: 3.14.10
| executable: E:\Desktop\Test\curlcffi-env\Scripts\python.exe
| platform: Windows-11-10.0.26300-SP0
| machine: AMD64
| curl_cffi: 0.17.0
| libcurl: libcurl/8.23.0-IMPERSONATE BoringSSL zlib/1.3.1 brotli/1.2.1 zstd/1.5.7 c-ares/1.34.9 WinIDN nghttp2/1.63.0 ngtcp2/1.20.0 nghttp3/1.15.0 curl-impersonate/2.3.1
| api_root: https://api.impersonate.pro/v1
| config_path: C:\Users\Loong\AppData\Roaming\impersonate\config.json
| config_present: False
| api_key_configured: False
| fingerprint_path: C:\Users\Loong\AppData\Roaming\impersonate\fingerprints.json
| fingerprint_present: False
| fingerprint_count: 47
doctor 里出现的键: ['api_key_configured', 'api_root', 'config_path', 'config_present', 'curl_cffi', 'executable', 'fingerprint_count', 'fingerprint_path', 'fingerprint_present', 'libcurl', 'machine', 'platform', 'python']
== 2. get --body(默认 impersonate=chrome) ==
$ curl-cffi get http://127.0.0.1:8050/cli --body
exit=0
| {
| "ok": true,
| "echo": ""
| }
靶站收到: {
"line": "GET /cli",
"order": [
"sec-ch-ua",
"sec-ch-ua-mobile",
"sec-ch-ua-platform",
"upgrade-insecure-requests",
"user-agent",
"accept",
"sec-fetch-site",
"sec-fetch-mode",
"sec-fetch-user",
"sec-fetch-dest",
"accept-encoding",
"accept-language",
"cache-control",
"Host",
"Upgrade",
"HTTP2-Settings",
"Connection"
],
"ua": "Chrome/152",
"body_len": 0,
"content_type": null
}
== 3. get --impersonate firefox133 ==
$ curl-cffi get http://127.0.0.1:8050/cli --impersonate firefox133 --body
exit=0
| {
| "ok": true,
| "echo": ""
| }
靶站收到: {
"line": "GET /cli",
"order": [
"Host",
"Upgrade",
"HTTP2-Settings",
"User-Agent",
"Accept",
"Accept-Language",
"Accept-Encoding",
"Upgrade-Insecure-Requests",
"Sec-Fetch-Dest",
"Sec-Fetch-Mode",
"Sec-Fetch-Site",
"Sec-Fetch-User",
"Priority",
"Te",
"Connection"
],
"ua": "Firefox/133",
"body_len": 0,
"content_type": null
}
== 4. get --impersonate none ==
$ curl-cffi get http://127.0.0.1:8050/cli --impersonate none --body
exit=1
stderr: Error: Impersonating none is not supported
靶站新收到请求数: 0(报错时根本没发出去)
== 5. post 带 form 字段与 header ==
$ curl-cffi post http://127.0.0.1:8050/cli name=Alice X-Test:1 --form --body
exit=0
| {
| "ok": true,
| "echo": "name=Alice"
| }
靶站收到: {
"line": "POST /cli",
"order": [
"sec-ch-ua",
"sec-ch-ua-mobile",
"sec-ch-ua-platform",
"upgrade-insecure-requests",
"user-agent",
"accept",
"sec-fetch-site",
"sec-fetch-mode",
"sec-fetch-user",
"sec-fetch-dest",
"accept-encoding",
"accept-language",
"cache-control",
"Host",
"Upgrade",
"HTTP2-Settings",
"X-Test",
"Content-Length",
"Content-Type",
"Connection"
],
"ua": "Chrome/152",
"body_len": 10,
"content_type": "application/x-www-form-urlencoded"
}
== 6. run 一个 .http 文件(两条请求共用 session) ==
$ curl-cffi run C:\Users\Loong\AppData\Local\Temp\probe-curlcffi.http --body
exit=0
| {
| "ok": true,
| "echo": ""
| }
| {
| "ok": true,
| "echo": "{\"a\": 1}"
| }
stderr: --- [1] GET http://127.0.0.1:8050/cli
--- [2] POST http://127.0.0.1:8050/cli?n=2
靶站累计收到: 5 条(来自 .http 文件)
最后两条: [
{
"line": "GET /cli",
"order": [
"sec-ch-ua",
"sec-ch-ua-mobile",
"sec-ch-ua-platform",
"upgrade-insecure-requests",
"user-agent",
"accept",
"sec-fetch-site",
"sec-fetch-mode",
"sec-fetch-user",
"sec-fetch-dest",
"accept-encoding",
"accept-language",
"cache-control",
"Host",
"Upgrade",
"HTTP2-Settings",
"Connection"
],
"ua": "Chrome/152",
"body_len": 0,
"content_type": null
},
{
"line": "POST /cli?n=2",
"order": [
"sec-ch-ua",
"sec-ch-ua-mobile",
"sec-ch-ua-platform",
"upgrade-insecure-requests",
"user-agent",
"accept",
"sec-fetch-site",
"sec-fetch-mode",
"sec-fetch-user",
"sec-fetch-dest",
"accept-encoding",
"accept-language",
"cache-control",
"Host",
"Upgrade",
"HTTP2-Settings",
"X-Run",
"Content-Type",
"Content-Length",
"Connection"
],
"ua": "Chrome/152",
"body_len": 8,
"content_type": "application/octet-stream"
}
]
== 7. python -m curl_cffi 走的是同一个入口 ==
exit=0 stdout='{\r\n "ok": true,\r\n "echo": ""\r\n}\r\n'
靶站收到: {
"line": "GET /cli",
"order": [
"sec-ch-ua",
"sec-ch-ua-mobile",
"sec-ch-ua-platform",
"upgrade-insecure-requests",
"user-agent",
"accept",
"sec-fetch-site",
"sec-fetch-mode",
"sec-fetch-user",
"sec-fetch-dest",
"accept-encoding",
"accept-language",
"cache-control",
"Host",
"Upgrade",
"HTTP2-Settings",
"Connection"
],
"ua": "Chrome/152",
"body_len": 0,
"content_type": null
}
== 8. list:离线可列的本机指纹表 ==
== 8a. curl-cffi list ==
$ curl-cffi list
exit=0
| type | name | browser | version | os | os_version | h3_fingerprints
| --------+-------------------+---------+---------+---------+------------+----------------
| builtin | chrome100 | Chrome | 100 | Windows | 10 | false
| builtin | chrome101 | Chrome | 101 | Windows | 10 | false
| builtin | chrome104 | Chrome | 104 | Windows | 10 | false
| builtin | chrome107 | Chrome | 107 | Windows | 10 | false
| builtin | chrome110 | Chrome | 110 | Windows | 10 | false
| builtin | chrome116 | Chrome | 116 | Windows | 10 | false
| builtin | chrome119 | Chrome | 119 | macOS | Sonoma | false
| builtin | chrome120 | Chrome | 120 | macOS | Sonoma | false
| builtin | chrome123 | Chrome | 123 | macOS | Sonoma | false
| builtin | chrome124 | Chrome | 124 | macOS | Sonoma | false
| builtin | chrome131 | Chrome | 131 | macOS | Sonoma | false
| builtin | chrome131_android | Chrome | 131 | Android | 14 | false
| builtin | chrome133a | Chrome | 133 | macOS | Sequoia | false
| builtin | chrome136 | Chrome | 136 | macOS | Sequoia | false
| builtin | chrome142 | Chrome | 142 | macOS | Tahoe | false
| builtin | chrome145 | Chrome | 145 | macOS | Tahoe | true
| builtin | chrome146 | Chrome | 146 | macOS | Tahoe | true
| builtin | chrome150 | Chrome | 150 | macOS | Tahoe | true
| builtin | chrome151 | Chrome | 151 | Windows | 10 | true
| builtin | chrome151_macos | Chrome | 151 | macOS | 26.0 | true
| builtin | chrome152 | Chrome | 152 | Windows | 10 | true
| builtin | chrome152_macos | Chrome | 152 | macOS | 26.0 | true
| builtin | chrome99 | Chrome | 99 | Windows | 10 | false
| builtin | chrome99_android | Chrome | 99 | Android | 12 | false
| builtin | edge101 | Edge | 101 | Windows | 10 | false
| builtin | edge99 | Edge | 99 | Windows | 10 | false
| builtin | firefox133 | Firefox | 133.0 | macOS | Sonoma | false
| builtin | firefox135 | Firefox | 135.0 | macOS | Sonoma | false
| builtin | firefox144 | Firefox | 144.0 | macOS | Tahoe | false
| builtin | firefox147 | Firefox | 147.0 | macOS | Tahoe | true
| builtin | firefox153 | Firefox | 153.0 | Windows | 10 | false
| builtin | firefox154 | Firefox | 154.0 | Windows | 10 | false
| builtin | firefox155 | Firefox | 155.0 | Windows | 10 | false
| builtin | safari153 | Safari | 15.3 | macOS | Big Sur | false
| builtin | safari155 | Safari | 15.5 | macOS | Monterey | false
| builtin | safari170 | Safari | 17.0 | macOS | Sonoma | false
| builtin | safari172_ios | Safari | 17.2 | iOS | 17.2 | false
| builtin | safari180 | Safari | 18.0 | macOS | Sequoia | false
| builtin | safari180_ios | Safari | 18.0 | iOS | 18.0 | false
| builtin | safari184 | Safari | 18.4 | macOS | Sequoia | false
| builtin | safari184_ios | Safari | 18.4 | iOS | 18.4 | false
| builtin | safari260 | Safari | 26.0 | macOS | Tahoe | false
| builtin | safari2601 | Safari | 26.0.1 | macOS | Tahoe | false
| builtin | safari260_ios | Safari | 26.0 | iOS | 26.0 | false
| builtin | safari270 | Safari | 27.0 | macOS | 27.0 | true
| builtin | safari270_ios | Safari | 27.0 | iOS | 27.0 | true
| builtin | tor145 | Tor | 14.5 | macOS | Sonoma | false
表格行数(不含表头): 47
第一行: builtin | chrome100 | Chrome | 100 | Windows | 10 | false
最后一条: builtin | tor145 | Tor | 14.5 | macOS | Sonoma | false
名字是否全小写+数字/下划线: True
== 9. 参数错误与不认识的子命令 ==
== 9a. 连不上的地址 ==
$ curl-cffi get http://127.0.0.1:8051/nope --body
exit=1
stderr: Error: Failed to perform, curl: (7) Failed to connect to 127.0.0.1:8051 after <耗时> ms: Could not connect to server. See https://curl.se/libcurl/c/libcurl-errors.html first for more details.
$ curl-cffi bogus -> exit=2 stderr 首行=['usage: curl-cffi [-h]']
CLI 里可用的子命令(从源码 SUPPORTED_METHODS): ['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PATCH', 'POST', 'PUT', 'QUERY', 'TRACE']
README 里的 `curl-cffi update` 要访问 api.impersonate.pro,本机未跑,不做取证。
靶站已 shutdown,共收到 6 条请求
  • doctor 打出 python 版本、libcurl 版本串、平台、api_root=https://api.impersonate.pro/v1、api_key_configured=False、fingerprint_count=47——这五个值就是「本机装好了没有」的判据,遇到指纹不对先跑它。
  • get/post/run 都实测可用,且默认就带 --impersonate chrome:靶站回签的头顺序是 sec-ch-ua 打头、UA 是 Chrome/152,和库内 impersonate="chrome" 完全一致;--body 只回正文(回响应头会带 Date,属于抖动项)。
  • --impersonate none 直接 Error: Impersonating none is not supported,而且请求根本没发出去(靶站新增请求数为 0)——命令行这一层不做静默降级。
  • list 离线可列 47 条内置指纹(表格行数不含表头正好 47)。
  • update 要访问 api.impersonate.pro,本机没配 api_key,未取证,所以这条只写「需要 api_key」,不写它的行为。

实战:抓电影站

把前面几节串起来:先按官方推荐用 impersonate,这台机器上不通就退回默认栈(判定逻辑见「在线站的可达性」),然后抓两页列表 + 三条详情,落 CSV 再回读核对。

15_mini_crawler.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
# -*- coding: utf-8 -*-
"""完整跑一遍:先试 impersonate,不行退回默认栈,然后抓 ssr1 两页列表 + 3 条详情,落 CSV。

本机网络对带指纹的 ClientHello 会丢包(见「在线站的可达性」一节),
所以这段把「先按官方推荐写法用 impersonate,失败再退默认栈」写成一个真的兜底流程。

列的取舍全部以本机实拿为准:
- 片名在 <a href="/detail/N" class="name"> 里套的 <h2>,不是 <h2 class="name">;
- 评分写的是 <p class="score m-t-md m-b-n-sm">9.5</p>,所以按 class="score 前缀取;
- 类型是 <div class="categories"> 里若干个 <button class="el-button category"><span>剧情</span>;
- 地区/时长在第一段 <div class="m-v-sm info"> 的三个 <span> 里,上映日期在第二段;
- 详情页导演是 <h2 class="subtitle">导演</h2> 之后 <div class="director"> 里的 <p class="name">,
不是「导演:xxx」这种文本,所以旧的正则一律取空。
"""

import csv
import os
import re
import tempfile

from curl_cffi import requests

BASE = "https://ssr1.scrape.center"
LIST_FIELDS = ["name", "score", "type", "region", "released", "cover"]
DETAIL_FIELDS = ["导演", "主演"]


def squeeze(s):
return re.sub(r"\s+", " ", s).strip()


def build_session():
"""按官方推荐先要 chrome 指纹;这台机器上会被网络掐掉,就退回默认栈。"""
try:
session = requests.Session(impersonate="chrome", timeout=20)
session.get(BASE + "/page/1")
return session, "impersonate=chrome"
except Exception:
print("impersonate=chrome 走不通,退回默认栈")
return requests.Session(timeout=20), "默认栈"


def parse_list(html):
"""一个卡片一个条目:按 el-card item 切块,块内各字段分别取。"""
rows = []
for chunk in re.split(r'class="el-card item', html)[1:]:
hit = re.search(r'href="(/detail/\d+)" class="name">\s*<h2[^>]*>([^<]+)</h2>', chunk)
if not hit:
continue
row = {"link": hit.group(1), "name": squeeze(hit.group(2))}
score = re.search(r'class="score[^"]*">\s*([0-9.]+)', chunk)
row["score"] = score.group(1) if score else ""
row["type"] = "、".join(
squeeze(x) for x in re.findall(
r'class="el-button category[^"]*"[^>]*>\s*<span>([^<]+)</span>', chunk))
row["region"] = row["released"] = ""
for block in re.findall(r'class="m-v-sm info">(.*?)</div>', chunk, re.S):
spans = [squeeze(x) for x in re.findall(r"<span[^>]*>([^<]*)</span>", block)]
spans = [x for x in spans if x and x != "/"]
if len(spans) >= 2 and spans[-1].endswith("分钟"):
row["region"] = spans[0]
for x in spans:
day = re.search(r"(\d{4}-\d{2}-\d{2}) 上映", x)
if day:
row["released"] = day.group(1)
img = re.search(r'src="(https?://[^"]+)"[^>]*class="cover"', chunk, re.S)
row["cover"] = img.group(1) if img else ""
rows.append(row)
return rows


def parse_detail(html):
"""详情页的导演/演员:小标题之后那一段里的 <p class="... name ...">。"""
out = {}
for label, key in [("导演", "导演"), ("演员", "主演")]:
sec = re.search(r'class="subtitle">%s</h2>(.*?)(?=<h2[^>]*class="subtitle"|\Z)' % label,
html, re.S)
names = []
if sec:
names = [squeeze(x) for x in re.findall(
r'<p[^>]*class="[^"]*\bname\b[^"]*"[^>]*>([^<]+)</p>', sec.group(1))]
out[key] = "、".join(names)
return out


session, how = build_session()
print("这一趟用的取法 = " + how)

rows = []
for page in (1, 2):
resp = session.get(BASE + "/page/%d" % page)
got = parse_list(resp.text)
print("第 %d 页 状态 %s 抓到 %d 条" % (page, resp.status_code, len(got)))
rows.extend(got)

for row in rows:
for key in DETAIL_FIELDS:
row[key] = ""
print("列表页字段每列非空的行数(共 %d 行)= %s"
% (len(rows), " | ".join("%s %d" % (k, sum(1 for r in rows if r[k])) for k in LIST_FIELDS)))

links = [r["link"] for r in rows][:3]
print("前 3 条详情链接 =", links)
for row, link in zip(rows, links):
row.update(parse_detail(session.get(BASE + link).text))
print("补详情之后 导演非空 %d 行 | 主演非空 %d 行"
% (sum(1 for r in rows if r["导演"]), sum(1 for r in rows if r["主演"])))

fields = LIST_FIELDS + DETAIL_FIELDS
path = os.path.join(tempfile.gettempdir(), "curl_cffi_mini_crawler_%s.csv" % os.getpid())
with open(path, "w", encoding="utf-8-sig", newline="") as fp:
writer = csv.DictWriter(fp, fieldnames=fields, extrasaction="ignore")
writer.writeheader()
writer.writerows(rows)
with open(path, encoding="utf-8-sig", newline="") as fp:
back = list(csv.DictReader(fp))
os.remove(path)

print("CSV 回读行数 = %d 列名 = %s" % (len(back), list(back[0])))
print("第一条片名 =", back[0]["name"])
print("最后一条片名 =", back[-1]["name"])
session.close()
1
2
3
4
5
6
7
8
9
10
impersonate=chrome 走不通,退回默认栈
这一趟用的取法 = 默认栈
第 1 页 状态 200 抓到 10 条
第 2 页 状态 200 抓到 10 条
列表页字段每列非空的行数(共 20 行)= name 20 | score 20 | type 20 | region 20 | released 19 | cover 20
前 3 条详情链接 = ['/detail/1', '/detail/2', '/detail/3']
补详情之后 导演非空 3 行 | 主演非空 3 行
CSV 回读行数 = 20 列名 = ['name', 'score', 'type', 'region', 'released', 'cover', '导演', '主演']
第一条片名 = 霸王别姬 - Farewell My Concubine
最后一条片名 = 迁徙的鸟 - The Travelling Birds

这段值得注意的四个地方,都是实测里踩出来的。

  • 兜底不是可选项。带 impersonate 的取样结果在 Timeout、SSLError、偶尔 200 之间跳(上一节那 20 次取样),所以「先按官方推荐试 impersonate → 失败退回默认栈 → 打印最终走了哪条路」必须写成代码的一部分。还有一个细节:兜底那一行只打印「走不通」,不打印异常名——异常名本身就是抖的,打出来这块就没法两遍一致了。
  • 列表页其实有评分,但要按前缀匹配。评分写在 <p class="score m-t-md m-b-n-sm">9.5</p> 里,按精确串 class="score" 找命中 0,按 class="score[^"]*" 找命中 10/页。这类「class 后面还跟着别的类名」的写法在 Element UI 生成的页面里到处都是,class="name" 同理(片名在 <a ... class="name"> 里套 <h2 class="m-b-sm">,写 <h2 class="name"> 一个都取不到)。
  • 详情页的导演/演员不是「导演:xxx」这种文本(导演: 在详情页命中 0)。真实结构是 <h2 class="subtitle">导演</h2> → <div class="directors"> → <div class="director ..."> → <p class="name text-center ...">陈凯歌</p>,演员段则是 <p class="el-tooltip name ...">张国荣</p>,而且旁边还有 class="... role ..." 的「饰:程蝶衣」要靠 class 里的 name 词边界排除掉。做法是先按 class="subtitle">导演</h2> 切到下一个 subtitle 为止,再在段内取 <p ... class="[^"]*name[^"]*">…</p>。
  • 一个自己写出来的 bug:re.findall(r'href="(/detail/\d+)"', html) 会把封面那个 <a href="/detail/1" class=""> 和片名那个 <a ... class="name"> 各算一次,页 1 命中 20 次,zip 到 20 行上就变成 ['/detail/1', '/detail/1', '/detail/2'],白抓重复详情。改成在切条目时就把 link 一起解析出来,前 3 条才是 ['/detail/1', '/detail/2', '/detail/3']。
  • 输出只留结构指标:每列非空的行数(released 是 19/20,有 1 部片列表页确实不给上映日期,空值如实保留)、详情链接、CSV 回读的行数与列名、首尾各一条片名。字节数、耗时、封面长地址这些会变的全部不打印,CSV 落盘文件名带上 os.getpid() 以免并发互踩。

与 requests 抓同一页

同一页在线站、两种客户端,把「换库会不会改变服务端返回的内容」这件事钉死。先看 requests 侧的基线:

same_page_requests.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# -*- coding: utf-8 -*-
"""同一个在线靶站在 requests 下的基线:状态码、字节数、能数出几个片名。"""

import re

import requests

URL = "https://ssr1.scrape.center/page/1"
resp = requests.get(URL, timeout=30, headers={"User-Agent": "Mozilla/5.0"})
html = resp.text
names = re.findall(r'class="name">\s*<h2[^>]*>([^<]+)', html)
print("requests 状态", resp.status_code)
print("字节数", len(resp.content))
print("编码", resp.encoding, "| Content-Type", resp.headers.get("Content-Type"))
print("片名个数", len(names))
print("前 3 个", names[:3])
1
2
3
4
5
requests 状态 200
字节数 41667
编码 utf-8 | Content-Type text/html; charset=utf-8
片名个数 10
前 3 个 ['霸王别姬 - Farewell My Concubine', '这个杀手不太冷 - Léon', '肖申克的救赎 - The Shawshank Redemption']

再用 curl_cffi 抓同一页,取法与打印口径完全一样:

same_page_curl_cffi.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# -*- coding: utf-8 -*-
"""同一个在线靶站在 curl_cffi 下的表现,输出口径与 requests 基线保持一致。"""

import re

from curl_cffi import requests

URL = "https://ssr1.scrape.center/page/1"
resp = requests.get(URL, timeout=30, headers={"User-Agent": "Mozilla/5.0"})
html = resp.text
names = re.findall(r'class="name">\s*<h2[^>]*>([^<]+)', html)
print("curl_cffi 状态", resp.status_code)
print("字节数", len(resp.content))
print("编码", resp.encoding, "| Content-Type", resp.headers.get("Content-Type"))
print("片名个数", len(names))
print("前 3 个", names[:3])
1
2
3
4
5
curl_cffi 状态 200
字节数 41667
编码 utf-8 | Content-Type text/html; charset=utf-8
片名个数 10
前 3 个 ['霸王别姬 - Farewell My Concubine', '这个杀手不太冷 - Léon', '肖申克的救赎 - The Shawshank Redemption']

两份输出除了第一行的库名,其余逐字相同:状态 200、resp.content 41667 字节、Content-Type: text/html; charset=utf-8(所以 encoding 直接是 utf-8,不需要 apparent_encoding)、片名 10 个、前三个是「霸王别姬 / 这个杀手不太冷 / 肖申克的救赎」。

这条结论的用意是:换 TLS/HTTP 指纹不会改变服务端返回的正文。指纹是用来骗过「你是不是浏览器」这一层检测的,不是用来改变渲染结果的。谁要是发现换了 impersonate 之后正文不一样,那多半是命中了不同的站点分支(比如按 UA 返回不同内容),不是 curl_cffi 在改你的 HTML。

顺带把本篇的定位说清楚,免得和系列其它篇混:

  • 只要 HTTP 客户端、要浏览器级指纹 → curl_cffi(本篇),它不开浏览器,成本低。
  • 要真的执行 JS、点按钮、处理 iframe → Selenium(见《爬虫-Selenium和PhantomJS》)或 DrissionPage(见《爬虫-DrissionPage》)。
  • 要「换指纹但仍只发 HTTP 请求」的高并发抓取 → curl_cffi 配 AsyncSession(max_clients=N);verify_ssl=0 这类证书开关是整数位标志(SSLVerify 枚举),requests 那种 verify="/路径/CA.pem" 的写法在这里不行,实测见「参数面」一节。

常见坑速查

把上面各节里最容易踩的收成一张表,左列是现象,右列是本篇里对应的小节。

现象 真正的原因 出处
带 impersonate 就超时 本机网络丢弃带指纹的 ClientHello 在线站的可达性
BrowserType.chrome 报 AttributeError 别名只在 BrowserTypeLiteral(63 个取值)里,枚举(53 个成员)不含别名 装进来有什么
__libcurl_version__ 拿到空值 本机版本里这个名字不存在,特性串在 __curl_version__ 装进来有什么
抓几次都是新连接、Cookie 为空 顶层 get() 每次新建 Session 三种用法
莫名出现 Timeout Session 默认 timeout=30,requests 是 None 参数面
files= 抛 NotImplementedError 上传改走 multipart=CurlMime(...) 参数面、从 requests 迁移
except RequestsError 接不住 抛的是 CurlError/UnicodeEncodeError/JSONDecodeError 异常体系、WebSocket、从 requests 迁移
gopher:// 不报错但拿不到东西 scheme 不认识时直接返回 status_code=0 异常体系
配了代理却没走代理 trust_env=True 读到 NO_PROXY 里的 127.0.0.1 代理:两种写法
想重试挂 HTTPAdapter 没有 mount/adapters,只有 RetryStrategy 从 requests 迁移
并发度数字乱跳 响应太快没拉开并发;要靠慢端点量峰值在途 下载与并发
headers 里同名头只拿到一个值 并成 'one, two',且没有 getlist() Response 对象
指纹串 ja3= 报 Cipher not found 十进制 ja3 串这条路本机走不通 自定义指纹、TLS 指纹怎么看

参考