testserver.py TLS and client auth support on WebSocket.

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
@@ skipping 75 matching lines @@
     self.port = port
     self.websock_handlers = data_dir
     self.scan_dir = None
     self.allow_handlers_outside_root_dir = False
     self.websock_handlers_map_file = None
     self.cgi_directories = []
     self.is_executable_method = None
     self.allow_draft75 = False
     self.strict = True
 
-    # TODO(toyoshim): Support SSL and authenticates (http://crbug.com/137639)
     self.use_tls = False
     self.private_key = None
     self.certificate = None
+    self.tls_client_auth = False
     self.tls_client_ca = None
     self.use_basic_auth = False
 
 class RecordingSSLSessionCache(object):
   """RecordingSSLSessionCache acts as a TLS session cache and maintains a log of
   lookups and inserts in order to test session cache behaviours."""
 
   def __init__(self):
     self.log = []
 
@@ skipping 2118 matching lines @@
             minica.GenerateCertKeyAndOCSP(
                 subject = "127.0.0.1",
                 ocsp_url = ("http://%s:%d/ocsp" %
                     (host, ocsp_server.server_port)),
                 ocsp_state = ocsp_state)
 
         ocsp_server.ocsp_response = ocsp_der
 
       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...'
+          print ('specified trusted client CA file not found: ' + ca_cert +
+                 ' exiting...')
           return
       server = HTTPSServer((host, port), TestPageHandler, pem_cert_and_key,
                            options.ssl_client_auth, options.ssl_client_ca,
                            options.ssl_bulk_cipher, options.record_resume,
                            options.tls_intolerant)
       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
     server._device_management_handler = None
     server.policy_keys = options.policy_keys
     server.policy_user = options.policy_user
     server.gdata_auth_token = options.auth_token
   elif options.server_type == SERVER_WEBSOCKET:
     # Launch pywebsocket via WebSocketServer.
     logger = logging.getLogger()
     logger.addHandler(logging.StreamHandler())
     # TODO(toyoshim): Remove following os.chdir. Currently this operation
     # is required to work correctly. It should be fixed from pywebsocket side.
     os.chdir(MakeDataDir())
-    server = WebSocketServer(WebSocketOptions(host, port, '.'))
+    websocket_options = WebSocketOptions(host, port, '.')
+    if options.cert_and_key_file:
+      websocket_options.use_tls = True
+      websocket_options.private_key = options.cert_and_key_file
+      websocket_options.certificate = options.cert_and_key_file
+    if options.ssl_client_auth:
+      websocket_options.tls_client_auth = True
+      if len(options.ssl_client_ca) != 1:
+        # TODO(toyoshim): Provide non-zero exit code for these error cases.
+        # Ditto on other paths here and there.
+        # http://crbug.com/156539
+        print 'one trusted client CA file should be specified'
+        return
+      if not os.path.isfile(options.ssl_client_ca[0]):
+        print ('specified trusted client CA file not found: ' +
+               options.ssl_client_ca[0] + ' exiting...')
+        return
+      websocket_options.tls_client_ca = options.ssl_client_ca[0]
+    server = WebSocketServer(websocket_options)
     print 'WebSocket server started on %s:%d...' % (host, server.server_port)
     server_data['port'] = server.server_port
   elif options.server_type == SERVER_SYNC:
     xmpp_port = options.xmpp_port
     server = SyncHTTPServer((host, port), xmpp_port, SyncPageHandler)
     print 'Sync HTTP server started on port %d...' % server.server_port
     print 'Sync XMPP server started on port %d...' % server.xmpp_port
     server_data['port'] = server.server_port
     server_data['xmpp_port'] = server.xmpp_port
   elif options.server_type == SERVER_TCP_ECHO:
@@ skipping 172 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))