text: # Test the support for SSL and sockets
  
  import sys
  import unittest
  from test import test_support
  support = test_support
  import asyncore
  import socket
  import select
  import time
  import gc
  import os
  import errno
  import pprint
  import urllib, urlparse
  import traceback
  import weakref
  import functools
  import platform
  from contextlib import closing
  
  from BaseHTTPServer import HTTPServer
  from SimpleHTTPServer import SimpleHTTPRequestHandler
  
  ssl = test_support.import_module("ssl")
  
  try:
      PROTOCOLS = sorted(ssl._PROTOCOL_NAMES)
  except AttributeError:
      PROTOCOLS = ()
  
  HOST = test_support.HOST
  CERTFILE = None
  SVN_PYTHON_ORG_ROOT_CERT = None
  
  def handle_error(prefix):
      exc_format = ' '.join(traceback.format_exception(*sys.exc_info()))
      if test_support.verbose:
          sys.stdout.write(prefix + exc_format)
  
  
  class BasicTests(unittest.TestCase):
  
      def test_sslwrap_simple(self):
          # A crude test for the legacy API
          try:
              ssl.sslwrap_simple(socket.socket(socket.AF_INET))
          except IOError, e:
              if e.errno == 32: # broken pipe when ssl_sock.do_handshake(), this test doesn't care about that
                  pass
              else:
                  raise
          try:
              ssl.sslwrap_simple(socket.socket(socket.AF_INET)._sock)
          except IOError, e:
              if e.errno == 32: # broken pipe when ssl_sock.do_handshake(), this test doesn't care about that
                  pass
              else:
                  raise
  
  # Issue #9415: Ubuntu hijacks their OpenSSL and forcefully disables SSLv2
  def skip_if_broken_ubuntu_ssl(func):
      if hasattr(ssl, 'PROTOCOL_SSLv2'):
          if hasattr(ssl, 'SSLContext'): # >= 2.7.9
              def f(*args, **kwargs):
                  try:
                      ssl.SSLContext(ssl.PROTOCOL_SSLv2)
                  except ssl.SSLError:
                      if (ssl.OPENSSL_VERSION_INFO == (0, 9, 8, 15, 15) and
                          platform.linux_distribution() == ('debian', 'squeeze/sid', '')):
                          raise unittest.SkipTest("Patched Ubuntu OpenSSL breaks behaviour")
                  return func(*args, **kwargs)
              return f
  
          # We need to access the lower-level wrapper in order to create an
          # implicit SSL context without trying to connect or listen.
          try:
              import _ssl
          except ImportError:
              # The returned function won't get executed, just ignore the error
              pass
  
          @functools.wraps(func)
          def f(*args, **kwargs):
              try:
                  s = socket.socket(socket.AF_INET)
                  _ssl.sslwrap(s._sock, 0, None, None,
                               ssl.CERT_NONE, ssl.PROTOCOL_SSLv2, None, None)
              except ssl.SSLError as e:
                  if (ssl.OPENSSL_VERSION_INFO == (0, 9, 8, 15, 15) and
                      platform.linux_distribution() == ('debian', 'squeeze/sid', '')
                      and 'Invalid SSL protocol variant specified' in str(e)):
                      raise unittest.SkipTest("Patched Ubuntu OpenSSL breaks behaviour")
              return func(*args, **kwargs)
          return f
      else:
          return func
  
  
  class BasicSocketTests(unittest.TestCase):
  
      def test_constants(self):
          #ssl.PROTOCOL_SSLv2
          ssl.PROTOCOL_SSLv23
          ssl.PROTOCOL_SSLv3
          ssl.PROTOCOL_TLSv1
          ssl.CERT_NONE
          ssl.CERT_OPTIONAL
          ssl.CERT_REQUIRED
  
      def test_random(self):
          v = ssl.RAND_status()
          if test_support.verbose:
              sys.stdout.write("\n RAND_status is %d (%s)\n"
                               % (v, (v and "sufficient randomness") or
                                  "insufficient randomness"))
          try:
              ssl.RAND_egd(1)
          except TypeError:
              pass
          else:
              print "didn't raise TypeError"
          ssl.RAND_add("this is a random string", 75.0)
  
      def test_parse_cert(self):
          # note that this uses an 'unofficial' function in _ssl.c,
          # provided solely for this test, to exercise the certificate
          # parsing code
          try:
              p = ssl._ssl._test_decode_cert(CERTFILE, False)
          except TypeError: # >= 2.7.9
              p = ssl._ssl._test_decode_cert(CERTFILE)
          if test_support.verbose:
              sys.stdout.write("\n" + pprint.pformat(p) + "\n")
          self.assertEqual(p['subject'],
                           ((('countryName', 'XY'),),
                            (('localityName', 'Castle Anthrax'),),
                            (('organizationName', 'Python Software Foundation'),),
                            (('commonName', 'localhost'),))
                          )
          self.assertEqual(p['subjectAltName'], (('DNS', 'localhost'),))
          # Issue #13034: the subjectAltName in some certificates
          # (notably projects.developer.nokia.com:443) wasn't parsed
          p = ssl._ssl._test_decode_cert(NOKIACERT)
          if test_support.verbose:
              sys.stdout.write("\n" + pprint.pformat(p) + "\n")
          self.assertEqual(p['subjectAltName'],
                           (('DNS', 'projects.developer.nokia.com'),
                            ('DNS', 'projects.forum.nokia.com'))
                          )
  
      def test_DER_to_PEM(self):
          with open(SVN_PYTHON_ORG_ROOT_CERT, 'r') as f:
              pem = f.read()
          d1 = ssl.PEM_cert_to_DER_cert(pem)
          p2 = ssl.DER_cert_to_PEM_cert(d1)
          d2 = ssl.PEM_cert_to_DER_cert(p2)
          self.assertEqual(d1, d2)
          if not p2.startswith(ssl.PEM_HEADER + '\n'):
              self.fail("DER-to-PEM didn't include correct header:\n%r\n" % p2)
          if not p2.endswith('\n' + ssl.PEM_FOOTER + '\n'):
              self.fail("DER-to-PEM didn't include correct footer:\n%r\n" % p2)
  
      def test_openssl_version(self):
          n = ssl.OPENSSL_VERSION_NUMBER
          t = ssl.OPENSSL_VERSION_INFO
          s = ssl.OPENSSL_VERSION
          self.assertIsInstance(n, (int, long))
          self.assertIsInstance(t, tuple)
          self.assertIsInstance(s, str)
          # Some sanity checks follow
          # >= 0.9
          self.assertGreaterEqual(n, 0x900000)
          # < 2.0
          self.assertLess(n, 0x20000000)
          major, minor, fix, patch, status = t
          self.assertGreaterEqual(major, 0)
          self.assertLess(major, 2)
          self.assertGreaterEqual(minor, 0)
          self.assertLess(minor, 256)
          self.assertGreaterEqual(fix, 0)
          self.assertLess(fix, 256)
          self.assertGreaterEqual(patch, 0)
          self.assertLessEqual(patch, 26)
          self.assertGreaterEqual(status, 0)
          self.assertLessEqual(status, 15)
          # Version string as returned by OpenSSL, the format might change
          self.assertTrue(s.startswith("OpenSSL {:d}.{:d}.{:d}".format(major, minor, fix)),
                          (s, t))
  
      def test_ciphers(self):
          if not test_support.is_resource_enabled('network'):
              return
          remote = ("svn.python.org", 443)
          with test_support.transient_internet(remote[0]):
              s = ssl.wrap_socket(socket.socket(socket.AF_INET),
                                  cert_reqs=ssl.CERT_NONE, ciphers="ALL")
              s.connect(remote)
              s = ssl.wrap_socket(socket.socket(socket.AF_INET),
                                  cert_reqs=ssl.CERT_NONE, ciphers="DEFAULT")
              s.connect(remote)
              # Error checking can happen at instantiation (>=2.7.9) or when connecting
              with self.assertRaisesRegexp(ssl.SSLError, "No cipher can be selected"):
                  s = ssl.wrap_socket(socket.socket(socket.AF_INET),
                                  cert_reqs=ssl.CERT_NONE, ciphers="^$:,;?*'dorothyx")
                  s.connect(remote)
  
      @test_support.cpython_only
      def test_refcycle(self):
          # Issue #7943: an SSL object doesn't create reference cycles with
          # itself.
          s = socket.socket(socket.AF_INET)
          ss = ssl.wrap_socket(s)
          wr = weakref.ref(ss)
          del ss
          self.assertEqual(wr(), None)
  
      def test_wrapped_unconnected(self):
          # The _delegate_methods in socket.py are correctly delegated to by an
          # unconnected SSLSocket, so they will raise a socket.error rather than
          # something unexpected like TypeError.
          s = socket.socket(socket.AF_INET)
          ss = ssl.wrap_socket(s)
          self.assertRaises(socket.error, ss.recv, 1)
          self.assertRaises(socket.error, ss.recv_into, bytearray(b'x'))
          self.assertRaises(socket.error, ss.recvfrom, 1)
          self.assertRaises(socket.error, ss.recvfrom_into, bytearray(b'x'), 1)
          self.assertRaises(socket.error, ss.send, b'x')
          self.assertRaises(socket.error, ss.sendto, b'x', ('0.0.0.0', 0))
  
  
  class NetworkedTests(unittest.TestCase):
  
      def test_connect(self):
          with test_support.transient_internet("svn.python.org"):
              s = ssl.wrap_socket(socket.socket(socket.AF_INET),
                                  cert_reqs=ssl.CERT_NONE)
              s.connect(("svn.python.org", 443))
              c = s.getpeercert()
              if c:
                  self.fail("Peer cert %s shouldn't be here!")
              s.close()
  
              # this should fail because we have no verification certs
              s = ssl.wrap_socket(socket.socket(socket.AF_INET),
                                  cert_reqs=ssl.CERT_REQUIRED)
              try:
                  s.connect(("svn.python.org", 443))
              except ssl.SSLError:
                  pass
              finally:
                  s.close()
  
              # this should succeed because we specify the root cert
              s = ssl.wrap_socket(socket.socket(socket.AF_INET),
                                  cert_reqs=ssl.CERT_REQUIRED,
                                  ca_certs=SVN_PYTHON_ORG_ROOT_CERT)
              try:
                  s.connect(("svn.python.org", 443))
              finally:
                  s.close()
  
      def test_connect_ex(self):
          # Issue #11326: check connect_ex() implementation
          with test_support.transient_internet("svn.python.org"):
              s = ssl.wrap_socket(socket.socket(socket.AF_INET),
                                  cert_reqs=ssl.CERT_REQUIRED,
                                  ca_certs=SVN_PYTHON_ORG_ROOT_CERT)
              try:
                  self.assertEqual(0, s.connect_ex(("svn.python.org", 443)))
                  self.assertTrue(s.getpeercert())
              finally:
                  s.close()
  
      def test_non_blocking_connect_ex(self):
          # Issue #11326: non-blocking connect_ex() should allow handshake
          # to proceed after the socket gets ready.
          with test_support.transient_internet("svn.python.org"):
              s = ssl.wrap_socket(socket.socket(socket.AF_INET),
                                  cert_reqs=ssl.CERT_REQUIRED,
                                  ca_certs=SVN_PYTHON_ORG_ROOT_CERT,
                                  do_handshake_on_connect=False)
              try:
                  s.setblocking(False)
                  rc = s.connect_ex(('svn.python.org', 443))
                  # EWOULDBLOCK under Windows, EINPROGRESS elsewhere
                  self.assertIn(rc, (0, errno.EINPROGRESS, errno.EWOULDBLOCK))
                  # Wait for connect to finish
                  select.select([], [s], [], 5.0)
                  # Non-blocking handshake
                  while True:
                      try:
                          s.do_handshake()
                          break
                      except ssl.SSLError as err:
                          if err.args[0] == ssl.SSL_ERROR_WANT_READ:
                              select.select([s], [], [], 5.0)
                          elif err.args[0] == ssl.SSL_ERROR_WANT_WRITE:
                              select.select([], [s], [], 5.0)
                          else:
                              raise
                  # SSL established
                  self.assertTrue(s.getpeercert())
              finally:
                  s.close()
  
      @unittest.skipIf(os.name == "nt", "Can't use a socket as a file under Windows")
      def test_makefile_close(self):
          # Issue #5238: creating a file-like object with makefile() shouldn't
          # delay closing the underlying "real socket" (here tested with its
          # file descriptor, hence skipping the test under Windows).
          with test_support.transient_internet("svn.python.org"):
              ss = ssl.wrap_socket(socket.socket(socket.AF_INET))
              ss.connect(("svn.python.org", 443))
              fd = ss.fileno()
              f = ss.makefile()
              f.close()
              # The fd is still open
              os.read(fd, 0)
              # Closing the SSL socket should close the fd too
              ss.close()
              gc.collect()
              with self.assertRaises(OSError) as e:
                  os.read(fd, 0)
              self.assertEqual(e.exception.errno, errno.EBADF)
  
      def test_non_blocking_handshake(self):
          with test_support.transient_internet("svn.python.org"):
              s = socket.socket(socket.AF_INET)
              s.connect(("svn.python.org", 443))
              s.setblocking(False)
              s = ssl.wrap_socket(s,
                                  cert_reqs=ssl.CERT_NONE,
                                  do_handshake_on_connect=False)
              count = 0
              while True:
                  try:
                      count += 1
                      s.do_handshake()
                      break
                  except ssl.SSLError, err:
                      if err.args[0] == ssl.SSL_ERROR_WANT_READ:
                          select.select([s], [], [])
                      elif err.args[0] == ssl.SSL_ERROR_WANT_WRITE:
                          select.select([], [s], [])
                      else:
                          raise
              s.close()
              if test_support.verbose:
                  sys.stdout.write("\nNeeded %d calls to do_handshake() to establish session.\n" % count)
  
      def test_get_server_certificate(self):
          with test_support.transient_internet("svn.python.org"):
              pem = ssl.get_server_certificate(("svn.python.org", 443))
              if not pem:
                  self.fail("No server certificate on svn.python.org:443!")
  
              try:
                  pem = ssl.get_server_certificate(("svn.python.org", 443), ca_certs=CERTFILE)
              except ssl.SSLError:
                  #should fail
                  pass
              else:
                  self.fail("Got server certificate %s for svn.python.org!" % pem)
  
              pem = ssl.get_server_certificate(("svn.python.org", 443), ca_certs=SVN_PYTHON_ORG_ROOT_CERT)
              if not pem:
                  self.fail("No server certificate on svn.python.org:443!")
              if test_support.verbose:
                  sys.stdout.write("\nVerified certificate for svn.python.org:443 is\n%s\n" % pem)
  
      def test_algorithms(self):
          # Issue #8484: all algorithms should be available when verifying a
          # certificate.
          # SHA256 was added in OpenSSL 0.9.8
          if ssl.OPENSSL_VERSION_INFO < (0, 9, 8, 0, 15):
              self.skipTest("SHA256 not available on %r" % ssl.OPENSSL_VERSION)
          if not hasattr(ssl, 'SSLContext') or not getattr(ssl, 'HAS_SNI', True):
              # The Python 2.7.8 test lib doesn't run this
              self.skipTest('remote host needs SNI, only available on Python 3.2+')
          # NOTE: https://sha256.tbs-internet.com is another possible test host
          remote = ("sha256.tbs-internet.com", 443)
          sha256_cert = os.path.join(os.path.dirname(__file__), "sha256.pem")
  
          with test_support.transient_internet("sha256.tbs-internet.com"):
              ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
              ctx.verify_mode = ssl.CERT_REQUIRED
              ctx.load_verify_locations(sha256_cert)
              s = ctx.wrap_socket(socket.socket(socket.AF_INET),
                                  server_hostname="sha256.tbs-internet.com")
              try:
                  s.connect(remote)
                  if test_support.verbose:
                      sys.stdout.write("\nCipher with %r is %r\n" %
                                       (remote, s.cipher()))
                      sys.stdout.write("Certificate is:\n%s\n" %
                                       pprint.pformat(s.getpeercert()))
              finally:
                  s.close()
  
  
  try:
      import threading
  except ImportError:
      _have_threads = False
  else:
      _have_threads = True
  
      class ThreadedEchoServer(threading.Thread):
  
          class ConnectionHandler(threading.Thread):
  
              """A mildly complicated class, because we want it to work both
              with and without the SSL wrapper around the socket connection, so
              that we can test the STARTTLS functionality."""
  
              def __init__(self, server, connsock):
                  self.server = server
                  self.running = False
                  self.sock = connsock
                  self.sock.setblocking(1)
                  self.sslconn = None
                  threading.Thread.__init__(self)
                  self.daemon = True
  
              def show_conn_details(self):
                  if self.server.certreqs == ssl.CERT_REQUIRED:
                      cert = self.sslconn.getpeercert()
                      if test_support.verbose and self.server.chatty:
                          sys.stdout.write(" client cert is " + pprint.pformat(cert) + "\n")
                      cert_binary = self.sslconn.getpeercert(True)
                      if test_support.verbose and self.server.chatty:
                          sys.stdout.write(" cert binary is " + str(len(cert_binary)) + " bytes\n")
                  cipher = self.sslconn.cipher()
                  if test_support.verbose and self.server.chatty:
                      sys.stdout.write(" server: connection cipher is now " + str(cipher) + "\n")
  
              if hasattr(ssl, 'SSLContext'): # >= 2.7.9
                  def wrap_conn(self):
                      try:
                          self.sslconn = self.server.context.wrap_socket(
                              self.sock, server_side=True)
                          self.server.selected_protocols.append(self.sslconn.selected_npn_protocol())
                      except socket.error as e:
                          # We treat ConnectionResetError as though it were an
                          # SSLError - OpenSSL on Ubuntu abruptly closes the
                          # connection when asked to use an unsupported protocol.
                          #
                          # XXX Various errors can have happened here, for example
                          # a mismatching protocol version, an invalid certificate,
                          # or a low-level bug. This should be made more discriminating.
                          if not isinstance(e, ssl.SSLError) and e.errno != errno.ECONNRESET:
                              raise
                          self.server.conn_errors.append(e)
                          if self.server.chatty:
                              handle_error("\n server:  bad connection attempt from " + repr(self.addr) + ":\n")
                          self.running = False
                          self.server.stop()
                          self.close()
                          return False
                      else:
                          if self.server.context.verify_mode == ssl.CERT_REQUIRED:
                              cert = self.sslconn.getpeercert()
                              if support.verbose and self.server.chatty:
                                  sys.stdout.write(" client cert is " + pprint.pformat(cert) + "\n")
                              cert_binary = self.sslconn.getpeercert(True)
                              if support.verbose and self.server.chatty:
                                  sys.stdout.write(" cert binary is " + str(len(cert_binary)) + " bytes\n")
                          cipher = self.sslconn.cipher()
                          if support.verbose and self.server.chatty:
                              sys.stdout.write(" server: connection cipher is now " + str(cipher) + "\n")
                              sys.stdout.write(" server: selected protocol is now "
                                      + str(self.sslconn.selected_npn_protocol()) + "\n")
                          return True
              else:
                  def wrap_conn(self):
                      try:
                          self.sslconn = ssl.wrap_socket(self.sock, server_side=True,
                                                         certfile=self.server.certificate,
                                                         ssl_version=self.server.protocol,
                                                         ca_certs=self.server.cacerts,
                                                         cert_reqs=self.server.certreqs,
                                                         ciphers=self.server.ciphers)
                      except ssl.SSLError as e:
                          # XXX Various errors can have happened here, for example
                          # a mismatching protocol version, an invalid certificate,
                          # or a low-level bug. This should be made more discriminating.
                          self.server.conn_errors.append(e)
                          if self.server.chatty:
                              handle_error("\n server:  bad connection attempt from " +
                                           str(self.sock.getpeername()) + ":\n")
                          self.close()
                          self.running = False
                          self.server.stop()
                          return False
                      else:
                          return True
  
              def read(self):
                  if self.sslconn:
                      return self.sslconn.read()
                  else:
                      return self.sock.recv(1024)
  
              def write(self, bytes):
                  if self.sslconn:
                      return self.sslconn.write(bytes)
                  else:
                      return self.sock.send(bytes)
  
              def close(self):
                  if self.sslconn:
                      self.sslconn.close()
                  else:
                      self.sock._sock.close()
  
              def run(self):
                  self.running = True
                  if not self.server.starttls_server:
                      if isinstance(self.sock, ssl.SSLSocket):
                          self.sslconn = self.sock
                      elif not self.wrap_conn():
                          return
  
                  while self.running:
                      try:
                          msg = self.read()
                          if not msg:
                              # eof, so quit this handler
                              self.running = False
                              self.close()
                          elif msg.strip() == 'over':
                              if test_support.verbose and self.server.connectionchatty:
                                  sys.stdout.write(" server: client closed connection\n")
                              self.close()
                              return
                          elif self.server.starttls_server and msg.strip() == 'STARTTLS':
                              if test_support.verbose and self.server.connectionchatty:
                                  sys.stdout.write(" server: read STARTTLS from client, sending OK...\n")
                              self.write("OK\n")
                              if not self.wrap_conn():
                                  return
                          elif self.server.starttls_server and self.sslconn and msg.strip() == 'ENDTLS':
                              if test_support.verbose and self.server.connectionchatty:
                                  sys.stdout.write(" server: read ENDTLS from client, sending OK...\n")
                              self.write("OK\n")
                              self.sslconn.unwrap()
                              self.sslconn = None
                              if test_support.verbose and self.server.connectionchatty:
                                  sys.stdout.write(" server: connection is now unencrypted...\n")
                          else:
                              if (test_support.verbose and
                                  self.server.connectionchatty):
                                  ctype = (self.sslconn and "encrypted") or "unencrypted"
                                  sys.stdout.write(" server: read %s (%s), sending back %s (%s)...\n"
                                                   % (repr(msg), ctype, repr(msg.lower()), ctype))
                              self.write(msg.lower())
                      except ssl.SSLError:
                          if self.server.chatty:
                              handle_error("Test server failure:\n")
                          self.close()
                          self.running = False
                          # normally, we'd just stop here, but for the test
                          # harness, we want to stop the server
                          self.server.stop()
  
          if hasattr(ssl, 'SSLContext'): # >= 2.7.9
              def __init__(self, certificate=None, ssl_version=None,
                           certreqs=None, cacerts=None,
                           chatty=True, connectionchatty=False, starttls_server=False,
                           npn_protocols=None, ciphers=None, context=None):
                  if context:
                      self.context = context
                  else:
                      self.context = ssl.SSLContext(ssl_version
                                                    if ssl_version is not None
                                                    else ssl.PROTOCOL_TLSv1)
                      self.context.verify_mode = (certreqs if certreqs is not None
                                                  else ssl.CERT_NONE)
                      if cacerts:
                          self.context.load_verify_locations(cacerts)
                      if certificate:
                          self.context.load_cert_chain(certificate)
                      if npn_protocols:
                          self.context.set_npn_protocols(npn_protocols)
                      if ciphers:
                          self.context.set_ciphers(ciphers)
                  self.chatty = chatty
                  self.connectionchatty = connectionchatty
                  self.starttls_server = starttls_server
                  self.sock = socket.socket()
                  self.port = support.bind_port(self.sock)
                  self.flag = None
                  self.active = False
                  self.selected_protocols = []
                  self.conn_errors = []
                  threading.Thread.__init__(self)
                  self.daemon = True
          else:
              def __init__(self, certificate, ssl_version=None,
                           certreqs=None, cacerts=None,
                           chatty=True, connectionchatty=False, starttls_server=False,
                           wrap_accepting_socket=False, ciphers=None):
  
                  if ssl_version is None:
                      ssl_version = ssl.PROTOCOL_TLSv1
                  if certreqs is None:
                      certreqs = ssl.CERT_NONE
                  self.certificate = certificate
                  self.protocol = ssl_version
                  self.certreqs = certreqs
                  self.cacerts = cacerts
                  self.ciphers = ciphers
                  self.chatty = chatty
                  self.connectionchatty = connectionchatty
                  self.starttls_server = starttls_server
                  self.sock = socket.socket()
                  self.flag = None
                  if wrap_accepting_socket:
                      self.sock = ssl.wrap_socket(self.sock, server_side=True,
                                                  certfile=self.certificate,
                                                  cert_reqs = self.certreqs,
                                                  ca_certs = self.cacerts,
                                                  ssl_version = self.protocol,
                                                  ciphers = self.ciphers)
                      if test_support.verbose and self.chatty:
                          sys.stdout.write(' server:  wrapped server socket as %s\n' % str(self.sock))
                  self.port = test_support.bind_port(self.sock)
                  self.active = False
                  self.conn_errors = []
                  threading.Thread.__init__(self)
                  self.daemon = True
  
          def __enter__(self):
              self.start(threading.Event())
              self.flag.wait()
              return self
  
          def __exit__(self, *args):
              self.stop()
              self.join()
  
          def start(self, flag=None):
              self.flag = flag
              threading.Thread.start(self)
  
          def run(self):
              self.sock.settimeout(0.05)
              self.sock.listen(5)
              self.active = True
              if self.flag:
                  # signal an event
                  self.flag.set()
              while self.active:
                  try:
                      newconn, connaddr = self.sock.accept()
                      if test_support.verbose and self.chatty:
                          sys.stdout.write(' server:  new connection from '
                                           + str(connaddr) + '\n')
                      handler = self.ConnectionHandler(self, newconn)
                      handler.start()
                      handler.join()
                  except socket.timeout:
                      pass
                  except KeyboardInterrupt:
                      self.stop()
              self.sock.close()
  
          def stop(self):
              self.active = False
  
      class AsyncoreEchoServer(threading.Thread):
  
          class EchoServer(asyncore.dispatcher):
  
              class ConnectionHandler(asyncore.dispatcher_with_send):
  
                  def __init__(self, conn, certfile):
                      asyncore.dispatcher_with_send.__init__(self, conn)
                      self.socket = ssl.wrap_socket(conn, server_side=True,
                                                    certfile=certfile,
                                                    do_handshake_on_connect=False)
                      self._ssl_accepting = True
  
                  def readable(self):
                      if isinstance(self.socket, ssl.SSLSocket):
                          while self.socket.pending() > 0:
                              self.handle_read_event()
                      return True
  
                  def _do_ssl_handshake(self):
                      try:
                          self.socket.do_handshake()
                      except ssl.SSLError, err:
                          if err.args[0] in (ssl.SSL_ERROR_WANT_READ,
                                             ssl.SSL_ERROR_WANT_WRITE):
                              return
                          elif err.args[0] == ssl.SSL_ERROR_EOF:
                              return self.handle_close()
                          raise
                      except socket.error, err:
                          if err.args[0] == errno.ECONNABORTED:
                              return self.handle_close()
                      else:
                          self._ssl_accepting = False
  
                  def handle_read(self):
                      if self._ssl_accepting:
                          self._do_ssl_handshake()
                      else:
                          data = self.recv(1024)
                          if data and data.strip() != 'over':
                              self.send(data.lower())
  
                  def handle_close(self):
                      self.close()
                      if test_support.verbose:
                          sys.stdout.write(" server:  closed connection %s\n" % self.socket)
  
                  def handle_error(self):
                      raise
  
              def __init__(self, certfile):
                  self.certfile = certfile
                  asyncore.dispatcher.__init__(self)
                  self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
                  self.port = test_support.bind_port(self.socket)
                  self.listen(5)
  
              def handle_accept(self):
                  sock_obj, addr = self.accept()
                  if test_support.verbose:
                      sys.stdout.write(" server:  new connection from %s:%s\n" %addr)
                  self.ConnectionHandler(sock_obj, self.certfile)
  
              def handle_error(self):
                  raise
  
          def __init__(self, certfile):
              self.flag = None
              self.active = False
              self.server = self.EchoServer(certfile)
              self.port = self.server.port
              threading.Thread.__init__(self)
              self.daemon = True
  
          def __str__(self):
              return "<%s %s>" % (self.__class__.__name__, self.server)
  
          def __enter__(self):
              self.start(threading.Event())
              self.flag.wait()
              return self
  
          def __exit__(self, *args):
              if test_support.verbose:
                  sys.stdout.write(" cleanup: stopping server.\n")
              self.stop()
              if test_support.verbose:
                  sys.stdout.write(" cleanup: joining server thread.\n")
              self.join()
              if test_support.verbose:
                  sys.stdout.write(" cleanup: successfully joined.\n")
  
          def start(self, flag=None):
              self.flag = flag
              threading.Thread.start(self)
  
          def run(self):
              self.active = True
              if self.flag:
                  self.flag.set()
              while self.active:
                  asyncore.loop(0.05)
  
          def stop(self):
              self.active = False
              self.server.close()
  
      class SocketServerHTTPSServer(threading.Thread):
  
          class HTTPSServer(HTTPServer):
  
              def __init__(self, server_address, RequestHandlerClass, certfile):
                  HTTPServer.__init__(self, server_address, RequestHandlerClass)
                  # we assume the certfile contains both private key and certificate
                  self.certfile = certfile
                  self.allow_reuse_address = True
  
              def __str__(self):
                  return ('<%s %s:%s>' %
                          (self.__class__.__name__,
                           self.server_name,
                           self.server_port))
  
              def get_request(self):
                  # override this to wrap socket with SSL
                  sock, addr = self.socket.accept()
                  sslconn = ssl.wrap_socket(sock, server_side=True,
                                            certfile=self.certfile)
                  return sslconn, addr
  
          class RootedHTTPRequestHandler(SimpleHTTPRequestHandler):
              # need to override translate_path to get a known root,
              # instead of using os.curdir, since the test could be
              # run from anywhere
  
              server_version = "TestHTTPS/1.0"
  
              root = None
  
              def translate_path(self, path):
                  """Translate a /-separated PATH to the local filename syntax.
  
                  Components that mean special things to the local file system
                  (e.g. drive or directory names) are ignored.  (XXX They should
                  probably be diagnosed.)
  
                  """
                  # abandon query parameters
                  path = urlparse.urlparse(path)[2]
                  path = os.path.normpath(urllib.unquote(path))
                  words = path.split('/')
                  words = filter(None, words)
                  path = self.root
                  for word in words:
                      drive, word = os.path.splitdrive(word)
                      head, word = os.path.split(word)
                      if word in self.root: continue
                      path = os.path.join(path, word)
                  return path
  
              def log_message(self, format, *args):
  
                  # we override this to suppress logging unless "verbose"
  
                  if test_support.verbose:
                      sys.stdout.write(" server (%s:%d %s):\n   [%s] %s\n" %
                                       (self.server.server_address,
                                        self.server.server_port,
                                        self.request.cipher(),
                                        self.log_date_time_string(),
                                        format%args))
  
  
          def __init__(self, certfile):
              self.flag = None
              self.RootedHTTPRequestHandler.root = os.path.split(CERTFILE)[0]
              self.server = self.HTTPSServer(
                  (HOST, 0), self.RootedHTTPRequestHandler, certfile)
              self.port = self.server.server_port
              threading.Thread.__init__(self)
              self.daemon = True
  
          def __str__(self):
              return "<%s %s>" % (self.__class__.__name__, self.server)
  
          def start(self, flag=None):
              self.flag = flag
              threading.Thread.start(self)
  
          def run(self):
              if self.flag:
                  self.flag.set()
              self.server.serve_forever(0.05)
  
          def stop(self):
              self.server.shutdown()
  
  
      def bad_cert_test(certfile):
          """
          Launch a server with CERT_REQUIRED, and check that trying to
          connect to it with the given client certificate fails.
          """
          server = ThreadedEchoServer(CERTFILE,
                                      certreqs=ssl.CERT_REQUIRED,
                                      cacerts=CERTFILE, chatty=False)
          with server:
              try:
                  s = ssl.wrap_socket(socket.socket(),
                                      certfile=certfile,
                                      ssl_version=ssl.PROTOCOL_TLSv1)
                  s.connect((HOST, server.port))
              except ssl.SSLError, x:
                  if test_support.verbose:
                      sys.stdout.write("\nSSLError is %s\n" % x[1])
              except socket.error, x:
                  if test_support.verbose:
                      sys.stdout.write("\nsocket.error is %s\n" % x[1])
              else:
                  raise AssertionError("Use of invalid cert should have failed!")
  
      if hasattr(ssl, 'SSLContext'): # >= 2.7.9
          def server_params_test(client_context, server_context, indata=b"FOO\n",
                             chatty=True, connectionchatty=False, sni_name=None):
              """
              Launch a server, connect a client to it and try various reads
              and writes.
              """
              stats = {}
              server = ThreadedEchoServer(context=server_context,
                                          chatty=chatty,
                                          connectionchatty=False)
              with server:
                  with closing(client_context.wrap_socket(socket.socket(),
                          server_hostname=sni_name)) as s:
                      s.connect((HOST, server.port))
                      for arg in [indata, bytearray(indata), memoryview(indata)]:
                          if connectionchatty:
                              if support.verbose:
                                  sys.stdout.write(
                                      " client:  sending %r...\n" % indata)
                          s.write(arg)
                          outdata = s.read()
                          if connectionchatty:
                              if support.verbose:
                                  sys.stdout.write(" client:  read %r\n" % outdata)
                          if outdata != indata.lower():
                              raise AssertionError(
                                  "bad data <<%r>> (%d) received; expected <<%r>> (%d)\n"
                                  % (outdata[:20], len(outdata),
                                     indata[:20].lower(), len(indata)))
                      s.write(b"over\n")
                      if connectionchatty:
                          if support.verbose:
                              sys.stdout.write(" client:  closing connection.\n")
                      stats.update({
                          'compression': s.compression(),
                          'cipher': s.cipher(),
                          'peercert': s.getpeercert(),
                          'client_npn_protocol': s.selected_npn_protocol(),
                          'version': s.version(),
                      })
                      s.close()
                  stats['server_npn_protocols'] = server.selected_protocols
              return stats
  
          def try_protocol_combo(server_protocol, client_protocol, expect_success,
                                 certsreqs=None, server_options=0, client_options=0):
              """
              Try to SSL-connect using *client_protocol* to *server_protocol*.
              If *expect_success* is true, assert that the connection succeeds,
              if it's false, assert that the connection fails.
              Also, if *expect_success* is a string, assert that it is the protocol
              version actually used by the connection.
              """
              if certsreqs is None:
                  certsreqs = ssl.CERT_NONE
              certtype = {
                  ssl.CERT_NONE: "CERT_NONE",
                  ssl.CERT_OPTIONAL: "CERT_OPTIONAL",
                  ssl.CERT_REQUIRED: "CERT_REQUIRED",
              }[certsreqs]
              if support.verbose:
                  formatstr = (expect_success and " %s->%s %s\n") or " {%s->%s} %s\n"
                  sys.stdout.write(formatstr %
                                   (ssl.get_protocol_name(client_protocol),
                                    ssl.get_protocol_name(server_protocol),
                                    certtype))
              client_context = ssl.SSLContext(client_protocol)
              client_context.options |= client_options
              server_context = ssl.SSLContext(server_protocol)
              server_context.options |= server_options
  
              # NOTE: we must enable "ALL" ciphers on the client, otherwise an
              # SSLv23 client will send an SSLv3 hello (rather than SSLv2)
              # starting from OpenSSL 1.0.0 (see issue #8322).
              if client_context.protocol == ssl.PROTOCOL_SSLv23:
                  client_context.set_ciphers("ALL")
  
              for ctx in (client_context, server_context):
                  ctx.verify_mode = certsreqs
                  ctx.load_cert_chain(CERTFILE)
                  ctx.load_verify_locations(CERTFILE)
              try:
                  stats = server_params_test(client_context, server_context,
                                             chatty=False, connectionchatty=False)
              # Protocol mismatch can result in either an SSLError, or a
              # "Connection reset by peer" error.
              except ssl.SSLError:
                  if expect_success:
                      raise
              except socket.error as e:
                  if expect_success or e.errno != errno.ECONNRESET:
                      raise
              else:
                  if not expect_success:
                      raise AssertionError(
                          "Client protocol %s succeeded with server protocol %s!"
                          % (ssl.get_protocol_name(client_protocol),
                             ssl.get_protocol_name(server_protocol)))
                  elif (expect_success is not True
                        and expect_success != stats['version']):
                      raise AssertionError("version mismatch: expected %r, got %r"
                                           % (expect_success, stats['version']))
  
      else:
          def server_params_test(certfile, protocol, certreqs, cacertsfile,
                                 client_certfile, client_protocol=None, indata="FOO\n",
                                 ciphers=None, chatty=True, connectionchatty=False,
                                 wrap_accepting_socket=False):
              """
              Launch a server, connect a client to it and try various reads
              and writes.
              """
              server = ThreadedEchoServer(certfile,
                                          certreqs=certreqs,
                                          ssl_version=protocol,
                                          cacerts=cacertsfile,
                                          ciphers=ciphers,
                                          chatty=chatty,
                                          connectionchatty=connectionchatty,
                                          wrap_accepting_socket=wrap_accepting_socket)
              with server:
                  # try to connect
                  if client_protocol is None:
                      client_protocol = protocol
                  s = ssl.wrap_socket(socket.socket(),
                                      certfile=client_certfile,
                                      ca_certs=cacertsfile,
                                      ciphers=ciphers,
                                      cert_reqs=certreqs,
                                      ssl_version=client_protocol)
                  s.connect((HOST, server.port))
                  for arg in [indata, bytearray(indata), memoryview(indata)]:
                      if connectionchatty:
                          if test_support.verbose:
                              sys.stdout.write(
                                  " client:  sending %s...\n" % (repr(arg)))
                      s.write(arg)
                      outdata = s.read()
                      if connectionchatty:
                          if test_support.verbose:
                              sys.stdout.write(" client:  read %s\n" % repr(outdata))
                      if outdata != indata.lower():
                          raise AssertionError(
                              "bad data <<%s>> (%d) received; expected <<%s>> (%d)\n"
                              % (outdata[:min(len(outdata),20)], len(outdata),
                                 indata[:min(len(indata),20)].lower(), len(indata)))
                  s.write("over\n")
                  if connectionchatty:
                      if test_support.verbose:
                          sys.stdout.write(" client:  closing connection.\n")
                  s.close()
  
          def try_protocol_combo(server_protocol,
                                 client_protocol,
                                 expect_success,
                                 certsreqs=None):
              if certsreqs is None:
                  certsreqs = ssl.CERT_NONE
              certtype = {
                  ssl.CERT_NONE: "CERT_NONE",
                  ssl.CERT_OPTIONAL: "CERT_OPTIONAL",
                  ssl.CERT_REQUIRED: "CERT_REQUIRED",
              }[certsreqs]
              if test_support.verbose:
                  formatstr = (expect_success and " %s->%s %s\n") or " {%s->%s} %s\n"
                  sys.stdout.write(formatstr %
                                   (ssl.get_protocol_name(client_protocol),
                                    ssl.get_protocol_name(server_protocol),
                                    certtype))
              try:
                  # NOTE: we must enable "ALL" ciphers, otherwise an SSLv23 client
                  # will send an SSLv3 hello (rather than SSLv2) starting from
                  # OpenSSL 1.0.0 (see issue #8322).
                  server_params_test(CERTFILE, server_protocol, certsreqs,
                                     CERTFILE, CERTFILE, client_protocol,
                                     ciphers="ALL", chatty=False)
              # Protocol mismatch can result in either an SSLError, or a
              # "Connection reset by peer" error.
              except ssl.SSLError:
                  if expect_success:
                      raise
              except socket.error as e:
                  if expect_success or e.errno != errno.ECONNRESET:
                      raise
              else:
                  if not expect_success:
                      raise AssertionError(
                          "Client protocol %s succeeded with server protocol %s!"
                          % (ssl.get_protocol_name(client_protocol),
                             ssl.get_protocol_name(server_protocol)))
  
  
      class ThreadedTests(unittest.TestCase):
  
          def test_rude_shutdown(self):
              """A brutal shutdown of an SSL server should raise an IOError
              in the client when attempting handshake.
              """
              listener_ready = threading.Event()
              listener_gone = threading.Event()
  
              s = socket.socket()
              port = test_support.bind_port(s, HOST)
  
              # `listener` runs in a thread.  It sits in an accept() until
              # the main thread connects.  Then it rudely closes the socket,
              # and sets Event `listener_gone` to let the main thread know
              # the socket is gone.
              def listener():
                  s.listen(5)
                  listener_ready.set()
                  s.accept()
                  s.close()
                  listener_gone.set()
  
              def connector():
                  listener_ready.wait()
                  c = socket.socket()
                  c.connect((HOST, port))
                  listener_gone.wait()
                  try:
                      ssl_sock = ssl.wrap_socket(c)
                  except IOError:
                      pass
                  else:
                      self.fail('connecting to closed SSL socket should have failed')
  
              t = threading.Thread(target=listener)
              t.start()
              try:
                  connector()
              finally:
                  t.join()
  
          @skip_if_broken_ubuntu_ssl
          def test_echo(self):
              """Basic test of an SSL client connecting to a server"""
              if test_support.verbose:
                  sys.stdout.write("\n")
              if hasattr(ssl, 'SSLContext'): # >= 2.7.9
                  for protocol in PROTOCOLS:
                      context = ssl.SSLContext(protocol)
                      context.load_cert_chain(CERTFILE)
                      server_params_test(context, context,
                                         chatty=True, connectionchatty=True)
              else:
                  server_params_test(CERTFILE, ssl.PROTOCOL_TLSv1, ssl.CERT_NONE,
                                     CERTFILE, CERTFILE, ssl.PROTOCOL_TLSv1,
                                     chatty=True, connectionchatty=True)
  
          def test_getpeercert(self):
              if test_support.verbose:
                  sys.stdout.write("\n")
              s2 = socket.socket()
              server = ThreadedEchoServer(CERTFILE,
                                          certreqs=ssl.CERT_NONE,
                                          ssl_version=ssl.PROTOCOL_SSLv23,
                                          cacerts=CERTFILE,
                                          chatty=False)
              with server:
                  s = ssl.wrap_socket(socket.socket(),
                                      certfile=CERTFILE,
                                      ca_certs=CERTFILE,
                                      cert_reqs=ssl.CERT_REQUIRED,
                                      ssl_version=ssl.PROTOCOL_SSLv23)
                  s.connect((HOST, server.port))
                  cert = s.getpeercert()
                  self.assertTrue(cert, "Can't get peer certificate.")
                  cipher = s.cipher()
                  if test_support.verbose:
                      sys.stdout.write(pprint.pformat(cert) + '\n')
                      sys.stdout.write("Connection cipher is " + str(cipher) + '.\n')
                  if 'subject' not in cert:
                      self.fail("No subject field in certificate: %s." %
                                pprint.pformat(cert))
                  if ((('organizationName', 'Python Software Foundation'),)
                      not in cert['subject']):
                      self.fail(
                          "Missing or invalid 'organizationName' field in certificate subject; "
                          "should be 'Python Software Foundation'.")
                  s.close()
  
          def test_empty_cert(self):
              """Connecting with an empty cert file"""
              bad_cert_test(os.path.join(os.path.dirname(__file__) or os.curdir,
                                        "nullcert.pem"))
          def test_malformed_cert(self):
              """Connecting with a badly formatted certificate (syntax error)"""
              bad_cert_test(os.path.join(os.path.dirname(__file__) or os.curdir,
                                         "badcert.pem"))
          def test_nonexisting_cert(self):
              """Connecting with a non-existing cert file"""
              bad_cert_test(os.path.join(os.path.dirname(__file__) or os.curdir,
                                         "wrongcert.pem"))
          def test_malformed_key(self):
              """Connecting with a badly formatted key (syntax error)"""
              bad_cert_test(os.path.join(os.path.dirname(__file__) or os.curdir,
                                         "badkey.pem"))
  
          @skip_if_broken_ubuntu_ssl
          def test_protocol_sslv2(self):
              """Connecting to an SSLv2 server with various client options"""
              if test_support.verbose:
                  sys.stdout.write("\n")
              if not hasattr(ssl, 'PROTOCOL_SSLv2'):
                  self.skipTest("PROTOCOL_SSLv2 needed")
              try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv2, True)
              try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv2, True, ssl.CERT_OPTIONAL)
              try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv2, True, ssl.CERT_REQUIRED)
              try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv23, False)
              try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv3, False)
              try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_TLSv1, False)
  
          @skip_if_broken_ubuntu_ssl
          def test_protocol_sslv23(self):
              """Connecting to an SSLv23 server with various client options"""
              if test_support.verbose:
                  sys.stdout.write("\n")
              try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv3, True)
              try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv23, True)
              try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_TLSv1, True)
  
              try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv3, True, ssl.CERT_OPTIONAL)
              try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv23, True, ssl.CERT_OPTIONAL)
              try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_TLSv1, True, ssl.CERT_OPTIONAL)
  
              try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv3, True, ssl.CERT_REQUIRED)
              try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv23, True, ssl.CERT_REQUIRED)
              try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_TLSv1, True, ssl.CERT_REQUIRED)
  
          @skip_if_broken_ubuntu_ssl
          def test_protocol_sslv3(self):
              """Connecting to an SSLv3 server with various client options"""
              if test_support.verbose:
                  sys.stdout.write("\n")
              try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, True)
              try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, True, ssl.CERT_OPTIONAL)
              try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, True, ssl.CERT_REQUIRED)
              if hasattr(ssl, 'PROTOCOL_SSLv2'):
                  try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv2, False)
              try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_TLSv1, False)
  
          @skip_if_broken_ubuntu_ssl
          def test_protocol_tlsv1(self):
              """Connecting to a TLSv1 server with various client options"""
              if test_support.verbose:
                  sys.stdout.write("\n")
              try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, True)
              try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, True, ssl.CERT_OPTIONAL)
              try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, True, ssl.CERT_REQUIRED)
              if hasattr(ssl, 'PROTOCOL_SSLv2'):
                  try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_SSLv2, False)
              try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_SSLv3, False)
  
          def test_starttls(self):
              """Switching from clear text to encrypted and back again."""
              msgs = ("msg 1", "MSG 2", "STARTTLS", "MSG 3", "msg 4", "ENDTLS", "msg 5", "msg 6")
  
              server = ThreadedEchoServer(CERTFILE,
                                          ssl_version=ssl.PROTOCOL_TLSv1,
                                          starttls_server=True,
                                          chatty=True,
                                          connectionchatty=True)
              wrapped = False
              with server:
                  s = socket.socket()
                  s.setblocking(1)
                  s.connect((HOST, server.port))
                  if test_support.verbose:
                      sys.stdout.write("\n")
                  for indata in msgs:
                      if test_support.verbose:
                          sys.stdout.write(
                              " client:  sending %s...\n" % repr(indata))
                      if wrapped:
                          conn.write(indata)
                          outdata = conn.read()
                      else:
                          s.send(indata)
                          outdata = s.recv(1024)
                      if (indata == "STARTTLS" and
                          outdata.strip().lower().startswith("ok")):
                          # STARTTLS ok, switch to secure mode
                          if test_support.verbose:
                              sys.stdout.write(
                                  " client:  read %s from server, starting TLS...\n"
                                  % repr(outdata))
                          conn = ssl.wrap_socket(s, ssl_version=ssl.PROTOCOL_TLSv1)
                          wrapped = True
                      elif (indata == "ENDTLS" and
                          outdata.strip().lower().startswith("ok")):
                          # ENDTLS ok, switch back to clear text
                          if test_support.verbose:
                              sys.stdout.write(
                                  " client:  read %s from server, ending TLS...\n"
                                  % repr(outdata))
                          s = conn.unwrap()
                          wrapped = False
                      else:
                          if test_support.verbose:
                              sys.stdout.write(
                                  " client:  read %s from server\n" % repr(outdata))
                  if test_support.verbose:
                      sys.stdout.write(" client:  closing connection.\n")
                  if wrapped:
                      conn.write("over\n")
                  else:
                      s.send("over\n")
                  s.close()
  
          def test_socketserver(self):
              """Using a SocketServer to create and manage SSL connections."""
              server = SocketServerHTTPSServer(CERTFILE)
              flag = threading.Event()
              server.start(flag)
              # wait for it to start
              flag.wait()
              # try to connect
              try:
                  if test_support.verbose:
                      sys.stdout.write('\n')
                  with open(CERTFILE, 'rb') as f:
                      d1 = f.read()
                  d2 = ''
                  # now fetch the same data from the HTTPS server
                  url = 'https://127.0.0.1:%d/%s' % (
                      server.port, os.path.split(CERTFILE)[1])
                  with test_support.check_py3k_warnings():
                      import ssl
                      if hasattr(ssl, '_create_unverified_context'):
                          # Disable verification for our self-signed cert
                          # on Python >= 2.7.9 and 3.4
                          ssl_ctx = ssl._create_unverified_context()
                          f = urllib.urlopen(url, context=ssl_ctx)
                      else:
                          f = urllib.urlopen(url)
                  dlen = f.info().getheader("content-length")
                  if dlen and (int(dlen) > 0):
                      d2 = f.read(int(dlen))
                      if test_support.verbose:
                          sys.stdout.write(
                              " client: read %d bytes from remote server '%s'\n"
                              % (len(d2), server))
                  f.close()
                  self.assertEqual(d1, d2)
              finally:
                  server.stop()
                  server.join()
  
          @unittest.skipIf(hasattr(ssl, 'SSLContext'), "<= 2.7.9 only")
          def test_wrapped_accept(self):
              """Check the accept() method on SSL sockets."""
              if test_support.verbose:
                  sys.stdout.write("\n")
              server_params_test(CERTFILE, ssl.PROTOCOL_SSLv23, ssl.CERT_REQUIRED,
                                 CERTFILE, CERTFILE, ssl.PROTOCOL_SSLv23,
                                 chatty=True, connectionchatty=True,
                                 wrap_accepting_socket=True)
  
          def test_asyncore_server(self):
              """Check the example asyncore integration."""
              indata = "TEST MESSAGE of mixed case\n"
  
              if test_support.verbose:
                  sys.stdout.write("\n")
              server = AsyncoreEchoServer(CERTFILE)
              with server:
                  s = ssl.wrap_socket(socket.socket())
                  s.connect(('127.0.0.1', server.port))
                  if test_support.verbose:
                      sys.stdout.write(
                          " client:  sending %s...\n" % (repr(indata)))
                  s.write(indata)
                  outdata = s.read()
                  if test_support.verbose:
                      sys.stdout.write(" client:  read %s\n" % repr(outdata))
                  if outdata != indata.lower():
                      self.fail(
                          "bad data <<%s>> (%d) received; expected <<%s>> (%d)\n"
                          % (outdata[:min(len(outdata),20)], len(outdata),
                             indata[:min(len(indata),20)].lower(), len(indata)))
                  s.write("over\n")
                  if test_support.verbose:
                      sys.stdout.write(" client:  closing connection.\n")
                  s.close()
  
          def test_recv_send(self):
              """Test recv(), send() and friends."""
              if test_support.verbose:
                  sys.stdout.write("\n")
  
              server = ThreadedEchoServer(CERTFILE,
                                          certreqs=ssl.CERT_NONE,
                                          ssl_version=ssl.PROTOCOL_TLSv1,
                                          cacerts=CERTFILE,
                                          chatty=True,
                                          connectionchatty=False)
              with server:
                  s = ssl.wrap_socket(socket.socket(),
                                      server_side=False,
                                      certfile=CERTFILE,
                                      ca_certs=CERTFILE,
                                      cert_reqs=ssl.CERT_NONE,
                                      ssl_version=ssl.PROTOCOL_TLSv1)
                  s.connect((HOST, server.port))
                  # helper methods for standardising recv* method signatures
                  def _recv_into():
                      b = bytearray("\0"*100)
                      count = s.recv_into(b)
                      return b[:count]
  
                  def _recvfrom_into():
                      b = bytearray("\0"*100)
                      count, addr = s.recvfrom_into(b)
                      return b[:count]
  
                  # (name, method, whether to expect success, *args)
                  send_methods = [
                      ('send', s.send, True, []),
                      ('sendto', s.sendto, False, ["some.address"]),
                      ('sendall', s.sendall, True, []),
                  ]
                  recv_methods = [
                      ('recv', s.recv, True, []),
                      ('recvfrom', s.recvfrom, False, ["some.address"]),
                      ('recv_into', _recv_into, True, []),
                      ('recvfrom_into', _recvfrom_into, False, []),
                  ]
                  data_prefix = u"PREFIX_"
  
                  for meth_name, send_meth, expect_success, args in send_methods:
                      indata = data_prefix + meth_name
                      try:
                          send_meth(indata.encode('ASCII', 'strict'), *args)
                          outdata = s.read()
                          outdata = outdata.decode('ASCII', 'strict')
                          if outdata != indata.lower():
                              self.fail(
                                  "While sending with <<%s>> bad data "
                                  "<<%r>> (%d) received; "
                                  "expected <<%r>> (%d)\n" % (
                                      meth_name, outdata[:20], len(outdata),
                                      indata[:20], len(indata)
                                  )
                              )
                      except ValueError as e:
                          if expect_success:
                              self.fail(
                                  "Failed to send with method <<%s>>; "
                                  "expected to succeed.\n" % (meth_name,)
                              )
                          if not str(e).startswith(meth_name):
                              self.fail(
                                  "Method <<%s>> failed with unexpected "
                                  "exception message: %s\n" % (
                                      meth_name, e
                                  )
                              )
  
                  for meth_name, recv_meth, expect_success, args in recv_methods:
                      indata = data_prefix + meth_name
                      try:
                          s.send(indata.encode('ASCII', 'strict'))
                          outdata = recv_meth(*args)
                          outdata = outdata.decode('ASCII', 'strict')
                          if outdata != indata.lower():
                              self.fail(
                                  "While receiving with <<%s>> bad data "
                                  "<<%r>> (%d) received; "
                                  "expected <<%r>> (%d)\n" % (
                                      meth_name, outdata[:20], len(outdata),
                                      indata[:20], len(indata)
                                  )
                              )
                      except ValueError as e:
                          if expect_success:
                              self.fail(
                                  "Failed to receive with method <<%s>>; "
                                  "expected to succeed.\n" % (meth_name,)
                              )
                          if not str(e).startswith(meth_name):
                              self.fail(
                                  "Method <<%s>> failed with unexpected "
                                  "exception message: %s\n" % (
                                      meth_name, e
                                  )
                              )
                          # consume data
                          s.read()
  
                  s.write("over\n".encode("ASCII", "strict"))
                  s.close()
  
          def test_handshake_timeout(self):
              # Issue #5103: SSL handshake must respect the socket timeout
              server = socket.socket(socket.AF_INET)
              host = "127.0.0.1"
              port = test_support.bind_port(server)
              started = threading.Event()
              finish = False
  
              def serve():
                  server.listen(5)
                  started.set()
                  conns = []
                  while not finish:
                      r, w, e = select.select([server], [], [], 0.1)
                      if server in r:
                          # Let the socket hang around rather than having
                          # it closed by garbage collection.
                          conns.append(server.accept()[0])
  
              t = threading.Thread(target=serve)
              t.start()
              started.wait()
  
              try:
                  try:
                      c = socket.socket(socket.AF_INET)
                      c.settimeout(0.2)
                      c.connect((host, port))
                      # Will attempt handshake and time out
                      self.assertRaisesRegexp(ssl.SSLError, "timed out",
                                              ssl.wrap_socket, c)
                  finally:
                      c.close()
                  try:
                      c = socket.socket(socket.AF_INET)
                      c.settimeout(0.2)
                      c = ssl.wrap_socket(c)
                      # Will attempt handshake and time out
                      self.assertRaisesRegexp(ssl.SSLError, "timed out",
                                              c.connect, (host, port))
                  finally:
                      c.close()
              finally:
                  finish = True
                  t.join()
                  server.close()
  
          def test_default_ciphers(self):
              with ThreadedEchoServer(CERTFILE,
                                      ssl_version=ssl.PROTOCOL_SSLv23,
                                      chatty=False) as server:
                  sock = socket.socket()
                  try:
                      # Force a set of weak ciphers on our client socket
                      try:
                          s = ssl.wrap_socket(sock,
                                              ssl_version=ssl.PROTOCOL_SSLv23,
                                              ciphers="DES")
                      except ssl.SSLError:
                          self.skipTest("no DES cipher available")
                      with self.assertRaises((OSError, ssl.SSLError)):
                          s.connect((HOST, server.port))
                  finally:
                      sock.close()
              self.assertIn("no shared cipher", str(server.conn_errors[0]))
  
  
  def test_main(verbose=False):
      global CERTFILE, SVN_PYTHON_ORG_ROOT_CERT, NOKIACERT
      CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir,
                              "keycert.pem")
      SVN_PYTHON_ORG_ROOT_CERT = os.path.join(
          os.path.dirname(__file__) or os.curdir,
          "https_svn_python_org_root.pem")
      NOKIACERT = os.path.join(os.path.dirname(__file__) or os.curdir,
                               "nokia.pem")
  
      if (not os.path.exists(CERTFILE) or
          not os.path.exists(SVN_PYTHON_ORG_ROOT_CERT) or
          not os.path.exists(NOKIACERT)):
          raise test_support.TestFailed("Can't read certificate files!")
  
      tests = [BasicTests, BasicSocketTests]
  
      if test_support.is_resource_enabled('network'):
          tests.append(NetworkedTests)
  
      if _have_threads:
          thread_info = test_support.threading_setup()
          if thread_info and test_support.is_resource_enabled('network'):
              tests.append(ThreadedTests)
  
      try:
          test_support.run_unittest(*tests)
      finally:
          if _have_threads:
              test_support.threading_cleanup(*thread_info)
  
  if __name__ == "__main__":
      test_main()
  