Fixed processing of encrypted private keys. Added tests for encrypted private keys

diff --git a/ctypescrypto/pkey.py b/ctypescrypto/pkey.py
index 59da723f7b93f926a44b2d766c19c00842ea939a..af31a6725bc5ed04726ee7c401bed94e3ee331e6 100644 (file)
--- a/ctypescrypto/pkey.py
+++ b/ctypescrypto/pkey.py
@@ -5,33 +5,45 @@ PKey object of this module is wrapper around OpenSSL EVP_PKEY object.
 """
 
 
 """
 
 

-from ctypes import c_char_p, c_void_p, c_int, c_long, POINTER
+from ctypes import c_char, c_char_p, c_void_p, c_int, c_long, POINTER

 from ctypes import create_string_buffer, byref, memmove, CFUNCTYPE
 from ctypescrypto import libcrypto
 from ctypescrypto.exception import LibCryptoError, clear_err_stack
 from ctypescrypto.bio import Membio
 
 from ctypes import create_string_buffer, byref, memmove, CFUNCTYPE
 from ctypescrypto import libcrypto
 from ctypescrypto.exception import LibCryptoError, clear_err_stack
 from ctypescrypto.bio import Membio
 

-__all__ = ['PKeyError', 'password_callback', 'PKey', 'PW_CALLBACK_FUNC']
+__all__ = ['PKeyError', 'PKey', 'PW_CALLBACK_FUNC']

 class PKeyError(LibCryptoError):
     """ Exception thrown if libcrypto finctions return an error """
     pass
 
 class PKeyError(LibCryptoError):
     """ Exception thrown if libcrypto finctions return an error """
     pass
 

-PW_CALLBACK_FUNC = CFUNCTYPE(c_int, c_char_p, c_int, c_int, c_char_p)
+PW_CALLBACK_FUNC = CFUNCTYPE(c_int, POINTER(c_char), c_int, c_int, c_char_p)

 """ Function type for pem password callback """
 
 """ Function type for pem password callback """
 

-def password_callback(buf, length, rwflag, userdata):
+def _password_callback(c):

     """
     """

-    Example password callback for private key. Assumes that
-    password is stored in the userdata parameter, so allows to pass password
-    from constructor arguments to the libcrypto keyloading functions
+    Converts given user function or string to C password callback
+    function, passable to openssl.
+
+    IF function is passed, it would be called upon reading or writing
+    PEM format private key with one argument which is True if we are
+    writing key and should verify passphrase and false if we are reading
+

     """
     """

-    cnt = len(userdata)
-    if length < cnt:
-        cnt = length
-    memmove(buf, userdata, cnt)
-    return cnt
+    if  c is None:
+        return PW_CALLBACK_FUNC(0)
+    if callable(c):
+         def __cb(buf, length, rwflag, userdata):
+             pwd = c(rwflag)
+             cnt = min(len(pwd),length)
+             memmove(buf,pwd, cnt)
+             return cnt
+    else:
+        def __cb(buf,length,rwflag,userdata):
+            cnt=min(len(c),length)
+            memmove(buf,c,cnt)
+            return cnt
+    return PW_CALLBACK_FUNC(__cb)        

 
 

-_cb = PW_CALLBACK_FUNC(password_callback)

 
 class PKey(object):
     """
 
 class PKey(object):
     """


@@ -48,7 +60,36 @@ class PKey(object):
          libcrypto routines
     """
     def __init__(self, ptr=None, privkey=None, pubkey=None, format="PEM",
          libcrypto routines
     """
     def __init__(self, ptr=None, privkey=None, pubkey=None, format="PEM",

-                 cansign=False, password=None, callback=_cb):
+                 cansign=False, password=None):
+        """
+        PKey object can be created from either private/public key blob or
+        from C language pointer, returned by some OpenSSL function
+
+        Following named arguments are recognized by constructor
+
+        privkey - private key blob. If this is specified, format and
+             password can be also specified
+
+        pubkey - public key blob. If this is specified, format can be
+                specified.
+
+        ptr - pointer, returned by openssl function. If it is specified,    
+            cansign should be also specified.
+
+        These three arguments are mutually exclusive.
+
+        format - can be either 'PEM' or 'DER'. Specifies format of blob.
+        
+        password - can be string with password for encrypted key, or
+            callable  with one boolean argument, which returns password.
+            During constructor call this argument would be false.
+
+        If key is in PEM format, its encrypted status and format is
+        autodetected. If key is in DER format, than if password is
+        specified, key is assumed to be encrypted PKCS8 key otherwise
+        it is assumed to be unencrypted.
+        """
+

         if not ptr is None:
             self.key = ptr
             self.cansign = cansign
         if not ptr is None:
             self.key = ptr
             self.cansign = cansign


@@ -63,10 +104,15 @@ class PKey(object):
             self.cansign = True
             if format == "PEM":
                 self.key = libcrypto.PEM_read_bio_PrivateKey(bio.bio, None,
             self.cansign = True
             if format == "PEM":
                 self.key = libcrypto.PEM_read_bio_PrivateKey(bio.bio, None,

-                                                             callback,
-                                                             c_char_p(password))
+                                              _password_callback(password),
+                                               None)

             else:
             else:

-                self.key = libcrypto.d2i_PrivateKey_bio(bio.bio, None)
+                if password is not None:
+                    self.key = libcrypto.d2i_PKCS8PrivateKey_bio(bio.bio,None,
+                                           _password_callback(password),
+                                           None)
+                else:
+                    self.key = libcrypto.d2i_PrivateKey_bio(bio.bio, None)

             if self.key is None:
                 raise PKeyError("error parsing private key")
         elif not pubkey is None:
             if self.key is None:
                 raise PKeyError("error parsing private key")
         elif not pubkey is None:


@@ -74,8 +120,8 @@ class PKey(object):
             self.cansign = False
             if format == "PEM":
                 self.key = libcrypto.PEM_read_bio_PUBKEY(bio.bio, None,
             self.cansign = False
             if format == "PEM":
                 self.key = libcrypto.PEM_read_bio_PUBKEY(bio.bio, None,

-                                                         callback,
-                                                         c_char_p(password))
+                                                         _password_callback(password),
+                                                         None)

             else:
                 self.key = libcrypto.d2i_PUBKEY_bio(bio.bio, None)
             if self.key is None:
             else:
                 self.key = libcrypto.d2i_PUBKEY_bio(bio.bio, None)
             if self.key is None:


@@ -249,13 +295,17 @@ class PKey(object):
             raise PKeyError("error serializing public key")
         return str(bio)
 
             raise PKeyError("error serializing public key")
         return str(bio)
 

-    def exportpriv(self, format="PEM", password=None, cipher=None,
-                   callback=_cb):
+    def exportpriv(self, format="PEM", password=None, cipher=None):

         """
         Returns private key as PEM or DER Structure.
         If password and cipher are specified, encrypts key
         on given password, using given algorithm. Cipher must be
         an ctypescrypto.cipher.CipherType object
         """
         Returns private key as PEM or DER Structure.
         If password and cipher are specified, encrypts key
         on given password, using given algorithm. Cipher must be
         an ctypescrypto.cipher.CipherType object

+
+        Password can be either string or function with one argument,
+        which returns password. It is called with argument True, which
+        means, that we are encrypting key, and password should be
+        verified (requested twice from user, for example).

         """
         bio = Membio()
         if cipher is None:
         """
         bio = Membio()
         if cipher is None:


