Properly handle alerts from the peer in SSL_read.

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 139 matching lines @@
                   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,
                tls_intolerance_type, signed_cert_timestamps,
-               fallback_scsv_enabled, ocsp_response):
+               fallback_scsv_enabled, ocsp_response,
+               alert_after_handshake):
     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
@@ skipping 24 matching lines @@
     self.ssl_handshake_settings = tlslite.api.HandshakeSettings()
     # Enable SSLv3 for testing purposes.
     self.ssl_handshake_settings.minVersion = (3, 0)
     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 alert_after_handshake:
+      self.ssl_handshake_settings.alertAfterHandshake = True
 
     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)
@@ skipping 1823 matching lines @@
                              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)
+                             stapled_ocsp_response,
+                             self.options.alert_after_handshake)
         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)
 
       server.data_dir = self.__make_data_dir()
       server.file_root_url = self.options.file_root_url
       server_data['port'] = server.server_port
@@ skipping 212 matching lines @@
     # 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',
                                   default=False, action='store_true',
                                   help='If set, the OCSP server will return '
                                   'a tryLater status rather than the actual '
                                   'OCSP response.')
+    self.option_parser.add_option('--alert-after-handshake',
+                                  dest='alert_after_handshake',
+                                  default=False, action='store_true',
+                                  help='If set, the server will send a fatal '
+                                  'alert immediately after the handshake.')
 
 
 if __name__ == '__main__':
   sys.exit(ServerRunner().main())