Add tests for TLS fallback on connection reset and close.

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 136 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, enable_npn,
-               record_resume_info, tls_intolerant, signed_cert_timestamps,
+               record_resume_info, tls_intolerant,
+               tls_intolerance_type, signed_cert_timestamps,
                fallback_scsv_enabled, ocsp_response):
     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 = []
     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
 
     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 tls_intolerant != 0:
+      self.ssl_handshake_settings.tlsIntolerant = (3, tls_intolerant)
+      self.ssl_handshake_settings.tlsIntoleranceType = tls_intolerance_type
 
     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.next_protos,
-                                    tlsIntolerant=self.tls_intolerant,
                                     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 1738 matching lines @@
 
         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.tls_intolerance_type,
                              self.options.signed_cert_timestamps_tls_ext.decode(
                                  "base64"),
                              self.options.fallback_scsv,
                              stapled_ocsp_response)
         print 'HTTPS server started on https://%s:%d...' % \
             (host, server.server_port)
       else:
         server = HTTPServer((host, port), TestPageHandler)
         print 'HTTP server started on http://%s:%d...' % \
             (host, server.server_port)
@@ skipping 126 matching lines @@
                                   help='If non-zero then the generated '
                                   'certificate will have this serial number')
     self.option_parser.add_option('--tls-intolerant', dest='tls_intolerant',
                                   default='0', type='int',
                                   help='If nonzero, certain TLS connections '
                                   'will be aborted in order to test version '
                                   'fallback. 1 means all TLS versions will be '
                                   'aborted. 2 means TLS 1.1 or higher will be '
                                   'aborted. 3 means TLS 1.2 or higher will be '
                                   'aborted.')
+    self.option_parser.add_option('--tls-intolerance-type',
+                                  dest='tls_intolerance_type',
+                                  default="alert",
+                                  help='Controls how the server reacts to a '
+                                  'TLS version it is intolerant to. Valid '
+                                  'values are "alert", "close", and "reset".')
     self.option_parser.add_option('--signed-cert-timestamps-tls-ext',
                                   dest='signed_cert_timestamps_tls_ext',
                                   default='',
                                   help='Base64 encoded SCT list. If set, '
                                   'server will respond with a '
                                   'signed_certificate_timestamp TLS extension '
                                   'whenever the client supports it.')
     self.option_parser.add_option('--fallback-scsv', dest='fallback_scsv',
                                   default=False, const=True,
                                   action='store_const',
@@ skipping 55 matching lines @@
                                   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.')
 
 
 if __name__ == '__main__':
   sys.exit(ServerRunner().main())