Adding protected method to convert PKCS12 key to PEM.

oauth2client/client.py

@@ -1503,6 +1503,31 @@ def _generate_assertion(self):
     return crypt.make_signed_jwt(crypt.Signer.from_string(
         private_key, self.private_key_password), payload)
 
+  def _private_key_as_pem(self):
+    """Convert the contents of the current key to PEM.
+
+    First tries to determine if the current key is PEM, then tries to
+    use OpenSSL to convert from PKCS12 to PEM.
+
+    Returns:
+        String. PEM contents of ``private_key``.
+
+    Raises:
+      ImportError: If key is PKCS12 and OpenSSL is not installed.
+    """
+    decoded_body = base64.b64decode(self.private_key)
+    pem_contents = crypt._parse_pem_key(decoded_body)
+    if pem_contents is None:
+      if not HAS_OPENSSL:
+        raise ImportError('OpenSSL not installed. Required to convert '
+                          'PKCS12 key to PEM.')
+      pkcs12 = crypt.crypto.load_pkcs12(decoded_body, self.private_key_password)
+      pem_contents = crypt.crypto.dump_privatekey(crypt.crypto.FILETYPE_PEM,
+                                                  pkcs12.get_privatekey())
+
+    return pem_contents
+
+
 # Only used in verify_id_token(), which is always calling to the same URI
 # for the certs.
 _cached_http = httplib2.Http(MemoryCache())

tests/test_jwt.py

@@ -23,19 +23,21 @@
 __author__ = '[email protected] (Joe Gregorio)'
 
 import os
+import mock
 import sys
 import tempfile
 import time
 import unittest
 
 from .http_mock import HttpMockSequence
-from oauth2client import crypt
+from oauth2client import client
 from oauth2client.client import Credentials
 from oauth2client.client import SignedJwtAssertionCredentials
 from oauth2client.client import VerifyJwtTokenError
 from oauth2client.client import verify_id_token
 from oauth2client.client import HAS_OPENSSL
 from oauth2client.client import HAS_CRYPTO
+from oauth2client import crypt
 from oauth2client.file import Storage
 
 
@@ -47,6 +49,7 @@ def datafile(filename):
 
 
 class CryptTests(unittest.TestCase):
+
   def setUp(self):
     self.format = 'p12'
     self.signer = crypt.OpenSSLSigner
@@ -185,6 +188,43 @@ def test_verify_id_token_bad_tokens(self):
     self._check_jwt_failure(jwt, 'Wrong recipient')
 
 
+class Test__private_key_as_pem(unittest.TestCase):
+
+  def _make_signed_jwt_creds(self, private_key_file='privatekey.p12',
+                             private_key=None):
+    private_key = private_key or datafile(private_key_file)
+    return SignedJwtAssertionCredentials(
+        '[email protected]',
+        private_key,
+        scope='read+write',
+        sub='[email protected]')
+
+  def test_succeeds(self):
+    self.assertEqual(True, HAS_OPENSSL)
+
+    credentials = self._make_signed_jwt_creds()
+    pem_contents = credentials._private_key_as_pem()
+
+    private_key_as_pem = datafile('pem_from_pkcs12.pem')
+    private_key_as_pem = crypt._parse_pem_key(private_key_as_pem)
+    self.assertEqual(pem_contents, private_key_as_pem)
+
+  def test_without_openssl(self):
+    credentials = self._make_signed_jwt_creds()
+    with mock.patch('oauth2client.client.HAS_OPENSSL', False):
+      self.assertRaises(ImportError, credentials._private_key_as_pem)
+
+  def test_with_pem_key(self):
+    credentials = self._make_signed_jwt_creds(private_key_file='privatekey.pem')
+    expected_pem_key = datafile('privatekey.pem')
+    self.assertEqual(credentials._private_key_as_pem(),
+                     expected_pem_key)
+
+  def test_with_nonsense_key(self):
+    credentials = self._make_signed_jwt_creds(private_key='NOT_A_KEY')
+    self.assertRaises(crypt.crypto.Error, credentials._private_key_as_pem)
+
+
 class PEMCryptTestsPyCrypto(CryptTests):
   def setUp(self):
     self.format = 'pem'
@@ -291,6 +331,7 @@ def setUp(self):
 
 
 class PKCSSignedJwtAssertionCredentialsPyCryptoTests(unittest.TestCase):
+
   def test_for_failure(self):
     crypt.Signer = crypt.PyCryptoSigner
     private_key = datafile('privatekey.p12')
@@ -311,5 +352,6 @@ def test_true(self):
     self.assertEqual(True, HAS_OPENSSL)
     self.assertEqual(True, HAS_CRYPTO)
 
+
 if __name__ == '__main__':
   unittest.main()