"""Bridge between the MCP console page and MCP servers / command-line tools.

Enabled by `python3 scripts/serve.py 8080 --mcp mcp.config.json`. The MCP client itself runs
in the browser (crates/tools/mcp, WebAssembly); this module only moves JSON-RPC
messages:

  * stdio servers: started as child processes, one message per line on stdin / stdout
  * HTTP servers (Streamable HTTP): requests forwarded, JSON or SSE responses unpacked
  * command-line tools: wrapped as a virtual MCP server whose tools run the registered
    programs (never through a shell)
  * restricted shell (scripts/mcp_shell.py): one tool that runs a typed command line whose
    program is on an allowlist, after the user confirms it in the terminal running serve.py

API (all under /api/mcp, same origin as the page):
  GET  /servers            configured servers: [{"id", "name", "kind"}], plus "confirm" for a
                           restricted shell
  POST /<id>/rpc           one JSON-RPC message; a request returns its response, anything
                           else 202
  GET  /events             server-sent events of every server on one stream, each tagged
                           with "server": <id> — the page uses this, since browsers allow only
                           6 connections per host and each stream holds one open
  GET  /<id>/events        the same for one server: messages it pushes, stderr lines,
                           process status
  POST /<id>/restart       restart the server process / reset the HTTP session
  GET  /whoami             {"mode": "local"} or {"mode": "remote", "expires_in": seconds}
  POST /login, /logout     remote access only: {"password": …} → session cookie

Security: whoever can reach this API can run the configured programs, so every request
must come from this machine (loopback address, Host header naming localhost, no proxy
forwarding headers), must not come from another origin (Origin header), and must carry the
random token that the page receives as a SameSite=Strict, HttpOnly cookie. With a "remote"
section in the config (scripts/mcp_remote.py), requests for the listed domain names are
served too, but only with a session from POST /login (password in WEBTT_MCP_PASSWORD), and
only for servers open to remote use (restricted shells need "remote": true). Only servers and tools listed in the config file can
be used; the browser can never name a program or its arguments — except in a restricted shell,
which the config must name explicitly and where every command is confirmed in the terminal.
"""
import hmac
import json
import os
import sys
import pathlib
import queue
import re
import secrets
import ssl
import subprocess
import threading
import time
import traceback
import urllib.error
import urllib.request

ID_RE = re.compile(r'^[A-Za-z0-9_-]{1,64}$')
TOOL_RE = re.compile(r'^[A-Za-z0-9_.-]{1,64}$')
PLACEHOLDER_RE = re.compile(r'\{([A-Za-z0-9_]+)\}')
COOKIE = 'mcp_token'
MAX_BODY = 4 * 1024 * 1024          # largest JSON-RPC message accepted from the page
MAX_OUTPUT = 1024 * 1024            # CLI stdout / stderr kept per call (the rest is read and dropped)
MAX_LINE = 16 * 1024 * 1024         # longest stdout line (one JSON-RPC message) read from a server
MAX_STDERR_LINE = 64 * 1024
DEFAULT_TIMEOUT = 30                # seconds to wait for a response
KEEPALIVE = 15                      # seconds between SSE comments on an idle stream
EVENT_QUEUE = 1000                  # events buffered per server for a stalled page (then dropped)
PROTOCOL_VERSIONS = ['2025-06-18', '2025-03-26', '2024-11-05']
MAX_LOGIN_BODY = 4096
# A proxy in front of this server adds these; a request carrying one is never "local", even
# if the proxy rewrote Host to localhost
FORWARDING_HEADERS = ('X-Forwarded-For', 'X-Forwarded-Host', 'X-Forwarded-Proto', 'X-Real-IP', 'Forwarded')


class BridgeError(Exception):
    def __init__(self, status, message, extra=None, headers=None):
        super().__init__(message)
        self.status = status
        self.extra = extra or {}       # more fields for the JSON body
        self.headers = headers or {}


# ── Config ─────────────────────────────────────────────