@@ -264,14 +314,14 @@ class PKey(object):
             evp_cipher = cipher.cipher
         if format == "PEM":
             ret = libcrypto.PEM_write_bio_PrivateKey(bio.bio, self.key,
             evp_cipher = cipher.cipher
         if format == "PEM":
             ret = libcrypto.PEM_write_bio_PrivateKey(bio.bio, self.key,

-                                                     evp_cipher, None, 0,
-                                                     callback,
-                                                     c_char_p(password))
+                                         evp_cipher, None, 0,
+                                         _password_callback(password),
+                                         None)

         else:
             ret = libcrypto.i2d_PKCS8PrivateKey_bio(bio.bio, self.key,
         else:
             ret = libcrypto.i2d_PKCS8PrivateKey_bio(bio.bio, self.key,

-                                                    evp_cipher, None, 0,
-                                                    callback,
-                                                    c_char_p(password))
+                                               evp_cipher, None, 0,
+                                              _password_callback(password),
+                                               None)

         if ret == 0:
             raise PKeyError("error serializing private key")
         return str(bio)
         if ret == 0:
             raise PKeyError("error serializing private key")
         return str(bio)


@@ -355,4 +405,7 @@ libcrypto.i2d_PUBKEY_bio.argtypes = (c_void_p, c_void_p)
 libcrypto.i2d_PKCS8PrivateKey_bio.argtypes = (c_void_p, c_void_p, c_void_p,
                                               c_char_p, c_int,
                                               PW_CALLBACK_FUNC, c_char_p)
 libcrypto.i2d_PKCS8PrivateKey_bio.argtypes = (c_void_p, c_void_p, c_void_p,
                                               c_char_p, c_int,
                                               PW_CALLBACK_FUNC, c_char_p)

+libcrypto.d2i_PKCS8PrivateKey_bio.restype = c_void_p                                              
+libcrypto.d2i_PKCS8PrivateKey_bio.argtypes = (c_void_p,c_void_p,
+                                              PW_CALLBACK_FUNC,c_void_p)                                            

 libcrypto.ENGINE_finish.argtypes = (c_void_p, )
 libcrypto.ENGINE_finish.argtypes = (c_void_p, )

diff --git a/tests/testpkey.py b/tests/testpkey.py
index 7e71e8c0cf88ab91bdabb22aac434f3bb89fa798..1018dc69a4caf94219de3fb09ed2a59f274f977f 100644 (file)
--- a/tests/testpkey.py
+++ b/tests/testpkey.py
@@ -1,6 +1,7 @@
 from ctypescrypto.pkey import PKey
 import unittest
 from base64 import b64decode, b16decode
 from ctypescrypto.pkey import PKey
 import unittest
 from base64 import b64decode, b16decode

+from subprocess import Popen,PIPE,CalledProcessError

 
 def pem2der(s):
     start=s.find('-----\n')
 
 def pem2der(s):
     start=s.find('-----\n')


@@ -8,6 +9,14 @@ def pem2der(s):
     data=s[start+6:finish]
     return b64decode(data)
 
     data=s[start+6:finish]
     return b64decode(data)
 

+def runopenssl(args,indata):
+    p=Popen(['openssl']+args,stdin=PIPE,stdout=PIPE,stderr=PIPE,universal_newlines=True)
+    (out,err)=p.communicate(indata)
+    if p.returncode:
+        raise CalledProcessError(p.returncode," ".join(['openssl']+args)+":"+err)
+    return out
+
+

 class TestPKey(unittest.TestCase):
     rsa="""-----BEGIN PRIVATE KEY-----
 MIICdQIBADANBgkqhkiG9w0BAQEFAASCAl8wggJbAgEAAoGBAL9CzVZu9bczTmB8
 class TestPKey(unittest.TestCase):
     rsa="""-----BEGIN PRIVATE KEY-----
 MIICdQIBADANBgkqhkiG9w0BAQEFAASCAl8wggJbAgEAAoGBAL9CzVZu9bczTmB8


@@ -26,6 +35,44 @@ Ao6uTm8fnkD4C836wS4mYAPqwRBK1JvnEXEQee9irf+ip89BAg74ViTcGF9lwJwQ
 gOM+X5Db+3pK
 -----END PRIVATE KEY-----
 """
 gOM+X5Db+3pK
 -----END PRIVATE KEY-----
 """

