Adding protected method to convert PKCS12 key to PEM.
This commit is contained in:
@@ -137,7 +137,6 @@ try:
|
||||
password = password.encode('utf-8')
|
||||
pkey = crypto.load_pkcs12(key, password).get_privatekey()
|
||||
return OpenSSLSigner(pkey)
|
||||
|
||||
except ImportError:
|
||||
OpenSSLVerifier = None
|
||||
OpenSSLSigner = None
|
||||
@@ -287,6 +286,39 @@ def _parse_pem_key(raw_key_input):
|
||||
return raw_key_input[offset:]
|
||||
|
||||
|
||||
def private_key_as_pem(private_key_text, private_key_password=None):
|
||||
"""Convert the contents of a key to PEM.
|
||||
|
||||
First tries to determine if the current key is PEM, then tries to
|
||||
use OpenSSL to convert from PKCS12 to PEM.
|
||||
|
||||
Args:
|
||||
private_key_text: String. Private key.
|
||||
private_key_password: Optional string. Password for PKCS12.
|
||||
|
||||
Returns:
|
||||
String. PEM contents of ``private_key_text``.
|
||||
|
||||
Raises:
|
||||
ImportError: If key is PKCS12 and OpenSSL is not installed.
|
||||
"""
|
||||
decoded_body = base64.b64decode(private_key_text)
|
||||
pem_contents = _parse_pem_key(decoded_body)
|
||||
if pem_contents is None:
|
||||
if OpenSSLVerifier is None or OpenSSLSigner is None:
|
||||
raise ImportError('OpenSSL not installed. Required to convert '
|
||||
'PKCS12 key to PEM.')
|
||||
|
||||
if isinstance(private_key_password, six.string_types):
|
||||
private_key_password = private_key_password.encode('ascii')
|
||||
|
||||
pkcs12 = crypto.load_pkcs12(decoded_body, private_key_password)
|
||||
pem_contents = crypto.dump_privatekey(crypto.FILETYPE_PEM,
|
||||
pkcs12.get_privatekey())
|
||||
|
||||
return pem_contents
|
||||
|
||||
|
||||
def _urlsafe_b64encode(raw_bytes):
|
||||
if isinstance(raw_bytes, six.text_type):
|
||||
raw_bytes = raw_bytes.encode('utf-8')
|
||||
|
||||
@@ -23,19 +23,21 @@ Unit tests for oauth2client.
|
||||
__author__ = 'jcgregorio@google.com (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,51 @@ class CryptTests(unittest.TestCase):
|
||||
self._check_jwt_failure(jwt, 'Wrong recipient')
|
||||
|
||||
|
||||
class Test_crypt_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(
|
||||
'some_account@example.com',
|
||||
private_key,
|
||||
scope='read+write',
|
||||
sub='joe@example.org')
|
||||
|
||||
def test_succeeds(self):
|
||||
self.assertEqual(True, HAS_OPENSSL)
|
||||
|
||||
credentials = self._make_signed_jwt_creds()
|
||||
pem_contents = crypt.private_key_as_pem(
|
||||
credentials.private_key,
|
||||
private_key_password=credentials.private_key_password)
|
||||
|
||||
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.crypt.OpenSSLSigner', None):
|
||||
self.assertRaises(ImportError, crypt.private_key_as_pem,
|
||||
credentials.private_key,
|
||||
private_key_password=credentials.private_key_password)
|
||||
|
||||
def test_with_pem_key(self):
|
||||
credentials = self._make_signed_jwt_creds(private_key_file='privatekey.pem')
|
||||
pem_contents = crypt.private_key_as_pem(
|
||||
credentials.private_key,
|
||||
private_key_password=credentials.private_key_password)
|
||||
expected_pem_key = datafile('privatekey.pem')
|
||||
self.assertEqual(pem_contents, expected_pem_key)
|
||||
|
||||
def test_with_nonsense_key(self):
|
||||
credentials = self._make_signed_jwt_creds(private_key=b'NOT_A_KEY')
|
||||
self.assertRaises(crypt.crypto.Error, crypt.private_key_as_pem,
|
||||
credentials.private_key,
|
||||
private_key_password=credentials.private_key_password)
|
||||
|
||||
|
||||
class PEMCryptTestsPyCrypto(CryptTests):
|
||||
def setUp(self):
|
||||
self.format = 'pem'
|
||||
@@ -291,6 +339,7 @@ class PEMSignedJwtAssertionCredentialsPyCryptoTests(
|
||||
|
||||
|
||||
class PKCSSignedJwtAssertionCredentialsPyCryptoTests(unittest.TestCase):
|
||||
|
||||
def test_for_failure(self):
|
||||
crypt.Signer = crypt.PyCryptoSigner
|
||||
private_key = datafile('privatekey.p12')
|
||||
@@ -311,5 +360,6 @@ class TestHasOpenSSLFlag(unittest.TestCase):
|
||||
self.assertEqual(True, HAS_OPENSSL)
|
||||
self.assertEqual(True, HAS_CRYPTO)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user