Skip to content

C
cpython
Kaydet (Commit) e472933e authored tarafından Guido van Rossum's avatar Guido van Rossum
Dosyalara gözat
Seçenekler

Bill Janssen wrote: 

Here's a patch which makes test_ssl a better player in the buildbots
environment.  I deep-ended on "try-except-else" clauses.
üst 7fc8e299
Hide whitespace changes
Inline Side-by-side
Showing with 87 additions and 43 deletions
Lib/ssl.py
Dosyayı görüntüle @ e472933e
...@@ -100,12 +100,13 @@ class sslsocket (socket): ...@@ -100,12 +100,13 @@ class sslsocket (socket):
# see if it's connected # see if it's connected
try: try:
socket.getpeername(self) socket.getpeername(self)
# yes
self._sslobj = _ssl.sslwrap(self._sock, 0, keyfile, certfile,
cert_reqs, ssl_version, ca_certs)
except: except:
# no # no, no connection yet
self._sslobj = None self._sslobj = None
else:
# yes, create the SSL object
self._sslobj = _ssl.sslwrap(self._sock, 0, keyfile, certfile,
cert_reqs, ssl_version, ca_certs)
self.keyfile = keyfile self.keyfile = keyfile
self.certfile = certfile self.certfile = certfile
self.cert_reqs = cert_reqs self.cert_reqs = cert_reqs
......
Lib/test/test_ssl.py
Dosyayı görüntüle @ e472933e
...@@ -91,38 +91,66 @@ class ConnectedTests(unittest.TestCase): ...@@ -91,38 +91,66 @@ class ConnectedTests(unittest.TestCase):
def testTLSecho (self): def testTLSecho (self):
s1 = socket.socket() s1 = socket.socket()
s1.connect(('127.0.0.1', 10024)) try:
c1 = ssl.sslsocket(s1, ssl_version=ssl.PROTOCOL_TLSv1) s1.connect(('127.0.0.1', 10024))
indata = "FOO\n" except:
c1.write(indata) sys.stdout.write("connection failure:\n" + string.join(
outdata = c1.read() traceback.format_exception(*sys.exc_info())))
if outdata != indata.lower(): raise test_support.TestFailed("Can't connect to test server")
sys.stderr.write("bad data <<%s>> received\n" % data) else:
c1.close() try:
c1 = ssl.sslsocket(s1, ssl_version=ssl.PROTOCOL_TLSv1)
except:
sys.stdout.write("SSL handshake failure:\n" + string.join(
traceback.format_exception(*sys.exc_info())))
raise test_support.TestFailed("Can't SSL-handshake with test server")
else:
if not c1:
raise test_support.TestFailed("Can't SSL-handshake with test server")
indata = "FOO\n"
c1.write(indata)
outdata = c1.read()
if outdata != indata.lower():
raise test_support.TestFailed("bad data <<%s>> received; expected <<%s>>\n" % (data, indata.lower()))
c1.close()
def testReadCert(self): def testReadCert(self):
s2 = socket.socket() s2 = socket.socket()
s2.connect(('127.0.0.1', 10024)) try:
c2 = ssl.sslsocket(s2, ssl_version=ssl.PROTOCOL_TLSv1, s2.connect(('127.0.0.1', 10024))
cert_reqs=ssl.CERT_REQUIRED, ca_certs=CERTFILE) except:
cert = c2.getpeercert() sys.stdout.write("connection failure:\n" + string.join(
if not cert: traceback.format_exception(*sys.exc_info())))
raise test_support.TestFailed("Can't get peer certificate.") raise test_support.TestFailed("Can't connect to test server")
if not cert.has_key('subject'): else:
raise test_support.TestFailed( try:
"No subject field in certificate: %s." % c2 = ssl.sslsocket(s2, ssl_version=ssl.PROTOCOL_TLSv1,
pprint.pformat(cert)) cert_reqs=ssl.CERT_REQUIRED, ca_certs=CERTFILE)
if not (cert['subject'].has_key('organizationName')): except:
raise test_support.TestFailed( sys.stdout.write("SSL handshake failure:\n" + string.join(
"No 'organizationName' field in certificate subject: %s." % traceback.format_exception(*sys.exc_info())))
pprint.pformat(cert)) raise test_support.TestFailed("Can't SSL-handshake with test server")
if (cert['subject']['organizationName'] != else:
"Python Software Foundation"): if not c2:
raise test_support.TestFailed( raise test_support.TestFailed("Can't SSL-handshake with test server")
"Invalid 'organizationName' field in certificate subject; " cert = c2.getpeercert()
"should be 'Python Software Foundation'."); if not cert:
c2.close() raise test_support.TestFailed("Can't get peer certificate.")
if not cert.has_key('subject'):
raise test_support.TestFailed(
"No subject field in certificate: %s." %
pprint.pformat(cert))
if not (cert['subject'].has_key('organizationName')):
raise test_support.TestFailed(
"No 'organizationName' field in certificate subject: %s." %
pprint.pformat(cert))
if (cert['subject']['organizationName'] !=
"Python Software Foundation"):
raise test_support.TestFailed(
"Invalid 'organizationName' field in certificate subject; "
"should be 'Python Software Foundation'.");
c2.close()
class threadedEchoServer(threading.Thread): class threadedEchoServer(threading.Thread):
...@@ -138,10 +166,22 @@ class threadedEchoServer(threading.Thread): ...@@ -138,10 +166,22 @@ class threadedEchoServer(threading.Thread):
def run (self): def run (self):
self.running = True self.running = True
sslconn = ssl.sslsocket(self.sock, server_side=True, try:
certfile=self.server.certificate, sslconn = ssl.sslsocket(self.sock, server_side=True,
ssl_version=self.server.protocol, certfile=self.server.certificate,
cert_reqs=self.server.certreqs) ssl_version=self.server.protocol,
cert_reqs=self.server.certreqs)
except:
# here, we want to stop the server, because this shouldn't
# happen in the context of our test case
sys.stdout.write("Test server failure:\n" + string.join(
traceback.format_exception(*sys.exc_info())))
self.running = False
# normally, we'd just stop here, but for the test
# harness, we want to stop the server
self.server.stop()
return
while self.running: while self.running:
try: try:
msg = sslconn.read() msg = sslconn.read()
...@@ -154,15 +194,18 @@ class threadedEchoServer(threading.Thread): ...@@ -154,15 +194,18 @@ class threadedEchoServer(threading.Thread):
self.server.stop() self.server.stop()
self.running = False self.running = False
else: else:
# print "server:", msg.strip().lower() sys.stdout.write("\nserver: %s\n" % msg.strip().lower())
sslconn.write(msg.lower()) sslconn.write(msg.lower())
except ssl.sslerror: except ssl.sslerror:
sys.stderr.write(string.join( sys.stdout.write("Test server failure:\n" + string.join(
traceback.format_exception(*sys.exc_info()))) traceback.format_exception(*sys.exc_info())))
sslconn.close() sslconn.close()
self.running = False self.running = False
# normally, we'd just stop here, but for the test
# harness, we want to stop the server
self.server.stop()
except: except:
sys.stderr.write(string.join( sys.stdout.write(string.join(
traceback.format_exception(*sys.exc_info()))) traceback.format_exception(*sys.exc_info())))
def __init__(self, port, certificate, ssl_version=None, def __init__(self, port, certificate, ssl_version=None,
...@@ -192,20 +235,20 @@ class threadedEchoServer(threading.Thread): ...@@ -192,20 +235,20 @@ class threadedEchoServer(threading.Thread):
while self.active: while self.active:
try: try:
newconn, connaddr = self.sock.accept() newconn, connaddr = self.sock.accept()
# sys.stderr.write('new connection from ' + str(connaddr)) sys.stdout.write('\nserver: new connection from ' + str(connaddr) + '\n')
handler = self.connectionHandler(self, newconn) handler = self.connectionHandler(self, newconn)
handler.start() handler.start()
except socket.timeout: except socket.timeout:
pass pass
except KeyboardInterrupt: except KeyboardInterrupt:
self.active = False self.stop()
except: except:
sys.stderr.write(string.join( sys.stdout.write("Test server failure:\n" + string.join(
traceback.format_exception(*sys.exc_info()))) traceback.format_exception(*sys.exc_info())))
def stop (self): def stop (self):
self.active = False self.active = False
self.sock.close()
CERTFILE_CONFIG_TEMPLATE = """ CERTFILE_CONFIG_TEMPLATE = """
# create RSA certs - Server # create RSA certs - Server
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment