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.

- 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.
This commit is contained in:
2026-10-26 12:06:16 +08:00
parent bd352051e0
commit 987fdc0e12
2 changed files with 51 additions and 24 deletions
+21 -23
View File
@@ -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.
# 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 signing_key is not None and is_private_key(signing_key):
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")
@@ -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: