Ensure that DS digest length is consistent with digest type

diff --git a/dns/rdtypes/dsbase.py b/dns/rdtypes/dsbase.py
index 38c95484920778630d0dff4135075ec1f22ab822..d125db21ef15492a11f3de882d568c050466e12a 100644 (file)
--- a/dns/rdtypes/dsbase.py
+++ b/dns/rdtypes/dsbase.py
@@ -24,6 +24,15 @@ import dns.rdata
 import dns.rdatatype
 
 
+# Digest types registry: https://www.iana.org/assignments/ds-rr-types/ds-rr-types.xhtml
+_digest_length_by_type = {
+    1: 20,  # SHA-1, RFC 3658 Sec. 2.4
+    2: 32,  # SHA-256, RFC 4509 Sec. 2.2
+    3: 32,  # GOST R 34.11-94, RFC 5933 Sec. 4 in conjunction with RFC 4490 Sec. 2.1
+    4: 48,  # SHA-384, RFC 6605 Sec. 2
+}
+
+
 @dns.immutable.immutable
 class DSBase(dns.rdata.Rdata):
 
@@ -39,6 +48,15 @@ class DSBase(dns.rdata.Rdata):
         self.digest_type = self._as_uint8(digest_type)
         self.digest = self._as_bytes(digest)
 
+        try:
+            if self.digest_type == 0:  # reserved, RFC 3658 Sec. 2.4
+                raise ValueError('digest type 0 is reserved')
+            expected_length = _digest_length_by_type[self.digest_type]
+        except KeyError:
+            raise ValueError('unknown digest type')
+        if len(self.digest) != expected_length:
+            raise ValueError('digest length inconsistent with digest type')
+
     def to_text(self, origin=None, relativize=True, **kw):
         kw = kw.copy()
         chunksize = kw.pop('chunksize', 128)

diff --git a/tests/test_dnssec.py b/tests/test_dnssec.py
index ea82d7b8e343b505f293b12199ccd5e31dfa480b..6ea51dc859bb77dbc358c8156e509366775ddeee 100644 (file)
--- a/tests/test_dnssec.py
+++ b/tests/test_dnssec.py
@@ -497,5 +497,35 @@ class DNSSECMakeDSTestCase(unittest.TestCase):
             with self.assertRaises(dns.dnssec.UnsupportedAlgorithm):
                 ds = dns.dnssec.make_ds(abs_example, example_sep_key, algorithm)
 
+    def testInvalidDigestType(self):  # type: () -> None
+        digest_type_errors = {
+            0: 'digest type 0 is reserved',
+            5: 'unknown digest type',
+        }
+        for digest_type, msg in digest_type_errors.items():
+            with self.assertRaises(dns.exception.SyntaxError) as cm:
+                dns.rdata.from_text(dns.rdataclass.IN,
+                                    dns.rdatatype.DS,
+                                    f'18673 3 {digest_type} 71b71d4f3e11bbd71b4eff12cde69f7f9215bbe7')
+            self.assertEqual(msg, str(cm.exception))
+
+    def testInvalidDigestLength(self):  # type: () -> None
+        test_records = []
+        for rdata in [example_ds_sha1, example_ds_sha256, example_ds_sha384]:
+            flags, digest = rdata.to_text().rsplit(' ', 1)
+
+            # Make sure the construction is working
+            dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.DS, f'{flags} {digest}')
+
+            test_records.append(f'{flags} {digest[:len(digest)//2]}')  # too short digest
+            test_records.append(f'{flags} {digest*2}')  # too long digest
+
+        for record in test_records:
+            with self.assertRaises(dns.exception.SyntaxError) as cm:
+                dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.DS, record)
+
+            self.assertEqual('digest length inconsistent with digest type', str(cm.exception))
+
+
 if __name__ == '__main__':
     unittest.main()