#!/usr/bin/env python3
"""chv-cdp-broker：单元外 CDP broker（客机侧、单元外独立进程）。

对齐 Muse P4（见 docs/chv-muse-p4-gap.md）：Chromium 的所有权
在这里，不在浏览器子代理手里。broker 独家持有 Playwright，
CDP 连接只存在于本进程内；子代理（chv-browser.py）只经本
socket 说话，绝不直连浏览器、绝不见 CDP 端点。

操作（Unix socket /run/chv/cdp-broker.sock，JSON 行）：
  open       {url}   -> 导航并返回 {title, url, a11y, truncated,
                    untrusted_input, injection_flags, form_risk,
                    checkout}
  snapshot   {}      -> 当前页同上
  screenshot {}      -> 截图落盘 /run/chv/shots/<uuid>.png，返回 {path}
  close      {}      -> 关当前页（浏览器进程继续跑）
  pause      {}      -> 暂停（用户接管/填表/自动暂停的统一状态）
  resume     {}      -> 解除暂停
  status     {}      -> {paused, url, form_risk}
  list_asks  {}      -> 待审批 ask 列表
  resolve_ask {ask_id, decision} -> 写回 allow/denied
  ping       {}      -> {"ok": true}

暂停时除 ping/resume/status/list_asks 外一切操作拒绝——
用户接管或凭证注入填表时，代理完全不能行动。

安全规则（纯函数在 chv_browser_guard.py，本文件只做接线）：
  导航前 validate_url + 静态拦截名单；快照恒带
  untrusted_input=true（外部页面文本永远是数据、不是指令），
  命中注入启发式带 injection_flags；高风险表单在页时 open
  导航即返回 ask（入队、执行停止）；结账页命中时 open 导航
  同样返回 ask 并暂停（round-12 修复轮项 4，P10 接线，
  detect_checkout 在快照时重算、导航前后两道门）；观测到
  导航 POST 即自动暂停 + ask；下载命中可执行扩展名即取消
  + 暂停 + ask。
  所有回包与日志过 sanitize_error；响应里绝无 ws 端点、
  DevTools 地址、原始 HTML。

Playwright 调用白名单：launch/goto/wait_for_load_state/title/
url/aria_snapshot/screenshot/close + 只读监听
on(request/download)。evaluate/add_script_tag/add_init_script/
expose_function/set_content 等脚本注入面不在代码里出现，
测试用 AST 扫本文件钉住。

只用标准库 + playwright（运行时，懒加载）。运行身份：系统用户
chv-browser，socket 0600。
"""
import importlib.util
import json
import os
import socket
import sys
import uuid

CDP_SOCK = "/run/chv/cdp-broker.sock"
ASK_QUEUE_DIR = "/run/chv/ask-queue"
BLOCKLIST_PATH = "/usr/local/share/chv/browser-blocklist.txt"
SHOTS_DIR = "/run/chv/shots"
PLAYWRIGHT_BROWSERS_PATH = "/usr/local/share/ms-playwright"
MAX_REQUEST_BYTES = 65536
NAV_TIMEOUT_MS = 30000

_GUARD = None
_CHECKOUT = None