def load_config(path):
    """Read and validate the config file. Returns (servers by id, base directory)."""
    path = pathlib.Path(path).resolve()
    data = json.loads(path.read_text(encoding='utf-8'))
    servers = {}
    for s in data.get('servers', []):
        sid = s.get('id', '')
        if not ID_RE.match(sid):
            raise ValueError(f'server id {sid!r}: use 1-64 letters, digits, "-" or "_"')
        if sid in servers:
            raise ValueError(f'duplicate server id {sid!r}')
        kinds = [k for k in ('command', 'url', 'cli', 'shell') if k in s]
        if len(kinds) != 1:
            raise ValueError(f'server {sid!r}: give exactly one of "command", "url", "cli" or "shell"')
        kind = kinds[0]
        if 'remote' in s and not isinstance(s['remote'], bool):
            raise ValueError(f'server {sid!r}: "remote" must be true or false')
        if kind == 'command' and not (isinstance(s['command'], list) and s['command'] and all(isinstance(a, str) for a in s['command'])):
            raise ValueError(f'server {sid!r}: "command" must be a non-empty list of strings')
        if kind == 'url' and not re.match(r'^https?://', s['url']):
            raise ValueError(f'server {sid!r}: "url" must start with http:// or https://')
        if kind == 'cli':
            names = set()
            for t in s['cli']:
                if not TOOL_RE.match(t.get('name', '')) or t['name'] in names:
                    raise ValueError(f'server {sid!r}: bad or duplicate tool name {t.get("name")!r}')
                names.add(t['name'])
                if not (isinstance(t.get('command'), list) and t['command'] and all(isinstance(a, str) for a in t['command'])):
                    raise ValueError(f'tool {t["name"]!r}: "command" must be a non-empty list of strings')
                if PLACEHOLDER_RE.search(t['command'][0]):
                    raise ValueError(f'tool {t["name"]!r}: the program (first element) must not contain a placeholder')
        if kind == 'shell':
            import mcp_shell  # imported here: mcp_shell builds on this module
            mcp_shell.validate(sid, s['shell'])
        servers[sid] = {**s, 'kind': {'command': 'stdio', 'url': 'http', 'cli': 'cli', 'shell': 'shell'}[kind]}
    return servers, path.parent


# ── Events (server-sent events to the page) ────────────

class Events:
    """One server's event hub. Every event is tagged with the server's id here, from where it
    was produced — never from the message a server sent — so a merged stream can't be fooled."""

    def __init__(self, sid=None):
        self.sid = sid
        self.lock = threading.Lock()
        self.subscribers = []

    def subscribe(self, q=None):
        """Deliver events to `q` (a new bounded queue by default; the merged stream passes one
        queue to every hub)."""
        q = q or queue.Queue(maxsize=EVENT_QUEUE)
        with self.lock:
            self.subscribers.append(q)
        return q

    def unsubscribe(self, q):
        with self.lock:
            if q in self.subscribers:
                self.subscribers.remove(q)

    def publish(self, event):
        text = json.dumps({**event, 'server': self.sid} if self.sid else event, ensure_ascii=False)
        with self.lock:
            for q in self.subscribers:
                try:
                    q.put_nowait(text)
                except queue.Full:  # a stalled page drops events rather than blocking the server
                    pass


def bounded_lines(pipe, limit, name, events):
    """Lines from a text pipe, never holding more than `limit` characters: an over-long line
    is read to its end, dropped and reported, so a misbehaving server can't exhaust memory."""
    while True:
        line = pipe.readline(limit + 1)
        if not line:
            return
        if len(line) > limit and not line.endswith('\n'):
            while True:
                rest = pipe.readline(limit + 1)
                if not rest or rest.endswith('\n'):
                    break
            events.publish({'type': 'stderr', 'line': f'[{name}] 一行超过 {limit} 个字符，已丢弃'})
            continue
        yield line


def message_id(msg):
    return json.dumps(msg.get('id'), sort_keys=True)


def is_request(msg):
    return 'method' in msg and 'id' in msg


def restore_id(response_text, client_id):
    reply = json.loads(response_text)
    reply['id'] = client_id
    return json.dumps(reply, ensure_ascii=False)


# ── stdio servers ──────────────────────────────────────

