Server-side SSL and certificate validation, by Bill Janssen.
While cleaning up Bill's C style, I may have cleaned up some code
he didn't touch as well (in _ssl.c).
diff --git a/Lib/test/test_ssl.py b/Lib/test/test_ssl.py
new file mode 100644
index 0000000..0188dc2
--- /dev/null
+++ b/Lib/test/test_ssl.py
@@ -0,0 +1,304 @@
+# Test the support for SSL and sockets
+
+import sys
+import unittest
+from test import test_support
+import socket
+import errno
+import threading
+import subprocess
+import time
+import os
+import pprint
+import urllib
+import shutil
+import string
+import traceback
+
+# Optionally test SSL support, if we have it in the tested platform
+skip_expected = False
+try:
+ import ssl
+except ImportError:
+ skip_expected = True
+
+CERTFILE = None
+GMAIL_POP_CERTFILE = None
+
+class BasicTests(unittest.TestCase):
+
+ def testRudeShutdown(self):
+ # Some random port to connect to.
+ PORT = [9934]
+
+ listener_ready = threading.Event()
+ listener_gone = threading.Event()
+
+ # `listener` runs in a thread. It opens a socket listening on
+ # PORT, and 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 = socket.socket()
+ PORT[0] = test_support.bind_port(s, '', PORT[0])
+ s.listen(5)
+ listener_ready.set()
+ s.accept()
+ s = None # reclaim the socket object, which also closes it
+ listener_gone.set()
+
+ def connector():
+ listener_ready.wait()
+ s = socket.socket()
+ s.connect(('localhost', PORT[0]))
+ listener_gone.wait()
+ try:
+ ssl_sock = socket.ssl(s)
+ except socket.sslerror:
+ pass
+ else:
+ raise test_support.TestFailed(
+ 'connecting to closed SSL socket should have failed')
+
+ t = threading.Thread(target=listener)
+ t.start()
+ connector()
+ t.join()
+
+ def testSSLconnect(self):
+ import os
+ with test_support.transient_internet():
+ s = ssl.sslsocket(socket.socket(socket.AF_INET),
+ cert_reqs=ssl.CERT_NONE)
+ s.connect(("pop.gmail.com", 995))
+ c = s.getpeercert()
+ if c:
+ raise test_support.TestFailed("Peer cert %s shouldn't be here!")
+ s.close()
+
+ # this should fail because we have no verification certs
+ s = ssl.sslsocket(socket.socket(socket.AF_INET),
+ cert_reqs=ssl.CERT_REQUIRED)
+ try:
+ s.connect(("pop.gmail.com", 995))
+ except ssl.sslerror:
+ pass
+ finally:
+ s.close()
+
+class ConnectedTests(unittest.TestCase):
+
+ def testTLSecho (self):
+
+ s1 = socket.socket()
+ s1.connect(('127.0.0.1', 10024))
+ c1 = ssl.sslsocket(s1, ssl_version=ssl.PROTOCOL_TLSv1)
+ indata = "FOO\n"
+ c1.write(indata)
+ outdata = c1.read()
+ if outdata != indata.lower():
+ sys.stderr.write("bad data <<%s>> received\n" % data)
+ c1.close()
+
+ def testReadCert(self):
+
+ s2 = socket.socket()
+ s2.connect(('127.0.0.1', 10024))
+ c2 = ssl.sslsocket(s2, ssl_version=ssl.PROTOCOL_TLSv1,
+ cert_reqs=ssl.CERT_REQUIRED, ca_certs=CERTFILE)
+ cert = c2.getpeercert()
+ if not cert:
+ 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 connectionHandler(threading.Thread):
+
+ def __init__(self, server, connsock):
+ self.server = server
+ self.running = False
+ self.sock = connsock
+ threading.Thread.__init__(self)
+ self.setDaemon(True)
+
+ def run (self):
+ self.running = True
+ sslconn = ssl.sslsocket(self.sock, server_side=True,
+ certfile=self.server.certificate,
+ ssl_version=self.server.protocol,
+ cert_reqs=self.server.certreqs)
+ while self.running:
+ try:
+ msg = sslconn.read()
+ if not msg:
+ # eof, so quit this handler
+ self.running = False
+ sslconn.close()
+ elif msg.strip() == 'over':
+ sslconn.close()
+ self.server.stop()
+ self.running = False
+ else:
+ # print "server:", msg.strip().lower()
+ sslconn.write(msg.lower())
+ except ssl.sslerror:
+ sys.stderr.write(string.join(
+ traceback.format_exception(*sys.exc_info())))
+ sslconn.close()
+ self.running = False
+ except:
+ sys.stderr.write(string.join(
+ traceback.format_exception(*sys.exc_info())))
+
+ def __init__(self, port, certificate, ssl_version=ssl.PROTOCOL_TLSv1,
+ certreqs=ssl.CERT_NONE, cacerts=None):
+ self.certificate = certificate
+ self.protocol = ssl_version
+ self.certreqs = certreqs
+ self.cacerts = cacerts
+ self.sock = socket.socket()
+ self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
+ self.sock.bind(('127.0.0.1', port))
+ self.active = False
+ threading.Thread.__init__(self)
+ self.setDaemon(False)
+
+ def run (self):
+ self.sock.settimeout(0.5)
+ self.sock.listen(5)
+ self.active = True
+ while self.active:
+ try:
+ newconn, connaddr = self.sock.accept()
+ # sys.stderr.write('new connection from ' + str(connaddr))
+ handler = self.connectionHandler(self, newconn)
+ handler.start()
+ except socket.timeout:
+ pass
+ except KeyboardInterrupt:
+ self.active = False
+ except:
+ sys.stderr.write(string.join(
+ traceback.format_exception(*sys.exc_info())))
+
+ def stop (self):
+ self.active = False
+
+
+CERTFILE_CONFIG_TEMPLATE = """
+# create RSA certs - Server
+
+[ req ]
+default_bits = 1024
+encrypt_key = yes
+distinguished_name = req_dn
+x509_extensions = cert_type
+
+[ req_dn ]
+countryName = Country Name (2 letter code)
+countryName_default = US
+countryName_min = 2
+countryName_max = 2
+
+stateOrProvinceName = State or Province Name (full name)
+stateOrProvinceName_default = %(state)s
+
+localityName = Locality Name (eg, city)
+localityName_default = %(city)s
+
+0.organizationName = Organization Name (eg, company)
+0.organizationName_default = %(organization)s
+
+organizationalUnitName = Organizational Unit Name (eg, section)
+organizationalUnitName_default = %(unit)s
+
+0.commonName = Common Name (FQDN of your server)
+0.commonName_default = %(common-name)s
+
+# To create a certificate for more than one name uncomment:
+# 1.commonName = DNS alias of your server
+# 2.commonName = DNS alias of your server
+# ...
+# See http://home.netscape.com/eng/security/ssl_2.0_certificate.html
+# to see how Netscape understands commonName.
+
+[ cert_type ]
+nsCertType = server
+"""
+
+def create_cert_files():
+
+ import tempfile, socket, os
+ d = tempfile.mkdtemp()
+ # now create a configuration file for the CA signing cert
+ fqdn = socket.getfqdn()
+ crtfile = os.path.join(d, "cert.pem")
+ conffile = os.path.join(d, "ca.conf")
+ fp = open(conffile, "w")
+ fp.write(CERTFILE_CONFIG_TEMPLATE %
+ {'state': "Delaware",
+ 'city': "Wilmington",
+ 'organization': "Python Software Foundation",
+ 'unit': "SSL",
+ 'common-name': fqdn,
+ })
+ fp.close()
+ os.system(
+ "openssl req -batch -new -x509 -days 10 -nodes -config %s "
+ "-keyout \"%s\" -out \"%s\" > /dev/null < /dev/null 2>&1" %
+ (conffile, crtfile, crtfile))
+ # now we have a self-signed server cert in crtfile
+ os.unlink(conffile)
+ #sf_certfile = os.path.join(d, "sourceforge-imap.pem")
+ #sf_cert = ssl.fetch_server_certificate('pop.gmail.com', 995)
+ #open(sf_certfile, 'w').write(sf_cert)
+ #return d, crtfile, sf_certfile
+ # sys.stderr.write(open(crtfile, 'r').read() + '\n')
+ return d, crtfile
+
+def test_main():
+ if skip_expected:
+ raise test_support.TestSkipped("socket module has no ssl support")
+
+ global CERTFILE
+ tdir, CERTFILE = create_cert_files()
+
+ tests = [BasicTests]
+
+ server = None
+ if test_support.is_resource_enabled('network'):
+ server = threadedEchoServer(10024, CERTFILE)
+ server.start()
+ time.sleep(1)
+ tests.append(ConnectedTests)
+
+ thread_info = test_support.threading_setup()
+
+ try:
+ test_support.run_unittest(*tests)
+ finally:
+ if server is not None and server.active:
+ server.stop()
+ # wait for it to stop
+ server.join()
+
+ shutil.rmtree(tdir)
+ test_support.threading_cleanup(*thread_info)
+
+if __name__ == "__main__":
+ test_main()