Moving private_key_as_pem->pkcs12_key_as_pem.
Also only defining if OpenSSL is installed and conditionally defining a method which raises NotImplementedError if not defined.
This commit is contained in:
@@ -137,9 +137,30 @@ try:
|
||||
password = password.encode('utf-8')
|
||||
pkey = crypto.load_pkcs12(key, password).get_privatekey()
|
||||
return OpenSSLSigner(pkey)
|
||||
|
||||
|
||||
def pkcs12_key_as_pem(private_key_text, private_key_password):
|
||||
"""Convert the contents of a PKCS12 key to PEM using OpenSSL.
|
||||
|
||||
Args:
|
||||
private_key_text: String. Private key.
|
||||
private_key_password: String. Password for PKCS12.
|
||||
|
||||
Returns:
|
||||
String. PEM contents of ``private_key_text``.
|
||||
"""
|
||||
decoded_body = base64.b64decode(private_key_text)
|
||||
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)
|
||||
return crypto.dump_privatekey(crypto.FILETYPE_PEM,
|
||||
pkcs12.get_privatekey())
|
||||
except ImportError:
|
||||
OpenSSLVerifier = None
|
||||
OpenSSLSigner = None
|
||||
def pkcs12_key_as_pem(*args, **kwargs):
|
||||
raise NotImplementedError('pkcs12_key_as_pem requires OpenSSL.')
|
||||
|
||||
|
||||
try:
|
||||
@@ -286,39 +307,6 @@ 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')
|
||||
|
||||
73
tests/test_crypt.py
Normal file
73
tests/test_crypt.py
Normal file
@@ -0,0 +1,73 @@
|
||||
# Copyright 2014 Google Inc. All rights reserved.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import mock
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
try:
|
||||
reload
|
||||
except NameError:
|
||||
# For Python3 (though importlib should be used, silly 3.3).
|
||||
from imp import reload
|
||||
|
||||
from oauth2client.client import HAS_OPENSSL
|
||||
from oauth2client.client import SignedJwtAssertionCredentials
|
||||
from oauth2client import crypt
|
||||
|
||||
|
||||
def datafile(filename):
|
||||
f = open(os.path.join(os.path.dirname(__file__), 'data', filename), 'rb')
|
||||
data = f.read()
|
||||
f.close()
|
||||
return data
|
||||
|
||||
|
||||
class Test_pkcs12_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.pkcs12_key_as_pem(credentials.private_key,
|
||||
credentials.private_key_password)
|
||||
pkcs12_key_as_pem = datafile('pem_from_pkcs12.pem')
|
||||
pkcs12_key_as_pem = crypt._parse_pem_key(pkcs12_key_as_pem)
|
||||
self.assertEqual(pem_contents, pkcs12_key_as_pem)
|
||||
|
||||
def test_without_openssl(self):
|
||||
openssl_mod = sys.modules['OpenSSL']
|
||||
try:
|
||||
sys.modules['OpenSSL'] = None
|
||||
reload(crypt)
|
||||
self.assertRaises(NotImplementedError, crypt.pkcs12_key_as_pem,
|
||||
'FOO', 'BAR')
|
||||
finally:
|
||||
sys.modules['OpenSSL'] = openssl_mod
|
||||
reload(crypt)
|
||||
|
||||
def test_with_nonsense_key(self):
|
||||
credentials = self._make_signed_jwt_creds(private_key=b'NOT_A_KEY')
|
||||
self.assertRaises(crypt.crypto.Error, crypt.pkcs12_key_as_pem,
|
||||
credentials.private_key, credentials.private_key_password)
|
||||
@@ -188,51 +188,6 @@ 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'
|
||||
|
||||
Reference in New Issue
Block a user