codex-app-server/chv-guest/chv_browser_guard.py

#!/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_a11y     P2 的无障碍树截断（200 节点/64KB）。
  sanitize_error    错误信息消毒：去 HTML 标签、截断 200 字符、
                    仍含尖括号即只留异常类型名——快照路径上不许有
                    原始 DOM/HTML 外泄口子。
  detect_injection  DOM 内提示注入启发式（中英文模式表，命中返标签；
                    是规则，不是模型分类器）。
  detect_form_risk  无障碍树扫密码/支付字段。
  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}(限制|规则|审查)"),
)

# 表单风险：无障碍树节点名/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_a11y(node, max_nodes=MAX_A11Y_NODES,
                  max_chars=MAX_A11Y_CHARS):
    """截断无障碍树：节点数超上限时剪掉多余子树。返回 (tree, truncated)。
    P2 原样搬入（两脚本共用，不再各存一份）。"""
    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
        tree = {"role": (node or {}).get("role"),
                "name": (node or {}).get("name"),
                "truncated_chars": True}
    return tree, state["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 内提示注入启发式：扫描快照全部文本，返回命中的模式标签
    列表（去重、按模式表顺序）。无命中即空列表。规则而已，不是
    模型分类器——命中只标注，不拦截。"""
    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)


def detect_form_risk(tree):
    """扫无障碍树里的密码/支付字段。返回 {level, fields}：level 是
    none/password/payment（支付优先于密码），fields 是命中的字段名
    （最多 5 个，供审批记录看，不进模型上下文原样只作标注）。"""
    found = {}
    _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