From 087df702931f32eeb3a29957a8fe3fa31749d642 Mon Sep 17 00:00:00 2001

From: Simo Sorce <simo@redhat.com>

Date: Tue, 28 Apr 2020 10:08:02 +0200

Subject: [PATCH] Switch to python cryptography

Original patch: https://github.com/simo5/dnspython/commit/bfe84d523bd4fde7b2655857d78bba85ed05f43c

The same change in master: https://github.com/rthalley/dnspython/pull/449

---

 dns/dnssec.py        | 168 ++++++++++++++++++++-----------------------

 setup.py             |   2 +-

 tests/test_dnssec.py |  12 +---

 3 files changed, 79 insertions(+), 103 deletions(-)

diff --git a/dns/dnssec.py b/dns/dnssec.py

index 35da6b5..73e92da 100644

--- a/dns/dnssec.py

+++ b/dns/dnssec.py

@@ -17,6 +17,7 @@

 """Common DNSSEC-related functions and constants."""

+import hashlib  # used in make_ds() to avoid pycrypto dependency

 from io import BytesIO

 import struct

 import time

@@ -165,10 +166,10 @@ def make_ds(name, key, algorithm, origin=None):

     if algorithm.upper() == 'SHA1':

         dsalg = 1

-        hash = SHA1.new()

+        hash = hashlib.sha1()

     elif algorithm.upper() == 'SHA256':

         dsalg = 2

-        hash = SHA256.new()

+        hash = hashlib.sha256()

     else:

         raise UnsupportedAlgorithm('unsupported algorithm "%s"' % algorithm)

@@ -214,7 +215,7 @@ def _is_dsa(algorithm):

 def _is_ecdsa(algorithm):

-    return _have_ecdsa and (algorithm in (ECDSAP256SHA256, ECDSAP384SHA384))

+    return (algorithm in (ECDSAP256SHA256, ECDSAP384SHA384))

 def _is_md5(algorithm):

@@ -240,18 +241,26 @@ def _is_sha512(algorithm):

 def _make_hash(algorithm):

     if _is_md5(algorithm):

-        return MD5.new()

+        return hashes.MD5()

     if _is_sha1(algorithm):

-        return SHA1.new()

+        return hashes.SHA1()

     if _is_sha256(algorithm):

-        return SHA256.new()

+        return hashes.SHA256()

     if _is_sha384(algorithm):

-        return SHA384.new()

+        return hashes.SHA384()

     if _is_sha512(algorithm):

-        return SHA512.new()

+        return hashes.SHA512()

+    if algorithm == ED25519:

+        return hashes.SHA512()

+    if algorithm == ED448:

+        return hashes.SHAKE256(114)

     raise ValidationFailure('unknown hash for algorithm %u' % algorithm)

+def _bytes_to_long(b):

+    return int.from_bytes(b, 'big')

+

+

 def _make_algorithm_id(algorithm):

     if _is_md5(algorithm):

         oid = [0x2a, 0x86, 0x48, 0x86, 0xf7, 0x0d, 0x02, 0x05]

@@ -316,8 +325,6 @@ def _validate_rrsig(rrset, rrsig, keys, origin=None, now=None):

         if rrsig.inception > now:

             raise ValidationFailure('not yet valid')

-        hash = _make_hash(rrsig.algorithm)

-

         if _is_rsa(rrsig.algorithm):

             keyptr = candidate_key.key

             (bytes_,) = struct.unpack('!B', keyptr[0:1])

@@ -328,9 +335,9 @@ def _validate_rrsig(rrset, rrsig, keys, origin=None, now=None):

             rsa_e = keyptr[0:bytes_]

             rsa_n = keyptr[bytes_:]

             try:

-                pubkey = CryptoRSA.construct(

-                    (number.bytes_to_long(rsa_n),

-                     number.bytes_to_long(rsa_e)))

+                public_key = rsa.RSAPublicNumbers(

+                    _bytes_to_long(rsa_e),

+                    _bytes_to_long(rsa_n)).public_key(default_backend())

             except ValueError:

                 raise ValidationFailure('invalid public key')

             sig = rrsig.signature

@@ -346,42 +353,47 @@ def _validate_rrsig(rrset, rrsig, keys, origin=None, now=None):

             dsa_g = keyptr[0:octets]

             keyptr = keyptr[octets:]

             dsa_y = keyptr[0:octets]

-            pubkey = CryptoDSA.construct(

-                (number.bytes_to_long(dsa_y),

-                 number.bytes_to_long(dsa_g),

-                 number.bytes_to_long(dsa_p),

-                 number.bytes_to_long(dsa_q)))

-            sig = rrsig.signature[1:]

+            try:

+                public_key = dsa.DSAPublicNumbers(

+                    _bytes_to_long(dsa_y),

+                    dsa.DSAParameterNumbers(

+                        _bytes_to_long(dsa_p),

+                        _bytes_to_long(dsa_q),

+                        _bytes_to_long(dsa_g))).public_key(default_backend())

+            except ValueError:

+                raise ValidationFailure('invalid public key')

+            sig_r = rrsig.signature[1:21]