+    rsaenc="""-----BEGIN RSA PRIVATE KEY-----
+Proc-Type: 4,ENCRYPTED
+DEK-Info: AES-256-CBC,7FF0E46291D60D35ACA881131C244655
+
+BeJoui1lRQDvvPr+gH8xCdqkcgKCLWpZTvFZmvrqXmPqMHpm20nK0ESAd6kKm8d1
+zaglRIHnnO6V7aDcwgOd3IYPEOnG2TIRQniZrwZdrXIfacscJ6Ekq+5YfLuyrRgq
+fscGl7ntm/eGLqwrhzuv7jAXpn9QWiiAld0EWcmZCAW7nGaUQtu4rc4ULwL5SC/M
+MOCPwpcD3SCQv55dX3cBOtfZ3lPbpgEpTpnNnj8OtxOkkIaG8yol7luwHvoOSyL/
+WuXGCpfJE4LzbxnSLhbiN7q+y/Sro3cGc9aO4tXToMqTFel4zqR0YgOeFazlDRi1
+mPuZcGLuSIef0kJn7Mg7jt0DQk579rTVxAhIu2rylTwEozkpCp5g4kGTJON++HQr
+BRrApm4XlAoH2GX1jqDBoWSnXCRH49jNGQrLwy469i+994cG8uVU9Z5cqm/LDIR9
+kwQfTJIvMi0g28NBMVgJ2gLj40OczxDGyNvBIbhPNswHljfsvPVr4vtxDGx8fS0N
+lUJUOL9me+XNZ5xGHYuT5DOr7GE+H3hKEg+XfrYEete9BeI4gm9cqESvrLY9EU5Q
+tOtnKKL7SglTZ5LxPMAedADC0o01nzr+D3gAiOhSgoZTrnQsSZ7iTJOtm3vNXwJx
+AgesYmXtr5mdiBPKQ1QA/jF5LUZji+5KENd5WHNQw7tOlMLDrPFVRfLZg1AQDljx
+u16kdyb71Kk3f6GCOfUntGr+kzppc3DDT+RcLetXphOOEQRy6C6/wmz08WlAPlu5
+mFfSDijpWxoUHooQISg5mE82oR8V81aBpbLtm7KevwY=
+-----END RSA PRIVATE KEY-----
+"""
+    pkcs8crypt="""-----BEGIN ENCRYPTED PRIVATE KEY-----
+MIICoTAbBgkqhkiG9w0BBQMwDgQIipVEnsV/gQoCAggABIICgE1i42C4aBhykhOi
+EItFRE+9iBgiklGxoCJtukdp1UwDRKy/GJJ1rcS385CQy4Rs0zN8NH1faVRbf4Vt
+iNACHtJx30qMCdo64CR+GJYHS4g2lGaz7PFNma8SjnAbGYXwXkdm5zhwmiU++wC7
+W59u8oWS8Dj9dZBMzoOQGQT6xzZwQ14H65zHvC16HdKSNtRgXDWkBnD2cQzuOyuf
+rFLyEf7/FH6B7/yKDcwsEfu97uPPxMvuusD1UubWnltO/Hc2oCPibN+dGw1PY9mC
+18yGQtZkf5z30yhLosF62IVy3XY9Yf/TJYojIExoASrThGRvENzWkQ3qfnErqmng
+l+dy66bmLjicobF5bO3xAhpU1dL+4/1ba2QuthVNlg6Or/iII1ntNN4PFyYcEwmX
+e09C3dyOtV7qCq13S1bRkbZhzwi2QbLKALAzrZpF6VYmayTz8KjQOZ8BncAM+BiI
+CtwuZJoXLW9kT4D7UsaSZdjUvzBIak5qdCGWpKmahMfjEEsCg6ApuIYmFrCgiY9c
+0keYjY8DJ+4bEvqsQvTIaU9F9mFytI1E3LnR0NP1jHuOA7Jc+oNQ2adgFNj12jKQ
+qNt1bEGNCqQHSrw7JNCrB7s+QAFNqJtno6fIq7vVNkqadJlnBbCIgm7NlJeGg9j6
+a5YVNGlbs0J4dQF4Jw13302IBn3piSzthWL2gL98v/1lEwGuernEpPAjry3YhzM9
+VA/oVt22n3yVA6dOSVL1oUTJyawEqASmH0jHAzXNDz+QLSLmz82ARcZPqPvVc45e
+5h0xtqtFVkQLNbYzpNWGrx7R1hdr84nOKa8EsIxTRgEL/w9Y4Z/3xEoK2+KVBpMk
+oxUuxuU=
+-----END ENCRYPTED PRIVATE KEY-----
+"""
+    password="1111"

     rsakeytext="""Public-Key: (1024 bit)
 Modulus:
     00:bf:42:cd:56:6e:f5:b7:33:4e:60:7c:ef:be:a9:
     rsakeytext="""Public-Key: (1024 bit)
 Modulus:
     00:bf:42:cd:56:6e:f5:b7:33:4e:60:7c:ef:be:a9:


