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

Author: c0r0n3r

cryptoparser/common/parse.py

@@ -457,11 +457,11 @@ def parse_time_delta(self, name):
 class ParserBinary(ParserBase):
     byte_order = attr.ib(default=ByteOrder.NETWORK, validator=attr.validators.in_(ByteOrder))
 
-    def parse_timestamp(self, name, milliseconds=False):
-        value, parsed_length = self._parse_numeric_array(name, 1, 8, int)
+    def parse_timestamp(self, name, milliseconds=False, item_size=8):
+        value, parsed_length = self._parse_numeric_array(name, 1, item_size, int)
 
         self._parsed_length += parsed_length
-        if value[0] == 0xffffffffffffffff:
+        if value[0] == (2 ** (8 * item_size) - 1):
             self._parsed_values[name] = None
         else:
             value = value[0]
@@ -798,7 +798,7 @@ def compose_time_delta(self, value):
 class ComposerBinary(ComposerBase):
     byte_order = attr.ib(default=ByteOrder.NETWORK, validator=attr.validators.in_(ByteOrder))
 
-    def compose_timestamp(self, value, milliseconds=False):
+    def compose_timestamp(self, value, milliseconds=False, item_size=8):
         if value is None:
             timestamp = 0xffffffffffffffff
         else:
@@ -808,7 +808,7 @@ def compose_timestamp(self, value, milliseconds=False):
                 timestamp *= 1000
                 timestamp += value.microsecond // 1000
 
-        return self._compose_numeric_array([timestamp, ], 8)
+        return self._compose_numeric_array([timestamp, ], item_size)
 
     def _compose_numeric_array(self, values, item_size):
         composed_bytes = bytearray()

cryptoparser/dnsrec/record.py

@@ -2,6 +2,7 @@
 
 import abc
 import collections
+import datetime
 import enum
 
 import attr
@@ -17,9 +18,9 @@
     PublicKeyParamsRsa,
 )
 
-from cryptodatahub.dnssec.algorithm import DnsSecAlgorithm, DnsSecDigestType
+from cryptodatahub.dnssec.algorithm import DnsRrType, DnsSecAlgorithm, DnsSecDigestType
 
-from cryptoparser.common.base import OneByteEnumParsable, Serializable
+from cryptoparser.common.base import OneByteEnumParsable, Serializable, TwoByteEnumParsable
 from cryptoparser.common.exception import NotEnoughData
 from cryptoparser.common.parse import ByteOrder, ComposerBinary, ParsableBase, ParserBinary
 
@@ -303,3 +304,119 @@ def compose(self):
         composer.compose_raw(self.digest)
 
         return composer.composed_bytes
+
+
+class DnsRrTypeFactory(TwoByteEnumParsable):
+    @classmethod
+    def get_enum_class(cls):
+        return DnsRrType
+
+    @abc.abstractmethod
+    def compose(self):
+        raise NotImplementedError()
+
+
+@attr.s
+class DnsNameUncompressed(ParsableBase, Serializable):
+    labels = attr.ib(
+        validator=attr.validators.deep_iterable(member_validator=attr.validators.instance_of(six.string_types))
+    )
+
+    def __str__(self):
+        return six.u('.').join(self.labels)
+
+    def _as_markdown(self, level):
+        return self._markdown_result(str(self), level)
+
+    @classmethod
+    def convert(cls, value):
+        if isinstance(value, cls):
+            return value
+        if isinstance(value, six.string_types):
+            if not value:
+                return cls([])
+
+            return cls(value.split('.'))
+
+        raise InvalidValue(value, cls, 'labels')
+
+    @classmethod
+    def _parse(cls, parsable):
+        parser = ParserBinary(parsable)
+
+        labels = []
+        while True:
+            parser.parse_string('label', 1, encoding='idna')
+            label = parser['label']
+
+            if not label:
+                break
+
+            labels.append(label)
+
+        return cls(labels), parser.parsed_length
+
+    def compose(self):
+        composer = ComposerBinary()
+
+        for label in self.labels:
+            composer.compose_string(label, 'idna', 1)
+
+        composer.compose_numeric(0, 1)
+
+        return composer.composed_bytes
+
+
+@attr.s
+class DnsRecordRrsig(ParsableBase):  # pylint: disable=too-many-instance-attributes
+    HEADER_SIZE = 24
+
+    type_covered = attr.ib(validator=attr.validators.instance_of(DnsRrType))
+    algorithm = attr.ib(validator=attr.validators.instance_of(DnsSecAlgorithm))
+    labels = attr.ib(validator=attr.validators.instance_of(six.integer_types))
+    original_ttl = attr.ib(
+        validator=attr.validators.instance_of(six.integer_types),
+        metadata={'human_readable_name': 'Original TTL'}
+    )
+    signature_expiration = attr.ib(validator=attr.validators.instance_of(datetime.datetime))
+    signature_inception = attr.ib(validator=attr.validators.instance_of(datetime.datetime))
+    key_tag = attr.ib(validator=attr.validators.instance_of(six.integer_types))
+    signers_name = attr.ib(
+        converter=DnsNameUncompressed.convert,
+        validator=attr.validators.instance_of(DnsNameUncompressed)
+    )
+    signature = 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_parsable('type_covered', DnsRrTypeFactory)
+        parser.parse_parsable('algorithm', DnsSecAlgorithmFactory)
+        parser.parse_numeric('labels', 1)
+        parser.parse_numeric('original_ttl', 4)
+        parser.parse_timestamp('signature_expiration', item_size=4)
+        parser.parse_timestamp('signature_inception', item_size=4)
+        parser.parse_numeric('key_tag', 2)
+        parser.parse_parsable('signers_name', DnsNameUncompressed)
+        parser.parse_raw('signature', parser.unparsed_length)
+
+        return cls(**parser), parser.parsed_length
+
+    def compose(self):
+        composer = ComposerBinary()
+
+        composer.compose_numeric_enum_coded(self.type_covered)
+        composer.compose_numeric_enum_coded(self.algorithm)
+        composer.compose_numeric(self.labels, 1)
+        composer.compose_numeric(self.original_ttl, 4)
+        composer.compose_timestamp(self.signature_expiration, item_size=4)
+        composer.compose_timestamp(self.signature_inception, item_size=4)
+        composer.compose_numeric(self.key_tag, 2)
+        composer.compose_parsable(self.signers_name)
+        composer.compose_raw(self.signature)
+
+        return composer.composed_bytes

test/common/test_parse.py

@@ -400,21 +400,40 @@ def test_parse_timestamp(self):
         parser.parse_timestamp('timestamp')
         self.assertEqual(parser['timestamp'], datetime.datetime.fromtimestamp(0, dateutil.tz.UTC))
 
+        parser = ParserBinary(b'\x00\x00\x00\x00')
+        parser.parse_timestamp('timestamp', item_size=4)
+        self.assertEqual(parser['timestamp'], datetime.datetime.fromtimestamp(0, dateutil.tz.UTC))
+
         parser = ParserBinary(b'\x00\x00\x00\x00\x00\x00\x00\xff')
         parser.parse_timestamp('timestamp', milliseconds=True)
         self.assertEqual(
             parser['timestamp'],
             datetime.datetime.fromtimestamp(0, dateutil.tz.UTC) + datetime.timedelta(microseconds=255000)
         )
 
+        parser = ParserBinary(b'\x00\x00\x00\xff')
+        parser.parse_timestamp('timestamp', milliseconds=True, item_size=4)
+        self.assertEqual(
+            parser['timestamp'],
+            datetime.datetime.fromtimestamp(0, dateutil.tz.UTC) + datetime.timedelta(microseconds=255000)
+        )
+
         parser = ParserBinary(b'\xff\xff\xff\xff\xff\xff\xff\xff')
         parser.parse_timestamp('timestamp')
         self.assertEqual(parser['timestamp'], None)
 
+        parser = ParserBinary(b'\xff\xff\xff\xff')
+        parser.parse_timestamp('timestamp', item_size=4)
+        self.assertEqual(parser['timestamp'], None)
+
         parser = ParserBinary(b'\x00\x00\x00\x00\xff\xff\xff\xff')
         parser.parse_timestamp('timestamp')
         self.assertEqual(parser['timestamp'], datetime.datetime.fromtimestamp(0xffffffff, dateutil.tz.UTC))
 
+        parser = ParserBinary(b'\x00\x00\xff\xff')
+        parser.parse_timestamp('timestamp', item_size=4)
+        self.assertEqual(parser['timestamp'], datetime.datetime.fromtimestamp(0x0000ffff, dateutil.tz.UTC))
+
 
 class TestParserText(TestParsableBase):
     def test_error(self):

test/dnsrec/test_record.py

@@ -3,8 +3,11 @@
 
 import base64
 import collections
+import datetime
 import unittest
 
+import dateutil
+
 from cryptodatahub.common.algorithm import Authentication, KeyExchange, NamedGroup
 from cryptodatahub.common.exception import InvalidValue
 from cryptodatahub.common.key import (
@@ -15,10 +18,17 @@
     PublicKeyParamsRsa,
 )
 
-from cryptodatahub.dnssec.algorithm import DnsSecAlgorithm, DnsSecDigestType
+from cryptodatahub.dnssec.algorithm import DnsSecAlgorithm, DnsSecDigestType, DnsRrType
 
 from cryptoparser.common.exception import NotEnoughData
