crypto._PassphraseHelper: pass non-callable passphrase using callback (#947)

* crypto._PassphraseHelper: pass non-callable passphrase using callback
Fixes #945

Before this commit, we would pass a bytes passphrase as a null terminated string.
This causes issue when a randomly generated key's first byte is null because
OpenSSL rightly determines the key length is 0.
This commit modifies the passphrase helper to pass the passphrase via the
 callback

* Update changelog to document bug fix

3 files changed, 68 insertions, 11 deletions

diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 2ba1f7f..5df0a05 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -38,7 +38,10 @@ Changes:
 - Make verification callback optional in ``Context.set_verify``.
   If omitted, OpenSSL's default verification is used.
   `#933 <https://github.com/pyca/pyopenssl/pull/933>`_
-
+- Fixed a bug that could truncate or cause a zero-length key error due to a
+  null byte in private key passphrase in ``OpenSSL.crypto.load_privatekey``
+  and ``OpenSSL.crypto.dump_privatekey``.
+  `#947 <https://github.com/pyca/pyopenssl/pull/947>`_
 
 19.1.0 (2019-11-18)
 -------------------
diff --git a/src/OpenSSL/crypto.py b/src/OpenSSL/crypto.py
index f89a28f..77fb821 100644
--- a/src/OpenSSL/crypto.py
+++ b/src/OpenSSL/crypto.py
@@ -2788,9 +2788,7 @@ class _PassphraseHelper(object):
     def callback(self):
         if self._passphrase is None:
             return _ffi.NULL
-        elif isinstance(self._passphrase, bytes):
-            return _ffi.NULL
-        elif callable(self._passphrase):
+        elif isinstance(self._passphrase, bytes) or callable(self._passphrase):
             return _ffi.callback("pem_password_cb", self._read_passphrase)
         else:
             raise TypeError(
@@ -2801,9 +2799,7 @@ class _PassphraseHelper(object):
     def callback_args(self):
         if self._passphrase is None:
             return _ffi.NULL
-        elif isinstance(self._passphrase, bytes):
-            return self._passphrase
-        elif callable(self._passphrase):
+        elif isinstance(self._passphrase, bytes) or callable(self._passphrase):
             return _ffi.NULL
         else:
             raise TypeError(
@@ -2823,12 +2819,15 @@ class _PassphraseHelper(object):
 
     def _read_passphrase(self, buf, size, rwflag, userdata):
         try:
-            if self._more_args:
-                result = self._passphrase(size, rwflag, userdata)
+            if callable(self._passphrase):
+                if self._more_args:
+                    result = self._passphrase(size, rwflag, userdata)
+                else:
+                    result = self._passphrase(rwflag)
             else:
-                result = self._passphrase(rwflag)
+                result = self._passphrase
             if not isinstance(result, bytes):
-                raise ValueError("String expected")
+                raise ValueError("Bytes expected")
             if len(result) > size:
                 if self._truncate:
                     result = result[:size]
diff --git a/tests/test_crypto.py b/tests/test_crypto.py
index 815bd8b..62b1690 100644
--- a/tests/test_crypto.py
+++ b/tests/test_crypto.py
@@ -516,6 +516,23 @@ MbzjS007Oe4qqBnCWaFPSnJX6uLApeTbqAxAeyCql56ULW5x6vDMNC3dwjvS/CEh
 
 encryptedPrivateKeyPEMPassphrase = b"foobar"
 
+cleartextPrivateKeyPEM = """-----BEGIN PRIVATE KEY-----
+MIICdwIBADANBgkqhkiG9w0BAQEFAASCAmEwggJdAgEAAoGBAMcRMugJ4kvkOEuT
+AvMFr9+3A6+HAB6nKYcXXZz93ube8rJpBZQEfWn73H10dQiQR/a+rhxYEeLy8dPc
+UkFcGR9miVkukJ59zex7iySJY76bdBD8gyx1LTKrkCstP2XHKEYqgbj+tm7VzJnY
+sQLqoaa5NeyWJnUC3MJympkAS7p3AgMBAAECgYAoBAcNqd75jnjaiETRgVUnTWzK
+PgMCJmwsob/JrSa/lhWHU6Exbe2f/mcGOQDFpesxaIcrX3DJBDkkc2d9h/vsfo5v
+JLk/rbHoItWxwuY5n5raAPeQPToKpTDxDrL6Ejhgcxd19wNht7/XSrYZ+dq3iU6G
+mOEvU2hrnfIW3kwVYQJBAP62G6R0gucNfaKGtHzfR3TN9G/DnCItchF+TxGTtpdh
+Cz32MG+7pirT/0xunekmUIp15QHdRy496sVxWTCooLkCQQDIEwXTAwhLNRGFEs5S
+jSkxNfTVeNiOzlG8jPBJJDAdlLt1gUqjZWnk9yU+itMSGi/6eeuH2n04FFk+SV/T
+7ryvAkB0y0ZDk5VOozX/p2rtc2iNm77A3N4kIdiTQuq4sZXhNgN0pwWwxke8jbcb
+8gEAnqwBwWt//locTxHu9TmjgT8pAkEAlbF16B0atXptM02QxT8MlN8z4gxaqu4/
+RX2FwpOq1FcVsqMbvwj/o+ouGY8wwRiK0TMrQCf/DFhdNTcc1aqHzQJBAKWtq4LI
+uVZjCAuyrqEnt7R1bOiLrar+/ezJPY2z+f2rb1TGr31ztPeFvO3edLw+QdhzwJGp
+QKImYzqMe+zkIOQ=
+-----END PRIVATE KEY-----
+"""
 
 cleartextPublicKeyPEM = b"""-----BEGIN PUBLIC KEY-----
 MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAxszlc+b71LvlLS0ypt/l
@@ -3167,6 +3184,44 @@ class TestFunction(object):
         with pytest.raises(ValueError):
             dump_privatekey(FILETYPE_PEM, key, GOOD_CIPHER, cb)
 
+    def test_dump_privatekey_truncated(self):
+        """
+        `crypto.dump_privatekey` should not truncate a passphrase that contains
+        a null byte.
+        """
+        key = load_privatekey(FILETYPE_PEM, cleartextPrivateKeyPEM)
+        passphrase = b"foo\x00bar"
+        truncated_passphrase = passphrase.split(b"\x00", 1)[0]
+
+        # By dumping with the full passphrase load should raise an error if we
+        # try to load using the truncated passphrase. If dump truncated the
+        # passphrase, then we WILL load the privatekey and the test fails
+        encrypted_key_pem = dump_privatekey(
+            FILETYPE_PEM, key, "AES-256-CBC", passphrase
+        )
+        with pytest.raises(Error):
+            load_privatekey(
+                FILETYPE_PEM, encrypted_key_pem, truncated_passphrase
+            )
+
+    def test_load_privatekey_truncated(self):
+        """
+        `crypto.load_privatekey` should not truncate a passphrase that contains
+        a null byte.
+        """
+        key = load_privatekey(FILETYPE_PEM, cleartextPrivateKeyPEM)
+        passphrase = b"foo\x00bar"
+        truncated_passphrase = passphrase.split(b"\x00", 1)[0]
+
+        # By dumping using the truncated passphrase load should raise an error
+        # if we try to load using the full passphrase. If load truncated the
+        # passphrase, then we WILL load the privatekey and the test fails
+        encrypted_key_pem = dump_privatekey(
+            FILETYPE_PEM, key, "AES-256-CBC", truncated_passphrase
+        )
+        with pytest.raises(Error):
+            load_privatekey(FILETYPE_PEM, encrypted_key_pem, passphrase)
+
     def test_load_pkcs7_data_pem(self):
         """
         `load_pkcs7_data` accepts a PKCS#7 string and returns an instance of