diff options
Diffstat (limited to 'chromium/net/tools/testserver/testserver.py')
| -rwxr-xr-x | chromium/net/tools/testserver/testserver.py | 142 |
1 files changed, 97 insertions, 45 deletions
diff --git a/chromium/net/tools/testserver/testserver.py b/chromium/net/tools/testserver/testserver.py index 0a1f59b0c59..33faf463d3a 100755 --- a/chromium/net/tools/testserver/testserver.py +++ b/chromium/net/tools/testserver/testserver.py @@ -27,6 +27,7 @@ import re import select import socket import SocketServer +import ssl import struct import sys import threading @@ -38,20 +39,36 @@ import zlib BASE_DIR = os.path.dirname(os.path.abspath(__file__)) ROOT_DIR = os.path.dirname(os.path.dirname(os.path.dirname(BASE_DIR))) -import echo_message -import testserver_base +# Temporary hack to deal with tlslite 0.3.8 -> 0.4.6 upgrade. +# +# TODO(davidben): Remove this when it has cycled through all the bots and +# developer checkouts or when http://crbug.com/356276 is resolved. +try: + os.remove(os.path.join(ROOT_DIR, 'third_party', 'tlslite', + 'tlslite', 'utils', 'hmac.pyc')) +except Exception: + pass # Append at the end of sys.path, it's fine to use the system library. sys.path.append(os.path.join(ROOT_DIR, 'third_party', 'pyftpdlib', 'src')) -sys.path.append(os.path.join(ROOT_DIR, 'third_party', 'tlslite')) -import pyftpdlib.ftpserver -import tlslite -import tlslite.api -# Insert at the beginning of the path, we want this to be used +# Insert at the beginning of the path, we want to use our copies of the library # unconditionally. sys.path.insert(0, os.path.join(ROOT_DIR, 'third_party', 'pywebsocket', 'src')) +sys.path.insert(0, os.path.join(ROOT_DIR, 'third_party', 'tlslite')) + +import mod_pywebsocket.standalone from mod_pywebsocket.standalone import WebSocketServer +# import manually +mod_pywebsocket.standalone.ssl = ssl + +import pyftpdlib.ftpserver + +import tlslite +import tlslite.api + +import echo_message +import testserver_base SERVER_HTTP = 0 SERVER_FTP = 1 @@ -84,6 +101,7 @@ class WebSocketOptions: self.certificate = None self.tls_client_auth = False self.tls_client_ca = None + self.tls_module = 'ssl' self.use_basic_auth = False @@ -134,10 +152,12 @@ class HTTPSServer(tlslite.api.TLSSocketServerMixIn, client verification.""" def __init__(self, server_address, request_hander_class, pem_cert_and_key, - ssl_client_auth, ssl_client_cas, ssl_bulk_ciphers, + ssl_client_auth, ssl_client_cas, ssl_client_cert_types, + ssl_bulk_ciphers, ssl_key_exchanges, enable_npn, record_resume_info, tls_intolerant, signed_cert_timestamps, fallback_scsv_enabled, ocsp_response): - self.cert_chain = tlslite.api.X509CertChain().parseChain(pem_cert_and_key) + self.cert_chain = tlslite.api.X509CertChain() + self.cert_chain.parsePemList(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 @@ -147,19 +167,38 @@ class HTTPSServer(tlslite.api.TLSSocketServerMixIn, implementations=['python']) self.ssl_client_auth = ssl_client_auth self.ssl_client_cas = [] - self.tls_intolerant = tls_intolerant + self.ssl_client_cert_types = [] + if enable_npn: + self.next_protos = ['http/1.1'] + else: + self.next_protos = None + if tls_intolerant == 0: + self.tls_intolerant = None + else: + self.tls_intolerant = (3, tls_intolerant) self.signed_cert_timestamps = signed_cert_timestamps self.fallback_scsv_enabled = fallback_scsv_enabled 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) + if ssl_client_auth: + 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) + + for cert_type in ssl_client_cert_types: + self.ssl_client_cert_types.append({ + "rsa_sign": tlslite.api.ClientCertificateType.rsa_sign, + "dss_sign": tlslite.api.ClientCertificateType.dss_sign, + "ecdsa_sign": tlslite.api.ClientCertificateType.ecdsa_sign, + }[cert_type]) + self.ssl_handshake_settings = tlslite.api.HandshakeSettings() if ssl_bulk_ciphers is not None: self.ssl_handshake_settings.cipherNames = ssl_bulk_ciphers + if ssl_key_exchanges is not None: + self.ssl_handshake_settings.keyExchangeNames = ssl_key_exchanges if record_resume_info: # If record_resume_info is true then we'll replace the session cache with @@ -182,6 +221,8 @@ class HTTPSServer(tlslite.api.TLSSocketServerMixIn, reqCert=self.ssl_client_auth, settings=self.ssl_handshake_settings, reqCAs=self.ssl_client_cas, + reqCertTypes=self.ssl_client_cert_types, + nextProtos=self.next_protos, tlsIntolerant=self.tls_intolerant, signedCertTimestamps= self.signed_cert_timestamps, @@ -288,7 +329,6 @@ class TestPageHandler(testserver_base.BasePageHandler): self.NoContentHandler, self.ServerRedirectHandler, self.ClientRedirectHandler, - self.MultipartHandler, self.GetSSLSessionCacheHandler, self.SSLManySmallRecords, self.GetChannelID, @@ -1398,29 +1438,6 @@ class TestPageHandler(testserver_base.BasePageHandler): return True - def MultipartHandler(self): - """Send a multipart response (10 text/html pages).""" - - test_name = '/multipart' - if not self._ShouldHandleRequest(test_name): - return False - - num_frames = 10 - bound = '12345' - self.send_response(200) - self.send_header('Content-Type', - 'multipart/x-mixed-replace;boundary=' + bound) - self.end_headers() - - for i in xrange(num_frames): - self.wfile.write('--' + bound + '\r\n') - self.wfile.write('Content-Type: text/html\r\n\r\n') - self.wfile.write('<title>page ' + str(i) + '</title>') - self.wfile.write('page ' + str(i)) - - self.wfile.write('--' + bound + '--') - return True - def GetSSLSessionCacheHandler(self): """Send a reply containing a log of the session cache operations.""" @@ -1431,11 +1448,14 @@ class TestPageHandler(testserver_base.BasePageHandler): self.send_header('Content-Type', 'text/plain') self.end_headers() try: - for (action, sessionID) in self.server.session_cache.log: - self.wfile.write('%s\t%s\n' % (action, sessionID.encode('hex'))) + log = self.server.session_cache.log except AttributeError: self.wfile.write('Pass --https-record-resume in order to use' + ' this request') + return True + + for (action, sessionID) in log: + self.wfile.write('%s\t%s\n' % (action, bytes(sessionID).encode('hex'))) return True def SSLManySmallRecords(self): @@ -1465,7 +1485,7 @@ class TestPageHandler(testserver_base.BasePageHandler): self.send_response(200) self.send_header('Content-Type', 'text/plain') self.end_headers() - channel_id = self.server.tlsConnection.channel_id.tostring() + channel_id = bytes(self.server.tlsConnection.channel_id) self.wfile.write(hashlib.sha256(channel_id).digest().encode('base64')) return True @@ -1956,17 +1976,22 @@ class ServerRunner(testserver_base.TestServerRunner): server = HTTPSServer((host, port), TestPageHandler, pem_cert_and_key, self.options.ssl_client_auth, self.options.ssl_client_ca, + self.options.ssl_client_cert_type, self.options.ssl_bulk_cipher, + self.options.ssl_key_exchange, + self.options.enable_npn, self.options.record_resume, self.options.tls_intolerant, self.options.signed_cert_timestamps_tls_ext.decode( "base64"), self.options.fallback_scsv, stapled_ocsp_response) - print 'HTTPS server started on %s:%d...' % (host, server.server_port) + print 'HTTPS server started on https://%s:%d...' % \ + (host, server.server_port) else: server = HTTPServer((host, port), TestPageHandler) - print 'HTTP server started on %s:%d...' % (host, server.server_port) + print 'HTTP server started on http://%s:%d...' % \ + (host, server.server_port) server.data_dir = self.__make_data_dir() server.file_root_url = self.options.file_root_url @@ -1979,7 +2004,9 @@ class ServerRunner(testserver_base.TestServerRunner): # is required to work correctly. It should be fixed from pywebsocket side. os.chdir(self.__make_data_dir()) websocket_options = WebSocketOptions(host, port, '.') + scheme = "ws" if self.options.cert_and_key_file: + scheme = "wss" websocket_options.use_tls = True websocket_options.private_key = self.options.cert_and_key_file websocket_options.certificate = self.options.cert_and_key_file @@ -1994,7 +2021,8 @@ class ServerRunner(testserver_base.TestServerRunner): self.options.ssl_client_ca[0] + ' exiting...') websocket_options.tls_client_ca = self.options.ssl_client_ca[0] server = WebSocketServer(websocket_options) - print 'WebSocket server started on %s:%d...' % (host, server.server_port) + print 'WebSocket server started on %s://%s:%d...' % \ + (scheme, host, server.server_port) server_data['port'] = server.server_port elif self.options.server_type == SERVER_TCP_ECHO: # Used for generating the key (randomly) that encodes the "echo request" @@ -2137,6 +2165,15 @@ class ServerRunner(testserver_base.TestServerRunner): 'file. This option may appear multiple ' 'times, indicating multiple CA names should ' 'be sent in the request.') + self.option_parser.add_option('--ssl-client-cert-type', action='append', + default=[], help='Specify that the client ' + 'certificate request should include the ' + 'specified certificate_type value. This ' + 'option may appear multiple times, ' + 'indicating multiple values should be send ' + 'in the request. Valid values are ' + '"rsa_sign", "dss_sign", and "ecdsa_sign". ' + 'If omitted, "rsa_sign" will be used.') self.option_parser.add_option('--ssl-bulk-cipher', action='append', help='Specify the bulk encryption ' 'algorithm(s) that will be accepted by the ' @@ -2145,6 +2182,21 @@ class ServerRunner(testserver_base.TestServerRunner): 'algorithms will be used. This option may ' 'appear multiple times, indicating ' 'multiple algorithms should be enabled.'); + self.option_parser.add_option('--ssl-key-exchange', action='append', + help='Specify the key exchange algorithm(s)' + 'that will be accepted by the SSL server. ' + 'Valid values are "rsa", "dhe_rsa". If ' + 'omitted, all algorithms will be used. This ' + 'option may appear multiple times, ' + 'indicating multiple algorithms should be ' + 'enabled.'); + # TODO(davidben): Add ALPN support to tlslite. + self.option_parser.add_option('--enable-npn', dest='enable_npn', + default=False, const=True, + action='store_const', + help='Enable server support for the NPN ' + 'extension. The server will advertise ' + 'support for exactly one protocol, http/1.1') self.option_parser.add_option('--file-root-url', default='/files/', help='Specify a root URL for files served.') |