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: