Commit de8efa76 authored by Szilárd Pfeiffer's avatar Szilárd Pfeiffer
Browse files

Merge branch '104-add-chrome-client-hello-extensions' 

Closes: #104
parents f05700a3 2f80cd54

cryptoparser/tls/extension.py

+93 −7
Original line number Diff line number Diff line
@@ -854,6 +854,12 @@ class TlsExtensionApplicationLayerProtocolSettings(TlsExtensionApplicationLayerP 
        return TlsExtensionType.APPLICATION_LAYER_PROTOCOL_SETTINGS





class TlsExtensionOldApplicationLayerProtocolSettings(TlsExtensionApplicationLayerProtocolBase):

    @classmethod

    def get_extension_type(cls):

        return TlsExtensionType.OLD_APPLICATION_LAYER_PROTOCOL_SETTINGS





class TlsExtensionNextProtocolNegotiationClient(TlsExtensionUnusedData):

    @classmethod

    def get_extension_type(cls):
@@ -1072,6 +1078,32 @@ class TlsExtensionRecordSizeLimit(TlsExtensionParsed): 
        return header_bytes + payload_composer.composed_bytes





@attr.s

class TlsExtensionServerPadding(TlsExtensionParsed):

    padding_size = attr.ib(validator=attr.validators.instance_of(int))



    @classmethod

    def get_extension_type(cls):

        return TlsExtensionType.SERVER_PADDING



    @classmethod

    def _parse(cls, parsable):

        parser = cls._parse_header(parsable)



        parser.parse_numeric('padding_size', 2)



        return TlsExtensionServerPadding(parser['padding_size']), parser.parsed_length



    def compose(self):

        payload_composer = ComposerBinary()



        payload_composer.compose_numeric(self.padding_size, 2)



        header_bytes = self._compose_header(payload_composer.composed_length)



        return header_bytes + payload_composer.composed_bytes





@attr.s

class TlsExtensionSignedCertificateTimestampClient(TlsExtensionUnusedData):

    @classmethod
@@ -1214,15 +1246,11 @@ class TlsExtensionEncryptedClientHelloBase(TlsExtensionParsed): 
        raise NotImplementedError()



    @classmethod

    def _parse_header(cls, parsable):

        parser = super()._parse_header(parsable)



    def _parse_client_hello_type(cls, parser):

        parser.parse_numeric('client_hello_type', 1, TlsEncryptedClientHelloType)

        if parser['client_hello_type'] != cls.get_encrypted_client_hello_type():

            raise InvalidType()



        return parser



    def compose_type(self):

        composer = ComposerBinary()
@@ -1240,6 +1268,10 @@ class TlsExtensionEncryptedClientHelloInner(TlsExtensionEncryptedClientHelloBase 
    @classmethod

    def _parse(cls, parsable):

        parser = cls._parse_header(parsable)

        parser.parse_raw('payload', parser['extension_length'])



        body_parser = ParserBinary(parser['payload'])

        cls._parse_client_hello_type(body_parser)



        return cls(), parser.parsed_length
@@ -1262,10 +1294,13 @@ class TlsExtensionEncryptedClientHelloOuter(TlsExtensionEncryptedClientHelloBase 
    @classmethod

    def _parse(cls, parsable):

        parser = cls._parse_header(parsable)

        parser.parse_raw('payload', parser['extension_length'])



        parser.parse_raw('data', parser.unparsed_length)

        body_parser = ParserBinary(parser['payload'])

        cls._parse_client_hello_type(body_parser)

        body_parser.parse_raw('data', body_parser.unparsed_length)



        return cls(parser['data']), parser.parsed_length

        return cls(body_parser['data']), parser.parsed_length



    def compose(self):

        payload_composer = self.compose_type()
@@ -1283,6 +1318,53 @@ class TlsExtensionPostHandshakeAuthentication(TlsExtensionUnusedData): 
        return TlsExtensionType.POST_HANDSHAKE_AUTH





class TlsTrustAnchorIdentifier(Opaque):

    @classmethod

    def get_param(cls):

        return OpaqueParam(

            min_byte_num=1, max_byte_num=2 ** 8 - 1

        )





class TlsTrustAnchorIdentifierList(VectorParsable):

    @classmethod

    def get_param(cls):

        return VectorParamParsable(

            item_class=TlsTrustAnchorIdentifier,

            fallback_class=None,

            min_byte_num=0, max_byte_num=2 ** 16 - 1

        )





@attr.s

class TlsExtensionTrustAnchors(TlsExtensionParsed):

    trust_anchor_identifiers = attr.ib(

        converter=TlsTrustAnchorIdentifierList,

        validator=attr.validators.instance_of(TlsTrustAnchorIdentifierList),

    )



    @classmethod

    def get_extension_type(cls):

        return TlsExtensionType.TRUST_ANCHORS



    @classmethod

    def _parse(cls, parsable):

        parser = cls._parse_header(parsable)



        parser.parse_parsable('trust_anchor_identifiers', TlsTrustAnchorIdentifierList)



        return TlsExtensionTrustAnchors(parser['trust_anchor_identifiers']), parser.parsed_length



    def compose(self):

        payload_composer = ComposerBinary()



        payload_composer.compose_parsable(self.trust_anchor_identifiers)



        header_bytes = self._compose_header(payload_composer.composed_length)



        return header_bytes + payload_composer.composed_bytes





class TlsExtensionVariantBase(VariantParsable):

    @classmethod

    @abc.abstractmethod
@@ -1310,6 +1392,8 @@ class TlsExtensionVariantClient(TlsExtensionVariantBase): 
                [TlsExtensionApplicationLayerProtocolNegotiation, ]),

            (TlsExtensionType.APPLICATION_LAYER_PROTOCOL_SETTINGS,

                [TlsExtensionApplicationLayerProtocolSettings, ]),

            (TlsExtensionType.OLD_APPLICATION_LAYER_PROTOCOL_SETTINGS,

                [TlsExtensionOldApplicationLayerProtocolSettings, ]),

            (TlsExtensionType.CHANNEL_ID, [TlsExtensionChannelId, ]),

            (TlsExtensionType.COMPRESS_CERTIFICATE, [TlsExtensionCompressCertificate, ]),

            (TlsExtensionType.ENCRYPT_THEN_MAC, [TlsExtensionEncryptThenMAC, ]),
@@ -1328,6 +1412,7 @@ class TlsExtensionVariantClient(TlsExtensionVariantBase): 
            (TlsExtensionType.POST_HANDSHAKE_AUTH, [TlsExtensionPostHandshakeAuthentication, ]),

            (TlsExtensionType.PSK_KEY_EXCHANGE_MODES, [TlsExtensionPskKeyExchangeModes, ]),

            (TlsExtensionType.RECORD_SIZE_LIMIT, [TlsExtensionRecordSizeLimit, ]),

            (TlsExtensionType.SERVER_PADDING, [TlsExtensionServerPadding, ]),

            (TlsExtensionType.SHORT_RECORD_HEADER, [TlsExtensionShortRecordHeader, ]),

            (TlsExtensionType.SIGNATURE_ALGORITHMS, [TlsExtensionSignatureAlgorithms, ]),

            (TlsExtensionType.SIGNATURE_ALGORITHMS_CERT, [TlsExtensionSignatureAlgorithmsCert, ]),
@@ -1336,6 +1421,7 @@ class TlsExtensionVariantClient(TlsExtensionVariantBase): 
            (TlsExtensionType.ENCRYPTED_CLIENT_HELLO, [TlsExtensionEncryptedClientHelloInner,

                                                       TlsExtensionEncryptedClientHelloOuter]),

            (TlsExtensionType.TOKEN_BINDING, [TlsExtensionTokenBinding, ]),

            (TlsExtensionType.TRUST_ANCHORS, [TlsExtensionTrustAnchors, ]),

        ])

