remove pyasn1 from conch and everywhere else

docs/installation/howto/optional.rst

@@ -33,7 +33,6 @@ The following optional dependencies are supported:
 
 * **conch** - packages for working with conch/SSH.
 
-  * `pyasn1`_
   * `cryptography`_
 
 * **conch-nacl** - **conch** options and `PyNaCl`_ to support Ed25519 keys on systems with OpenSSL < 1.1.1b.
@@ -65,7 +64,6 @@ The following optional dependencies are supported:
 .. _pydoctor: https://pypi.python.org/pypi/pydoctor
 .. _pyOpenSSL: https://pypi.python.org/pypi/pyOpenSSL
 .. _service_identity: https://pypi.python.org/pypi/service_identity
-.. _pyasn1: https://pypi.python.org/pypi/pyasn1
 .. _cryptography: https://pypi.python.org/pypi/cryptography
 .. _PyNaCl: https://pypi.python.org/pypi/PyNaCl
 .. _SOAPpy: https://pypi.python.org/pypi/SOAPpy

pyproject.toml

@@ -85,7 +85,6 @@ tls = [
 ]
 
 conch = [
-    "pyasn1 >= 0.4",
     "cryptography >= 3.3",
     "appdirs >= 1.4.0",
     "bcrypt >= 3.1.3",

src/twisted/conch/newsfragments/11843.removal

@@ -0,0 +1 @@
+PyAsn1 has been removed as a conch dependency.

src/twisted/conch/ssh/keys.py

@@ -8,7 +8,6 @@
 
 
 import binascii
-import itertools
 import struct
 import unicodedata
 import warnings
@@ -27,12 +26,6 @@
     load_pem_private_key,
     load_ssh_public_key,
 )
-from pyasn1.codec.ber import (  # type: ignore[import]
-    decoder as berDecoder,
-    encoder as berEncoder,
-)
-from pyasn1.error import PyAsn1Error  # type: ignore[import]
-from pyasn1.type import univ  # type: ignore[import]
 
 from twisted.conch.ssh import common, sexpy
 from twisted.conch.ssh.common import int_to_bytes
@@ -516,90 +509,21 @@ def _fromPrivateOpenSSH_PEM(cls, data, passphrase):
         """
         lines = data.strip().splitlines()
         kind = lines[0][11:-17]
-        if lines[1].startswith(b"Proc-Type: 4,ENCRYPTED"):
-            if not passphrase:
+        # cryptography considers an empty byte string a passphrase, but
+        # twisted considers that to be "no password". So we need to convert
+        # to None on empty.
+        if not passphrase:
+            passphrase = None
+        if kind in (b"EC", b"RSA", b"DSA"):
+            try:
+                key = load_pem_private_key(data, passphrase, default_backend())
+            except TypeError:
                 raise EncryptedKeyError(
-                    "Passphrase must be provided " "for an encrypted key"
+                    "Passphrase must be provided for an encrypted key"
                 )
-
-            # Determine cipher and initialization vector
-            try:
-                _, cipherIVInfo = lines[2].split(b" ", 1)
-                cipher, ivdata = cipherIVInfo.rstrip().split(b",", 1)
             except ValueError:
-                raise BadKeyError(f"invalid DEK-info {lines[2]!r}")
-
-            if cipher in (b"AES-128-CBC", b"AES-256-CBC"):
-                algorithmClass = algorithms.AES
-                keySize = int(cipher.split(b"-")[1]) // 8
-                if len(ivdata) != 32:
-                    raise BadKeyError("AES encrypted key with a bad IV")
-            elif cipher == b"DES-EDE3-CBC":
-                algorithmClass = algorithms.TripleDES
-                keySize = 24
-                if len(ivdata) != 16:
-                    raise BadKeyError("DES encrypted key with a bad IV")
-            else:
-                raise BadKeyError(f"unknown encryption type {cipher!r}")
-
-            # Extract keyData for decoding
-            iv = bytes(
-                bytearray(int(ivdata[i : i + 2], 16) for i in range(0, len(ivdata), 2))
-            )
-            ba = md5(passphrase + iv[:8]).digest()
-            bb = md5(ba + passphrase + iv[:8]).digest()
-            decKey = (ba + bb)[:keySize]
-            b64Data = decodebytes(b"".join(lines[3:-1]))
-
-            decryptor = Cipher(
-                algorithmClass(decKey), modes.CBC(iv), backend=default_backend()
-            ).decryptor()
-            keyData = decryptor.update(b64Data) + decryptor.finalize()
-
-            removeLen = ord(keyData[-1:])
-            keyData = keyData[:-removeLen]
-        else:
-            b64Data = b"".join(lines[1:-1])
-            keyData = decodebytes(b64Data)
-
-        try:
-            decodedKey = berDecoder.decode(keyData)[0]
-        except PyAsn1Error as asn1Error:
-            raise BadKeyError(f"Failed to decode key (Bad Passphrase?): {asn1Error}")
-
-        if kind == b"EC":
-            return cls(load_pem_private_key(data, passphrase, default_backend()))
-
-        if kind == b"RSA":
-            if len(decodedKey) == 2:  # Alternate RSA key
-                decodedKey = decodedKey[0]
-            if len(decodedKey) < 6:
-                raise BadKeyError("RSA key failed to decode properly")
-
-            n, e, d, p, q, dmp1, dmq1, iqmp = (int(value) for value in decodedKey[1:9])
-            return cls(
-                rsa.RSAPrivateNumbers(
-                    p=p,
-                    q=q,
-                    d=d,
-                    dmp1=dmp1,
-                    dmq1=dmq1,
-                    iqmp=iqmp,
-                    public_numbers=rsa.RSAPublicNumbers(e=e, n=n),
-                ).private_key(default_backend())
-            )
-        elif kind == b"DSA":
-            p, q, g, y, x = (int(value) for value in decodedKey[1:6])
-            if len(decodedKey) < 6:
-                raise BadKeyError("DSA key failed to decode properly")
-            return cls(
-                dsa.DSAPrivateNumbers(
-                    x=x,
-                    public_numbers=dsa.DSAPublicNumbers(
-                        y=y, parameter_numbers=dsa.DSAParameterNumbers(p=p, q=q, g=g)
-                    ),
-                ).private_key(backend=default_backend())
-            )
+                raise BadKeyError("Failed to decode key (Bad Passphrase?)")
+            return cls(key)
         else:
             raise BadKeyError(f"unknown key type {kind}")
 
@@ -1563,75 +1487,24 @@ def _toPrivateOpenSSH_PEM(self, passphrase=None):
         @param passphrase: The passphrase to encrypt the key with, or L{None}
         if it is not encrypted.
         """