def _load_guard():
    """加载同目录的 chv_browser_guard（纯函数模块）。运行时
    安装在 /usr/local/bin/chv_browser_guard.py；测试时与本
    文件同在 chv-guest/。只在首次调用加载一次。"""
    global _GUARD
    if _GUARD is not None:
        return _GUARD
    here = os.path.dirname(os.path.abspath(__file__))
    path = os.path.join(here, "chv_browser_guard.py")
    spec = importlib.util.spec_from_file_location(
        "chv_browser_guard", path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    _GUARD = module
    return module


def _load_checkout():
    """加载结账检测模块 checkout_detect（纯函数模块）。

    运行时安装在 /usr/local/bin/checkout_detect.py；测试时在
    仓库 cloud/chv/checkout_detect.py。找到即只加载一次；两处
    都找不到时返回 None（降级：结账检测关闭，broker 继续跑并
    写警告——不因缺一个检测模块把浏览器整路 crash-loop，但完
    整镜像必须含它，build.sh 有登记、BUILDINFO 有登记）。
    """
    global _CHECKOUT
    if _CHECKOUT is not None:
        return _CHECKOUT
    here = os.path.dirname(os.path.abspath(__file__))
    candidates = [
        os.path.join(here, "checkout_detect.py"),
        os.path.join(os.path.dirname(here), "cloud", "chv",
                     "checkout_detect.py"),
    ]
    for path in candidates:
        if not os.path.exists(path):
            continue
        spec = importlib.util.spec_from_file_location(
            "checkout_detect", path)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        _CHECKOUT = module
        return module
    sys.stderr.write(
        "chv-cdp-broker: 警告：checkout_detect 未找到（%s），"
        "结账检测关闭\n" % candidates)
    _CHECKOUT = None
    return None


def _load_playwright():
    """懒加载 playwright；缺失时报明错（不静默假装有浏览器）。
    返回调用后的上下文管理器（sync_playwright()），不是函数本身——
    round-11 复测回归（原在 chv-browser.py，P4 起 Playwright
    所有权移到本文件，同形 bug 随之移到此处；合入时移植）：
    曾返回裸 sync_playwright 函数，_ensure_browser 对它调
    __enter__ 即 AttributeError，open 永远失败；单测的假工厂
    直接返回带 __enter__ 的对象，盖不住这个形状（见
    test_chv_muse_p4.py 的 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-cdp-broker: playwright 未装（%s）" % e)
    return sync_playwright()


class CdpBroker:
    """CDP broker：独家持有 Chromium。playwright_factory 供单测
    注入假实现；正常运行时用 _load_playwright()。"""

    def __init__(self, playwright_factory=None,
                 blocklist_path=BLOCKLIST_PATH,
                 ask_queue_dir=ASK_QUEUE_DIR, shots_dir=SHOTS_DIR):
        self._guard = _load_guard()
        # round-12 修复轮（项 4，P10）：结账检测接线。模块缺失
        # 时 _checkout_mod 为 None，检测降级关闭（见 _load_checkout）。
        self._checkout_mod = _load_checkout()
        self._checkout = {"is_checkout": False, "confidence": "low",
                          "reasons": [], "form_risk": None,
                          "amount_text": None}
        # 本次会话里已批过 ask 的结账页 URL：批过之后同一页
        # 再 open 不再重复 ask（与 form_risk“批准即放行”同口径）。
        self._checkout_approved = set()
        self._factory = playwright_factory or _load_playwright
        self._blocklist_path = blocklist_path
        self._ask_queue_dir = ask_queue_dir
        self._shots_dir = shots_dir
        self._pw = None
        self._browser = None
        self._page = None
        self._paused = False
        self._form_risk = {"level": "none", "fields": []}
        self._current_url = ""

    def _ensure_browser(self):
        if self._browser is not None:
            return
        ok, err = self._guard.check_launch_args(
            self._guard.LAUNCH_ARGS)
        if not ok:
            raise RuntimeError("启动参数不合规：%s" % err)
        self._pw = self._factory()
        entered = self._pw.__enter__()
        self._browser = entered.chromium.launch(
            headless=True, args=list(self._guard.LAUNCH_ARGS))

    def _new_page(self):
        if self._page is not None:
            try:
                self._page.close()
            except Exception:
                pass
        self._page = self._browser.new_page()
        self._page.on("request", self._on_request)
        self._page.on("download", self._on_download)

    def _on_request(self, request):
        """只读监听：导航 POST（表单提交）即自动暂停 + 入队。
        非导航 POST（beacon/fetch）不在此处理——它不代表提交。"""
        try:
            method = request.method
            is_nav = request.is_navigation_request()
        except Exception:
            return
        if method != "POST" or not is_nav:
            return
        self._paused = True
        self._guard.enqueue_ask(
            self._ask_queue_dir, "form_submit_observed",
            self._current_url, "观测到导航 POST（疑似表单提交）")

    def _on_download(self, download):
        """只读监听：命中可执行扩展名即取消下载 + 暂停 + 入队。"""
        try:
            name = download.suggested_filename
        except Exception:
            return
        blocked, reason = self._guard.scan_download(name)
        if not blocked:
            return
        try:
            download.cancel()
        except Exception:
            pass
        self._paused = True
        self._guard.enqueue_ask(
            self._ask_queue_dir, "download_blocked",
            self._current_url, "下载命中拦截规则：%s" % reason)

    def _paused_resp(self):
        return {"ok": False,
                "error": "浏览器已暂停（用户接管/填表/自动暂停）"}

    def _ask_resp(self, ask_id, detail):
        return {"ok": False, "ask": True, "ask_id": ask_id,
                "error": "需审批：%s" % detail}

    def _checkout_verdict(self, url, aria):
        """结账检测：URL 信号 + ARIA 快照行拼成的伪节点。

        broker 只有 aria_snapshot() 字符串，没有节点树；每行
        快照文本作一个节点（name=行文本），与 detect_checkout
        的节点口径一致（_node_text 拼 role/name/label/type）。
        模块缺失时返回“非结账页”（降级，见 _load_checkout）。
        """
        if self._checkout_mod is None:
            return {"is_checkout": False, "confidence": "low",
                    "reasons": [], "form_risk": None,
                    "amount_text": None}
        nodes = [{"role": "", "name": line, "label": "", "type": ""}
                 for line in (aria or "").splitlines()
                 if line.strip()]
        verdict = self._checkout_mod.detect_checkout(url, nodes)
        return {"is_checkout": verdict.is_checkout,
                "confidence": verdict.confidence,
                "reasons": list(verdict.reasons),
                "form_risk": verdict.form_risk,
                "amount_text": verdict.amount_text}

    def _checkout_ask(self, url, pause):
        """结账页 ask：与 P4 的 form_submit 同种（P10 gap §4.1
        对齐），入队并按需暂停，返回 ask 响应。"""
        reasons = ",".join(self._checkout.get("reasons") or [])
        ask_id = self._guard.enqueue_ask(
            self._ask_queue_dir, "form_submit", url,
            "结账页命中（%s），购买前须审批" % reasons)
        if pause:
            self._paused = True
        return self._ask_resp(ask_id, "结账页命中，购买前须审批")

    def open(self, url):
        if self._paused:
            return self._paused_resp()
        ok, err = self._guard.validate_url(url)
        if not ok:
            return {"ok": False, "error": err}
        host = self._guard.url_host(url)
        entries = self._guard.load_blocklist(self._blocklist_path)
        if self._guard.check_blocklist(host, entries):
            return {"ok": False, "error": "命中恶意站拦截名单：%s"
                                          % host}
        if self._form_risk.get("level") != "none":
            ask_id = self._guard.enqueue_ask(
                self._ask_queue_dir, "form_submit", url,
                "高风险表单在页（%s），导航须审批"
                % self._form_risk["level"])
            return self._ask_resp(ask_id, "高风险表单在页，导航须审批")
        # round-12 修复轮（项 4，P10）：导航前门——上一页已是
        # 结账页时，下一次导航须先审批（与 _form_risk 门同形）。
        if (self._checkout.get("is_checkout")
                and url not in self._checkout_approved):
            return self._checkout_ask(url, pause=False)
        try:
            self._ensure_browser()
            self._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": self._guard.sanitize_error(e)}
        result = self.snapshot()
        # 导航后门——落点本身是结账页时，返回 ask 并暂停
        #（P10 gap §4.1：open 导航返回 ask 并暂停）。
        if (result.get("ok") and self._checkout.get("is_checkout")
                and url not in self._checkout_approved):
            return self._checkout_ask(url, pause=True)
        return result

    def snapshot(self):
        if self._paused:
            return self._paused_resp()
        if self._page is None:
            return {"ok": False, "error": "还没有打开的页面（先 open）"}
        try:
            title = self._page.title()
            url = self._page.url
            # round-12 移植 round-11 的 179a86d 纠正：Playwright
            # 没有 page.accessibility（Puppeteer 的 API），原生
            # 快照是 Page.aria_snapshot()（字符串）；a11y 字段仍
            # 叫 a11y，值改为 ARIA 快照字符串（截断到 200 节点
            # 行/64KB）。合入 round-12 时这处遗漏，仍调不存在的
            # accessibility.snapshot()，真机上 snapshot 必走异常
            # 分支返回错误。
            aria = self._page.aria_snapshot()
        except Exception as e:
            return {"ok": False,
                    "error": self._guard.sanitize_error(e)}
        aria, truncated = self._guard.truncate_aria(aria or "")
        flags = self._guard.detect_injection(aria)
        self._form_risk = self._guard.detect_form_risk(aria)
        # round-12 修复轮（项 4，P10）：每次快照重算结账 verdict，
        # 与 _form_risk 同形（快照缓存，open 的两道门读它）。
        self._checkout = self._checkout_verdict(url, aria)
        self._current_url = url
        return {"ok": True, "title": title, "url": url,
                "a11y": aria, "truncated": truncated,
                "untrusted_input": True, "injection_flags": flags,
                "form_risk": self._form_risk,
                "checkout": self._checkout}

    def screenshot(self):
        if self._paused:
            return self._paused_resp()
        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": self._guard.sanitize_error(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
        self._form_risk = {"level": "none", "fields": []}
        self._checkout = {"is_checkout": False, "confidence": "low",
                          "reasons": [], "form_risk": None,
                          "amount_text": None}
        self._current_url = ""
        return {"ok": True, "closed": True}

    def pause(self):
        self._paused = True
        return {"ok": True, "paused": True}

    def ping(self):
        return {"ok": True, "broker": True}

    def resume(self):
        self._paused = False
        return {"ok": True, "paused": False}

    def status(self):
        return {"ok": True, "paused": self._paused,
                "url": self._current_url,
                "form_risk": self._form_risk,
                "checkout": self._checkout}

    def list_asks(self):
        return {"ok": True,
                "asks": self._guard.list_asks(self._ask_queue_dir)}

    def resolve_ask(self, ask_id, decision):
        # 批之前先读回 ask 记录：批后它不再是 pending，
        # list_asks 读不到——结账 ask 批过后要把 URL 记进
        # _checkout_approved（同页再 open 不再重复 ask）。
        record = None
        for ask in self._guard.list_asks(self._ask_queue_dir):
            if ask.get("ask_id") == ask_id:
                record = ask
                break
        ok, err = self._guard.resolve_ask(
            self._ask_queue_dir, ask_id, decision)
        if not ok:
            return {"ok": False, "error": err}
        if decision == "allowed":
            self._paused = False
            # 批准即放行这次操作：清掉快照缓存的表单风险（下一次
            # snapshot 会按新页面重算，不会丢检测）。
            self._form_risk = {"level": "none", "fields": []}
            # round-12 修复轮（项 4，P10）：结账 ask 批过后，
            # 同 URL 再 open 不重复 ask；_checkout 本身也清掉，
            # 下一次 snapshot 按新页面重算。
            if record is not None and str(
                    record.get("detail", "")).startswith("结账页命中"):
                self._checkout_approved.add(record.get("url"))
            self._checkout = {"is_checkout": False,
                              "confidence": "low", "reasons": [],
                              "form_risk": None, "amount_text": None}
        else:
            self.close()
            self._paused = False
        return {"ok": True, "ask_id": ask_id,
                "decision": decision}

    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


def handle_request(broker, req):
    """处理一个请求 dict，返回响应 dict（纯逻辑，可单测）。
    未知操作与禁用操作一律拒绝，不转发、不执行。"""
    guard = _load_guard()
    if not isinstance(req, dict):
        return {"ok": False, "error": "请求须是 JSON 对象"}
    op = req.get("op")
    if op in guard.FORBIDDEN_OPS:
        return {"ok": False,
                "error": "操作 %r 不暴露（无 JS/无原始 DOM/无交互）"
                         % op}
    if op == "ping":
        return {"ok": True, "broker": True}
    if op == "open":
        return broker.open(req.get("url"))
    if op == "snapshot":
        return broker.snapshot()
    if op == "screenshot":
        return broker.screenshot()
    if op == "close":
        return broker.close()
    if op == "pause":
        return broker.pause()
    if op == "resume":
        return broker.resume()
    if op == "status":
        return broker.status()
    if op == "list_asks":
        return broker.list_asks()
    if op == "resolve_ask":
        return broker.resolve_ask(req.get("ask_id"),
                                  req.get("decision"))
    return {"ok": False,
            "error": "未知操作：%r（白名单：open/snapshot/screenshot/"
                     "close/pause/resume/status/list_asks/resolve_ask/"
                     "ping）" % (op,)}


def _handle_conn(broker, conn):
    """一个连接一请求一响应；回包先过泄露检查，不过即换通用错。"""
    guard = _load_guard()
    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(broker, req)
        except Exception as e:
            resp = {"ok": False,
                    "error": guard.sanitize_error(e)}
    if guard.response_leaks(resp):
        resp = {"ok": False, "error": "响应泄露检查未通过（已拦截）"}
    conn.sendall((json.dumps(resp, ensure_ascii=False) + "\n").encode())


def serve(sock_path=CDP_SOCK, shots_dir=SHOTS_DIR,
          blocklist_path=BLOCKLIST_PATH, ask_queue_dir=ASK_QUEUE_DIR,
          playwright_factory=None):
    """监听 Unix socket，逐连接处理。socket 0600。建不起来即抛错，
    不要静默带病跑。playwright_factory 只供测试注入假实现。"""
    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-cdp-broker: 监听 %s" % sock_path,
          file=sys.stderr, flush=True)
    broker = CdpBroker(playwright_factory=playwright_factory,
                       shots_dir=shots_dir,
                       blocklist_path=blocklist_path,
                       ask_queue_dir=ask_queue_dir)
    try:
        while True:
            try:
                conn, _ = srv.accept()
            except OSError:
                continue
            try:
                _handle_conn(broker, conn)
            except OSError as e:
                print("chv-cdp-broker: 连接处理异常：%s"
                      % _load_guard().sanitize_error(e),
                      file=sys.stderr, flush=True)
            finally:
                try:
                    conn.close()
                except OSError:
                    pass
    finally:
        broker.shutdown()


def main(argv):
    sock_path = CDP_SOCK
    shots_dir = SHOTS_DIR
    blocklist_path = BLOCKLIST_PATH
    ask_queue_dir = ASK_QUEUE_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
        elif argv[i] == "--blocklist" and i + 1 < len(argv):
            blocklist_path = argv[i + 1]
            i += 2
        elif argv[i] == "--ask-queue" and i + 1 < len(argv):
            ask_queue_dir = argv[i + 1]
            i += 2
        else:
            print("chv-cdp-broker: 未知参数：%s" % argv[i],
                  file=sys.stderr)
            return 2
    serve(sock_path, shots_dir, blocklist_path, ask_queue_dir)
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))