Implement ALPN in tlslite.

net/tools/testserver/testserver.py

 #!/usr/bin/env python
 # Copyright 2013 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/TCP/UDP/BASIC_AUTH_PROXY/WEBSOCKET 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
@@ skipping 143 matching lines @@
 
 class HTTPSServer(tlslite.api.TLSSocketServerMixIn,
                   testserver_base.ClientRestrictingServerMixIn,
                   testserver_base.BrokenPipeHandlerMixIn,
                   testserver_base.StoppableHTTPServer):
   """This is a specialization of StoppableHTTPServer that add https support and
   client verification."""
 
   def __init__(self, server_address, request_hander_class, pem_cert_and_key,
                ssl_client_auth, ssl_client_cas, ssl_client_cert_types,
-               ssl_bulk_ciphers, ssl_key_exchanges, npn_protocols,
-               record_resume_info, tls_intolerant,
+               ssl_bulk_ciphers, ssl_key_exchanges, alpn_protocols,
+               npn_protocols, record_resume_info, tls_intolerant,
                tls_intolerance_type, signed_cert_timestamps,
-               fallback_scsv_enabled, ocsp_response,
-               alert_after_handshake, disable_channel_id, disable_ems,
-               token_binding_params):
+               fallback_scsv_enabled, ocsp_response, alert_after_handshake,
+               disable_channel_id, disable_ems, token_binding_params):
     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
     # the hood.
     self.private_key = tlslite.api.parsePEMKey(pem_cert_and_key,
                                                private=True,
                                                implementations=['python'])
     self.ssl_client_auth = ssl_client_auth
     self.ssl_client_cas = []
     self.ssl_client_cert_types = []
-    self.npn_protocols = npn_protocols
     self.signed_cert_timestamps = signed_cert_timestamps
     self.fallback_scsv_enabled = fallback_scsv_enabled
     self.ocsp_response = ocsp_response
 
     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)
@@ skipping 15 matching lines @@
       self.ssl_handshake_settings.tlsIntolerant = (3, tls_intolerant)
       self.ssl_handshake_settings.tlsIntoleranceType = tls_intolerance_type
     if alert_after_handshake:
       self.ssl_handshake_settings.alertAfterHandshake = True
     if disable_channel_id:
       self.ssl_handshake_settings.enableChannelID = False
     if disable_ems:
       self.ssl_handshake_settings.enableExtendedMasterSecret = False
     self.ssl_handshake_settings.supportedTokenBindingParams = \
         token_binding_params
+    self.ssl_handshake_settings.alpnProtos=alpn_protocols;
+    self.ssl_handshake_settings.nextProtos=npn_protocols;
 
     if record_resume_info:
       # If record_resume_info is true then we'll replace the session cache with
       # an object that records the lookups and inserts that it sees.
       self.session_cache = RecordingSSLSessionCache()
     else:
       self.session_cache = tlslite.api.SessionCache()
     testserver_base.StoppableHTTPServer.__init__(self,
                                                  server_address,
                                                  request_hander_class)
 
   def handshake(self, tlsConnection):
     """Creates the SSL connection."""
 
     try:
       self.tlsConnection = tlsConnection
       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,
                                     reqCertTypes=self.ssl_client_cert_types,
-                                    nextProtos=self.npn_protocols,
                                     signedCertTimestamps=
                                     self.signed_cert_timestamps,
                                     fallbackSCSV=self.fallback_scsv_enabled,
                                     ocspResponse = self.ocsp_response)
       tlsConnection.ignoreAbruptClose = True
       return True
     except tlslite.api.TLSAbruptCloseError:
       # Ignore abrupt close.
       return True
     except tlslite.api.TLSError, error:
@@ skipping 1733 matching lines @@
         stapled_ocsp_response = None
         if self.options.staple_ocsp_response:
           stapled_ocsp_response = ocsp_der
 
         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.alpn_protocols,
                              self.options.npn_protocols,
                              self.options.record_resume,
                              self.options.tls_intolerant,
                              self.options.tls_intolerance_type,
                              self.options.signed_cert_timestamps_tls_ext.decode(
                                  "base64"),
                              self.options.fallback_scsv,
                              stapled_ocsp_response,
                              self.options.alert_after_handshake,
                              self.options.disable_channel_id,
@@ skipping 214 matching lines @@
                                   '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", '
                                   '"ecdhe_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('--alpn-protocols', action='append',
+                                  help='Specify the list of ALPN protocols.  '
+                                  'The server will not send an ALPN response '
+                                  'if this list does not overlap with the '
+                                  'list of protocols the client advertises.')
     self.option_parser.add_option('--npn-protocols', action='append',
-                                  help='Specify the list of protocols sent in'
+                                  help='Specify the list of protocols sent in '
                                   'an NPN response.  The server will not'
                                   'support NPN if the list is empty.')
     self.option_parser.add_option('--file-root-url', default='/files/',
                                   help='Specify a root URL for files served.')
     # TODO(ricea): Generalize this to support basic auth for HTTP too.
     self.option_parser.add_option('--ws-basic-auth', action='store_true',
                                   dest='ws_basic_auth',
                                   help='Enable basic-auth for WebSocket')
     self.option_parser.add_option('--ocsp-server-unavailable',
                                   dest='ocsp_server_unavailable',
@@ skipping 13 matching lines @@
                                   'an anonymous user.')
     self.option_parser.add_option('--disable-channel-id', action='store_true')
     self.option_parser.add_option('--disable-extended-master-secret',
                                   action='store_true')
     self.option_parser.add_option('--token-binding-params', action='append',
                                   default=[], type='int')
 
 
 if __name__ == '__main__':
   sys.exit(ServerRunner().main())