host
179a86d655
Playwright 1.64.1 的 Page 根本没有 accessibility 属性（那是 Puppeteer 的 API），open/snapshot 在真实库上必报 'Page' object has no attribute 'accessibility'。原生快照是 Page.aria_snapshot()（返回 ARIA 快照字符串），a11y 字段仍叫 a11y，值改为字符串；截断改字符串版 truncate_aria（200 节点 行/64KB）。FakePage 同步改真形状（旧 fake 自带 .accessibility 与真实库分叉，是与 sync_playwright bug 同款模式）。 wheel 1.64.1 实测：sync_api/_generated.py 无 def accessibility， Page 与 Locator 均有 aria_snapshot() -> str。
309 lines
12 KiB
Python
309 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""chv-browser：浏览器子代理（客机侧、单元外独立进程）。
|
||
|
||
对齐 Muse P2（见 docs/chv-muse-p2-gap.md）：浏览器任务由专门子代理
|
||
完成，与 codex 单元分离。子代理只向单元返回页面无障碍快照
|
||
（Playwright Page.aria_snapshot() 的 ARIA 快照字符串），不见原始
|
||
DOM，不能在页面上下文跑 JS，不暴露 evaluate/content，不开 DevTools。
|
||
|
||
操作（Unix socket /run/chv/browser.sock，JSON 行）：
|
||
open {url} -> 导航并返回 {title, url, a11y, truncated}
|
||
snapshot {} -> 当前页的 {title, url, a11y, truncated}
|
||
screenshot {} -> 截图落盘 /run/chv/shots/<uuid>.png，返回 {path}
|
||
close {} -> 关当前页（浏览器进程继续跑，等下一次 open）
|
||
ping {} -> {"ok": true}
|
||
|
||
URL 只允许 http/https；localhost 与私有地址字面量直接拒绝（主机名
|
||
判定，不做 DNS 解析——DNS 重绑定防护是 P3 Sentinel 的事，这里不
|
||
冒充已做）。无障碍树截断到 200 节点 / 64KB，超了标 truncated=true，
|
||
绝不把整棵树无上限灌进 socket。
|
||
|
||
playwright 是懒加载：import 本模块不需要 playwright 已装（单测
|
||
在无 playwright 环境跑）；真正 open 时才加载，缺失即报错。
|
||
|
||
只用标准库 + playwright（运行时）。运行身份：系统用户 chv-browser，
|
||
socket 0600（broker 以 root 连，不受模式限制）。
|
||
"""
|
||
import json
|
||
import os
|
||
import re
|
||
import socket
|
||
import sys
|
||
import uuid
|
||
|
||
# round-11 真机修复（缺陷 4）：自有目录 /run/chv-browser，
|
||
# 0755 chv-browser 所有——/run/chv 是 0755 root root，
|
||
# 在此 bind 会 EACCES 而无限 crash loop（见 chv-broker.py）。
|
||
BROWSER_SOCK = "/run/chv-browser/browser.sock"
|
||
SHOTS_DIR = "/run/chv/shots"
|
||
PLAYWRIGHT_BROWSERS_PATH = "/usr/local/share/ms-playwright"
|
||
MAX_REQUEST_BYTES = 65536
|
||
MAX_A11Y_NODES = 200
|
||
MAX_A11Y_CHARS = 65536
|
||
NAV_TIMEOUT_MS = 30000
|
||
|
||
# 拒绝的 URL 主机：localhost 与私有地址字面量（主机名判定）。
|
||
_PRIVATE_HOST_RE = re.compile(
|
||
r"^(localhost|127\.\d+\.\d+\.\d+|10\.\d+\.\d+\.\d+|"
|
||
r"172\.(1[6-9]|2\d|3[01])\.\d+\.\d+|192\.168\.\d+\.\d+|"
|
||
r"\[::1\]|\[fe8[0-9a-f]*:.*\])$",
|
||
re.IGNORECASE)
|
||
|
||
|
||
def validate_url(url):
|
||
"""URL 校验：只允许 http/https，且主机不是 localhost/私有地址
|
||
字面量。返回 (ok, error)。DNS 重绑定不在本层防（见模块 docstring）。"""
|
||
if not isinstance(url, str) or not url:
|
||
return False, "url 不能为空"
|
||
m = re.match(r"^(https?)://([^/]+)", url, re.IGNORECASE)
|
||
if not m:
|
||
return False, "只允许 http/https URL：%s" % url[:80]
|
||
host = m.group(2).split("@")[-1].split(":")[0].lower()
|
||
if _PRIVATE_HOST_RE.match(host):
|
||
return False, "拒绝 localhost/私有地址：%s" % host
|
||
return True, ""
|
||
|
||
|
||
def truncate_aria(text, max_nodes=MAX_A11Y_NODES,
|
||
max_chars=MAX_A11Y_CHARS):
|
||
"""截断 ARIA 快照字符串：节点行（行首 "- "）超上限时剪掉多余
|
||
行，字符超上限时截断。返回 (text, truncated)。纯函数，可单测。
|
||
round-11 复测纠正（第二轮）：Playwright 根本没有
|
||
page.accessibility（那是 Puppeteer 的 API），原生快照是
|
||
Page.aria_snapshot() 返回的字符串；旧的 dict 版 truncate_a11y
|
||
随之废止。a11y 字段仍叫 a11y，值改为 ARIA 快照字符串。"""
|
||
text = text or ""
|
||
lines = text.splitlines()
|
||
kept = []
|
||
nodes = 0
|
||
truncated = False
|
||
for line in lines:
|
||
if line.lstrip().startswith("- "):
|
||
nodes += 1
|
||
if nodes > max_nodes:
|
||
truncated = True
|
||
break
|
||
kept.append(line)
|
||
out = "\n".join(kept)
|
||
if len(out) > max_chars:
|
||
out = out[:max_chars]
|
||
truncated = True
|
||
return out, truncated
|
||
|
||
|
||
def _load_playwright():
|
||
"""懒加载 playwright；缺失时报明错（不静默假装有浏览器）。
|
||
返回调用后的上下文管理器（sync_playwright()），不是函数本身——
|
||
round-11 复测回归：曾返回裸 sync_playwright 函数，
|
||
_ensure_browser 对它调 __enter__ 即 AttributeError，open
|
||
永远失败；单测的假工厂直接返回带 __enter__ 的对象，盖不住
|
||
这个形状（见 test_load_playwright_returns_called_cm）。"""
|
||
os.environ.setdefault("PLAYWRIGHT_BROWSERS_PATH",
|
||
PLAYWRIGHT_BROWSERS_PATH)
|
||
try:
|
||
from playwright.sync_api import sync_playwright
|
||
except ImportError as e:
|
||
raise RuntimeError(
|
||
"chv-browser: playwright 未装（%s），浏览器子代理不可用"
|
||
% e)
|
||
return sync_playwright()
|
||
|
||
|
||
class BrowserAgent:
|
||
"""单页浏览器子代理。playwright_factory 供单测注入假实现；
|
||
正常运行时用 _load_playwright()。"""
|
||
|
||
def __init__(self, playwright_factory=None, shots_dir=SHOTS_DIR):
|
||
self._factory = playwright_factory or _load_playwright
|
||
self._shots_dir = shots_dir
|
||
self._pw = None
|
||
self._browser = None
|
||
self._page = None
|
||
|
||
def _ensure_browser(self):
|
||
if self._browser is not None:
|
||
return
|
||
self._pw = self._factory()
|
||
# sync_playwright() 返回上下文管理器；__enter__ 后拿 chromium。
|
||
entered = self._pw.__enter__()
|
||
self._browser = entered.chromium.launch(headless=True)
|
||
|
||
def open(self, url):
|
||
ok, err = validate_url(url)
|
||
if not ok:
|
||
return {"ok": False, "error": err}
|
||
try:
|
||
self._ensure_browser()
|
||
if self._page is not None:
|
||
try:
|
||
self._page.close()
|
||
except Exception:
|
||
pass
|
||
self._page = self._browser.new_page()
|
||
self._page.goto(url, timeout=NAV_TIMEOUT_MS)
|
||
self._page.wait_for_load_state("domcontentloaded",
|
||
timeout=NAV_TIMEOUT_MS)
|
||
except Exception as e:
|
||
return {"ok": False,
|
||
"error": "打开 %s 失败：%s" % (url[:80], e)}
|
||
return self.snapshot()
|
||
|
||
def snapshot(self):
|
||
if self._page is None:
|
||
return {"ok": False, "error": "还没有打开的页面（先 open）"}
|
||
try:
|
||
title = self._page.title()
|
||
url = self._page.url
|
||
# round-11 复测纠正（第二轮）：Playwright 没有
|
||
# page.accessibility（Puppeteer 的 API），原生快照是
|
||
# Page.aria_snapshot()（字符串）；a11y 字段仍叫 a11y，
|
||
# 值改为 ARIA 快照字符串（截断到 200 节点行/64KB）。
|
||
aria = self._page.aria_snapshot()
|
||
except Exception as e:
|
||
return {"ok": False, "error": "取快照失败：%s" % e}
|
||
aria, truncated = truncate_aria(aria or "")
|
||
return {"ok": True, "title": title, "url": url,
|
||
"a11y": aria, "truncated": truncated}
|
||
|
||
def screenshot(self):
|
||
if self._page is None:
|
||
return {"ok": False, "error": "还没有打开的页面（先 open）"}
|
||
try:
|
||
os.makedirs(self._shots_dir, exist_ok=True)
|
||
path = os.path.join(self._shots_dir, "%s.png"
|
||
% uuid.uuid4().hex)
|
||
self._page.screenshot(path=path)
|
||
except Exception as e:
|
||
return {"ok": False, "error": "截图失败：%s" % e}
|
||
return {"ok": True, "path": path}
|
||
|
||
def close(self):
|
||
if self._page is not None:
|
||
try:
|
||
self._page.close()
|
||
except Exception:
|
||
pass
|
||
self._page = None
|
||
return {"ok": True, "closed": True}
|
||
|
||
def shutdown(self):
|
||
try:
|
||
if self._browser is not None:
|
||
self._browser.close()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
if self._pw is not None:
|
||
self._pw.__exit__(None, None, None)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# 本层不暴露的操作：evaluate / content（原始 DOM）/ DevTools。
|
||
# 收到即拒绝，不转发、不执行。
|
||
FORBIDDEN_OPS = ("evaluate", "content", "devtools", "js")
|
||
|
||
|
||
def handle_request(agent, req):
|
||
"""处理一个请求 dict，返回响应 dict（纯逻辑，可单测）。
|
||
op "browser" 是 broker 原样转发的形状（round-11 真机修复）：
|
||
内层 op 在 action 里；无 action 但带 url 时默认 open（驱动
|
||
只发 url）；action 同样受 FORBIDDEN_OPS 约束。"""
|
||
if not isinstance(req, dict):
|
||
return {"ok": False, "error": "请求须是 JSON 对象"}
|
||
op = req.get("op")
|
||
if op == "browser":
|
||
op = req.get("action")
|
||
if op is None and req.get("url"):
|
||
op = "open"
|
||
if op in FORBIDDEN_OPS:
|
||
return {"ok": False,
|
||
"error": "操作 %r 不暴露（无 JS/无原始 DOM）" % op}
|
||
if op == "ping":
|
||
return {"ok": True, "browser": True}
|
||
if op == "open":
|
||
return agent.open(req.get("url"))
|
||
if op == "snapshot":
|
||
return agent.snapshot()
|
||
if op == "screenshot":
|
||
return agent.screenshot()
|
||
if op == "close":
|
||
return agent.close()
|
||
return {"ok": False,
|
||
"error": "未知操作：%r（白名单：open/snapshot/screenshot/"
|
||
"close/ping）" % (op,)}
|
||
|
||
|
||
def serve(sock_path=BROWSER_SOCK, shots_dir=SHOTS_DIR):
|
||
"""监听 Unix socket，逐连接处理。socket 0600（只给属主 +
|
||
broker 的 root 连）。建不起来即抛错，不要静默带病跑。"""
|
||
if os.path.exists(sock_path):
|
||
os.remove(sock_path)
|
||
os.makedirs(os.path.dirname(sock_path), exist_ok=True)
|
||
srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
||
srv.bind(sock_path)
|
||
srv.listen(16)
|
||
os.chmod(sock_path, 0o600)
|
||
print("chv-browser: 监听 %s" % sock_path, file=sys.stderr, flush=True)
|
||
agent = BrowserAgent(shots_dir=shots_dir)
|
||
try:
|
||
while True:
|
||
try:
|
||
conn, _ = srv.accept()
|
||
except OSError:
|
||
continue
|
||
try:
|
||
conn.settimeout(120)
|
||
data = b""
|
||
while len(data) < MAX_REQUEST_BYTES:
|
||
chunk = conn.recv(4096)
|
||
if not chunk:
|
||
break
|
||
data += chunk
|
||
if data.endswith(b"\n"):
|
||
break
|
||
try:
|
||
req = json.loads(data.decode())
|
||
except (ValueError, UnicodeDecodeError):
|
||
resp = {"ok": False, "error": "请求不是合法 JSON"}
|
||
else:
|
||
try:
|
||
resp = handle_request(agent, req)
|
||
except Exception as e:
|
||
resp = {"ok": False,
|
||
"error": "浏览器子代理内部错误：%s" % e}
|
||
conn.sendall((json.dumps(resp, ensure_ascii=False)
|
||
+ "\n").encode())
|
||
except OSError as e:
|
||
print("chv-browser: 连接处理异常：%s" % e, file=sys.stderr,
|
||
flush=True)
|
||
finally:
|
||
try:
|
||
conn.close()
|
||
except OSError:
|
||
pass
|
||
finally:
|
||
agent.shutdown()
|
||
|
||
|
||
def main(argv):
|
||
sock_path = BROWSER_SOCK
|
||
shots_dir = SHOTS_DIR
|
||
i = 0
|
||
while i < len(argv):
|
||
if argv[i] == "--socket" and i + 1 < len(argv):
|
||
sock_path = argv[i + 1]
|
||
i += 2
|
||
elif argv[i] == "--shots-dir" and i + 1 < len(argv):
|
||
shots_dir = argv[i + 1]
|
||
i += 2
|
||
else:
|
||
print("chv-browser: 未知参数：%s" % argv[i], file=sys.stderr)
|
||
return 2
|
||
serve(sock_path, shots_dir)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main(sys.argv[1:]))
|