class StdioSession:
    def __init__(self, cfg, base, events):
        self.cfg, self.base, self.events = cfg, base, events
        self.timeout = float(cfg.get('timeout_s', DEFAULT_TIMEOUT))
        self.proc = None
        self.lock = threading.Lock()          # start / stop / write / ids
        self.pending = {}                     # bridge's request id → [Event, response text or None]
        self.counter = 0

    def _alive(self):
        return self.proc is not None and self.proc.poll() is None

    @staticmethod
    def _close(proc):
        for pipe in (proc.stdin, proc.stdout, proc.stderr):
            try:
                pipe.close()
            except OSError:
                pass

    def start(self):
        if self.proc is not None:  # an exited process: release its pipes before replacing it
            self._close(self.proc)
        cwd = (self.base / self.cfg.get('cwd', '.')).resolve()
        env = {**os.environ, **{k: str(v) for k, v in self.cfg.get('env', {}).items()}}
        try:
            self.proc = subprocess.Popen(self.cfg['command'], cwd=cwd, env=env, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                                         stderr=subprocess.PIPE, text=True, encoding='utf-8', errors='replace', bufsize=1)
        except OSError as e:
            raise BridgeError(502, f'无法启动 {self.cfg["command"][0]}：{e}')
        threading.Thread(target=self._read_stdout, args=(self.proc,), daemon=True).start()
        threading.Thread(target=self._read_stderr, args=(self.proc,), daemon=True).start()
        self.events.publish({'type': 'status', 'status': 'started', 'pid': self.proc.pid})

    def _read_stdout(self, proc):
        for line in bounded_lines(proc.stdout, MAX_LINE, 'stdout', self.events):
            line = line.strip()
            if not line:
                continue
            try:
                msg = json.loads(line)
            except json.JSONDecodeError:
                self.events.publish({'type': 'stderr', 'line': f'[stdout 不是 JSON] {line[:500]}'})
                continue
            waiter = self.pending.get(message_id(msg)) if isinstance(msg, dict) and 'method' not in msg else None
            if waiter:
                waiter[1] = line
                waiter[0].set()
            else:
                self.events.publish({'type': 'message', 'message': msg})
        code = proc.wait()
        self.events.publish({'type': 'status', 'status': 'exited', 'code': code})
        for waiter in list(self.pending.values()):  # fail whatever was still waiting
            waiter[0].set()

    def _read_stderr(self, proc):
        for line in bounded_lines(proc.stderr, MAX_STDERR_LINE, 'stderr', self.events):
            self.events.publish({'type': 'stderr', 'line': line.rstrip('\n')[:2000]})

    def rpc(self, msg, raw):
        waiter, key = None, None
        with self.lock:
            if not self._alive():
                self.start()
            if is_request(msg):
                # Several clients (tabs) may use the same ids: the server sees the bridge's own
                # unique id, and the response gets the client's id back
                self.counter += 1
                internal = f'bridge-{self.counter}'
                key = json.dumps(internal)
                waiter = [threading.Event(), None]
                self.pending[key] = waiter
                raw = json.dumps({**msg, 'id': internal}, ensure_ascii=False)
            try:
                self.proc.stdin.write(raw.replace('\n', ' ') + '\n')
                self.proc.stdin.flush()
            except (BrokenPipeError, OSError):
                if key:
                    self.pending.pop(key, None)
                raise BridgeError(502, '服务端进程已退出')
        if waiter is None:
            return None
        try:
            if not waiter[0].wait(self.timeout):
                raise BridgeError(504, f'{self.timeout:g} 秒内没有收到响应')
            if waiter[1] is None:
                raise BridgeError(502, '服务端进程在响应前退出了')
            return restore_id(waiter[1], msg['id'])
        finally:
            self.pending.pop(key, None)

    def stop(self):
        with self.lock:
            if self.proc is None:
                return
            if self._alive():
                self.proc.terminate()
                try:
                    self.proc.wait(3)
                except subprocess.TimeoutExpired:
                    self.proc.kill()
                    self.proc.wait()
            self._close(self.proc)
            self.proc = None


# ── HTTP servers (Streamable HTTP) ─────────────────────

