From 6f933e1c80cebc2797ed4610d7cb1a8d9c462fb5 Mon Sep 17 00:00:00 2001 From: Donald Date: Sat, 3 Oct 2026 02:51:44 +0000 Subject: [PATCH] Add Herdr update and live handoff menu actions The Ctrl+A h command menu gains two host-scoped actions: a plain `herdr update` that streams updater output into the popup, and an experimental update-and-hot-restart flow that verifies a live server handoff and reconnects the originating Hub tab once the popup closes. Remote updates use the remote host's own proxy settings. A standalone helper reads literal proxy assignments from shell profiles and running Clash/Mihomo listeners without executing startup files; Windows reads process, user and machine proxy variables through PowerShell. --- README.md | 53 ++++++- herdr_hub/cli.py | 12 +- herdr_hub/control.py | 279 ++++++++++++++++++++++++++++----- herdr_hub/proxy_env.py | 184 ++++++++++++++++++++++ herdr_hub/remotes.py | 16 +- herdr_hub/restart.py | 97 ++++++++++++ tests/test_control.py | 227 +++++++++++++++++++++++++++ tests/test_proxy_env.py | 126 +++++++++++++++ tests/test_restart.py | 152 ++++++++++++++++++ tests/test_tmux_integration.py | 22 +++ 10 files changed, 1130 insertions(+), 38 deletions(-) create mode 100644 herdr_hub/proxy_env.py create mode 100644 herdr_hub/restart.py create mode 100644 tests/test_proxy_env.py create mode 100644 tests/test_restart.py diff --git a/README.md b/README.md index d02a1dc..424a6ae 100644 --- a/README.md +++ b/README.md @@ -110,7 +110,7 @@ Hub 使用独立 tmux socket `herdr-hub`，不会接管普通 tmux server。外 ### Fork 当前 Agent 到新 Tab 按 **Ctrl+A h** 打开独立的 Herdr 命令菜单，选择“Fork 当前会话到新 Tab”， -按 Enter 或单击执行，Esc 关闭。菜单只显示当前 Agent、Workspace、主操作和必要的状态信息。 +按 Enter 或单击执行，Esc 关闭。菜单显示当前 Agent、Workspace、Fork 和更新操作及必要的状态信息。 操作正在执行时需等待结果后再关闭，以免将关闭弹窗误认为取消已经发送的请求。 Sessions 弹窗仍使用 `Ctrl+A g`；两个菜单都使用 Hub 外层的 `Ctrl+A` 前缀。 快捷键集中定义在 `herdr_hub/control.py` 的 `CONTROL_KEY` 常量中。 @@ -124,6 +124,9 @@ Codex 执行 `codex fork UUID`；Claude 执行 `claude --resume SESSION_ID --for 支持本地 default / named session，以及通过 SSH 控制的远端 native / Windows SSH endpoint。 控制目标取自触发菜单的 Hub window metadata，不使用 popup 继承的 Herdr socket 或 pane 环境变量。 目标必须运行 Herdr 0.9+，并安装对应 Agent；当前 pane 必须已报告可用的 Agent session ID。 +Fork 显示灰色时，菜单会说明原因；按 `Ctrl+R` 可重新读取当前 Agent 信息。 +若提示尚未上报会话 ID，在该主机检查 `herdr integration status`，确认对应 Agent 的集成已加载。 +识别出 Codex / Claude 并不代表已经取得 conversation ID；Hub 不会根据目录猜测会话。 远端控制使用非交互 SSH（需要已信任的主机密钥和可用认证）；Windows 需要 PowerShell。 启动 blocked、失败、超时或响应不完整时，菜单保留错误信息，**不会自动关闭任何 Tab**。 超时并不代表远端没有创建或启动成功，请先检查提示中的新 Tab，再重新打开菜单重试。 @@ -132,6 +135,54 @@ CLI 入口是 `herdr-hub control`。弹窗打开时会锁定 Hub 当前激活的 再从其 Endpoint 对应的 Herdr server 读取当前页面的 Agent conversation。 更新后运行 `herdr-hub refresh` 配置现有 Hub 的快捷键。`herdr-hub doctor` 会显示入口和版本要求。 +### 从命令菜单更新 Herdr + +需要让运行中的服务器也使用新版本时，按 **Ctrl+A h → 更新并热重启 Herdr**。 +此操作先更新当前 Tab 所在主机的 Herdr，再对当前 Session 执行实验性的 live handoff， +尝试保留 Agent、shell 和其他 pane 进程。它会核对服务器版本；确认使用更新后的版本后， +自动关闭菜单并重新连接打开菜单时的那个 Hub Tab。即使此前已安装新版本、服务器仍是旧版， +也会尝试热切换。该 Session 内的所有 Workspace/Tab 共用服务器，因此都会随之切换。 + +更新阶段使用 `herdr --session NAME update --handoff`，兼容 0.8.x 在非交互 SSH 下的更新要求。 +原生更新已经尝试热切换却未切换成功时，不会再次发送请求；仅在安装文件未变化、旧服务器仍运行时补做热切换。 + +热切换适用于支持该能力的 Linux/macOS Herdr 服务器；远端还需有 `python3`。 +Windows 不支持此入口，可以使用下面的普通更新。失败、超时、服务器不支持或未确认新版本时， +菜单保留诊断，不自动执行普通重启，也不重试热切换。连接中断时操作可能已经完成，请先检查 +目标 Session。普通重启会结束 pane 进程；缺少会话 ID 的 Agent 不能保证自动恢复。 + +按 **Ctrl+A h**，选择“更新 Herdr”，按 Enter 或单击执行。更新目标跟随打开菜单时的 +当前 Hub Tab：本地 Session 更新本机，远端 Session 通过 SSH 更新对应主机（包括 Windows）。 +菜单会显示目标主机；更新作用于该主机的 Herdr 安装，而非单个 Session。当前 pane 没有 +Agent 或无法读取 Agent 信息时，仍可使用更新操作。 + +该操作执行 `herdr update`，沿用目标主机已有的更新渠道，不传入 `--handoff`。 +它适用于 Herdr 自带安装器管理的安装；Homebrew、mise、Nix 等安装需使用对应包管理器更新。 +这个普通更新入口不自动重启 Session 或重新连接客户端。完成后可关闭菜单，按 `Ctrl+A R` 重新连接； +服务器端新功能可能仍需另行重启服务器。兼容的服务器会继续运行；旧版本升级若要求停止 +Session 或交互确认，本菜单不会代为确认，请根据命令输出在终端处理。 + +更新在后台执行，实时显示命令输出，最长等待 5 分钟。小窗口自动使用紧凑布局，保证结果区可见。 +下载器没有新输出时，菜单显示已执行时间和无新输出时间。 +读取 Agent 信息期间也可选择更新。结果留在菜单内，可用 Page Up/Down 翻阅，Esc 关闭。 +执行期间需等待结果；同一次菜单内更新只能执行一次，失败或超时也不会自动重试。 +超时或 SSH 断开不代表目标主机已取消更新，请检查目标状态后再重新打开菜单。 + +远端更新使用远端自己的代理。优先保留 SSH 进程已有的 `http_proxy`、`https_proxy`、 +`all_proxy`、`no_proxy`（兼容大写）；Linux/macOS 还通过远端 `python3` 读取 +`/etc/environment`、`/etc/profile`、`/etc/profile.d/*.sh`、`~/.config/environment.d/*.conf` +及用户 profile、bash/zsh 配置中的独立、无缩进的字面量赋值或 `export` 行。 +不会执行整个 shell 配置文件，也不解析命令替换、变量引用或条件表达式。 +没有上述代理变量时，还会读取远端 Clash Verge、Mihomo/Clash 常见运行配置中的 +`mixed-port`、`port` 或 `socks-port`，确认对应回环端口可连接后用于本次更新。 +Linux 下会优先从正在运行的 Mihomo/Clash 进程读取 `-f` / `-d` 指定的配置路径，支持 systemd 服务使用自定义目录。 +这覆盖了 TUN 模式下没有环境变量的情况，不会修改桌面系统代理开关。 +需要认证的应用代理、不在已支持路径中的配置及未运行的端口不会自动采用。 +Windows 普通更新会补读远端用户和系统的代理环境变量。 +检测结果只显示配置来源，不显示代理地址或凭据；配置仅传给本次更新进程。 +没有检测到代理时沿用远端网络和 curl 自身配置，不自动借用本机代理。 +“未检测到代理”只是网络方式说明；更新失败时，菜单会先显示末尾的具体错误，再列完整输出。 + ## 远端 Session 弹窗 弹窗由 tmux `display-popup` 和 Hub 内置选择器组成，不依赖 fzf。它会立即显示已打开的 diff --git a/herdr_hub/cli.py b/herdr_hub/cli.py index c40906e..2512aa4 100644 --- a/herdr_hub/cli.py +++ b/herdr_hub/cli.py @@ -1452,7 +1452,17 @@ def command_control(arguments: argparse.Namespace) -> int: # Endpoint metadata is authoritative; popup environment may belong to another server. binary = resolve_herdr() if window.endpoint.kind == "local" else "herdr" control = HerdrControl(window.endpoint, binary, hub.ssh_bin) - return curses.wrapper(ControlMenu(control).run) + menu = ControlMenu(control) + result = curses.wrapper(menu.run) + if menu.reconnect_requested: + # Close curses first; respawning the client must not destroy its popup. + current = next((item for item in hub.managed_windows() if item.window_id == window.window_id), None) + if current is None or current.endpoint != window.endpoint: + raise HubError("服务器已更新，但原 Hub Tab 已改变；请手动连接目标 Session") + hub.reconnect(window.window_id) + hub.wait_for_window_startup(window.window_id) + hub.display_message("Herdr 服务器已更新，当前 Tab 已重新连接") + return result def command_keys(arguments: argparse.Namespace) -> int: diff --git a/herdr_hub/control.py b/herdr_hub/control.py index d0c617b..898a34d 100644 --- a/herdr_hub/control.py +++ b/herdr_hub/control.py @@ -1,4 +1,4 @@ -"""Herdr command menu and non-destructive conversation forks.""" +"""Herdr command menu, updates, and conversation forks.""" from __future__ import annotations from concurrent.futures import ThreadPoolExecutor @@ -6,7 +6,11 @@ import curses from dataclasses import dataclass import json import os +from pathlib import Path +import shlex import subprocess +import sys +import tempfile import threading import time from typing import TYPE_CHECKING, Callable, Sequence @@ -26,6 +30,7 @@ if TYPE_CHECKING: CONTROL_KEY = "h" CONTROL_SHORTCUT = "C-a h" ROUTING_ENV = ( + "HERDR_ENV", "HERDR_SOCKET_PATH", "HERDR_CLIENT_SOCKET_PATH", "HERDR_SESSION", "HERDR_PANE_ID", "HERDR_TAB_ID", "HERDR_WORKSPACE_ID", "HERDR_ACTIVE_PANE_ID", "HERDR_ACTIVE_TAB_ID", "HERDR_ACTIVE_WORKSPACE_ID", @@ -40,6 +45,16 @@ class ControlCancelled(RuntimeError): """A read-only control request cancelled while closing the popup.""" +def update_failure(header: str, output: str) -> str: + """Put the final diagnostic ahead of setup chatter in small popups.""" + lines = [line.strip() for line in output.splitlines() if line.strip()] + diagnostic = next((line for line in reversed(lines) if line.startswith("curl:")), "") + if not diagnostic: + diagnostic = next((line for line in reversed(lines) + if not line.startswith("没有自动执行普通重启")), "无错误输出") + return f"{header}\n{diagnostic}\n\n完整命令输出（可翻页）：\n{output}" + + def object_field(data: dict, name: str) -> dict: value = data.get(name) if not isinstance(value, dict): @@ -99,7 +114,12 @@ def parse_snapshot(payload: object) -> ForkSource: raise ControlError("当前 pane 没有运行 Agent") if kind not in ("codex", "claude"): raise ControlError(f"暂不支持 Fork {kind}；支持 Codex / Claude") - session = object_field(pane, "agent_session") + session = pane.get("agent_session") + if not isinstance(session, dict): + raise ControlError( + f"当前 {kind.capitalize()} 尚未向 Herdr 上报会话 ID，暂时无法 Fork。\n" + f"在目标主机检查 herdr integration status；确认集成已加载后按 Ctrl+R 重读。" + ) if session.get("kind") != "id" or session.get("agent") != kind: raise ControlError("当前 Agent 尚未报告可 Fork 的 conversation ID") session_id = string_field(session, "value") @@ -135,8 +155,53 @@ class HerdrControl: command: Sequence[str], timeout: float, cancel_event: threading.Event | None = None, + progress: Callable[[str], None] | None = None, ) -> subprocess.CompletedProcess: environment = {k: v for k, v in os.environ.items() if k not in ROUTING_ENV} + if progress is not None: + # A file avoids pipe deadlocks and lets us show output before exit, + # including progress indicators that do not end in a newline. + try: + with tempfile.TemporaryFile() as output: + process = subprocess.Popen(command, stdout=output, stderr=subprocess.STDOUT, + stdin=subprocess.DEVNULL, env=environment) + deadline = time.monotonic() + timeout + started = time.monotonic() + last_output = started + last_notice = started + position = 0 + captured = b"" + try: + while True: + finished = process.poll() is not None + chunk = os.pread(output.fileno(), 65536, position) + if chunk: + position += len(chunk) + captured = (captured + chunk)[-65536:] + progress(decode_remote_output(captured).replace("\r", "\n")) + last_output = time.monotonic() + now = time.monotonic() + if not finished and now - last_notice >= 1: + progress( + decode_remote_output(captured).replace("\r", "\n") + + f"\n已执行 {int(now - started)} 秒 · {int(now - last_output)} 秒无新输出" + ) + last_notice = now + if finished and not chunk: + return subprocess.CompletedProcess(command, process.returncode, captured, b"") + if time.monotonic() >= deadline: + raise ControlError( + f"命令超时（{timeout:g}s）；操作可能已完成，请检查 Herdr\n" + + decode_remote_output(captured) + ) + if not chunk: + time.sleep(0.1) + finally: + if process.poll() is None: + process.kill() + process.wait() + except OSError as error: + raise ControlError(f"无法执行 Herdr 命令: {error}") from error if cancel_event is not None and cancel_event.is_set(): raise ControlCancelled() if cancel_event is not None: @@ -231,15 +296,7 @@ class HerdrControl: cancel_event: threading.Event | None = None, ) -> dict: argv = ["--session", self.endpoint.session, *arguments] - if self.endpoint.kind == "remote": - entry = RemoteEntry(self.endpoint.target or "", self.endpoint.remote_keybindings, - self.endpoint.remote_transport) - command = remote_control_command(self.ssh_bin, entry, argv, - platform=self._remote_platform(entry, cancel_event), - unset_environment=ROUTING_ENV) - else: - command = [self.herdr_bin, *argv] - result = self._execute(command, timeout, cancel_event) + result = self._execute(self._command(argv, cancel_event), timeout, cancel_event) stdout, stderr = decode_remote_output(result.stdout), decode_remote_output(result.stderr) if result.returncode != 0: raise ControlError((stderr or stdout or f"Herdr exited {result.returncode}").strip()) @@ -248,6 +305,85 @@ class HerdrControl: except json.JSONDecodeError as error: raise ControlError(f"Herdr 返回无效 JSON: {error}") from error + def _command( + self, argv: Sequence[str], cancel_event: threading.Event | None = None, + ) -> list[str]: + if self.endpoint.kind == "remote": + entry = RemoteEntry(self.endpoint.target or "", self.endpoint.remote_keybindings, + self.endpoint.remote_transport) + return remote_control_command(self.ssh_bin, entry, argv, + platform=self._remote_platform(entry, cancel_event), + unset_environment=ROUTING_ENV) + return [self.herdr_bin, *argv] + + def update(self, progress: Callable[[str], None] | None = None) -> str: + """Run the host updater once; its output is text, not control API JSON.""" + if progress: + progress("正在连接更新目标…") + command = self._command(["update"]) + if self.endpoint.kind == "remote": + command = self._update_proxy(command, progress) + result = self._execute(command, timeout=300, progress=progress) + stdout, stderr = decode_remote_output(result.stdout), decode_remote_output(result.stderr) + output = "\n".join(part.strip() for part in (stdout, stderr) if part.strip()) + if result.returncode != 0: + raise ControlError(update_failure(f"更新失败（退出码 {result.returncode}）", output)) + return "更新命令已完成。重新连接可使用更新后的客户端。\n" + output + + def _update_proxy( + self, command: list[str], progress: Callable[[str], None] | None, + ) -> list[str]: + """Load proxy settings on the remote host, never from the Hub host.""" + if self.platform == "windows": + entry = RemoteEntry(self.endpoint.target or "", self.endpoint.remote_keybindings, + self.endpoint.remote_transport) + return remote_control_command( + self.ssh_bin, entry, ["update"], platform="windows", + unset_environment=ROUTING_ENV, load_proxy_environment=True, + ) + if self.platform != "posix": + return command + if progress: + progress("正在检测远端自身的代理配置…") + helper = Path(__file__).with_name("proxy_env.py").read_text(encoding="utf-8") + result = list(command) + result[-1] = ( + "if command -v python3 >/dev/null 2>&1; then exec python3 -u -c " + + shlex.quote(helper) + " " + shlex.quote(command[-1]) + + "; else echo '远端没有 python3，沿用远端现有代理环境及 curl 配置'; " + + command[-1] + "; fi" + ) + return result + + def update_and_restart(self, progress: Callable[[str], None]) -> str: + """Update and verify a live server handoff on the selected POSIX session.""" + helper = str(Path(__file__).with_name("restart.py")) + if self.endpoint.kind == "local": + command = [sys.executable, helper, self.herdr_bin, self.endpoint.session] + else: + entry = RemoteEntry(self.endpoint.target or "", self.endpoint.remote_keybindings, + self.endpoint.remote_transport) + progress("正在检查远端热切换支持…") + if self._remote_platform(entry) != "posix": + raise ControlError("Windows 不支持热切换；请使用“更新 Herdr”，再安排普通重启。") + # Reuse the SSH destination/options, but run the standalone helper + # on the host that owns the Unix socket. No socket crosses SSH. + command = remote_control_command(self.ssh_bin, entry, [], platform="posix") + script = Path(helper).read_text(encoding="utf-8") + command[-1] = ( + "".join(f"unset {key}; " for key in ROUTING_ENV) + + 'if [ -x "$HOME/.local/bin/herdr" ]; then hub_herdr="$HOME/.local/bin/herdr"; ' + 'else hub_herdr=$(command -v herdr) || exit 127; fi; ' + + "exec python3 -u -c " + shlex.quote(script) + + ' "$hub_herdr" ' + shlex.quote(self.endpoint.session) + ) + command = self._update_proxy(command, progress) + result = self._execute(command, timeout=400, progress=progress) + output = decode_remote_output(result.stdout) + if result.returncode != 0: + raise ControlError(update_failure("更新/热切换失败；未自动执行普通重启。", output)) + return output + def source( self, cancel_event: threading.Event | None = None, @@ -319,10 +455,13 @@ class ControlMenu: "Fork 当前会话到新 Tab", "保留当前 Tab，在同一 Workspace 和工作目录继续", ), + ("update", "更新 Herdr", "更新当前主机的 Herdr（使用已配置的更新渠道）"), + ("update-restart", "更新并热重启 Herdr", "实验性热切换，保留进程；成功后自动重连当前 Tab"), ) def __init__(self, control: HerdrControl) -> None: self.control = control + self.reconnect_requested = False def run(self, screen) -> int: # type: ignore[no-untyped-def] import queue @@ -336,8 +475,13 @@ class ControlMenu: except (AttributeError, curses.error): pass source: ForkSource | None = None + source_error = "" status = "" selected = 0 + update_attempted = False + pending_update: str | None = None + status_offset = 0 + target = "本机" if self.control.endpoint.kind == "local" else self.control.endpoint.target cancel_loading = threading.Event() with ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(self.control.source, cancel_loading) @@ -346,22 +490,46 @@ class ControlMenu: while not messages.empty(): status = messages.get() if future is not None and future.done(): + # The worker is finished: discard queued progress before + # publishing its final result so stale output cannot replace it. + while not messages.empty(): + messages.get() + status_offset = 0 try: value = future.result() if phase == "load": source = value + source_error = "" status = "" + elif phase == "update": + status = value + elif phase == "update-restart": + self.reconnect_requested = True + return 0 else: return 0 except ControlCancelled: - return 0 + if not pending_update: + return 0 except Exception as error: status = str(error) + if phase == "load": + source_error = status if phase == "fork": source = None # Avoid accidental duplicate forks after ambiguous failure. + source_error = status future = None + if pending_update and future is None: + phase = pending_update + pending_update = None + update_attempted = True + method = self.control.update if phase == "update" else self.control.update_and_restart + future = executor.submit(method, messages.put) + status = f"正在更新 {target} 的 Herdr…" + status_offset = 0 screen.erase() height, width = screen.getmaxyx() + compact = height < 16 if source is not None: context = ( f"当前：{source.kind.capitalize()} · " @@ -371,11 +539,18 @@ class ControlMenu: context = "正在读取当前页面…" else: context = "无法读取当前页面" - self._add(screen, 1, 2, context, curses.A_BOLD) - action_top = 3 - for index, (_, label, detail) in enumerate(self.ACTIONS): - y = action_top + index * 2 - enabled = source is not None and future is None + if not compact: + self._add(screen, 1, 2, context, curses.A_BOLD) + heading = context if compact and selected == 0 else f"更新目标：{target}" + self._add(screen, 0 if compact else 2, 2, heading, curses.A_BOLD) + action_top = 1 if compact else 3 + action_stride = 1 if compact else 2 + for index, (action, label, detail) in enumerate(self.ACTIONS): + y = action_top + index * action_stride + enabled = ( + source is not None and future is None if action == "fork" + else not update_attempted and (future is None or phase == "load") + ) attribute = ( curses.A_REVERSE if enabled and index == selected @@ -383,21 +558,27 @@ class ControlMenu: if enabled else curses.A_DIM ) + if action == "fork" and source is None: + label += "（读取中）" if phase == "load" and future is not None else "（不可用）" self._add(screen, y, 2, f" {label} ", attribute) - self._add(screen, y + 1, 4, detail, curses.A_DIM) + if not compact: + self._add(screen, y + 1, 4, detail, curses.A_DIM) - status_top = action_top + len(self.ACTIONS) * 2 + 1 - for offset, line in enumerate( - display_chunks(status, max(1, width - 5)) if status else [] - ): + status_top = action_top + len(self.ACTIONS) * action_stride + (0 if compact else 1) + status_lines = display_chunks(status, max(1, width - 5)) if status else [] + visible_lines = max(1, height - status_top - 1) + status_offset = min(status_offset, max(0, len(status_lines) - visible_lines)) + if future is not None and phase in ("update", "update-restart"): + status_offset = max(0, len(status_lines) - visible_lines) + for offset, line in enumerate(status_lines[status_offset:status_offset + visible_lines]): self._add(screen, status_top + offset, 2, line, curses.A_DIM) footer = ( "操作执行中，请等待结果" - if future is not None and phase == "fork" - else "Enter 执行 · Esc 关闭" + if future is not None and phase != "load" + else "Enter 执行 · Ctrl+R 重读 · PgUp/PgDn 结果 · Esc 关闭" ) - self._add(screen, height - 2, 2, footer, curses.A_DIM) + self._add(screen, height - 1, 2, footer, curses.A_DIM) screen.refresh() key = screen.getch() if key == 27: @@ -406,7 +587,19 @@ class ControlMenu: return 0 if future is None: return 0 - status = "操作执行中；完成后可关闭（避免误以为已取消 Fork）。" + status = "操作执行中，请等待结果；关闭菜单不会取消已发送的操作。" + elif key == 18 and future is None: + source = None + source_error = "" + cancel_loading = threading.Event() + future = executor.submit(self.control.source, cancel_loading) + phase = "load" + status = "正在重新读取当前页面…" + status_offset = 0 + elif key in (curses.KEY_PPAGE, curses.KEY_NPAGE): + status_offset = max(0, status_offset + ( + visible_lines if key == curses.KEY_NPAGE else -visible_lines + )) elif key in (curses.KEY_UP, curses.KEY_DOWN): selected = (selected + (1 if key == curses.KEY_DOWN else -1)) % len(self.ACTIONS) else: @@ -414,14 +607,10 @@ class ControlMenu: if key == curses.KEY_MOUSE: try: _, x, y, _, state = curses.getmouse() - action_index = (y - action_top) // 2 + action_index = (y - action_top) // action_stride execute = ( 1 <= x < width - 1 and 0 <= action_index < len(self.ACTIONS) - and y in { - action_top + action_index * 2, - action_top + action_index * 2 + 1, - } and bool( state & ( @@ -434,10 +623,30 @@ class ControlMenu: selected = action_index except curses.error: execute = False - if execute and future is None and source is not None: - phase = "fork" - future = executor.submit(self.control.fork, source, messages.put) - status = "正在 Fork…" + if execute: + action = self.ACTIONS[selected][0] + if action in ("update", "update-restart") and not update_attempted and phase == "load" and future is not None: + pending_update = action + cancel_loading.set() + status = "正在准备更新…" + elif action == "fork" and source is not None and future is None: + phase = "fork" + future = executor.submit(self.control.fork, source, messages.put) + status = "正在 Fork…" + status_offset = 0 + elif action == "fork": + if future is None: + status = source_error or "尚未读取 Fork 会话信息，请按 Ctrl+R 重读。" + status_offset = 0 + elif phase == "load": + status = "正在读取 Fork 会话信息，请稍候再执行。" + elif action in ("update", "update-restart") and not update_attempted and future is None: + phase = action + update_attempted = True + method = self.control.update if action == "update" else self.control.update_and_restart + future = executor.submit(method, messages.put) + status = f"正在更新 {target} 的 Herdr，请等待…" + status_offset = 0 @staticmethod def _add( diff --git a/herdr_hub/proxy_env.py b/herdr_hub/proxy_env.py new file mode 100644 index 0000000..b2c7da5 --- /dev/null +++ b/herdr_hub/proxy_env.py @@ -0,0 +1,184 @@ +"""Read literal remote proxy settings without executing shell startup files. + +This standalone helper runs on the remote host; proxy values never leave it. +""" +from __future__ import annotations + +import os +from pathlib import Path +import re +import shlex +import socket +import sys + + +PROXY_KEYS = {"http_proxy", "https_proxy", "all_proxy", "no_proxy"} + + +def proxy_assignments(text: str) -> dict[str, str]: + values = {} + for line in text.splitlines(): + # Only standalone, unindented literal assignments. Do not evaluate + # functions, shell commands, substitutions, or conditional expressions. + if not re.match(r"^(?:export\s+)?(?:https?_proxy|all_proxy|no_proxy)=", line, re.I): + continue + if any(marker in line for marker in ("$", "`", ";", "&&", "||")): + continue + try: + words = shlex.split(line, comments=True) + except ValueError: + continue + if words and words[0] == "export": + words = words[1:] + assignments = [word.split("=", 1) for word in words] + if not assignments or any(len(item) != 2 or item[0].lower() not in PROXY_KEYS for item in assignments): + continue + for key, value in assignments: + if "\0" not in value: + values[key.lower()] = value + return values + + +def config_paths(home: Path, shell: str) -> list[Path]: + paths = [Path("/etc/environment"), Path("/etc/profile")] + paths.extend(sorted(Path("/etc/profile.d").glob("*.sh"))) + paths.extend(sorted((home / ".config/environment.d").glob("*.conf"))) + paths.append(home / ".profile") + if Path(shell).name in ("zsh",): + paths.extend([home / ".zshenv", home / ".zprofile", home / ".zshrc"]) + else: + paths.extend([home / ".bash_profile", home / ".bash_login", home / ".bashrc"]) + return paths + + +def configure_proxy(environment: dict[str, str], paths: list[Path]) -> list[str]: + # An existing remote SSH environment (even an explicit empty value) wins. + inherited = {key.lower() for key in environment if key.lower() in PROXY_KEYS} + loaded = {} + sources = [] + for path in paths: + try: + values = proxy_assignments(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError): + continue + values = {key: value for key, value in values.items() if key not in inherited} + if values: + loaded.update(values) + sources.append(str(path)) + environment.update(loaded) + # curl ignores uppercase HTTP_PROXY; normalize remote values, preserving + # explicit lowercase precedence and NO_PROXY exclusions. + for key in PROXY_KEYS: + if key not in environment and key.upper() in environment: + environment[key] = environment[key.upper()] + return sources + + +def application_proxy_paths(home: Path) -> list[Path]: + """Known runtime config locations; never inspect downloaded subscriptions.""" + return running_proxy_paths() + [ + home / ".local/share/io.github.clash-verge-rev.clash-verge-rev/clash-verge.yaml", + home / ".config/io.github.clash-verge-rev.clash-verge-rev/clash-verge.yaml", + home / ".config/clash-verge/clash-verge.yaml", + home / ".config/mihomo/config.yaml", + home / ".config/clash/config.yaml", + ] + + +def running_proxy_paths(proc_root: Path = Path("/proc")) -> list[Path]: + """Find the config explicitly used by a running Clash/Mihomo core.""" + paths = [] + try: + processes = list(proc_root.iterdir()) + except OSError: + return paths + for process in processes: + if not process.name.isdigit(): + continue + try: + arguments = (process / "cmdline").read_bytes().decode().rstrip("\0").split("\0") + except (OSError, UnicodeError): + continue + if not arguments or not re.fullmatch(r"(?:mihomo|clash)(?:[-.][A-Za-z0-9_.-]+)?", Path(arguments[0]).name): + continue + config = None + directory = None + for index, argument in enumerate(arguments[1:], 1): + flag, separator, value = argument.partition("=") + if not separator and index + 1 < len(arguments): + value = arguments[index + 1] + if flag in ("-f", "--config") and value: + config = Path(value) + elif flag in ("-d", "--config-dir") and value: + directory = Path(value) + if config is None and directory is not None: + config = directory / "config.yaml" + if config is None: + continue + if not config.is_absolute(): + try: + config = (process / "cwd").resolve(strict=True) / config + except OSError: + continue + if config not in paths: + paths.append(config) + return paths + + +def configure_application_proxy(environment: dict[str, str], paths: list[Path]) -> str | None: + # An explicit environment setting (including an empty/disabled value) + # takes precedence over a proxy application's independent listener. + if any(key.lower() in {"http_proxy", "https_proxy", "all_proxy"} for key in environment): + return None + for path in paths: + try: + text = path.read_text(encoding="utf-8") + except (OSError, UnicodeError): + continue + # This deliberately reads only simple top-level listener settings, + # not arbitrary YAML, provider credentials or subscription profiles. + fields = dict(re.findall( + r"^(mixed-port|port|socks-port|bind-address|authentication):[ \t]*(.*)$", + text, re.M, + )) + if "authentication" in fields and fields["authentication"].strip() != "[]": + continue + try: + bind = shlex.split(fields.get("bind-address", "'*'"), comments=True) + except ValueError: + continue + if len(bind) != 1 or bind[0] not in ("*", "0.0.0.0", "127.0.0.1", "::", "::1", "localhost"): + continue + host = "::1" if bind[0] in ("::", "::1") else "127.0.0.1" + for key, scheme in (("mixed-port", "http"), ("port", "http"), ("socks-port", "socks5h")): + match = re.fullmatch(r"[\"']?(\d+)[\"']?\s*(?:#.*)?", fields.get(key, "")) + if not match: + continue + port = int(match[1]) + if not 1 <= port <= 65535: + continue + try: + with socket.create_connection((host, port), timeout=1): + pass + except OSError: + continue + address = f"[{host}]" if host == "::1" else host + proxy = f"{scheme}://{address}:{port}" + environment.update(http_proxy=proxy, https_proxy=proxy, all_proxy=proxy) + return str(path) + return None + + +if __name__ == "__main__": + environment = dict(os.environ) + sources = configure_proxy(environment, config_paths(Path.home(), environment.get("SHELL", "bash"))) + application_source = configure_application_proxy(environment, application_proxy_paths(Path.home())) + if application_source: + sources.append(application_source + "（应用代理端口可连接）") + active = any(environment.get(key) for key in ("https_proxy", "all_proxy", "http_proxy")) + if active: + source = "、".join(sources) if sources else "SSH 环境变量" + print("使用远端自身代理，配置来源：" + source, flush=True) + else: + print("网络方式：远端直连（未发现可用代理配置；此行不是错误）。", flush=True) + os.execve("/bin/sh", ["sh", "-c", sys.argv[1]], environment) diff --git a/herdr_hub/remotes.py b/herdr_hub/remotes.py index a74904c..53daedc 100644 --- a/herdr_hub/remotes.py +++ b/herdr_hub/remotes.py @@ -717,6 +717,7 @@ def remote_control_command( *, platform: str, unset_environment: Sequence[str] = (), + load_proxy_environment: bool = False, ) -> list[str]: """Build one non-replayed mutation with platform-specific argv encoding. @@ -746,8 +747,21 @@ def remote_control_command( return "'" + value.replace("'", "''") + "'" unset = "".join(f"Remove-Item Env:{key} -ErrorAction SilentlyContinue; " for key in unset_environment) + proxy_setup = "" + if load_proxy_environment: + proxy_setup = ( + "foreach ($key in @('http_proxy','https_proxy','all_proxy','no_proxy')) { " + "$value=[Environment]::GetEnvironmentVariable($key,'Process'); " + "if ([string]::IsNullOrEmpty($value)) { " + "$value=[Environment]::GetEnvironmentVariable($key,'User'); " + "if ([string]::IsNullOrEmpty($value)) { $value=[Environment]::GetEnvironmentVariable($key,'Machine') }; " + "if (![string]::IsNullOrEmpty($value)) { [Environment]::SetEnvironmentVariable($key,$value,'Process') } } }; " + "if ($env:https_proxy -or $env:http_proxy -or $env:all_proxy) { " + "Write-Output 'Using remote proxy environment (process/user/machine).' " + "} else { Write-Output 'No remote proxy environment found; using remote network defaults.' }; " + ) script_text = ( - "$ErrorActionPreference='Stop'; " + unset + "$ErrorActionPreference='Stop'; " + unset + proxy_setup + "$exe=Join-Path $HOME '.local\\bin\\herdr.exe'; " "if (!(Test-Path -LiteralPath $exe)) { $exe=(Get-Command herdr -ErrorAction Stop).Source }; " "$p=New-Object System.Diagnostics.Process; " diff --git a/herdr_hub/restart.py b/herdr_hub/restart.py new file mode 100644 index 0000000..dfaa892 --- /dev/null +++ b/herdr_hub/restart.py @@ -0,0 +1,97 @@ +"""Standalone POSIX update/handoff helper, also sent to remote Python over SSH.""" +from __future__ import annotations + +import json +import os +from pathlib import Path +import socket +import subprocess +import sys +import time +import uuid + + +def request(socket_path: str, method: str, params: dict, timeout: float = 10) -> dict: + request_id = uuid.uuid4().hex + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection: + connection.settimeout(timeout) + connection.connect(socket_path) + connection.sendall((json.dumps({"id": request_id, "method": method, "params": params}) + "\n").encode()) + with connection.makefile("rb") as stream: + raw = stream.readline(1024 * 1024) + if not raw.endswith(b"\n"): + raise RuntimeError("Herdr 控制响应不完整；请检查 Session 状态后再重试") + response = json.loads(raw) + if not isinstance(response, dict) or response.get("id") != request_id: + raise RuntimeError("Herdr 控制响应 ID 不匹配") + if response.get("error"): + raise RuntimeError(json.dumps(response["error"], ensure_ascii=False)) + result = response.get("result") + if not isinstance(result, dict): + raise RuntimeError("Herdr 控制响应缺少结果") + return result + + +def installed_version(binary: str) -> str: + version = subprocess.check_output([binary, "--version"], text=True, timeout=10).strip() + if not version.startswith("herdr ") or not version[6:].strip(): + raise RuntimeError("无法识别已安装的 Herdr 版本：" + version) + return version[6:].strip() + + +def binary_identity(binary: str) -> tuple[int, int, int, int]: + info = os.stat(binary) + return info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns + + +def update_and_handoff(binary: str, session: str) -> None: + binary = str(Path(binary).absolute()) + status = subprocess.check_output( + [binary, "--session", session, "status", "server"], text=True, timeout=10, + ) + fields = dict(line.split(": ", 1) for line in status.splitlines() if ": " in line) + socket_path = fields.get("socket", "") + if fields.get("status") != "running" or not os.path.isabs(socket_path): + raise RuntimeError("当前 Session 没有可用的运行中服务器，不能执行热切换") + before = request(socket_path, "ping", {}) + if before.get("capabilities", {}).get("live_handoff") is not True: + raise RuntimeError("当前服务器不支持热切换；可先使用“更新 Herdr”，再安排普通重启") + print(f"更新目标 Session：{session}；当前服务器：{before.get('version', '未知')}", flush=True) + identity_before = binary_identity(binary) + # 0.8.x plain update refuses to complete with live sessions and no TTY. + # Let the native updater perform its opt-in handoff, scoped to this session. + subprocess.run([binary, "--session", session, "update", "--handoff"], check=True, timeout=300) + version = installed_version(binary) + current = request(socket_path, "ping", {}) + if current.get("version") != version: + if binary_identity(binary) != identity_before: + raise RuntimeError( + f"已安装 {version}，但原生热切换后服务器仍是 {current.get('version', '未知')}；" + "不会重复发送热切换请求或强制重启，请检查上方原生命令输出" + ) + print(f"正在热切换服务器到 {version}，保留运行中的进程…", flush=True) + # An unchanged binary means the updater may have returned 'up to date' + # before touching an older running server. Only then send a handoff. + request(socket_path, "server.live_handoff", { + "import_exe": str(Path(binary).resolve()), "expected_version": version, + }, timeout=40) + deadline = time.monotonic() + 20 + while True: + try: + current = request(socket_path, "ping", {}, timeout=2) + if current.get("version") == version: + break + except (OSError, ValueError, RuntimeError): + pass + if time.monotonic() >= deadline: + raise RuntimeError("热切换后未确认新服务器就绪；请检查状态，不会自动停止 Session") + time.sleep(.2) + print(f"服务器已确认使用 {version}，可以重新连接。", flush=True) + + +if __name__ == "__main__": + try: + update_and_handoff(sys.argv[1], sys.argv[2]) + except Exception as error: + print(f"更新/热切换未完成：{error}\n没有自动执行普通重启；请检查当前 Session。", file=sys.stderr, flush=True) + raise SystemExit(1) diff --git a/tests/test_control.py b/tests/test_control.py index 60032de..5f11380 100644 --- a/tests/test_control.py +++ b/tests/test_control.py @@ -6,6 +6,9 @@ import curses import json import os import subprocess +import sys +import tempfile +from pathlib import Path import threading import time import unittest @@ -18,6 +21,7 @@ from herdr_hub.control import ( ControlMenu, HerdrControl, parse_snapshot, + update_failure, ) from herdr_hub.remotes import RemoteEntry, remote_control_command @@ -41,6 +45,13 @@ def completed(payload, code=0): class SnapshotTests(unittest.TestCase): + def test_missing_session_reports_identity_problem_not_version(self): + payload = snapshot() + del payload["result"]["snapshot"]["panes"][0]["agent_session"] + with self.assertRaisesRegex(ControlError, "尚未向 Herdr 上报会话 ID") as caught: + parse_snapshot(payload) + self.assertNotIn("需要 Herdr 0.9", str(caught.exception)) + def test_focused_pane_and_exact_native_arguments(self): for kind, args in (("codex", ["fork", SESSION_ID]), ("claude", ["--resume", SESSION_ID, "--fork-session"])): @@ -68,6 +79,99 @@ class SnapshotTests(unittest.TestCase): class ControlTests(unittest.TestCase): + def test_windows_hot_restart_rejected_before_update(self): + control = HerdrControl(Endpoint.remote("windows-host", remote_transport="ssh"), "unused") + control.platform = "windows" + with patch.object(control, "_execute") as execute: + with self.assertRaisesRegex(ControlError, "Windows 不支持热切换"): + control.update_and_restart(lambda _: None) + execute.assert_not_called() + + def test_remote_hot_restart_runs_helper_on_exact_host(self): + control = HerdrControl(Endpoint.remote("ssh://me@host:2222", "named"), "unused", "fake-ssh") + control.platform = "posix" + with patch.object(control, "_execute", return_value=subprocess.CompletedProcess([], 0, b"ok", b"")) as execute: + control.update_and_restart(lambda _: None) + command = execute.call_args.args[0] + self.assertEqual(command[0], "fake-ssh") + self.assertIn("host", command) + self.assertIn("2222", command) + self.assertIn("exec python3 -u -c", command[-1]) + self.assertIn(' "$hub_herdr" named', command[-1]) + self.assertIn("unset HERDR_ENV;", command[-1]) + + def test_streaming_shows_unterminated_output_before_process_exits(self): + with tempfile.TemporaryDirectory() as directory: + marker = Path(directory) / "seen" + script = ( + "import os, sys, time\n" + "from pathlib import Path\n" + "os.write(2, b'downloading 10%')\n" + "deadline = time.monotonic() + 3\n" + "while not Path(sys.argv[1]).exists():\n" + " if time.monotonic() > deadline: sys.exit(9)\n" + " time.sleep(.01)\n" + "print(' done')\n" + ) + updates = [] + def progress(value): + updates.append(value) + marker.touch() + result = HerdrControl(Endpoint.local(), "unused")._execute( + [sys.executable, "-c", script, str(marker)], 5, progress=progress + ) + self.assertEqual(result.returncode, 0) + self.assertIn("downloading 10%", updates[0]) + self.assertIn(b"done", result.stdout) + + def test_streaming_timeout_preserves_diagnostic(self): + with self.assertRaisesRegex(ControlError, "waiting for network"): + HerdrControl(Endpoint.local(), "unused")._execute( + [sys.executable, "-c", "import os,time; os.write(2,b'waiting for network'); time.sleep(10)"], + .3, progress=lambda _: None, + ) + + def test_update_local_is_host_scoped_and_accepts_plain_text(self): + result = subprocess.CompletedProcess([], 0, b"Already up to date", b"notice") + with patch.dict(os.environ, {"HERDR_SESSION": "wrong"}), patch( + "herdr_hub.control.subprocess.run", return_value=result + ) as run: + output = HerdrControl(Endpoint.local("named"), "/fake/herdr").update() + self.assertIn("Already up to date", output) + self.assertIn("notice", output) + self.assertEqual(run.call_args.args[0], ["/fake/herdr", "update"]) + self.assertEqual(run.call_args.kwargs["timeout"], 300) + self.assertEqual(run.call_args.kwargs["stdin"], subprocess.DEVNULL) + self.assertNotIn("HERDR_SESSION", run.call_args.kwargs["env"]) + + def test_update_remote_routes_once_on_both_platforms(self): + for transport, path in (("native", "/home/me/sessions"), + ("ssh", "C:\\Users\\me\\sessions")): + listing = {"sessions": [{"name": "named", "running": True, "session_dir": path}]} + failure = subprocess.CompletedProcess([], 1, b"download started", b"network error") + with self.subTest(transport=transport), patch( + "herdr_hub.control.subprocess.run", side_effect=[completed(listing), failure] + ) as run: + control = HerdrControl(Endpoint.remote("host", "named", remote_transport=transport), "unused", "fake-ssh") + with self.assertRaisesRegex(ControlError, "network error"): + control.update() + self.assertEqual(run.call_count, 2) + command = run.call_args.args[0] + self.assertEqual(command[0], "fake-ssh") + self.assertIn("host", command) + script = command[-1] + if transport == "ssh": + script = base64.b64decode(script.split()[-1]).decode("utf-16le") + self.assertIn("update", script) + self.assertNotIn("--session", script) + self.assertNotIn("--handoff", script) + + def test_update_timeout_is_not_replayed(self): + with patch("herdr_hub.control.subprocess.run", side_effect=subprocess.TimeoutExpired("herdr", 300)) as run: + with self.assertRaisesRegex(ControlError, "操作可能已完成"): + HerdrControl(Endpoint.local(), "herdr").update() + run.assert_called_once() + def test_local_explicit_session_and_clean_environment(self): for session in ("default", "named"): with self.subTest(session=session), patch.dict(os.environ, { @@ -234,6 +338,35 @@ class ImmediateExecutor: class MenuTests(unittest.TestCase): + def test_small_popup_shows_download_failure_before_proxy_notice(self): + self.screen.getmaxyx.return_value = (8, 71) + self.control.update_and_restart.side_effect = ControlError(update_failure( + "更新失败", "未检测到代理\n正在下载\ncurl: (28) Connection timed out\n没有自动执行普通重启；请检查当前 Session。" + )) + self.run_menu([curses.KEY_DOWN, curses.KEY_DOWN, 10, 27]) + rendered = str(self.screen.addstr.call_args_list) + self.assertIn("curl: (28) Connection timed out", rendered) + self.assertNotIn("未检测到代理", rendered) + + def test_hot_restart_success_requests_reconnect(self): + self.control.update_and_restart.return_value = "verified" + self.screen.getch.side_effect = [curses.KEY_DOWN, curses.KEY_DOWN, 10] + menu = ControlMenu(self.control) + with patch("herdr_hub.control.ThreadPoolExecutor", ImmediateExecutor), patch("curses.set_escdelay"), patch("curses.curs_set"), patch("curses.mousemask"): + self.assertEqual(menu.run(self.screen), 0) + self.assertTrue(menu.reconnect_requested) + self.control.update_and_restart.assert_called_once() + + def test_hot_restart_failure_stays_visible_without_reconnect(self): + self.control.update_and_restart.side_effect = ControlError("handoff failed") + self.screen.getch.side_effect = [curses.KEY_DOWN, curses.KEY_DOWN, 10, 10, 27] + menu = ControlMenu(self.control) + with patch("herdr_hub.control.ThreadPoolExecutor", ImmediateExecutor), patch("curses.set_escdelay"), patch("curses.curs_set"), patch("curses.mousemask"): + self.assertEqual(menu.run(self.screen), 0) + self.assertFalse(menu.reconnect_requested) + self.control.update_and_restart.assert_called_once() + self.assertIn("handoff failed", str(self.screen.addstr.call_args_list)) + def setUp(self): self.control = Mock(endpoint=Endpoint.local()) self.control.source.return_value = parse_snapshot(snapshot()) @@ -268,8 +401,102 @@ class MenuTests(unittest.TestCase): self.run_menu([10, 27]) self.control.fork.assert_not_called() + def test_refresh_restores_fork_after_session_identity_becomes_available(self): + self.screen.getmaxyx.return_value = (8, 71) + self.control.source.side_effect = [ControlError("尚未上报会话 ID"), parse_snapshot(snapshot())] + self.run_menu([10, 18, 10]) + self.assertEqual(self.control.source.call_count, 2) + self.control.fork.assert_called_once() + self.assertIn("不可用", str(self.screen.addstr.call_args_list)) + + def test_compact_mouse_fork_still_executes(self): + self.screen.getmaxyx.return_value = (8, 71) + with patch("curses.getmouse", return_value=(0, 3, 1, 0, curses.BUTTON1_PRESSED)): + self.run_menu([curses.KEY_MOUSE]) + self.control.fork.assert_called_once() + self.control.update.assert_not_called() + + def test_update_without_agent_keeps_result_and_prevents_repeat(self): + self.control.source.side_effect = ControlError("没有 Agent") + self.control.update.return_value = "更新成功" + self.run_menu([curses.KEY_DOWN, 10, 10, 27]) + self.control.update.assert_called_once() + self.control.fork.assert_not_called() + rendered = str(self.screen.addstr.call_args_list) + self.assertIn("更新成功", rendered) + self.assertIn("更新目标：本机", rendered) + + def test_click_update_reports_remote_target_and_failure(self): + self.control.endpoint = Endpoint.remote("my-host") + self.control.update.side_effect = ControlError("更新失败：连接断开") + with patch("curses.getmouse", return_value=(0, 3, 5, 0, curses.BUTTON1_PRESSED)): + self.run_menu([curses.KEY_MOUSE, 10, 27]) + self.control.update.assert_called_once() + rendered = str(self.screen.addstr.call_args_list) + self.assertIn("更新目标：my-host", rendered) + self.assertIn("更新失败：连接断开", rendered) + + def test_update_output_can_be_scrolled(self): + self.control.update.return_value = "\n".join(f"line-{i}" for i in range(60)) + self.run_menu([curses.KEY_DOWN, 10, curses.KEY_NPAGE, curses.KEY_PPAGE, 27]) + self.assertIn("line-20", str(self.screen.addstr.call_args_list)) + + def test_update_result_visible_in_actual_small_popup(self): + self.screen.getmaxyx.return_value = (8, 71) + self.control.update.return_value = "更新命令已完成\nAlready up to date" + self.run_menu([curses.KEY_DOWN, 10, 27]) + calls = self.screen.addstr.call_args_list + results = [call.args for call in calls if "Already up to date" in call.args[2]] + self.assertTrue(results, "结果必须在真实的 8 行弹窗内绘制") + self.assertTrue(all(3 <= args[0] < 7 for args in results)) + self.assertTrue(all(0 <= call.args[0] < 8 for call in calls)) + + def test_compact_mouse_update_row(self): + self.screen.getmaxyx.return_value = (8, 71) + self.control.update.return_value = "完成" + with patch("curses.getmouse", return_value=(0, 3, 2, 0, curses.BUTTON1_PRESSED)): + self.run_menu([curses.KEY_MOUSE, 27]) + self.control.update.assert_called_once() + self.control.fork.assert_not_called() + + def test_update_during_snapshot_load_is_not_ignored(self): + cancel = None + class LoadingFuture: + def done(self): + return cancel.is_set() + def result(self): + raise ControlCancelled() + class LoadingExecutor(ImmediateExecutor): + def submit(inner, fn, *args): + nonlocal cancel + if fn == self.control.source: + cancel = args[0] + return LoadingFuture() + return ImmediateFuture(fn, *args) + self.control.update.return_value = "完成" + self.screen.getch.side_effect = [curses.KEY_DOWN, 10, -1, 27] + with patch("herdr_hub.control.ThreadPoolExecutor", LoadingExecutor), patch("curses.set_escdelay"), patch("curses.curs_set"), patch("curses.mousemask"): + ControlMenu(self.control).run(self.screen) + self.control.update.assert_called_once() + class CliControlTests(unittest.TestCase): + def test_hot_restart_reconnects_original_window_after_popup_closes(self): + args = build_parser().parse_args(["control"]) + endpoint = Endpoint.local() + original = ManagedWindow("@17", endpoint.key, "Local", False, True, endpoint) + other = ManagedWindow("@18", "local:other", "Other", False, True, Endpoint.local("other")) + hub = Mock(ssh_bin="fake-ssh") + hub.managed_windows.side_effect = [[original], [original, other]] + def close_popup(run): + hub.reconnect.assert_not_called() + run.__self__.reconnect_requested = True + return 0 + with patch("herdr_hub.cli.make_hub", return_value=hub), patch("herdr_hub.cli.curses.wrapper", side_effect=close_popup): + self.assertEqual(command_control(args), 0) + hub.reconnect.assert_called_once_with("@17") + hub.wait_for_window_startup.assert_called_once_with("@17") + def test_active_window_metadata_selects_control_endpoint(self): args = build_parser().parse_args(["control"]) hub = Mock(ssh_bin="fake-ssh") diff --git a/tests/test_proxy_env.py b/tests/test_proxy_env.py new file mode 100644 index 0000000..dbcff35 --- /dev/null +++ b/tests/test_proxy_env.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +import base64 +import os +from pathlib import Path +import shlex +import subprocess +import sys +import tempfile +import unittest +from unittest.mock import patch + +from herdr_hub.cli import Endpoint +from herdr_hub.control import HerdrControl +from herdr_hub.proxy_env import configure_application_proxy, configure_proxy, proxy_assignments, running_proxy_paths + + +class RemoteProxyTests(unittest.TestCase): + def test_running_mihomo_custom_config_is_discovered(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + core = root / "1277" + core.mkdir() + (core / "cmdline").write_bytes( + b"/opt/mihomo-linux-arm64\0-d\0/home/orangepi/repo/clash\0-f\0/home/orangepi/repo/clash/config-systemd.yaml\0" + ) + unrelated = root / "1278" + unrelated.mkdir() + (unrelated / "cmdline").write_bytes(b"python3\0-c\0mihomo\0-f\0/irrelevant.yaml\0") + self.assertEqual(running_proxy_paths(root), [Path("/home/orangepi/repo/clash/config-systemd.yaml")]) + + def test_running_clash_listener_used_when_shell_has_no_proxy(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "clash-verge.yaml" + path.write_text("mixed-port: 7897\nbind-address: '*'\nallow-lan: false\n") + environment = {"no_proxy": "localhost,internal"} + with patch("herdr_hub.proxy_env.socket.create_connection") as connect: + source = configure_application_proxy(environment, [path]) + self.assertEqual(source, str(path)) + self.assertEqual(environment["https_proxy"], "http://127.0.0.1:7897") + self.assertEqual(environment["no_proxy"], "localhost,internal") + connect.assert_called_once_with(("127.0.0.1", 7897), timeout=1) + + def test_stopped_application_and_authenticated_proxy_not_adopted(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "config.yaml" + for contents in ("mixed-port: 7897\n", "mixed-port: 7897\nauthentication:\n - user:secret\n"): + path.write_text(contents) + environment = {} + with patch("herdr_hub.proxy_env.socket.create_connection", side_effect=ConnectionRefusedError): + self.assertIsNone(configure_application_proxy(environment, [path])) + self.assertEqual(environment, {}) + + def test_application_detection_respects_existing_proxy_and_disable(self): + for value in ("", "http://configured:8080"): + environment = {"HTTPS_PROXY": value} + with patch("herdr_hub.proxy_env.socket.create_connection") as connect: + self.assertIsNone(configure_application_proxy(environment, [Path("/unused")])) + connect.assert_not_called() + self.assertEqual(environment, {"HTTPS_PROXY": value}) + + def test_literal_exports_and_uppercase_are_read_without_shell_evaluation(self): + values = proxy_assignments(''' +export https_proxy="http://remote-user:secret@127.0.0.1:7890" # local to remote +HTTP_PROXY=http://remote-host:8080 +export no_proxy='localhost,127.0.0.1' +export all_proxy=$(dangerous_command) +export all_proxy=http://other:8080; dangerous_command + export all_proxy=http://inside-function:8080 +''') + self.assertEqual(values, { + "https_proxy": "http://remote-user:secret@127.0.0.1:7890", + "http_proxy": "http://remote-host:8080", + "no_proxy": "localhost,127.0.0.1", + }) + + def test_remote_environment_wins_including_explicit_disable(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / ".bashrc" + path.write_text("export https_proxy=http://file:7890\nexport all_proxy=http://fallback:7890\n") + environment = {"HTTPS_PROXY": "http://remote-env:7890", "all_proxy": ""} + configure_proxy(environment, [path]) + self.assertEqual(environment["https_proxy"], "http://remote-env:7890") + self.assertEqual(environment["all_proxy"], "") + + def test_wrapper_loads_remote_bashrc_and_never_embeds_local_proxy(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + marker = root / "should-not-exist" + remote_proxy = "http://remote-user:remote-secret@127.0.0.1:7891" + (root / ".bashrc").write_text( + "case $- in *i*) ;; *) return;; esac\n" + + f"export https_proxy={shlex.quote(remote_proxy)}\n" + + f"touch {shlex.quote(str(marker))}\n" + ) + controller = HerdrControl(Endpoint.remote("host"), "unused") + controller.platform = "posix" + child = shlex.join([sys.executable, "-c", "import os; assert os.environ['https_proxy'] == " + repr(remote_proxy)]) + with patch.dict(os.environ, {"https_proxy": "http://local-secret@127.0.0.1:9999"}): + command = controller._update_proxy(["ssh", "host", child], None) + self.assertEqual(command[:-1], ["ssh", "host"]) + self.assertNotIn("local-secret", command[-1]) + environment = {k: v for k, v in os.environ.items() if k.lower() not in { + "http_proxy", "https_proxy", "all_proxy", "no_proxy", + }} + environment.update(HOME=directory, SHELL="/bin/bash") + result = subprocess.run(["sh", "-c", command[-1]], env=environment, capture_output=True, text=True, timeout=10) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("使用远端自身代理", result.stdout) + self.assertNotIn("remote-secret", result.stdout + result.stderr) + self.assertFalse(marker.exists()) + + def test_windows_loads_only_remote_environment_scopes(self): + controller = HerdrControl(Endpoint.remote("windows-host"), "unused") + controller.platform = "windows" + with patch.dict(os.environ, {"https_proxy": "http://local-secret@127.0.0.1:9999"}): + command = controller._update_proxy(["ssh", "windows-host", "unused"], None) + script = base64.b64decode(command[-1].split()[-1]).decode("utf-16le") + for scope in ("Process", "User", "Machine"): + self.assertIn(f"GetEnvironmentVariable($key,'{scope}')", script) + self.assertNotIn("local-secret", script) + self.assertNotIn("-R", command) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_restart.py b/tests/test_restart.py new file mode 100644 index 0000000..6c5061e --- /dev/null +++ b/tests/test_restart.py @@ -0,0 +1,152 @@ +from __future__ import annotations + +import contextlib +import io +import json +import os +from pathlib import Path +import socket +import subprocess +import sys +import tempfile +import threading +import unittest +from unittest.mock import patch + +from herdr_hub.cli import Endpoint +from herdr_hub.control import ControlError, HerdrControl +from herdr_hub.restart import update_and_handoff + + +class HandoffTests(unittest.TestCase): + def setUp(self): + identity = patch("herdr_hub.restart.binary_identity", return_value=(1, 1, 1, 1)) + identity.start() + self.addCleanup(identity.stop) + + def test_failed_update_never_sends_handoff(self): + with patch("herdr_hub.restart.subprocess.check_output", return_value="status: running\nsocket: /tmp/fake.sock\n"), patch( + "herdr_hub.restart.request", return_value={"capabilities": {"live_handoff": True}} + ) as call, patch("herdr_hub.restart.subprocess.run", side_effect=subprocess.CalledProcessError(1, "update")), contextlib.redirect_stdout(io.StringIO()): + with self.assertRaises(subprocess.CalledProcessError): + update_and_handoff("/fake/herdr", "named") + self.assertEqual([c.args[1] for c in call.call_args_list], ["ping"]) + + def test_native_082_update_uses_handoff_and_does_not_repeat_it(self): + old = {"version": "0.8.2", "capabilities": {"live_handoff": True}} + with patch("herdr_hub.restart.subprocess.check_output", return_value="status: running\nsocket: /tmp/fake.sock\n"), patch( + "herdr_hub.restart.request", side_effect=[old, {"version": "0.9.1"}] + ) as call, patch("herdr_hub.restart.subprocess.run") as run, patch( + "herdr_hub.restart.installed_version", return_value="0.9.1" + ), contextlib.redirect_stdout(io.StringIO()): + update_and_handoff("/fake/herdr", "named") + self.assertEqual(run.call_args.args[0], ["/fake/herdr", "--session", "named", "update", "--handoff"]) + self.assertEqual([c.args[1] for c in call.call_args_list], ["ping", "ping"]) + + def test_native_handoff_failure_after_install_is_not_replayed(self): + old = {"version": "0.8.2", "capabilities": {"live_handoff": True}} + with patch("herdr_hub.restart.subprocess.check_output", return_value="status: running\nsocket: /tmp/fake.sock\n"), patch( + "herdr_hub.restart.request", side_effect=[old, old] + ) as call, patch("herdr_hub.restart.subprocess.run"), patch( + "herdr_hub.restart.installed_version", return_value="0.9.1" + ), patch("herdr_hub.restart.binary_identity", side_effect=[(1, 1, 1, 1), (1, 2, 1, 1)]), contextlib.redirect_stdout(io.StringIO()): + with self.assertRaisesRegex(RuntimeError, "不会重复发送热切换"): + update_and_handoff("/fake/herdr", "named") + self.assertEqual([c.args[1] for c in call.call_args_list], ["ping", "ping"]) + + def test_unverified_handoff_is_failure_not_success(self): + old = {"version": "0.9.0", "capabilities": {"live_handoff": True}} + with patch("herdr_hub.restart.subprocess.check_output", return_value="status: running\nsocket: /tmp/fake.sock\n"), patch( + "herdr_hub.restart.request", side_effect=[old, old, {}, old] + ) as call, patch("herdr_hub.restart.subprocess.run"), patch( + "herdr_hub.restart.installed_version", return_value="0.9.1" + ), patch("herdr_hub.restart.time.monotonic", side_effect=[0, 21]), contextlib.redirect_stdout(io.StringIO()): + with self.assertRaisesRegex(RuntimeError, "未确认新服务器就绪"): + update_and_handoff("/fake/herdr", "named") + self.assertEqual([c.args[1] for c in call.call_args_list].count("server.live_handoff"), 1) + + def test_real_helper_updates_old_server_even_when_binary_already_latest(self): + self._run_real_helper() + + def test_real_helper_failure_does_not_stop_or_retry_server(self): + self._run_real_helper(fail_handoff=True) + + def _run_real_helper(self, fail_handoff=False): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + socket_path = root / "api.sock" + record = root / "commands.jsonl" + binary = root / "herdr" + binary.write_text( + f"#!{sys.executable}\n" + "import json, os, sys\n" + f"with open({str(record)!r}, 'a') as f: f.write(json.dumps(sys.argv[1:])+'\\n')\n" + "if os.environ.get('HERDR_ENV') == '1': sys.exit(8)\n" + "if sys.argv[1:] == ['--version']: print('herdr 0.9.1')\n" + "elif sys.argv[1:] == ['--session', 'named', 'status', 'server']:\n" + f" print('status: running\\nsocket: {socket_path}')\n" + "elif sys.argv[1:] == ['--session', 'named', 'update', '--handoff']:\n" + " print('already up to date (0.9.1)', flush=True)\n" + "else: sys.exit(9)\n" + ) + binary.chmod(0o755) + methods = [] + done = threading.Event() + errors = [] + server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + server.bind(str(socket_path)) + server.listen() + server.settimeout(.1) + def serve(): + version = "0.9.0" + try: + while not done.is_set(): + try: + connection, _ = server.accept() + except socket.timeout: + continue + with connection, connection.makefile("rb") as stream: + message = json.loads(stream.readline()) + methods.append(message) + response = {"id": message["id"]} + if message["method"] == "server.live_handoff": + if fail_handoff: + response["error"] = {"message": "handoff refused"} + else: + version = "0.9.1" + response["result"] = {} + elif message["method"] == "ping": + response["result"] = {"version": version, "capabilities": {"live_handoff": True}} + else: + raise AssertionError(message) + connection.sendall((json.dumps(response) + "\n").encode()) + except Exception as error: + errors.append(error) + thread = threading.Thread(target=serve) + thread.start() + output = [] + try: + with patch.dict(os.environ, {"HERDR_ENV": "1", "HERDR_SESSION": "wrong"}): + control = HerdrControl(Endpoint.local("named"), str(binary)) + if fail_handoff: + with self.assertRaisesRegex(ControlError, "handoff refused"): + control.update_and_restart(output.append) + else: + result = control.update_and_restart(output.append) + self.assertIn("服务器已确认使用 0.9.1", result) + finally: + done.set() + thread.join(3) + server.close() + self.assertFalse(errors) + handoffs = [m for m in methods if m["method"] == "server.live_handoff"] + self.assertEqual(len(handoffs), 1) + self.assertEqual(handoffs[0]["params"]["import_exe"], str(binary.resolve())) + self.assertEqual(handoffs[0]["params"]["expected_version"], "0.9.1") + commands = [json.loads(line) for line in record.read_text().splitlines()] + self.assertEqual(commands.count(["--session", "named", "update", "--handoff"]), 1) + self.assertFalse(any("stop" in command for command in commands)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_tmux_integration.py b/tests/test_tmux_integration.py index 66f88c5..ea9bde6 100644 --- a/tests/test_tmux_integration.py +++ b/tests/test_tmux_integration.py @@ -8,6 +8,7 @@ import tempfile import time import unittest import uuid +from unittest.mock import patch from herdr_hub.cli import ( BOOTSTRAP_WINDOW_NAME, @@ -15,11 +16,32 @@ from herdr_hub.cli import ( HubError, TmuxHub, WINDOW_SHORT_LABEL_OPTION, + build_parser, + command_control, ) @unittest.skipUnless(shutil.which("tmux"), "tmux is required") class TmuxIntegrationTests(unittest.TestCase): + def test_verified_handoff_reconnects_only_originating_client(self) -> None: + self.hub.ensure_server() + original = self.hub.create_window(Endpoint.local("first"), str(self.fake_herdr)) + other = self.hub.create_window(Endpoint.local("second"), str(self.fake_herdr)) + self.hub.select_window(original.window_id) + def pid(window): + return self.hub.run(["display-message", "-p", "-t", window.window_id, "#{pane_pid}"]).stdout.strip() + original_pid, other_pid = pid(original), pid(other) + def close_popup(run): + self.hub.select_window(other.window_id) + run.__self__.reconnect_requested = True + return 0 + with patch("herdr_hub.cli.make_hub", return_value=self.hub), patch( + "herdr_hub.cli.resolve_herdr", return_value=str(self.fake_herdr) + ), patch("herdr_hub.cli.curses.wrapper", side_effect=close_popup): + self.assertEqual(command_control(build_parser().parse_args(["control"])), 0) + self.assertNotEqual(pid(original), original_pid) + self.assertEqual(pid(other), other_pid) + def setUp(self) -> None: self.temporary_directory = tempfile.TemporaryDirectory() self.addCleanup(self.temporary_directory.cleanup)