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)

[wtc, 2013/12/03 21:04:25]

Nit: should "OCSPResponse" start in lowercase "ocspResponse"?

[ekasper, 2013/12/04 19:25:15]

Done.

       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 1691 matching lines @@
                                  "test", "data")
 
       #TODO(ibrar): Must use Find* funtion defined in google\tools
       #i.e my_data_dir = FindUpward(my_data_dir, "test", "data")
 
     return my_data_dir
 
   def create_server(self, server_data):
     port = self.options.port
     host = self.options.host
+    ocsp_der = None
 
     if self.options.server_type == SERVER_HTTP:
       if self.options.https:
         pem_cert_and_key = None
         if self.options.cert_and_key_file:
           if not os.path.isfile(self.options.cert_and_key_file):
             raise testserver_base.OptionError(
                 'specified server cert file not found: ' +
                 self.options.cert_and_key_file + ' exiting...')
           pem_cert_and_key = file(self.options.cert_and_key_file, 'r').read()
         else:
           # generate a new certificate and run an OCSP server for it.
           self.__ocsp_server = OCSPServer((host, 0), OCSPHandler)
           print ('OCSP server started on %s:%d...' %
               (host, self.__ocsp_server.server_port))
 
-          ocsp_der = None
           ocsp_state = None
 
           if self.options.ocsp == 'ok':
             ocsp_state = minica.OCSP_STATE_GOOD
           elif self.options.ocsp == 'revoked':
             ocsp_state = minica.OCSP_STATE_REVOKED
           elif self.options.ocsp == 'invalid':
             ocsp_state = minica.OCSP_STATE_INVALID
           elif self.options.ocsp == 'unauthorized':
             ocsp_state = minica.OCSP_STATE_UNAUTHORIZED
           elif self.options.ocsp == 'unknown':
             ocsp_state = minica.OCSP_STATE_UNKNOWN
           else:
             raise testserver_base.OptionError('unknown OCSP status: ' +
                 self.options.ocsp_status)
 
           (pem_cert_and_key, ocsp_der) = minica.GenerateCertKeyAndOCSP(
               subject = "127.0.0.1",
               ocsp_url = ("http://%s:%d/ocsp" %
                   (host, self.__ocsp_server.server_port)),
               ocsp_state = ocsp_state,
-              serial = self.options.cert_serial)
+              serial = self.options.cert_serial,
+              # Signed Certificate Timestamps are only accepted in a stapled
+              # response: when included in a non-stapled response, the client
+              # will simply ignore the extension.

[wtc, 2013/12/03 21:04:25]

Is this comment true? If I get an OCSP response from an OCSP responder, should I
ignore the SCT extension in the OCSP response?

[ekasper, 2013/12/04 19:25:15]

I've reverted the extension code here as the testing is now broken up between
receiving a stapled response and decoding it.

But yes, this comment is true. CT must be side-channel free so we only accept
stapled responses.

+              sct_extension = (
+                  self.options.signed_cert_timestamps_ocsp.decode("base64")))
 
           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...')
+
         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"),
+                             ocsp_der)

[wtc, 2013/12/03 21:04:25]

Should we use self.__ocsp_server.ocsp_response here instead? This will make it
unnecessary to move the "ocsp_der = None" line to the beginning of this
function.

[ekasper, 2013/12/04 19:25:15]

Changed.

         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('--signed-cert-timestamps-ocsp',
+                                  dest='signed_cert_timestamps_ocsp',
+                                  default='',
+                                  help='Base64 encoded SCT list. If set, '
+                                  'server will include the list in a stapled '
+                                  'OCSP response whenever 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())