class HttpSession:
    def __init__(self, cfg, base, events):
        self.cfg, self.events = cfg, events
        self.timeout = float(cfg.get('timeout_s', DEFAULT_TIMEOUT))
        self.session_id = None
        self.protocol = None
        self.lock = threading.Lock()
        self.counter = 0

    def rpc(self, msg, raw):
        headers = {'Content-Type': 'application/json', 'Accept': 'application/json, text/event-stream', **self.cfg.get('headers', {})}
        with self.lock:
            if self.session_id:
                headers['Mcp-Session-Id'] = self.session_id
            if self.protocol:
                headers['MCP-Protocol-Version'] = self.protocol
            if is_request(msg):  # unique ids per bridge, as for stdio servers
                self.counter += 1
                internal = f'bridge-{self.counter}'
                raw = json.dumps({**msg, 'id': internal}, ensure_ascii=False)
        req = urllib.request.Request(self.cfg['url'], data=raw.encode(), headers=headers, method='POST')
        try:
            resp = urllib.request.urlopen(req, timeout=self.timeout, context=ssl.create_default_context())
        except urllib.error.HTTPError as e:
            body = e.read(2000).decode('utf-8', 'replace')
            raise BridgeError(502, f'MCP 服务端返回 HTTP {e.code}：{body}')
        except (urllib.error.URLError, OSError) as e:
            raise BridgeError(502, f'无法连接 {self.cfg["url"]}：{e}')
        with resp:
            if resp.headers.get('Mcp-Session-Id'):
                with self.lock:
                    self.session_id = resp.headers['Mcp-Session-Id']
            if resp.status == 202 or not is_request(msg):
                return None
            kind = resp.headers.get('Content-Type', '')
            if kind.startswith('text/event-stream'):
                answer = self._read_sse(resp, json.dumps(internal))
            else:
                answer = resp.read(MAX_BODY).decode('utf-8', 'replace')
        try:
            answer = restore_id(answer, msg['id'])
        except (ValueError, TypeError):
            raise BridgeError(502, 'MCP 服务端返回的不是 JSON')
        if msg.get('method') == 'initialize':  # later requests must say which version was agreed
            try:
                with self.lock:
                    self.protocol = json.loads(answer)['result']['protocolVersion']
            except (ValueError, KeyError, TypeError):
                pass
        return answer

    def _read_sse(self, resp, wanted):
        """Unpack an SSE response: return the matching response, publish everything else."""
        data = []
        for raw in resp:
            line = raw.decode('utf-8', 'replace').rstrip('\r\n')
            if line.startswith('data:'):
                data.append(line[5:].lstrip())
            elif line == '' and data:
                text, data = '\n'.join(data), []
                try:
                    msg = json.loads(text)
                except json.JSONDecodeError:
                    continue
                if 'method' not in msg and message_id(msg) == wanted:
                    return text
                self.events.publish({'type': 'message', 'message': msg})
        raise BridgeError(502, 'SSE 流结束了，但没有收到响应')

    def stop(self):
        with self.lock:
            self.session_id = None
            self.protocol = None


# ── Command-line tools as a virtual MCP server ─────────

def argv_for(command, args, allow_option_values=False):
    """Fill {placeholders} with argument values. An element whose placeholder has no value
    is left out, so optional flags disappear when not given. Values are separate argv
    elements for execve: no shell ever sees them.

    A value that fills a whole element and starts with "-" would be read by most programs
    as an option ("--upload-pack=…"), so such string values are refused unless the tool sets
    allow_option_values (numbers are fine). Putting a literal "--" before positional
    placeholders in the template is the other common fix."""
    out = []
    for part in command:
        names = PLACEHOLDER_RE.findall(part)
        if any(n not in args for n in names):
            continue
        whole = PLACEHOLDER_RE.fullmatch(part)
        if whole and not allow_option_values and isinstance(args[whole.group(1)], str) and args[whole.group(1)].startswith('-'):
            raise ValueError(f'参数 {whole.group(1)} 以 "-" 开头，可能被程序当成选项；如需允许，请在配置里给这个工具加上 "allow_option_values": true')
        def text(v):
            if isinstance(v, bool):
                return 'true' if v else 'false'
            return v if isinstance(v, str) else json.dumps(v, ensure_ascii=False)
        out.append(PLACEHOLDER_RE.sub(lambda m: text(args[m.group(1)]), part))
    return out


