x/cloudbase-init
Code Issues Proposed changes

Modified test_x509.py to work on Linux

Browse Source
This commit is contained in:
Robert Tingirica 2014-09-09 14:17:22 +03:00
parent f3e5b8d2f7
commit 8d4e3b3948
1 changed files with 102 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

184
cloudbaseinit/tests/utils/windows/test_x509.py
View File

@ -14,64 +14,68 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import importlib
import mock import mock
import sys
import unittest import unittest
from oslo.config import cfg
from cloudbaseinit.utils import s from cloudbaseinit.utils import s
from cloudbaseinit.utils import x509constants from cloudbaseinit.utils import x509constants
if sys.platform == 'win32':
from cloudbaseinit.utils.windows import cryptoapi
from cloudbaseinit.utils.windows import x509
CONF = cfg.CONF
@unittest.skipUnless(sys.platform == "win32", "requires Windows")
class CryptoAPICertManagerTests(unittest.TestCase): class CryptoAPICertManagerTests(unittest.TestCase):
def setUp(self): def setUp(self):
self._x509 = x509.CryptoAPICertManager() self._ctypes = mock.MagicMock()
self._module_patcher = mock.patch.dict(
'sys.modules', {'ctypes': self._ctypes})
self._module_patcher.start()
self.x509 = importlib.import_module("cloudbaseinit.utils.windows.x509")
self._x509_manager = self.x509.CryptoAPICertManager()
def tearDown(self):
self._module_patcher.stop()
@mock.patch('cloudbaseinit.utils.windows.x509.free') @mock.patch('cloudbaseinit.utils.windows.x509.free')
@mock.patch('ctypes.c_ubyte')
@mock.patch('ctypes.POINTER')
@mock.patch('ctypes.cast')
@mock.patch('cloudbaseinit.utils.windows.x509.malloc') @mock.patch('cloudbaseinit.utils.windows.x509.malloc')
@mock.patch('ctypes.byref')
@mock.patch('ctypes.wintypes.DWORD')
@mock.patch('cloudbaseinit.utils.windows.cryptoapi.' @mock.patch('cloudbaseinit.utils.windows.cryptoapi.'
'CertGetCertificateContextProperty') 'CertGetCertificateContextProperty')
def _test_get_cert_thumprint(self, mock_CertGetCertificateContextProperty, def _test_get_cert_thumprint(self, mock_CertGetCertificateContextProperty,
mock_DWORD, mock_byref, mock_malloc, mock_malloc, mock_free, ret_val):
mock_cast, mock_POINTER, mock_c_ubyte, mock_DWORD = self._ctypes.wintypes.DWORD
mock_free, ret_val): mock_cast = self._ctypes.cast
mock_POINTER = self._ctypes.POINTER
mock_byref = self._ctypes.byref
mock_pointer = mock.MagicMock() mock_pointer = mock.MagicMock()
fake_cert_context_p = 'fake context' fake_cert_context_p = 'fake context'
mock_DWORD().value = 10 mock_DWORD.return_value.value = 10
mock_CertGetCertificateContextProperty.return_value = ret_val mock_CertGetCertificateContextProperty.return_value = ret_val
mock_POINTER.return_value = mock_pointer mock_POINTER.return_value = mock_pointer
mock_cast().contents = [16] mock_cast.return_value.contents = [16]
if not ret_val: if not ret_val:
self.assertRaises(cryptoapi.CryptoAPIException, self.assertRaises(self.x509.cryptoapi.CryptoAPIException,
self._x509._get_cert_thumprint, self._x509_manager._get_cert_thumprint,
fake_cert_context_p) fake_cert_context_p)
else: else:
expected = [mock.call(fake_cert_context_p, expected = [mock.call(fake_cert_context_p,
cryptoapi.CERT_SHA1_HASH_PROP_ID, self.x509.cryptoapi.CERT_SHA1_HASH_PROP_ID,
None, mock_byref()), None, mock_byref.return_value),
mock.call(fake_cert_context_p, mock.call(fake_cert_context_p,
cryptoapi.CERT_SHA1_HASH_PROP_ID, self.x509.cryptoapi.CERT_SHA1_HASH_PROP_ID,
mock_malloc(), mock_byref())] mock_malloc.return_value,
response = self._x509._get_cert_thumprint(fake_cert_context_p) mock_byref.return_value)]
response = self._x509_manager._get_cert_thumprint(
fake_cert_context_p)
self.assertEqual( self.assertEqual(
mock_CertGetCertificateContextProperty.call_args_list, mock_CertGetCertificateContextProperty.call_args_list,
expected) expected)
mock_malloc.assert_called_with(mock_DWORD()) mock_malloc.assert_called_with(mock_DWORD.return_value)
mock_cast.assert_called_with(mock_malloc(), mock_pointer) mock_cast.assert_called_with(mock_malloc(), mock_pointer)
mock_free.assert_called_with(mock_malloc()) mock_free.assert_called_with(mock_malloc())
self.assertEqual(response, '10') self.assertEqual(response, '10')
@ -86,30 +90,34 @@ class CryptoAPICertManagerTests(unittest.TestCase):
@mock.patch('cloudbaseinit.utils.windows.cryptoapi.CryptReleaseContext') @mock.patch('cloudbaseinit.utils.windows.cryptoapi.CryptReleaseContext')
@mock.patch('cloudbaseinit.utils.windows.cryptoapi.CryptGenKey') @mock.patch('cloudbaseinit.utils.windows.cryptoapi.CryptGenKey')
@mock.patch('cloudbaseinit.utils.windows.cryptoapi.CryptAcquireContext') @mock.patch('cloudbaseinit.utils.windows.cryptoapi.CryptAcquireContext')
@mock.patch('ctypes.byref') def _test_generate_key(self, mock_CryptAcquireContext, mock_CryptGenKey,
@mock.patch('ctypes.wintypes.HANDLE')
def _test_generate_key(self, mock_HANDLE, mock_byref,
mock_CryptAcquireContext, mock_CryptGenKey,
mock_CryptReleaseContext, mock_CryptDestroyKey, mock_CryptReleaseContext, mock_CryptDestroyKey,
acquired_context, generate_key_ret_val): acquired_context, generate_key_ret_val):
mock_HANDLE = self._ctypes.wintypes.HANDLE
mock_byref = self._ctypes.byref
mock_CryptAcquireContext.return_value = acquired_context mock_CryptAcquireContext.return_value = acquired_context
mock_CryptGenKey.return_value = generate_key_ret_val mock_CryptGenKey.return_value = generate_key_ret_val
if not acquired_context: if not acquired_context:
self.assertRaises(cryptoapi.CryptoAPIException, self.assertRaises(self.x509.cryptoapi.CryptoAPIException,
self._x509._generate_key, self._x509_manager._generate_key,
'fake container', True) 'fake container', True)
else: else:
if generate_key_ret_val is None: if not generate_key_ret_val:
self.assertRaises(cryptoapi.CryptoAPIException, self.assertRaises(self.x509.cryptoapi.CryptoAPIException,
self._x509._generate_key, 'fake container', self._x509_manager._generate_key,
True) 'fake container', True)
else: else:
self._x509._generate_key('fake container', True) self._x509_manager._generate_key('fake container', True)
mock_CryptAcquireContext.assert_called_with( mock_CryptAcquireContext.assert_called_with(
mock_byref(), 'fake container', None, mock_byref(), 'fake container', None,
cryptoapi.PROV_RSA_FULL, cryptoapi.CRYPT_MACHINE_KEYSET) self.x509.cryptoapi.PROV_RSA_FULL,
mock_CryptGenKey.assert_called_with(mock_HANDLE(), self.x509.cryptoapi.CRYPT_MACHINE_KEYSET)
cryptoapi.AT_SIGNATURE, mock_CryptGenKey.assert_called_with(
mock_HANDLE(), self.x509.cryptoapi.AT_SIGNATURE,
0x08000000, mock_HANDLE()) 0x08000000, mock_HANDLE())
mock_CryptDestroyKey.assert_called_once_with( mock_CryptDestroyKey.assert_called_once_with(
mock_HANDLE()) mock_HANDLE())
@ -130,16 +138,12 @@ class CryptoAPICertManagerTests(unittest.TestCase):
@mock.patch('cloudbaseinit.utils.windows.x509.free') @mock.patch('cloudbaseinit.utils.windows.x509.free')
@mock.patch('copy.copy') @mock.patch('copy.copy')
@mock.patch('ctypes.byref')
@mock.patch('cloudbaseinit.utils.windows.x509.malloc') @mock.patch('cloudbaseinit.utils.windows.x509.malloc')
@mock.patch('ctypes.POINTER')
@mock.patch('ctypes.cast')
@mock.patch('cloudbaseinit.utils.windows.x509.CryptoAPICertManager' @mock.patch('cloudbaseinit.utils.windows.x509.CryptoAPICertManager'
'._generate_key') '._generate_key')
@mock.patch('cloudbaseinit.utils.windows.x509.CryptoAPICertManager' @mock.patch('cloudbaseinit.utils.windows.x509.CryptoAPICertManager'
'._get_cert_thumprint') '._get_cert_thumprint')
@mock.patch('uuid.uuid4') @mock.patch('uuid.uuid4')
@mock.patch('ctypes.wintypes.DWORD')
@mock.patch('cloudbaseinit.utils.windows.cryptoapi.' @mock.patch('cloudbaseinit.utils.windows.cryptoapi.'
'CertStrToName') 'CertStrToName')
@mock.patch('cloudbaseinit.utils.windows.cryptoapi.' @mock.patch('cloudbaseinit.utils.windows.cryptoapi.'
@ -174,13 +178,16 @@ class CryptoAPICertManagerTests(unittest.TestCase):
mock_CRYPT_ALGORITHM_IDENTIFIER, mock_CRYPT_ALGORITHM_IDENTIFIER,
mock_CRYPT_KEY_PROV_INFO, mock_CRYPT_KEY_PROV_INFO,
mock_CRYPTOAPI_BLOB, mock_CRYPTOAPI_BLOB,
mock_CertStrToName, mock_DWORD, mock_CertStrToName,
mock_uuid4, mock_get_cert_thumprint, mock_uuid4, mock_get_cert_thumprint,
mock_generate_key, mock_cast, mock_generate_key, mock_malloc,
mock_POINTER, mock_malloc, mock_byref,
mock_copy, mock_free, certstr, mock_copy, mock_free, certstr,
certificate, enhanced_key, certificate, enhanced_key, store_handle,
store_handle, context_to_store): context_to_store):
mock_POINTER = self._ctypes.POINTER
mock_byref = self._ctypes.byref
mock_cast = self._ctypes.cast
mock_uuid4.return_value = 'fake_name' mock_uuid4.return_value = 'fake_name'
mock_CertCreateSelfSignCertificate.return_value = certificate mock_CertCreateSelfSignCertificate.return_value = certificate
@ -190,11 +197,12 @@ class CryptoAPICertManagerTests(unittest.TestCase):
mock_CertAddCertificateContextToStore.return_value = context_to_store mock_CertAddCertificateContextToStore.return_value = context_to_store
if (certstr is None or certificate is None or enhanced_key is None if (certstr is None or certificate is None or enhanced_key is None
or store_handle is None or context_to_store is None): or store_handle is None or context_to_store is None):
self.assertRaises(cryptoapi.CryptoAPIException, self.assertRaises(self.x509.cryptoapi.CryptoAPIException,
self._x509.create_self_signed_cert, self._x509_manager.create_self_signed_cert,
'fake subject', 10, True, x509.STORE_NAME_MY) 'fake subject', 10, True,
self.x509.STORE_NAME_MY)
else: else:
response = self._x509.create_self_signed_cert( response = self._x509_manager.create_self_signed_cert(
subject='fake subject') subject='fake subject')
mock_cast.assert_called_with(mock_malloc(), mock_POINTER()) mock_cast.assert_called_with(mock_malloc(), mock_POINTER())
mock_CRYPTOAPI_BLOB.assert_called_once_with() mock_CRYPTOAPI_BLOB.assert_called_once_with()
@ -208,11 +216,11 @@ class CryptoAPICertManagerTests(unittest.TestCase):
mock_byref(), mock_byref(), mock_byref(), None) mock_byref(), mock_byref(), mock_byref(), None)
mock_CertAddEnhancedKeyUsageIdentifier.assert_called_with( mock_CertAddEnhancedKeyUsageIdentifier.assert_called_with(
mock_CertCreateSelfSignCertificate(), mock_CertCreateSelfSignCertificate(),
cryptoapi.szOID_PKIX_KP_SERVER_AUTH) self.x509.cryptoapi.szOID_PKIX_KP_SERVER_AUTH)
mock_CertOpenStore.assert_called_with( mock_CertOpenStore.assert_called_with(
cryptoapi.CERT_STORE_PROV_SYSTEM, 0, 0, self.x509.cryptoapi.CERT_STORE_PROV_SYSTEM, 0, 0,
cryptoapi.CERT_SYSTEM_STORE_LOCAL_MACHINE, self.x509.cryptoapi.CERT_SYSTEM_STORE_LOCAL_MACHINE,
s.unicode(x509.STORE_NAME_MY)) s.unicode(self.x509.STORE_NAME_MY))
mock_get_cert_thumprint.assert_called_once_with( mock_get_cert_thumprint.assert_called_once_with(
mock_CertCreateSelfSignCertificate()) mock_CertCreateSelfSignCertificate())
@ -272,7 +280,7 @@ class CryptoAPICertManagerTests(unittest.TestCase):
fake_cert_data += x509constants.PEM_HEADER + '\n' fake_cert_data += x509constants.PEM_HEADER + '\n'
fake_cert_data += 'fake cert' + '\n' fake_cert_data += 'fake cert' + '\n'
fake_cert_data += x509constants.PEM_FOOTER fake_cert_data += x509constants.PEM_FOOTER
response = self._x509._get_cert_base64(fake_cert_data) response = self._x509_manager._get_cert_base64(fake_cert_data)
self.assertEqual(response, 'fake cert') self.assertEqual(response, 'fake cert')
@mock.patch('cloudbaseinit.utils.windows.x509.free') @mock.patch('cloudbaseinit.utils.windows.x509.free')
@ -292,15 +300,8 @@ class CryptoAPICertManagerTests(unittest.TestCase):
'CryptStringToBinaryA') 'CryptStringToBinaryA')
@mock.patch('cloudbaseinit.utils.windows.x509.CryptoAPICertManager' @mock.patch('cloudbaseinit.utils.windows.x509.CryptoAPICertManager'
'._get_cert_base64') '._get_cert_base64')
@mock.patch('ctypes.POINTER')
@mock.patch('cloudbaseinit.utils.windows.x509.malloc') @mock.patch('cloudbaseinit.utils.windows.x509.malloc')
@mock.patch('ctypes.cast') def _test_import_cert(self, mock_malloc, mock_get_cert_base64,
@mock.patch('ctypes.byref')
@mock.patch('ctypes.wintypes.DWORD')
@mock.patch('ctypes.create_unicode_buffer')
def _test_import_cert(self, mock_create_unicode_buffer, mock_DWORD,
mock_byref, mock_cast,
mock_malloc, mock_POINTER, mock_get_cert_base64,
mock_CryptStringToBinaryA, mock_CertOpenStore, mock_CryptStringToBinaryA, mock_CertOpenStore,
mock_CertAddEncodedCertificateToStore, mock_CertAddEncodedCertificateToStore,
mock_CertGetNameString, mock_CertGetNameString,
@ -308,6 +309,13 @@ class CryptoAPICertManagerTests(unittest.TestCase):
mock_CertCloseStore, mock_get_cert_thumprint, mock_CertCloseStore, mock_get_cert_thumprint,
mock_free, crypttstr, store_handle, add_enc_cert, mock_free, crypttstr, store_handle, add_enc_cert,
upn_len): upn_len):
mock_POINTER = self._ctypes.POINTER
mock_cast = self._ctypes.cast
mock_byref = self._ctypes.byref
mock_DWORD = self._ctypes.wintypes.DWORD
mock_create_unicode_buffer = self._ctypes.create_unicode_buffer
fake_cert_data = '' fake_cert_data = ''
fake_cert_data += x509constants.PEM_HEADER + '\n' fake_cert_data += x509constants.PEM_HEADER + '\n'
fake_cert_data += 'fake cert' + '\n' fake_cert_data += 'fake cert' + '\n'
@ -319,45 +327,55 @@ class CryptoAPICertManagerTests(unittest.TestCase):
mock_CertGetNameString.side_effect = [2, upn_len] mock_CertGetNameString.side_effect = [2, upn_len]
expected = [mock.call('fake cert', len('fake cert'), expected = [mock.call('fake cert', len('fake cert'),
cryptoapi.CRYPT_STRING_BASE64, None, self.x509.cryptoapi.CRYPT_STRING_BASE64, None,
mock_byref(), None, None), mock_byref(), None, None),
mock.call('fake cert', len('fake cert'), mock.call('fake cert', len('fake cert'),
cryptoapi.CRYPT_STRING_BASE64, mock_cast(), self.x509.cryptoapi.CRYPT_STRING_BASE64,
mock_byref(), None, None)] mock_cast(), mock_byref(), None, None)]
expected2 = [mock.call(mock_POINTER()(), cryptoapi.CERT_NAME_UPN_TYPE, expected2 = [mock.call(mock_POINTER()(),
self.x509.cryptoapi.CERT_NAME_UPN_TYPE,
0, None, None, 0), 0, None, None, 0),
mock.call(mock_POINTER()(), cryptoapi.CERT_NAME_UPN_TYPE, mock.call(mock_POINTER()(),
self.x509.cryptoapi.CERT_NAME_UPN_TYPE,
0, None, mock_create_unicode_buffer(), 2)] 0, None, mock_create_unicode_buffer(), 2)]
if (not crypttstr or store_handle is None or add_enc_cert is None or if (not crypttstr or store_handle is None or add_enc_cert is None or
upn_len != 2): upn_len != 2):
self.assertRaises(cryptoapi.CryptoAPIException, self.assertRaises(self.x509.cryptoapi.CryptoAPIException,
self._x509.import_cert, fake_cert_data, True, self._x509_manager.import_cert, fake_cert_data,
x509.STORE_NAME_MY) True, self.x509.STORE_NAME_MY)
else: else:
response = self._x509.import_cert(fake_cert_data) response = self._x509_manager.import_cert(fake_cert_data)
mock_cast.assert_called_with(mock_malloc(), mock_POINTER()) mock_cast.assert_called_with(mock_malloc(), mock_POINTER())
self.assertEqual(mock_CryptStringToBinaryA.call_args_list, self.assertEqual(mock_CryptStringToBinaryA.call_args_list,
expected) expected)
mock_CertOpenStore.assert_called_with( mock_CertOpenStore.assert_called_with(
cryptoapi.CERT_STORE_PROV_SYSTEM, 0, 0, self.x509.cryptoapi.CERT_STORE_PROV_SYSTEM, 0, 0,
cryptoapi.CERT_SYSTEM_STORE_LOCAL_MACHINE, self.x509.cryptoapi.CERT_SYSTEM_STORE_LOCAL_MACHINE,
s.unicode(x509.STORE_NAME_MY)) s.unicode(self.x509.STORE_NAME_MY))
mock_CertAddEncodedCertificateToStore.assert_called_with( mock_CertAddEncodedCertificateToStore.assert_called_with(
mock_CertOpenStore(), mock_CertOpenStore(),
cryptoapi.X509_ASN_ENCODING | cryptoapi.PKCS_7_ASN_ENCODING, self.x509.cryptoapi.X509_ASN_ENCODING |
self.x509.cryptoapi.PKCS_7_ASN_ENCODING,
mock_cast(), mock_DWORD(), mock_cast(), mock_DWORD(),
cryptoapi.CERT_STORE_ADD_REPLACE_EXISTING, mock_byref()) self.x509.cryptoapi.CERT_STORE_ADD_REPLACE_EXISTING,
mock_byref())
mock_create_unicode_buffer.assert_called_with(2) mock_create_unicode_buffer.assert_called_with(2)
self.assertEqual(mock_CertGetNameString.call_args_list, expected2) self.assertEqual(mock_CertGetNameString.call_args_list, expected2)
mock_get_cert_thumprint.assert_called_once_with(mock_POINTER()()) mock_get_cert_thumprint.assert_called_once_with(mock_POINTER()())
mock_CertFreeCertificateContext.assert_called_once_with( mock_CertFreeCertificateContext.assert_called_once_with(
mock_POINTER()()) mock_POINTER()())
mock_CertCloseStore.assert_called_once_with( mock_CertCloseStore.assert_called_once_with(
mock_CertOpenStore(), 0) mock_CertOpenStore(), 0)
mock_free.assert_called_once_with(mock_cast()) mock_free.assert_called_once_with(mock_cast())
self.assertEqual(response, (mock_get_cert_thumprint(), self.assertEqual(response, (mock_get_cert_thumprint(),
mock_create_unicode_buffer().value)) mock_create_unicode_buffer().value))
mock_get_cert_base64.assert_called_with(fake_cert_data) mock_get_cert_base64.assert_called_with(fake_cert_data)
def test_import_cert(self): def test_import_cert(self):