Merge pull request #75 from openqj/feat-deploy-key

feat: deploy 无密码时走密钥直连

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):
+def _deploy_key():
-    """返回一条 ssh 命令。优先用已验证稳定的 helper，否则用
+    """无密码时的部署密钥路径：显式环境变量优先，其次两处惯常
-    SSH_PROXY_CMD 环境变量，否则直连。绝不把密码拼进命令。"""
+    位置。/root/.ssh 的那份会在 VM 回收时被清空，~/.ssh 的那份
-    helper = helper_path("ssh_run.py")
+    持久；两处都没有时返回 None（此时靠 ssh 配置或 agent，不
-    if helper:
+    硬编造路径）。这里只返回路径，不读密钥内容。"""
-        return ["python3", helper, "ssh", host, user, remote_cmd]
+    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)]
 
 

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()