class CliSession:
    def __init__(self, cfg, base, events):
        self.cfg, self.base, self.events = cfg, base, events
        self.tools = {t['name']: t for t in cfg['cli']}

    def rpc(self, msg, raw):
        if not is_request(msg):
            return None
        method, params = msg['method'], msg.get('params') or {}
        if method == 'initialize':
            asked = params.get('protocolVersion')
            result = {'protocolVersion': asked if asked in PROTOCOL_VERSIONS else PROTOCOL_VERSIONS[0],
                      'capabilities': {'tools': {}},
                      'serverInfo': {'name': f'cli:{self.cfg["id"]}', 'title': self.cfg.get('name', self.cfg['id']), 'version': '1.0.0'},
                      'instructions': '命令行工具包装成的虚拟 MCP 服务端，每个工具运行一个登记在配置文件里的程序。'}
        elif method == 'ping':
            result = {}
        elif method == 'tools/list':
            result = {'tools': [{'name': t['name'], 'description': t.get('description', ''),
                                 'inputSchema': t.get('input_schema', {'type': 'object', 'properties': {}})} for t in self.cfg['cli']]}
        elif method == 'tools/call':
            tool = self.tools.get(params.get('name'))
            if tool is None:
                return json.dumps({'jsonrpc': '2.0', 'id': msg['id'], 'error': {'code': -32602, 'message': f'Unknown tool: {params.get("name")}'}})
            result = self._run(tool, params.get('arguments') or {})
        else:
            return json.dumps({'jsonrpc': '2.0', 'id': msg['id'], 'error': {'code': -32601, 'message': f'Method not found: {method}'}})
        return json.dumps({'jsonrpc': '2.0', 'id': msg['id'], 'result': result}, ensure_ascii=False)

    def _run(self, tool, args):
        error = tool_error
        try:
            argv = argv_for(tool['command'], args, tool.get('allow_option_values', False))
        except ValueError as e:
            return error(str(e))
        if not argv:
            return error('命令为空：模板里的元素都缺少参数值')
        stdin_arg = tool.get('stdin')
        stdin = args.get(stdin_arg) if stdin_arg else None
        if stdin is not None and not isinstance(stdin, str):
            stdin = json.dumps(stdin, ensure_ascii=False)
        timeout = float(tool.get('timeout_s', DEFAULT_TIMEOUT))
        cwd = (self.base / tool.get('cwd', self.cfg.get('cwd', '.'))).resolve()
        return run_program(argv, cwd, stdin, timeout, self.events)

    def stop(self):
        pass


def tool_error(text):
    return {'content': [{'type': 'text', 'text': text}], 'isError': True}


