Extract Certificate Transparency SCTs from stapled OCSP responses

net/tools/testserver/testserver.py

 #!/usr/bin/env python
 # Copyright 2013 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """This is a simple HTTP/FTP/TCP/UDP/BASIC_AUTH_PROXY/WEBSOCKET server used for
 testing Chrome.
 
 It supports several test URLs, as specified by the handlers in TestPageHandler.
 By default, it listens on an ephemeral port and sends the port number back to
@@ skipping 110 matching lines @@
 
 class HTTPSServer(tlslite.api.TLSSocketServerMixIn,
                   testserver_base.ClientRestrictingServerMixIn,
                   testserver_base.BrokenPipeHandlerMixIn,
                   testserver_base.StoppableHTTPServer):
   """This is a specialization of StoppableHTTPServer that add https support and
   client verification."""
 
   def __init__(self, server_address, request_hander_class, pem_cert_and_key,
                ssl_client_auth, ssl_client_cas, ssl_bulk_ciphers,
-               record_resume_info, tls_intolerant, signed_cert_timestamps):
+               record_resume_info, tls_intolerant,
+               signed_cert_timestamps, ocsp_response):
     self.cert_chain = tlslite.api.X509CertChain().parseChain(pem_cert_and_key)
     # Force using only python implementation - otherwise behavior is different
     # depending on whether m2crypto Python module is present (error is thrown
     # when it is). m2crypto uses a C (based on OpenSSL) implementation under
     # the hood.
     self.private_key = tlslite.api.parsePEMKey(pem_cert_and_key,
                                                private=True,
                                                implementations=['python'])
     self.ssl_client_auth = ssl_client_auth
     self.ssl_client_cas = []
     self.tls_intolerant = tls_intolerant
     self.signed_cert_timestamps = signed_cert_timestamps
+    self.ocsp_response = ocsp_response
 
     for ca_file in ssl_client_cas:
       s = open(ca_file).read()
       x509 = tlslite.api.X509()
       x509.parse(s)
       self.ssl_client_cas.append(x509.subject)
     self.ssl_handshake_settings = tlslite.api.HandshakeSettings()
     if ssl_bulk_ciphers is not None:
       self.ssl_handshake_settings.cipherNames = ssl_bulk_ciphers
 
@@ skipping 13 matching lines @@
     try:
       self.tlsConnection = tlsConnection
       tlsConnection.handshakeServer(certChain=self.cert_chain,
                                     privateKey=self.private_key,
                                     sessionCache=self.session_cache,
                                     reqCert=self.ssl_client_auth,
                                     settings=self.ssl_handshake_settings,
                                     reqCAs=self.ssl_client_cas,
                                     tlsIntolerant=self.tls_intolerant,
                                     signedCertTimestamps=
-                                    self.signed_cert_timestamps)
+                                    self.signed_cert_timestamps,
+                                    ocspResponse = self.ocsp_response)
       tlsConnection.ignoreAbruptClose = True
       return True
     except tlslite.api.TLSAbruptCloseError:
       # Ignore abrupt close.
       return True
     except tlslite.api.TLSError, error:
       print "Handshake failure:", str(error)
       return False
 
 
@@ skipping 1738 matching lines @@
               ocsp_state = ocsp_state,
               serial = self.options.cert_serial)
 
           self.__ocsp_server.ocsp_response = ocsp_der
 
         for ca_cert in self.options.ssl_client_ca:
           if not os.path.isfile(ca_cert):
             raise testserver_base.OptionError(
                 'specified trusted client CA file not found: ' + ca_cert +
                 ' exiting...')
+
+        stapled_ocsp_response = None
+        if self.__ocsp_server and self.options.staple_ocsp_response:
+          stapled_ocsp_response = self.__ocsp_server.ocsp_response
+
         server = HTTPSServer((host, port), TestPageHandler, pem_cert_and_key,
                              self.options.ssl_client_auth,
                              self.options.ssl_client_ca,
                              self.options.ssl_bulk_cipher,
                              self.options.record_resume,
                              self.options.tls_intolerant,
-                             self.options.signed_cert_timestamps.decode(
-                                 "base64"))
+                             self.options.signed_cert_timestamps_tls_ext.decode(
+                                 "base64"),
+                             stapled_ocsp_response)
         print 'HTTPS server started on %s:%d...' % (host, server.server_port)
       else:
         server = HTTPServer((host, port), TestPageHandler)
         print 'HTTP server started on %s:%d...' % (host, server.server_port)
 
       server.data_dir = self.__make_data_dir()
       server.file_root_url = self.options.file_root_url
       server_data['port'] = server.server_port
     elif self.options.server_type == SERVER_WEBSOCKET:
       # Launch pywebsocket via WebSocketServer.
@@ skipping 117 matching lines @@
                                   help='If non-zero then the generated '
                                   'certificate will have this serial number')
     self.option_parser.add_option('--tls-intolerant', dest='tls_intolerant',
                                   default='0', type='int',
                                   help='If nonzero, certain TLS connections '
                                   'will be aborted in order to test version '
                                   'fallback. 1 means all TLS versions will be '
                                   'aborted. 2 means TLS 1.1 or higher will be '
                                   'aborted. 3 means TLS 1.2 or higher will be '
                                   'aborted.')
-    self.option_parser.add_option('--signed-cert-timestamps',
-                                  dest='signed_cert_timestamps',
+    self.option_parser.add_option('--signed-cert-timestamps-tls-ext',
+                                  dest='signed_cert_timestamps_tls_ext',
                                   default='',
                                   help='Base64 encoded SCT list. If set, '
                                   'server will respond with a '
                                   'signed_certificate_timestamp TLS extension '
                                   'whenever the client supports it.')
+    self.option_parser.add_option('--staple-ocsp-response',
+                                  dest='staple_ocsp_response',
+                                  default=False, action='store_true',
+                                  help='If set, server will staple the OCSP '
+                                  'response whenever OCSP is on and the client '
+                                  'supports OCSP stapling.')
     self.option_parser.add_option('--https-record-resume',
                                   dest='record_resume', const=True,
                                   default=False, action='store_const',
                                   help='Record resumption cache events rather '
                                   'than resuming as normal. Allows the use of '
                                   'the /ssl-session-cache request')
     self.option_parser.add_option('--ssl-client-auth', action='store_true',
                                   help='Require SSL client auth on every '
                                   'connection.')
     self.option_parser.add_option('--ssl-client-ca', action='append',
@@ skipping 11 matching lines @@
                                   '"aes128", "3des", "rc4". If omitted, all '
                                   'algorithms will be used. This option may '
                                   'appear multiple times, indicating '
                                   'multiple algorithms should be enabled.');
     self.option_parser.add_option('--file-root-url', default='/files/',
                                   help='Specify a root URL for files served.')
 
 
 if __name__ == '__main__':
   sys.exit(ServerRunner().main())