feat(dnssec)!: Implement parser for DNSSEC DS records (#72)

Author: c0r0n3r

cryptoparser/dnsrec/record.py

@@ -5,6 +5,7 @@
 import enum
 
 import attr
+import six
 
 from cryptodatahub.common.algorithm import Authentication, NamedGroup, Signature
 from cryptodatahub.common.exception import InvalidValue
@@ -15,11 +16,12 @@
     PublicKeyParamsEddsa,
     PublicKeyParamsRsa,
 )
-from cryptodatahub.dnssec.algorithm import DnsSecAlgorithm
+
+from cryptodatahub.dnssec.algorithm import DnsSecAlgorithm, DnsSecDigestType
 
 from cryptoparser.common.base import OneByteEnumParsable, Serializable
 from cryptoparser.common.exception import NotEnoughData
-from cryptoparser.common.parse import ByteOrder, ParsableBase, ParserBinary, ComposerBinary
+from cryptoparser.common.parse import ByteOrder, ComposerBinary, ParsableBase, ParserBinary
 
 
 class DnsSecProtocol(enum.Enum):
@@ -257,3 +259,47 @@ def compose(self):
         key_bytes = self.compose_key(self.key)
 
         return composer.composed_bytes + key_bytes
+
+
+class DnsSecDigestTypeFactory(OneByteEnumParsable):
+    @classmethod
+    def get_enum_class(cls):
+        return DnsSecDigestType
+
+    @abc.abstractmethod
+    def compose(self):
+        raise NotImplementedError()
+
+
+@attr.s
+class DnsRecordDs(ParsableBase):
+    HEADER_SIZE = 4
+
+    key_tag = attr.ib(validator=attr.validators.instance_of(six.integer_types))
+    algorithm = attr.ib(validator=attr.validators.instance_of(DnsSecAlgorithm))
+    digest_type = attr.ib(validator=attr.validators.instance_of(DnsSecDigestType))
+    digest = attr.ib(validator=attr.validators.instance_of((bytes, bytearray)))
+
+    @classmethod
+    def _parse(cls, parsable):
+        if len(parsable) < cls.HEADER_SIZE:
+            raise NotEnoughData(cls.HEADER_SIZE - len(parsable))
+
+        parser = ParserBinary(parsable)
+
+        parser.parse_numeric('key_tag', 2)
+        parser.parse_parsable('algorithm', DnsSecAlgorithmFactory)
+        parser.parse_parsable('digest_type', DnsSecDigestTypeFactory)
+        parser.parse_raw('digest', parser.unparsed_length)
+
+        return cls(**parser), parser.parsed_length
+
+    def compose(self):
+        composer = ComposerBinary()
+
+        composer.compose_numeric(self.key_tag, 2)
+        composer.compose_numeric_enum_coded(self.algorithm)
+        composer.compose_numeric_enum_coded(self.digest_type)
+        composer.compose_raw(self.digest)
+
+        return composer.composed_bytes

test/dnsrec/test_record.py

@@ -15,10 +15,10 @@
     PublicKeyParamsRsa,
 )
 
-from cryptodatahub.dnssec.algorithm import DnsSecAlgorithm
+from cryptodatahub.dnssec.algorithm import DnsSecAlgorithm, DnsSecDigestType
 
 from cryptoparser.common.exception import NotEnoughData
-from cryptoparser.dnsrec.record import DnsRecordDnskey, DnsSecFlag, DnsSecProtocol
+from cryptoparser.dnsrec.record import DnsRecordDnskey, DnsRecordDs, DnsSecFlag, DnsSecProtocol
 
 
 class TestDnsRecordDnskey(unittest.TestCase):
@@ -330,3 +330,35 @@ def test_real(self):
             protocol=DnsSecProtocol.V3,
         )
         self.assertEqual(dns_record.key_tag, 59732)
+
+
+class TestDnsRecordDs(unittest.TestCase):
+    def setUp(self):
+        self.record_bytes = bytes(
+            b'\x00\x01' +        # key_tag: 1
+            b'\x01' +            # algorithm: RSAMD5
+            b'\x02' +            # digest_type: SHA_256
+            32 * b'\xff' +       # digest
+            b''
+        )
+        self.record = DnsRecordDs(
+            key_tag=1,
+            algorithm=DnsSecAlgorithm.RSAMD5,
+            digest_type=DnsSecDigestType.SHA_256,
+            digest=32 * b'\xff',
+        )
+
+    def test_error_not_enough_data(self):
+        with self.assertRaises(NotEnoughData) as context_manager:
+            DnsRecordDs.parse_exact_size(b'\x00')
+
+        self.assertEqual(
+            context_manager.exception.bytes_needed,
+            DnsRecordDs.HEADER_SIZE - 1
+        )
+
+    def test_parse(self):
+        self.assertEqual(DnsRecordDs.parse_exact_size(self.record_bytes), self.record)
+
+    def test_compose(self):
+        self.assertEqual(self.record.compose(), self.record_bytes)