Skip to content

Commit

Permalink
remove pyasn1 from conch and everywhere else
Browse files Browse the repository at this point in the history
  • Loading branch information
@reaperhulk
reaperhulk committed Apr 21, 2023
1 parent 525da4a commit bdee0eb
Show file tree
Hide file tree
Showing 23 changed files with 55 additions and 264 deletions.
2 changes: 0 additions & 2 deletions docs/installation/howto/optional.rst
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down
1 change: 0 additions & 1 deletion pyproject.toml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,6 @@ tls = [
]

conch = [
"pyasn1 >= 0.4",
"cryptography >= 3.3",
"appdirs >= 1.4.0",
"bcrypt >= 3.1.3",
Expand Down
1 change: 1 addition & 0 deletions src/twisted/conch/newsfragments/11843.removal
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
PyAsn1 has been removed as a conch dependency.
169 changes: 21 additions & 148 deletions src/twisted/conch/ssh/keys.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@


import binascii
import itertools
import struct
import unicodedata
import warnings
Expand All @@ -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
Expand Down Expand Up @@ -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}")

Expand Down Expand Up @@ -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
Expand Down
33 changes: 0 additions & 33 deletions src/twisted/conch/test/keydata.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -653,5 +621,4 @@
"publicDSA_openssh",
"publicRSA_lsh",
"publicRSA_openssh",
"privateRSA_openssh_alternate",
]
9 changes: 1 addition & 8 deletions src/twisted/conch/test/test_agent.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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
Expand Down
15 changes: 7 additions & 8 deletions src/twisted/conch/test/test_cftp.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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}.
Expand Down Expand Up @@ -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}.
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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):
Expand Down
4 changes: 2 additions & 2 deletions src/twisted/conch/test/test_checkers.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading

0 comments on commit bdee0eb

Please sign in to comment.