Parse client_certificate_url extension and support CertificateURL message

tls/_common/enums.py

@@ -45,6 +45,8 @@ class HandshakeType(Enum):
     CERTIFICATE_VERIFY = 15
     CLIENT_KEY_EXCHANGE = 16
     FINISHED = 20
+    CERTIFICATE_URL = 21
+    CERTIFICATE_STATUS = 22
 
 
 class ContentType(Enum):
@@ -85,6 +87,10 @@ class AlertDescription(Enum):
     USER_CANCELED = 90
     NO_RENEGOTIATION = 100
     UNSUPPORTED_EXTENSION = 110
+    CERTIFICATE_UNOBTAINABLE = 111
+    UNRECOGNIZED_NAME = 112
+    BAD_CERTIFICATE_STATUS_RESPONSE = 113
+    BAD_CERTIFICATE_HASH_VALUE = 114
 
 
 class ExtensionType(Enum):
@@ -128,3 +134,8 @@ class CompressionMethod(Enum):
 
 class NameType(Enum):
     HOST_NAME = 0
+
+
+class CertChainType(Enum):
+    INDIVIDUAL_CERTS = 0
+    PKIPATH = 1

tls/_constructs.py

@@ -84,6 +84,11 @@
 
 ServerNameList = TLSPrefixedArray("server_name_list", ServerName)
 
+ClientCertificateURL = Struct(
+    "client_certificate_url",
+    # The "extension_data" field of this extension SHALL be empty.
+)
+
 SignatureAndHashAlgorithm = Struct(
     "algorithms",
     EnumClass(UBInt8("hash"), enums.HashAlgorithm),
@@ -107,6 +112,9 @@
             enums.ExtensionType.SIGNATURE_ALGORITHMS: Opaque(
                 SupportedSignatureAlgorithms
             ),
+            enums.ExtensionType.CLIENT_CERTIFICATE_URL: Opaque(
+                ClientCertificateURL
+            ),
         },
         default=Pass,
     )
@@ -195,3 +203,18 @@
     UBInt8("level"),
     UBInt8("description"),
 )
+
+URLAndHash = Struct(
+    "url_and_hash",
+    SizeWithin(UBInt16("length"),
+               min_size=1, max_size=2 ** 16 - 1),
+    Bytes("url", lambda ctx: ctx.length),
+    UBInt8("padding"),
+    Bytes("sha1_hash", 20),
+)
+
+CertificateURL = Struct(
+    "CertificateURL",
+    EnumClass(UBInt8("type"), enums.CertChainType),
+    TLSPrefixedArray("url_and_hash_list", URLAndHash),
+)

tls/exceptions.py

@@ -11,3 +11,10 @@ class UnsupportedCipherException(Exception):
 
 class UnsupportedExtensionException(Exception):
     pass
+
+
+class InvalidPaddingException(Exception):
+    # XXX: Is leaking this info in CertificateURL (i.e. revealing the specific
+    # byte that is invalid) a good idea? Need to check -- could be a
+    # vulnerability in disguise.
+    pass

tls/message.py

@@ -14,6 +14,8 @@
 
 from tls._common import enums
 
+from tls.exceptions import InvalidPaddingException
+
 from tls.hello_message import ClientHello, ProtocolVersion, ServerHello
 
 
@@ -204,6 +206,73 @@ def from_bytes(cls, bytes):
         )
 
 
+@attr.s
+class URLAndHash(object):
+    """
+    An object representing a URLAndHash struct.
+    """
+    url = attr.ib()
+    padding = attr.ib()
+    sha1_hash = attr.ib()
+
+    # XXX: The "padding" byte MUST be 0x01.  It is
+    # present to make the structure backwards
+    # compatible.
+    # Should the validation for padding go here?
+
+
+@attr.s
+class CertificateURL(object):
+    """
+    An object representing a CertificateURL struct.
+    """
+    type = attr.ib()
+    url_and_hash_list = attr.ib()
+
+    def as_bytes(self):
+        for url_and_hash in self.url_and_hash_list:
+            if url_and_hash.padding != 1:
+                raise InvalidPaddingException
+        return _constructs.CertificateURL.build(Container(
+            type=self.type,
+            url_and_hash_list=ListContainer(
+                Container(
+                    length=len(url_and_hash.url),
+                    url=url_and_hash.url,
+                    padding=url_and_hash.padding,
+                    sha1_hash=url_and_hash.sha1_hash,
+                )
+                for url_and_hash in self.url_and_hash_list
+            )
+        ))
+
+    @classmethod
+    def from_bytes(cls, bytes):
+        """
+        Parse a ``CertificateURL`` struct.
+
+        :param bytes: the bytes representing the input.
+        :return: CertificateURL object.
+        """
+        construct = _constructs.CertificateURL.parse(bytes)
+        url_and_hash_list = []
+        for url_and_hash in construct.url_and_hash_list:
+            # TODO: Figure how attrs does validation and move this assert to
+            # URLAndHash
+            if url_and_hash.padding != 1:
+                raise InvalidPaddingException
+            url_and_hash_list.append(URLAndHash(
+                url=url_and_hash.url,
+                padding=url_and_hash.padding,
+                sha1_hash=url_and_hash.sha1_hash,
+            ))
+
+        return cls(
+            type=construct.type,
+            url_and_hash_list=url_and_hash_list,
+        )
+
+
 @attr.s
 class Finished(object):
     verify_data = attr.ib()
