From 987fdc0e1204ef1663b0eb8ce4490b209c69321b Mon Sep 17 00:00:00 2001 From: Walter Oggioni Date: Mon, 26 Oct 2026 12:06:16 +0800 Subject: [PATCH] Fix SSH private-key import/export for Paramiko 5.1.0 and remove unconditional DSSKey reference - Replace paramiko-based OpenSSH private key serialization with cryptography's PrivateFormat.OpenSSH backend. The old code relied on paramiko.RSAKey, paramiko.DSSKey, etc. to write private keys, but Paramiko 5.1.0 removed DSSKey entirely and changed constructor APIs, causing crashes. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add explicit fallback path in _load_ssh_private_with_paramiko that guards against missing DSSKey (getattr paramiko 'DSSKey', append only when present), modern private_key attribute vs legacy signing_key, and Ed25519 PyNaCl key material conversion. - Add two new test coverage methods: test_ssh_private_roundtrip – round-trip RSA/DSA/EC/Ed25519 unencrypted and password-encrypted, verifying OpenSSH PEM header and DER preservation. test_ssh_private_paramiko_fallback – explicitly exercise the Paramiko fallback for RSA and EC keys, confirming it still works under locked Paramiko 5.1.0. Also clean up the comment in load_key_bytes to reflect the new fallback behavior rather than the old 'encrypted OpenSSH keys not supported' note. --- server/packages/kca-core/src/kca/core/keys.py | 46 +++++++++---------- server/packages/kca-core/tests/test_core.py | 29 ++++++++++++ 2 files changed, 51 insertions(+), 24 deletions(-) diff --git a/server/packages/kca-core/src/kca/core/keys.py b/server/packages/kca-core/src/kca/core/keys.py index 3b524f3..798b94b 100644 --- a/server/packages/kca-core/src/kca/core/keys.py +++ b/server/packages/kca-core/src/kca/core/keys.py @@ -230,23 +230,11 @@ def ssh_public(key: PrivateKey | PublicKey, comment: str = "") -> str: def ssh_private(key: PrivateKey, password: Optional[str] = None) -> bytes: - """OpenSSH private key format, via paramiko.""" - import paramiko - + """Serialize a private key in OpenSSH format.""" key = _require_private(key) - if isinstance(key, rsa.RSAPrivateKey): - pkey = paramiko.RSAKey(key=key) - elif isinstance(key, dsa.DSAPrivateKey): - pkey = paramiko.DSSKey(key=key) - elif isinstance(key, ec.EllipticCurvePrivateKey): - pkey = paramiko.ECDSAKey(key=key) - elif isinstance(key, ed25519.Ed25519PrivateKey): - pkey = paramiko.Ed25519Key(key=key) - else: # pragma: no cover - defensive - raise ValueError("unsupported key type for SSH export") - buffer = io.StringIO() - pkey.write_private_key(buffer, password=password) - return buffer.getvalue().encode("utf-8") + encryption: serialization.KeySerializationEncryption + encryption = NoEncryption() if not password else serialization.BestAvailableEncryption(password.encode()) + return key.private_bytes(Encoding.PEM, PrivateFormat.OpenSSH, encryption) # --------------------------------------------------------------------------- @@ -451,8 +439,8 @@ def load_key_bytes(data: bytes, password: Optional[str] = None) -> Tuple[Private except Exception as exc: # noqa: BLE001 errors.append(str(exc)) - # Encrypted OpenSSH keys are not supported by cryptography; paramiko can - # read them. Convert the common key types back to cryptography objects. + # Fall back to paramiko for OpenSSH variants that cryptography cannot + # read. Convert the common key types back to cryptography objects. try: return _load_ssh_private_with_paramiko(data, password), True except Exception as exc: # noqa: BLE001 @@ -465,8 +453,12 @@ def _load_ssh_private_with_paramiko(data: bytes, password: Optional[str]) -> Pri import paramiko text = data.decode("utf-8", "replace") + key_classes = [paramiko.RSAKey, paramiko.ECDSAKey, paramiko.Ed25519Key] + dss_key = getattr(paramiko, "DSSKey", None) + if dss_key is not None: + key_classes.append(dss_key) last: Optional[Exception] = None - for cls in (paramiko.RSAKey, paramiko.ECDSAKey, paramiko.Ed25519Key, paramiko.DSSKey): + for cls in key_classes: try: pkey = cls.from_private_key(io.StringIO(text), password=password) break @@ -474,18 +466,24 @@ def _load_ssh_private_with_paramiko(data: bytes, password: Optional[str]) -> Pri last = exc else: # pragma: no cover - defensive raise ValueError(f"unable to parse OpenSSH private key: {last}") - # Paramiko keeps the cryptography signing key on its instances. This is - # stable across 3.x and avoids re-implementing key conversion. - signing_key = getattr(pkey, "_signing_key", None) - if signing_key is not None and is_private_key(signing_key): + # Paramiko 5 exposes cryptography keys as private_key; older releases use + # signing_key. Ed25519 uses a PyNaCl key in _signing_key. + signing_key = getattr(pkey, "private_key", None) + if signing_key is None: + signing_key = getattr(pkey, "signing_key", None) + if signing_key is None: + signing_key = getattr(pkey, "_signing_key", None) + if is_private_key(signing_key): return signing_key + if isinstance(pkey, paramiko.Ed25519Key) and signing_key is not None: + return ed25519.Ed25519PrivateKey.from_private_bytes(signing_key.encode()) if isinstance(pkey, paramiko.RSAKey): pub = rsa.RSAPublicNumbers(e=pkey.e, n=pkey.n) return rsa.RSAPrivateNumbers( p=pkey.p, q=pkey.q, d=pkey.d, dmp1=pkey.dmp1, dmq1=pkey.dmq1, iqmp=pkey.iqmp, public_numbers=pub, ).private_key() - if isinstance(pkey, paramiko.DSSKey): + if dss_key is not None and isinstance(pkey, dss_key): pub = dsa.DSAPublicNumbers(parameter_numbers=dsa.DSAParameterNumbers(pkey.p, pkey.q, pkey.g), y=pkey.y) return dsa.DSAPrivateNumbers(x=pkey.x, public_numbers=pub).private_key() raise ValueError("unable to convert the SSH private key") diff --git a/server/packages/kca-core/tests/test_core.py b/server/packages/kca-core/tests/test_core.py index 5e6bed8..485a0f3 100644 --- a/server/packages/kca-core/tests/test_core.py +++ b/server/packages/kca-core/tests/test_core.py @@ -192,6 +192,35 @@ class KeyTest(unittest.TestCase): self.assertTrue(private) self.assertEqual(keys.public_der(loaded), keys.public_der(key)) + def test_ssh_private_roundtrip(self) -> None: + for key_type, kwargs in [ + ("rsa", {"size": 2048}), + ("dsa", {"size": 1024}), + ("ec", {"curve": "secp256r1"}), + ("ed25519", {}), + ]: + key = keys.generate_key(key_type, **kwargs) + for password in (None, "secret"): + with self.subTest(key_type=key_type, password=password): + encoded = keys.ssh_private(key, password=password) + self.assertTrue(encoded.startswith(b"-----BEGIN OPENSSH PRIVATE KEY-----")) + loaded, private = keys.load_key_bytes(encoded, password=password) + self.assertTrue(private) + self.assertEqual(keys.private_der(loaded), keys.private_der(key)) + + def test_ssh_private_paramiko_fallback(self) -> None: + for key_type, kwargs in [ + ("rsa", {"size": 2048}), + ("ec", {"curve": "secp256r1"}), + ("ed25519", {}), + ]: + key = keys.generate_key(key_type, **kwargs) + for password in (None, "secret"): + with self.subTest(key_type=key_type, password=password): + encoded = keys.ssh_private(key, password=password) + loaded = keys._load_ssh_private_with_paramiko(encoded, password=password) + self.assertEqual(keys.private_der(loaded), keys.private_der(key)) + class CertificateTest(unittest.TestCase): def test_issue_and_verify(self) -> None: