| |
@@ -0,0 +1,343 @@
|
| |
+ 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
|
| |
+
|
| |
jcapitao
commented 3 years ago
Hello @kevin
I bundled last version of dns1 (1.16.0) with patch submited in [1] as Miro suggested in [2].
Scratch build result: https://koji.fedoraproject.org/koji/taskinfo?taskID=56440651
I installed it in my rawhide environment, it looks fine but better if you could to double check.
[1] https://src.fedoraproject.org/rpms/python-dns/c/84c2874c7a39e72e7c074afb8f7682fa46ab2859?branch=master
[2] https://bugzilla.redhat.com/show_bug.cgi?id=1896191#c5