host
04287f6053
复测（RETEST.md §browser-defect、§broker-proto）：
1. _load_playwright 返回裸 sync_playwright 函数（未调用），
_ensure_browser 对它调 __enter__ 即 AttributeError，open
永远失败。改返回 sync_playwright()（调用后的上下文管理器），
与单测假工厂的形状一致。回归测试用 stub 的 playwright 模块
钉住“返回调用结果、不是函数本身”。
2. broker 按设计原样转发 {op: browser, …}，chv-browser 白名单
却不认 browser。在 chv-browser 的 handle_request 里归一化：
内层 op 在 action 里；无 action 但带 url 时默认 open（驱动
只发 url）；action 同样受 FORBIDDEN_OPS 约束。broker 保持
原样转发不动（其 docstring 与单测均以此为契约）。
附：本提交含 test_chv_muse_p2.py 里 CellArgvTest 的配套更新
（c1d00b3 的 --volatile=yes 断言，同文件无法拆分提交）。
317 lines
12 KiB
Python
317 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""chv-browser：浏览器子代理（客机侧、单元外独立进程）。
|
||
|
||
对齐 Muse P2（见 docs/chv-muse-p2-gap.md）：浏览器任务由专门子代理
|
||
完成，与 codex 单元分离。子代理只向单元返回页面无障碍树快照
|
||
（Playwright accessibility.snapshot），不见原始 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_a11y(node, max_nodes=MAX_A11Y_NODES,
|
||
max_chars=MAX_A11Y_CHARS):
|
||
"""截断无障碍树：节点数超上限时剪掉多余子树。返回 (tree, truncated)。
|
||
纯函数，可单测。节点计数按 dict 节点数（含 children 展开）。"""
|
||
state = {"count": 0, "truncated": False}
|
||
|
||
def walk(n):
|
||
if not isinstance(n, dict):
|
||
return n
|
||
state["count"] += 1
|
||
if state["count"] > max_nodes:
|
||
state["truncated"] = True
|
||
return None
|
||
out = {}
|
||
for key, value in n.items():
|
||
if key == "children" and isinstance(value, list):
|
||
children = []
|
||
for child in value:
|
||
walked = walk(child)
|
||
if walked is None:
|
||
state["truncated"] = True
|
||
break
|
||
children.append(walked)
|
||
out[key] = children
|
||
else:
|
||
out[key] = value
|
||
return out
|
||
|
||
tree = walk(node)
|
||
text = json.dumps(tree, ensure_ascii=False)
|
||
if len(text) > max_chars:
|
||
state["truncated"] = True
|
||
# 字符超限时只留顶层角色/名字/URL 信息，不灌整棵树。
|
||
tree = {"role": (node or {}).get("role"),
|
||
"name": (node or {}).get("name"),
|
||
"truncated_chars": True}
|
||
return tree, state["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
|
||
tree = self._page.accessibility.snapshot()
|
||
except Exception as e:
|
||
return {"ok": False, "error": "取快照失败：%s" % e}
|
||
tree, truncated = truncate_a11y(tree or {})
|
||
return {"ok": True, "title": title, "url": url,
|
||
"a11y": tree, "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:]))
|