-        if self.type() == "EC":
-            # EC keys has complex ASN.1 structure hence we do this this way.
-            if not passphrase:
-                # unencrypted private key
-                encryptor = serialization.NoEncryption()
-            else:
-                encryptor = serialization.BestAvailableEncryption(passphrase)
-
+        if not passphrase:
+            # unencrypted private key
+            encryptor = serialization.NoEncryption()
+        else:
+            encryptor = serialization.BestAvailableEncryption(passphrase)
+        if self.type() != "Ed25519":
             return self._keyObject.private_bytes(
                 serialization.Encoding.PEM,
                 serialization.PrivateFormat.TraditionalOpenSSL,
                 encryptor,
             )
-        elif self.type() == "Ed25519":
+        else:
+            # TODO: why not just support serialization here
+            assert self.type() == "Ed25519"
             raise ValueError(
                 "cannot serialize Ed25519 key to OpenSSH PEM format; use v1 " "instead"
             )
 
-        data = self.data()
-        lines = [
-            b"".join(
-                (b"-----BEGIN ", self.type().encode("ascii"), b" PRIVATE KEY-----")
-            )
-        ]
-        if self.type() == "RSA":
-            p, q = data["p"], data["q"]
-            iqmp = rsa.rsa_crt_iqmp(p, q)
-            objData = (
-                0,
-                data["n"],
-                data["e"],
-                data["d"],
-                p,
-                q,
-                data["d"] % (p - 1),
-                data["d"] % (q - 1),
-                iqmp,
-            )
-        else:
-            objData = (0, data["p"], data["q"], data["g"], data["y"], data["x"])
-        asn1Sequence = univ.Sequence()
-        for index, value in zip(itertools.count(), objData):
-            asn1Sequence.setComponentByPosition(index, univ.Integer(value))
-        asn1Data = berEncoder.encode(asn1Sequence)
-        if passphrase:
-            iv = randbytes.secureRandom(8)
-            hexiv = "".join([f"{ord(x):02X}" for x in iterbytes(iv)])
-            hexiv = hexiv.encode("ascii")
-            lines.append(b"Proc-Type: 4,ENCRYPTED")
-            lines.append(b"DEK-Info: DES-EDE3-CBC," + hexiv + b"\n")
-            ba = md5(passphrase + iv).digest()
-            bb = md5(ba + passphrase + iv).digest()
-            encKey = (ba + bb)[:24]
-            padLen = 8 - (len(asn1Data) % 8)
-            asn1Data += bytes((padLen,)) * padLen
-
-            encryptor = Cipher(
-                algorithms.TripleDES(encKey), modes.CBC(iv), backend=default_backend()
-            ).encryptor()
-
-            asn1Data = encryptor.update(asn1Data) + encryptor.finalize()
-
-        b64Data = encodebytes(asn1Data).replace(b"\n", b"")
-        lines += [b64Data[i : i + 64] for i in range(0, len(b64Data), 64)]
-        lines.append(
-            b"".join((b"-----END ", self.type().encode("ascii"), b" PRIVATE KEY-----"))
-        )
-        return b"\n".join(lines)
-
     def _toString_OPENSSH(self, subtype=None, comment=None, passphrase=None):
         """
         Return a public or private OpenSSH string.  See

src/twisted/conch/test/keydata.py

@@ -283,38 +283,6 @@
 rXp0IjScTxfLP+Cq5V6lJ94/pX8Ppoj1FdZfNxeS4NYFSRA7kvY=
 -----END RSA PRIVATE KEY-----"""
 
