From 5aad2c24292ce427ac2d65c69282f05502598dd6 Mon Sep 17 00:00:00 2001 From: sully Date: Tue, 16 Jun 2026 11:43:04 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=E6=94=AF=E6=8C=81=20SSH=20=E5=AF=86?= =?UTF-8?q?=E9=92=A5=E5=AF=86=E7=A0=81=E7=9F=AD=E8=AF=AD=E9=AA=8C=E8=AF=81?= =?UTF-8?q?=EF=BC=8C=E6=B7=BB=E5=8A=A0=20cryptography=20=E4=BE=9D=E8=B5=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 _resolve_key_passphrase() 函数，通过 cryptography 库解析密钥并提示输入密码短语 - HostConfig 添加 get_password() 方法处理整数类型密码 - 添加 poetry.toml 配置 in-project 虚拟环境 - 更新 dev-dependencies 为新的 poetry group 格式 --- poetry.toml | 2 ++ pyproject.toml | 3 ++- sshg.py | 42 ++++++++++++++++++++++++++++++++++++++++-- 3 files changed, 44 insertions(+), 3 deletions(-) create mode 100644 poetry.toml diff --git a/poetry.toml b/poetry.toml new file mode 100644 index 0000000..ab1033b --- /dev/null +++ b/poetry.toml @@ -0,0 +1,2 @@ +[virtualenvs] +in-project = true diff --git a/pyproject.toml b/pyproject.toml index 2708b2f..6aa4a57 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,11 +18,12 @@ dataclasses-json = "*" pyyaml = "*" prompt_toolkit = "*" pexpect = "*" +cryptography = "<43" [tool.poetry.scripts] sshg = "sshg:main" -[tool.poetry.dev-dependencies] +[tool.poetry.group.dev.dependencies] pytest = "^7.2.0" pytest-cov = "^2" diff --git a/sshg.py b/sshg.py index da0f6bb..d1e3e9f 100644 --- a/sshg.py +++ b/sshg.py @@ -96,6 +96,11 @@ class HostConfig(DataClassJsonMixin): via: typing.Optional["HostConfig"] = make_field(mm_field=fields.Field(), default=None) _parent: typing.Optional["HostConfig"] = make_field(mm_field=fields.Field(), default=None, init=False, repr=False) + def get_password(self) -> str: + if isinstance(self.password, int): + return str(self.password) + return self.password or "" + def post_load(self): if self._parent: if not self.user: @@ -151,6 +156,37 @@ def output_filter(line): s.interact() +def _resolve_key_passphrase(keypath: pathlib.Path) -> str: + """Try loading key without passphrase, prompt until correct if needed.""" + from cryptography.hazmat.primitives.serialization import load_pem_private_key, load_ssh_private_key + + key_data = keypath.read_bytes() + + def _try_load(password: bytes | None) -> bool: + try: + load_ssh_private_key(key_data, password=password) + return True + except TypeError: + return False + except ValueError as e: + if "password" in str(e).lower(): + return False + try: + load_pem_private_key(key_data, password=password) + return True + except (TypeError, ValueError) as e: + return False + + if _try_load(None): + return "" + + while True: + password = getpass.getpass(f"Enter passphrase for key {keypath}: ") + if _try_load(password.encode()): + return password + print("Wrong passphrase, try again.") + + def spawn_ssh(host_config: HostConfig, is_local: bool = True, ssh_client: pxssh.pxssh = None, reset_prompt: bool = None) -> pxssh.pxssh: # https://pexpect.readthedocs.io/en/stable/api/pxssh.html cmdargs = host_config.build_cmdargs() @@ -164,6 +200,8 @@ def spawn_ssh(host_config: HostConfig, is_local: bool = True, ssh_client: pxssh. if keypath.stat().st_mode & 0o077 != 0: print("Warning: keypath mode change to 0600") keypath.chmod(0o600) + if not host_config.get_password(): + host_config.password = _resolve_key_passphrase(keypath) s.SSH_OPTS += " -o StrictHostKeyChecking=no" if reset_prompt is None: @@ -172,7 +210,7 @@ def spawn_ssh(host_config: HostConfig, is_local: bool = True, ssh_client: pxssh. if is_local: s.login(host_config.host, username=host_config.user, - password=host_config.password, + password=host_config.get_password(), port=host_config.port, ssh_key=keypath, quiet=False, @@ -182,7 +220,7 @@ def spawn_ssh(host_config: HostConfig, is_local: bool = True, ssh_client: pxssh. else: s.login(host_config.host, username=host_config.user, - password=host_config.password, + password=host_config.get_password(), port=host_config.port, ssh_key=keypath, quiet=False, From afb494d97ad306b57186001345bfc02b231b51ad Mon Sep 17 00:00:00 2001 From: codeskyblue Date: Tue, 16 Jun 2026 13:08:43 +0800 Subject: [PATCH 2/2] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- sshg.py | 32 +++++++++++++++++++------------- 1 file changed, 19 insertions(+), 13 deletions(-) diff --git a/sshg.py b/sshg.py index d1e3e9f..bbb0da9 100644 --- a/sshg.py +++ b/sshg.py @@ -162,27 +162,33 @@ def _resolve_key_passphrase(keypath: pathlib.Path) -> str: key_data = keypath.read_bytes() - def _try_load(password: bytes | None) -> bool: - try: - load_ssh_private_key(key_data, password=password) - return True - except TypeError: - return False - except ValueError as e: - if "password" in str(e).lower(): - return False + def _try_load(password: typing.Optional[bytes]) -> typing.Optional[bool]: + """Return True if the key loads, False if a passphrase is required/incorrect, None if the key is invalid.""" + errors: typing.List[Exception] = [] + for loader in (load_ssh_private_key, load_pem_private_key): try: - load_pem_private_key(key_data, password=password) + loader(key_data, password=password) return True except (TypeError, ValueError) as e: - return False + errors.append(e) + + msg = " ".join(str(e).lower() for e in errors) + if any(k in msg for k in ("password", "passphrase", "bad decrypt", "incorrect")): + return False + return None - if _try_load(None): + res = _try_load(None) + if res is None: + raise ValueError(f"Unsupported or invalid private key: {keypath}") + if res: return "" while True: password = getpass.getpass(f"Enter passphrase for key {keypath}: ") - if _try_load(password.encode()): + res = _try_load(password.encode()) + if res is None: + raise ValueError(f"Unsupported or invalid private key: {keypath}") + if res: return password print("Wrong passphrase, try again.")