@@ -230,7 +299,8 @@ def as_bytes(self):
             enums.HandshakeType.CERTIFICATE_REQUEST,
             enums.HandshakeType.HELLO_REQUEST,
             enums.HandshakeType.SERVER_HELLO_DONE,
-            enums.HandshakeType.FINISHED
+            enums.HandshakeType.FINISHED,
+            enums.HandshakeType.CERTIFICATE_URL,
         ]:
             _body_as_bytes = self.body.as_bytes()
         else:
@@ -271,6 +341,8 @@ def _get_handshake_message(msg_type, body):
                 CertificateRequest.from_bytes,
             #    15: parse_certificate_verify,
             #    16: parse_client_key_exchange,
+            enums.HandshakeType.CERTIFICATE_URL: CertificateURL.from_bytes,
+            #    22: parse certificate_status,
         }
 
         try:
@@ -282,7 +354,8 @@ def _get_handshake_message(msg_type, body):
                 return Finished(verify_data=body)
             elif msg_type in [enums.HandshakeType.SERVER_KEY_EXCHANGE,
                               enums.HandshakeType.CERTIFICATE_VERIFY,
-                              enums.HandshakeType.CLIENT_KEY_EXCHANGE]:
+                              enums.HandshakeType.CLIENT_KEY_EXCHANGE,
+                              enums.HandshakeType.CERTIFICATE_STATUS]:
                 raise NotImplementedError
             else:
                 return _handshake_message_parser[msg_type](body)

tls/test/test_hello_message.py

@@ -104,6 +104,17 @@ class TestClientHello(object):
         b'\x00\x12'
     ) + server_name_extension_data
 
+    client_certificate_url_extension = (
+        b'\x00\x02'  # Extension Type: Server Certificate Type
+        b'\x00\x00'  # Length
+        b''  # Data
+    )
+
+    client_hello_packet_with_client_certificate_url_extension = (
+        common_client_hello_data +
+        b'\x00\x04'
+    ) + client_certificate_url_extension
+
     def test_resumption_no_extensions(self):
         """
         :func:`parse_client_hello` returns an instance of
@@ -219,6 +230,30 @@ def test_hello_from_bytes_with_unsupported_extension(self):
                 client_hello_packet
             )
 
+    def test_parse_client_certificate_url_extension(self):
+        """
+        :py:func:`tls.hello_message.ClientHello` parses a packet with
+        CLIENT_CERTIFICATE_URL extension.
+        """
+        record = ClientHello.from_bytes(
+            self.client_hello_packet_with_client_certificate_url_extension
+        )
+        assert len(record.extensions) == 1
+        assert (record.extensions[0].type ==
+                enums.ExtensionType.CLIENT_CERTIFICATE_URL)
+        assert record.extensions[0].data == {}
+
+    def test_as_bytes_client_certificate_url_extension(self):
+        """
+        :py:func:`tls.hello_message.ClientHello` serializes a message
+        containing the CLIENT_CERTIFICATE_URL extension.
+        """
+        record = ClientHello.from_bytes(
+            self.client_hello_packet_with_client_certificate_url_extension
+        )
+        assert (record.as_bytes() ==
+                self.client_hello_packet_with_client_certificate_url_extension)
+
     def test_as_bytes_unsupported_extension(self):
         """
         :func:`ClientHello.as_bytes` fails to serialize a message that

tls/test/test_message.py

@@ -10,12 +10,14 @@
 
 from tls._common import enums
 
+from tls.exceptions import InvalidPaddingException
+
 from tls.hello_message import ClientHello, ProtocolVersion, ServerHello
 
 from tls.message import (ASN1Cert, Certificate, CertificateRequest,
-                         Finished, Handshake, HelloRequest,
-                         PreMasterSecret, ServerDHParams,
-                         ServerHelloDone)
+                         CertificateURL, Finished, Handshake, HelloRequest,
+                         PreMasterSecret, ServerDHParams, ServerHelloDone,
+                         URLAndHash)
 
 
 class TestCertificateRequestParsing(object):
@@ -283,6 +285,73 @@ def test_as_bytes_too_long(self):
             certificate.as_bytes()
 
 