-# Some versions of OpenSSH generate these (slightly different keys): the PKCS#1
-# structure is wrapped in an extra ASN.1 SEQUENCE and there's an empty SEQUENCE
-# following it. It is not any standard key format and was probably a bug in
-# OpenSSH at some point.
-privateRSA_openssh_alternate = b"""-----BEGIN RSA PRIVATE KEY-----
-MIIEqDCCBKICAQACggEBANVqrHgj1tYb7CWhUMR3Y1CERQFVQhQqKuDQYO7U6aOt
-Svo5Bl6EVXVfADa/b6oqP4MmN8FpLlv98PPSfdaYzTpAeNXKqBjAEZMkCQyBTI/3
-nO0TFmqkBOlJd8PkVWSzeWieLAjrrOgELSF3BaeO71MwDaXluz1q4gk2b/00031v
-Rv+H2qkpJ6r/rfWF5j4auHodSrHqwFr3MN8fwqTk7z+RSZZA1Rl3LTfDXuydpjpE
-pcKkKd3Vupw9RbPGLBhk1bo936t/zUKsp/EYC6BYFWILpCpuQ8PkBJ81o0eORu0z
-pWW9vDspbgILV9906BO0NzV+g18gJmCm3K2Lxmx5mPcCAwEAAQKCAQAhTAhmoijV
-tPuOD3IbhQkAufJON/AcV0vjUX+eI6fkOphVG+qLepgevNi6sfmJEhhgrOjMC04J
-WkBqui+Z+LMkYIS5zmmVmvni/B9RTScV2ysnre+0aay+fRDrhkdwc7QAh5UVOzf5
-5xTngLtoHhvm3btzY7ln5rInf8/PMJvCmP3ZGDYvNi7xPYF6n+EDLUfbNFFiOd1P
-6ayoi9nW84TEF7lxnQYIQnhNu8Uq9MNYzVUr7b4zXwTqe+YEJGPyLdc9G2zVnGND
-L5KIjT5u2hg32A8lZ4kduUY0XsnOxIvtklozBw/fhgj5kunb6zgINsnNzQoBSFs5
-PnrKxoCp3NQ5AoGBANlwBtjivNR4kVCU1MEbiThsRmRaUaCaBz1IjwNRzGsSjn0a
-sWXncXU54DIFdY0YTK+TsUmxZl94YnrRDMrmTUOznPRrfeYMmNzPIWKO1S4S3gSu
-1yRugzGiFaJEPSKpYiYiubLtVAqdCIOnBw3/GRiO2Ksd2kicMWgRoWZt49gdAoGB
-APtEF4ukNr4eNx2n9mFsBMSq3Xg+B4weMwKuAxSHg3rlnn0IZ6jyqr8ScM9yqafH
-Cx2I1SD9nGPKRzBVTovEz/R/FqSSEnShCcLEbpyMM++l5ffgK61PXBGqGoQ3W/16
-6sPNfLDI5B9UY7XHr9/0Caf8xyX8XOmR15LFmB5W07EjAoGAUa/pkp+UC0qEZT6U
-szuSELV0uIzJ78kOATL6L2gSoQMmrs9RaBRMJpsopAIzCF/hp3CYATR5XlKOxM82
-vB9LVazrwVOEx+FhqErUov9ADYAfEqlQwCoYdZQMBpsWUKhL7EHNe+/3S8l1AmjE
-mLiGiBhaQ+cCM5ciZJODDEUqfO0CgYBpZjfGQN0hxQTzsLg+R5R8dvwt6z85PJXD
-QwFRxEKX8+gWpMbu7NRJEFA4BO47zdfQzMwyaZAHoBtan/4xzR46fnEeGZQaTk8M
-319S1dEXbuzXnLZVnduOIV+8JIi2/K+r8O+kLLDcn4awAxK4i+LdD8DuIz1KUP4v
-uClGWL+2JwKBgGYW+SA00FQlvGExrIL775w1Hn5KVQJolQ0Kk74ev+FA+pCnVHAx
-6Xj84Ga3Inea693V0jBGyuLXXkGbz7VINVGqJdze2zQpSCHb1nT8fuUvU/ecCXC5
-5KB2pq16dCI0nE8Xyz/gquVepSfeP6V/D6aI9RXWXzcXkuDWBUkQO5L2MAA=
------END RSA PRIVATE KEY-----"""
-
 # New format introduced in OpenSSH 6.5
 privateRSA_openssh_new = b"""-----BEGIN OPENSSH PRIVATE KEY-----
 b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAABFwAAAAdzc2gtcn
@@ -653,5 +621,4 @@
     "publicDSA_openssh",
     "publicRSA_lsh",
     "publicRSA_openssh",
-    "privateRSA_openssh_alternate",
 ]

src/twisted/conch/test/test_agent.py

@@ -17,13 +17,6 @@
 else:
     cryptography = _cryptography
 
-try:
-    import pyasn1 as _pyasn1  # type: ignore[import]
-except ImportError:
-    pyasn1 = None
-else:
-    pyasn1 = _pyasn1
-
 try:
     from twisted.conch.ssh import agent as _agent, keys as _keys
 except ImportError:
@@ -51,7 +44,7 @@ class AgentTestBase(unittest.TestCase):
     """
 
     if agent is None or keys is None:
