Merge pull request #210 from reaperhulk/docker-travis
travis docker support + osx support ~~madness~~ (switch to container-based Docker builds for speed; also introduce an OS X builder using OpenSSL 1.0.2).
diff --git a/ChangeLog b/ChangeLog
index e872803..95c891f 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,9 @@
+2015-04-12 Jean-Paul Calderone <exarkun@twistedmatrix.com>
+
+ * OpenSSL/rand.py, OpenSSL/SSL.py: APIs which previously accepted
+ filenames only as bytes now accept them as either bytes or
+ unicode (and respect sys.getfilesystemencoding()).
+
2015-03-23 Jean-Paul Calderone <exarkun@twistedmatrix.com>
* OpenSSL/SSL.py: Add Cory Benfield's next-protocol-negotiation
diff --git a/OpenSSL/SSL.py b/OpenSSL/SSL.py
index 6d00e13..1215526 100644
--- a/OpenSSL/SSL.py
+++ b/OpenSSL/SSL.py
@@ -12,7 +12,9 @@
ffi as _ffi,
lib as _lib,
exception_from_error_queue as _exception_from_error_queue,
- native as _native)
+ native as _native,
+ path_string as _path_string,
+)
from OpenSSL.crypto import (
FILETYPE_PEM, _PassphraseHelper, PKey, X509Name, X509, X509Store)
@@ -420,19 +422,22 @@
Let SSL know where we can find trusted certificates for the certificate
chain
- :param cafile: In which file we can find the certificates
+ :param cafile: In which file we can find the certificates (``bytes`` or
+ ``unicode``).
:param capath: In which directory we can find the certificates
+ (``bytes`` or ``unicode``).
+
:return: None
"""
if cafile is None:
cafile = _ffi.NULL
- elif not isinstance(cafile, bytes):
- raise TypeError("cafile must be None or a byte string")
+ else:
+ cafile = _path_string(cafile)
if capath is None:
capath = _ffi.NULL
- elif not isinstance(capath, bytes):
- raise TypeError("capath must be None or a byte string")
+ else:
+ capath = _path_string(capath)
load_result = _lib.SSL_CTX_load_verify_locations(self._context, cafile, capath)
if not load_result:
@@ -482,15 +487,12 @@
"""
Load a certificate chain from a file
- :param certfile: The name of the certificate chain file
+ :param certfile: The name of the certificate chain file (``bytes`` or
+ ``unicode``).
+
:return: None
"""
- if isinstance(certfile, _text_type):
- # Perhaps sys.getfilesystemencoding() could be better?
- certfile = certfile.encode("utf-8")
-
- if not isinstance(certfile, bytes):
- raise TypeError("certfile must be bytes or unicode")
+ certfile = _path_string(certfile)
result = _lib.SSL_CTX_use_certificate_chain_file(self._context, certfile)
if not result:
@@ -501,15 +503,13 @@
"""
Load a certificate from a file
- :param certfile: The name of the certificate file
+ :param certfile: The name of the certificate file (``bytes`` or
+ ``unicode``).
:param filetype: (optional) The encoding of the file, default is PEM
+
:return: None
"""
- if isinstance(certfile, _text_type):
- # Perhaps sys.getfilesystemencoding() could be better?
- certfile = certfile.encode("utf-8")
- if not isinstance(certfile, bytes):
- raise TypeError("certfile must be bytes or unicode")
+ certfile = _path_string(certfile)
if not isinstance(filetype, integer_types):
raise TypeError("filetype must be an integer")
@@ -563,16 +563,12 @@
"""
Load a private key from a file
- :param keyfile: The name of the key file
+ :param keyfile: The name of the key file (``bytes`` or ``unicode``)
:param filetype: (optional) The encoding of the file, default is PEM
+
:return: None
"""
- if isinstance(keyfile, _text_type):
- # Perhaps sys.getfilesystemencoding() could be better?
- keyfile = keyfile.encode("utf-8")
-
- if not isinstance(keyfile, bytes):
- raise TypeError("keyfile must be a byte string")
+ keyfile = _path_string(keyfile)
if filetype is _unspecified:
filetype = FILETYPE_PEM
@@ -708,11 +704,12 @@
"""
Load parameters for Ephemeral Diffie-Hellman
- :param dhfile: The file to load EDH parameters from
+ :param dhfile: The file to load EDH parameters from (``bytes`` or
+ ``unicode``).
+
:return: None
"""
- if not isinstance(dhfile, bytes):
- raise TypeError("dhfile must be a byte string")
+ dhfile = _path_string(dhfile)
bio = _lib.BIO_new_file(dhfile, b"r")
if bio == _ffi.NULL:
diff --git a/OpenSSL/_util.py b/OpenSSL/_util.py
index 5e2b256..1eec17e 100644
--- a/OpenSSL/_util.py
+++ b/OpenSSL/_util.py
@@ -1,3 +1,5 @@
+import sys
+
from six import PY3, binary_type, text_type
from cryptography.hazmat.bindings.openssl.binding import Binding
@@ -68,6 +70,23 @@
+def path_string(s):
+ """
+ Convert a Python string to a :py:class:`bytes` string identifying the same
+ path and which can be passed into an OpenSSL API accepting a filename.
+
+ :param s: An instance of :py:class:`bytes` or :py:class:`unicode`.
+
+ :return: An instance of :py:class:`bytes`.
+ """
+ if isinstance(s, binary_type):
+ return s
+ elif isinstance(s, text_type):
+ return s.encode(sys.getfilesystemencoding())
+ else:
+ raise TypeError("Path must be represented as bytes or unicode string")
+
+
if PY3:
def byte_string(s):
return s.encode("charmap")
diff --git a/OpenSSL/rand.py b/OpenSSL/rand.py
index e754378..3adf693 100644
--- a/OpenSSL/rand.py
+++ b/OpenSSL/rand.py
@@ -11,7 +11,8 @@
from OpenSSL._util import (
ffi as _ffi,
lib as _lib,
- exception_from_error_queue as _exception_from_error_queue)
+ exception_from_error_queue as _exception_from_error_queue,
+ path_string as _path_string)
class Error(Exception):
@@ -131,13 +132,13 @@
"""
Seed the PRNG with data from a file
- :param filename: The file to read data from
- :param maxbytes: (optional) The number of bytes to read, default is
- to read the entire file
+ :param filename: The file to read data from (``bytes`` or ``unicode``).
+ :param maxbytes: (optional) The number of bytes to read, default is to read
+ the entire file
+
:return: The number of bytes read
"""
- if not isinstance(filename, _builtin_bytes):
- raise TypeError("filename must be a string")
+ filename = _path_string(filename)
if maxbytes is _unspecified:
maxbytes = -1
@@ -152,12 +153,11 @@
"""
Save PRNG state to a file
- :param filename: The file to write data to
+ :param filename: The file to write data to (``bytes`` or ``unicode``).
+
:return: The number of bytes written
"""
- if not isinstance(filename, _builtin_bytes):
- raise TypeError("filename must be a string")
-
+ filename = _path_string(filename)
return _lib.RAND_write_file(filename)
diff --git a/OpenSSL/test/test_rand.py b/OpenSSL/test/test_rand.py
index c52cb6b..3d5c290 100644
--- a/OpenSSL/test/test_rand.py
+++ b/OpenSSL/test/test_rand.py
@@ -10,7 +10,7 @@
import stat
import sys
-from OpenSSL.test.util import TestCase, b
+from OpenSSL.test.util import NON_ASCII, TestCase, b
from OpenSSL import rand
@@ -176,27 +176,47 @@
self.assertRaises(TypeError, rand.write_file, None)
self.assertRaises(TypeError, rand.write_file, "foo", None)
+ def _read_write_test(self, path):
+ """
+ Verify that ``rand.write_file`` and ``rand.load_file`` can be used.
+ """
+ # Create the file so cleanup is more straightforward
+ with open(path, "w"):
+ pass
- def test_files(self):
- """
- Test reading and writing of files via rand functions.
- """
- # Write random bytes to a file
- tmpfile = self.mktemp()
- # Make sure it exists (so cleanup definitely succeeds)
- fObj = open(tmpfile, 'w')
- fObj.close()
try:
- rand.write_file(tmpfile)
+ # Write random bytes to a file
+ rand.write_file(path)
+
# Verify length of written file
- size = os.stat(tmpfile)[stat.ST_SIZE]
+ size = os.stat(path)[stat.ST_SIZE]
self.assertEqual(1024, size)
+
# Read random bytes from file
- rand.load_file(tmpfile)
- rand.load_file(tmpfile, 4) # specify a length
+ rand.load_file(path)
+ rand.load_file(path, 4) # specify a length
finally:
# Cleanup
- os.unlink(tmpfile)
+ os.unlink(path)
+
+
+ def test_bytes_paths(self):
+ """
+ Random data can be saved and loaded to files with paths specified as
+ bytes.
+ """
+ path = self.mktemp()
+ path += NON_ASCII.encode(sys.getfilesystemencoding())
+ self._read_write_test(path)
+
+
+ def test_unicode_paths(self):
+ """
+ Random data can be saved and loaded to files with paths specified as
+ unicode.
+ """
+ path = self.mktemp().decode('utf-8') + NON_ASCII
+ self._read_write_test(path)
if __name__ == '__main__':
diff --git a/OpenSSL/test/test_ssl.py b/OpenSSL/test/test_ssl.py
index 188d559..c82dea6 100644
--- a/OpenSSL/test/test_ssl.py
+++ b/OpenSSL/test/test_ssl.py
@@ -7,7 +7,7 @@
from gc import collect, get_referrers
from errno import ECONNREFUSED, EINPROGRESS, EWOULDBLOCK, EPIPE, ESHUTDOWN
-from sys import platform, version_info
+from sys import platform, version_info, getfilesystemencoding
from socket import SHUT_RDWR, error, socket
from os import makedirs
from os.path import join
@@ -22,7 +22,6 @@
from OpenSSL.crypto import dump_certificate, load_certificate
from OpenSSL.crypto import get_elliptic_curves
-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
@@ -43,7 +42,7 @@
from OpenSSL.SSL import (
Context, ContextType, Session, Connection, ConnectionType, SSLeay_version)
-from OpenSSL.test.util import TestCase, b
+from OpenSSL.test.util import NON_ASCII, TestCase, b
from OpenSSL.test.test_crypto import (
cleartextCertificatePEM, cleartextPrivateKeyPEM)
from OpenSSL.test.test_crypto import (
@@ -95,6 +94,23 @@
"""
+def join_bytes_or_unicode(prefix, suffix):
+ """
+ Join two path components of either ``bytes`` or ``unicode``.
+
+ The return type is the same as the type of ``prefix``.
+ """
+ # If the types are the same, nothing special is necessary.
+ if type(prefix) == type(suffix):
+ return join(prefix, suffix)
+
+ # Otherwise, coerce suffix to the type of prefix.
+ if isinstance(prefix, text_type):
+ return join(prefix, suffix.decode(getfilesystemencoding()))
+ else:
+ return join(prefix, suffix.encode(getfilesystemencoding()))
+
+
def verify_cb(conn, cert, errnum, depth, ok):
return ok
@@ -395,23 +411,52 @@
self.assertRaises(Error, ctx.use_privatekey_file, self.mktemp())
+ def _use_privatekey_file_test(self, pemfile, filetype):
+ """
+ Verify that calling ``Context.use_privatekey_file`` with the given
+ arguments does not raise an exception.
+ """
+ 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, filetype)
+
+
+ def test_use_privatekey_file_bytes(self):
+ """
+ A private key can be specified from a file by passing a ``bytes``
+ instance giving the file name to ``Context.use_privatekey_file``.
+ """
+ self._use_privatekey_file_test(
+ self.mktemp() + NON_ASCII.encode(getfilesystemencoding()),
+ FILETYPE_PEM,
+ )
+
+
+ def test_use_privatekey_file_unicode(self):
+ """
+ A private key can be specified from a file by passing a ``unicode``
+ instance giving the file name to ``Context.use_privatekey_file``.
+ """
+ self._use_privatekey_file_test(
+ self.mktemp().decode(getfilesystemencoding()) + NON_ASCII,
+ FILETYPE_PEM,
+ )
+
+
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))
+ self._use_privatekey_file_test(self.mktemp(), long(FILETYPE_PEM))
def test_use_certificate_wrong_args(self):
@@ -476,21 +521,40 @@
self.assertRaises(Error, ctx.use_certificate_file, self.mktemp())
- def test_use_certificate_file(self):
+ def _use_certificate_file_test(self, certificate_file):
"""
- :py:obj:`Context.use_certificate` sets the certificate which will be
- used to identify connections created using the context.
+ Verify that calling ``Context.use_certificate_file`` with the given
+ filename doesn't raise an exception.
"""
# 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:
+ with open(certificate_file, "wb") as pem_file:
pem_file.write(cleartextCertificatePEM)
ctx = Context(TLSv1_METHOD)
- ctx.use_certificate_file(pem_filename)
+ ctx.use_certificate_file(certificate_file)
+
+
+ def test_use_certificate_file_bytes(self):
+ """
+ :py:obj:`Context.use_certificate_file` sets the certificate (given as a
+ ``bytes`` filename) which will be used to identify connections created
+ using the context.
+ """
+ filename = self.mktemp() + NON_ASCII.encode(getfilesystemencoding())
+ self._use_certificate_file_test(filename)
+
+
+ def test_use_certificate_file_unicode(self):
+ """
+ :py:obj:`Context.use_certificate_file` sets the certificate (given as a
+ ``bytes`` filename) which will be used to identify connections created
+ using the context.
+ """
+ filename = self.mktemp().decode(getfilesystemencoding()) + NON_ASCII
+ self._use_certificate_file_test(filename)
if not PY3:
@@ -904,12 +968,13 @@
self.assertEqual(cert.get_subject().CN, 'Testing Root CA')
- def test_load_verify_file(self):
+ def _load_verify_cafile(self, cafile):
"""
- :py:obj:`Context.load_verify_locations` accepts a file name and uses the
- certificates within for verification purposes.
+ Verify that if path to a file containing a certificate is passed to
+ ``Context.load_verify_locations`` for the ``cafile`` parameter, that
+ certificate is used as a trust root for the purposes of verifying
+ connections created using that ``Context``.
"""
- cafile = self.mktemp()
fObj = open(cafile, 'w')
fObj.write(cleartextCertificatePEM.decode('ascii'))
fObj.close()
@@ -917,6 +982,27 @@
self._load_verify_locations_test(cafile)
+ def test_load_verify_bytes_cafile(self):
+ """
+ :py:obj:`Context.load_verify_locations` accepts a file name as a
+ ``bytes`` instance and uses the certificates within for verification
+ purposes.
+ """
+ cafile = self.mktemp() + NON_ASCII.encode(getfilesystemencoding())
+ self._load_verify_cafile(cafile)
+
+
+ def test_load_verify_unicode_cafile(self):
+ """
+ :py:obj:`Context.load_verify_locations` accepts a file name as a
+ ``unicode`` instance and uses the certificates within for verification
+ purposes.
+ """
+ self._load_verify_cafile(
+ self.mktemp().decode(getfilesystemencoding()) + NON_ASCII
+ )
+
+
def test_load_verify_invalid_file(self):
"""
:py:obj:`Context.load_verify_locations` raises :py:obj:`Error` when passed a
@@ -927,25 +1013,47 @@
Error, clientContext.load_verify_locations, self.mktemp())
- def test_load_verify_directory(self):
+ def _load_verify_directory_locations_capath(self, capath):
"""
- :py:obj:`Context.load_verify_locations` accepts a directory name and uses
- the certificates within for verification purposes.
+ Verify that if path to a directory containing certificate files is
+ passed to ``Context.load_verify_locations`` for the ``capath``
+ parameter, those certificates are used as trust roots for the purposes
+ of verifying connections created using that ``Context``.
"""
- 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()
+ cafile = join_bytes_or_unicode(capath, name)
+ with open(cafile, 'w') as fObj:
+ fObj.write(cleartextCertificatePEM.decode('ascii'))
self._load_verify_locations_test(None, capath)
+ def test_load_verify_directory_bytes_capath(self):
+ """
+ :py:obj:`Context.load_verify_locations` accepts a directory name as a
+ ``bytes`` instance and uses the certificates within for verification
+ purposes.
+ """
+ self._load_verify_directory_locations_capath(
+ self.mktemp() + NON_ASCII.encode(getfilesystemencoding())
+ )
+
+
+ def test_load_verify_directory_unicode_capath(self):
+ """
+ :py:obj:`Context.load_verify_locations` accepts a directory name as a
+ ``unicode`` instance and uses the certificates within for verification
+ purposes.
+ """
+ self._load_verify_directory_locations_capath(
+ self.mktemp().decode(getfilesystemencoding()) + NON_ASCII
+ )
+
+
def test_load_verify_locations_wrong_args(self):
"""
:py:obj:`Context.load_verify_locations` raises :py:obj:`TypeError` if called with
@@ -1131,43 +1239,67 @@
self._handshake_test(serverContext, clientContext)
- def test_use_certificate_chain_file(self):
+ def _use_certificate_chain_file_test(self, certdir):
"""
- :py:obj:`Context.use_certificate_chain_file` reads a certificate chain from
- the specified file.
+ Verify that :py:obj:`Context.use_certificate_chain_file` reads a
+ certificate chain from a 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
+ 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
+ makedirs(certdir)
+
+ chainFile = join_bytes_or_unicode(certdir, "chain.pem")
+ caFile = join_bytes_or_unicode(certdir, "ca.pem")
+
# 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()
+ with open(chainFile, 'wb') as fObj:
+ # 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))
+
+ with open(caFile, 'w') as fObj:
+ fObj.write(dump_certificate(FILETYPE_PEM, cacert).decode('ascii'))
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")
+ clientContext.load_verify_locations(caFile)
self._handshake_test(serverContext, clientContext)
+ def test_use_certificate_chain_file_bytes(self):
+ """
+ ``Context.use_certificate_chain_file`` accepts the name of a file (as
+ an instance of ``bytes``) to specify additional certificates to use to
+ construct and verify a trust chain.
+ """
+ self._use_certificate_chain_file_test(
+ self.mktemp() + NON_ASCII.encode(getfilesystemencoding())
+ )
+
+
+ def test_use_certificate_chain_file_unicode(self):
+ """
+ ``Context.use_certificate_chain_file`` accepts the name of a file (as
+ an instance of ``unicode``) to specify additional certificates to use
+ to construct and verify a trust chain.
+ """
+ self._use_certificate_chain_file_test(
+ self.mktemp().decode(getfilesystemencoding()) + NON_ASCII
+ )
+
+
def test_use_certificate_chain_file_wrong_args(self):
"""
:py:obj:`Context.use_certificate_chain_file` raises :py:obj:`TypeError`
@@ -1242,20 +1374,39 @@
self.assertRaises(Error, context.load_tmp_dh, b"hello")
- def test_load_tmp_dh(self):
+ def _load_tmp_dh_test(self, dhfilename):
"""
- :py:obj:`Context.load_tmp_dh` loads Diffie-Hellman parameters from the
- specified file.
+ Verify that calling ``Context.load_tmp_dh`` with the given filename
+ does not raise an exception.
"""
context = Context(TLSv1_METHOD)
- dhfilename = self.mktemp()
- dhfile = open(dhfilename, "w")
- dhfile.write(dhparam)
- dhfile.close()
+ with open(dhfilename, "w") as dhfile:
+ dhfile.write(dhparam)
+
context.load_tmp_dh(dhfilename)
# XXX What should I assert here? -exarkun
+ def test_load_tmp_dh_bytes(self):
+ """
+ :py:obj:`Context.load_tmp_dh` loads Diffie-Hellman parameters from the
+ specified file (given as ``bytes``).
+ """
+ self._load_tmp_dh_test(
+ self.mktemp() + NON_ASCII.encode(getfilesystemencoding()),
+ )
+
+
+ def test_load_tmp_dh_unicode(self):
+ """
+ :py:obj:`Context.load_tmp_dh` loads Diffie-Hellman parameters from the
+ specified file (given as ``unicode``).
+ """
+ self._load_tmp_dh_test(
+ self.mktemp().decode(getfilesystemencoding()) + NON_ASCII,
+ )
+
+
def test_set_tmp_ecdh(self):
"""
:py:obj:`Context.set_tmp_ecdh` sets the elliptic curve for
diff --git a/OpenSSL/test/util.py b/OpenSSL/test/util.py
index 4260eb0..b24a73b 100644
--- a/OpenSSL/test/util.py
+++ b/OpenSSL/test/util.py
@@ -25,6 +25,11 @@
from OpenSSL._util import ffi, lib, byte_string as b
+
+# This is the UTF-8 encoding of the SNOWMAN unicode code point.
+NON_ASCII = b("\xe2\x98\x83").decode("utf-8")
+
+
class TestCase(TestCase):
"""
:py:class:`TestCase` adds useful testing functionality beyond what is available