def run_program(argv, cwd, stdin, timeout, events):
    """Run argv (no shell) and return an MCP tool result: stdout, stderr, exit code. Output
    beyond MAX_OUTPUT is read and dropped; the program is killed after `timeout` seconds."""
    error = tool_error
    events.publish({'type': 'status', 'status': 'run', 'argv': argv})
    t0 = time.monotonic()
    try:
        proc = subprocess.Popen(argv, cwd=cwd, stdin=subprocess.PIPE if stdin is not None else subprocess.DEVNULL,
                                stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    except FileNotFoundError:
        return error(f'找不到程序：{argv[0]}')
    except OSError as e:
        return error(f'无法运行：{e}')
    out, err, truncated = bytearray(), bytearray(), [False]

    def pump(pipe, buf):
        # Keep at most MAX_OUTPUT bytes, but keep reading so the program never blocks
        while chunk := pipe.read1(65536):
            room = MAX_OUTPUT - len(buf)
            if len(chunk) > room:
                truncated[0] = True
            buf += chunk[:max(room, 0)]
        pipe.close()

    def feed():
        try:
            proc.stdin.write(stdin.encode())
            proc.stdin.close()
        except OSError:
            pass
    workers = [threading.Thread(target=pump, args=(proc.stdout, out), daemon=True),
               threading.Thread(target=pump, args=(proc.stderr, err), daemon=True)]
    if stdin is not None:
        workers.append(threading.Thread(target=feed, daemon=True))
    for w in workers:
        w.start()
    try:
        code = proc.wait(timeout)
    except subprocess.TimeoutExpired:
        proc.kill()
        proc.wait()
        return error(f'超过 {timeout:g} 秒未结束，已终止')
    finally:
        for w in workers:
            w.join(5)
    ms = (time.monotonic() - t0) * 1000
    content = [{'type': 'text', 'text': out.decode('utf-8', 'replace')}]
    if err:
        content.append({'type': 'text', 'text': '[stderr]\n' + err.decode('utf-8', 'replace')})
    note = f'（输出超过 {MAX_OUTPUT} 字节，已截断）' if truncated[0] else ''
    content.append({'type': 'text', 'text': f'退出码 {code}  ·  {ms:.0f}ms{note}'})
    return {'content': content, 'isError': code != 0,
            'structuredContent': {'exitCode': code, 'milliseconds': round(ms), 'truncated': truncated[0]}}


# ── The bridge ─────────────────────────────────────────

class Bridge:
    def __init__(self, config_path, port, confirm=None, env=None):
        import mcp_remote
        self.servers, self.base = load_config(config_path)
        self.config_path = pathlib.Path(config_path).resolve()
        section = json.loads(self.config_path.read_text(encoding='utf-8')).get('remote')
        self.remote = mcp_remote.RemoteAccess.from_config(section, os.environ if env is None else env)
        self.port = port
        self.confirm = confirm  # restricted shell prompt; None → ask in the terminal (mcp_shell)
        self.token = secrets.token_urlsafe(32)
        self.events = {sid: Events(sid) for sid in self.servers}
        self.sessions = {}
        self.lock = threading.Lock()

    def startup_lines(self):
        """What serve.py prints at startup: the config file actually loaded, and for each
        restricted shell whether its commands wait for the terminal."""
        lines = [f'MCP bridge: {len(self.servers)} server(s) from {self.config_path}']
        for s in self.servers.values():
            if s['kind'] == 'shell':
                mode = '每条命令在终端确认' if s['shell'].get('confirm', True) else '免确认，直接执行'
                lines.append(f'  受限 shell {s["id"]}：{mode}（允许 {"、".join(s["shell"]["allow"])}）')
        if self.remote:
            lines += self.remote.startup_lines()
            reachable = [sid for sid in self.servers if self.open_remotely(sid)]
            lines.append(f'  远程可用的服务端：{"、".join(reachable) or "（无）"}')
        return lines

    def open_remotely(self, sid):
        """Whether a remote (logged-in) page may use this server: yes by default, except
        restricted shells, which need "remote": true."""
        s = self.servers[sid]
        return s.get('remote', s['kind'] != 'shell')

    def server_list(self, remote=False):
        """What the page shows: id, name, kind, and for a restricted shell whether each
        command waits for the terminal. A remote page sees only the servers open to it."""
        out = []
        for s in self.servers.values():
            if remote and not self.open_remotely(s['id']):
                continue
            entry = {'id': s['id'], 'name': s.get('name', s['id']), 'kind': s['kind']}
            if s['kind'] == 'shell':
                entry['confirm'] = s['shell'].get('confirm', True)
            out.append(entry)
        return out

    def session(self, sid):
        with self.lock:
            if sid not in self.sessions:
                cfg = self.servers[sid]
                if cfg['kind'] == 'shell':
                    import mcp_shell
                    self.sessions[sid] = mcp_shell.ShellSession(cfg, self.base, self.events[sid], confirm=self.confirm,
                                                                source=self.config_path)
                else:
                    cls = {'stdio': StdioSession, 'http': HttpSession, 'cli': CliSession}[cfg['kind']]
                    self.sessions[sid] = cls(cfg, self.base, self.events[sid])
            return self.sessions[sid]

    # Requests must come from this machine and this site, with the token cookie
    def allowed_hosts(self):
        return {f'localhost:{self.port}', f'127.0.0.1:{self.port}', f'[::1]:{self.port}'}

    def local_page_request(self, handler):
        client = handler.client_address[0]
        loopback = client in ('127.0.0.1', '::1') or client.startswith('::ffff:127.')
        forwarded = any(h in handler.headers for h in FORWARDING_HEADERS)
        # With remote access on, a proxy rewriting Host could fake all of the above, so
        # localhost is trusted only if the config says so
        trusted = self.remote is None or self.remote.trust_localhost
        return trusted and loopback and not forwarded and handler.headers.get('Host', '') in self.allowed_hosts()

    @staticmethod
    def cookies(handler):
        return dict(c.strip().split('=', 1) for c in handler.headers.get('Cookie', '').split(';') if '=' in c)

    def check(self, handler):
        """Refuse the request unless it is local (with the token) or for a remote host.
        Returns None for local, or the Host of a remote request (whose session the caller
        checks, since /login comes before there is one)."""
        if self.local_page_request(handler):
            origin = handler.headers.get('Origin')
            if origin is not None and origin not in {f'http://{h}' for h in self.allowed_hosts()}:
                raise BridgeError(403, f'拒绝来自 {origin} 的请求')
            if not hmac.compare_digest(self.cookies(handler).get(COOKIE, ''), self.token):
                raise BridgeError(403, '缺少或错误的令牌：请从本机打开 /www/mcp/ 页面')
            return None
        host = handler.headers.get('Host', '')
        local_name = host.lower() in self.allowed_hosts() and not self.remote.trust_localhost if self.remote else False
        if self.remote and (self.remote.is_remote_host(host) or local_name):
            origin = handler.headers.get('Origin')
            if origin is not None and not self.remote.origin_ok(host, origin):
                raise BridgeError(403, f'拒绝来自 {origin} 的请求')
            if handler.headers.get('Sec-Fetch-Site') not in (None, 'same-origin'):
                raise BridgeError(403, '拒绝跨站请求')
            return host
        raise BridgeError(403, '只接受本机通过 localhost 发来的请求')

    def may_issue_cookie(self, handler):
        """Give the token only to a local, same-site page load (not to an <img>/<iframe>
        pointing here from another site)."""
        if not self.local_page_request(handler):
            return False
        site = handler.headers.get('Sec-Fetch-Site')
        origin = handler.headers.get('Origin')
        return site in (None, 'same-origin', 'none') and (origin is None or origin in {f'http://{h}' for h in self.allowed_hosts()})

    def cookie_header(self):
        # Path=/api: the compute API (scripts/compute_bridge.py) uses the same access rules
        return f'{COOKIE}={self.token}; Path=/api; HttpOnly; SameSite=Strict'

    def handle(self, handler, method):
        """Serve an /api/mcp request. Returns False if the path isn't ours."""
        path = handler.path.split('?', 1)[0]
        if not path.startswith('/api/mcp/'):
            return False
        try:
            remote_host = self.check(handler)
            remote = remote_host is not None
            parts = path[len('/api/mcp/'):].split('/')
            token = None
            if remote:
                if method == 'POST' and parts == ['login']:
                    self.login(handler, remote_host)
                    return True
                token = self.session_token(handler)
            visible = [sid for sid in self.servers if not remote or self.open_remotely(sid)]
            if method == 'GET' and parts == ['whoami']:
                self.reply_json(handler, 200, {'mode': 'remote', 'expires_in': self.remote.expires_in(token)} if remote else {'mode': 'local'})
            elif method == 'POST' and parts == ['logout'] and remote:
                self.remote.logout(token)
                self.log_remote(handler, 'logout')
                self.reply_json(handler, 200, {'ok': True}, {'Set-Cookie': self.remote.clear_cookie_header()})
            elif method == 'GET' and parts == ['servers']:
                self.reply_json(handler, 200, self.server_list(remote))
            elif method == 'GET' and parts == ['events']:
                self.stream(handler, [self.events[sid] for sid in visible])
            elif len(parts) == 2 and parts[0] in visible:
                sid, action = parts
                if method == 'POST' and action == 'rpc':
                    self.rpc(handler, sid)
                elif method == 'GET' and action == 'events':
                    self.stream(handler, [self.events[sid]])
                elif method == 'POST' and action == 'restart':
                    self.session(sid).stop()
                    self.reply_json(handler, 200, {'ok': True})
                else:
                    raise BridgeError(404, '未知操作')
            else:
                raise BridgeError(404, '未知的服务端或路径')
        except BridgeError as e:
            self.reply_json(handler, e.status, {'error': str(e), **e.extra}, e.headers)
        except Exception as e:  # never leave a request without an answer
            traceback.print_exc()
            try:
                self.reply_json(handler, 500, {'error': f'桥接服务内部错误：{e}'})
            except OSError:
                pass
        return True

    def session_token(self, handler):
        """The remote request's session token; 401 (with login: true) if it has none."""
        token = self.cookies(handler).get(self.remote_cookie_name(), '')
        if not self.remote.valid(token):
            raise BridgeError(401, '需要登录', {'login': True})
        return token

    def authorize(self, handler):
        """The access rules of this API, for other APIs of serve.py (the compute API):
        'local' or 'remote', or raises BridgeError (403, or 401 when a login would help)."""
        if self.check(handler) is None:
            return 'local'
        self.session_token(handler)
        return 'remote'

    @staticmethod
    def remote_cookie_name():
        import mcp_remote
        return mcp_remote.SESSION_COOKIE

    def login(self, handler, host):
        """POST /login from a remote page: password → session cookie."""
        origin = handler.headers.get('Origin')
        if origin is None or not self.remote.origin_ok(host, origin):  # a login form on another site can't log in
            raise BridgeError(403, '登录请求必须来自本站页面')
        try:
            length = int(handler.headers.get('Content-Length') or 0)
        except ValueError:
            raise BridgeError(400, 'Content-Length 不是数字')
        if not 0 < length <= MAX_LOGIN_BODY:
            raise BridgeError(400, '请求体为空或过大')
        try:
            body = json.loads(handler.rfile.read(length))
        except (json.JSONDecodeError, UnicodeDecodeError):
            raise BridgeError(400, '不是合法的 JSON')
        if not (isinstance(body, dict) and isinstance(body.get('password'), str)):
            raise BridgeError(400, '需要 {"password": "…"}')
        result = self.remote.login(body['password'])
        if result.retry_after:
            self.log_remote(handler, f'login refused: locked for {result.retry_after} s')
            raise BridgeError(429, f'失败次数过多，请 {result.retry_after} 秒后再试', {'retry_after': result.retry_after, 'login': True},
                              {'Retry-After': str(result.retry_after)})
        if not result.token:
            self.log_remote(handler, 'login failed')
            raise BridgeError(401, '密码错误', {'login': True})
        self.log_remote(handler, 'login ok')
        self.reply_json(handler, 200, {'ok': True, 'expires_in': self.remote.session_seconds},
                        {'Set-Cookie': self.remote.cookie_header(result.token)})

    @staticmethod
    def log_remote(handler, what):
        """One line per login event for the log (pm2 logs), with the forwarded client address
        frp adds, escaped so a crafted header can't write control characters."""
        via = handler.headers.get('X-Forwarded-For') or handler.client_address[0]
        print(f'[mcp remote] {what} · client {via!r}', file=sys.stderr, flush=True)

    def rpc(self, handler, sid):
        try:
            length = int(handler.headers.get('Content-Length') or 0)
        except ValueError:
            raise BridgeError(400, 'Content-Length 不是数字')
        if length <= 0 or length > MAX_BODY:
            raise BridgeError(413 if length > MAX_BODY else 400, '请求体为空或过大')
        raw = handler.rfile.read(length).decode('utf-8', 'replace')
        try:
            msg = json.loads(raw)
        except json.JSONDecodeError as e:
            raise BridgeError(400, f'不是合法的 JSON：{e}')
        if not isinstance(msg, dict) or msg.get('jsonrpc') != '2.0':
            raise BridgeError(400, '需要一条 JSON-RPC 2.0 消息')
        answer = self.session(sid).rpc(msg, raw)
        if answer is None:
            handler.send_response(202)
            handler.send_header('Content-Length', '0')
            handler.end_headers()
        else:
            self.reply_raw(handler, 200, answer.encode())

    def stream(self, handler, hubs):
        """Server-sent events from `hubs` on one response, until the page goes away."""
        q = queue.Queue(maxsize=EVENT_QUEUE * len(hubs))  # so one busy server can't crowd out the rest
        for hub in hubs:
            hub.subscribe(q)
        try:
            handler.send_response(200)
            handler.send_header('Content-Type', 'text/event-stream; charset=utf-8')
            handler.send_header('Cache-Control', 'no-cache')
            handler.end_headers()
            handler.wfile.write(b': connected\n\n')
            handler.wfile.flush()
            while True:
                try:
                    text = q.get(timeout=KEEPALIVE)
                    handler.wfile.write(f'data: {text}\n\n'.encode())
                except queue.Empty:
                    handler.wfile.write(b': keepalive\n\n')
                handler.wfile.flush()
        except (BrokenPipeError, ConnectionResetError, OSError):
            pass
        finally:
            for hub in hubs:
                hub.unsubscribe(q)

    @staticmethod
    def reply_raw(handler, status, body, kind='application/json; charset=utf-8', headers=None):
        handler.send_response(status)
        handler.send_header('Content-Type', kind)
        for name, value in (headers or {}).items():
            handler.send_header(name, value)
        handler.send_header('Content-Length', str(len(body)))
        handler.end_headers()
        handler.wfile.write(body)

    def reply_json(self, handler, status, obj, headers=None):
        self.reply_raw(handler, status, json.dumps(obj, ensure_ascii=False).encode(), headers=headers)

    def shutdown(self):
        for s in self.sessions.values():
            s.stop()