pydtls/dtls/test/unit.py

1414 lines
56 KiB
Python

# Test the support for DTLS through the SSL module. Adapted from the Python
# standard library's test_ssl.py regression test module by Ray Brown.
import sys
import unittest
import asyncore
import socket
import select
import gc
import os
import errno
import pprint
import urllib, urlparse
import traceback
import weakref
import platform
import threading
import time
import datetime
import SocketServer
from SimpleHTTPServer import SimpleHTTPRequestHandler
from collections import OrderedDict
import ssl
from dtls import do_patch, force_routing_demux, reset_default_demux
HOST = "localhost"
CONNECTION_TIMEOUT = datetime.timedelta(seconds=30)
class TestSupport(object):
verbose = True
class Ctx(object):
def __enter__(self):
self.server = AsyncoreEchoServer(CERTFILE)
flag = threading.Event()
self.server.start(flag)
flag.wait()
return self.server.sockname
def __exit__(self, exc_type, exc_value, traceback):
self.server.stop()
self.server = None
def transient_internet(self):
return self.Ctx()
test_support = TestSupport()
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(AF_INET4_6, socket.SOCK_DGRAM))
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(AF_INET4_6,
socket.SOCK_DGRAM)._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
class BasicSocketTests(unittest.TestCase):
def test_constants(self):
ssl.PROTOCOL_SSLv23
ssl.PROTOCOL_TLSv1
ssl.PROTOCOL_DTLSv1 # added
ssl.PROTOCOL_DTLSv1_2 # added
ssl.PROTOCOL_DTLS # added
ssl.CERT_NONE
ssl.CERT_OPTIONAL
ssl.CERT_REQUIRED
def test_dtls_openssl_version(self):
n = ssl.DTLS_OPENSSL_VERSION_NUMBER
t = ssl.DTLS_OPENSSL_VERSION_INFO
s = ssl.DTLS_OPENSSL_VERSION
self.assertIsInstance(n, (int, long))
self.assertIsInstance(t, tuple)
self.assertIsInstance(s, str)
# Some sanity checks follow
# >= 1.0.2
self.assertGreaterEqual(n, 0x10002000)
# < 2.0
self.assertLess(n, 0x20000000)
major, minor, fix, patch, status = t
self.assertGreaterEqual(major, 1)
self.assertLess(major, 2)
self.assertGreaterEqual(minor, 0)
self.assertLess(minor, 256)
self.assertGreaterEqual(fix, 2)
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):
server = AsyncoreEchoServer(CERTFILE)
flag = threading.Event()
server.start(flag)
flag.wait()
remote = (HOST, server.port)
try:
s = ssl.wrap_socket(socket.socket(AF_INET4_6, socket.SOCK_DGRAM),
cert_reqs=ssl.CERT_NONE, ciphers="ALL")
s.connect(remote)
s = ssl.wrap_socket(socket.socket(AF_INET4_6, socket.SOCK_DGRAM),
cert_reqs=ssl.CERT_NONE, ciphers="DEFAULT")
s.connect(remote)
# Error checking occurs when connecting, because the SSL context
# isn't created before.
s = ssl.wrap_socket(socket.socket(AF_INET4_6, socket.SOCK_DGRAM),
cert_reqs=ssl.CERT_NONE,
ciphers="^$:,;?*'dorothyx")
with self.assertRaisesRegexp(ssl.SSLError,
"No cipher can be selected"):
s.connect(remote)
finally:
server.stop()
@unittest.skipIf(platform.python_implementation() != "CPython",
"Reference cycle test feasible under CPython only")
def test_refcycle(self):
# Issue #7943: an SSL object doesn't create reference cycles with
# itself.
s = socket.socket(AF_INET4_6, socket.SOCK_DGRAM)
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(AF_INET4_6, socket.SOCK_DGRAM)
ss = ssl.wrap_socket(s)
if os.name != "posix":
# On Linux, unconnected, unbound datagram sockets can receive and
# the following calls will therefore block
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) if AF_INET4_6 == socket.AF_INET else
('::', 0))
class NetworkedTests(unittest.TestCase):
def test_connect(self):
with test_support.transient_internet() as remote:
s = ssl.wrap_socket(socket.socket(AF_INET4_6, socket.SOCK_DGRAM),
cert_reqs=ssl.CERT_NONE)
s.connect(remote)
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(AF_INET4_6, socket.SOCK_DGRAM),
cert_reqs=ssl.CERT_REQUIRED)
try:
s.connect(remote)
except ssl.SSLError:
pass
finally:
s.close()
# this should succeed because we specify the root cert
s = ssl.wrap_socket(socket.socket(AF_INET4_6, socket.SOCK_DGRAM),
cert_reqs=ssl.CERT_REQUIRED,
ca_certs=ISSUER_CERTFILE)
try:
s.connect(remote)
finally:
s.close()
def test_connect_ex(self):
# Issue #11326: check connect_ex() implementation
with test_support.transient_internet() as remote:
s = ssl.wrap_socket(socket.socket(AF_INET4_6, socket.SOCK_DGRAM),
cert_reqs=ssl.CERT_REQUIRED,
ca_certs=ISSUER_CERTFILE)
try:
self.assertEqual(0, s.connect_ex(remote))
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() as remote:
s = ssl.wrap_socket(socket.socket(AF_INET4_6, socket.SOCK_DGRAM),
cert_reqs=ssl.CERT_REQUIRED,
ca_certs=ISSUER_CERTFILE,
do_handshake_on_connect=False)
try:
s.setblocking(False)
rc = s.connect_ex(remote)
# EWOULDBLOCK under Windows, EINPROGRESS elsewhere
self.assertIn(rc, (0, errno.EINPROGRESS, errno.EWOULDBLOCK))
# Non-blocking handshake
while True:
try:
s.do_handshake()
break
except ssl.SSLError as err:
if err.args[0] == ssl.SSL_ERROR_WANT_READ:
while True:
to = s.get_timeout()
to = to.total_seconds() if to else 5.0
sel = select.select([s], [], [], to)
if sel[0]:
break
s.handle_timeout()
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() as remote:
ss = ssl.wrap_socket(socket.socket(AF_INET4_6, socket.SOCK_DGRAM))
ss.connect(remote)
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() as remote:
s = socket.socket(AF_INET4_6, socket.SOCK_DGRAM)
s.connect(remote)
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:
while True:
to = s.get_timeout()
if to:
sel = select.select([s], [], [],
to.total_seconds())
if sel[0]:
break
s.handle_timeout()
continue
select.select([s], [], [])
break
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):
for prot in (ssl.PROTOCOL_DTLSv1, ssl.PROTOCOL_DTLSv1_2, ssl.PROTOCOL_DTLS):
with test_support.transient_internet() as remote:
pem = ssl.get_server_certificate(remote,
prot)
if not pem:
self.fail("No server certificate!")
try:
pem = ssl.get_server_certificate(remote,
prot,
ca_certs=OTHER_CERTFILE)
except ssl.SSLError:
# should fail
pass
else:
self.fail("Got server certificate %s!" % pem)
pem = ssl.get_server_certificate(remote,
prot,
ca_certs=ISSUER_CERTFILE)
if not pem:
self.fail("No server certificate!")
if test_support.verbose:
sys.stdout.write("\nVerified certificate is\n%s\n" % pem)
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.settimeout(CONNECTION_TIMEOUT.total_seconds())
self.sslconn = connsock
threading.Thread.__init__(self)
server.register_handler(True)
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")
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:
# 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 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):
self.server.register_handler(False)
if self.sslconn:
self.sslconn.close()
else:
self.sock._sock.close()
def run(self):
self.running = True
# Complete the handshake
try:
self.sock.do_handshake()
except ssl.SSLError:
if self.server.chatty:
handle_error("\n server: failed to handshake with " +
str(self.sock.getpeername()) + ":\n")
self.close()
self.running = False
self.server.stop()
return
if self.server.starttls_server:
self.sock = self.sock.unwrap()
self.sslconn = None
else:
self.show_conn_details()
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 not self.sslconn \
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()
def __init__(self, certificate, ssl_version=None,
certreqs=None, cacerts=None,
chatty=True, connectionchatty=False, starttls_server=False,
ciphers=None):
if ssl_version is None:
ssl_version = ssl.PROTOCOL_DTLSv1
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(AF_INET4_6, socket.SOCK_DGRAM)
self.flag = None
self.num_handlers = 0
self.num_handlers_lock = threading.Lock()
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,
do_handshake_on_connect=False,
ciphers=self.ciphers)
if test_support.verbose and self.chatty:
sys.stdout.write(' server: wrapped server ' +
'socket as %s\n' % str(self.sock))
self.sock.bind((HOST, 0))
self.port = self.sock.getsockname()[1]
self.active = False
threading.Thread.__init__(self)
self.daemon = True
def start(self, flag=None):
self.flag = flag
self.starter = threading.current_thread().ident
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:
acc_ret = self.sock.accept()
if acc_ret:
newconn, connaddr = acc_ret
if test_support.verbose and self.chatty:
sys.stdout.write(' server: new connection from '
+ str(connaddr) + '\n')
handler = self.ConnectionHandler(self, newconn)
handler.start()
except socket.timeout:
pass
except ssl.SSLError:
pass
except KeyboardInterrupt:
self.stop()
self.sock.close()
def register_handler(self, add):
with self.num_handlers_lock:
if add:
self.num_handlers += 1
else:
self.num_handlers -= 1
assert self.num_handlers >= 0
def stop(self):
self.active = False
if self.starter != threading.current_thread().ident:
return
self.join() # don't allow spawning new handlers after we've checked
last_msg = datetime.datetime.now()
while self.num_handlers:
time.sleep(0.05)
now = datetime.datetime.now()
if now > last_msg + datetime.timedelta(seconds=1):
sys.stdout.write(' server: waiting for connections to close\n')
last_msg = now
class AsyncoreEchoServer(threading.Thread):
class EchoServer(asyncore.dispatcher):
class ConnectionHandler(asyncore.dispatcher):
def __init__(self, conn, timeout_tracker, server):
asyncore.dispatcher.__init__(self, conn)
self._timeout_tracker = timeout_tracker
self._server = server
self._ssl_accepting = True
# Complete the handshake
self.handle_read_event()
def __hash__(self):
return hash(self.socket)
def readable(self):
while self.socket.pending() > 0:
self.handle_read_event()
if self._timeout_tracker.has_key(self) and \
datetime.datetime.now() >= self._timeout_tracker[self]:
self._timeout_tracker.pop(self)
try:
self.socket.handle_timeout()
except:
self.handle_close()
return False
return True
def writable(self):
return False
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,
ssl.SSL_ERROR_SSL):
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())
if self.connected:
self._server.reset_timeout(self)
self._server.check_timeout()
if not self.connected: # above called handle_close
return
delta = self.socket.get_timeout()
if delta:
self._timeout_tracker[self] = \
datetime.datetime.now() + delta
def handle_close(self):
if self._timeout_tracker.has_key(self):
self._timeout_tracker.pop(self)
self._server._handlers.pop(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, timeout_tracker):
asyncore.dispatcher.__init__(self)
self._timeout_tracker = timeout_tracker
self._handlers = OrderedDict()
sock = socket.socket(AF_INET4_6, socket.SOCK_DGRAM)
sock.setblocking(False)
sock.bind((HOST, 0))
self.sockname = sock.getsockname()
self.port = self.sockname[1]
self.set_socket(ssl.wrap_socket(sock, server_side=True,
certfile=certfile,
do_handshake_on_connect=False))
self.listen(5)
def writable(self):
return False
def handle_accept(self):
self.check_timeout()
acc_ret = self.accept()
if acc_ret:
sock_obj, addr = acc_ret
if test_support.verbose:
sys.stdout.write(" server: new connection from " +
"%s:%s\n" % (addr[0], str(addr[1:])))
self._handlers[self.ConnectionHandler(sock_obj,
self._timeout_tracker,
self)] = \
datetime.datetime.now()
def handle_error(self):
raise
def reset_timeout(self, handler):
if self._handlers.has_key(handler):
self._handlers.pop(handler)
self._handlers[handler] = datetime.datetime.now()
def check_timeout(self):
now = datetime.datetime.now()
while True:
try:
handler = self._handlers.__iter__().next() # oldest handler
except StopIteration:
break # there are no more handlers
if now > self._handlers[handler] + CONNECTION_TIMEOUT:
handler.handle_close()
else:
break # the oldest handlers has not yet timed out
def close(self):
map(lambda x: x.handle_close(), self._handlers.keys())
assert not self._handlers
asyncore.dispatcher.close(self)
def __init__(self, certfile):
self.flag = None
self.active = False
self.timeout_tracker = {}
self.server = self.EchoServer(certfile, self.timeout_tracker)
self.sockname = self.server.sockname
self.port = self.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):
self.active = True
if self.flag:
self.flag.set()
while self.active:
now = datetime.datetime.now()
future_timeouts = filter(lambda x: x > now,
self.timeout_tracker.values())
future_timeouts.append(now + datetime.timedelta(seconds=0.05))
first_timeout = min(future_timeouts) - now
asyncore.loop(first_timeout.total_seconds(), count=1)
def stop(self):
self.active = False
self.join()
self.server.close()
# Note that this HTTP-over-UDP server does not implement packet recovery and
# reordering, but it's good enough for testing on a loopback interface
class SocketServerHTTPSServer(threading.Thread):
class HTTPSServerUDP(SocketServer.ThreadingTCPServer):
def __init__(self, server_address, RequestHandlerClass, certfile):
SocketServer.ThreadingTCPServer.__init__(self, server_address,
RequestHandlerClass, False)
# account for dealing with a datagram socket
self.socket = ssl.wrap_socket(socket.socket(AF_INET4_6,
socket.SOCK_DGRAM),
server_side=True,
certfile=certfile,
do_handshake_on_connect=False)
self.server_bind()
self.server_activate()
def __str__(self):
return ('<%s %s:%s>' %
(self.__class__.__name__,
self.server_name,
self.server_port))
def server_bind(self):
"""Override server_bind to store the server name."""
SocketServer.ThreadingTCPServer.server_bind(self)
host, port = self.socket.getsockname()[:2]
self.server_name = socket.getfqdn(host)
self.server_port = port
def get_request(self):
# account for the fact that accept can return nothing, and
# according to BaseServer documentation, we should not block here
acc_ret = self.socket.accept()
if not acc_ret:
raise socket.error("No new connection")
return acc_ret
def shutdown_request(self, request):
# Notify client of termination
request.unwrap()
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-UDP/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.HTTPSServerUDP(
(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=ISSUER_CERTFILE, chatty=False)
flag = threading.Event()
server.start(flag)
# wait for it to start
flag.wait()
# try to connect
try:
try:
s = ssl.wrap_socket(socket.socket(AF_INET4_6, socket.SOCK_DGRAM),
certfile=certfile,
ssl_version=ssl.PROTOCOL_DTLSv1)
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!")
finally:
server.stop()
def server_params_test(certfile, protocol, certreqs, cacertsfile,
client_certfile, client_protocol=None,
indata="FOO\n", ciphers=None, chatty=True,
connectionchatty=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)
flag = threading.Event()
server.start(flag)
# wait for it to start
flag.wait()
# try to connect
if client_protocol is None:
client_protocol = protocol
try:
s = ssl.wrap_socket(socket.socket(AF_INET4_6, socket.SOCK_DGRAM),
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()
finally:
server.stop()
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,
ISSUER_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_unreachable(self):
server = socket.socket(AF_INET4_6, socket.SOCK_DGRAM)
server.bind((HOST, 0))
port = server.getsockname()[1]
server.close()
s = ssl.wrap_socket(socket.socket(AF_INET4_6, socket.SOCK_DGRAM))
self.assertRaisesRegexp(ssl.SSLError,
"The peer address is not reachable",
s.connect, (HOST, port))
def test_echo(self):
"""Basic test of an SSL client connecting to a server"""
if test_support.verbose:
sys.stdout.write("\n")
server_params_test(CERTFILE, ssl.PROTOCOL_DTLSv1, ssl.CERT_NONE,
CERTFILE, CERTFILE, ssl.PROTOCOL_DTLSv1,
chatty=True, connectionchatty=True)
def test_getpeercert(self):
if test_support.verbose:
sys.stdout.write("\n")
server = ThreadedEchoServer(CERTFILE,
certreqs=ssl.CERT_NONE,
ssl_version=ssl.PROTOCOL_DTLSv1,
cacerts=CERTFILE,
chatty=False)
flag = threading.Event()
server.start(flag)
# wait for it to start
flag.wait()
# try to connect
try:
s = ssl.wrap_socket(socket.socket(AF_INET4_6, socket.SOCK_DGRAM),
certfile=CERTFILE,
ca_certs=ISSUER_CERTFILE,
cert_reqs=ssl.CERT_REQUIRED,
ssl_version=ssl.PROTOCOL_DTLSv1)
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', 'Ray Srv Inc'),)
not in cert['subject']):
self.fail(
"Missing or invalid 'organizationName' field in "
"certificate subject; should be 'Ray Srv Inc'.")
s.write("over\n")
s.close()
finally:
server.stop()
def test_empty_cert(self):
"""Connecting with an empty cert file"""
bad_cert_test(os.path.join(os.path.dirname(__file__) or os.curdir,
"certs", "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,
"certs", "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,
"certs", "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,
"certs", "badkey.pem"))
def test_protocol_dtlsv1(self):
"""Connecting to a DTLSv1 server with various client options"""
if test_support.verbose:
sys.stdout.write("\n")
# server: 1.0 - client: 1.0 -> ok
try_protocol_combo(ssl.PROTOCOL_DTLSv1, ssl.PROTOCOL_DTLSv1, True)
try_protocol_combo(ssl.PROTOCOL_DTLSv1, ssl.PROTOCOL_DTLSv1, True,
ssl.CERT_OPTIONAL)
try_protocol_combo(ssl.PROTOCOL_DTLSv1, ssl.PROTOCOL_DTLSv1, True,
ssl.CERT_REQUIRED)
# server: any - client: 1.0 and 1.2(any) -> ok
try_protocol_combo(ssl.PROTOCOL_DTLS, ssl.PROTOCOL_DTLSv1, True)
try_protocol_combo(ssl.PROTOCOL_DTLS, ssl.PROTOCOL_DTLSv1, True,
ssl.CERT_REQUIRED)
try_protocol_combo(ssl.PROTOCOL_DTLS, ssl.PROTOCOL_DTLSv1_2, True)
try_protocol_combo(ssl.PROTOCOL_DTLS, ssl.PROTOCOL_DTLSv1_2, True,
ssl.CERT_REQUIRED)
try_protocol_combo(ssl.PROTOCOL_DTLS, ssl.PROTOCOL_DTLS, True)
try_protocol_combo(ssl.PROTOCOL_DTLS, ssl.PROTOCOL_DTLS, True,
ssl.CERT_REQUIRED)
# server: 1.0 - client: 1.2 -> fail
try_protocol_combo(ssl.PROTOCOL_DTLSv1, ssl.PROTOCOL_DTLSv1_2, False)
try_protocol_combo(ssl.PROTOCOL_DTLSv1, ssl.PROTOCOL_DTLSv1_2, False,
ssl.CERT_REQUIRED)
# server: 1.2 - client: 1.0 -> fail
try_protocol_combo(ssl.PROTOCOL_DTLSv1_2, ssl.PROTOCOL_DTLSv1, False)
try_protocol_combo(ssl.PROTOCOL_DTLSv1_2, ssl.PROTOCOL_DTLSv1, False,
ssl.CERT_REQUIRED)
# server: 1.2 - client: 1.2 -> ok
try_protocol_combo(ssl.PROTOCOL_DTLSv1_2, ssl.PROTOCOL_DTLSv1_2, True)
try_protocol_combo(ssl.PROTOCOL_DTLSv1_2, ssl.PROTOCOL_DTLSv1_2, True,
ssl.CERT_REQUIRED)
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_DTLSv1,
starttls_server=True,
chatty=True,
connectionchatty=True)
flag = threading.Event()
server.start(flag)
# wait for it to start
flag.wait()
# try to connect
wrapped = False
try:
s = ssl.wrap_socket(socket.socket(AF_INET4_6, socket.SOCK_DGRAM), ssl_version=ssl.PROTOCOL_DTLSv1)
s.connect((HOST, server.port))
s = s.unwrap()
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_DTLSv1)
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()
finally:
server.stop()
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-UDP server
s = ssl.wrap_socket(socket.socket(AF_INET4_6, socket.SOCK_DGRAM))
s.connect((HOST, server.port))
fl = "/" + os.path.split(CERTFILE)[1]
s.write("GET " + fl + " HTTP/1.1\r\n" +
"Host: " + HOST + "\r\n\r\n")
content = False
last_buf = ""
while True:
try:
buf = last_buf + s.read()
except ssl.SSLError as err:
if err.args[0] == ssl.SSL_ERROR_ZERO_RETURN:
s = s.unwrap() # complete shutdown protocol with server
break
raise
if test_support.verbose:
sys.stdout.write(
" client: read %d bytes from remote server '%s'\n"
% (len(buf), server))
if content:
d2.append(buf)
continue
ind = buf.find("\r\n\r\n")
if ind < 0:
last_buf = buf[-3:] # find double-newline across buffers
continue
d2.append(buf[ind + 4:])
content = True
last_buf = ""
s.close()
self.assertEqual(d1, ''.join(d2))
finally:
server.stop()
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)
flag = threading.Event()
server.start(flag)
# wait for it to start
flag.wait()
# try to connect
try:
s = ssl.wrap_socket(socket.socket(AF_INET4_6, socket.SOCK_DGRAM))
s.connect((HOST, 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()
finally:
server.stop()
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)
flag = threading.Event()
server.start(flag)
# wait for it to start
flag.wait()
# try to connect
s = ssl.wrap_socket(socket.socket(AF_INET4_6, socket.SOCK_DGRAM),
server_side=False,
certfile=CERTFILE,
ca_certs=CERTFILE,
cert_reqs=ssl.CERT_NONE,
ssl_version=ssl.PROTOCOL_DTLSv1)
s.connect((HOST, server.port))
try:
# 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()
finally:
server.stop()
def test_handshake_timeout(self):
# Issue #5103: SSL handshake must respect the socket timeout
server = socket.socket(AF_INET4_6, socket.SOCK_DGRAM)
server.bind((HOST, 0))
port = server.getsockname()[1]
try:
try:
c = socket.socket(AF_INET4_6, socket.SOCK_DGRAM)
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(AF_INET4_6, socket.SOCK_DGRAM)
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:
server.close()
def hostname_for_protocol(protocol):
global HOST
# We can't quite predict the content of the hosts file, but we prefer names
# to numbers in order to test name resolution; if we can't find a name,
# then fall back to a number for the given protocol
for name in HOST, "localhost", "ip6-localhost", "127.0.0.1", "::1":
try:
socket.getaddrinfo(name, 0, protocol)
except socket.error:
pass
else:
HOST = name
return
# Is the loopback interface enabled along with ipv6 for that interface?
raise Exception("Failed to select hostname for protocol %d" % protocol)
def test_main(verbose=True):
global CERTFILE, ISSUER_CERTFILE, OTHER_CERTFILE, AF_INET4_6
CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir,
"certs", "keycert.pem")
ISSUER_CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir,
"certs", "ca-cert.pem")
OTHER_CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir,
"certs", "yahoo-cert.pem")
for fl in CERTFILE, ISSUER_CERTFILE, OTHER_CERTFILE:
if not os.path.exists(fl):
raise Exception("Can't read certificate files!")
TestSupport.verbose = verbose
reset_default_demux()
do_patch()
for demux in "platform-native", "routing":
for AF_INET4_6 in socket.AF_INET, socket.AF_INET6:
print "Suite run: demux: %s, protocol: %d" % (demux, AF_INET4_6)
hostname_for_protocol(AF_INET4_6)
res = unittest.main(exit=False).result.wasSuccessful()
if not res:
print "Suite run failed: demux: %s, protocol: %d" % (
demux, AF_INET4_6)
sys.exit(True)
if not force_routing_demux():
break
if __name__ == "__main__":
verbose = True if len(sys.argv) > 1 and sys.argv[1] == "-v" else False
test_main(verbose)