mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-12-09 02:35:14 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

Merged revisions 80596 via svnmerge from

Browse source 
svn+ssh://pythondev@svn.python.org/python/trunk

........
  r80596 | antoine.pitrou | 2010-04-28 23:11:01 +0200 (mer., 28 avril 2010) | 3 lines

  Fix style issues in test_ssl
........
This commit is contained in:
Antoine Pitrou 2010-04-28 21:37:09 +00:00
parent f9fc08f9f3
commit 480a124973
1 changed files with 185 additions and 187 deletions
Download patch file Download diff file Expand all files Collapse all files

372
Lib/test/test_ssl.py
View file

@ -36,7 +36,7 @@ def handle_error(prefix):
class BasicTests(unittest.TestCase): class BasicTests(unittest.TestCase):
def testSSLconnect(self): def test_connect(self):
if not support.is_resource_enabled('network'): if not support.is_resource_enabled('network'):
return return
s = ssl.wrap_socket(socket.socket(socket.AF_INET), s = ssl.wrap_socket(socket.socket(socket.AF_INET),
@ -57,7 +57,7 @@ class BasicTests(unittest.TestCase):
finally: finally:
s.close() s.close()
def testCrucialConstants(self): def test_constants(self):
ssl.PROTOCOL_SSLv2 ssl.PROTOCOL_SSLv2
ssl.PROTOCOL_SSLv23 ssl.PROTOCOL_SSLv23
ssl.PROTOCOL_SSLv3 ssl.PROTOCOL_SSLv3
@ -66,7 +66,7 @@ class BasicTests(unittest.TestCase):
ssl.CERT_OPTIONAL ssl.CERT_OPTIONAL
ssl.CERT_REQUIRED ssl.CERT_REQUIRED
def testRAND(self): def test_random(self):
v = ssl.RAND_status() v = ssl.RAND_status()
if support.verbose: if support.verbose:
sys.stdout.write("\n RAND_status is %d (%s)\n" sys.stdout.write("\n RAND_status is %d (%s)\n"
@ -80,7 +80,7 @@ class BasicTests(unittest.TestCase):
print("didn't raise TypeError") print("didn't raise TypeError")
ssl.RAND_add("this is a random string", 75.0) ssl.RAND_add("this is a random string", 75.0)
def testParseCert(self): def test_parse_cert(self):
# note that this uses an 'unofficial' function in _ssl.c, # note that this uses an 'unofficial' function in _ssl.c,
# provided solely for this test, to exercise the certificate # provided solely for this test, to exercise the certificate
# parsing code # parsing code
@ -88,9 +88,9 @@ class BasicTests(unittest.TestCase):
if support.verbose: if support.verbose:
sys.stdout.write("\n" + pprint.pformat(p) + "\n") sys.stdout.write("\n" + pprint.pformat(p) + "\n")
def testDERtoPEM(self): def test_DER_to_PEM(self):
with open(SVN_PYTHON_ORG_ROOT_CERT, 'r') as f:
pem = open(SVN_PYTHON_ORG_ROOT_CERT, 'r').read() pem = f.read()
d1 = ssl.PEM_cert_to_DER_cert(pem) d1 = ssl.PEM_cert_to_DER_cert(pem)
p2 = ssl.DER_cert_to_PEM_cert(d1) p2 = ssl.DER_cert_to_PEM_cert(d1)
d2 = ssl.PEM_cert_to_DER_cert(p2) d2 = ssl.PEM_cert_to_DER_cert(p2)
@ -166,7 +166,7 @@ class BasicTests(unittest.TestCase):
class NetworkedTests(unittest.TestCase): class NetworkedTests(unittest.TestCase):
def testConnect(self): def test_connect(self):
s = ssl.wrap_socket(socket.socket(socket.AF_INET), s = ssl.wrap_socket(socket.socket(socket.AF_INET),
cert_reqs=ssl.CERT_NONE) cert_reqs=ssl.CERT_NONE)
s.connect(("svn.python.org", 443)) s.connect(("svn.python.org", 443))
@ -213,7 +213,7 @@ class NetworkedTests(unittest.TestCase):
os.read(fd, 0) os.read(fd, 0)
self.assertEqual(e.exception.errno, errno.EBADF) self.assertEqual(e.exception.errno, errno.EBADF)
def testNonBlockingHandshake(self): def test_non_blocking_handshake(self):
s = socket.socket(socket.AF_INET) s = socket.socket(socket.AF_INET)
s.connect(("svn.python.org", 443)) s.connect(("svn.python.org", 443))
s.setblocking(False) s.setblocking(False)
@ -237,8 +237,7 @@ class NetworkedTests(unittest.TestCase):
if support.verbose: if support.verbose:
sys.stdout.write("\nNeeded %d calls to do_handshake() to establish session.\n" % count) sys.stdout.write("\nNeeded %d calls to do_handshake() to establish session.\n" % count)
def testFetchServerCert(self): def test_get_server_certificate(self):
pem = ssl.get_server_certificate(("svn.python.org", 443)) pem = ssl.get_server_certificate(("svn.python.org", 443))
if not pem: if not pem:
self.fail("No server certificate on svn.python.org:443!") self.fail("No server certificate on svn.python.org:443!")
@ -289,7 +288,6 @@ try:
except ImportError: except ImportError:
_have_threads = False _have_threads = False
else: else:
_have_threads = True _have_threads = True
class ThreadedEchoServer(threading.Thread): class ThreadedEchoServer(threading.Thread):
@ -310,7 +308,7 @@ else:
threading.Thread.__init__(self) threading.Thread.__init__(self)
self.daemon = True self.daemon = True
def wrap_conn (self): def wrap_conn(self):
try: try:
self.sslconn = ssl.wrap_socket(self.sock, server_side=True, self.sslconn = ssl.wrap_socket(self.sock, server_side=True,
certfile=self.server.certificate, certfile=self.server.certificate,
@ -359,7 +357,7 @@ else:
else: else:
self.sock.close() self.sock.close()
def run (self): def run(self):
self.running = True self.running = True
if not self.server.starttls_server: if not self.server.starttls_server:
if not self.wrap_conn(): if not self.wrap_conn():
@ -367,28 +365,28 @@ else:
while self.running: while self.running:
try: try:
msg = self.read() msg = self.read()
amsg = (msg and str(msg, 'ASCII', 'strict')) or '' stripped = msg.strip()
if not msg: if not stripped:
# eof, so quit this handler # eof, so quit this handler
self.running = False self.running = False
self.close() self.close()
elif amsg.strip() == 'over': elif stripped == b'over':
if support.verbose and self.server.connectionchatty: if support.verbose and self.server.connectionchatty:
sys.stdout.write(" server: client closed connection\n") sys.stdout.write(" server: client closed connection\n")
self.close() self.close()
return return
elif (self.server.starttls_server and elif (self.server.starttls_server and
amsg.strip() == 'STARTTLS'): stripped == 'STARTTLS'):
if support.verbose and self.server.connectionchatty: if support.verbose and self.server.connectionchatty:
sys.stdout.write(" server: read STARTTLS from client, sending OK...\n") sys.stdout.write(" server: read STARTTLS from client, sending OK...\n")
self.write("OK\n".encode("ASCII", "strict")) self.write(b"OK\n")
if not self.wrap_conn(): if not self.wrap_conn():
return return
elif (self.server.starttls_server and self.sslconn elif (self.server.starttls_server and self.sslconn
and amsg.strip() == 'ENDTLS'): and stripped == 'ENDTLS'):
if support.verbose and self.server.connectionchatty: if support.verbose and self.server.connectionchatty:
sys.stdout.write(" server: read ENDTLS from client, sending OK...\n") sys.stdout.write(" server: read ENDTLS from client, sending OK...\n")
self.write("OK\n".encode("ASCII", "strict")) self.write(b"OK\n")
self.sock = self.sslconn.unwrap() self.sock = self.sslconn.unwrap()
self.sslconn = None self.sslconn = None
if support.verbose and self.server.connectionchatty: if support.verbose and self.server.connectionchatty:
@ -397,9 +395,9 @@ else:
if (support.verbose and if (support.verbose and
self.server.connectionchatty): self.server.connectionchatty):
ctype = (self.sslconn and "encrypted") or "unencrypted" ctype = (self.sslconn and "encrypted") or "unencrypted"
sys.stdout.write(" server: read %s (%s), sending back %s (%s)...\n" sys.stdout.write(" server: read %r (%s), sending back %r (%s)...\n"
% (repr(msg), ctype, repr(msg.lower()), ctype)) % (msg, ctype, msg.lower(), ctype))
self.write(amsg.lower().encode('ASCII', 'strict')) self.write(msg.lower())
except socket.error: except socket.error:
if self.server.chatty: if self.server.chatty:
handle_error("Test server failure:\n") handle_error("Test server failure:\n")
@ -432,11 +430,11 @@ else:
threading.Thread.__init__(self) threading.Thread.__init__(self)
self.daemon = True self.daemon = True
def start (self, flag=None): def start(self, flag=None):
self.flag = flag self.flag = flag
threading.Thread.start(self) threading.Thread.start(self)
def run (self): def run(self):
self.sock.settimeout(0.05) self.sock.settimeout(0.05)
self.sock.listen(5) self.sock.listen(5)
self.active = True self.active = True
@ -457,7 +455,7 @@ else:
self.stop() self.stop()
self.sock.close() self.sock.close()
def stop (self): def stop(self):
self.active = False self.active = False
class OurHTTPSServer(threading.Thread): class OurHTTPSServer(threading.Thread):
@ -467,12 +465,9 @@ else:
class HTTPSServer(HTTPServer): class HTTPSServer(HTTPServer):
def __init__(self, server_address, RequestHandlerClass, certfile): def __init__(self, server_address, RequestHandlerClass, certfile):
HTTPServer.__init__(self, server_address, RequestHandlerClass) HTTPServer.__init__(self, server_address, RequestHandlerClass)
# we assume the certfile contains both private key and certificate # we assume the certfile contains both private key and certificate
self.certfile = certfile self.certfile = certfile
self.active = False
self.active_lock = threading.Lock()
self.allow_reuse_address = True self.allow_reuse_address = True
def __str__(self): def __str__(self):
@ -481,7 +476,7 @@ else:
self.server_name, self.server_name,
self.server_port)) self.server_port))
def get_request (self): def get_request(self):
# override this to wrap socket with SSL # override this to wrap socket with SSL
sock, addr = self.socket.accept() sock, addr = self.socket.accept()
sslconn = ssl.wrap_socket(sock, server_side=True, sslconn = ssl.wrap_socket(sock, server_side=True,
@ -489,7 +484,6 @@ else:
return sslconn, addr return sslconn, addr
class RootedHTTPRequestHandler(SimpleHTTPRequestHandler): class RootedHTTPRequestHandler(SimpleHTTPRequestHandler):
# need to override translate_path to get a known root, # need to override translate_path to get a known root,
# instead of using os.curdir, since the test could be # instead of using os.curdir, since the test could be
# run from anywhere # run from anywhere
@ -520,7 +514,6 @@ else:
return path return path
def log_message(self, format, *args): def log_message(self, format, *args):
# we override this to suppress logging unless "verbose" # we override this to suppress logging unless "verbose"
if support.verbose: if support.verbose:
@ -534,7 +527,6 @@ else:
def __init__(self, certfile): def __init__(self, certfile):
self.flag = None self.flag = None
self.active = False
self.RootedHTTPRequestHandler.root = os.path.split(CERTFILE)[0] self.RootedHTTPRequestHandler.root = os.path.split(CERTFILE)[0]
self.server = self.HTTPSServer( self.server = self.HTTPSServer(
(HOST, 0), self.RootedHTTPRequestHandler, certfile) (HOST, 0), self.RootedHTTPRequestHandler, certfile)
@ -545,19 +537,16 @@ else:
def __str__(self): def __str__(self):
return "<%s %s>" % (self.__class__.__name__, self.server) return "<%s %s>" % (self.__class__.__name__, self.server)
def start (self, flag=None): def start(self, flag=None):
self.flag = flag self.flag = flag
threading.Thread.start(self) threading.Thread.start(self)
def run (self): def run(self):
self.active = True
if self.flag: if self.flag:
self.flag.set() self.flag.set()
self.server.serve_forever(0.05) self.server.serve_forever(0.05)
self.active = False
def stop (self): def stop(self):
self.active = False
self.server.shutdown() self.server.shutdown()
@ -609,7 +598,7 @@ else:
if not data: if not data:
self.close() self.close()
else: else:
self.send(str(data, 'ASCII', 'strict').lower().encode('ASCII', 'strict')) self.send(data.lower())
def handle_close(self): def handle_close(self):
self.close() self.close()
@ -650,7 +639,7 @@ else:
self.flag = flag self.flag = flag
threading.Thread.start(self) threading.Thread.start(self)
def run (self): def run(self):
self.active = True self.active = True
if self.flag: if self.flag:
self.flag.set() self.flag.set()
@ -660,11 +649,15 @@ else:
except: except:
pass pass
def stop (self): def stop(self):
self.active = False self.active = False
self.server.close() self.server.close()
def badCertTest (certfile): 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, server = ThreadedEchoServer(CERTFILE,
certreqs=ssl.CERT_REQUIRED, certreqs=ssl.CERT_REQUIRED,
cacerts=CERTFILE, chatty=False, cacerts=CERTFILE, chatty=False,
@ -684,7 +677,7 @@ else:
if support.verbose: if support.verbose:
sys.stdout.write("\nSSLError is %s\n" % x.args[1]) sys.stdout.write("\nSSLError is %s\n" % x.args[1])
except socket.error as x: except socket.error as x:
if test_support.verbose: if support.verbose:
sys.stdout.write("\nsocket.error is %s\n" % x[1]) sys.stdout.write("\nsocket.error is %s\n" % x[1])
else: else:
self.fail("Use of invalid cert should have failed!") self.fail("Use of invalid cert should have failed!")
@ -692,11 +685,13 @@ else:
server.stop() server.stop()
server.join() server.join()
def serverParamsTest (certfile, protocol, certreqs, cacertsfile, def server_params_test(certfile, protocol, certreqs, cacertsfile,
client_certfile, client_protocol=None, client_certfile, client_protocol=None, indata=b"FOO\n",
indata="FOO\n", ciphers=None, chatty=True, connectionchatty=False):
ciphers=None, chatty=False, connectionchatty=False): """
Launch a server, connect a client to it and try various reads
and writes.
"""
server = ThreadedEchoServer(certfile, server = ThreadedEchoServer(certfile,
certreqs=certreqs, certreqs=certreqs,
ssl_version=protocol, ssl_version=protocol,
@ -719,24 +714,22 @@ else:
cert_reqs=certreqs, cert_reqs=certreqs,
ssl_version=client_protocol) ssl_version=client_protocol)
s.connect((HOST, server.port)) s.connect((HOST, server.port))
bindata = indata.encode('ASCII', 'strict') for arg in [indata, bytearray(indata), memoryview(indata)]:
for arg in [bindata, bytearray(bindata), memoryview(bindata)]:
if connectionchatty: if connectionchatty:
if support.verbose: if support.verbose:
sys.stdout.write( sys.stdout.write(
" client: sending %s...\n" % (repr(indata))) " client: sending %r...\n" % indata)
s.write(arg) s.write(arg)
outdata = s.read() outdata = s.read()
if connectionchatty: if connectionchatty:
if support.verbose: if support.verbose:
sys.stdout.write(" client: read %s\n" % repr(outdata)) sys.stdout.write(" client: read %r\n" % outdata)
outdata = str(outdata, 'ASCII', 'strict')
if outdata != indata.lower(): if outdata != indata.lower():
self.fail( self.fail(
"bad data <<%s>> (%d) received; expected <<%s>> (%d)\n" "bad data <<%r>> (%d) received; expected <<%r>> (%d)\n"
% (repr(outdata[:min(len(outdata),20)]), len(outdata), % (outdata[:20], len(outdata),
repr(indata[:min(len(indata),20)].lower()), len(indata))) indata[:20].lower(), len(indata)))
s.write("over\n".encode("ASCII", "strict")) s.write(b"over\n")
if connectionchatty: if connectionchatty:
if support.verbose: if support.verbose:
sys.stdout.write(" client: closing connection.\n") sys.stdout.write(" client: closing connection.\n")
@ -745,22 +738,19 @@ else:
server.stop() server.stop()
server.join() server.join()
def tryProtocolCombo (server_protocol, def try_protocol_combo(server_protocol,
client_protocol, client_protocol,
expectedToWork, expect_success,
certsreqs=None): certsreqs=None):
if certsreqs is None: if certsreqs is None:
certsreqs = ssl.CERT_NONE certsreqs = ssl.CERT_NONE
certtype = {
if certsreqs == ssl.CERT_NONE: ssl.CERT_NONE: "CERT_NONE",
certtype = "CERT_NONE" ssl.CERT_OPTIONAL: "CERT_OPTIONAL",
elif certsreqs == ssl.CERT_OPTIONAL: ssl.CERT_REQUIRED: "CERT_REQUIRED",
certtype = "CERT_OPTIONAL" }[certsreqs]
elif certsreqs == ssl.CERT_REQUIRED:
certtype = "CERT_REQUIRED"
if support.verbose: if support.verbose:
formatstr = (expectedToWork and " %s->%s %s\n") or " {%s->%s} %s\n" formatstr = (expect_success and " %s->%s %s\n") or " {%s->%s} %s\n"
sys.stdout.write(formatstr % sys.stdout.write(formatstr %
(ssl.get_protocol_name(client_protocol), (ssl.get_protocol_name(client_protocol),
ssl.get_protocol_name(server_protocol), ssl.get_protocol_name(server_protocol),
@ -769,20 +759,20 @@ else:
# NOTE: we must enable "ALL" ciphers, otherwise an SSLv23 client # NOTE: we must enable "ALL" ciphers, otherwise an SSLv23 client
# will send an SSLv3 hello (rather than SSLv2) starting from # will send an SSLv3 hello (rather than SSLv2) starting from
# OpenSSL 1.0.0 (see issue #8322). # OpenSSL 1.0.0 (see issue #8322).
serverParamsTest(CERTFILE, server_protocol, certsreqs, server_params_test(CERTFILE, server_protocol, certsreqs,
CERTFILE, CERTFILE, client_protocol, CERTFILE, CERTFILE, client_protocol,
ciphers="ALL", ciphers="ALL", chatty=False,
chatty=False, connectionchatty=False) connectionchatty=False)
# Protocol mismatch can result in either an SSLError, or a # Protocol mismatch can result in either an SSLError, or a
# "Connection reset by peer" error. # "Connection reset by peer" error.
except ssl.SSLError: except ssl.SSLError:
if expectedToWork: if expect_success:
raise raise
except socket.error as e: except socket.error as e:
if expectedToWork or e.errno != errno.ECONNRESET: if expect_success or e.errno != errno.ECONNRESET:
raise raise
else: else:
if not expectedToWork: if not expect_success:
self.fail( self.fail(
"Client protocol %s succeeded with server protocol %s!" "Client protocol %s succeeded with server protocol %s!"
% (ssl.get_protocol_name(client_protocol), % (ssl.get_protocol_name(client_protocol),
@ -791,16 +781,15 @@ else:
class ThreadedTests(unittest.TestCase): class ThreadedTests(unittest.TestCase):
def testEcho (self): def test_echo(self):
"""Basic test of an SSL client connecting to a server"""
if support.verbose: if support.verbose:
sys.stdout.write("\n") sys.stdout.write("\n")
serverParamsTest(CERTFILE, ssl.PROTOCOL_TLSv1, ssl.CERT_NONE, server_params_test(CERTFILE, ssl.PROTOCOL_TLSv1, ssl.CERT_NONE,
CERTFILE, CERTFILE, ssl.PROTOCOL_TLSv1, CERTFILE, CERTFILE, ssl.PROTOCOL_TLSv1,
chatty=True, connectionchatty=True) chatty=True, connectionchatty=True)
def testReadCert(self):
def test_getpeercert(self):
if support.verbose: if support.verbose:
sys.stdout.write("\n") sys.stdout.write("\n")
s2 = socket.socket() s2 = socket.socket()
@ -840,23 +829,30 @@ else:
server.stop() server.stop()
server.join() server.join()
def testNULLcert(self): def test_empty_cert(self):
badCertTest(os.path.join(os.path.dirname(__file__) or os.curdir, """Connecting with an empty cert file"""
"nullcert.pem")) bad_cert_test(os.path.join(os.path.dirname(__file__) or os.curdir,
def testMalformedCert(self): "nullcert.pem"))
badCertTest(os.path.join(os.path.dirname(__file__) or os.curdir, def test_malformed_cert(self):
"badcert.pem")) """Connecting with a badly formatted certificate (syntax error)"""
def testWrongCert(self): bad_cert_test(os.path.join(os.path.dirname(__file__) or os.curdir,
badCertTest(os.path.join(os.path.dirname(__file__) or os.curdir, "badcert.pem"))
"wrongcert.pem")) def test_nonexisting_cert(self):
def testMalformedKey(self): """Connecting with a non-existing cert file"""
badCertTest(os.path.join(os.path.dirname(__file__) or os.curdir, bad_cert_test(os.path.join(os.path.dirname(__file__) or os.curdir,
"badkey.pem")) "wrongcert.pem"))
def test_malformed_key(self):
def testRudeShutdown(self): """Connecting with a badly formatted key (syntax error)"""
bad_cert_test(os.path.join(os.path.dirname(__file__) or os.curdir,
"badkey.pem"))
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_ready = threading.Event()
listener_gone = threading.Event() listener_gone = threading.Event()
s = socket.socket() s = socket.socket()
port = support.bind_port(s, HOST) port = support.bind_port(s, HOST)
@ -890,62 +886,66 @@ else:
finally: finally:
t.join() t.join()
def testProtocolSSL2(self): def test_protocol_sslv2(self):
"""Connecting to an SSLv2 server with various client options"""
if support.verbose: if support.verbose:
sys.stdout.write("\n") sys.stdout.write("\n")
tryProtocolCombo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv2, True) try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv2, True)
tryProtocolCombo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv2, True, ssl.CERT_OPTIONAL) try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv2, True, ssl.CERT_OPTIONAL)
tryProtocolCombo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv2, True, ssl.CERT_REQUIRED) try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv2, True, ssl.CERT_REQUIRED)
tryProtocolCombo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv23, True) try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv23, True)
tryProtocolCombo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv3, False) try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv3, False)
tryProtocolCombo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_TLSv1, False) try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_TLSv1, False)
def testProtocolSSL23(self): def test_protocol_sslv23(self):
"""Connecting to an SSLv23 server with various client options"""
if support.verbose: if support.verbose:
sys.stdout.write("\n") sys.stdout.write("\n")
try: try:
tryProtocolCombo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv2, True) try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv2, True)
except (ssl.SSLError, socket.error) as x: except (ssl.SSLError, socket.error) as x:
# this fails on some older versions of OpenSSL (0.9.7l, for instance) # this fails on some older versions of OpenSSL (0.9.7l, for instance)
if support.verbose: if support.verbose:
sys.stdout.write( sys.stdout.write(
" SSL2 client to SSL23 server test unexpectedly failed:\n %s\n" " SSL2 client to SSL23 server test unexpectedly failed:\n %s\n"
% str(x)) % str(x))
tryProtocolCombo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv3, True) try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv3, True)
tryProtocolCombo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv23, True) try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv23, True)
tryProtocolCombo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_TLSv1, True) try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_TLSv1, True)
tryProtocolCombo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv3, True, ssl.CERT_OPTIONAL) try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv3, True, ssl.CERT_OPTIONAL)
tryProtocolCombo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv23, True, ssl.CERT_OPTIONAL) try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv23, True, ssl.CERT_OPTIONAL)
tryProtocolCombo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_TLSv1, True, ssl.CERT_OPTIONAL) try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_TLSv1, True, ssl.CERT_OPTIONAL)
tryProtocolCombo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv3, True, ssl.CERT_REQUIRED) try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv3, True, ssl.CERT_REQUIRED)
tryProtocolCombo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv23, True, ssl.CERT_REQUIRED) try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv23, True, ssl.CERT_REQUIRED)
tryProtocolCombo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_TLSv1, True, ssl.CERT_REQUIRED) try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_TLSv1, True, ssl.CERT_REQUIRED)
def testProtocolSSL3(self): def test_protocol_sslv3(self):
"""Connecting to an SSLv3 server with various client options"""
if support.verbose: if support.verbose:
sys.stdout.write("\n") sys.stdout.write("\n")
tryProtocolCombo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, True) try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, True)
tryProtocolCombo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, True, ssl.CERT_OPTIONAL) try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, True, ssl.CERT_OPTIONAL)
tryProtocolCombo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, True, ssl.CERT_REQUIRED) try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, True, ssl.CERT_REQUIRED)
tryProtocolCombo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv2, False) try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv2, False)
tryProtocolCombo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv23, False) try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv23, False)
tryProtocolCombo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_TLSv1, False) try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_TLSv1, False)
def testProtocolTLS1(self): def test_protocol_tlsv1(self):
"""Connecting to a TLSv1 server with various client options"""
if support.verbose: if support.verbose:
sys.stdout.write("\n") sys.stdout.write("\n")
tryProtocolCombo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, True) try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, True)
tryProtocolCombo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, True, ssl.CERT_OPTIONAL) try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, True, ssl.CERT_OPTIONAL)
tryProtocolCombo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, True, ssl.CERT_REQUIRED) try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, True, ssl.CERT_REQUIRED)
tryProtocolCombo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_SSLv2, False) try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_SSLv2, False)
tryProtocolCombo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_SSLv3, False) try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_SSLv3, False)
tryProtocolCombo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_SSLv23, False) try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_SSLv23, False)
def testSTARTTLS (self): 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") msgs = (b"msg 1", b"MSG 2", b"STARTTLS", b"MSG 3", b"msg 4", b"ENDTLS", b"msg 5", b"msg 6")
server = ThreadedEchoServer(CERTFILE, server = ThreadedEchoServer(CERTFILE,
ssl_version=ssl.PROTOCOL_TLSv1, ssl_version=ssl.PROTOCOL_TLSv1,
@ -965,45 +965,42 @@ else:
if support.verbose: if support.verbose:
sys.stdout.write("\n") sys.stdout.write("\n")
for indata in msgs: for indata in msgs:
msg = indata.encode('ASCII', 'replace')
if support.verbose: if support.verbose:
sys.stdout.write( sys.stdout.write(
" client: sending %s...\n" % repr(msg)) " client: sending %r...\n" % indata)
if wrapped: if wrapped:
conn.write(msg) conn.write(indata)
outdata = conn.read() outdata = conn.read()
else: else:
s.send(msg) s.send(indata)
outdata = s.recv(1024) outdata = s.recv(1024)
if (indata == "STARTTLS" and msg = outdata.strip().lower()
str(outdata, 'ASCII', 'replace').strip().lower().startswith("ok")): if indata == b"STARTTLS" and msg.startswith(b"ok"):
# STARTTLS ok, switch to secure mode
if support.verbose: if support.verbose:
msg = str(outdata, 'ASCII', 'replace')
sys.stdout.write( sys.stdout.write(
" client: read %s from server, starting TLS...\n" " client: read %r from server, starting TLS...\n"
% repr(msg)) % msg)
conn = ssl.wrap_socket(s, ssl_version=ssl.PROTOCOL_TLSv1) conn = ssl.wrap_socket(s, ssl_version=ssl.PROTOCOL_TLSv1)
wrapped = True wrapped = True
elif (indata == "ENDTLS" and elif indata == b"ENDTLS" and msg.startswith(b"ok"):
str(outdata, 'ASCII', 'replace').strip().lower().startswith("ok")): # ENDTLS ok, switch back to clear text
if support.verbose: if support.verbose:
msg = str(outdata, 'ASCII', 'replace')
sys.stdout.write( sys.stdout.write(
" client: read %s from server, ending TLS...\n" " client: read %r from server, ending TLS...\n"
% repr(msg)) % msg)
s = conn.unwrap() s = conn.unwrap()
wrapped = False wrapped = False
else: else:
if support.verbose: if support.verbose:
msg = str(outdata, 'ASCII', 'replace')
sys.stdout.write( sys.stdout.write(
" client: read %s from server\n" % repr(msg)) " client: read %r from server\n" % msg)
if support.verbose: if support.verbose:
sys.stdout.write(" client: closing connection.\n") sys.stdout.write(" client: closing connection.\n")
if wrapped: if wrapped:
conn.write("over\n".encode("ASCII", "strict")) conn.write(b"over\n")
else: else:
s.send("over\n".encode("ASCII", "strict")) s.send(b"over\n")
if wrapped: if wrapped:
conn.close() conn.close()
else: else:
@ -1012,8 +1009,8 @@ else:
server.stop() server.stop()
server.join() server.join()
def testSocketServer(self): def test_socketserver(self):
"""Using a SocketServer to create and manage SSL connections."""
server = OurHTTPSServer(CERTFILE) server = OurHTTPSServer(CERTFILE)
flag = threading.Event() flag = threading.Event()
server.start(flag) server.start(flag)
@ -1023,7 +1020,8 @@ else:
try: try:
if support.verbose: if support.verbose:
sys.stdout.write('\n') sys.stdout.write('\n')
d1 = open(CERTFILE, 'rb').read() with open(CERTFILE, 'rb') as f:
d1 = f.read()
d2 = '' d2 = ''
# now fetch the same data from the HTTPS server # now fetch the same data from the HTTPS server
url = 'https://%s:%d/%s' % ( url = 'https://%s:%d/%s' % (
@ -1046,12 +1044,14 @@ else:
sys.stdout.write('joining thread\n') sys.stdout.write('joining thread\n')
server.join() server.join()
def testAsyncoreServer(self): def test_asyncore_server(self):
"""Check the example asyncore integration."""
indata = "TEST MESSAGE of mixed case\n"
if support.verbose: if support.verbose:
sys.stdout.write("\n") sys.stdout.write("\n")
indata="FOO\n" indata = b"FOO\n"
server = AsyncoreEchoServer(CERTFILE) server = AsyncoreEchoServer(CERTFILE)
flag = threading.Event() flag = threading.Event()
server.start(flag) server.start(flag)
@ -1063,18 +1063,17 @@ else:
s.connect(('127.0.0.1', server.port)) s.connect(('127.0.0.1', server.port))
if support.verbose: if support.verbose:
sys.stdout.write( sys.stdout.write(
" client: sending %s...\n" % (repr(indata))) " client: sending %r...\n" % indata)
s.write(indata.encode('ASCII', 'strict')) s.write(indata)
outdata = s.read() outdata = s.read()
if support.verbose: if support.verbose:
sys.stdout.write(" client: read %s\n" % repr(outdata)) sys.stdout.write(" client: read %r\n" % outdata)
outdata = str(outdata, 'ASCII', 'strict')
if outdata != indata.lower(): if outdata != indata.lower():
self.fail( self.fail(
"bad data <<%s>> (%d) received; expected <<%s>> (%d)\n" "bad data <<%r>> (%d) received; expected <<%r>> (%d)\n"
% (outdata[:min(len(outdata),20)], len(outdata), % (outdata[:20], len(outdata),
indata[:min(len(indata),20)].lower(), len(indata))) indata[:20].lower(), len(indata)))
s.write("over\n".encode("ASCII", "strict")) s.write(b"over\n")
if support.verbose: if support.verbose:
sys.stdout.write(" client: closing connection.\n") sys.stdout.write(" client: closing connection.\n")
s.close() s.close()
@ -1082,8 +1081,8 @@ else:
server.stop() server.stop()
server.join() server.join()
def testAllRecvAndSendMethods(self): def test_recv_send(self):
"""Test recv(), send() and friends."""
if support.verbose: if support.verbose:
sys.stdout.write("\n") sys.stdout.write("\n")
@ -1132,19 +1131,18 @@ else:
data_prefix = "PREFIX_" data_prefix = "PREFIX_"
for meth_name, send_meth, expect_success, args in send_methods: for meth_name, send_meth, expect_success, args in send_methods:
indata = data_prefix + meth_name indata = (data_prefix + meth_name).encode('ascii')
try: try:
send_meth(indata.encode('ASCII', 'strict'), *args) send_meth(indata, *args)
outdata = s.read() outdata = s.read()
outdata = str(outdata, 'ASCII', 'strict')
if outdata != indata.lower(): if outdata != indata.lower():
self.fail( self.fail(
"While sending with <<{name:s}>> bad data " "While sending with <<{name:s}>> bad data "
"<<{outdata:s}>> ({nout:d}) received; " "<<{outdata:r}>> ({nout:d}) received; "
"expected <<{indata:s}>> ({nin:d})\n".format( "expected <<{indata:r}>> ({nin:d})\n".format(
name=meth_name, outdata=repr(outdata[:20]), name=meth_name, outdata=outdata[:20],
nout=len(outdata), nout=len(outdata),
indata=repr(indata[:20]), nin=len(indata) indata=indata[:20], nin=len(indata)
) )
) )
except ValueError as e: except ValueError as e:
@ -1162,19 +1160,18 @@ else:
) )
for meth_name, recv_meth, expect_success, args in recv_methods: for meth_name, recv_meth, expect_success, args in recv_methods:
indata = data_prefix + meth_name indata = (data_prefix + meth_name).encode('ascii')
try: try:
s.send(indata.encode('ASCII', 'strict')) s.send(indata)
outdata = recv_meth(*args) outdata = recv_meth(*args)
outdata = str(outdata, 'ASCII', 'strict')
if outdata != indata.lower(): if outdata != indata.lower():
self.fail( self.fail(
"While receiving with <<{name:s}>> bad data " "While receiving with <<{name:s}>> bad data "
"<<{outdata:s}>> ({nout:d}) received; " "<<{outdata:r}>> ({nout:d}) received; "
"expected <<{indata:s}>> ({nin:d})\n".format( "expected <<{indata:r}>> ({nin:d})\n".format(
name=meth_name, outdata=repr(outdata[:20]), name=meth_name, outdata=outdata[:20],
nout=len(outdata), nout=len(outdata),
indata=repr(indata[:20]), nin=len(indata) indata=indata[:20], nin=len(indata)
) )
) )
except ValueError as e: except ValueError as e:
@ -1193,7 +1190,7 @@ else:
# consume data # consume data
s.read() s.read()
s.write("over\n".encode("ASCII", "strict")) s.write(b"over\n")
s.close() s.close()
finally: finally:
server.stop() server.stop()
@ -1272,10 +1269,11 @@ def test_main(verbose=False):
if thread_info and support.is_resource_enabled('network'): if thread_info and support.is_resource_enabled('network'):
tests.append(ThreadedTests) tests.append(ThreadedTests)
support.run_unittest(*tests) try:
support.run_unittest(*tests)
if _have_threads: finally:
support.threading_cleanup(*thread_info) if _have_threads:
support.threading_cleanup(*thread_info)
if __name__ == "__main__": if __name__ == "__main__":
test_main() test_main()