Add re-add deprecated support for unicode in a variety of APIs

OpenSSL/SSL.py

@@ -14,6 +14,7 @@
     lib as _lib,
     exception_from_error_queue as _exception_from_error_queue,
     native as _native,
+    text_to_bytes_and_warn as _text_to_bytes_and_warn,
     path_string as _path_string,
 )
 
@@ -1239,6 +1240,9 @@ def send(self, buf, flags=0):
                       API, the value is ignored
         :return: The number of bytes written
         """
+        # Backward compatibility
+        buf = _text_to_bytes_and_warn("buf", buf)
+
         if isinstance(buf, _memoryview):
             buf = buf.tobytes()
         if isinstance(buf, _buffer):
@@ -1263,6 +1267,8 @@ def sendall(self, buf, flags=0):
                       API, the value is ignored
         :return: The number of bytes written
         """
+        buf = _text_to_bytes_and_warn("buf", buf)
+
         if isinstance(buf, _memoryview):
             buf = buf.tobytes()
         if isinstance(buf, _buffer):
@@ -1387,6 +1393,8 @@ def bio_write(self, buf):
         :param buf: The string to put into the memory BIO.
         :return: The number of bytes written
         """
+        buf = _text_to_bytes_and_warn("buf", buf)
+
         if self._into_ssl is None:
             raise TypeError("Connection sock was not None")
 

OpenSSL/_util.py

@@ -1,3 +1,4 @@
+from warnings import warn
 import sys
 
 from six import PY3, binary_type, text_type
@@ -93,3 +94,29 @@ def byte_string(s):
 else:
     def byte_string(s):
         return s
+
+_TEXT_WARNING = (
+    text_type.__name__ + " for {0} is no longer accepted, use bytes"
+)
+
+def text_to_bytes_and_warn(label, obj):
+    """
+    If ``obj`` is text, emit a warning that it should be bytes instead and try
+    to convert it to bytes automatically.
+
+    :param str label: The name of the parameter from which ``obj`` was taken
+        (so a developer can easily find the source of the problem and correct
+        it).
+
+    :return: If ``obj`` is the text string type, a ``bytes`` object giving the
+        UTF-8 encoding of that text is returned.  Otherwise, ``obj`` itself is
+        returned.
+    """
+    if isinstance(obj, text_type):
+        warn(
+            _TEXT_WARNING.format(label),
+            category=DeprecationWarning,
+            stacklevel=3
+        )
+        return obj.encode('utf-8')
+    return obj

OpenSSL/crypto.py

@@ -13,7 +13,9 @@
     lib as _lib,
     exception_from_error_queue as _exception_from_error_queue,
     byte_string as _byte_string,
-    native as _native)
+    native as _native,
+    text_to_bytes_and_warn as _text_to_bytes_and_warn,
+)
 
 FILETYPE_PEM = _lib.SSL_FILETYPE_PEM
 FILETYPE_ASN1 = _lib.SSL_FILETYPE_ASN1
@@ -2074,6 +2076,8 @@ def export(self, passphrase=None, iter=2048, maciter=1):
 
         :return: The string containing the PKCS12
         """
+        passphrase = _text_to_bytes_and_warn("passphrase", passphrase)
+
         if self._cacerts is None:
             cacerts = _ffi.NULL
         else:
@@ -2371,6 +2375,8 @@ def sign(pkey, data, digest):
     :param digest: message digest to use
     :return: signature
     """
+    data = _text_to_bytes_and_warn("data", data)
+
     digest_obj = _lib.EVP_get_digestbyname(_byte_string(digest))
     if digest_obj == _ffi.NULL:
         raise ValueError("No such digest method")
@@ -2405,6 +2411,8 @@ def verify(cert, signature, data, digest):
     :param digest: message digest to use
     :return: None if the signature is correct, raise exception otherwise
     """
+    data = _text_to_bytes_and_warn("data", data)
+
     digest_obj = _lib.EVP_get_digestbyname(_byte_string(digest))
     if digest_obj == _ffi.NULL:
         raise ValueError("No such digest method")