+            sig_s = rrsig.signature[21:]

+            sig = utils.encode_dss_signature(_bytes_to_long(sig_r),

+                                             _bytes_to_long(sig_s))

         elif _is_ecdsa(rrsig.algorithm):

-            # use ecdsa for NIST-384p -- not currently supported by pycryptodome

-

             keyptr = candidate_key.key

-

             if rrsig.algorithm == ECDSAP256SHA256:

-                curve = ecdsa.curves.NIST256p

-                key_len = 32

+                curve = ec.SECP256R1()

+                octets = 32

             elif rrsig.algorithm == ECDSAP384SHA384:

-                curve = ecdsa.curves.NIST384p

-                key_len = 48

-

-            x = number.bytes_to_long(keyptr[0:key_len])

-            y = number.bytes_to_long(keyptr[key_len:key_len * 2])

-            if not ecdsa.ecdsa.point_is_valid(curve.generator, x, y):

-                raise ValidationFailure('invalid ECDSA key')

-            point = ecdsa.ellipticcurve.Point(curve.curve, x, y, curve.order)

-            verifying_key = ecdsa.keys.VerifyingKey.from_public_point(point,

-                                                                      curve)

-            pubkey = ECKeyWrapper(verifying_key, key_len)

-            r = rrsig.signature[:key_len]

-            s = rrsig.signature[key_len:]

-            sig = ecdsa.ecdsa.Signature(number.bytes_to_long(r),

-                                        number.bytes_to_long(s))

+                curve = ec.SECP384R1()

+                octets = 48

+            ecdsa_x = keyptr[0:octets]

+            ecdsa_y = keyptr[octets:octets * 2]

+            try:

+                public_key = ec.EllipticCurvePublicNumbers(

+                    curve=curve,

+                    x=_bytes_to_long(ecdsa_x),

+                    y=_bytes_to_long(ecdsa_y)).public_key(default_backend())

+            except ValueError:

+                raise ValidationFailure('invalid public key')

+            sig_r = rrsig.signature[0:octets]

+            sig_s = rrsig.signature[octets:]

+            sig = utils.encode_dss_signature(_bytes_to_long(sig_r),

+                                             _bytes_to_long(sig_s))

         else:

             raise ValidationFailure('unknown algorithm %u' % rrsig.algorithm)

-        hash.update(_to_rdata(rrsig, origin)[:18])

-        hash.update(rrsig.signer.to_digestable(origin))

+        data = b''

+        data += _to_rdata(rrsig, origin)[:18]

+        data += rrsig.signer.to_digestable(origin)

         if rrsig.labels < len(rrname) - 1:

             suffix = rrname.split(rrsig.labels + 1)[1]

@@ -391,25 +403,21 @@ def _validate_rrsig(rrset, rrsig, keys, origin=None, now=None):

                               rrsig.original_ttl)

         rrlist = sorted(rdataset)

         for rr in rrlist:

-            hash.update(rrnamebuf)

-            hash.update(rrfixed)

+            data += rrnamebuf

+            data += rrfixed

             rrdata = rr.to_digestable(origin)

             rrlen = struct.pack('!H', len(rrdata))

-            hash.update(rrlen)

-            hash.update(rrdata)

+            data += rrlen

+            data += rrdata

+        chosen_hash = _make_hash(rrsig.algorithm)

         try:

             if _is_rsa(rrsig.algorithm):

-                verifier = pkcs1_15.new(pubkey)

-                # will raise ValueError if verify fails:

-                verifier.verify(hash, sig)

+                public_key.verify(sig, data, padding.PKCS1v15(), chosen_hash)

             elif _is_dsa(rrsig.algorithm):

-                verifier = DSS.new(pubkey, 'fips-186-3')

-                verifier.verify(hash, sig)

+                public_key.verify(sig, data, chosen_hash)

             elif _is_ecdsa(rrsig.algorithm):

-                digest = hash.digest()

-                if not pubkey.verify(digest, sig):

-                    raise ValueError

+                public_key.verify(sig, data, ec.ECDSA(chosen_hash))

             else:

                 # Raise here for code clarity; this won't actually ever happen

                 # since if the algorithm is really unknown we'd already have

@@ -417,7 +425,7 @@ def _validate_rrsig(rrset, rrsig, keys, origin=None, now=None):

                 raise ValidationFailure('unknown algorithm %u' % rrsig.algorithm)

             # If we got here, we successfully verified so we can return without error

             return

-        except ValueError:

+        except InvalidSignature:

             # this happens on an individual validation failure

             continue

     # nothing verified -- raise failure:

@@ -472,48 +480,24 @@ def _validate(rrset, rrsigset, keys, origin=None, now=None):

     raise ValidationFailure("no RRSIGs validated")

-def _need_pycrypto(*args, **kwargs):

-    raise NotImplementedError("DNSSEC validation requires pycryptodome/pycryptodomex")

+def _need_pyca(*args, **kwargs):

+    raise NotImplementedError("DNSSEC validation requires python cryptography")

 try:

-    try:

-        # test we're using pycryptodome, not pycrypto (which misses SHA1 for example)

