| # Copyright (C) Jean-Paul Calderone |
| # See LICENSE for details. |
| |
| """ |
| Unit tests for :py:obj:`OpenSSL.SSL`. |
| """ |
| |
| from gc import collect, get_referrers |
| from errno import ECONNREFUSED, EINPROGRESS, EWOULDBLOCK, EPIPE, ESHUTDOWN |
| from sys import platform, version_info |
| from socket import SHUT_RDWR, error, socket |
| from os import makedirs |
| from os.path import join |
| from unittest import main |
| from weakref import ref |
| |
| from six import PY3, text_type, u |
| |
| from OpenSSL.crypto import TYPE_RSA, FILETYPE_PEM |
| from OpenSSL.crypto import PKey, X509, X509Extension, X509Store |
| from OpenSSL.crypto import dump_privatekey, load_privatekey |
| from OpenSSL.crypto import dump_certificate, load_certificate |
| |
| from OpenSSL.SSL import _lib |
| from OpenSSL.SSL import OPENSSL_VERSION_NUMBER, SSLEAY_VERSION, SSLEAY_CFLAGS |
| from OpenSSL.SSL import SSLEAY_PLATFORM, SSLEAY_DIR, SSLEAY_BUILT_ON |
| from OpenSSL.SSL import SENT_SHUTDOWN, RECEIVED_SHUTDOWN |
| from OpenSSL.SSL import ( |
| SSLv2_METHOD, SSLv3_METHOD, SSLv23_METHOD, TLSv1_METHOD, |
| TLSv1_1_METHOD, TLSv1_2_METHOD) |
| from OpenSSL.SSL import OP_SINGLE_DH_USE, OP_NO_SSLv2, OP_NO_SSLv3 |
| from OpenSSL.SSL import ( |
| VERIFY_PEER, VERIFY_FAIL_IF_NO_PEER_CERT, VERIFY_CLIENT_ONCE, VERIFY_NONE) |
| |
| from OpenSSL.SSL import ( |
| SESS_CACHE_OFF, SESS_CACHE_CLIENT, SESS_CACHE_SERVER, SESS_CACHE_BOTH, |
| SESS_CACHE_NO_AUTO_CLEAR, SESS_CACHE_NO_INTERNAL_LOOKUP, |
| SESS_CACHE_NO_INTERNAL_STORE, SESS_CACHE_NO_INTERNAL) |
| from OpenSSL.SSL import ( |
| _Cryptography_HAS_EC, ELLIPTIC_CURVE_DESCRIPTIONS, |
| ECNotAvailable, UnknownObject) |
| |
| from OpenSSL.SSL import ( |
| Error, SysCallError, WantReadError, WantWriteError, ZeroReturnError) |
| from OpenSSL.SSL import ( |
| Context, ContextType, Session, Connection, ConnectionType, SSLeay_version) |
| |
| from OpenSSL.test.util import TestCase, b |
| from OpenSSL.test.test_crypto import ( |
| cleartextCertificatePEM, cleartextPrivateKeyPEM) |
| from OpenSSL.test.test_crypto import ( |
| client_cert_pem, client_key_pem, server_cert_pem, server_key_pem, |
| root_cert_pem) |
| |
| try: |
| from OpenSSL.SSL import OP_NO_QUERY_MTU |
| except ImportError: |
| OP_NO_QUERY_MTU = None |
| try: |
| from OpenSSL.SSL import OP_COOKIE_EXCHANGE |
| except ImportError: |
| OP_COOKIE_EXCHANGE = None |
| try: |
| from OpenSSL.SSL import OP_NO_TICKET |
| except ImportError: |
| OP_NO_TICKET = None |
| |
| try: |
| from OpenSSL.SSL import OP_NO_COMPRESSION |
| except ImportError: |
| OP_NO_COMPRESSION = None |
| |
| try: |
| from OpenSSL.SSL import MODE_RELEASE_BUFFERS |
| except ImportError: |
| MODE_RELEASE_BUFFERS = None |
| |
| try: |
| from OpenSSL.SSL import OP_NO_TLSv1, OP_NO_TLSv1_1, OP_NO_TLSv1_2 |
| except ImportError: |
| OP_NO_TLSv1 = OP_NO_TLSv1_1 = OP_NO_TLSv1_2 = None |
| |
| from OpenSSL.SSL import ( |
| SSL_ST_CONNECT, SSL_ST_ACCEPT, SSL_ST_MASK, SSL_ST_INIT, SSL_ST_BEFORE, |
| SSL_ST_OK, SSL_ST_RENEGOTIATE, |
| SSL_CB_LOOP, SSL_CB_EXIT, SSL_CB_READ, SSL_CB_WRITE, SSL_CB_ALERT, |
| SSL_CB_READ_ALERT, SSL_CB_WRITE_ALERT, SSL_CB_ACCEPT_LOOP, |
| SSL_CB_ACCEPT_EXIT, SSL_CB_CONNECT_LOOP, SSL_CB_CONNECT_EXIT, |
| SSL_CB_HANDSHAKE_START, SSL_CB_HANDSHAKE_DONE) |
| |
| # openssl dhparam 128 -out dh-128.pem (note that 128 is a small number of bits |
| # to use) |
| dhparam = """\ |
| -----BEGIN DH PARAMETERS----- |
| MBYCEQCobsg29c9WZP/54oAPcwiDAgEC |
| -----END DH PARAMETERS----- |
| """ |
| |
| |
| def verify_cb(conn, cert, errnum, depth, ok): |
| return ok |
| |
| |
| def socket_pair(): |
| """ |
| Establish and return a pair of network sockets connected to each other. |
| """ |
| # Connect a pair of sockets |
| port = socket() |
| port.bind(('', 0)) |
| port.listen(1) |
| client = socket() |
| client.setblocking(False) |
| client.connect_ex(("127.0.0.1", port.getsockname()[1])) |
| client.setblocking(True) |
| server = port.accept()[0] |
| |
| # Let's pass some unencrypted data to make sure our socket connection is |
| # fine. Just one byte, so we don't have to worry about buffers getting |
| # filled up or fragmentation. |
| server.send(b("x")) |
| assert client.recv(1024) == b("x") |
| client.send(b("y")) |
| assert server.recv(1024) == b("y") |
| |
| # Most of our callers want non-blocking sockets, make it easy for them. |
| server.setblocking(False) |
| client.setblocking(False) |
| |
| return (server, client) |
| |
| |
| |
| def handshake(client, server): |
| conns = [client, server] |
| while conns: |
| for conn in conns: |
| try: |
| conn.do_handshake() |
| except WantReadError: |
| pass |
| else: |
| conns.remove(conn) |
| |
| |
| def _create_certificate_chain(): |
| """ |
| Construct and return a chain of certificates. |
| |
| 1. A new self-signed certificate authority certificate (cacert) |
| 2. A new intermediate certificate signed by cacert (icert) |
| 3. A new server certificate signed by icert (scert) |
| """ |
| caext = X509Extension(b('basicConstraints'), False, b('CA:true')) |
| |
| # Step 1 |
| cakey = PKey() |
| cakey.generate_key(TYPE_RSA, 512) |
| cacert = X509() |
| cacert.get_subject().commonName = "Authority Certificate" |
| cacert.set_issuer(cacert.get_subject()) |
| cacert.set_pubkey(cakey) |
| cacert.set_notBefore(b("20000101000000Z")) |
| cacert.set_notAfter(b("20200101000000Z")) |
| cacert.add_extensions([caext]) |
| cacert.set_serial_number(0) |
| cacert.sign(cakey, "sha1") |
| |
| # Step 2 |
| ikey = PKey() |
| ikey.generate_key(TYPE_RSA, 512) |
| icert = X509() |
| icert.get_subject().commonName = "Intermediate Certificate" |
| icert.set_issuer(cacert.get_subject()) |
| icert.set_pubkey(ikey) |
| icert.set_notBefore(b("20000101000000Z")) |
| icert.set_notAfter(b("20200101000000Z")) |
| icert.add_extensions([caext]) |
| icert.set_serial_number(0) |
| icert.sign(cakey, "sha1") |
| |
| # Step 3 |
| skey = PKey() |
| skey.generate_key(TYPE_RSA, 512) |
| scert = X509() |
| scert.get_subject().commonName = "Server Certificate" |
| scert.set_issuer(icert.get_subject()) |
| scert.set_pubkey(skey) |
| scert.set_notBefore(b("20000101000000Z")) |
| scert.set_notAfter(b("20200101000000Z")) |
| scert.add_extensions([ |
| X509Extension(b('basicConstraints'), True, b('CA:false'))]) |
| scert.set_serial_number(0) |
| scert.sign(ikey, "sha1") |
| |
| return [(cakey, cacert), (ikey, icert), (skey, scert)] |
| |
| |
| |
| class _LoopbackMixin: |
| """ |
| Helper mixin which defines methods for creating a connected socket pair and |
| for forcing two connected SSL sockets to talk to each other via memory BIOs. |
| """ |
| def _loopbackClientFactory(self, socket): |
| client = Connection(Context(TLSv1_METHOD), socket) |
| client.set_connect_state() |
| return client |
| |
| |
| def _loopbackServerFactory(self, socket): |
| ctx = Context(TLSv1_METHOD) |
| ctx.use_privatekey(load_privatekey(FILETYPE_PEM, server_key_pem)) |
| ctx.use_certificate(load_certificate(FILETYPE_PEM, server_cert_pem)) |
| server = Connection(ctx, socket) |
| server.set_accept_state() |
| return server |
| |
| |
| def _loopback(self, serverFactory=None, clientFactory=None): |
| if serverFactory is None: |
| serverFactory = self._loopbackServerFactory |
| if clientFactory is None: |
| clientFactory = self._loopbackClientFactory |
| |
| (server, client) = socket_pair() |
| server = serverFactory(server) |
| client = clientFactory(client) |
| |
| handshake(client, server) |
| |
| server.setblocking(True) |
| client.setblocking(True) |
| return server, client |
| |
| |
| def _interactInMemory(self, client_conn, server_conn): |
| """ |
| Try to read application bytes from each of the two :py:obj:`Connection` |
| objects. Copy bytes back and forth between their send/receive buffers |
| for as long as there is anything to copy. When there is nothing more |
| to copy, return :py:obj:`None`. If one of them actually manages to deliver |
| some application bytes, return a two-tuple of the connection from which |
| the bytes were read and the bytes themselves. |
| """ |
| wrote = True |
| while wrote: |
| # Loop until neither side has anything to say |
| wrote = False |
| |
| # Copy stuff from each side's send buffer to the other side's |
| # receive buffer. |
| for (read, write) in [(client_conn, server_conn), |
| (server_conn, client_conn)]: |
| |
| # Give the side a chance to generate some more bytes, or |
| # succeed. |
| try: |
| data = read.recv(2 ** 16) |
| except WantReadError: |
| # It didn't succeed, so we'll hope it generated some |
| # output. |
| pass |
| else: |
| # It did succeed, so we'll stop now and let the caller deal |
| # with it. |
| return (read, data) |
| |
| while True: |
| # Keep copying as long as there's more stuff there. |
| try: |
| dirty = read.bio_read(4096) |
| except WantReadError: |
| # Okay, nothing more waiting to be sent. Stop |
| # processing this send buffer. |
| break |
| else: |
| # Keep track of the fact that someone generated some |
| # output. |
| wrote = True |
| write.bio_write(dirty) |
| |
| |
| |
| class VersionTests(TestCase): |
| """ |
| Tests for version information exposed by |
| :py:obj:`OpenSSL.SSL.SSLeay_version` and |
| :py:obj:`OpenSSL.SSL.OPENSSL_VERSION_NUMBER`. |
| """ |
| def test_OPENSSL_VERSION_NUMBER(self): |
| """ |
| :py:obj:`OPENSSL_VERSION_NUMBER` is an integer with status in the low |
| byte and the patch, fix, minor, and major versions in the |
| nibbles above that. |
| """ |
| self.assertTrue(isinstance(OPENSSL_VERSION_NUMBER, int)) |
| |
| |
| def test_SSLeay_version(self): |
| """ |
| :py:obj:`SSLeay_version` takes a version type indicator and returns |
| one of a number of version strings based on that indicator. |
| """ |
| versions = {} |
| for t in [SSLEAY_VERSION, SSLEAY_CFLAGS, SSLEAY_BUILT_ON, |
| SSLEAY_PLATFORM, SSLEAY_DIR]: |
| version = SSLeay_version(t) |
| versions[version] = t |
| self.assertTrue(isinstance(version, bytes)) |
| self.assertEqual(len(versions), 5) |
| |
| |
| |
| class ContextTests(TestCase, _LoopbackMixin): |
| """ |
| Unit tests for :py:obj:`OpenSSL.SSL.Context`. |
| """ |
| def test_method(self): |
| """ |
| :py:obj:`Context` can be instantiated with one of :py:obj:`SSLv2_METHOD`, |
| :py:obj:`SSLv3_METHOD`, :py:obj:`SSLv23_METHOD`, :py:obj:`TLSv1_METHOD`, |
| :py:obj:`TLSv1_1_METHOD`, or :py:obj:`TLSv1_2_METHOD`. |
| """ |
| methods = [ |
| SSLv3_METHOD, SSLv23_METHOD, TLSv1_METHOD] |
| for meth in methods: |
| Context(meth) |
| |
| |
| maybe = [SSLv2_METHOD, TLSv1_1_METHOD, TLSv1_2_METHOD] |
| for meth in maybe: |
| try: |
| Context(meth) |
| except (Error, ValueError): |
| # Some versions of OpenSSL have SSLv2 / TLSv1.1 / TLSv1.2, some |
| # don't. Difficult to say in advance. |
| pass |
| |
| self.assertRaises(TypeError, Context, "") |
| self.assertRaises(ValueError, Context, 10) |
| |
| |
| if not PY3: |
| def test_method_long(self): |
| """ |
| On Python 2 :py:class:`Context` accepts values of type |
| :py:obj:`long` as well as :py:obj:`int`. |
| """ |
| Context(long(TLSv1_METHOD)) |
| |
| |
| |
| def test_type(self): |
| """ |
| :py:obj:`Context` and :py:obj:`ContextType` refer to the same type object and can be |
| used to create instances of that type. |
| """ |
| self.assertIdentical(Context, ContextType) |
| self.assertConsistentType(Context, 'Context', TLSv1_METHOD) |
| |
| |
| def test_use_privatekey(self): |
| """ |
| :py:obj:`Context.use_privatekey` takes an :py:obj:`OpenSSL.crypto.PKey` instance. |
| """ |
| key = PKey() |
| key.generate_key(TYPE_RSA, 128) |
| ctx = Context(TLSv1_METHOD) |
| ctx.use_privatekey(key) |
| self.assertRaises(TypeError, ctx.use_privatekey, "") |
| |
| |
| def test_use_privatekey_file_missing(self): |
| """ |
| :py:obj:`Context.use_privatekey_file` raises :py:obj:`OpenSSL.SSL.Error` |
| when passed the name of a file which does not exist. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| self.assertRaises(Error, ctx.use_privatekey_file, self.mktemp()) |
| |
| |
| if not PY3: |
| def test_use_privatekey_file_long(self): |
| """ |
| On Python 2 :py:obj:`Context.use_privatekey_file` accepts a |
| filetype of type :py:obj:`long` as well as :py:obj:`int`. |
| """ |
| pemfile = self.mktemp() |
| |
| key = PKey() |
| key.generate_key(TYPE_RSA, 128) |
| |
| with open(pemfile, "wt") as pem: |
| pem.write( |
| dump_privatekey(FILETYPE_PEM, key).decode("ascii")) |
| |
| ctx = Context(TLSv1_METHOD) |
| ctx.use_privatekey_file(pemfile, long(FILETYPE_PEM)) |
| |
| |
| def test_use_certificate_wrong_args(self): |
| """ |
| :py:obj:`Context.use_certificate_wrong_args` raises :py:obj:`TypeError` |
| when not passed exactly one :py:obj:`OpenSSL.crypto.X509` instance as an |
| argument. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, ctx.use_certificate) |
| self.assertRaises(TypeError, ctx.use_certificate, "hello, world") |
| self.assertRaises(TypeError, ctx.use_certificate, X509(), "hello, world") |
| |
| |
| def test_use_certificate_uninitialized(self): |
| """ |
| :py:obj:`Context.use_certificate` raises :py:obj:`OpenSSL.SSL.Error` |
| when passed a :py:obj:`OpenSSL.crypto.X509` instance which has not been |
| initialized (ie, which does not actually have any certificate data). |
| """ |
| ctx = Context(TLSv1_METHOD) |
| self.assertRaises(Error, ctx.use_certificate, X509()) |
| |
| |
| def test_use_certificate(self): |
| """ |
| :py:obj:`Context.use_certificate` sets the certificate which will be |
| used to identify connections created using the context. |
| """ |
| # TODO |
| # Hard to assert anything. But we could set a privatekey then ask |
| # OpenSSL if the cert and key agree using check_privatekey. Then as |
| # long as check_privatekey works right we're good... |
| ctx = Context(TLSv1_METHOD) |
| ctx.use_certificate(load_certificate(FILETYPE_PEM, cleartextCertificatePEM)) |
| |
| |
| def test_use_certificate_file_wrong_args(self): |
| """ |
| :py:obj:`Context.use_certificate_file` raises :py:obj:`TypeError` if |
| called with zero arguments or more than two arguments, or if the first |
| argument is not a byte string or the second argumnent is not an integer. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, ctx.use_certificate_file) |
| self.assertRaises(TypeError, ctx.use_certificate_file, b"somefile", object()) |
| self.assertRaises( |
| TypeError, ctx.use_certificate_file, b"somefile", FILETYPE_PEM, object()) |
| self.assertRaises( |
| TypeError, ctx.use_certificate_file, object(), FILETYPE_PEM) |
| self.assertRaises( |
| TypeError, ctx.use_certificate_file, b"somefile", object()) |
| |
| |
| def test_use_certificate_file_missing(self): |
| """ |
| :py:obj:`Context.use_certificate_file` raises |
| `:py:obj:`OpenSSL.SSL.Error` if passed the name of a file which does not |
| exist. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| self.assertRaises(Error, ctx.use_certificate_file, self.mktemp()) |
| |
| |
| def test_use_certificate_file(self): |
| """ |
| :py:obj:`Context.use_certificate` sets the certificate which will be |
| used to identify connections created using the context. |
| """ |
| # TODO |
| # Hard to assert anything. But we could set a privatekey then ask |
| # OpenSSL if the cert and key agree using check_privatekey. Then as |
| # long as check_privatekey works right we're good... |
| pem_filename = self.mktemp() |
| with open(pem_filename, "wb") as pem_file: |
| pem_file.write(cleartextCertificatePEM) |
| |
| ctx = Context(TLSv1_METHOD) |
| ctx.use_certificate_file(pem_filename) |
| |
| |
| if not PY3: |
| def test_use_certificate_file_long(self): |
| """ |
| On Python 2 :py:obj:`Context.use_certificate_file` accepts a |
| filetype of type :py:obj:`long` as well as :py:obj:`int`. |
| """ |
| pem_filename = self.mktemp() |
| with open(pem_filename, "wb") as pem_file: |
| pem_file.write(cleartextCertificatePEM) |
| |
| ctx = Context(TLSv1_METHOD) |
| ctx.use_certificate_file(pem_filename, long(FILETYPE_PEM)) |
| |
| |
| def test_set_app_data_wrong_args(self): |
| """ |
| :py:obj:`Context.set_app_data` raises :py:obj:`TypeError` if called with other than |
| one argument. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.set_app_data) |
| self.assertRaises(TypeError, context.set_app_data, None, None) |
| |
| |
| def test_get_app_data_wrong_args(self): |
| """ |
| :py:obj:`Context.get_app_data` raises :py:obj:`TypeError` if called with any |
| arguments. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.get_app_data, None) |
| |
| |
| def test_app_data(self): |
| """ |
| :py:obj:`Context.set_app_data` stores an object for later retrieval using |
| :py:obj:`Context.get_app_data`. |
| """ |
| app_data = object() |
| context = Context(TLSv1_METHOD) |
| context.set_app_data(app_data) |
| self.assertIdentical(context.get_app_data(), app_data) |
| |
| |
| def test_set_options_wrong_args(self): |
| """ |
| :py:obj:`Context.set_options` raises :py:obj:`TypeError` if called with the wrong |
| number of arguments or a non-:py:obj:`int` argument. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.set_options) |
| self.assertRaises(TypeError, context.set_options, None) |
| self.assertRaises(TypeError, context.set_options, 1, None) |
| |
| |
| def test_set_options(self): |
| """ |
| :py:obj:`Context.set_options` returns the new options value. |
| """ |
| context = Context(TLSv1_METHOD) |
| options = context.set_options(OP_NO_SSLv2) |
| self.assertTrue(OP_NO_SSLv2 & options) |
| |
| |
| if not PY3: |
| def test_set_options_long(self): |
| """ |
| On Python 2 :py:obj:`Context.set_options` accepts values of type |
| :py:obj:`long` as well as :py:obj:`int`. |
| """ |
| context = Context(TLSv1_METHOD) |
| options = context.set_options(long(OP_NO_SSLv2)) |
| self.assertTrue(OP_NO_SSLv2 & options) |
| |
| |
| def test_set_mode_wrong_args(self): |
| """ |
| :py:obj:`Context.set`mode} raises :py:obj:`TypeError` if called with the wrong |
| number of arguments or a non-:py:obj:`int` argument. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.set_mode) |
| self.assertRaises(TypeError, context.set_mode, None) |
| self.assertRaises(TypeError, context.set_mode, 1, None) |
| |
| |
| if MODE_RELEASE_BUFFERS is not None: |
| def test_set_mode(self): |
| """ |
| :py:obj:`Context.set_mode` accepts a mode bitvector and returns the newly |
| set mode. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertTrue( |
| MODE_RELEASE_BUFFERS & context.set_mode(MODE_RELEASE_BUFFERS)) |
| |
| if not PY3: |
| def test_set_mode_long(self): |
| """ |
| On Python 2 :py:obj:`Context.set_mode` accepts values of type |
| :py:obj:`long` as well as :py:obj:`int`. |
| """ |
| context = Context(TLSv1_METHOD) |
| mode = context.set_mode(long(MODE_RELEASE_BUFFERS)) |
| self.assertTrue(MODE_RELEASE_BUFFERS & mode) |
| else: |
| "MODE_RELEASE_BUFFERS unavailable - OpenSSL version may be too old" |
| |
| |
| def test_set_timeout_wrong_args(self): |
| """ |
| :py:obj:`Context.set_timeout` raises :py:obj:`TypeError` if called with the wrong |
| number of arguments or a non-:py:obj:`int` argument. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.set_timeout) |
| self.assertRaises(TypeError, context.set_timeout, None) |
| self.assertRaises(TypeError, context.set_timeout, 1, None) |
| |
| |
| def test_get_timeout_wrong_args(self): |
| """ |
| :py:obj:`Context.get_timeout` raises :py:obj:`TypeError` if called with any arguments. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.get_timeout, None) |
| |
| |
| def test_timeout(self): |
| """ |
| :py:obj:`Context.set_timeout` sets the session timeout for all connections |
| created using the context object. :py:obj:`Context.get_timeout` retrieves this |
| value. |
| """ |
| context = Context(TLSv1_METHOD) |
| context.set_timeout(1234) |
| self.assertEquals(context.get_timeout(), 1234) |
| |
| |
| if not PY3: |
| def test_timeout_long(self): |
| """ |
| On Python 2 :py:obj:`Context.set_timeout` accepts values of type |
| `long` as well as int. |
| """ |
| context = Context(TLSv1_METHOD) |
| context.set_timeout(long(1234)) |
| self.assertEquals(context.get_timeout(), 1234) |
| |
| |
| def test_set_verify_depth_wrong_args(self): |
| """ |
| :py:obj:`Context.set_verify_depth` raises :py:obj:`TypeError` if called with the wrong |
| number of arguments or a non-:py:obj:`int` argument. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.set_verify_depth) |
| self.assertRaises(TypeError, context.set_verify_depth, None) |
| self.assertRaises(TypeError, context.set_verify_depth, 1, None) |
| |
| |
| def test_get_verify_depth_wrong_args(self): |
| """ |
| :py:obj:`Context.get_verify_depth` raises :py:obj:`TypeError` if called with any arguments. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.get_verify_depth, None) |
| |
| |
| def test_verify_depth(self): |
| """ |
| :py:obj:`Context.set_verify_depth` sets the number of certificates in a chain |
| to follow before giving up. The value can be retrieved with |
| :py:obj:`Context.get_verify_depth`. |
| """ |
| context = Context(TLSv1_METHOD) |
| context.set_verify_depth(11) |
| self.assertEquals(context.get_verify_depth(), 11) |
| |
| |
| if not PY3: |
| def test_verify_depth_long(self): |
| """ |
| On Python 2 :py:obj:`Context.set_verify_depth` accepts values of |
| type `long` as well as int. |
| """ |
| context = Context(TLSv1_METHOD) |
| context.set_verify_depth(long(11)) |
| self.assertEquals(context.get_verify_depth(), 11) |
| |
| |
| def _write_encrypted_pem(self, passphrase): |
| """ |
| Write a new private key out to a new file, encrypted using the given |
| passphrase. Return the path to the new file. |
| """ |
| key = PKey() |
| key.generate_key(TYPE_RSA, 128) |
| pemFile = self.mktemp() |
| fObj = open(pemFile, 'w') |
| pem = dump_privatekey(FILETYPE_PEM, key, "blowfish", passphrase) |
| fObj.write(pem.decode('ascii')) |
| fObj.close() |
| return pemFile |
| |
| |
| def test_set_passwd_cb_wrong_args(self): |
| """ |
| :py:obj:`Context.set_passwd_cb` raises :py:obj:`TypeError` if called with the |
| wrong arguments or with a non-callable first argument. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.set_passwd_cb) |
| self.assertRaises(TypeError, context.set_passwd_cb, None) |
| self.assertRaises(TypeError, context.set_passwd_cb, lambda: None, None, None) |
| |
| |
| def test_set_passwd_cb(self): |
| """ |
| :py:obj:`Context.set_passwd_cb` accepts a callable which will be invoked when |
| a private key is loaded from an encrypted PEM. |
| """ |
| passphrase = b("foobar") |
| pemFile = self._write_encrypted_pem(passphrase) |
| calledWith = [] |
| def passphraseCallback(maxlen, verify, extra): |
| calledWith.append((maxlen, verify, extra)) |
| return passphrase |
| context = Context(TLSv1_METHOD) |
| context.set_passwd_cb(passphraseCallback) |
| context.use_privatekey_file(pemFile) |
| self.assertTrue(len(calledWith), 1) |
| self.assertTrue(isinstance(calledWith[0][0], int)) |
| self.assertTrue(isinstance(calledWith[0][1], int)) |
| self.assertEqual(calledWith[0][2], None) |
| |
| |
| def test_passwd_callback_exception(self): |
| """ |
| :py:obj:`Context.use_privatekey_file` propagates any exception raised by the |
| passphrase callback. |
| """ |
| pemFile = self._write_encrypted_pem(b("monkeys are nice")) |
| def passphraseCallback(maxlen, verify, extra): |
| raise RuntimeError("Sorry, I am a fail.") |
| |
| context = Context(TLSv1_METHOD) |
| context.set_passwd_cb(passphraseCallback) |
| self.assertRaises(RuntimeError, context.use_privatekey_file, pemFile) |
| |
| |
| def test_passwd_callback_false(self): |
| """ |
| :py:obj:`Context.use_privatekey_file` raises :py:obj:`OpenSSL.SSL.Error` if the |
| passphrase callback returns a false value. |
| """ |
| pemFile = self._write_encrypted_pem(b("monkeys are nice")) |
| def passphraseCallback(maxlen, verify, extra): |
| return b"" |
| |
| context = Context(TLSv1_METHOD) |
| context.set_passwd_cb(passphraseCallback) |
| self.assertRaises(Error, context.use_privatekey_file, pemFile) |
| |
| |
| def test_passwd_callback_non_string(self): |
| """ |
| :py:obj:`Context.use_privatekey_file` raises :py:obj:`OpenSSL.SSL.Error` if the |
| passphrase callback returns a true non-string value. |
| """ |
| pemFile = self._write_encrypted_pem(b("monkeys are nice")) |
| def passphraseCallback(maxlen, verify, extra): |
| return 10 |
| |
| context = Context(TLSv1_METHOD) |
| context.set_passwd_cb(passphraseCallback) |
| self.assertRaises(ValueError, context.use_privatekey_file, pemFile) |
| |
| |
| def test_passwd_callback_too_long(self): |
| """ |
| If the passphrase returned by the passphrase callback returns a string |
| longer than the indicated maximum length, it is truncated. |
| """ |
| # A priori knowledge! |
| passphrase = b("x") * 1024 |
| pemFile = self._write_encrypted_pem(passphrase) |
| def passphraseCallback(maxlen, verify, extra): |
| assert maxlen == 1024 |
| return passphrase + b("y") |
| |
| context = Context(TLSv1_METHOD) |
| context.set_passwd_cb(passphraseCallback) |
| # This shall succeed because the truncated result is the correct |
| # passphrase. |
| context.use_privatekey_file(pemFile) |
| |
| |
| def test_set_info_callback(self): |
| """ |
| :py:obj:`Context.set_info_callback` accepts a callable which will be invoked |
| when certain information about an SSL connection is available. |
| """ |
| (server, client) = socket_pair() |
| |
| clientSSL = Connection(Context(TLSv1_METHOD), client) |
| clientSSL.set_connect_state() |
| |
| called = [] |
| def info(conn, where, ret): |
| called.append((conn, where, ret)) |
| context = Context(TLSv1_METHOD) |
| context.set_info_callback(info) |
| context.use_certificate( |
| load_certificate(FILETYPE_PEM, cleartextCertificatePEM)) |
| context.use_privatekey( |
| load_privatekey(FILETYPE_PEM, cleartextPrivateKeyPEM)) |
| |
| serverSSL = Connection(context, server) |
| serverSSL.set_accept_state() |
| |
| handshake(clientSSL, serverSSL) |
| |
| # The callback must always be called with a Connection instance as the |
| # first argument. It would probably be better to split this into |
| # separate tests for client and server side info callbacks so we could |
| # assert it is called with the right Connection instance. It would |
| # also be good to assert *something* about `where` and `ret`. |
| notConnections = [ |
| conn for (conn, where, ret) in called |
| if not isinstance(conn, Connection)] |
| self.assertEqual( |
| [], notConnections, |
| "Some info callback arguments were not Connection instaces.") |
| |
| |
| def _load_verify_locations_test(self, *args): |
| """ |
| Create a client context which will verify the peer certificate and call |
| its :py:obj:`load_verify_locations` method with the given arguments. |
| Then connect it to a server and ensure that the handshake succeeds. |
| """ |
| (server, client) = socket_pair() |
| |
| clientContext = Context(TLSv1_METHOD) |
| clientContext.load_verify_locations(*args) |
| # Require that the server certificate verify properly or the |
| # connection will fail. |
| clientContext.set_verify( |
| VERIFY_PEER, |
| lambda conn, cert, errno, depth, preverify_ok: preverify_ok) |
| |
| clientSSL = Connection(clientContext, client) |
| clientSSL.set_connect_state() |
| |
| serverContext = Context(TLSv1_METHOD) |
| serverContext.use_certificate( |
| load_certificate(FILETYPE_PEM, cleartextCertificatePEM)) |
| serverContext.use_privatekey( |
| load_privatekey(FILETYPE_PEM, cleartextPrivateKeyPEM)) |
| |
| serverSSL = Connection(serverContext, server) |
| serverSSL.set_accept_state() |
| |
| # Without load_verify_locations above, the handshake |
| # will fail: |
| # Error: [('SSL routines', 'SSL3_GET_SERVER_CERTIFICATE', |
| # 'certificate verify failed')] |
| handshake(clientSSL, serverSSL) |
| |
| cert = clientSSL.get_peer_certificate() |
| self.assertEqual(cert.get_subject().CN, 'Testing Root CA') |
| |
| |
| def test_load_verify_file(self): |
| """ |
| :py:obj:`Context.load_verify_locations` accepts a file name and uses the |
| certificates within for verification purposes. |
| """ |
| cafile = self.mktemp() |
| fObj = open(cafile, 'w') |
| fObj.write(cleartextCertificatePEM.decode('ascii')) |
| fObj.close() |
| |
| self._load_verify_locations_test(cafile) |
| |
| |
| def test_load_verify_invalid_file(self): |
| """ |
| :py:obj:`Context.load_verify_locations` raises :py:obj:`Error` when passed a |
| non-existent cafile. |
| """ |
| clientContext = Context(TLSv1_METHOD) |
| self.assertRaises( |
| Error, clientContext.load_verify_locations, self.mktemp()) |
| |
| |
| def test_load_verify_directory(self): |
| """ |
| :py:obj:`Context.load_verify_locations` accepts a directory name and uses |
| the certificates within for verification purposes. |
| """ |
| capath = self.mktemp() |
| makedirs(capath) |
| # Hash values computed manually with c_rehash to avoid depending on |
| # c_rehash in the test suite. One is from OpenSSL 0.9.8, the other |
| # from OpenSSL 1.0.0. |
| for name in [b'c7adac82.0', b'c3705638.0']: |
| cafile = join(capath, name) |
| fObj = open(cafile, 'w') |
| fObj.write(cleartextCertificatePEM.decode('ascii')) |
| fObj.close() |
| |
| self._load_verify_locations_test(None, capath) |
| |
| |
| def test_load_verify_locations_wrong_args(self): |
| """ |
| :py:obj:`Context.load_verify_locations` raises :py:obj:`TypeError` if called with |
| the wrong number of arguments or with non-:py:obj:`str` arguments. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.load_verify_locations) |
| self.assertRaises(TypeError, context.load_verify_locations, object()) |
| self.assertRaises(TypeError, context.load_verify_locations, object(), object()) |
| self.assertRaises(TypeError, context.load_verify_locations, None, None, None) |
| |
| |
| if platform == "win32": |
| "set_default_verify_paths appears not to work on Windows. " |
| "See LP#404343 and LP#404344." |
| else: |
| def test_set_default_verify_paths(self): |
| """ |
| :py:obj:`Context.set_default_verify_paths` causes the platform-specific CA |
| certificate locations to be used for verification purposes. |
| """ |
| # Testing this requires a server with a certificate signed by one of |
| # the CAs in the platform CA location. Getting one of those costs |
| # money. Fortunately (or unfortunately, depending on your |
| # perspective), it's easy to think of a public server on the |
| # internet which has such a certificate. Connecting to the network |
| # in a unit test is bad, but it's the only way I can think of to |
| # really test this. -exarkun |
| |
| # Arg, verisign.com doesn't speak TLSv1 |
| context = Context(SSLv3_METHOD) |
| context.set_default_verify_paths() |
| context.set_verify( |
| VERIFY_PEER, |
| lambda conn, cert, errno, depth, preverify_ok: preverify_ok) |
| |
| client = socket() |
| client.connect(('verisign.com', 443)) |
| clientSSL = Connection(context, client) |
| clientSSL.set_connect_state() |
| clientSSL.do_handshake() |
| clientSSL.send(b"GET / HTTP/1.0\r\n\r\n") |
| self.assertTrue(clientSSL.recv(1024)) |
| |
| |
| def test_set_default_verify_paths_signature(self): |
| """ |
| :py:obj:`Context.set_default_verify_paths` takes no arguments and raises |
| :py:obj:`TypeError` if given any. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.set_default_verify_paths, None) |
| self.assertRaises(TypeError, context.set_default_verify_paths, 1) |
| self.assertRaises(TypeError, context.set_default_verify_paths, "") |
| |
| |
| def test_add_extra_chain_cert_invalid_cert(self): |
| """ |
| :py:obj:`Context.add_extra_chain_cert` raises :py:obj:`TypeError` if called with |
| other than one argument or if called with an object which is not an |
| instance of :py:obj:`X509`. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.add_extra_chain_cert) |
| self.assertRaises(TypeError, context.add_extra_chain_cert, object()) |
| self.assertRaises(TypeError, context.add_extra_chain_cert, object(), object()) |
| |
| |
| def _handshake_test(self, serverContext, clientContext): |
| """ |
| Verify that a client and server created with the given contexts can |
| successfully handshake and communicate. |
| """ |
| serverSocket, clientSocket = socket_pair() |
| |
| server = Connection(serverContext, serverSocket) |
| server.set_accept_state() |
| |
| client = Connection(clientContext, clientSocket) |
| client.set_connect_state() |
| |
| # Make them talk to each other. |
| # self._interactInMemory(client, server) |
| for i in range(3): |
| for s in [client, server]: |
| try: |
| s.do_handshake() |
| except WantReadError: |
| pass |
| |
| |
| def test_set_verify_callback_exception(self): |
| """ |
| If the verify callback passed to :py:obj:`Context.set_verify` raises an |
| exception, verification fails and the exception is propagated to the |
| caller of :py:obj:`Connection.do_handshake`. |
| """ |
| serverContext = Context(TLSv1_METHOD) |
| serverContext.use_privatekey( |
| load_privatekey(FILETYPE_PEM, cleartextPrivateKeyPEM)) |
| serverContext.use_certificate( |
| load_certificate(FILETYPE_PEM, cleartextCertificatePEM)) |
| |
| clientContext = Context(TLSv1_METHOD) |
| def verify_callback(*args): |
| raise Exception("silly verify failure") |
| clientContext.set_verify(VERIFY_PEER, verify_callback) |
| |
| exc = self.assertRaises( |
| Exception, self._handshake_test, serverContext, clientContext) |
| self.assertEqual("silly verify failure", str(exc)) |
| |
| |
| def test_add_extra_chain_cert(self): |
| """ |
| :py:obj:`Context.add_extra_chain_cert` accepts an :py:obj:`X509` instance to add to |
| the certificate chain. |
| |
| See :py:obj:`_create_certificate_chain` for the details of the certificate |
| chain tested. |
| |
| The chain is tested by starting a server with scert and connecting |
| to it with a client which trusts cacert and requires verification to |
| succeed. |
| """ |
| chain = _create_certificate_chain() |
| [(cakey, cacert), (ikey, icert), (skey, scert)] = chain |
| |
| # Dump the CA certificate to a file because that's the only way to load |
| # it as a trusted CA in the client context. |
| for cert, name in [(cacert, 'ca.pem'), (icert, 'i.pem'), (scert, 's.pem')]: |
| fObj = open(name, 'w') |
| fObj.write(dump_certificate(FILETYPE_PEM, cert).decode('ascii')) |
| fObj.close() |
| |
| for key, name in [(cakey, 'ca.key'), (ikey, 'i.key'), (skey, 's.key')]: |
| fObj = open(name, 'w') |
| fObj.write(dump_privatekey(FILETYPE_PEM, key).decode('ascii')) |
| fObj.close() |
| |
| # Create the server context |
| serverContext = Context(TLSv1_METHOD) |
| serverContext.use_privatekey(skey) |
| serverContext.use_certificate(scert) |
| # The client already has cacert, we only need to give them icert. |
| serverContext.add_extra_chain_cert(icert) |
| |
| # Create the client |
| clientContext = Context(TLSv1_METHOD) |
| clientContext.set_verify( |
| VERIFY_PEER | VERIFY_FAIL_IF_NO_PEER_CERT, verify_cb) |
| clientContext.load_verify_locations(b"ca.pem") |
| |
| # Try it out. |
| self._handshake_test(serverContext, clientContext) |
| |
| |
| def test_use_certificate_chain_file(self): |
| """ |
| :py:obj:`Context.use_certificate_chain_file` reads a certificate chain from |
| the specified file. |
| |
| The chain is tested by starting a server with scert and connecting |
| to it with a client which trusts cacert and requires verification to |
| succeed. |
| """ |
| chain = _create_certificate_chain() |
| [(cakey, cacert), (ikey, icert), (skey, scert)] = chain |
| |
| # Write out the chain file. |
| chainFile = self.mktemp() |
| fObj = open(chainFile, 'wb') |
| # Most specific to least general. |
| fObj.write(dump_certificate(FILETYPE_PEM, scert)) |
| fObj.write(dump_certificate(FILETYPE_PEM, icert)) |
| fObj.write(dump_certificate(FILETYPE_PEM, cacert)) |
| fObj.close() |
| |
| serverContext = Context(TLSv1_METHOD) |
| serverContext.use_certificate_chain_file(chainFile) |
| serverContext.use_privatekey(skey) |
| |
| fObj = open('ca.pem', 'w') |
| fObj.write(dump_certificate(FILETYPE_PEM, cacert).decode('ascii')) |
| fObj.close() |
| |
| clientContext = Context(TLSv1_METHOD) |
| clientContext.set_verify( |
| VERIFY_PEER | VERIFY_FAIL_IF_NO_PEER_CERT, verify_cb) |
| clientContext.load_verify_locations(b"ca.pem") |
| |
| self._handshake_test(serverContext, clientContext) |
| |
| |
| def test_use_certificate_chain_file_wrong_args(self): |
| """ |
| :py:obj:`Context.use_certificate_chain_file` raises :py:obj:`TypeError` |
| if passed zero or more than one argument or when passed a non-byte |
| string single argument. It also raises :py:obj:`OpenSSL.SSL.Error` when |
| passed a bad chain file name (for example, the name of a file which does |
| not exist). |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.use_certificate_chain_file) |
| self.assertRaises(TypeError, context.use_certificate_chain_file, object()) |
| self.assertRaises(TypeError, context.use_certificate_chain_file, b"foo", object()) |
| |
| self.assertRaises(Error, context.use_certificate_chain_file, self.mktemp()) |
| |
| # XXX load_client_ca |
| # XXX set_session_id |
| |
| def test_get_verify_mode_wrong_args(self): |
| """ |
| :py:obj:`Context.get_verify_mode` raises :py:obj:`TypeError` if called with any |
| arguments. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.get_verify_mode, None) |
| |
| |
| def test_set_verify_mode(self): |
| """ |
| :py:obj:`Context.get_verify_mode` returns the verify mode flags previously |
| passed to :py:obj:`Context.set_verify`. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertEquals(context.get_verify_mode(), 0) |
| context.set_verify( |
| VERIFY_PEER | VERIFY_CLIENT_ONCE, lambda *args: None) |
| self.assertEquals( |
| context.get_verify_mode(), VERIFY_PEER | VERIFY_CLIENT_ONCE) |
| |
| |
| if not PY3: |
| def test_set_verify_mode_long(self): |
| """ |
| On Python 2 :py:obj:`Context.set_verify_mode` accepts values of |
| type :py:obj:`long` as well as :py:obj:`int`. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertEquals(context.get_verify_mode(), 0) |
| context.set_verify( |
| long(VERIFY_PEER | VERIFY_CLIENT_ONCE), lambda *args: None) |
| self.assertEquals( |
| context.get_verify_mode(), VERIFY_PEER | VERIFY_CLIENT_ONCE) |
| |
| |
| def test_load_tmp_dh_wrong_args(self): |
| """ |
| :py:obj:`Context.load_tmp_dh` raises :py:obj:`TypeError` if called with the wrong |
| number of arguments or with a non-:py:obj:`str` argument. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.load_tmp_dh) |
| self.assertRaises(TypeError, context.load_tmp_dh, "foo", None) |
| self.assertRaises(TypeError, context.load_tmp_dh, object()) |
| |
| |
| def test_load_tmp_dh_missing_file(self): |
| """ |
| :py:obj:`Context.load_tmp_dh` raises :py:obj:`OpenSSL.SSL.Error` if the specified file |
| does not exist. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(Error, context.load_tmp_dh, b"hello") |
| |
| |
| def test_load_tmp_dh(self): |
| """ |
| :py:obj:`Context.load_tmp_dh` loads Diffie-Hellman parameters from the |
| specified file. |
| """ |
| context = Context(TLSv1_METHOD) |
| dhfilename = self.mktemp() |
| dhfile = open(dhfilename, "w") |
| dhfile.write(dhparam) |
| dhfile.close() |
| context.load_tmp_dh(dhfilename) |
| # XXX What should I assert here? -exarkun |
| |
| |
| def test_set_tmp_ecdh_curve(self): |
| """ |
| :py:obj:`Context.set_tmp_ecdh_curve` sets the elliptic curve for |
| Diffie-Hellman to the specified named curve. |
| """ |
| context = Context(TLSv1_METHOD) |
| for curve in ELLIPTIC_CURVE_DESCRIPTIONS.keys(): |
| context.set_tmp_ecdh_curve(curve) # Must not throw. |
| |
| |
| def test_set_tmp_ecdh_curve_not_available(self): |
| """ |
| :py:obj:`Context.set_tmp_ecdh_curve` raises :py:obj:`ECNotAvailable` if |
| elliptic curve support is not available from the underlying OpenSSL |
| version at all. |
| """ |
| self.addCleanup( |
| setattr, |
| _lib, "Cryptography_HAS_EC", _lib.Cryptography_HAS_EC) |
| _lib.Cryptography_HAS_EC = False |
| |
| context = Context(TLSv1_METHOD) |
| self.assertRaises( |
| ECNotAvailable, |
| context.set_tmp_ecdh_curve, next(iter(ELLIPTIC_CURVE_DESCRIPTIONS))) |
| |
| |
| def test_set_tmp_ecdh_curve_bad_sn(self): |
| """ |
| :py:obj:`Context.set_tmp_ecdh_curve` raises :py:obj:`UnknownObject` if |
| passed a curve_name that OpenSSL does not recognize and EC is |
| available. |
| """ |
| context = Context(TLSv1_METHOD) |
| try: |
| context.set_tmp_ecdh_curve('not_an_elliptic_curve') |
| except ECNotAvailable: |
| self.assertFalse(_Cryptography_HAS_EC) |
| except UnknownObject: |
| self.assertTrue(_Cryptography_HAS_EC) |
| else: |
| self.assertFalse(True) |
| |
| |
| def test_set_tmp_ecdh_curve_not_a_curve(self): |
| """ |
| :py:obj:`Context.set_tmp_ecdh_curve` raises |
| :py:obj:`UnsupportedEllipticCurve` if passed a curve_name that OpenSSL |
| cannot instantiate as an elliptic curve. It raises |
| :py:obj:`ECNotAvailable` if EC is not available at all. |
| """ |
| context = Context(TLSv1_METHOD) |
| try: |
| context.set_tmp_ecdh_curve('sha256') |
| except ECNotAvailable: |
| self.assertFalse(_Cryptography_HAS_EC) |
| except UnknownObject: |
| self.assertTrue(_Cryptography_HAS_EC) |
| else: |
| self.assertFalse(True) |
| |
| |
| def test_has_curve_descriptions(self): |
| """ |
| If the underlying cryptography bindings claim to have elliptic |
| curve support, there should be at least one curve. |
| |
| (In theory there could be an OpenSSL that violates this |
| assumption. If so, this test will fail and we'll find out.) |
| |
| """ |
| if _Cryptography_HAS_EC: |
| self.assertNotEqual(len(ELLIPTIC_CURVE_DESCRIPTIONS), 0) |
| |
| |
| def test_set_cipher_list_bytes(self): |
| """ |
| :py:obj:`Context.set_cipher_list` accepts a :py:obj:`bytes` naming the |
| ciphers which connections created with the context object will be able |
| to choose from. |
| """ |
| context = Context(TLSv1_METHOD) |
| context.set_cipher_list(b"hello world:EXP-RC4-MD5") |
| conn = Connection(context, None) |
| self.assertEquals(conn.get_cipher_list(), ["EXP-RC4-MD5"]) |
| |
| |
| def test_set_cipher_list_text(self): |
| """ |
| :py:obj:`Context.set_cipher_list` accepts a :py:obj:`unicode` naming |
| the ciphers which connections created with the context object will be |
| able to choose from. |
| """ |
| context = Context(TLSv1_METHOD) |
| context.set_cipher_list(u("hello world:EXP-RC4-MD5")) |
| conn = Connection(context, None) |
| self.assertEquals(conn.get_cipher_list(), ["EXP-RC4-MD5"]) |
| |
| |
| def test_set_cipher_list_wrong_args(self): |
| """ |
| :py:obj:`Context.set_cipher_list` raises :py:obj:`TypeError` when |
| passed zero arguments or more than one argument or when passed a |
| non-string single argument and raises :py:obj:`OpenSSL.SSL.Error` when |
| passed an incorrect cipher list string. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.set_cipher_list) |
| self.assertRaises(TypeError, context.set_cipher_list, object()) |
| self.assertRaises(TypeError, context.set_cipher_list, b"EXP-RC4-MD5", object()) |
| |
| self.assertRaises(Error, context.set_cipher_list, "imaginary-cipher") |
| |
| |
| def test_set_session_cache_mode_wrong_args(self): |
| """ |
| :py:obj:`Context.set_session_cache_mode` raises :py:obj:`TypeError` if |
| called with other than one integer argument. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.set_session_cache_mode) |
| self.assertRaises(TypeError, context.set_session_cache_mode, object()) |
| |
| |
| def test_get_session_cache_mode_wrong_args(self): |
| """ |
| :py:obj:`Context.get_session_cache_mode` raises :py:obj:`TypeError` if |
| called with any arguments. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.get_session_cache_mode, 1) |
| |
| |
| def test_session_cache_mode(self): |
| """ |
| :py:obj:`Context.set_session_cache_mode` specifies how sessions are |
| cached. The setting can be retrieved via |
| :py:obj:`Context.get_session_cache_mode`. |
| """ |
| context = Context(TLSv1_METHOD) |
| context.set_session_cache_mode(SESS_CACHE_OFF) |
| off = context.set_session_cache_mode(SESS_CACHE_BOTH) |
| self.assertEqual(SESS_CACHE_OFF, off) |
| self.assertEqual(SESS_CACHE_BOTH, context.get_session_cache_mode()) |
| |
| if not PY3: |
| def test_session_cache_mode_long(self): |
| """ |
| On Python 2 :py:obj:`Context.set_session_cache_mode` accepts values |
| of type :py:obj:`long` as well as :py:obj:`int`. |
| """ |
| context = Context(TLSv1_METHOD) |
| context.set_session_cache_mode(long(SESS_CACHE_BOTH)) |
| self.assertEqual( |
| SESS_CACHE_BOTH, context.get_session_cache_mode()) |
| |
| |
| def test_get_cert_store(self): |
| """ |
| :py:obj:`Context.get_cert_store` returns a :py:obj:`X509Store` instance. |
| """ |
| context = Context(TLSv1_METHOD) |
| store = context.get_cert_store() |
| self.assertIsInstance(store, X509Store) |
| |
| |
| |
| class ServerNameCallbackTests(TestCase, _LoopbackMixin): |
| """ |
| Tests for :py:obj:`Context.set_tlsext_servername_callback` and its interaction with |
| :py:obj:`Connection`. |
| """ |
| def test_wrong_args(self): |
| """ |
| :py:obj:`Context.set_tlsext_servername_callback` raises :py:obj:`TypeError` if called |
| with other than one argument. |
| """ |
| context = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, context.set_tlsext_servername_callback) |
| self.assertRaises( |
| TypeError, context.set_tlsext_servername_callback, 1, 2) |
| |
| |
| def test_old_callback_forgotten(self): |
| """ |
| If :py:obj:`Context.set_tlsext_servername_callback` is used to specify a new |
| callback, the one it replaces is dereferenced. |
| """ |
| def callback(connection): |
| pass |
| |
| def replacement(connection): |
| pass |
| |
| context = Context(TLSv1_METHOD) |
| context.set_tlsext_servername_callback(callback) |
| |
| tracker = ref(callback) |
| del callback |
| |
| context.set_tlsext_servername_callback(replacement) |
| |
| # One run of the garbage collector happens to work on CPython. PyPy |
| # doesn't collect the underlying object until a second run for whatever |
| # reason. That's fine, it still demonstrates our code has properly |
| # dropped the reference. |
| collect() |
| collect() |
| |
| callback = tracker() |
| if callback is not None: |
| referrers = get_referrers(callback) |
| if len(referrers) > 1: |
| self.fail("Some references remain: %r" % (referrers,)) |
| |
| |
| def test_no_servername(self): |
| """ |
| When a client specifies no server name, the callback passed to |
| :py:obj:`Context.set_tlsext_servername_callback` is invoked and the result of |
| :py:obj:`Connection.get_servername` is :py:obj:`None`. |
| """ |
| args = [] |
| def servername(conn): |
| args.append((conn, conn.get_servername())) |
| context = Context(TLSv1_METHOD) |
| context.set_tlsext_servername_callback(servername) |
| |
| # Lose our reference to it. The Context is responsible for keeping it |
| # alive now. |
| del servername |
| collect() |
| |
| # Necessary to actually accept the connection |
| context.use_privatekey(load_privatekey(FILETYPE_PEM, server_key_pem)) |
| context.use_certificate(load_certificate(FILETYPE_PEM, server_cert_pem)) |
| |
| # Do a little connection to trigger the logic |
| server = Connection(context, None) |
| server.set_accept_state() |
| |
| client = Connection(Context(TLSv1_METHOD), None) |
| client.set_connect_state() |
| |
| self._interactInMemory(server, client) |
| |
| self.assertEqual([(server, None)], args) |
| |
| |
| def test_servername(self): |
| """ |
| When a client specifies a server name in its hello message, the callback |
| passed to :py:obj:`Contexts.set_tlsext_servername_callback` is invoked and the |
| result of :py:obj:`Connection.get_servername` is that server name. |
| """ |
| args = [] |
| def servername(conn): |
| args.append((conn, conn.get_servername())) |
| context = Context(TLSv1_METHOD) |
| context.set_tlsext_servername_callback(servername) |
| |
| # Necessary to actually accept the connection |
| context.use_privatekey(load_privatekey(FILETYPE_PEM, server_key_pem)) |
| context.use_certificate(load_certificate(FILETYPE_PEM, server_cert_pem)) |
| |
| # Do a little connection to trigger the logic |
| server = Connection(context, None) |
| server.set_accept_state() |
| |
| client = Connection(Context(TLSv1_METHOD), None) |
| client.set_connect_state() |
| client.set_tlsext_host_name(b("foo1.example.com")) |
| |
| self._interactInMemory(server, client) |
| |
| self.assertEqual([(server, b("foo1.example.com"))], args) |
| |
| |
| |
| class SessionTests(TestCase): |
| """ |
| Unit tests for :py:obj:`OpenSSL.SSL.Session`. |
| """ |
| def test_construction(self): |
| """ |
| :py:class:`Session` can be constructed with no arguments, creating a new |
| instance of that type. |
| """ |
| new_session = Session() |
| self.assertTrue(isinstance(new_session, Session)) |
| |
| |
| def test_construction_wrong_args(self): |
| """ |
| If any arguments are passed to :py:class:`Session`, :py:obj:`TypeError` |
| is raised. |
| """ |
| self.assertRaises(TypeError, Session, 123) |
| self.assertRaises(TypeError, Session, "hello") |
| self.assertRaises(TypeError, Session, object()) |
| |
| |
| |
| class ConnectionTests(TestCase, _LoopbackMixin): |
| """ |
| Unit tests for :py:obj:`OpenSSL.SSL.Connection`. |
| """ |
| # XXX get_peer_certificate -> None |
| # XXX sock_shutdown |
| # XXX master_key -> TypeError |
| # XXX server_random -> TypeError |
| # XXX state_string |
| # XXX connect -> TypeError |
| # XXX connect_ex -> TypeError |
| # XXX set_connect_state -> TypeError |
| # XXX set_accept_state -> TypeError |
| # XXX renegotiate_pending |
| # XXX do_handshake -> TypeError |
| # XXX bio_read -> TypeError |
| # XXX recv -> TypeError |
| # XXX send -> TypeError |
| # XXX bio_write -> TypeError |
| |
| def test_type(self): |
| """ |
| :py:obj:`Connection` and :py:obj:`ConnectionType` refer to the same type object and |
| can be used to create instances of that type. |
| """ |
| self.assertIdentical(Connection, ConnectionType) |
| ctx = Context(TLSv1_METHOD) |
| self.assertConsistentType(Connection, 'Connection', ctx, None) |
| |
| |
| def test_get_context(self): |
| """ |
| :py:obj:`Connection.get_context` returns the :py:obj:`Context` instance used to |
| construct the :py:obj:`Connection` instance. |
| """ |
| context = Context(TLSv1_METHOD) |
| connection = Connection(context, None) |
| self.assertIdentical(connection.get_context(), context) |
| |
| |
| def test_get_context_wrong_args(self): |
| """ |
| :py:obj:`Connection.get_context` raises :py:obj:`TypeError` if called with any |
| arguments. |
| """ |
| connection = Connection(Context(TLSv1_METHOD), None) |
| self.assertRaises(TypeError, connection.get_context, None) |
| |
| |
| def test_set_context_wrong_args(self): |
| """ |
| :py:obj:`Connection.set_context` raises :py:obj:`TypeError` if called with a |
| non-:py:obj:`Context` instance argument or with any number of arguments other |
| than 1. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| connection = Connection(ctx, None) |
| self.assertRaises(TypeError, connection.set_context) |
| self.assertRaises(TypeError, connection.set_context, object()) |
| self.assertRaises(TypeError, connection.set_context, "hello") |
| self.assertRaises(TypeError, connection.set_context, 1) |
| self.assertRaises(TypeError, connection.set_context, 1, 2) |
| self.assertRaises( |
| TypeError, connection.set_context, Context(TLSv1_METHOD), 2) |
| self.assertIdentical(ctx, connection.get_context()) |
| |
| |
| def test_set_context(self): |
| """ |
| :py:obj:`Connection.set_context` specifies a new :py:obj:`Context` instance to be used |
| for the connection. |
| """ |
| original = Context(SSLv23_METHOD) |
| replacement = Context(TLSv1_METHOD) |
| connection = Connection(original, None) |
| connection.set_context(replacement) |
| self.assertIdentical(replacement, connection.get_context()) |
| # Lose our references to the contexts, just in case the Connection isn't |
| # properly managing its own contributions to their reference counts. |
| del original, replacement |
| collect() |
| |
| |
| def test_set_tlsext_host_name_wrong_args(self): |
| """ |
| If :py:obj:`Connection.set_tlsext_host_name` is called with a non-byte string |
| argument or a byte string with an embedded NUL or other than one |
| argument, :py:obj:`TypeError` is raised. |
| """ |
| conn = Connection(Context(TLSv1_METHOD), None) |
| self.assertRaises(TypeError, conn.set_tlsext_host_name) |
| self.assertRaises(TypeError, conn.set_tlsext_host_name, object()) |
| self.assertRaises(TypeError, conn.set_tlsext_host_name, 123, 456) |
| self.assertRaises( |
| TypeError, conn.set_tlsext_host_name, b("with\0null")) |
| |
| if version_info >= (3,): |
| # On Python 3.x, don't accidentally implicitly convert from text. |
| self.assertRaises( |
| TypeError, |
| conn.set_tlsext_host_name, b("example.com").decode("ascii")) |
| |
| |
| def test_get_servername_wrong_args(self): |
| """ |
| :py:obj:`Connection.get_servername` raises :py:obj:`TypeError` if called with any |
| arguments. |
| """ |
| connection = Connection(Context(TLSv1_METHOD), None) |
| self.assertRaises(TypeError, connection.get_servername, object()) |
| self.assertRaises(TypeError, connection.get_servername, 1) |
| self.assertRaises(TypeError, connection.get_servername, "hello") |
| |
| |
| def test_pending(self): |
| """ |
| :py:obj:`Connection.pending` returns the number of bytes available for |
| immediate read. |
| """ |
| connection = Connection(Context(TLSv1_METHOD), None) |
| self.assertEquals(connection.pending(), 0) |
| |
| |
| def test_pending_wrong_args(self): |
| """ |
| :py:obj:`Connection.pending` raises :py:obj:`TypeError` if called with any arguments. |
| """ |
| connection = Connection(Context(TLSv1_METHOD), None) |
| self.assertRaises(TypeError, connection.pending, None) |
| |
| |
| def test_connect_wrong_args(self): |
| """ |
| :py:obj:`Connection.connect` raises :py:obj:`TypeError` if called with a non-address |
| argument or with the wrong number of arguments. |
| """ |
| connection = Connection(Context(TLSv1_METHOD), socket()) |
| self.assertRaises(TypeError, connection.connect, None) |
| self.assertRaises(TypeError, connection.connect) |
| self.assertRaises(TypeError, connection.connect, ("127.0.0.1", 1), None) |
| |
| |
| def test_connect_refused(self): |
| """ |
| :py:obj:`Connection.connect` raises :py:obj:`socket.error` if the underlying socket |
| connect method raises it. |
| """ |
| client = socket() |
| context = Context(TLSv1_METHOD) |
| clientSSL = Connection(context, client) |
| exc = self.assertRaises(error, clientSSL.connect, ("127.0.0.1", 1)) |
| self.assertEquals(exc.args[0], ECONNREFUSED) |
| |
| |
| def test_connect(self): |
| """ |
| :py:obj:`Connection.connect` establishes a connection to the specified address. |
| """ |
| port = socket() |
| port.bind(('', 0)) |
| port.listen(3) |
| |
| clientSSL = Connection(Context(TLSv1_METHOD), socket()) |
| clientSSL.connect(('127.0.0.1', port.getsockname()[1])) |
| # XXX An assertion? Or something? |
| |
| |
| if platform == "darwin": |
| "connect_ex sometimes causes a kernel panic on OS X 10.6.4" |
| else: |
| def test_connect_ex(self): |
| """ |
| If there is a connection error, :py:obj:`Connection.connect_ex` returns the |
| errno instead of raising an exception. |
| """ |
| port = socket() |
| port.bind(('', 0)) |
| port.listen(3) |
| |
| clientSSL = Connection(Context(TLSv1_METHOD), socket()) |
| clientSSL.setblocking(False) |
| result = clientSSL.connect_ex(port.getsockname()) |
| expected = (EINPROGRESS, EWOULDBLOCK) |
| self.assertTrue( |
| result in expected, "%r not in %r" % (result, expected)) |
| |
| |
| def test_accept_wrong_args(self): |
| """ |
| :py:obj:`Connection.accept` raises :py:obj:`TypeError` if called with any arguments. |
| """ |
| connection = Connection(Context(TLSv1_METHOD), socket()) |
| self.assertRaises(TypeError, connection.accept, None) |
| |
| |
| def test_accept(self): |
| """ |
| :py:obj:`Connection.accept` accepts a pending connection attempt and returns a |
| tuple of a new :py:obj:`Connection` (the accepted client) and the address the |
| connection originated from. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| ctx.use_privatekey(load_privatekey(FILETYPE_PEM, server_key_pem)) |
| ctx.use_certificate(load_certificate(FILETYPE_PEM, server_cert_pem)) |
| port = socket() |
| portSSL = Connection(ctx, port) |
| portSSL.bind(('', 0)) |
| portSSL.listen(3) |
| |
| clientSSL = Connection(Context(TLSv1_METHOD), socket()) |
| |
| # Calling portSSL.getsockname() here to get the server IP address sounds |
| # great, but frequently fails on Windows. |
| clientSSL.connect(('127.0.0.1', portSSL.getsockname()[1])) |
| |
| serverSSL, address = portSSL.accept() |
| |
| self.assertTrue(isinstance(serverSSL, Connection)) |
| self.assertIdentical(serverSSL.get_context(), ctx) |
| self.assertEquals(address, clientSSL.getsockname()) |
| |
| |
| def test_shutdown_wrong_args(self): |
| """ |
| :py:obj:`Connection.shutdown` raises :py:obj:`TypeError` if called with the wrong |
| number of arguments or with arguments other than integers. |
| """ |
| connection = Connection(Context(TLSv1_METHOD), None) |
| self.assertRaises(TypeError, connection.shutdown, None) |
| self.assertRaises(TypeError, connection.get_shutdown, None) |
| self.assertRaises(TypeError, connection.set_shutdown) |
| self.assertRaises(TypeError, connection.set_shutdown, None) |
| self.assertRaises(TypeError, connection.set_shutdown, 0, 1) |
| |
| |
| def test_shutdown(self): |
| """ |
| :py:obj:`Connection.shutdown` performs an SSL-level connection shutdown. |
| """ |
| server, client = self._loopback() |
| self.assertFalse(server.shutdown()) |
| self.assertEquals(server.get_shutdown(), SENT_SHUTDOWN) |
| self.assertRaises(ZeroReturnError, client.recv, 1024) |
| self.assertEquals(client.get_shutdown(), RECEIVED_SHUTDOWN) |
| client.shutdown() |
| self.assertEquals(client.get_shutdown(), SENT_SHUTDOWN|RECEIVED_SHUTDOWN) |
| self.assertRaises(ZeroReturnError, server.recv, 1024) |
| self.assertEquals(server.get_shutdown(), SENT_SHUTDOWN|RECEIVED_SHUTDOWN) |
| |
| |
| def test_set_shutdown(self): |
| """ |
| :py:obj:`Connection.set_shutdown` sets the state of the SSL connection shutdown |
| process. |
| """ |
| connection = Connection(Context(TLSv1_METHOD), socket()) |
| connection.set_shutdown(RECEIVED_SHUTDOWN) |
| self.assertEquals(connection.get_shutdown(), RECEIVED_SHUTDOWN) |
| |
| |
| if not PY3: |
| def test_set_shutdown_long(self): |
| """ |
| On Python 2 :py:obj:`Connection.set_shutdown` accepts an argument |
| of type :py:obj:`long` as well as :py:obj:`int`. |
| """ |
| connection = Connection(Context(TLSv1_METHOD), socket()) |
| connection.set_shutdown(long(RECEIVED_SHUTDOWN)) |
| self.assertEquals(connection.get_shutdown(), RECEIVED_SHUTDOWN) |
| |
| |
| def test_app_data_wrong_args(self): |
| """ |
| :py:obj:`Connection.set_app_data` raises :py:obj:`TypeError` if called with other than |
| one argument. :py:obj:`Connection.get_app_data` raises :py:obj:`TypeError` if called |
| with any arguments. |
| """ |
| conn = Connection(Context(TLSv1_METHOD), None) |
| self.assertRaises(TypeError, conn.get_app_data, None) |
| self.assertRaises(TypeError, conn.set_app_data) |
| self.assertRaises(TypeError, conn.set_app_data, None, None) |
| |
| |
| def test_app_data(self): |
| """ |
| Any object can be set as app data by passing it to |
| :py:obj:`Connection.set_app_data` and later retrieved with |
| :py:obj:`Connection.get_app_data`. |
| """ |
| conn = Connection(Context(TLSv1_METHOD), None) |
| app_data = object() |
| conn.set_app_data(app_data) |
| self.assertIdentical(conn.get_app_data(), app_data) |
| |
| |
| def test_makefile(self): |
| """ |
| :py:obj:`Connection.makefile` is not implemented and calling that method raises |
| :py:obj:`NotImplementedError`. |
| """ |
| conn = Connection(Context(TLSv1_METHOD), None) |
| self.assertRaises(NotImplementedError, conn.makefile) |
| |
| |
| def test_get_peer_cert_chain_wrong_args(self): |
| """ |
| :py:obj:`Connection.get_peer_cert_chain` raises :py:obj:`TypeError` if called with any |
| arguments. |
| """ |
| conn = Connection(Context(TLSv1_METHOD), None) |
| self.assertRaises(TypeError, conn.get_peer_cert_chain, 1) |
| self.assertRaises(TypeError, conn.get_peer_cert_chain, "foo") |
| self.assertRaises(TypeError, conn.get_peer_cert_chain, object()) |
| self.assertRaises(TypeError, conn.get_peer_cert_chain, []) |
| |
| |
| def test_get_peer_cert_chain(self): |
| """ |
| :py:obj:`Connection.get_peer_cert_chain` returns a list of certificates which |
| the connected server returned for the certification verification. |
| """ |
| chain = _create_certificate_chain() |
| [(cakey, cacert), (ikey, icert), (skey, scert)] = chain |
| |
| serverContext = Context(TLSv1_METHOD) |
| serverContext.use_privatekey(skey) |
| serverContext.use_certificate(scert) |
| serverContext.add_extra_chain_cert(icert) |
| serverContext.add_extra_chain_cert(cacert) |
| server = Connection(serverContext, None) |
| server.set_accept_state() |
| |
| # Create the client |
| clientContext = Context(TLSv1_METHOD) |
| clientContext.set_verify(VERIFY_NONE, verify_cb) |
| client = Connection(clientContext, None) |
| client.set_connect_state() |
| |
| self._interactInMemory(client, server) |
| |
| chain = client.get_peer_cert_chain() |
| self.assertEqual(len(chain), 3) |
| self.assertEqual( |
| "Server Certificate", chain[0].get_subject().CN) |
| self.assertEqual( |
| "Intermediate Certificate", chain[1].get_subject().CN) |
| self.assertEqual( |
| "Authority Certificate", chain[2].get_subject().CN) |
| |
| |
| def test_get_peer_cert_chain_none(self): |
| """ |
| :py:obj:`Connection.get_peer_cert_chain` returns :py:obj:`None` if the peer sends no |
| certificate chain. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| ctx.use_privatekey(load_privatekey(FILETYPE_PEM, server_key_pem)) |
| ctx.use_certificate(load_certificate(FILETYPE_PEM, server_cert_pem)) |
| server = Connection(ctx, None) |
| server.set_accept_state() |
| client = Connection(Context(TLSv1_METHOD), None) |
| client.set_connect_state() |
| self._interactInMemory(client, server) |
| self.assertIdentical(None, server.get_peer_cert_chain()) |
| |
| |
| def test_get_session_wrong_args(self): |
| """ |
| :py:obj:`Connection.get_session` raises :py:obj:`TypeError` if called |
| with any arguments. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| server = Connection(ctx, None) |
| self.assertRaises(TypeError, server.get_session, 123) |
| self.assertRaises(TypeError, server.get_session, "hello") |
| self.assertRaises(TypeError, server.get_session, object()) |
| |
| |
| def test_get_session_unconnected(self): |
| """ |
| :py:obj:`Connection.get_session` returns :py:obj:`None` when used with |
| an object which has not been connected. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| server = Connection(ctx, None) |
| session = server.get_session() |
| self.assertIdentical(None, session) |
| |
| |
| def test_server_get_session(self): |
| """ |
| On the server side of a connection, :py:obj:`Connection.get_session` |
| returns a :py:class:`Session` instance representing the SSL session for |
| that connection. |
| """ |
| server, client = self._loopback() |
| session = server.get_session() |
| self.assertIsInstance(session, Session) |
| |
| |
| def test_client_get_session(self): |
| """ |
| On the client side of a connection, :py:obj:`Connection.get_session` |
| returns a :py:class:`Session` instance representing the SSL session for |
| that connection. |
| """ |
| server, client = self._loopback() |
| session = client.get_session() |
| self.assertIsInstance(session, Session) |
| |
| |
| def test_set_session_wrong_args(self): |
| """ |
| If called with an object that is not an instance of :py:class:`Session`, |
| or with other than one argument, :py:obj:`Connection.set_session` raises |
| :py:obj:`TypeError`. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| connection = Connection(ctx, None) |
| self.assertRaises(TypeError, connection.set_session) |
| self.assertRaises(TypeError, connection.set_session, 123) |
| self.assertRaises(TypeError, connection.set_session, "hello") |
| self.assertRaises(TypeError, connection.set_session, object()) |
| self.assertRaises( |
| TypeError, connection.set_session, Session(), Session()) |
| |
| |
| def test_client_set_session(self): |
| """ |
| :py:obj:`Connection.set_session`, when used prior to a connection being |
| established, accepts a :py:class:`Session` instance and causes an |
| attempt to re-use the session it represents when the SSL handshake is |
| performed. |
| """ |
| key = load_privatekey(FILETYPE_PEM, server_key_pem) |
| cert = load_certificate(FILETYPE_PEM, server_cert_pem) |
| ctx = Context(TLSv1_METHOD) |
| ctx.use_privatekey(key) |
| ctx.use_certificate(cert) |
| ctx.set_session_id("unity-test") |
| |
| def makeServer(socket): |
| server = Connection(ctx, socket) |
| server.set_accept_state() |
| return server |
| |
| originalServer, originalClient = self._loopback( |
| serverFactory=makeServer) |
| originalSession = originalClient.get_session() |
| |
| def makeClient(socket): |
| client = self._loopbackClientFactory(socket) |
| client.set_session(originalSession) |
| return client |
| resumedServer, resumedClient = self._loopback( |
| serverFactory=makeServer, |
| clientFactory=makeClient) |
| |
| # This is a proxy: in general, we have no access to any unique |
| # identifier for the session (new enough versions of OpenSSL expose a |
| # hash which could be usable, but "new enough" is very, very new). |
| # Instead, exploit the fact that the master key is re-used if the |
| # session is re-used. As long as the master key for the two connections |
| # is the same, the session was re-used! |
| self.assertEqual( |
| originalServer.master_key(), resumedServer.master_key()) |
| |
| |
| def test_set_session_wrong_method(self): |
| """ |
| If :py:obj:`Connection.set_session` is passed a :py:class:`Session` |
| instance associated with a context using a different SSL method than the |
| :py:obj:`Connection` is using, a :py:class:`OpenSSL.SSL.Error` is |
| raised. |
| """ |
| key = load_privatekey(FILETYPE_PEM, server_key_pem) |
| cert = load_certificate(FILETYPE_PEM, server_cert_pem) |
| ctx = Context(TLSv1_METHOD) |
| ctx.use_privatekey(key) |
| ctx.use_certificate(cert) |
| ctx.set_session_id("unity-test") |
| |
| def makeServer(socket): |
| server = Connection(ctx, socket) |
| server.set_accept_state() |
| return server |
| |
| originalServer, originalClient = self._loopback( |
| serverFactory=makeServer) |
| originalSession = originalClient.get_session() |
| |
| def makeClient(socket): |
| # Intentionally use a different, incompatible method here. |
| client = Connection(Context(SSLv3_METHOD), socket) |
| client.set_connect_state() |
| client.set_session(originalSession) |
| return client |
| |
| self.assertRaises( |
| Error, |
| self._loopback, clientFactory=makeClient, serverFactory=makeServer) |
| |
| |
| def test_wantWriteError(self): |
| """ |
| :py:obj:`Connection` methods which generate output raise |
| :py:obj:`OpenSSL.SSL.WantWriteError` if writing to the connection's BIO |
| fail indicating a should-write state. |
| """ |
| client_socket, server_socket = socket_pair() |
| # Fill up the client's send buffer so Connection won't be able to write |
| # anything. |
| msg = b"x" * 512 |
| for i in range(2048): |
| try: |
| client_socket.send(msg) |
| except error as e: |
| if e.errno == EWOULDBLOCK: |
| break |
| raise |
| else: |
| self.fail( |
| "Failed to fill socket buffer, cannot test BIO want write") |
| |
| ctx = Context(TLSv1_METHOD) |
| conn = Connection(ctx, client_socket) |
| # Client's speak first, so make it an SSL client |
| conn.set_connect_state() |
| self.assertRaises(WantWriteError, conn.do_handshake) |
| |
| # XXX want_read |
| |
| def test_get_finished_before_connect(self): |
| """ |
| :py:obj:`Connection.get_finished` returns :py:obj:`None` before TLS |
| handshake is completed. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| connection = Connection(ctx, None) |
| self.assertEqual(connection.get_finished(), None) |
| |
| |
| def test_get_peer_finished_before_connect(self): |
| """ |
| :py:obj:`Connection.get_peer_finished` returns :py:obj:`None` before |
| TLS handshake is completed. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| connection = Connection(ctx, None) |
| self.assertEqual(connection.get_peer_finished(), None) |
| |
| |
| def test_get_finished(self): |
| """ |
| :py:obj:`Connection.get_finished` method returns the TLS Finished |
| message send from client, or server. Finished messages are send during |
| TLS handshake. |
| """ |
| |
| server, client = self._loopback() |
| |
| self.assertNotEqual(server.get_finished(), None) |
| self.assertTrue(len(server.get_finished()) > 0) |
| |
| |
| def test_get_peer_finished(self): |
| """ |
| :py:obj:`Connection.get_peer_finished` method returns the TLS Finished |
| message received from client, or server. Finished messages are send |
| during TLS handshake. |
| """ |
| server, client = self._loopback() |
| |
| self.assertNotEqual(server.get_peer_finished(), None) |
| self.assertTrue(len(server.get_peer_finished()) > 0) |
| |
| |
| def test_tls_finished_message_symmetry(self): |
| """ |
| The TLS Finished message send by server must be the TLS Finished message |
| received by client. |
| |
| The TLS Finished message send by client must be the TLS Finished message |
| received by server. |
| """ |
| server, client = self._loopback() |
| |
| self.assertEqual(server.get_finished(), client.get_peer_finished()) |
| self.assertEqual(client.get_finished(), server.get_peer_finished()) |
| |
| |
| def test_get_cipher_name_before_connect(self): |
| """ |
| :py:obj:`Connection.get_cipher_name` returns :py:obj:`None` if no |
| connection has been established. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| conn = Connection(ctx, None) |
| self.assertIdentical(conn.get_cipher_name(), None) |
| |
| |
| def test_get_cipher_name(self): |
| """ |
| :py:obj:`Connection.get_cipher_name` returns a :py:class:`unicode` |
| string giving the name of the currently used cipher. |
| """ |
| server, client = self._loopback() |
| server_cipher_name, client_cipher_name = \ |
| server.get_cipher_name(), client.get_cipher_name() |
| |
| self.assertIsInstance(server_cipher_name, text_type) |
| self.assertIsInstance(client_cipher_name, text_type) |
| |
| self.assertEqual(server_cipher_name, client_cipher_name) |
| |
| |
| def test_get_cipher_version_before_connect(self): |
| """ |
| :py:obj:`Connection.get_cipher_version` returns :py:obj:`None` if no |
| connection has been established. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| conn = Connection(ctx, None) |
| self.assertIdentical(conn.get_cipher_version(), None) |
| |
| |
| def test_get_cipher_version(self): |
| """ |
| :py:obj:`Connection.get_cipher_version` returns a :py:class:`unicode` |
| string giving the protocol name of the currently used cipher. |
| """ |
| server, client = self._loopback() |
| server_cipher_version, client_cipher_version = \ |
| server.get_cipher_version(), client.get_cipher_version() |
| |
| self.assertIsInstance(server_cipher_version, text_type) |
| self.assertIsInstance(client_cipher_version, text_type) |
| |
| self.assertEqual(server_cipher_version, client_cipher_version) |
| |
| |
| def test_get_cipher_bits_before_connect(self): |
| """ |
| :py:obj:`Connection.get_cipher_bits` returns :py:obj:`None` if no |
| connection has been established. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| conn = Connection(ctx, None) |
| self.assertIdentical(conn.get_cipher_bits(), None) |
| |
| |
| def test_get_cipher_bits(self): |
| """ |
| :py:obj:`Connection.get_cipher_bits` returns the number of secret bits |
| of the currently used cipher. |
| """ |
| server, client = self._loopback() |
| server_cipher_bits, client_cipher_bits = \ |
| server.get_cipher_bits(), client.get_cipher_bits() |
| |
| self.assertIsInstance(server_cipher_bits, int) |
| self.assertIsInstance(client_cipher_bits, int) |
| |
| self.assertEqual(server_cipher_bits, client_cipher_bits) |
| |
| |
| |
| class ConnectionGetCipherListTests(TestCase): |
| """ |
| Tests for :py:obj:`Connection.get_cipher_list`. |
| """ |
| def test_wrong_args(self): |
| """ |
| :py:obj:`Connection.get_cipher_list` raises :py:obj:`TypeError` if called with any |
| arguments. |
| """ |
| connection = Connection(Context(TLSv1_METHOD), None) |
| self.assertRaises(TypeError, connection.get_cipher_list, None) |
| |
| |
| def test_result(self): |
| """ |
| :py:obj:`Connection.get_cipher_list` returns a :py:obj:`list` of |
| :py:obj:`bytes` giving the names of the ciphers which might be used. |
| """ |
| connection = Connection(Context(TLSv1_METHOD), None) |
| ciphers = connection.get_cipher_list() |
| self.assertTrue(isinstance(ciphers, list)) |
| for cipher in ciphers: |
| self.assertTrue(isinstance(cipher, str)) |
| |
| |
| |
| class ConnectionSendTests(TestCase, _LoopbackMixin): |
| """ |
| Tests for :py:obj:`Connection.send` |
| """ |
| def test_wrong_args(self): |
| """ |
| When called with arguments other than string argument for its first |
| parameter or more than two arguments, :py:obj:`Connection.send` raises |
| :py:obj:`TypeError`. |
| """ |
| connection = Connection(Context(TLSv1_METHOD), None) |
| self.assertRaises(TypeError, connection.send) |
| self.assertRaises(TypeError, connection.send, object()) |
| self.assertRaises(TypeError, connection.send, "foo", object(), "bar") |
| |
| |
| def test_short_bytes(self): |
| """ |
| When passed a short byte string, :py:obj:`Connection.send` transmits all of it |
| and returns the number of bytes sent. |
| """ |
| server, client = self._loopback() |
| count = server.send(b('xy')) |
| self.assertEquals(count, 2) |
| self.assertEquals(client.recv(2), b('xy')) |
| |
| try: |
| memoryview |
| except NameError: |
| "cannot test sending memoryview without memoryview" |
| else: |
| def test_short_memoryview(self): |
| """ |
| When passed a memoryview onto a small number of bytes, |
| :py:obj:`Connection.send` transmits all of them and returns the number of |
| bytes sent. |
| """ |
| server, client = self._loopback() |
| count = server.send(memoryview(b('xy'))) |
| self.assertEquals(count, 2) |
| self.assertEquals(client.recv(2), b('xy')) |
| |
| |
| |
| class ConnectionSendallTests(TestCase, _LoopbackMixin): |
| """ |
| Tests for :py:obj:`Connection.sendall`. |
| """ |
| def test_wrong_args(self): |
| """ |
| When called with arguments other than a string argument for its first |
| parameter or with more than two arguments, :py:obj:`Connection.sendall` |
| raises :py:obj:`TypeError`. |
| """ |
| connection = Connection(Context(TLSv1_METHOD), None) |
| self.assertRaises(TypeError, connection.sendall) |
| self.assertRaises(TypeError, connection.sendall, object()) |
| self.assertRaises( |
| TypeError, connection.sendall, "foo", object(), "bar") |
| |
| |
| def test_short(self): |
| """ |
| :py:obj:`Connection.sendall` transmits all of the bytes in the string passed to |
| it. |
| """ |
| server, client = self._loopback() |
| server.sendall(b('x')) |
| self.assertEquals(client.recv(1), b('x')) |
| |
| |
| try: |
| memoryview |
| except NameError: |
| "cannot test sending memoryview without memoryview" |
| else: |
| def test_short_memoryview(self): |
| """ |
| When passed a memoryview onto a small number of bytes, |
| :py:obj:`Connection.sendall` transmits all of them. |
| """ |
| server, client = self._loopback() |
| server.sendall(memoryview(b('x'))) |
| self.assertEquals(client.recv(1), b('x')) |
| |
| |
| def test_long(self): |
| """ |
| :py:obj:`Connection.sendall` transmits all of the bytes in the string passed to |
| it even if this requires multiple calls of an underlying write function. |
| """ |
| server, client = self._loopback() |
| # Should be enough, underlying SSL_write should only do 16k at a time. |
| # On Windows, after 32k of bytes the write will block (forever - because |
| # no one is yet reading). |
| message = b('x') * (1024 * 32 - 1) + b('y') |
| server.sendall(message) |
| accum = [] |
| received = 0 |
| while received < len(message): |
| data = client.recv(1024) |
| accum.append(data) |
| received += len(data) |
| self.assertEquals(message, b('').join(accum)) |
| |
| |
| def test_closed(self): |
| """ |
| If the underlying socket is closed, :py:obj:`Connection.sendall` propagates the |
| write error from the low level write call. |
| """ |
| server, client = self._loopback() |
| server.sock_shutdown(2) |
| exc = self.assertRaises(SysCallError, server.sendall, b"hello, world") |
| if platform == "win32": |
| self.assertEqual(exc.args[0], ESHUTDOWN) |
| else: |
| self.assertEqual(exc.args[0], EPIPE) |
| |
| |
| |
| class ConnectionRenegotiateTests(TestCase, _LoopbackMixin): |
| """ |
| Tests for SSL renegotiation APIs. |
| """ |
| def test_renegotiate_wrong_args(self): |
| """ |
| :py:obj:`Connection.renegotiate` raises :py:obj:`TypeError` if called with any |
| arguments. |
| """ |
| connection = Connection(Context(TLSv1_METHOD), None) |
| self.assertRaises(TypeError, connection.renegotiate, None) |
| |
| |
| def test_total_renegotiations_wrong_args(self): |
| """ |
| :py:obj:`Connection.total_renegotiations` raises :py:obj:`TypeError` if called with |
| any arguments. |
| """ |
| connection = Connection(Context(TLSv1_METHOD), None) |
| self.assertRaises(TypeError, connection.total_renegotiations, None) |
| |
| |
| def test_total_renegotiations(self): |
| """ |
| :py:obj:`Connection.total_renegotiations` returns :py:obj:`0` before any |
| renegotiations have happened. |
| """ |
| connection = Connection(Context(TLSv1_METHOD), None) |
| self.assertEquals(connection.total_renegotiations(), 0) |
| |
| |
| # def test_renegotiate(self): |
| # """ |
| # """ |
| # server, client = self._loopback() |
| |
| # server.send("hello world") |
| # self.assertEquals(client.recv(len("hello world")), "hello world") |
| |
| # self.assertEquals(server.total_renegotiations(), 0) |
| # self.assertTrue(server.renegotiate()) |
| |
| # server.setblocking(False) |
| # client.setblocking(False) |
| # while server.renegotiate_pending(): |
| # client.do_handshake() |
| # server.do_handshake() |
| |
| # self.assertEquals(server.total_renegotiations(), 1) |
| |
| |
| |
| |
| class ErrorTests(TestCase): |
| """ |
| Unit tests for :py:obj:`OpenSSL.SSL.Error`. |
| """ |
| def test_type(self): |
| """ |
| :py:obj:`Error` is an exception type. |
| """ |
| self.assertTrue(issubclass(Error, Exception)) |
| self.assertEqual(Error.__name__, 'Error') |
| |
| |
| |
| class ConstantsTests(TestCase): |
| """ |
| Tests for the values of constants exposed in :py:obj:`OpenSSL.SSL`. |
| |
| These are values defined by OpenSSL intended only to be used as flags to |
| OpenSSL APIs. The only assertions it seems can be made about them is |
| their values. |
| """ |
| # unittest.TestCase has no skip mechanism |
| if OP_NO_QUERY_MTU is not None: |
| def test_op_no_query_mtu(self): |
| """ |
| The value of :py:obj:`OpenSSL.SSL.OP_NO_QUERY_MTU` is 0x1000, the value of |
| :py:const:`SSL_OP_NO_QUERY_MTU` defined by :file:`openssl/ssl.h`. |
| """ |
| self.assertEqual(OP_NO_QUERY_MTU, 0x1000) |
| else: |
| "OP_NO_QUERY_MTU unavailable - OpenSSL version may be too old" |
| |
| |
| if OP_COOKIE_EXCHANGE is not None: |
| def test_op_cookie_exchange(self): |
| """ |
| The value of :py:obj:`OpenSSL.SSL.OP_COOKIE_EXCHANGE` is 0x2000, the value |
| of :py:const:`SSL_OP_COOKIE_EXCHANGE` defined by :file:`openssl/ssl.h`. |
| """ |
| self.assertEqual(OP_COOKIE_EXCHANGE, 0x2000) |
| else: |
| "OP_COOKIE_EXCHANGE unavailable - OpenSSL version may be too old" |
| |
| |
| if OP_NO_TICKET is not None: |
| def test_op_no_ticket(self): |
| """ |
| The value of :py:obj:`OpenSSL.SSL.OP_NO_TICKET` is 0x4000, the value of |
| :py:const:`SSL_OP_NO_TICKET` defined by :file:`openssl/ssl.h`. |
| """ |
| self.assertEqual(OP_NO_TICKET, 0x4000) |
| else: |
| "OP_NO_TICKET unavailable - OpenSSL version may be too old" |
| |
| |
| if OP_NO_COMPRESSION is not None: |
| def test_op_no_compression(self): |
| """ |
| The value of :py:obj:`OpenSSL.SSL.OP_NO_COMPRESSION` is 0x20000, the value |
| of :py:const:`SSL_OP_NO_COMPRESSION` defined by :file:`openssl/ssl.h`. |
| """ |
| self.assertEqual(OP_NO_COMPRESSION, 0x20000) |
| else: |
| "OP_NO_COMPRESSION unavailable - OpenSSL version may be too old" |
| |
| |
| def test_sess_cache_off(self): |
| """ |
| The value of :py:obj:`OpenSSL.SSL.SESS_CACHE_OFF` 0x0, the value of |
| :py:obj:`SSL_SESS_CACHE_OFF` defined by ``openssl/ssl.h``. |
| """ |
| self.assertEqual(0x0, SESS_CACHE_OFF) |
| |
| |
| def test_sess_cache_client(self): |
| """ |
| The value of :py:obj:`OpenSSL.SSL.SESS_CACHE_CLIENT` 0x1, the value of |
| :py:obj:`SSL_SESS_CACHE_CLIENT` defined by ``openssl/ssl.h``. |
| """ |
| self.assertEqual(0x1, SESS_CACHE_CLIENT) |
| |
| |
| def test_sess_cache_server(self): |
| """ |
| The value of :py:obj:`OpenSSL.SSL.SESS_CACHE_SERVER` 0x2, the value of |
| :py:obj:`SSL_SESS_CACHE_SERVER` defined by ``openssl/ssl.h``. |
| """ |
| self.assertEqual(0x2, SESS_CACHE_SERVER) |
| |
| |
| def test_sess_cache_both(self): |
| """ |
| The value of :py:obj:`OpenSSL.SSL.SESS_CACHE_BOTH` 0x3, the value of |
| :py:obj:`SSL_SESS_CACHE_BOTH` defined by ``openssl/ssl.h``. |
| """ |
| self.assertEqual(0x3, SESS_CACHE_BOTH) |
| |
| |
| def test_sess_cache_no_auto_clear(self): |
| """ |
| The value of :py:obj:`OpenSSL.SSL.SESS_CACHE_NO_AUTO_CLEAR` 0x80, the |
| value of :py:obj:`SSL_SESS_CACHE_NO_AUTO_CLEAR` defined by |
| ``openssl/ssl.h``. |
| """ |
| self.assertEqual(0x80, SESS_CACHE_NO_AUTO_CLEAR) |
| |
| |
| def test_sess_cache_no_internal_lookup(self): |
| """ |
| The value of :py:obj:`OpenSSL.SSL.SESS_CACHE_NO_INTERNAL_LOOKUP` 0x100, |
| the value of :py:obj:`SSL_SESS_CACHE_NO_INTERNAL_LOOKUP` defined by |
| ``openssl/ssl.h``. |
| """ |
| self.assertEqual(0x100, SESS_CACHE_NO_INTERNAL_LOOKUP) |
| |
| |
| def test_sess_cache_no_internal_store(self): |
| """ |
| The value of :py:obj:`OpenSSL.SSL.SESS_CACHE_NO_INTERNAL_STORE` 0x200, |
| the value of :py:obj:`SSL_SESS_CACHE_NO_INTERNAL_STORE` defined by |
| ``openssl/ssl.h``. |
| """ |
| self.assertEqual(0x200, SESS_CACHE_NO_INTERNAL_STORE) |
| |
| |
| def test_sess_cache_no_internal(self): |
| """ |
| The value of :py:obj:`OpenSSL.SSL.SESS_CACHE_NO_INTERNAL` 0x300, the |
| value of :py:obj:`SSL_SESS_CACHE_NO_INTERNAL` defined by |
| ``openssl/ssl.h``. |
| """ |
| self.assertEqual(0x300, SESS_CACHE_NO_INTERNAL) |
| |
| |
| |
| class MemoryBIOTests(TestCase, _LoopbackMixin): |
| """ |
| Tests for :py:obj:`OpenSSL.SSL.Connection` using a memory BIO. |
| """ |
| def _server(self, sock): |
| """ |
| Create a new server-side SSL :py:obj:`Connection` object wrapped around |
| :py:obj:`sock`. |
| """ |
| # Create the server side Connection. This is mostly setup boilerplate |
| # - use TLSv1, use a particular certificate, etc. |
| server_ctx = Context(TLSv1_METHOD) |
| server_ctx.set_options(OP_NO_SSLv2 | OP_NO_SSLv3 | OP_SINGLE_DH_USE ) |
| server_ctx.set_verify(VERIFY_PEER|VERIFY_FAIL_IF_NO_PEER_CERT|VERIFY_CLIENT_ONCE, verify_cb) |
| server_store = server_ctx.get_cert_store() |
| server_ctx.use_privatekey(load_privatekey(FILETYPE_PEM, server_key_pem)) |
| server_ctx.use_certificate(load_certificate(FILETYPE_PEM, server_cert_pem)) |
| server_ctx.check_privatekey() |
| server_store.add_cert(load_certificate(FILETYPE_PEM, root_cert_pem)) |
| # Here the Connection is actually created. If None is passed as the 2nd |
| # parameter, it indicates a memory BIO should be created. |
| server_conn = Connection(server_ctx, sock) |
| server_conn.set_accept_state() |
| return server_conn |
| |
| |
| def _client(self, sock): |
| """ |
| Create a new client-side SSL :py:obj:`Connection` object wrapped around |
| :py:obj:`sock`. |
| """ |
| # Now create the client side Connection. Similar boilerplate to the |
| # above. |
| client_ctx = Context(TLSv1_METHOD) |
| client_ctx.set_options(OP_NO_SSLv2 | OP_NO_SSLv3 | OP_SINGLE_DH_USE ) |
| client_ctx.set_verify(VERIFY_PEER|VERIFY_FAIL_IF_NO_PEER_CERT|VERIFY_CLIENT_ONCE, verify_cb) |
| client_store = client_ctx.get_cert_store() |
| client_ctx.use_privatekey(load_privatekey(FILETYPE_PEM, client_key_pem)) |
| client_ctx.use_certificate(load_certificate(FILETYPE_PEM, client_cert_pem)) |
| client_ctx.check_privatekey() |
| client_store.add_cert(load_certificate(FILETYPE_PEM, root_cert_pem)) |
| client_conn = Connection(client_ctx, sock) |
| client_conn.set_connect_state() |
| return client_conn |
| |
| |
| def test_memoryConnect(self): |
| """ |
| Two :py:obj:`Connection`s which use memory BIOs can be manually connected by |
| reading from the output of each and writing those bytes to the input of |
| the other and in this way establish a connection and exchange |
| application-level bytes with each other. |
| """ |
| server_conn = self._server(None) |
| client_conn = self._client(None) |
| |
| # There should be no key or nonces yet. |
| self.assertIdentical(server_conn.master_key(), None) |
| self.assertIdentical(server_conn.client_random(), None) |
| self.assertIdentical(server_conn.server_random(), None) |
| |
| # First, the handshake needs to happen. We'll deliver bytes back and |
| # forth between the client and server until neither of them feels like |
| # speaking any more. |
| self.assertIdentical( |
| self._interactInMemory(client_conn, server_conn), None) |
| |
| # Now that the handshake is done, there should be a key and nonces. |
| self.assertNotIdentical(server_conn.master_key(), None) |
| self.assertNotIdentical(server_conn.client_random(), None) |
| self.assertNotIdentical(server_conn.server_random(), None) |
| self.assertEquals(server_conn.client_random(), client_conn.client_random()) |
| self.assertEquals(server_conn.server_random(), client_conn.server_random()) |
| self.assertNotEquals(server_conn.client_random(), server_conn.server_random()) |
| self.assertNotEquals(client_conn.client_random(), client_conn.server_random()) |
| |
| # Here are the bytes we'll try to send. |
| important_message = b('One if by land, two if by sea.') |
| |
| server_conn.write(important_message) |
| self.assertEquals( |
| self._interactInMemory(client_conn, server_conn), |
| (client_conn, important_message)) |
| |
| client_conn.write(important_message[::-1]) |
| self.assertEquals( |
| self._interactInMemory(client_conn, server_conn), |
| (server_conn, important_message[::-1])) |
| |
| |
| def test_socketConnect(self): |
| """ |
| Just like :py:obj:`test_memoryConnect` but with an actual socket. |
| |
| This is primarily to rule out the memory BIO code as the source of |
| any problems encountered while passing data over a :py:obj:`Connection` (if |
| this test fails, there must be a problem outside the memory BIO |
| code, as no memory BIO is involved here). Even though this isn't a |
| memory BIO test, it's convenient to have it here. |
| """ |
| server_conn, client_conn = self._loopback() |
| |
| important_message = b("Help me Obi Wan Kenobi, you're my only hope.") |
| client_conn.send(important_message) |
| msg = server_conn.recv(1024) |
| self.assertEqual(msg, important_message) |
| |
| # Again in the other direction, just for fun. |
| important_message = important_message[::-1] |
| server_conn.send(important_message) |
| msg = client_conn.recv(1024) |
| self.assertEqual(msg, important_message) |
| |
| |
| def test_socketOverridesMemory(self): |
| """ |
| Test that :py:obj:`OpenSSL.SSL.bio_read` and :py:obj:`OpenSSL.SSL.bio_write` don't |
| work on :py:obj:`OpenSSL.SSL.Connection`() that use sockets. |
| """ |
| context = Context(SSLv3_METHOD) |
| client = socket() |
| clientSSL = Connection(context, client) |
| self.assertRaises( TypeError, clientSSL.bio_read, 100) |
| self.assertRaises( TypeError, clientSSL.bio_write, "foo") |
| self.assertRaises( TypeError, clientSSL.bio_shutdown ) |
| |
| |
| def test_outgoingOverflow(self): |
| """ |
| If more bytes than can be written to the memory BIO are passed to |
| :py:obj:`Connection.send` at once, the number of bytes which were written is |
| returned and that many bytes from the beginning of the input can be |
| read from the other end of the connection. |
| """ |
| server = self._server(None) |
| client = self._client(None) |
| |
| self._interactInMemory(client, server) |
| |
| size = 2 ** 15 |
| sent = client.send(b"x" * size) |
| # Sanity check. We're trying to test what happens when the entire |
| # input can't be sent. If the entire input was sent, this test is |
| # meaningless. |
| self.assertTrue(sent < size) |
| |
| receiver, received = self._interactInMemory(client, server) |
| self.assertIdentical(receiver, server) |
| |
| # We can rely on all of these bytes being received at once because |
| # _loopback passes 2 ** 16 to recv - more than 2 ** 15. |
| self.assertEquals(len(received), sent) |
| |
| |
| def test_shutdown(self): |
| """ |
| :py:obj:`Connection.bio_shutdown` signals the end of the data stream from |
| which the :py:obj:`Connection` reads. |
| """ |
| server = self._server(None) |
| server.bio_shutdown() |
| e = self.assertRaises(Error, server.recv, 1024) |
| # We don't want WantReadError or ZeroReturnError or anything - it's a |
| # handshake failure. |
| self.assertEquals(e.__class__, Error) |
| |
| |
| def test_unexpectedEndOfFile(self): |
| """ |
| If the connection is lost before an orderly SSL shutdown occurs, |
| :py:obj:`OpenSSL.SSL.SysCallError` is raised with a message of |
| "Unexpected EOF". |
| """ |
| server_conn, client_conn = self._loopback() |
| client_conn.sock_shutdown(SHUT_RDWR) |
| exc = self.assertRaises(SysCallError, server_conn.recv, 1024) |
| self.assertEqual(exc.args, (-1, "Unexpected EOF")) |
| |
| |
| def _check_client_ca_list(self, func): |
| """ |
| Verify the return value of the :py:obj:`get_client_ca_list` method for server and client connections. |
| |
| :param func: A function which will be called with the server context |
| before the client and server are connected to each other. This |
| function should specify a list of CAs for the server to send to the |
| client and return that same list. The list will be used to verify |
| that :py:obj:`get_client_ca_list` returns the proper value at various |
| times. |
| """ |
| server = self._server(None) |
| client = self._client(None) |
| self.assertEqual(client.get_client_ca_list(), []) |
| self.assertEqual(server.get_client_ca_list(), []) |
| ctx = server.get_context() |
| expected = func(ctx) |
| self.assertEqual(client.get_client_ca_list(), []) |
| self.assertEqual(server.get_client_ca_list(), expected) |
| self._interactInMemory(client, server) |
| self.assertEqual(client.get_client_ca_list(), expected) |
| self.assertEqual(server.get_client_ca_list(), expected) |
| |
| |
| def test_set_client_ca_list_errors(self): |
| """ |
| :py:obj:`Context.set_client_ca_list` raises a :py:obj:`TypeError` if called with a |
| non-list or a list that contains objects other than X509Names. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| self.assertRaises(TypeError, ctx.set_client_ca_list, "spam") |
| self.assertRaises(TypeError, ctx.set_client_ca_list, ["spam"]) |
| self.assertIdentical(ctx.set_client_ca_list([]), None) |
| |
| |
| def test_set_empty_ca_list(self): |
| """ |
| If passed an empty list, :py:obj:`Context.set_client_ca_list` configures the |
| context to send no CA names to the client and, on both the server and |
| client sides, :py:obj:`Connection.get_client_ca_list` returns an empty list |
| after the connection is set up. |
| """ |
| def no_ca(ctx): |
| ctx.set_client_ca_list([]) |
| return [] |
| self._check_client_ca_list(no_ca) |
| |
| |
| def test_set_one_ca_list(self): |
| """ |
| If passed a list containing a single X509Name, |
| :py:obj:`Context.set_client_ca_list` configures the context to send that CA |
| name to the client and, on both the server and client sides, |
| :py:obj:`Connection.get_client_ca_list` returns a list containing that |
| X509Name after the connection is set up. |
| """ |
| cacert = load_certificate(FILETYPE_PEM, root_cert_pem) |
| cadesc = cacert.get_subject() |
| def single_ca(ctx): |
| ctx.set_client_ca_list([cadesc]) |
| return [cadesc] |
| self._check_client_ca_list(single_ca) |
| |
| |
| def test_set_multiple_ca_list(self): |
| """ |
| If passed a list containing multiple X509Name objects, |
| :py:obj:`Context.set_client_ca_list` configures the context to send those CA |
| names to the client and, on both the server and client sides, |
| :py:obj:`Connection.get_client_ca_list` returns a list containing those |
| X509Names after the connection is set up. |
| """ |
| secert = load_certificate(FILETYPE_PEM, server_cert_pem) |
| clcert = load_certificate(FILETYPE_PEM, server_cert_pem) |
| |
| sedesc = secert.get_subject() |
| cldesc = clcert.get_subject() |
| |
| def multiple_ca(ctx): |
| L = [sedesc, cldesc] |
| ctx.set_client_ca_list(L) |
| return L |
| self._check_client_ca_list(multiple_ca) |
| |
| |
| def test_reset_ca_list(self): |
| """ |
| If called multiple times, only the X509Names passed to the final call |
| of :py:obj:`Context.set_client_ca_list` are used to configure the CA names |
| sent to the client. |
| """ |
| cacert = load_certificate(FILETYPE_PEM, root_cert_pem) |
| secert = load_certificate(FILETYPE_PEM, server_cert_pem) |
| clcert = load_certificate(FILETYPE_PEM, server_cert_pem) |
| |
| cadesc = cacert.get_subject() |
| sedesc = secert.get_subject() |
| cldesc = clcert.get_subject() |
| |
| def changed_ca(ctx): |
| ctx.set_client_ca_list([sedesc, cldesc]) |
| ctx.set_client_ca_list([cadesc]) |
| return [cadesc] |
| self._check_client_ca_list(changed_ca) |
| |
| |
| def test_mutated_ca_list(self): |
| """ |
| If the list passed to :py:obj:`Context.set_client_ca_list` is mutated |
| afterwards, this does not affect the list of CA names sent to the |
| client. |
| """ |
| cacert = load_certificate(FILETYPE_PEM, root_cert_pem) |
| secert = load_certificate(FILETYPE_PEM, server_cert_pem) |
| |
| cadesc = cacert.get_subject() |
| sedesc = secert.get_subject() |
| |
| def mutated_ca(ctx): |
| L = [cadesc] |
| ctx.set_client_ca_list([cadesc]) |
| L.append(sedesc) |
| return [cadesc] |
| self._check_client_ca_list(mutated_ca) |
| |
| |
| def test_add_client_ca_errors(self): |
| """ |
| :py:obj:`Context.add_client_ca` raises :py:obj:`TypeError` if called with a non-X509 |
| object or with a number of arguments other than one. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| cacert = load_certificate(FILETYPE_PEM, root_cert_pem) |
| self.assertRaises(TypeError, ctx.add_client_ca) |
| self.assertRaises(TypeError, ctx.add_client_ca, "spam") |
| self.assertRaises(TypeError, ctx.add_client_ca, cacert, cacert) |
| |
| |
| def test_one_add_client_ca(self): |
| """ |
| A certificate's subject can be added as a CA to be sent to the client |
| with :py:obj:`Context.add_client_ca`. |
| """ |
| cacert = load_certificate(FILETYPE_PEM, root_cert_pem) |
| cadesc = cacert.get_subject() |
| def single_ca(ctx): |
| ctx.add_client_ca(cacert) |
| return [cadesc] |
| self._check_client_ca_list(single_ca) |
| |
| |
| def test_multiple_add_client_ca(self): |
| """ |
| Multiple CA names can be sent to the client by calling |
| :py:obj:`Context.add_client_ca` with multiple X509 objects. |
| """ |
| cacert = load_certificate(FILETYPE_PEM, root_cert_pem) |
| secert = load_certificate(FILETYPE_PEM, server_cert_pem) |
| |
| cadesc = cacert.get_subject() |
| sedesc = secert.get_subject() |
| |
| def multiple_ca(ctx): |
| ctx.add_client_ca(cacert) |
| ctx.add_client_ca(secert) |
| return [cadesc, sedesc] |
| self._check_client_ca_list(multiple_ca) |
| |
| |
| def test_set_and_add_client_ca(self): |
| """ |
| A call to :py:obj:`Context.set_client_ca_list` followed by a call to |
| :py:obj:`Context.add_client_ca` results in using the CA names from the first |
| call and the CA name from the second call. |
| """ |
| cacert = load_certificate(FILETYPE_PEM, root_cert_pem) |
| secert = load_certificate(FILETYPE_PEM, server_cert_pem) |
| clcert = load_certificate(FILETYPE_PEM, server_cert_pem) |
| |
| cadesc = cacert.get_subject() |
| sedesc = secert.get_subject() |
| cldesc = clcert.get_subject() |
| |
| def mixed_set_add_ca(ctx): |
| ctx.set_client_ca_list([cadesc, sedesc]) |
| ctx.add_client_ca(clcert) |
| return [cadesc, sedesc, cldesc] |
| self._check_client_ca_list(mixed_set_add_ca) |
| |
| |
| def test_set_after_add_client_ca(self): |
| """ |
| A call to :py:obj:`Context.set_client_ca_list` after a call to |
| :py:obj:`Context.add_client_ca` replaces the CA name specified by the former |
| call with the names specified by the latter cal. |
| """ |
| cacert = load_certificate(FILETYPE_PEM, root_cert_pem) |
| secert = load_certificate(FILETYPE_PEM, server_cert_pem) |
| clcert = load_certificate(FILETYPE_PEM, server_cert_pem) |
| |
| cadesc = cacert.get_subject() |
| sedesc = secert.get_subject() |
| |
| def set_replaces_add_ca(ctx): |
| ctx.add_client_ca(clcert) |
| ctx.set_client_ca_list([cadesc]) |
| ctx.add_client_ca(secert) |
| return [cadesc, sedesc] |
| self._check_client_ca_list(set_replaces_add_ca) |
| |
| |
| |
| class ConnectionBIOTests(TestCase): |
| """ |
| Tests for :py:obj:`Connection.bio_read` and :py:obj:`Connection.bio_write`. |
| """ |
| def test_wantReadError(self): |
| """ |
| :py:obj:`Connection.bio_read` raises :py:obj:`OpenSSL.SSL.WantReadError` |
| if there are no bytes available to be read from the BIO. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| conn = Connection(ctx, None) |
| self.assertRaises(WantReadError, conn.bio_read, 1024) |
| |
| |
| def test_buffer_size(self): |
| """ |
| :py:obj:`Connection.bio_read` accepts an integer giving the maximum |
| number of bytes to read and return. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| conn = Connection(ctx, None) |
| conn.set_connect_state() |
| try: |
| conn.do_handshake() |
| except WantReadError: |
| pass |
| data = conn.bio_read(2) |
| self.assertEqual(2, len(data)) |
| |
| |
| if not PY3: |
| def test_buffer_size_long(self): |
| """ |
| On Python 2 :py:obj:`Connection.bio_read` accepts values of type |
| :py:obj:`long` as well as :py:obj:`int`. |
| """ |
| ctx = Context(TLSv1_METHOD) |
| conn = Connection(ctx, None) |
| conn.set_connect_state() |
| try: |
| conn.do_handshake() |
| except WantReadError: |
| pass |
| data = conn.bio_read(long(2)) |
| self.assertEqual(2, len(data)) |
| |
| |
| |
| |
| class InfoConstantTests(TestCase): |
| """ |
| Tests for assorted constants exposed for use in info callbacks. |
| """ |
| def test_integers(self): |
| """ |
| All of the info constants are integers. |
| |
| This is a very weak test. It would be nice to have one that actually |
| verifies that as certain info events happen, the value passed to the |
| info callback matches up with the constant exposed by OpenSSL.SSL. |
| """ |
| for const in [ |
| SSL_ST_CONNECT, SSL_ST_ACCEPT, SSL_ST_MASK, SSL_ST_INIT, |
| SSL_ST_BEFORE, SSL_ST_OK, SSL_ST_RENEGOTIATE, |
| SSL_CB_LOOP, SSL_CB_EXIT, SSL_CB_READ, SSL_CB_WRITE, SSL_CB_ALERT, |
| SSL_CB_READ_ALERT, SSL_CB_WRITE_ALERT, SSL_CB_ACCEPT_LOOP, |
| SSL_CB_ACCEPT_EXIT, SSL_CB_CONNECT_LOOP, SSL_CB_CONNECT_EXIT, |
| SSL_CB_HANDSHAKE_START, SSL_CB_HANDSHAKE_DONE]: |
| |
| self.assertTrue(isinstance(const, int)) |
| |
| |
| if __name__ == '__main__': |
| main() |