Add TLS-SRP (RFC 5054) support

net/tools/testserver/testserver.py

 #!/usr/bin/python2.4
 # Copyright (c) 2011 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 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
@@ skipping 55 matching lines @@
     self.stop = False
     self.nonce_time = None
     while not self.stop:
       self.handle_request()
     self.socket.close()
 
 class HTTPSServer(tlslite.api.TLSSocketServerMixIn, StoppableHTTPServer):
   """This is a specialization of StoppableHTTPerver that add https support."""
 
   def __init__(self, server_address, request_hander_class, cert_path,
-               ssl_client_auth, ssl_client_cas, ssl_bulk_ciphers):
+               ssl_client_auth, ssl_client_cas, ssl_bulk_ciphers,
+               use_tls_srp, only_tls_srp):
     s = open(cert_path).read()
     x509 = tlslite.api.X509()
     x509.parse(s)
     self.cert_chain = tlslite.api.X509CertChain([x509])
     s = open(cert_path).read()
     self.private_key = tlslite.api.parsePEMKey(s, 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
+    self.only_tls_srp = only_tls_srp
+    self.srp_verifier_db = None
+    if use_tls_srp:
+      # Make dummy SRP verifier database
+      self.srp_verifier_db = tlslite.api.VerifierDB()
+      self.srp_verifier_db.create()
+      entry = tlslite.api.VerifierDB.makeVerifier('user', 'secret', 1536)
+      self.srp_verifier_db['user'] = entry
 
     self.session_cache = tlslite.api.SessionCache()
     StoppableHTTPServer.__init__(self, server_address, request_hander_class)
 
   def handshake(self, tlsConnection):
     """Creates the SSL connection."""
     try:
-      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)
+      if not self.only_tls_srp:
+        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,
+                                      verifierDB=self.srp_verifier_db)
+      else:
+        tlsConnection.handshakeServer(verifierDB=self.srp_verifier_db)
       tlsConnection.ignoreAbruptClose = True
+      self.tlsConnection = tlsConnection
       return True
     except tlslite.api.TLSAbruptCloseError:
       # Ignore abrupt close.
       return True
     except tlslite.api.TLSError, error:
       print "Handshake failure:", str(error)
       return False
 
 
 class SyncHTTPServer(StoppableHTTPServer):
@@ skipping 87 matching lines @@
                          asyncore.dispatcher.handle_write_event)
 
       for fd in exceptional_fds:
         HandleXmppSocket(fd, self._xmpp_socket_map,
                          asyncore.dispatcher.handle_expt_event)
 
 
 class BasePageHandler(BaseHTTPServer.BaseHTTPRequestHandler):
 
   def __init__(self, request, client_address, socket_server,
-               connect_handlers, get_handlers, post_handlers, put_handlers):
+               connect_handlers, get_handlers, get_with_socket_handlers,
+               post_handlers, put_handlers):
     self._connect_handlers = connect_handlers
     self._get_handlers = get_handlers
+    self._get_with_socket_handlers = get_with_socket_handlers
     self._post_handlers = post_handlers
     self._put_handlers = put_handlers
+    self._socket_server = socket_server
     BaseHTTPServer.BaseHTTPRequestHandler.__init__(
       self, request, client_address, socket_server)
 
   def log_request(self, *args, **kwargs):
     # Disable request logging to declutter test log output.
     pass
 
   def _ShouldHandleRequest(self, handler_name):
     """Determines if the path can be handled by the handler.
 
     We consider a handler valid if the path begins with the
     handler name. It can optionally be followed by "?*", "/*".
     """
 
     pattern = re.compile('%s($|\?|/).*' % handler_name)
     return pattern.match(self.path)
 
   def do_CONNECT(self):
     for handler in self._connect_handlers:
       if handler():
         return
 
   def do_GET(self):
+    for handler in self._get_with_socket_handlers:
+      if handler(self._socket_server):
+        return
     for handler in self._get_handlers:
       if handler():
         return
 
   def do_POST(self):
     for handler in self._post_handlers:
       if handler():
         return
 
   def do_PUT(self):
@@ skipping 33 matching lines @@
       self.RealBZ2FileWithCommonHeaderHandler,
       self.SetCookieHandler,
       self.AuthBasicHandler,
       self.AuthDigestHandler,
       self.SlowServerHandler,
       self.ContentTypeHandler,
       self.ServerRedirectHandler,
       self.ClientRedirectHandler,
       self.MultipartHandler,
       self.DefaultResponseHandler]