-        from Crypto.Hash import MD5, SHA1, SHA256, SHA384, SHA512

-        from Crypto.PublicKey import RSA as CryptoRSA, DSA as CryptoDSA

-        from Crypto.Signature import pkcs1_15, DSS

-        from Crypto.Util import number

-    except ImportError:

-        from Cryptodome.Hash import MD5, SHA1, SHA256, SHA384, SHA512

-        from Cryptodome.PublicKey import RSA as CryptoRSA, DSA as CryptoDSA

-        from Cryptodome.Signature import pkcs1_15, DSS

-        from Cryptodome.Util import number

+    from cryptography.exceptions import InvalidSignature

+    from cryptography.hazmat.backends import default_backend

+    from cryptography.hazmat.primitives import hashes

+    from cryptography.hazmat.primitives.asymmetric import padding

+    from cryptography.hazmat.primitives.asymmetric import utils

+    from cryptography.hazmat.primitives.asymmetric import dsa

+    from cryptography.hazmat.primitives.asymmetric import ec

+    from cryptography.hazmat.primitives.asymmetric import rsa

 except ImportError:

-    validate = _need_pycrypto

-    validate_rrsig = _need_pycrypto

-    _have_pycrypto = False

-    _have_ecdsa = False

+    validate = _need_pyca

+    validate_rrsig = _need_pyca

+    _have_pyca = False

 else:

     validate = _validate

     validate_rrsig = _validate_rrsig

-    _have_pycrypto = True

-

-    try:

-        import ecdsa

-        import ecdsa.ecdsa

-        import ecdsa.ellipticcurve

-        import ecdsa.keys

-    except ImportError:

-        _have_ecdsa = False

-    else:

-        _have_ecdsa = True

-

-        class ECKeyWrapper(object):

-

-            def __init__(self, key, key_len):

-                self.key = key

-                self.key_len = key_len

-

-            def verify(self, digest, sig):

-                diglong = number.bytes_to_long(digest)

-                return self.key.pubkey.verifies(diglong, sig)

+    _have_pyca = True

diff --git a/setup.py b/setup.py

index 743d43c..2ee38a7 100755

--- a/setup.py

+++ b/setup.py

@@ -75,7 +75,7 @@ direct manipulation of DNS zones, messages, names, and records.""",

     'provides': ['dns'],

     'extras_require': {

         'IDNA': ['idna>=2.1'],

-        'DNSSEC': ['pycryptodome', 'ecdsa>=0.13'],

+        'DNSSEC': ['cryptography>=2.3'],

         },

     'ext_modules': ext_modules if compile_cython else None,

     'zip_safe': False if compile_cython else None,

diff --git a/tests/test_dnssec.py b/tests/test_dnssec.py

index c87862a..20b52b2 100644

--- a/tests/test_dnssec.py

+++ b/tests/test_dnssec.py

@@ -151,8 +151,8 @@ abs_ecdsa384_soa_rrsig = dns.rrset.from_text('example.', 86400, 'IN', 'RRSIG',

-@unittest.skipUnless(dns.dnssec._have_pycrypto,

-                     "Pycryptodome cannot be imported")

+@unittest.skipUnless(dns.dnssec._have_pyca,

+                     "Python Cryptography cannot be imported")

 class DNSSECValidatorTestCase(unittest.TestCase):

     def testAbsoluteRSAGood(self): # type: () -> None

@@ -199,28 +199,20 @@ class DNSSECValidatorTestCase(unittest.TestCase):

         ds = dns.dnssec.make_ds(abs_example, example_sep_key, 'SHA256')

         self.failUnless(ds == example_ds_sha256)

-    @unittest.skipUnless(dns.dnssec._have_ecdsa,

-                         "python ECDSA cannot be imported")

     def testAbsoluteECDSA256Good(self): # type: () -> None

         dns.dnssec.validate(abs_ecdsa256_soa, abs_ecdsa256_soa_rrsig,

                             abs_ecdsa256_keys, None, when3)

-    @unittest.skipUnless(dns.dnssec._have_ecdsa,

-                         "python ECDSA cannot be imported")

     def testAbsoluteECDSA256Bad(self): # type: () -> None

         def bad(): # type: () -> None

             dns.dnssec.validate(abs_other_ecdsa256_soa, abs_ecdsa256_soa_rrsig,

                                 abs_ecdsa256_keys, None, when3)

         self.failUnlessRaises(dns.dnssec.ValidationFailure, bad)

-    @unittest.skipUnless(dns.dnssec._have_ecdsa,

-                         "python ECDSA cannot be imported")

     def testAbsoluteECDSA384Good(self): # type: () -> None

         dns.dnssec.validate(abs_ecdsa384_soa, abs_ecdsa384_soa_rrsig,

                             abs_ecdsa384_keys, None, when4)

-    @unittest.skipUnless(dns.dnssec._have_ecdsa,

-                         "python ECDSA cannot be imported")

     def testAbsoluteECDSA384Bad(self): # type: () -> None

         def bad(): # type: () -> None

             dns.dnssec.validate(abs_other_ecdsa384_soa, abs_ecdsa384_soa_rrsig,

--

2.26.2