Extract Certificate Transparency SCTs from stapled OCSP responses

third_party/tlslite/tlslite/TLSConnection.py

 """
 MAIN CLASS FOR TLS LITE (START HERE!).
 """
 from __future__ import generators
 
 import socket
 from utils.compat import formatExceptionTrace
 from TLSRecordLayer import TLSRecordLayer
 from Session import Session
 from constants import *
@@ skipping 919 matching lines @@
             #Mark the connection as open
             self.session._setResumable(True)
             self._handshakeDone(resumed=False)
 
 
 
     def handshakeServer(self, sharedKeyDB=None, verifierDB=None,
                         certChain=None, privateKey=None, reqCert=False,
                         sessionCache=None, settings=None, checker=None,
                         reqCAs=None, tlsIntolerant=0,
-                        signedCertTimestamps=None,
-                        fallbackSCSV=False):
+                        signedCertTimestamps=None, fallbackSCSV=False,
+                        ocspResponse=None):
         """Perform a handshake in the role of server.
 
         This function performs an SSL or TLS handshake.  Depending on
         the arguments and the behavior of the client, this function can
         perform a shared-key, SRP, or certificate-based handshake.  It
         can also perform a combined SRP and server-certificate
         handshake.
 
         Like any handshake function, this can be called on a closed
         TLS connection, or on a TLS connection that is already open.
@@ skipping 55 matching lines @@
         @type reqCAs: list of L{array.array} of unsigned bytes
         @param reqCAs: A collection of DER-encoded DistinguishedNames that
         will be sent along with a certificate request. This does not affect
         verification.
 
         @type signedCertTimestamps: str
         @param signedCertTimestamps: A SignedCertificateTimestampList (as a
         binary 8-bit string) that will be sent as a TLS extension whenever
         the client announces support for the extension.
 
+        @type ocspResponse: str
+        @param ocspResponse: An OCSP response (as a binary 8-bit string) that
+        will be sent stapled in the handshake whenever the client announces
+        support for the status_request extension.
+        Note that the response is sent independent of the ClientHello
+        status_request extension contents, and is thus only meant for testing
+        environments. Real OCSP stapling is more complicated as it requires
+        choosing a suitable response based on the ClientHello status_request
+        extension contents.
+
         @raise socket.error: If a socket error occurs.
         @raise tlslite.errors.TLSAbruptCloseError: If the socket is closed
         without a preceding alert.
         @raise tlslite.errors.TLSAlert: If a TLS alert is signalled.
         @raise tlslite.errors.TLSAuthenticationError: If the checker
         doesn't like the other party's authentication credentials.
         """
         for result in self.handshakeServerAsync(sharedKeyDB, verifierDB,
                 certChain, privateKey, reqCert, sessionCache, settings,
                 checker, reqCAs, tlsIntolerant, signedCertTimestamps,
-                fallbackSCSV):
+                fallbackSCSV, ocspResponse):
             pass
 
 
     def handshakeServerAsync(self, sharedKeyDB=None, verifierDB=None,
                              certChain=None, privateKey=None, reqCert=False,
                              sessionCache=None, settings=None, checker=None,
                              reqCAs=None, tlsIntolerant=0,
                              signedCertTimestamps=None,
-                             fallbackSCSV=False):
+                             fallbackSCSV=False, ocspResponse=None):
         """Start a server handshake operation on the TLS connection.
 
         This function returns a generator which behaves similarly to
         handshakeServer().  Successive invocations of the generator
         will return 0 if it is waiting to read from the socket, 1 if it is
         waiting to write to the socket, or it will raise StopIteration
         if the handshake operation is complete.
 
         @rtype: iterable
         @return: A generator; see above for details.
         """
         handshaker = self._handshakeServerAsyncHelper(\
             sharedKeyDB=sharedKeyDB,
             verifierDB=verifierDB, certChain=certChain,
             privateKey=privateKey, reqCert=reqCert,
             sessionCache=sessionCache, settings=settings,
             reqCAs=reqCAs,
             tlsIntolerant=tlsIntolerant,
             signedCertTimestamps=signedCertTimestamps,
-            fallbackSCSV=fallbackSCSV)
+            fallbackSCSV=fallbackSCSV, ocspResponse=ocspResponse)
+
         for result in self._handshakeWrapperAsync(handshaker, checker):
             yield result
 
 
     def _handshakeServerAsyncHelper(self, sharedKeyDB, verifierDB,
                                     certChain, privateKey, reqCert,
                                     sessionCache, settings, reqCAs,
                                     tlsIntolerant, signedCertTimestamps,
-                                    fallbackSCSV):
+                                    fallbackSCSV, ocspResponse):
 
         self._handshakeStart(client=False)
 
         if (not sharedKeyDB) and (not verifierDB) and (not certChain):
             raise ValueError("Caller passed no authentication credentials")
         if certChain and not privateKey:
             raise ValueError("Caller passed a certChain but no privateKey")
         if privateKey and not certChain:
             raise ValueError("Caller passed a privateKey but no certChain")
         if reqCAs and not reqCert:
@@ skipping 356 matching lines @@
         elif cipherSuite in CipherSuite.rsaSuites:
 
             #Send ServerHello, Certificate[, CertificateRequest],
             #ServerHelloDone
             msgs = []
             serverHello = ServerHello().create(
                     self.version, serverRandom,
                     sessionID, cipherSuite, certificateType)
             serverHello.channel_id = clientHello.channel_id
             if clientHello.support_signed_cert_timestamps:
-                serverHello.signed_cert_timestamps = signedCertTimestamps
+              serverHello.signed_cert_timestamps = signedCertTimestamps
+            serverHello.status_request = (clientHello.status_request and
+                                          ocspResponse)
             doingChannelID = clientHello.channel_id
             msgs.append(serverHello)
             msgs.append(Certificate(certificateType).create(serverCertChain))
+            if serverHello.status_request:
+                msgs.append(CertificateStatus().create(ocspResponse))
             if reqCert and reqCAs:
                 msgs.append(CertificateRequest().create([], reqCAs))
             elif reqCert:
                 msgs.append(CertificateRequest())
             msgs.append(ServerHelloDone())
             for result in self._sendMsgs(msgs):
                 yield result
 
             #From here on, the client's messages must have the right version
             self._versionCheck = True
@@ skipping 205 matching lines @@
         if len(publicKey) < settings.minKeySize:
             for result in self._sendError(AlertDescription.handshake_failure,
                     "Other party's public key too small: %d" % len(publicKey)):
                 yield result
         if len(publicKey) > settings.maxKeySize:
             for result in self._sendError(AlertDescription.handshake_failure,
                     "Other party's public key too large: %d" % len(publicKey)):
                 yield result
 
         yield publicKey, certChain