diff options
| -rw-r--r-- | netlib/tcp.py | 46 | ||||
| -rw-r--r-- | test/test_tcp.py | 25 |
2 files changed, 66 insertions, 5 deletions
diff --git a/netlib/tcp.py b/netlib/tcp.py index bb0a00b9..54148172 100644 --- a/netlib/tcp.py +++ b/netlib/tcp.py @@ -2,6 +2,37 @@ import select, socket, threading, traceback, sys from OpenSSL import SSL import certutils +SSLv2_METHOD = SSL.SSLv2_METHOD +SSLv3_METHOD = SSL.SSLv3_METHOD +SSLv23_METHOD = SSL.SSLv23_METHOD +TLSv1_METHOD = SSL.TLSv1_METHOD + +OP_ALL = SSL.OP_ALL +OP_CIPHER_SERVER_PREFERENCE = SSL.OP_CIPHER_SERVER_PREFERENCE +OP_COOKIE_EXCHANGE = SSL.OP_COOKIE_EXCHANGE +OP_DONT_INSERT_EMPTY_FRAGMENTS = SSL.OP_DONT_INSERT_EMPTY_FRAGMENTS +OP_EPHEMERAL_RSA = SSL.OP_EPHEMERAL_RSA +OP_MICROSOFT_BIG_SSLV3_BUFFER = SSL.OP_MICROSOFT_BIG_SSLV3_BUFFER +OP_MICROSOFT_SESS_ID_BUG = SSL.OP_MICROSOFT_SESS_ID_BUG +OP_MSIE_SSLV2_RSA_PADDING = SSL.OP_MSIE_SSLV2_RSA_PADDING +OP_NETSCAPE_CA_DN_BUG = SSL.OP_NETSCAPE_CA_DN_BUG +OP_NETSCAPE_CHALLENGE_BUG = SSL.OP_NETSCAPE_CHALLENGE_BUG +OP_NETSCAPE_DEMO_CIPHER_CHANGE_BUG = SSL.OP_NETSCAPE_DEMO_CIPHER_CHANGE_BUG +OP_NETSCAPE_REUSE_CIPHER_CHANGE_BUG = SSL.OP_NETSCAPE_REUSE_CIPHER_CHANGE_BUG +OP_NO_QUERY_MTU = SSL.OP_NO_QUERY_MTU +OP_NO_SSLv2 = SSL.OP_NO_SSLv2 +OP_NO_SSLv3 = SSL.OP_NO_SSLv3 +OP_NO_TICKET = SSL.OP_NO_TICKET +OP_NO_TLSv1 = SSL.OP_NO_TLSv1 +OP_PKCS1_CHECK_1 = SSL.OP_PKCS1_CHECK_1 +OP_PKCS1_CHECK_2 = SSL.OP_PKCS1_CHECK_2 +OP_SINGLE_DH_USE = SSL.OP_SINGLE_DH_USE +OP_SSLEAY_080_CLIENT_DH_BUG = SSL.OP_SSLEAY_080_CLIENT_DH_BUG +OP_SSLREF2_REUSE_CERT_TYPE_BUG = SSL.OP_SSLREF2_REUSE_CERT_TYPE_BUG +OP_TLS_BLOCK_PADDING_BUG = SSL.OP_TLS_BLOCK_PADDING_BUG +OP_TLS_D5_BUG = SSL.OP_TLS_D5_BUG +OP_TLS_ROLLBACK_BUG = SSL.OP_TLS_ROLLBACK_BUG + class NetLibError(Exception): pass @@ -58,8 +89,10 @@ class TCPClient: self.cert = None self.ssl_established = False - def convert_to_ssl(self, clientcert=None, sni=None): - context = SSL.Context(SSL.SSLv23_METHOD) + def convert_to_ssl(self, clientcert=None, sni=None, method=TLSv1_METHOD, options=None): + context = SSL.Context(method) + if not options is None: + ctx.set_options(options) if clientcert: context.use_certificate_file(self.clientcert) self.connection = SSL.Connection(context, self.connection) @@ -103,8 +136,13 @@ class BaseHandler: self.finished = False self.ssl_established = False - def convert_to_ssl(self, cert, key): - ctx = SSL.Context(SSL.SSLv23_METHOD) + def convert_to_ssl(self, cert, key, method=SSLv23_METHOD, options=None): + """ + method: One of SSLv2_METHOD, SSLv3_METHOD, SSLv23_METHOD, or TLSv1_METHOD + """ + ctx = SSL.Context(method) + if not options is None: + ctx.set_options(options) ctx.set_tlsext_servername_callback(self.handle_sni) ctx.use_privatekey_file(key) ctx.use_certificate_file(cert) diff --git a/test/test_tcp.py b/test/test_tcp.py index 969daf1e..b9f274ae 100644 --- a/test/test_tcp.py +++ b/test/test_tcp.py @@ -55,17 +55,26 @@ class DisconnectHandler(tcp.BaseHandler): class TServer(tcp.TCPServer): - def __init__(self, addr, ssl, q, handler): + def __init__(self, addr, ssl, q, handler, v3_only=False): tcp.TCPServer.__init__(self, addr) self.ssl, self.q = ssl, q + self.v3_only = v3_only self.handler = handler def handle_connection(self, request, client_address): h = self.handler(request, client_address, self) if self.ssl: + if self.v3_only: + method = tcp.SSLv3_METHOD + options = tcp.OP_NO_SSLv2|tcp.OP_NO_TLSv1 + else: + method = tcp.SSLv23_METHOD + options = None h.convert_to_ssl( tutils.test_data.path("data/server.crt"), tutils.test_data.path("data/server.key"), + method = method, + options = options, ) h.handle() h.finish() @@ -114,6 +123,20 @@ class TestServerSSL(ServerTestBase): assert certutils.get_remote_cert("127.0.0.1", self.port, None).digest("sha1") +class TestSSLv3Only(ServerTestBase): + @classmethod + def makeserver(cls): + cls.q = Queue.Queue() + s = TServer(("127.0.0.1", 0), True, cls.q, EchoHandler, True) + cls.port = s.port + return s + + def test_failure(self): + c = tcp.TCPClient("127.0.0.1", self.port) + c.connect() + tutils.raises(tcp.NetLibError, c.convert_to_ssl, sni="foo.com", method=tcp.TLSv1_METHOD) + + class TestSNI(ServerTestBase): @classmethod def makeserver(cls): |