net: add OCSP tests.

net/tools/testserver/testserver.py

Index: net/tools/testserver/testserver.py
diff --git a/net/tools/testserver/testserver.py b/net/tools/testserver/testserver.py
index 28405ffc17e871cabafb5d47ddf1c23c34af7745..f694bf718cbe969f71b15b22fb0bfa2e905ed0ca 100755
--- a/net/tools/testserver/testserver.py
+++ b/net/tools/testserver/testserver.py
@@ -19,15 +19,17 @@ import BaseHTTPServer
 import cgi
 import errno
 import httplib
+import minica
 import optparse
 import os
 import random
 import re
 import select
-import SocketServer
 import socket
-import sys
+import SocketServer
 import struct
+import sys
+import threading
 import time
 import urllib
 import urlparse
@@ -105,25 +107,35 @@ class StoppableHTTPServer(BaseHTTPServer.HTTPServer):
 
 
 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:
@@ -1876,6 +1888,21 @@ def MakeDataDir():
 
   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.
@@ -1956,19 +1983,53 @@ def main(options, args):
   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)
@@ -2048,10 +2109,15 @@ def main(options, args):
     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__':