net: add OCSP tests.

net/tools/testserver/testserver.py

 #!/usr/bin/env python
 # Copyright (c) 2012 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/SYNC/TCP/UDP/ 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
 the originating process over a pipe. The originating process can specify an
 explicit port if necessary.
 It can use https if you specify the flag --https=CERT where CERT is the path
 to a pem file containing the certificate and private key that should be used.
 """
 
 import asyncore
 import base64
 import BaseHTTPServer
 import cgi
 import errno
 import httplib
+import minica
 import optparse
 import os
 import random
 import re
 import select
+import socket
 import SocketServer
-import socket
+import struct
 import sys
-import struct
+import threading
 import time
 import urllib
 import urlparse
 import warnings
 import zlib
 
 # Ignore deprecation warnings, they make our output more cluttered.
 warnings.filterwarnings("ignore", category=DeprecationWarning)
 
 import echo_message
@@ skipping 57 matching lines @@
 
   def serve_forever(self):
     self.stop = False
     self.nonce_time = None
     while not self.stop:
       self.handle_request()
     self.socket.close()
 
 
 class HTTPServer(ClientRestrictingServerMixIn, StoppableHTTPServer):
-  """This is a specialization of StoppableHTTPerver that adds client
+  """This is a specialization of StoppableHTTPServer that adds client
   verification."""
 
   pass
 
+class OCSPServer(ClientRestrictingServerMixIn, BaseHTTPServer.HTTPServer):
+  """This is a specialization of HTTPServer that serves an
+  OCSP response"""
+
+  def serve_forever_on_thread(self):
+    self.thread = threading.Thread(target = self.serve_forever,
+                                   name = "OCSPServerThread")
+    self.thread.start()
+
+  def stop_serving(self):
+    self.shutdown()
+    self.thread.join()
 
 class HTTPSServer(tlslite.api.TLSSocketServerMixIn,
                   ClientRestrictingServerMixIn,
                   StoppableHTTPServer):