-from cryptoparser.dnsrec.record import DnsRecordDnskey, DnsRecordDs, DnsSecFlag, DnsSecProtocol
+from cryptoparser.dnsrec.record import (
+    DnsNameUncompressed,
+    DnsRecordDnskey,
+    DnsRecordDs,
+    DnsRecordRrsig,
+    DnsSecFlag,
+    DnsSecProtocol,
+)
 
 
 class TestDnsRecordDnskey(unittest.TestCase):
@@ -362,3 +372,90 @@ def test_parse(self):
 
     def test_compose(self):
         self.assertEqual(self.record.compose(), self.record_bytes)
+
+
+class TestDnsNameUncompressed(unittest.TestCase):
+    def setUp(self):
+        self.label_empty_bytes = b'\x00'
+        self.label_empty_name = DnsNameUncompressed([])
+        self.label_single_bytes = b'\x01a\x00'
+        self.label_single_name = DnsNameUncompressed(['a'])
+        self.label_multiple_bytes = b'\x01a\x02bb\x03ccc\x00'
+        self.label_multiple_name = DnsNameUncompressed(['a', 'bb', 'ccc'])
+
+    def test_error_convert_invalid_value(self):
+        with self.assertRaises(InvalidValue) as context_manager:
+            DnsNameUncompressed.convert(None)
+
+        self.assertEqual(context_manager.exception.value, None)
+
+    def test_parse(self):
+        self.assertEqual(DnsNameUncompressed.parse_exact_size(self.label_empty_bytes), self.label_empty_name)
+        self.assertEqual(DnsNameUncompressed.parse_exact_size(self.label_single_bytes), self.label_single_name)
+        self.assertEqual(DnsNameUncompressed.parse_exact_size(self.label_multiple_bytes), self.label_multiple_name)
+
+    def test_compose(self):
+        self.assertEqual(self.label_empty_name.compose(), self.label_empty_bytes)
+        self.assertEqual(self.label_single_name.compose(), self.label_single_bytes)
+        self.assertEqual(self.label_multiple_name.compose(), self.label_multiple_bytes)
+
+    def test_convert(self):
+        self.assertEqual(DnsNameUncompressed.convert(self.label_empty_name), self.label_empty_name)
+        self.assertEqual(DnsNameUncompressed.convert(self.label_single_name), self.label_single_name)
+        self.assertEqual(DnsNameUncompressed.convert(self.label_multiple_name), self.label_multiple_name)
+
+        self.assertEqual(DnsNameUncompressed.convert(''), self.label_empty_name)
+        self.assertEqual(DnsNameUncompressed.convert('a'), self.label_single_name)
+        self.assertEqual(DnsNameUncompressed.convert('a.bb.ccc'), self.label_multiple_name)
+
+    def test_str(self):
+        self.assertEqual(str(self.label_empty_name), '')
+        self.assertEqual(str(self.label_single_name), 'a')
+        self.assertEqual(str(self.label_multiple_name), 'a.bb.ccc')
+
+    def test_as_markdown(self):
+        self.assertEqual(self.label_empty_name.as_markdown(), '')
+        self.assertEqual(self.label_single_name.as_markdown(), 'a')
+        self.assertEqual(self.label_multiple_name.as_markdown(), 'a.bb.ccc')
+
+
+class TestDnsRecordRrsig(unittest.TestCase):
+    def setUp(self):
+        self.record_bytes = bytes(
+            b'\x00\x01' +          # type_covered: A
+            b'\x01' +              # algorithm: RSAMD5
+            b'\x03' +              # labels
+            b'\x00\x00\x0e\x10' +  # original_ttl: 3600
+            b'\x00\x00\x00\x01' +  # signature_expiration
+            b'\x00\x00\x00\x02' +  # signature_inception
+            b'\xab\xcd' +          # key_tag
+            b'\x06signer\x00' +    # signers_name
+            32 * b'\xff' +         # signature
+            b''
+        )
+        self.record = DnsRecordRrsig(
+            type_covered=DnsRrType.A,
+            algorithm=DnsSecAlgorithm.RSAMD5,
+            labels=3,
+            original_ttl=3600,
+            signature_expiration=datetime.datetime(1970, 1, 1, 0, 0, 1, tzinfo=dateutil.tz.UTC),
+            signature_inception=datetime.datetime(1970, 1, 1, 0, 0, 2, tzinfo=dateutil.tz.UTC),
+            key_tag=0xabcd,
+            signers_name='signer',
+            signature=32 * b'\xff',
+        )
+
+    def test_error_not_enough_data(self):
+        with self.assertRaises(NotEnoughData) as context_manager:
+            DnsRecordRrsig.parse_exact_size(b'\x00')
+
+        self.assertEqual(
+            context_manager.exception.bytes_needed,
+            DnsRecordRrsig.HEADER_SIZE - 1
+        )
+
+    def test_parse(self):
+        self.assertEqual(DnsRecordRrsig.parse_exact_size(self.record_bytes), self.record)
+
+    def test_compose(self):
+        self.assertEqual(self.record.compose(), self.record_bytes)