@@ -2495,6 +2503,8 @@ def load_pkcs12(buffer, passphrase=None):
     :param passphrase: (Optional) The password to decrypt the PKCS12 lump
     :returns: The PKCS12 object
     """
+    passphrase = _text_to_bytes_and_warn("passphrase", passphrase)
+
     if isinstance(buffer, _text_type):
         buffer = buffer.encode("ascii")
 

OpenSSL/test/test_crypto.py

@@ -13,7 +13,9 @@
 from subprocess import PIPE, Popen
 from datetime import datetime, timedelta
 
-from six import u, b, binary_type
+from six import u, b, binary_type, PY3
+from warnings import simplefilter
+from warnings import catch_warnings
 
 from OpenSSL.crypto import TYPE_RSA, TYPE_DSA, Error, PKey, PKeyType
 from OpenSSL.crypto import X509, X509Type, X509Name, X509NameType
@@ -30,7 +32,9 @@
 from OpenSSL.crypto import NetscapeSPKI, NetscapeSPKIType
 from OpenSSL.crypto import (
     sign, verify, get_elliptic_curve, get_elliptic_curves)
-from OpenSSL.test.util import EqualityTestsMixin, TestCase
+from OpenSSL.test.util import (
+    EqualityTestsMixin, TestCase, WARNING_TYPE_EXPECTED
+)
 from OpenSSL._util import native, lib
 
 def normalize_certificate_pem(pem):
@@ -2061,6 +2065,31 @@ def test_load_pkcs12(self):
         self.verify_pkcs12_container(p12)
 
 
+    def test_load_pkcs12_text_passphrase(self):
+        """
+        A PKCS12 string generated using the openssl command line can be loaded
+        with :py:obj:`load_pkcs12` and its components extracted and examined.
+        Using text as passphrase instead of bytes. DeprecationWarning expected.
+        """
+        pem = client_key_pem + client_cert_pem
+        passwd = b"whatever"
+        p12_str = _runopenssl(pem, b"pkcs12", b"-export", b"-clcerts",
+                              b"-passout", b"pass:" + passwd)
+        with catch_warnings(record=True) as w:
+            simplefilter("always")
+            p12 = load_pkcs12(p12_str, passphrase=b"whatever".decode("ascii"))
+
+            self.assertEqual(
+                "{0} for passphrase is no longer accepted, use bytes".format(
+                    WARNING_TYPE_EXPECTED
+                ),
+                str(w[-1].message)
+            )
+            self.assertIs(w[-1].category, DeprecationWarning)
+
+        self.verify_pkcs12_container(p12)
+
+
     def test_load_pkcs12_no_passphrase(self):
         """
         A PKCS12 string generated using openssl command line can be loaded with
@@ -2262,6 +2291,26 @@ def test_export_without_args(self):
             dumped_p12, key=server_key_pem, cert=server_cert_pem, passwd=b"")
 
 
+    def test_export_without_bytes(self):
+        """
+        Test :py:obj:`PKCS12.export` with text not bytes as passphrase
+        """
+        p12 = self.gen_pkcs12(server_cert_pem, server_key_pem, root_cert_pem)
+
+        with catch_warnings(record=True) as w:
+            simplefilter("always")
+            dumped_p12 = p12.export(passphrase=b"randomtext".decode("ascii"))
+            self.assertEqual(
+                "{0} for passphrase is no longer accepted, use bytes".format(
+                    WARNING_TYPE_EXPECTED
+                ),
+                str(w[-1].message)
+            )
+            self.assertIs(w[-1].category, DeprecationWarning)
+        self.check_recovery(
+            dumped_p12, key=server_key_pem, cert=server_cert_pem, passwd=b"randomtext")
+
+
     def test_key_cert_mismatch(self):
         """
         :py:obj:`PKCS12.export` raises an exception when a key and certificate
@@ -3321,6 +3370,47 @@ def test_sign_verify(self):
             ValueError, verify, good_cert, sig, content, "strange-digest")
 
 
+    def test_sign_verify_with_text(self):
+        """
+        :py:obj:`sign` generates a cryptographic signature which :py:obj:`verify` can check.
+        Deprecation warnings raised because using text instead of bytes as content
+        """
+        content = (
+            b"It was a bright cold day in April, and the clocks were striking "
+            b"thirteen. Winston Smith, his chin nuzzled into his breast in an "
+            b"effort to escape the vile wind, slipped quickly through the "
+            b"glass doors of Victory Mansions, though not quickly enough to "
+            b"prevent a swirl of gritty dust from entering along with him."
+        ).decode("ascii")
+
+        priv_key = load_privatekey(FILETYPE_PEM, root_key_pem)
+        cert = load_certificate(FILETYPE_PEM, root_cert_pem)
+        for digest in ['md5', 'sha1']:
+            with catch_warnings(record=True) as w:
+                simplefilter("always")
+                sig = sign(priv_key, content, digest)
+
+                self.assertEqual(
+                    "{0} for data is no longer accepted, use bytes".format(
+                        WARNING_TYPE_EXPECTED
+                    ),
+                    str(w[-1].message)
+                )
+                self.assertIs(w[-1].category, DeprecationWarning)
+
+            with catch_warnings(record=True) as w:
+                simplefilter("always")
+                verify(cert, sig, content, digest)
+
+                self.assertEqual(
+                    "{0} for data is no longer accepted, use bytes".format(
+                        WARNING_TYPE_EXPECTED
+                    ),
+                    str(w[-1].message)
+                )
+                self.assertIs(w[-1].category, DeprecationWarning)
+
+
     def test_sign_nulls(self):
         """
         :py:obj:`sign` produces a signature for a string with embedded nulls.

