Extract Certificate Transparency SCTs from stapled OCSP responses

third_party/tlslite/tlslite/messages.py

 """Classes representing TLS messages."""
 
 from utils.compat import *
 from utils.cryptomath import *
 from errors import *
 from utils.codec import *
 from constants import *
 from X509 import X509
 from X509CertChain import X509CertChain
 
@@ skipping 114 matching lines @@
         self.ssl2 = ssl2
         self.client_version = (0,0)
         self.random = createByteArrayZeros(32)
         self.session_id = createByteArraySequence([])
         self.cipher_suites = []         # a list of 16-bit values
         self.certificate_types = [CertificateType.x509]
         self.compression_methods = []   # a list of 8-bit values
         self.srp_username = None        # a string
         self.channel_id = False
         self.support_signed_cert_timestamps = False
+        self.status_request = False
 
     def create(self, version, random, session_id, cipher_suites,
                certificate_types=None, srp_username=None):
         self.client_version = version
         self.random = random
         self.session_id = session_id
         self.cipher_suites = cipher_suites
         self.certificate_types = certificate_types
         self.compression_methods = [0]
         self.srp_username = srp_username
@@ skipping 30 matching lines @@
                     if extType == 6:
                         self.srp_username = bytesToString(p.getVarBytes(1))
                     elif extType == 7:
                         self.certificate_types = p.getVarList(1, 1)
                     elif extType == ExtensionType.channel_id:
                         self.channel_id = True
                     elif extType == ExtensionType.signed_cert_timestamps:
                         if extLength:
                             raise SyntaxError()
                         self.support_signed_cert_timestamps = True
+                    elif extType == ExtensionType.status_request:
+                        # Extension contents are currently ignored.
+                        # According to RFC 6066, this is not strictly forbidden
+                        # (although it is suboptimal):
+                        # Servers that receive a client hello containing the
+                        # "status_request" extension MAY return a suitable
+                        # certificate status response to the client along with
+                        # their certificate.  If OCSP is requested, they
+                        # SHOULD use the information contained in the extension
+                        # when selecting an OCSP responder and SHOULD include
+                        # request_extensions in the OCSP request.
+                        p.getFixBytes(extLength)
+                        self.status_request = True
                     else:
                         p.getFixBytes(extLength)
                     soFar += 4 + extLength
             p.stopLengthCheck()
         return self
 
     def write(self, trial=False):
         w = HandshakeMsg.preWrite(self, HandshakeType.client_hello, trial)
         w.add(self.client_version[0], 1)
         w.add(self.client_version[1], 1)
@@ skipping 28 matching lines @@
     def __init__(self):
         self.contentType = ContentType.handshake
         self.server_version = (0,0)
         self.random = createByteArrayZeros(32)
         self.session_id = createByteArraySequence([])
         self.cipher_suite = 0
         self.certificate_type = CertificateType.x509
         self.compression_method = 0
         self.channel_id = False
         self.signed_cert_timestamps = None
+        self.status_request = False
 
     def create(self, version, random, session_id, cipher_suite,
                certificate_type):
         self.server_version = version
         self.random = random
         self.session_id = session_id
         self.cipher_suite = cipher_suite
         self.certificate_type = certificate_type
         self.compression_method = 0
         return self
@@ skipping 32 matching lines @@
         if self.certificate_type and self.certificate_type != \
                 CertificateType.x509:
             extLength += 5
 
         if self.channel_id:
             extLength += 4
 
         if self.signed_cert_timestamps:
             extLength += 4 + len(self.signed_cert_timestamps)
 
+        if self.status_request:
+            extLength += 4
+
         if extLength != 0:
             w.add(extLength, 2)
 
         if self.certificate_type and self.certificate_type != \
                 CertificateType.x509:
             w.add(7, 2)
             w.add(1, 2)
             w.add(self.certificate_type, 1)
 
         if self.channel_id:
             w.add(ExtensionType.channel_id, 2)
             w.add(0, 2)
 
         if self.signed_cert_timestamps:
             w.add(ExtensionType.signed_cert_timestamps, 2)
             w.addVarSeq(stringToBytes(self.signed_cert_timestamps), 1, 2)
 
+        if self.status_request:
+            w.add(ExtensionType.status_request, 2)
+            w.add(0, 2)
+
         return HandshakeMsg.postWrite(self, w, trial)
 
 class Certificate(HandshakeMsg):
     def __init__(self, certificateType):
         self.certificateType = certificateType
         self.contentType = ContentType.handshake
         self.certChain = None
 
     def create(self, certChain):
         self.certChain = certChain
@@ skipping 48 matching lines @@
         elif self.certificateType == CertificateType.cryptoID:
             if self.certChain:
                 bytes = stringToBytes(self.certChain.write())
             else:
                 bytes = createByteArraySequence([])
             w.addVarSeq(bytes, 1, 2)
         else:
             raise AssertionError()
         return HandshakeMsg.postWrite(self, w, trial)
 
+class CertificateStatus(HandshakeMsg):
+    def __init__(self):
+        self.contentType = ContentType.handshake
+
+    def create(self, ocsp_response):
+        self.ocsp_response = ocsp_response
+        return self
+
+    # Defined for the sake of completeness, even though we currently only
+    # support sending the status message (server-side), not requesting
+    # or receiving it (client-side).
+    def parse(self, p):
+        p.startLengthCheck(3)
+        status_type = p.get(1)
+        # Only one type is specified, so hardwire it.
+        if status_type != CertificateStatusType.ocsp:
+            raise SyntaxError()
+        ocsp_response = p.getVarBytes(3)
+        if not ocsp_response:
+            # Can't be empty
+            raise SyntaxError()
+        self.ocsp_response = ocsp_response
+        return self
+
+    def write(self, trial=False):
+        w = HandshakeMsg.preWrite(self, HandshakeType.certificate_status,
+                                  trial)
+        w.add(CertificateStatusType.ocsp, 1)
+        w.addVarSeq(stringToBytes(self.ocsp_response), 1, 3)
+        return HandshakeMsg.postWrite(self, w, trial)
+
 class CertificateRequest(HandshakeMsg):
     def __init__(self):
         self.contentType = ContentType.handshake
         #Apple's Secure Transport library rejects empty certificate_types, so
         #default to rsa_sign.
         self.certificate_types = [ClientCertificateType.rsa_sign]
         self.certificate_authorities = []
 
     def create(self, certificate_types, certificate_authorities):
         self.certificate_types = certificate_types
@@ skipping 241 matching lines @@
     def create(self, bytes):
         self.bytes = bytes
         return self
 
     def parse(self, p):
         self.bytes = p.bytes
         return self
 
     def write(self):
         return self.bytes