cryptoparser/tls/grease.py

+6 −6
Original line number Diff line number Diff line
@@ -19,7 +19,7 @@ class TlsInvalidType(enum.IntEnum): 
    UNKNOWN = 1





@attr.s

@attr.s(frozen=True)

class TlsInvalidTypeParamsBase:

    code = attr.ib(validator=attr.validators.instance_of(int))

    value_type = attr.ib(validator=attr.validators.in_(TlsInvalidType))
@@ -41,7 +41,7 @@ class TlsInvalidTypeParamsTwoByte(TlsInvalidTypeParamsBase): 
        return 2





@attr.s

@attr.s(frozen=True)

class TlsInvalidTypeBase(ParsableBase):

    code = attr.ib(validator=attr.validators.instance_of((int, CryptoDataEnumCodedBase)))

    value = attr.ib(init=False, validator=attr.validators.instance_of(TlsInvalidTypeParamsBase))
@@ -49,17 +49,17 @@ class TlsInvalidTypeBase(ParsableBase): 
    def __attrs_post_init__(self):

        if isinstance(self.code, self.get_grease_enum()):

            value_type = TlsInvalidType.GREASE

            self.code = self.code.value.code

            object.__setattr__(self, 'code', self.code.value.code)

        elif isinstance(self.code, CryptoDataEnumCodedBase):

            value_type = TlsInvalidType.UNKNOWN

            self.code = self.code.value.code

            object.__setattr__(self, 'code', self.code.value.code)

        else:

            try:

                self.code = self.get_grease_enum().from_code(self.code).value.code

                object.__setattr__(self, 'code', self.get_grease_enum().from_code(self.code).value.code)

                value_type = TlsInvalidType.GREASE

            except InvalidValue:

                value_type = TlsInvalidType.UNKNOWN

        self.value = self.get_param_class()(self.code, value_type)

        object.__setattr__(self, 'value', self.get_param_class()(self.code, value_type))



    @classmethod

    @abc.abstractmethod

