From 420a7fecfd43259972e5aaf6d0f4c423595d9d1a Mon Sep 17 00:00:00 2001 From: host Date: Tue, 27 Oct 2026 05:47:33 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20deploy=20=E6=97=A0=E5=AF=86=E7=A0=81?= =?UTF-8?q?=E6=97=B6=E8=B5=B0=E5=AF=86=E9=92=A5=E7=9B=B4=E8=BF=9E=EF=BC=88?= =?UTF-8?q?=E6=94=B6=E8=BF=9B=20main=EF=BC=8C=E6=9B=BF=E4=BB=A3=E4=B8=A2?= =?UTF-8?q?=E5=A4=B1=E7=9A=84=E5=B7=A5=E4=BD=9C=E5=8C=BA=E8=A1=A5=E4=B8=81?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 背景：2026-10-27 另一条线给 deploy.py 打的密钥补丁只放在 env-console 工作区未提交，切分支后丢失；部署一度只能靠它 + --allow-dirty。本批把逻辑收进 main，并加一处加固。 分发：有 SSH_PASSWORD 时走 helper（pty 喂密码，实测最稳的一 路）；无密码时 helper 不可用（会卡在密码提示），改走密钥直连： ProxyCommand 显式环境变量优先、其次 helper 目录的 proxy_tunnel.py（存在才用），密钥按 CODEX_DEPLOY_KEY → /root/.ssh/id_codex_deploy → ~/.ssh/id_codex_deploy 顺序显式传 -i——不能只依赖 /root/.ssh/config 的 Host 条目，那个目录会在 VM 回收时被清空（2026-10-27 05:04 实测），而 ~/.ssh 的那份 持久。两处都没有时不硬编造，靠 ssh 配置或 agent。 测试：deploy/tests 新增 TestSshScpCmd 5 个（有密码走 helper、 无密码绝不用 helper、无 helper 无代理时直连、显式代理优先、 显式密钥优先）；全量 deploy 19、host 62、cloud 598 全过。 --- deploy/deploy.py | 53 ++++++++++++++++++++++++----- deploy/tests/test_deploy.py | 67 +++++++++++++++++++++++++++++++++++++ 2 files changed, 112 insertions(+), 8 deletions(-) diff --git a/deploy/deploy.py b/deploy/deploy.py index 5f30343..3a44bcf 100644 --- a/deploy/deploy.py +++ b/deploy/deploy.py @@ -156,30 +156,67 @@ def helper_path(name): return path if os.path.isfile(path) else None -def ssh_cmd(host, user, remote_cmd): - """返回一条 ssh 命令。优先用已验证稳定的 helper，否则用 - SSH_PROXY_CMD 环境变量，否则直连。绝不把密码拼进命令。""" - helper = helper_path("ssh_run.py") - if helper: - return ["python3", helper, "ssh", host, user, remote_cmd] +def _deploy_key(): + """无密码时的部署密钥路径：显式环境变量优先，其次两处惯常 + 位置。/root/.ssh 的那份会在 VM 回收时被清空，~/.ssh 的那份 + 持久；两处都没有时返回 None（此时靠 ssh 配置或 agent，不 + 硬编造路径）。这里只返回路径，不读密钥内容。""" + env_key = os.environ.get("CODEX_DEPLOY_KEY") + if env_key: + return env_key + for cand in ("/root/.ssh/id_codex_deploy", + os.path.expanduser("~/.ssh/id_codex_deploy")): + if os.path.isfile(cand): + return cand + return None + + +def _proxy_cmd(): + """无密码时的 ProxyCommand：显式环境变量优先，其次 helper + 目录的 proxy_tunnel.py（存在才用）；都没有时返回 None（直 + 连，不把不存在的脚本写进命令）。""" proxy = os.environ.get("SSH_PROXY_CMD") + if proxy: + return proxy + default_proxy = os.path.join(HELPER_DIR, "proxy_tunnel.py") + if os.path.isfile(default_proxy): + return "python3 %s %%h %%p" % default_proxy + return None + + +def ssh_cmd(host, user, remote_cmd): + """返回一条 ssh 命令。有 SSH_PASSWORD 时优先用已验证稳定的 + helper（pty 喂密码）；无密码时走密钥直连：ProxyCommand 按 + _proxy_cmd()，密钥按 _deploy_key() 显式传 -i（不能只依赖 + /root/.ssh/config 的 Host 条目，那个目录会在 VM 回收时被 + 清空）。绝不把密码拼进命令。""" + helper = helper_path("ssh_run.py") + if helper and os.environ.get("SSH_PASSWORD"): + return ["python3", helper, "ssh", host, user, remote_cmd] base = ["ssh", "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null"] + proxy = _proxy_cmd() if proxy: base += ["-o", "ProxyCommand %s" % proxy] + key = _deploy_key() + if key: + base += ["-i", key] return base + ["%s@%s" % (user, host), remote_cmd] def scp_cmd(host, user, local, remote): helper = helper_path("scp_run.py") - if helper: + if helper and os.environ.get("SSH_PASSWORD"): # helper 自带 egress proxy + 密码 pty，是此前实测最稳的一路。 return ["python3", helper, "scp", host, user, local, remote] - proxy = os.environ.get("SSH_PROXY_CMD") base = ["scp", "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null"] + proxy = _proxy_cmd() if proxy: base += ["-o", "ProxyCommand %s" % proxy] + key = _deploy_key() + if key: + base += ["-i", key] return base + [local, "%s@%s:%s" % (user, host, remote)] diff --git a/deploy/tests/test_deploy.py b/deploy/tests/test_deploy.py index 145bbd5..950b5b8 100644 --- a/deploy/tests/test_deploy.py +++ b/deploy/tests/test_deploy.py @@ -118,5 +118,72 @@ class TestBuildRelease(unittest.TestCase): self.assertIn("hello.py", names) +class TestSshScpCmd(unittest.TestCase): + """ssh_cmd/scp_cmd 的三路分发：有密码走 helper、无密码走 + 密钥直连、无 helper 无代理时直连。2026-10-27 密钥补丁曾只 + 放在 env-console 工作区未提交、切分支后丢失，本批把它收进 + main；测试钉住分发逻辑，不依赖真实主机与真实密钥文件。""" + + def setUp(self): + self._env = dict(os.environ) + os.environ.pop("SSH_PASSWORD", None) + os.environ.pop("SSH_PROXY_CMD", None) + os.environ.pop("CODEX_DEPLOY_KEY", None) + + def tearDown(self): + os.environ.clear() + os.environ.update(self._env) + + def _cmds(self, helper=None, proxy_file=False, key_file=False): + with mock.patch.object(deploy_mod, "helper_path", + return_value=helper): + with mock.patch.object(deploy_mod.os.path, "isfile", + return_value=(proxy_file or key_file)): + # isfile 对 proxy 与 key 都返回同一值：测试只断言 + # 分支形状（是否出现 ProxyCommand/-i），不区分两者。 + ssh = deploy_mod.ssh_cmd("h", "u", "true") + scp = deploy_mod.scp_cmd("h", "u", "a", "b") + return ssh, scp + + def test_password_uses_helper_even_without_proxy(self): + os.environ["SSH_PASSWORD"] = "x" + ssh, scp = self._cmds(helper="/helpers/ssh_run.py") + self.assertEqual(ssh[:4], + ["python3", "/helpers/ssh_run.py", "ssh", "h"]) + self.assertEqual(scp[:4], + ["python3", "/helpers/ssh_run.py", "scp", "h"]) + + def test_no_password_never_uses_helper(self): + # 无密码时 helper（pty 喂密码）不可用：即使 helper 文件 + # 存在也不走它，否则会在密码提示处卡住。 + ssh, scp = self._cmds(helper="/helpers/ssh_run.py", + proxy_file=True, key_file=True) + self.assertNotIn("/helpers/ssh_run.py", ssh) + self.assertNotIn("/helpers/ssh_run.py", scp) + # ProxyCommand 与参数拼在一个元素里，查前缀而不是整元素。 + self.assertTrue(any(x.startswith("ProxyCommand") for x in ssh)) + self.assertIn("-i", ssh) + self.assertIn("-i", scp) + + def test_no_password_no_helpers_goes_direct(self): + ssh, scp = self._cmds(helper=None, + proxy_file=False, key_file=False) + self.assertEqual(ssh[0], "ssh") + self.assertNotIn("ProxyCommand", ssh) + self.assertNotIn("-i", ssh) + self.assertEqual(scp[0], "scp") + + def test_explicit_proxy_cmd_wins(self): + os.environ["SSH_PROXY_CMD"] = "proxy-cmd %h %p" + ssh, _ = self._cmds(helper=None, proxy_file=False) + self.assertIn("ProxyCommand proxy-cmd %h %p", ssh) + + def test_explicit_key_env_wins(self): + os.environ["CODEX_DEPLOY_KEY"] = "/keys/d" + ssh, _ = self._cmds(helper=None, proxy_file=False, + key_file=False) + self.assertIn("/keys/d", ssh) + + if __name__ == "__main__": unittest.main()