返回 last30days-skill
test_chrome_cdp.py
根目录 / tests / test_chrome_cdp.py
1 """U6/U5: live Chrome cookie reader over CDP (lib/chrome_cdp.py).
2
3 Unit tests for port derivation and cookie filtering, plus an end-to-end test
4 against a FAKE CDP server (stdlib socket) that exercises the real HTTP target
5 lookup and the hand-rolled RFC 6455 websocket client. Only obvious dummy
6 cookie values are used (test-auth-token / test-ct0).
7 """
8
9 import base64
10 import hashlib
11 import json
12 import socket
13 import struct
14 import threading
15 from unittest import mock
16
17 from lib import chrome_cdp
18
19 _WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
20
21
22 # --- Unit: endpoint derivation + cookie filtering --------------------------
23
24
25 def test_display_number_parsing():
26 with mock.patch.dict("os.environ", {"DISPLAY": ":99"}, clear=False):
27 assert chrome_cdp._display_number() == 99
28 with mock.patch.dict("os.environ", {"DISPLAY": "localhost:10.0"}, clear=False):
29 assert chrome_cdp._display_number() == 10
30
31
32 def test_candidate_endpoints_default_order_18800_then_display():
33 with mock.patch.dict("os.environ", {"DISPLAY": ":7"}, clear=False):
34 # No BROWSER_CDP_URL in env for this check.
35 with mock.patch.dict("os.environ", {"BROWSER_CDP_URL": ""}, clear=False):
36 endpoints = chrome_cdp.candidate_endpoints({})
37 assert endpoints == ["http://127.0.0.1:18800", "http://127.0.0.1:9229"]
38
39
40 def test_candidate_endpoints_prefers_browser_cdp_url_exclusively():
41 endpoints = chrome_cdp.candidate_endpoints({"BROWSER_CDP_URL": "http://127.0.0.1:5555"})
42 assert endpoints == ["http://127.0.0.1:5555"]
43
44
45 def test_pair_from_cookies_requires_both_and_filters_domain():
46 cookies = [
47 {"name": "auth_token", "value": "test-auth-token", "domain": ".x.com"},
48 {"name": "ct0", "value": "test-ct0", "domain": ".x.com"},
49 {"name": "auth_token", "value": "someone-else", "domain": ".example.com"},
50 ]
51 pair = chrome_cdp._pair_from_cookies(cookies)
52 assert pair == {"auth_token": "test-auth-token", "ct0": "test-ct0"}
53
54
55 def test_pair_from_cookies_half_pair_is_incomplete():
56 cookies = [{"name": "auth_token", "value": "test-auth-token", "domain": ".x.com"}]
57 pair = chrome_cdp._pair_from_cookies(cookies)
58 assert "ct0" not in pair
59
60
61 def test_pair_from_cookies_never_mixes_hosts():
62 """A lookalike host (notx.com) must not contribute; the later same-host
63 x.com pair wins, never a cross-host mix (P1)."""
64 cookies = [
65 {"name": "auth_token", "value": "notx-token", "domain": "notx.com"},
66 {"name": "ct0", "value": "test-ct0", "domain": ".x.com"},
67 {"name": "auth_token", "value": "test-auth-token", "domain": ".x.com"},
68 ]
69 pair = chrome_cdp._pair_from_cookies(cookies)
70 assert pair == {"auth_token": "test-auth-token", "ct0": "test-ct0"}
71 assert pair["auth_token"] != "notx-token"
72
73
74 def test_pair_from_cookies_prefers_x_com_over_twitter():
75 cookies = [
76 {"name": "auth_token", "value": "tw-token", "domain": ".twitter.com"},
77 {"name": "ct0", "value": "tw-ct0", "domain": ".twitter.com"},
78 {"name": "auth_token", "value": "test-auth-token", "domain": ".x.com"},
79 {"name": "ct0", "value": "test-ct0", "domain": ".x.com"},
80 ]
81 pair = chrome_cdp._pair_from_cookies(cookies)
82 assert pair == {"auth_token": "test-auth-token", "ct0": "test-ct0"}
83
84
85 def test_pair_from_cookies_skips_partial_host_for_complete_one():
86 """x.com has only auth_token; twitter.com has both -> the twitter pair wins
87 (same-host), never x.com's auth_token merged with twitter's ct0."""
88 cookies = [
89 {"name": "auth_token", "value": "x-only-token", "domain": ".x.com"},
90 {"name": "auth_token", "value": "test-auth-token", "domain": ".twitter.com"},
91 {"name": "ct0", "value": "test-ct0", "domain": ".twitter.com"},
92 ]
93 pair = chrome_cdp._pair_from_cookies(cookies)
94 assert pair == {"auth_token": "test-auth-token", "ct0": "test-ct0"}
95
96
97 def test_pair_from_cookies_does_not_mix_across_paths():
98 """auth_token on path / and ct0 on path /i are different scopes -> no pair."""
99 cookies = [
100 {"name": "auth_token", "value": "test-auth-token", "domain": ".x.com", "path": "/"},
101 {"name": "ct0", "value": "test-ct0", "domain": ".x.com", "path": "/i"},
102 ]
103 pair = chrome_cdp._pair_from_cookies(cookies)
104 assert set(pair) != {"auth_token", "ct0"} # never a cross-path pair
105
106
107 def test_pair_from_cookies_does_not_mix_across_partitions():
108 """Same host+path but different partitionKey -> different scopes -> no pair."""
109 cookies = [
110 {"name": "auth_token", "value": "test-auth-token", "domain": ".x.com",
111 "path": "/", "partitionKey": "https://a.example"},
112 {"name": "ct0", "value": "test-ct0", "domain": ".x.com",
113 "path": "/", "partitionKey": "https://b.example"},
114 ]
115 pair = chrome_cdp._pair_from_cookies(cookies)
116 assert set(pair) != {"auth_token", "ct0"}
117
118
119 def test_pair_from_cookies_prefers_unpartitioned_root_path_scope():
120 """A complete unpartitioned path-/ pair wins over later partitioned or
121 other-path complete pairs (object-form partitionKey must not crash)."""
122 cookies = [
123 # other-path complete pair
124 {"name": "auth_token", "value": "ipath-token", "domain": ".x.com", "path": "/i"},
125 {"name": "ct0", "value": "ipath-ct0", "domain": ".x.com", "path": "/i"},
126 # partitioned complete pair (object form)
127 {"name": "auth_token", "value": "part-token", "domain": ".x.com", "path": "/",
128 "partitionKey": {"topLevelSite": "https://x.com", "hasCrossSiteAncestor": True}},
129 {"name": "ct0", "value": "part-ct0", "domain": ".x.com", "path": "/",
130 "partitionKey": {"topLevelSite": "https://x.com", "hasCrossSiteAncestor": True}},
131 # the winner: unpartitioned, path /
132 {"name": "auth_token", "value": "test-auth-token", "domain": ".x.com", "path": "/"},
133 {"name": "ct0", "value": "test-ct0", "domain": ".x.com", "path": "/"},
134 ]
135 pair = chrome_cdp._pair_from_cookies(cookies)
136 assert pair == {"auth_token": "test-auth-token", "ct0": "test-ct0"}
137
138
139 def test_from_browser_off_skips_endpoints():
140 with mock.patch.object(
141 chrome_cdp, "candidate_endpoints", side_effect=AssertionError("must not probe")
142 ):
143 assert chrome_cdp.read_x_cookies({"FROM_BROWSER": "off"}) is None
144
145
146 # --- Fake CDP server (stdlib socket) ---------------------------------------
147
148
149 class _FakeCDPServer:
150 """Minimal Chrome-debug endpoint: GET /json/version + /json + a CDP websocket."""
151
152 def __init__(self, cookies, browser="Chrome/120.0.0.0"):
153 self._cookies = cookies
154 self._browser = browser
155 self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
156 self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
157 self._sock.bind(("127.0.0.1", 0))
158 self._sock.listen(8)
159 self._sock.settimeout(5)
160 self.port = self._sock.getsockname()[1]
161 self._stop = False
162 self._thread = threading.Thread(target=self._serve, daemon=True)
163
164 def __enter__(self):
165 self._thread.start()
166 return self
167
168 def __exit__(self, *exc):
169 self._stop = True
170 try:
171 self._sock.close()
172 except OSError:
173 pass
174
175 def _serve(self):
176 while not self._stop:
177 try:
178 conn, _ = self._sock.accept()
179 except OSError:
180 return
181 try:
182 self._handle(conn)
183 except OSError:
184 pass
185 finally:
186 try:
187 conn.close()
188 except OSError:
189 pass
190
191 def _read_request(self, conn):
192 buf = b""
193 conn.settimeout(5)
194 while b"\r\n\r\n" not in buf:
195 chunk = conn.recv(4096)
196 if not chunk:
197 return None, b""
198 buf += chunk
199 head, _, rest = buf.partition(b"\r\n\r\n")
200 return head.decode("latin-1"), rest
201
202 def _handle(self, conn):
203 head, rest = self._read_request(conn)
204 if head is None:
205 return # bare reachability probe (connect then close)
206 if "upgrade: websocket" in head.lower():
207 self._handle_ws(conn, head, rest)
208 return
209 request_line = head.splitlines()[0]
210 if request_line.startswith("GET /json/version"):
211 body = json.dumps({"Browser": self._browser, "Protocol-Version": "1.3"}).encode()
212 self._send_http_json(conn, body)
213 elif request_line.startswith("GET /json"):
214 body = json.dumps([
215 {
216 "type": "page",
217 "webSocketDebuggerUrl": f"ws://127.0.0.1:{self.port}/devtools/page/ABC",
218 }
219 ]).encode()
220 self._send_http_json(conn, body)
221
222 def _send_http_json(self, conn, body):
223 conn.sendall(
224 b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n"
225 b"Content-Length: " + str(len(body)).encode()
226 + b"\r\nConnection: close\r\n\r\n" + body
227 )
228
229 def _handle_ws(self, conn, head, rest):
230 key = ""
231 for line in head.splitlines():
232 if line.lower().startswith("sec-websocket-key:"):
233 key = line.split(":", 1)[1].strip()
234 accept = base64.b64encode(
235 hashlib.sha1((key + _WS_GUID).encode()).digest()
236 ).decode()
237 conn.sendall(
238 b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\n"
239 b"Connection: Upgrade\r\nSec-WebSocket-Accept: "
240 + accept.encode() + b"\r\n\r\n"
241 )
242 buf = bytearray(rest)
243 # Respond to each client command until getAllCookies has been answered.
244 for _ in range(4):
245 msg, buf = self._read_frame(conn, buf)
246 if msg is None:
247 return
248 try:
249 data = json.loads(msg)
250 except json.JSONDecodeError:
251 continue
252 mid = data.get("id")
253 if data.get("method") == "Network.getAllCookies":
254 self._send_frame(conn, json.dumps(
255 {"id": mid, "result": {"cookies": self._cookies}}
256 ))
257 return
258 self._send_frame(conn, json.dumps({"id": mid, "result": {}}))
259
260 def _read_frame(self, conn, buf):
261 def need(n):
262 nonlocal buf
263 while len(buf) < n:
264 chunk = conn.recv(4096)
265 if not chunk:
266 return False
267 buf += chunk
268 return True
269
270 if not need(2):
271 return None, buf
272 length = buf[1] & 0x7F
273 masked = buf[1] & 0x80
274 idx = 2
275 if length == 126:
276 if not need(4):
277 return None, buf
278 length = struct.unpack(">H", buf[2:4])[0]
279 idx = 4
280 elif length == 127:
281 if not need(10):
282 return None, buf
283 length = struct.unpack(">Q", buf[2:10])[0]
284 idx = 10
285 mask = b""
286 if masked:
287 if not need(idx + 4):
288 return None, buf
289 mask = bytes(buf[idx:idx + 4])
290 idx += 4
291 if not need(idx + length):
292 return None, buf
293 payload = bytes(buf[idx:idx + length])
294 if masked:
295 payload = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
296 del buf[:idx + length]
297 return payload.decode("utf-8"), buf
298
299 def _send_frame(self, conn, text):
300 payload = text.encode("utf-8")
301 header = bytearray([0x81]) # FIN + text, unmasked (server)
302 length = len(payload)
303 if length < 126:
304 header.append(length)
305 elif length < 65536:
306 header.append(126)
307 header += struct.pack(">H", length)
308 else:
309 header.append(127)
310 header += struct.pack(">Q", length)
311 conn.sendall(bytes(header) + payload)
312
313
314 def test_read_x_cookies_via_fake_cdp():
315 cookies = [
316 {"name": "auth_token", "value": "test-auth-token", "domain": ".x.com"},
317 {"name": "ct0", "value": "test-ct0", "domain": ".x.com"},
318 {"name": "guest_id", "value": "irrelevant", "domain": ".x.com"},
319 ]
320 with _FakeCDPServer(cookies) as server:
321 config = {"BROWSER_CDP_URL": f"http://127.0.0.1:{server.port}"}
322 result = chrome_cdp.read_x_cookies(config)
323 assert result == {"auth_token": "test-auth-token", "ct0": "test-ct0"}
324
325
326 def test_read_x_cookies_incomplete_pair_returns_none():
327 cookies = [{"name": "auth_token", "value": "test-auth-token", "domain": ".x.com"}]
328 with _FakeCDPServer(cookies) as server:
329 config = {"BROWSER_CDP_URL": f"http://127.0.0.1:{server.port}"}
330 result = chrome_cdp.read_x_cookies(config)
331 assert result is None
332
333
334 def test_read_x_cookies_rejects_node_inspector():
335 """A Node --inspect endpoint (Browser=node.js/...) is not Chrome -> None."""
336 cookies = [
337 {"name": "auth_token", "value": "test-auth-token", "domain": ".x.com"},
338 {"name": "ct0", "value": "test-ct0", "domain": ".x.com"},
339 ]
340 with _FakeCDPServer(cookies, browser="node.js/v20.0.0") as server:
341 config = {"BROWSER_CDP_URL": f"http://127.0.0.1:{server.port}"}
342 result = chrome_cdp.read_x_cookies(config)
343 assert result is None
344
345
346 def test_read_x_cookies_from_browser_off_opens_no_socket():
347 with mock.patch("socket.create_connection", side_effect=AssertionError("no socket")):
348 with mock.patch("urllib.request.urlopen", side_effect=AssertionError("no http")):
349 assert chrome_cdp.read_x_cookies({"FROM_BROWSER": "off"}) is None
350
351
352 def test_read_x_cookies_rejects_wss_without_plaintext_connect():
353 """wss:// (TLS) is refused; the plaintext client must not connect (P2)."""
354 with (
355 mock.patch("socket.create_connection", side_effect=AssertionError("no plaintext connect to TLS endpoint")),
356 mock.patch("urllib.request.urlopen", side_effect=AssertionError("no http probe of TLS endpoint")),
357 ):
358 assert chrome_cdp.read_x_cookies({"BROWSER_CDP_URL": "wss://127.0.0.1:9222/devtools/page/ABC"}) is None
359
360
361 def test_read_x_cookies_rejects_https_base_without_connect():
362 """An https:// debug base (would yield wss) is refused without probing."""
363 with (
364 mock.patch("socket.create_connection", side_effect=AssertionError("no connect")),
365 mock.patch("urllib.request.urlopen", side_effect=AssertionError("no TLS http probe")),
366 ):
367 assert chrome_cdp.read_x_cookies({"BROWSER_CDP_URL": "https://127.0.0.1:18800"}) is None
368
369
370 def test_wsconn_connect_refuses_wss():
371 assert chrome_cdp._WSConn.connect("wss://127.0.0.1:9222/devtools/page/ABC", 1.0) is None
372
373
374 def test_read_x_cookies_no_reachable_endpoint_returns_none():
375 # A port with nothing listening: connection refused, returns None.
376 with socket.socket() as s:
377 s.bind(("127.0.0.1", 0))
378 dead_port = s.getsockname()[1]
379 config = {"BROWSER_CDP_URL": f"http://127.0.0.1:{dead_port}"}
380 assert chrome_cdp.read_x_cookies(config) is None
381
381 lines PYTHON