-  """This is a specialization of StoppableHTTPerver that add https support and
+  """This is a specialization of StoppableHTTPServer that add https support and
   client verification."""
 
-  def __init__(self, server_address, request_hander_class, cert_path,
+  def __init__(self, server_address, request_hander_class, pem_cert_and_key,
                ssl_client_auth, ssl_client_cas, ssl_bulk_ciphers,
                record_resume_info):
-    s = open(cert_path).read()
-    self.cert_chain = tlslite.api.X509CertChain().parseChain(s)
-    s = open(cert_path).read()
-    self.private_key = tlslite.api.parsePEMKey(s, private=True)
+    self.cert_chain = tlslite.api.X509CertChain().parseChain(pem_cert_and_key)
+    self.private_key = tlslite.api.parsePEMKey(pem_cert_and_key, private=True)
     self.ssl_client_auth = ssl_client_auth
     self.ssl_client_cas = []
     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 1732 matching lines @@
     # Create the default path to our data dir, relative to the exe dir.
     my_data_dir = os.path.dirname(sys.argv[0])
     my_data_dir = os.path.join(my_data_dir, "..", "..", "..", "..",
                                "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
 
+class OCSPHandler(BasePageHandler):
+  def __init__(self, request, client_address, socket_server):
+    handlers = [self.OCSPResponse]
+    self.ocsp_response = socket_server.ocsp_response
+    BasePageHandler.__init__(self, request, client_address, socket_server,
+                             handlers, handlers, handlers,
+                             handlers, handlers)
+
+  def OCSPResponse(self):
+    self.send_response(200)
+    self.send_header('Content-Type', 'application/ocsp-response')
+    self.send_header('Content-Length', str(len(self.ocsp_response)))
+    self.end_headers()
+
+    self.wfile.write(self.ocsp_response)
 
 class TCPEchoHandler(SocketServer.BaseRequestHandler):
   """The RequestHandler class for TCP echo server.
 
   It is instantiated once per connection to the server, and overrides the
   handle() method to implement communication to the client.
   """
 
   def handle(self):
     """Handles the request from the client and constructs a response."""
@@ skipping 60 matching lines @@
     sys.stdout = FileMultiplexer(sys.stdout, logfile)
   else:
     sys.stdout = logfile
 
   port = options.port
   host = options.host
 
   server_data = {}
   server_data['host'] = host
 
+  ocsp_server = None
+
   if options.server_type == SERVER_HTTP:
     if options.cert:
-      # let's make sure the cert file exists.
-      if not os.path.isfile(options.cert):
-        print 'specified server cert file not found: ' + options.cert + \
-              ' exiting...'
-        return
+      pem_cert_and_key = None
+      if options.cert.startswith('__'):
+        ocsp_server = OCSPServer((host, 0), OCSPHandler)
+        print 'OCSP server on port ' + str(ocsp_server.server_port)
+
+        ocsp_der = None
+        ocsp_revoked = False
+        ocsp_invalid = False
+
+        if options.cert == '__ocsp_ok__':
+          pass
+        elif options.cert == '__ocsp_revoked__':
+          ocsp_revoked = True
+        elif options.cert == '__ocsp_invalid__':
+          ocsp_invalid = True
+        else:
+          print 'unknown special certificate type: ' + options.cert
+          return
+
+        (pem_cert_and_key, ocsp_der) = \
+            minica.GenerateCertKeyAndOCSP(
+                subject = "127.0.0.1",
+                ocsp_url = "http://127.0.0.1:%d/ocsp" % ocsp_server.server_port,
+                ocsp_revoked = ocsp_revoked)
+
+        if ocsp_invalid:
+          ocsp_der = '3'
+
+        ocsp_server.ocsp_response = ocsp_der
+      else:
+        # options.cert is a filename. Let's make sure the cert file exists.
+        if not os.path.isfile(options.cert):
+          print 'specified server cert file not found: ' + options.cert + \
+                ' exiting...'
+          return
+        pem_cert_and_key = file(options.cert, 'r').read()
+
       for ca_cert in options.ssl_client_ca:
         if not os.path.isfile(ca_cert):
           print 'specified trusted client CA file not found: ' + ca_cert + \
                 ' exiting...'
           return
-      server = HTTPSServer((host, port), TestPageHandler, options.cert,
+      server = HTTPSServer((host, port), TestPageHandler, pem_cert_and_key,
                            options.ssl_client_auth, options.ssl_client_ca,
                            options.ssl_bulk_cipher, options.record_resume)
       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 = MakeDataDir()
     server.file_root_url = options.file_root_url
     server_data['port'] = server.server_port
@@ skipping 59 matching lines @@
     else:
       fd = options.startup_pipe
     startup_pipe = os.fdopen(fd, "w")
     # First write the data length as an unsigned 4-byte value.  This
     # is _not_ using network byte ordering since the other end of the
     # pipe is on the same machine.
     startup_pipe.write(struct.pack('=L', server_data_len))
     startup_pipe.write(server_data_json)
     startup_pipe.close()
 
+  if ocsp_server is not None:
+    ocsp_server.serve_forever_on_thread()
+
   try:
     server.serve_forever()
   except KeyboardInterrupt:
     print 'shutting down server'
+    if ocsp_server is not None:
+      ocsp_server.stop_serving()
     server.stop = True
 
 if __name__ == '__main__':
   option_parser = optparse.OptionParser()
   option_parser.add_option("-f", '--ftp', action='store_const',
                            const=SERVER_FTP, default=SERVER_HTTP,
                            dest='server_type',
                            help='start up an FTP server.')
   option_parser.add_option('', '--sync', action='store_const',
                            const=SERVER_SYNC, default=SERVER_HTTP,
@@ skipping 66 matching lines @@
                            dest='host',
                            help='Hostname or IP upon which the server will '
                            'listen. Client connections will also only be '
                            'allowed from this address.')
   option_parser.add_option('', '--auth-token', dest='auth_token',
                            help='Specify the auth token which should be used'
                            'in the authorization header for GData.')
   options, args = option_parser.parse_args()
 
   sys.exit(main(options, args))