ukamnya/pydtls
1
0 Fork 0
Code Issues Pull Requests Releases Wiki Activity

Added new methods for DTLSv1.2

Browse Source 
* dtls/err.py: Added error code ERR_WRONG_VERSION_NUMBER
* dtls/openssl.py: Added DTLS_server_method(), DTLSv1_2_server_method() and DTLSv1_2_client_method()
* dtls/patch.py: Default protocol DTLS for ssl.wrap_socket() and ssl.SSLSocket()
* dtls/sslconnection.py:
	- Introduced PROTOCOL_DTLSv1_2 and PROTOCOL_DTLS (the latter one is a synonym for the "higher" version)
	- Updated _init_client() and _init_server() with the new protocol methods
	- Default protocol DTLS for SSLConnection()
	- Return on ERR_WRONG_VERSION_NUMBER if client and server cannot agree on protocol version
* dtls/test/unit.py:
	- Extended test_get_server_certificate() to iterate over the different protocol combinations
	- Extended test_protocol_dtlsv1() to try the different protocol combinations between client and server
This commit is contained in:
mcfreis committed 2017-03-20 14:48:42 +01:00
1 parent 8b07f3f46d
commit 83204e8c4d
6 files changed
+225 -141
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

dtls/test/unit.py
+93 -66
View File
@@ -78,10 +78,12 @@ class BasicSocketTests(unittest.TestCase):
def test_constants(self):
ssl.PROTOCOL_SSLv23
ssl.PROTOCOL_TLSv1
ssl.PROTOCOL_DTLSv1 # added
ssl.CERT_NONE
ssl.CERT_OPTIONAL
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):
@@ -89,22 +91,22 @@ class BasicSocketTests(unittest.TestCase):
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
self.assertGreaterEqual(n, 0x10000000)
# < 2.0
self.assertLess(n, 0x20000000)
major, minor, fix, patch, status = t
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, 0)
self.assertLess(fix, 256)
self.assertGreaterEqual(patch, 0)
self.assertLessEqual(patch, 26)
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
@@ -297,35 +299,37 @@ class NetworkedTests(unittest.TestCase):
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() as remote:
pem = ssl.get_server_certificate(remote, ssl.PROTOCOL_DTLSv1)
if not pem:
self.fail("No server certificate!")
try:
pem = ssl.get_server_certificate(remote,
ssl.PROTOCOL_DTLSv1,
ca_certs=OTHER_CERTFILE)
except ssl.SSLError:
#should fail
pass
else:
self.fail("Got server certificate %s!" % pem)
pem = ssl.get_server_certificate(remote,
ssl.PROTOCOL_DTLSv1,
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):
"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
@@ -1036,17 +1040,40 @@ class ThreadedTests(unittest.TestCase):
"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")
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)
def test_starttls(self):
"""Switching from clear text to encrypted and back again."""
"""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")
@@ -1059,13 +1086,13 @@ class ThreadedTests(unittest.TestCase):
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))
s.connect((HOST, server.port))
s = s.unwrap()
if test_support.verbose:
# 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: