host
114b444554
Playwright 1.64.1 没有 page.accessibility（Puppeteer 的 API）， broker 的 snapshot 调它必走异常分支。改调 Page.aria_snapshot() （字符串）；guard 的 truncate_a11y 换成字符串版 truncate_aria （179a86d 原样），detect_injection/detect_form_risk 加字符串 分支（dict 分支保留）；两边 fake 与 P4 文档同步改。a11y-only 契约不变：不读原始 DOM、不走 JS eval。 测试：test_chv_muse_p2 + test_chv_muse_p4 114 OK；全量 2523 OK （skipped=10）。
409 lines
16 KiB
Python
409 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""chv_browser_guard：浏览器安全共享纯函数（客机侧）。
|
||
|
||
对齐 Muse P4（见 docs/chv-muse-p4-gap.md）：本模块只放纯函数，
|
||
不 import playwright、不起进程、不连网络，供 chv-cdp-broker.py
|
||
与 chv-browser.py 两脚本共用，避免三份拷贝漂移。
|
||
|
||
内容：
|
||
validate_url P2 的 URL 规则：只 http/https，拒 localhost/
|
||
私有地址字面量（DNS 重绑定是 Sentinel 的事）。
|
||
check_blocklist 静态恶意站名单（browser-blocklist.txt）后缀匹配。
|
||
truncate_aria ARIA 快照字符串截断（200 节点行/64KB）。
|
||
sanitize_error 错误信息消毒：去 HTML 标签、截断 200 字符、
|
||
仍含尖括号即只留异常类型名——快照路径上不许有
|
||
原始 DOM/HTML 外泄口子。
|
||
detect_injection DOM 内提示注入启发式（中英文模式表，命中返标签；
|
||
是规则，不是模型分类器）。
|
||
detect_form_risk ARIA 快照字符串（或旧 dict 树）扫密码/支付字段。
|
||
scan_download 下载扫描钩子：一条可执行扩展名规则。
|
||
enqueue/list/resolve_ask 文件型待审批队列（与 P3 ApprovalQueue
|
||
同语义：pending/allowed/denied）。
|
||
LAUNCH_ARGS / FORBIDDEN_PAGE_ATTRS Chromium 启动参数白名单与
|
||
禁用的 Playwright 页面调用名表。
|
||
"""
|
||
import json
|
||
import os
|
||
import re
|
||
import time
|
||
import uuid
|
||
|
||
MAX_A11Y_NODES = 200
|
||
MAX_A11Y_CHARS = 65536
|
||
MAX_ERROR_CHARS = 200
|
||
|
||
# 拒绝的 URL 主机：localhost 与私有地址字面量（P2 原样）。
|
||
_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)
|
||
|
||
# Chromium 启动参数白名单：均为真实开关；刻意没有
|
||
# --remote-debugging-port（DevTools 不暴露）与 --no-sandbox
|
||
#（浏览器以非 root 的 chv-browser 身份跑，不降级沙箱）。
|
||
LAUNCH_ARGS = [
|
||
"--disable-gpu",
|
||
"--no-first-run",
|
||
"--disable-dev-shm-usage",
|
||
"--disable-extensions",
|
||
"--disable-background-networking",
|
||
"--disable-sync",
|
||
"--disable-translate",
|
||
"--mute-audio",
|
||
]
|
||
|
||
# 禁用的 Playwright 页面调用名：页面上下文 JS 与进程内 exec
|
||
# 的全部入口。broker 与子代理的代码里不得出现这些属性名，
|
||
# 测试用 AST 扫两个脚本钉住（字符串常量里的出现不算）。
|
||
FORBIDDEN_PAGE_ATTRS = (
|
||
"evaluate", "evaluate_handle", "add_script_tag",
|
||
"add_init_script", "expose_function", "set_content",
|
||
"keyboard", "mouse", "touchscreen",
|
||
)
|
||
|
||
# 浏览器 socket 白名单之外的操作名（子代理层拒绝）。
|
||
FORBIDDEN_OPS = (
|
||
"evaluate", "content", "devtools", "js", "click", "fill",
|
||
"type", "press", "hover", "select", "check", "route",
|
||
"expose", "add_init_script", "add_script_tag", "set_content",
|
||
"eval",
|
||
)
|
||
|
||
# 下载拦截：可执行扩展名（一条规则；后续规则只加条目）。
|
||
BLOCKED_DOWNLOAD_EXTS = frozenset({
|
||
".exe", ".msi", ".bat", ".cmd", ".ps1", ".sh", ".dmg",
|
||
".pkg", ".apk", ".scr", ".vbs", ".com", ".gadget",
|
||
})
|
||
|
||
# 提示注入启发式模式表（中英文各 6 条）：字面匹配，不是语义判断。
|
||
# 命中只做标注（untrusted + injection_flags），不做拦截决定。
|
||
INJECTION_PATTERNS = (
|
||
("en_ignore_instructions", r"ignore (all )?(previous|prior|above) instructions"),
|
||
("en_you_are_now", r"you are now (a|an|the)"),
|
||
("en_system_prompt", r"(reveal|show|print|disregard).{0,40}system prompt"),
|
||
("en_disregard", r"disregard (all )?(previous|prior|above)"),
|
||
("en_new_instructions", r"new instructions:? (you|to|do)"),
|
||
("en_jailbreak", r"(jailbreak|DAN mode|do anything now)"),
|
||
("zh_ignore", r"忽略(以上|之前|所有|先前)的?(指令|指示|提示)"),
|
||
("zh_pretend", r"假装你是"),
|
||
("zh_now_you_are", r"你现在是"),
|
||
("zh_leak_prompt", r"(泄露|泄漏|输出|打印).{0,20}系统提示"),
|
||
("zh_admin", r"以管理员(身份|权限)"),
|
||
("zh_no_limit", r"(无视|绕过).{0,20}(限制|规则|审查)"),
|
||
)
|
||
|
||
# 表单风险：ARIA 快照行里的字段名/autocomplete 里的敏感词（小写匹配）。
|
||
PASSWORD_HINTS = ("password", "密码", "current-password",
|
||
"new-password", "passwd", "pass word")
|
||
PAYMENT_HINTS = ("payment", "支付", "信用卡", "银行卡", "卡号",
|
||
"cc-number", "cc-exp", "cc-csc", "cc-name",
|
||
"card number", "cvv", "expiry", "有效期")
|
||
|
||
|
||
def validate_url(url):
|
||
"""URL 校验：只允许 http/https，且主机不是 localhost/私有地址
|
||
字面量。返回 (ok, error)。"""
|
||
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 url_host(url):
|
||
"""取 URL 的主机小写形式；取不到返回空串。"""
|
||
m = re.match(r"^(https?)://([^/]+)", url or "", re.IGNORECASE)
|
||
if not m:
|
||
return ""
|
||
return m.group(2).split("@")[-1].split(":")[0].lower()
|
||
|
||
|
||
def load_blocklist(path):
|
||
"""读静态拦截名单：每行一个条目（域名或 .后缀），# 开头是注释，
|
||
空行跳过。文件缺失即空名单（不静默编造条目）。"""
|
||
entries = set()
|
||
try:
|
||
with open(path) as fh:
|
||
for line in fh:
|
||
line = line.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
entries.add(line.lower())
|
||
except OSError:
|
||
return set()
|
||
return entries
|
||
|
||
|
||
def check_blocklist(host, entries):
|
||
"""主机是否命中拦截名单：精确匹配或后缀匹配（条目 .example.com
|
||
命中 a.example.com；条目 example.com 同样命中子域）。"""
|
||
host = (host or "").lower()
|
||
if not host or not entries:
|
||
return False
|
||
for entry in entries:
|
||
entry = entry.lstrip(".")
|
||
if host == entry or host.endswith("." + entry):
|
||
return True
|
||
return False
|
||
|
||
|
||
def truncate_aria(text, max_nodes=MAX_A11Y_NODES,
|
||
max_chars=MAX_A11Y_CHARS):
|
||
"""截断 ARIA 快照字符串：节点行（行首 "- "）超上限时剪掉多余
|
||
行，字符超上限时截断。返回 (text, truncated)。纯函数，可单测。
|
||
round-11 复测纠正（第二轮，179a86d）：Playwright 根本没有
|
||
page.accessibility（那是 Puppeteer 的 API），原生快照是
|
||
Page.aria_snapshot() 返回的字符串；旧的 dict 版 truncate_a11y
|
||
随之废止。round-12 合入时从 chv-browser.py 移植到共享 guard
|
||
（P4 架构里快照走 chv-cdp-broker.py）。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 sanitize_error(exc):
|
||
"""错误信息消毒：只留异常类型名 + 去 HTML 标签、截断后的首行；
|
||
凭据形内容（token=/password= 等 key=value 与 sk- 形 token）打码；
|
||
清洗后仍含尖括号或 javascript: 即丢弃原文。绝不把页面原文
|
||
或凭据原样放进回包或日志。"""
|
||
name = type(exc).__name__ if not isinstance(exc, str) else "Error"
|
||
text = exc if isinstance(exc, str) else str(exc)
|
||
text = re.sub(r"<[^>]{0,500}>", "", text)
|
||
text = re.sub(r"(?i)(token|password|passwd|secret|api[-_]?key)"
|
||
r"\s*=\s*[^\s&;\"']+", r"\1=<redacted>", text)
|
||
text = re.sub(r"sk-[A-Za-z0-9_\-]{8,}", "<redacted>", text)
|
||
text = " ".join(text.split())[:MAX_ERROR_CHARS]
|
||
if "<" in text or ">" in text or "javascript:" in text.lower():
|
||
return "浏览器操作失败（%s）" % name
|
||
if not text:
|
||
return "浏览器操作失败（%s）" % name
|
||
return "浏览器操作失败（%s）：%s" % (name, text)
|
||
|
||
|
||
def _collect_text(node, out):
|
||
"""收集无障碍树里全部字符串值（name/value/description 等），
|
||
供注入启发式扫描。只收字符串，不收 role 名以外的结构。"""
|
||
if isinstance(node, dict):
|
||
for value in node.values():
|
||
_collect_text(value, out)
|
||
elif isinstance(node, list):
|
||
for item in node:
|
||
_collect_text(item, out)
|
||
elif isinstance(node, str):
|
||
out.append(node)
|
||
|
||
|
||
def detect_injection(tree):
|
||
"""DOM 内提示注入启发式：扫描快照全部文本，返回命中的模式标签
|
||
列表（去重、按模式表顺序）。无命中即空列表。规则而已，不是
|
||
模型分类器——命中只标注，不拦截。输入是 ARIA 快照字符串
|
||
（round-12 移植后 broker 传的就是它）；旧 dict 树仍接受，
|
||
收集全部字符串值后同样扫描。"""
|
||
if isinstance(tree, str):
|
||
blob = tree
|
||
else:
|
||
texts = []
|
||
_collect_text(tree or {}, texts)
|
||
blob = "\n".join(texts)
|
||
hits = []
|
||
for label, pattern in INJECTION_PATTERNS:
|
||
if re.search(pattern, blob, re.IGNORECASE):
|
||
hits.append(label)
|
||
return hits
|
||
|
||
|
||
def _node_hints(node, hints):
|
||
"""一个无障碍树节点的 name/autocomplete/type 里是否含敏感词。"""
|
||
blob = " ".join(str(node.get(k, "")) for k in
|
||
("name", "autocomplete", "type", "description"))
|
||
blob = blob.lower()
|
||
return any(h in blob for h in hints)
|
||
|
||
|
||
def _walk_form_risk(node, found):
|
||
if not isinstance(node, dict):
|
||
return
|
||
role = str(node.get("role", "")).lower()
|
||
if role in ("textbox", "textfield", "combobox"):
|
||
if _node_hints(node, PASSWORD_HINTS):
|
||
found.setdefault("password", []).append(
|
||
str(node.get("name", ""))[:40])
|
||
elif _node_hints(node, PAYMENT_HINTS):
|
||
found.setdefault("payment", []).append(
|
||
str(node.get("name", ""))[:40])
|
||
for child in node.get("children") or []:
|
||
_walk_form_risk(child, found)
|
||
|
||
|
||
_ARIA_FIELD_RE = re.compile(
|
||
r"^\s*-\s+(textbox|textfield|combobox)\b", re.IGNORECASE)
|
||
_ARIA_NAME_RE = re.compile(r'"([^"]{0,40})"')
|
||
|
||
|
||
def _form_risk_from_aria(text, found):
|
||
"""ARIA 快照字符串版表单风险扫描：每行行首 "- textbox/
|
||
textfield/combobox" 的字段行，整行小写匹配敏感词（字段名与
|
||
[autocomplete=...] 都在行内）；字段名取行内第一个引号串。"""
|
||
for line in (text or "").splitlines():
|
||
if not _ARIA_FIELD_RE.match(line):
|
||
continue
|
||
blob = line.lower()
|
||
m = _ARIA_NAME_RE.search(line)
|
||
name = m.group(1) if m else ""
|
||
if any(h in blob for h in PASSWORD_HINTS):
|
||
found.setdefault("password", []).append(name)
|
||
elif any(h in blob for h in PAYMENT_HINTS):
|
||
found.setdefault("payment", []).append(name)
|
||
|
||
|
||
def detect_form_risk(tree):
|
||
"""扫快照里的密码/支付字段。返回 {level, fields}：level 是
|
||
none/password/payment（支付优先于密码），fields 是命中的字段名
|
||
（最多 5 个，供审批记录看，不进模型上下文原样只作标注）。
|
||
输入是 ARIA 快照字符串（round-12 移植后 broker 传的就是它）；
|
||
旧 dict 树仍接受，走原来的节点遍历。"""
|
||
found = {}
|
||
if isinstance(tree, str):
|
||
_form_risk_from_aria(tree, found)
|
||
else:
|
||
_walk_form_risk(tree or {}, found)
|
||
if found.get("payment"):
|
||
return {"level": "payment",
|
||
"fields": found["payment"][:5]}
|
||
if found.get("password"):
|
||
return {"level": "password",
|
||
"fields": found["password"][:5]}
|
||
return {"level": "none", "fields": []}
|
||
|
||
|
||
def scan_download(filename):
|
||
"""下载扫描钩子（一条规则）：可执行扩展名、或中间段含可执行
|
||
扩展名（双扩展名把戏，如 report.pdf.exe / archive.exe.pdf），
|
||
即拦。返回 (blocked, reason)。"""
|
||
name = (filename or "").strip().lower()
|
||
if not name:
|
||
return False, ""
|
||
parts = [p for p in name.split(".") if p]
|
||
if len(parts) < 2:
|
||
return False, ""
|
||
final_ext = "." + parts[-1]
|
||
if final_ext in BLOCKED_DOWNLOAD_EXTS:
|
||
return True, "可执行扩展名：%s" % final_ext
|
||
for middle in parts[1:-1]:
|
||
if ("." + middle) in BLOCKED_DOWNLOAD_EXTS:
|
||
return True, "双扩展名可疑：%s" % filename[:60]
|
||
return False, ""
|
||
|
||
|
||
def check_launch_args(args):
|
||
"""Chromium 启动参数校验：不得含 --remote-debugging-*（DevTools
|
||
不暴露），不得含 devtools 字串。返回 (ok, error)。"""
|
||
for arg in args or []:
|
||
if arg.startswith("--remote-debugging"):
|
||
return False, "启动参数不得含 %s" % arg
|
||
if "devtools" in arg.lower():
|
||
return False, "启动参数不得含 DevTools：%s" % arg
|
||
return True, ""
|
||
|
||
|
||
def _ask_path(queue_dir, ask_id):
|
||
return os.path.join(queue_dir, "ask-%s.json" % ask_id)
|
||
|
||
|
||
def enqueue_ask(queue_dir, kind, url, detail):
|
||
"""待审批入队：一条 ask 一个 JSON 文件（原子写：先写 tmp 再
|
||
rename）。返回 ask_id。队列目录不存在即建。"""
|
||
os.makedirs(queue_dir, exist_ok=True)
|
||
ask_id = uuid.uuid4().hex[:12]
|
||
record = {"ask_id": ask_id, "kind": kind, "url": url or "",
|
||
"detail": detail or "", "status": "pending",
|
||
"created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ",
|
||
time.gmtime())}
|
||
tmp_path = _ask_path(queue_dir, ask_id) + ".tmp"
|
||
with open(tmp_path, "w") as fh:
|
||
json.dump(record, fh, ensure_ascii=False)
|
||
fh.write("\n")
|
||
os.replace(tmp_path, _ask_path(queue_dir, ask_id))
|
||
return ask_id
|
||
|
||
|
||
def list_asks(queue_dir):
|
||
"""列出待审批（status pending）的 ask，按创建时间排序。"""
|
||
out = []
|
||
try:
|
||
names = sorted(os.listdir(queue_dir))
|
||
except OSError:
|
||
return []
|
||
for name in names:
|
||
if not (name.startswith("ask-") and name.endswith(".json")):
|
||
continue
|
||
try:
|
||
with open(os.path.join(queue_dir, name)) as fh:
|
||
record = json.load(fh)
|
||
except (OSError, ValueError):
|
||
continue
|
||
if record.get("status") == "pending":
|
||
out.append(record)
|
||
out.sort(key=lambda r: r.get("created_at", ""))
|
||
return out
|
||
|
||
|
||
def resolve_ask(queue_dir, ask_id, decision):
|
||
"""写回审批决定：decision 只认 allowed/denied。返回 (ok, error)。
|
||
不存在的 ask_id 即报错，不静默成功。"""
|
||
if decision not in ("allowed", "denied"):
|
||
return False, "decision 只认 allowed/denied：%s" % decision
|
||
path = _ask_path(queue_dir, ask_id)
|
||
try:
|
||
with open(path) as fh:
|
||
record = json.load(fh)
|
||
except (OSError, ValueError):
|
||
return False, "ask 不存在：%s" % ask_id
|
||
if record.get("status") != "pending":
|
||
return False, "ask 已处理：%s" % ask_id
|
||
record["status"] = decision
|
||
tmp_path = path + ".tmp"
|
||
with open(tmp_path, "w") as fh:
|
||
json.dump(record, fh, ensure_ascii=False)
|
||
fh.write("\n")
|
||
os.replace(tmp_path, path)
|
||
return True, ""
|
||
|
||
|
||
def response_leaks(resp):
|
||
"""回包泄露检查（测试与 broker 自检共用）：递归扫响应里的字符串，
|
||
命中 ws://、DevTools 字串或 HTML 标签即返回问题列表。空列表即干净。"""
|
||
problems = []
|
||
text = json.dumps(resp, ensure_ascii=False)
|
||
if "ws://" in text or "wss://" in text:
|
||
problems.append("响应含 ws 端点")
|
||
# 只查 CDP 形态（devtools/ 路径、remote-debugging），不查裸词——
|
||
# 正常 URL 里也可能含 devtools（如 devtools.example.com）。
|
||
if re.search(r"devtools/", text, re.IGNORECASE):
|
||
problems.append("响应含 DevTools 路径")
|
||
if "remote-debugging" in text.lower():
|
||
problems.append("响应含 remote-debugging")
|
||
if re.search(r"<[a-zA-Z][^>]{0,200}>", text):
|
||
problems.append("响应含 HTML 标签")
|
||
return problems
|