Phone Chrome CDP
通过 ADB 连接手机 Chrome DevTools Protocol,用原始 WebSocket 控制浏览器。
Step 1: Setup
启动 Chrome 并转发端口:
adb shell am start -n com.android.chrome/com.google.android.apps.chrome.Main
sleep 3
adb forward --remove tcp:9222 2>/dev/null
adb forward tcp:9222 localabstract:chrome_devtools_remote
验证:curl -s http://localhost:9222/json/version
获取 tab 列表:curl -s http://localhost:9222/json/list
Step 2: CDPClient Class
关键点:手动构造 WebSocket 帧,不加 Origin 头(Chrome Android 会拒绝带 Origin 的连接)。
import socket, base64, os, struct, json, time
class CDPClient:
def __init__(self, host, port, tab_id):
self.sock = socket.create_connection((host, port), timeout=10)
key = base64.b64encode(os.urandom(16)).decode()
path = "/devtools/page/" + str(tab_id)
req = "GET " + path + " HTTP/1.1\r\nHost: " + host + ":" + str(port) + "\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: " + key + "\r\nSec-WebSocket-Version: 13\r\n\r\n"
self.sock.sendall(req.encode())
resp = b""
while b"\r\n\r\n" not in resp:
resp += self.sock.recv(4096)
self._msg_id = 0
def send(self, method, params=None):
self._msg_id += 1
cmd = {"id": self._msg_id, "method": method}
if params:
cmd["params"] = params
data = json.dumps(cmd).encode()
mask = os.urandom(4)
frame = bytearray([0x81])
length = len(data)
if length < 126:
frame.append(0x80 | length)
elif length < 65536:
frame.append(0x80 | 126)
frame.extend(struct.pack(">H", length))
frame.extend(mask)
frame.extend(b ^ mask[i % 4] for i, b in enumerate(data))
self.sock.sendall(bytes(frame))
return self._msg_id
def recv_until(self, target_id, timeout=10):
deadline = time.time() + timeout
while time.time() < deadline:
self.sock.settimeout(max(0.1, deadline - time.time()))
try:
header = self.sock.recv(2)
if len(header) < 2:
continue
length = header[1] & 0x7f
if length == 126:
length = struct.unpack(">H", self.sock.recv(2))[0]
if header[1] & 0x80:
self.sock.recv(4)
data = bytearray()
while len(data) < length:
chunk = self.sock.recv(length - len(data))
if not chunk:
break
data.extend(chunk)
msg = json.loads(data.decode())
if msg.get("id") == target_id:
return msg
except socket.timeout:
continue
return None
def drain(self, timeout=1):
deadline = time.time() + timeout
while time.time() < deadline:
self.sock.settimeout(0.1)
try:
self.sock.recv(4096)
except:
pass
def eval_js(self, expression):
msg_id = self.send("Runtime.evaluate", {"expression": expression, "returnByValue": True})
resp = self.recv_until(msg_id)
if resp and "result" in resp:
return resp["result"].get("result", {}).get("value")
return None
def navigate(self, url):
self.send("Page.navigate", {"url": url})
time.sleep(3)
def close(self):
self.sock.close()
Step 3: Usage Examples
cdp = CDPClient("localhost", 9222, tab_id=110)
# 读页面信息
title = cdp.eval_js("document.title")
url = cdp.eval_js("window.location.href")
# 导航
cdp.navigate("https://www.baidu.com")
# 注入 JS 搜索
cdp.eval_js("document.querySelector('#kw').value = '美食'")
cdp.eval_js("document.querySelector('#su').click()")
time.sleep(4)
# 读搜索结果
results_json = cdp.eval_js("JSON.stringify(Array.from(document.querySelectorAll('h3 a')).slice(0,5).map(a=>({title:a.textContent,href:a.href})))")
# 读 Cookie(含 httpOnly)
cdp.send("Network.enable")
cdp.drain(0.5)
msg_id = cdp.send("Network.getAllCookies")
all_cookies = cdp.recv_until(msg_id)
cookie_count = len(all_cookies["result"]["cookies"])
# 写 Cookie
cdp.eval_js("document.cookie = 'test=hello; path=/'")
# 滚动
cdp.eval_js("window.scrollBy(0, 500)")
# 读页面文本
text = cdp.eval_js("document.body.innerText.substring(0, 2000)")
cdp.close()
截图(非 CDP)
用 adb screencap:
adb shell input keyevent KEYCODE_WAKEUP
adb shell screencap -p /sdcard/screen.png
adb pull /sdcard/screen.png ./screen.png
LAN 共享
当前机器当跳板,用端口转发工具把 localhost:9222 暴露给局域网。其他机器连接 your_ip:转发端口。
已验证
- 导航 / 读标题URL / 注入JS / 读写Cookie / 点击 / 滚动 / 读DOM
- 截图通过 screencap(非 CDP 原生)
坑
- websocket-client 库连不上(403):因为自动加 Origin 头。用 CDPClient 手动构造帧
- /json/list 超时:重启 Chrome
- screencap 黑屏:先唤醒屏幕
- Chrome 重启后 tab ID 变:重新查 tab 列表