-        skip = "Cannot run without cryptography or PyASN1"
+        skip = "Cannot run without cryptography"
 
     def setUp(self):
         # wire up our client <-> server

src/twisted/conch/test/test_cftp.py

@@ -33,11 +33,10 @@
 from twisted.python.reflect import requireModule
 from twisted.trial.unittest import TestCase
 
-pyasn1 = requireModule("pyasn1")
 cryptography = requireModule("cryptography")
 unix = requireModule("twisted.conch.unix")
 
-if cryptography and pyasn1:
+if cryptography:
     try:
         from twisted.conch.scripts import cftp
         from twisted.conch.scripts.cftp import SSHSession
@@ -50,11 +49,11 @@
         pass
 
 skipTests = False
-if None in (unix, cryptography, pyasn1, interfaces.IReactorProcess(reactor, None)):
+if None in (unix, cryptography, interfaces.IReactorProcess(reactor, None)):
     skipTests = True
 
 
-@skipIf(skipTests, "don't run w/o spawnProcess or cryptography or pyasn1")
+@skipIf(skipTests, "don't run w/o spawnProcess or cryptography")
 class SSHSessionTests(TestCase):
     """
     Tests for L{twisted.conch.scripts.cftp.SSHSession}.
@@ -388,7 +387,7 @@ def getvalue(self):
         return BytesIO.getvalue(self)
 
 
-@skipIf(skipTests, "don't run w/o spawnProcess or cryptography or pyasn1")
+@skipIf(skipTests, "don't run w/o spawnProcess or cryptography")
 class StdioClientTests(TestCase):
     """
     Tests for L{cftp.StdioClient}.
@@ -931,7 +930,7 @@ def tearDown(self):
         return SFTPTestBase.tearDown(self)
 
 
-@skipIf(skipTests, "don't run w/o spawnProcess or cryptography or pyasn1")
+@skipIf(skipTests, "don't run w/o spawnProcess or cryptography")
 class OurServerCmdLineClientTests(CFTPClientTestBase):
     """
     Functional tests which launch a SFTP server over TCP on localhost and check
@@ -1330,7 +1329,7 @@ def _check(results):
         return d
 
 
-@skipIf(skipTests, "don't run w/o spawnProcess or cryptography or pyasn1")
+@skipIf(skipTests, "don't run w/o spawnProcess or cryptography")
 class OurServerBatchFileTests(CFTPClientTestBase):
     """
     Functional tests which launch a SFTP server over localhost and checks csftp
@@ -1434,7 +1433,7 @@ def _cbCheckResult(res):
         return d
 
 
-@skipIf(skipTests, "don't run w/o spawnProcess or cryptography or pyasn1")
+@skipIf(skipTests, "don't run w/o spawnProcess or cryptography")
 @skipIf(not which("ssh"), "no ssh command-line client available")
 @skipIf(not which("sftp"), "no sftp command-line client available")
 class OurServerSftpClientTests(CFTPClientTestBase):

src/twisted/conch/test/test_checkers.py

@@ -38,14 +38,14 @@
 from twisted.test.test_process import MockOS
 from twisted.trial.unittest import TestCase
 
-if requireModule("cryptography") and requireModule("pyasn1"):
+if requireModule("cryptography"):
     dependencySkip = None
     from twisted.conch import checkers
     from twisted.conch.error import NotEnoughAuthentication, ValidPublicKey
     from twisted.conch.ssh import keys
     from twisted.conch.test import keydata
 else:
-    dependencySkip = "can't run without cryptography and PyASN1"
+    dependencySkip = "can't run without cryptography"
 
 if getattr(os, "geteuid", None) is not None:
     euidSkip = None