+    get_with_socket_handlers = [
+      self.TLSLoginInfoHandler]
     post_handlers = [
       self.EchoTitleHandler,
       self.EchoAllHandler,
       self.EchoHandler,
       self.DeviceManagementHandler] + get_handlers
     put_handlers = [
       self.EchoTitleHandler,
       self.EchoAllHandler,
       self.EchoHandler] + get_handlers
 
     self._mime_types = {
       'crx' : 'application/x-chrome-extension',
       'exe' : 'application/octet-stream',
       'gif': 'image/gif',
       'jpeg' : 'image/jpeg',
       'jpg' : 'image/jpeg',
       'pdf' : 'application/pdf',
       'xml' : 'text/xml'
     }
     self._default_mime_type = 'text/html'
 
     BasePageHandler.__init__(self, request, client_address, socket_server,
-                             connect_handlers, get_handlers, post_handlers,
+                             connect_handlers, get_handlers,
+                             get_with_socket_handlers, post_handlers,
                              put_handlers)
 
   def GetMIMETypeFromName(self, file_name):
     """Returns the mime type for the specified file_name. So far it only looks
     at the file extension."""
 
     (shortname, extension) = os.path.splitext(file_name.split("?")[0])
     if len(extension) == 0:
       # no extension.
       return self._default_mime_type
@@ skipping 278 matching lines @@
     while True:
       line = self.rfile.readline()
       length = int(line, 16)
       if length == 0:
         self.rfile.readline()
         break
       body += self.rfile.read(length)
       self.rfile.read(2)
     return body
 
+  def TLSLoginInfoHandler(self, socket_server):
+    """This handler echoes back the username used to log in via TLS."""
+
+    if not self._ShouldHandleRequest("/tlslogininfo"):
+      return False
+
+    self.send_response(200)
+    self.send_header('Content-type', 'text/html')
+    self.end_headers()
+
+    srp_user = socket_server.tlsConnection.session.srpUsername
+    if srp_user:
+      self.wfile.write('username: ' + srp_user)
+    else:
+      self.wfile.write('not using TLS-SRP')
+    return True
+
   def EchoHandler(self):
     """This handler just echoes back the payload of the request, for testing
     form submission."""
 
     if not self._ShouldHandleRequest("/echo"):
       return False
 
     self.send_response(200)
     self.send_header('Content-type', 'text/html')
     self.end_headers()
@@ skipping 780 matching lines @@
         print 'specified server cert file not found: ' + options.cert + \
               ' exiting...'
         return
       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(('127.0.0.1', port), TestPageHandler, options.cert,
                            options.ssl_client_auth, options.ssl_client_ca,
-                           options.ssl_bulk_cipher)
+                           options.ssl_bulk_cipher, options.use_tls_srp,
+                           options.only_tls_srp)
       print 'HTTPS server started on port %d...' % server.server_port
     else:
       server = StoppableHTTPServer(('127.0.0.1', port), TestPageHandler)
       print 'HTTP server started on port %d...' % server.server_port
 
     server.data_dir = MakeDataDir()
     server.file_root_url = options.file_root_url
     server_data['port'] = server.server_port
     server._device_management_handler = None
     server.policy_cert_chain = options.policy_cert_chain
@@ skipping 88 matching lines @@
                            'specified file. This option may appear multiple '
                            'times, indicating multiple CA names should be '
                            'sent in the request.')
   option_parser.add_option('', '--ssl-bulk-cipher', action='append',
                            help='Specify the bulk encryption algorithm(s)'
                            'that will be accepted by the SSL server. Valid '
                            'values are "aes256", "aes128", "3des", "rc4". If '
                            'omitted, all algorithms will be used. This '
                            'option may appear multiple times, indicating '
                            'multiple algorithms should be enabled.');
+  option_parser.add_option('', '--use-tls-srp', action='store_true',
+                           help='Allow TLS authentication using TLS-SRP'
+                           ' (user jsmith, password asdf)')
+  option_parser.add_option('', '--only-tls-srp', action='store_true',
+                           help='Only allow connections using TLS-SRP')
   option_parser.add_option('', '--file-root-url', default='/files/',
                            help='Specify a root URL for files served.')
   option_parser.add_option('', '--startup-pipe', type='int',
                            dest='startup_pipe',
                            help='File handle of pipe to parent process')
   option_parser.add_option('', '--policy-cert-chain', action='append',
                            help='Specify a path to a certificate file to sign '
                                 'policy responses. This option may be used '
                                 'multiple times to define a certificate chain. '
                                 'The first element will be used for signing, '
                                 'the last element should be the root '
                                 'certificate.')
   options, args = option_parser.parse_args()
 
   sys.exit(main(options, args))