+class TestCertificateURLParsing(object):
+    """
+    Tests for parsing of :py:class:`tls.message.CertificateURL` messages.
+    """
+    url_and_hash_list_bytes = (
+        b'\x00\x10'  # url length
+        b'cert.example.com'  # url
+        b'\x01'  # padding
+        b'abcdefghijklmnopqrst'  # SHA1Hash[20]
+    )
+
+    certificate_url_packet = (
+        b'\x00'  # CertChainType
+        b'\x00\x27'  # url_and_hash_list length
+    ) + url_and_hash_list_bytes
+
+    def test_parse_certificate_url(self):
+        """
+        :py:meth:`tls.message.CertificateURL.from_bytes` parses a valid
+        packet.
+        """
+        record = CertificateURL.from_bytes(self.certificate_url_packet)
+        assert isinstance(record, CertificateURL)
+        assert record.type == enums.CertChainType.INDIVIDUAL_CERTS
+        assert len(record.url_and_hash_list) == 1
+        assert record.url_and_hash_list[0].url == b'cert.example.com'
+        assert record.url_and_hash_list[0].padding == 1
+        assert record.url_and_hash_list[0].sha1_hash == b'abcdefghijklmnopqrst'
+
+    def test_incorrect_padding(self):
+        """
+        :py:meth:`tls.message.CertificateURL.from_bytes` rejects a packet
+        whose ``padding`` is not 1.
+        """
+        bad_padding_bytes = (
+            b'\x00\x10'  # url length
+            b'cert.example.com'  # url
+            b'\x03'  # padding
+            b'abcdefghijklmnopqrst'  # SHA1Hash[20]
+        )
+        certificate_url_packet = (
+            b'\x00'  # CertChainType
+            b'\x00\x27'  # url_and_hash_list length
+        ) + bad_padding_bytes
+
+        with pytest.raises(InvalidPaddingException):
+            CertificateURL.from_bytes(certificate_url_packet)
+
+    def test_as_bytes(self):
+        """
+        :py:meth:`tls.message.CertificateUrl.as_bytes` returns a valid
+        packet.
+        """
+        record = CertificateURL.from_bytes(self.certificate_url_packet)
+        assert record.as_bytes() == self.certificate_url_packet
+
+    def test_as_bytes_with_bad_padding(self):
+        """
+        :py:meth:`tls.message.CertificateURL.as_bytes` fails to serialize  a
+        record whose ``padding`` is not 1.
+        """
+        record = CertificateURL.from_bytes(self.certificate_url_packet)
+        record.url_and_hash_list[0].padding = 5
+        with pytest.raises(InvalidPaddingException):
+            record.as_bytes()
+
+
 class TestHandshakeStructParsing(object):
     """
     Tests for parsing of :py:class:`tls.messages.Handshake` structs.
@@ -387,6 +456,20 @@ class TestHandshakeStructParsing(object):
         b'some-encrypted-bytes'
     )
 
+    certificate_url_packet = (
+        b'\x00'  # CertChainType
+        b'\x00\x27'  # url_and_hash_list length
+        b'\x00\x10'  # url length
+        b'cert.example.com'  # url
+        b'\x01'  # padding
+        b'abcdefghijklmnopqrst'  # SHA1Hash[20]
+    )
+
+    certificate_url_handshake_packet = (
+        b'\x15'
+        b'\x00\x00\x2a'
+    ) + certificate_url_packet
+
     def test_parse_client_hello_in_handshake(self):
         record = Handshake.from_bytes(self.client_hello_handshake_packet)
         assert isinstance(record, Handshake)
@@ -477,3 +560,29 @@ def test_as_bytes_server_hello_done(self):
     def test_as_bytes_finished(self):
         record = Handshake.from_bytes(self.finished_handshake)
         assert record.as_bytes() == self.finished_handshake
+
+    def test_from_bytes_certificate_url(self):
+        """
+        :py:class:`tls.messages.Handshake` parses a valid packet with a
+        ``CertificateURL`` message.
+        """
+        record = Handshake.from_bytes(self.certificate_url_handshake_packet)
+        assert isinstance(record, Handshake)
+        assert record.msg_type == enums.HandshakeType.CERTIFICATE_URL
+        assert record.length == 42
+        assert isinstance(record.body, CertificateURL)
+        assert record.body.type == enums.CertChainType.INDIVIDUAL_CERTS
+        assert len(record.body.url_and_hash_list) == 1
+        assert isinstance(record.body.url_and_hash_list[0], URLAndHash)
+        assert record.body.url_and_hash_list[0].url == b'cert.example.com'
+        assert record.body.url_and_hash_list[0].padding == 1
+        assert (record.body.url_and_hash_list[0].sha1_hash ==
+                b'abcdefghijklmnopqrst')
+
+    def test_as_bytes_certificate_url(self):
+        """
+        :py:meth:`tls.message.Handshake.as_bytes` returns a valid packet when
+        the body contains a ``CertificateURL`` message.
+        """
+        record = Handshake.from_bytes(self.certificate_url_handshake_packet)
+        assert record.as_bytes() == self.certificate_url_handshake_packet