@@ -62,12 +109,48 @@ AvhqdgCWLMG0D4Rj4oCqJcyG2WH8J5+0DnGujfEA4TwJ90ECvLa2SA==
     
     def test_unencrypted_pem(self):
         key=PKey(privkey=self.rsa)
     
     def test_unencrypted_pem(self):
         key=PKey(privkey=self.rsa)

+        self.assertTrue(key.cansign)
+        self.assertIsNotNone(key.key)
+        self.assertEqual(str(key),self.rsakeytext)
+    def test_encrypted_pem(self):
+        key=PKey(privkey=self.rsaenc,password=self.password)
+        self.assertIsNotNone(key.key)
+        self.assertEqual(str(key),self.rsakeytext)
+    def test_encrypted_pem_cb(self):
+        cb=lambda x:self.password
+        key=PKey(privkey=self.rsaenc,password=cb)
+        self.assertIsNotNone(key.key)
+        self.assertEqual(str(key),self.rsakeytext)
+    def test_encryped_pem_pkcs8(self):
+        key=PKey(privkey=self.pkcs8crypt,password=self.password)
+        self.assertIsNotNone(key.key)
+        self.assertEqual(str(key),self.rsakeytext)
+    def test_encrypted_der_pkcs8(self):
+        pkcs8der = pem2der(self.pkcs8crypt)
+        key=PKey(privkey=pkcs8der,password=self.password,format="DER")

         self.assertIsNotNone(key.key)
         self.assertEqual(str(key),self.rsakeytext)
     def test_export_priv_pem(self):
         key=PKey(privkey=self.ec1priv)
         out=key.exportpriv()
         self.assertEqual(self.ec1priv,out)
         self.assertIsNotNone(key.key)
         self.assertEqual(str(key),self.rsakeytext)
     def test_export_priv_pem(self):
         key=PKey(privkey=self.ec1priv)
         out=key.exportpriv()
         self.assertEqual(self.ec1priv,out)

+    def test_export_priv_encrypt(self):
+        from ctypescrypto.cipher import CipherType
+        key=PKey(privkey=self.rsa)
+        pem=key.exportpriv(password='2222',cipher=CipherType("aes256"))
+        self.assertEqual(runopenssl(["pkey","-text_pub","-noout","-passin","pass:2222"],
+                                    pem),self.rsakeytext)
+    def test_export_priv_der(self):
+        key=PKey(privkey=self.rsa)
+        der=key.exportpriv(format="DER")
+        self.assertEqual(runopenssl(["pkey","-text_pub","-noout","-inform","DER"],
+            der),self.rsakeytext)
+    def test_export_priv_der_enc(self):
+        from ctypescrypto.cipher import CipherType
+        key=PKey(privkey=self.rsa)
+        der=key.exportpriv(format="DER",password='2222',cipher=CipherType("aes256"))
+        self.assertEqual(runopenssl(["pkcs8","-passin","pass:2222","-inform","DER"],
+            der),self.rsa)

     def test_unencrypted_pem_ec(self):
         
         key=PKey(privkey=self.ec1priv)
     def test_unencrypted_pem_ec(self):
         
         key=PKey(privkey=self.ec1priv)