cryptodatahub @ eb9ebb7b

Compare a88f6a94 to eb9ebb7b
Original line number Diff line number Diff line 
Subproject commit a88f6a94c6cc9cee82517fc94c716cb992b65101
 
Subproject commit eb9ebb7b079b9bd8d11bd100c8c25bc788f786c7

test/tls/test_extension.py

+100 −1
Original line number Diff line number Diff line 
# SPDX-License-Identifier: MPL-2.0

# pylint: disable=too-many-lines



import collections

import datetime
@@ -43,6 +44,7 @@ from cryptoparser.tls.extension import ( 
    TlsExtensionKeyShareReservedClient,

    TlsExtensionNextProtocolNegotiationClient,

    TlsExtensionNextProtocolNegotiationServer,

    TlsExtensionOldApplicationLayerProtocolSettings,

    TlsExtensionPadding,

    TlsExtensionPostHandshakeAuthentication,

    TlsExtensionPskKeyExchangeModes,
@@ -50,6 +52,7 @@ from cryptoparser.tls.extension import ( 
    TlsExtensionRenegotiationInfo,

    TlsExtensionServerNameClient,

    TlsExtensionServerNameServer,

    TlsExtensionServerPadding,

    TlsExtensionSessionTicket,

    TlsExtensionShortRecordHeader,

    TlsExtensionSignatureAlgorithms,
@@ -59,6 +62,7 @@ from cryptoparser.tls.extension import ( 
    TlsExtensionSupportedVersionsClient,

    TlsExtensionSupportedVersionsServer,

    TlsExtensionTokenBinding,

    TlsExtensionTrustAnchors,

    TlsExtensionUnparsed,

    TlsExtensionParsed,

    TlsExtensionType,
@@ -66,6 +70,8 @@ from cryptoparser.tls.extension import ( 
    TlsProtocolNameList,

    TlsRenegotiatedConnection,

    TlsTokenBindingProtocolVersion,

    TlsTrustAnchorIdentifier,

    TlsTrustAnchorIdentifierList,

)

from cryptoparser.tls.grease import TlsGreaseOneByte, TlsGreaseTwoByte, TlsInvalidTypeOneByte, TlsInvalidTypeTwoByte

from cryptoparser.tls.version import TlsVersion, TlsProtocolVersion
@@ -772,7 +778,7 @@ class TestExtensionApplicationLayerProtocolNegotiation(unittest.TestCase): 
class TestExtensionApplicationLayerProtocolSettings(unittest.TestCase):

    def test_parse(self):

        extension_alpn_dict = collections.OrderedDict([

            ('extension_type', b'\x44\x69'),

            ('extension_type', b'\x44\xcd'),

            ('extension_length', b'\x00\x09'),

            ('protocol_name_list_length', b'\x00\x07'),

            ('protocol_name_h2_length', b'\x02'),
@@ -788,6 +794,25 @@ class TestExtensionApplicationLayerProtocolSettings(unittest.TestCase): 
        self.assertEqual(extension_alpn.compose(), extension_alpn_bytes)





class TestExtensionOldApplicationLayerProtocolSettings(unittest.TestCase):

    def test_parse(self):

        extension_alpn_dict = collections.OrderedDict([

            ('extension_type', b'\x44\x69'),

            ('extension_length', b'\x00\x09'),

            ('protocol_name_list_length', b'\x00\x07'),

            ('protocol_name_h2_length', b'\x02'),

            ('protocol_name_h2', b'h2'),

            ('protocol_name_h2c_length', b'\x03'),

            ('protocol_name_h2c', b'h2c'),

        ])

        extension_alpn_bytes = b''.join(extension_alpn_dict.values())

        extension_alpn = TlsExtensionOldApplicationLayerProtocolSettings.parse_exact_size(

            extension_alpn_bytes

        )

        self.assertEqual(extension_alpn.protocol_names, TlsProtocolNameList([TlsProtocolName.H2, TlsProtocolName.H2C]))

        self.assertEqual(extension_alpn.compose(), extension_alpn_bytes)





class TestExtensionUnusedData(unittest.TestCase):

    def test_error(self):

        extension_unused_data_dict = collections.OrderedDict([
@@ -852,6 +877,53 @@ class TestExtensionRecordSizeLimit(unittest.TestCase): 
        self.assertEqual(extension_record_size_limit.compose(), extension_record_size_limit_bytes)





class TestExtensionServerPadding(unittest.TestCase):

    def test_parse(self):

        extension_server_padding_dict = collections.OrderedDict([

            ('extension_type', b'\x12\xe0'),

            ('extension_length', b'\x00\x02'),

            ('padding_size', b'\x00\x00'),

        ])

        extension_server_padding_bytes = b''.join(extension_server_padding_dict.values())

        extension_server_padding = TlsExtensionServerPadding.parse_exact_size(extension_server_padding_bytes)

        self.assertEqual(extension_server_padding.padding_size, 0)

        self.assertEqual(extension_server_padding.compose(), extension_server_padding_bytes)





class TestExtensionTrustAnchors(unittest.TestCase):

    def test_parse(self):

        extension_trust_anchors_dict = collections.OrderedDict([

            ('extension_type', b'\xca\x34'),

            ('extension_length', b'\x00\x02'),

            ('trust_anchor_identifiers_length', b'\x00\x00'),

        ])

        extension_trust_anchors_bytes = b''.join(extension_trust_anchors_dict.values())

        extension_trust_anchors = TlsExtensionTrustAnchors.parse_exact_size(extension_trust_anchors_bytes)

        self.assertEqual(extension_trust_anchors.trust_anchor_identifiers, TlsTrustAnchorIdentifierList([]))

        self.assertEqual(extension_trust_anchors.compose(), extension_trust_anchors_bytes)



    def test_parse_non_empty_list(self):

        extension_trust_anchors_dict = collections.OrderedDict([

            ('extension_type', b'\xca\x34'),

            ('extension_length', b'\x00\x07'),

            ('trust_anchor_identifiers_length', b'\x00\x05'),

            ('trust_anchor_identifier_1_length', b'\x02'),

            ('trust_anchor_identifier_1', b'\x01\x02'),

            ('trust_anchor_identifier_2_length', b'\x01'),

            ('trust_anchor_identifier_2', b'\x03'),

        ])

        extension_trust_anchors_bytes = b''.join(extension_trust_anchors_dict.values())

        extension_trust_anchors = TlsExtensionTrustAnchors.parse_exact_size(extension_trust_anchors_bytes)

        self.assertEqual(

            extension_trust_anchors.trust_anchor_identifiers,

            TlsTrustAnchorIdentifierList([

                TlsTrustAnchorIdentifier(b'\x01\x02'),

                TlsTrustAnchorIdentifier(b'\x03'),

            ])

        )

        self.assertEqual(extension_trust_anchors.compose(), extension_trust_anchors_bytes)





class TestExtensionPadding(unittest.TestCase):

    def test_error_non_zero_padding(self):

        extension_padding_dict = collections.OrderedDict([
@@ -936,6 +1008,33 @@ class ExtensionEncryptedClientHelloOuter(unittest.TestCase): 
            extension_encrypted_client_hello_bytes

        )



    def test_parse_followed_by_another_extension(self):

        extension_encrypted_client_hello_dict = collections.OrderedDict([

            ('extension_type', b'\xfe\x0d'),

            ('extension_length', b'\x00\x11'),

            ('hello_type', b'\x00'),

            ('hello_data', b'\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f'),

        ])

        extension_encrypted_client_hello_bytes = b''.join(extension_encrypted_client_hello_dict.values())

        extension_extended_master_secret_dict = collections.OrderedDict([

            ('extension_type', b'\x00\x17'),

            ('extension_length', b'\x00\x00'),

        ])

        extension_extended_master_secret_bytes = b''.join(extension_extended_master_secret_dict.values())



        extension_encrypted_client_hello, parsed_length = TlsExtensionEncryptedClientHelloOuter.parse_immutable(

            extension_encrypted_client_hello_bytes + extension_extended_master_secret_bytes

        )

        self.assertEqual(parsed_length, len(extension_encrypted_client_hello_bytes))

        self.assertEqual(

            extension_encrypted_client_hello.data,

            extension_encrypted_client_hello_dict['hello_data']

        )

        self.assertEqual(

            extension_encrypted_client_hello.compose(),

            extension_encrypted_client_hello_bytes

        )





class TestExtensionPostHandshakeAuthentication(unittest.TestCase):

    def test_parse(self):