OpenSSL/test/test_ssl.py

@@ -7,12 +7,13 @@
 
 from gc import collect, get_referrers
 from errno import ECONNREFUSED, EINPROGRESS, EWOULDBLOCK, EPIPE, ESHUTDOWN
-from sys import platform, version_info, getfilesystemencoding
+from sys import platform, getfilesystemencoding
 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 warnings import catch_warnings, simplefilter
 
 from six import PY3, text_type, u
 
@@ -43,11 +44,9 @@
     Context, ContextType, Session, Connection, ConnectionType, SSLeay_version)
 
 from OpenSSL._util import lib as _lib
-
-from OpenSSL.test.util import NON_ASCII, TestCase, b
-from OpenSSL.test.test_crypto import (
-    cleartextCertificatePEM, cleartextPrivateKeyPEM)
+from OpenSSL.test.util import WARNING_TYPE_EXPECTED, NON_ASCII, TestCase, b
 from OpenSSL.test.test_crypto import (
+    cleartextCertificatePEM, cleartextPrivateKeyPEM,
     client_cert_pem, client_key_pem, server_cert_pem, server_key_pem,
     root_cert_pem)
 
@@ -2108,7 +2107,7 @@ def test_set_tlsext_host_name_wrong_args(self):
         self.assertRaises(
             TypeError, conn.set_tlsext_host_name, b("with\0null"))
 
-        if version_info >= (3,):
+        if PY3:
             # On Python 3.x, don't accidentally implicitly convert from text.
             self.assertRaises(
                 TypeError,
@@ -2749,6 +2748,26 @@ def test_short_bytes(self):
         self.assertEquals(count, 2)
         self.assertEquals(client.recv(2), b('xy'))
 
+
+    def test_text(self):
+        """
+        When passed a text, :py:obj:`Connection.send` transmits all of it and
+        returns the number of bytes sent. It also raises a DeprecationWarning.
+        """
+        server, client = self._loopback()
+        with catch_warnings(record=True) as w:
+            simplefilter("always")
+            count = server.send(b"xy".decode("ascii"))
+            self.assertEqual(
+                "{0} for buf is no longer accepted, use bytes".format(
+                    WARNING_TYPE_EXPECTED
+                ),
+                str(w[-1].message)
+            )
+            self.assertIs(w[-1].category, DeprecationWarning)
+        self.assertEquals(count, 2)
+        self.assertEquals(client.recv(2), b"xy")
+
     try:
         memoryview
     except NameError:
@@ -2969,6 +2988,25 @@ def test_short(self):
         self.assertEquals(client.recv(1), b('x'))
 
 
+    def test_text(self):
+        """
+        :py:obj:`Connection.sendall` transmits all the content in the string
+        passed to it raising a DeprecationWarning in case of this being a text.
+        """
+        server, client = self._loopback()
+        with catch_warnings(record=True) as w:
+            simplefilter("always")
+            server.sendall(b"x".decode("ascii"))
+            self.assertEqual(
+                "{0} for buf is no longer accepted, use bytes".format(
+                    WARNING_TYPE_EXPECTED
+                ),
+                str(w[-1].message)
+            )
+            self.assertIs(w[-1].category, DeprecationWarning)
+        self.assertEquals(client.recv(1), b"x")
+
+
     try:
         memoryview
     except NameError:

OpenSSL/test/util.py

@@ -14,6 +14,8 @@
 from unittest import TestCase
 import sys
 
+from six import PY3
+
 from OpenSSL._util import exception_from_error_queue
 from OpenSSL.crypto import Error
 
@@ -452,3 +454,10 @@ def __ne__(self, other):
         a = self.anInstance()
         b = Delegate()
         self.assertEqual(a != b, [b])
+
+
+# The type name expected in warnings about using the wrong string type.
+if PY3:
+    WARNING_TYPE_EXPECTED = "str"
+else:
+    WARNING_TYPE_EXPECTED = "unicode"