Merge "Create nova cert worker for x509 support"

2 files changed, 55 insertions, 31 deletions

diff --git a/nova/tests/test_auth.py b/nova/tests/test_auth.py
index b1feb0856..bdc7f3142 100644
--- a/nova/tests/test_auth.py
+++ b/nova/tests/test_auth.py
@@ -16,7 +16,6 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-from M2Crypto import X509
 import unittest
 
 from nova import crypto
@@ -245,28 +244,6 @@ class _AuthManagerBaseTestCase(test.TestCase):
                                                        project))
                 self.assertFalse(self.manager.is_project_member(user, project))
 
-    def test_can_generate_x509(self):
-        # NOTE(todd): this doesn't assert against the auth manager
-        #             so it probably belongs in crypto_unittest
-        #             but I'm leaving it where I found it.
-        with user_and_project_generator(self.manager) as (user, project):
-            # NOTE(vish): Setup runs genroot.sh if it hasn't been run
-            cloud.CloudController().setup()
-            _key, cert_str = crypto.generate_x509_cert(user.id, project.id)
-            LOG.debug(cert_str)
-
-            int_cert = crypto.fetch_ca(project_id=project.id)
-            cloud_cert = crypto.fetch_ca()
-            signed_cert = X509.load_cert_string(cert_str)
-            int_cert = X509.load_cert_string(int_cert)
-            cloud_cert = X509.load_cert_string(cloud_cert)
-            self.assertTrue(signed_cert.verify(int_cert.get_pubkey()))
-
-            if not FLAGS.use_project_ca:
-                self.assertTrue(signed_cert.verify(cloud_cert.get_pubkey()))
-            else:
-                self.assertFalse(signed_cert.verify(cloud_cert.get_pubkey()))
-
     def test_adding_role_to_project_is_ignored_unless_added_to_user(self):
         with user_and_project_generator(self.manager) as (user, project):
             self.assertFalse(self.manager.has_role(user, 'sysadmin', project))
diff --git a/nova/tests/test_crypto.py b/nova/tests/test_crypto.py
index 6c25b396e..b5f70ea72 100644
--- a/nova/tests/test_crypto.py
+++ b/nova/tests/test_crypto.py
@@ -16,12 +16,20 @@
 Tests for Crypto module.
 """
 
+import os
+import shutil
+import tempfile
+
 import mox
-import stubout
+from M2Crypto import X509
 
 from nova import crypto
 from nova import db
+from nova import flags
 from nova import test
+from nova import utils
+
+FLAGS = flags.FLAGS
 
 
 class SymmetricKeyTestCase(test.TestCase):
@@ -52,15 +60,54 @@ class SymmetricKeyTestCase(test.TestCase):
         self.assertEquals(plain_text, plain)
 
 
-class RevokeCertsTest(test.TestCase):
+class X509Test(test.TestCase):
+    def test_can_generate_x509(self):
+        tmpdir = tempfile.mkdtemp()
+        self.flags(ca_path=tmpdir)
+        try:
+            crypto.ensure_ca_filesystem()
+            _key, cert_str = crypto.generate_x509_cert('fake', 'fake')
+
+            project_cert = crypto.fetch_ca(project_id='fake')
+            cloud_cert = crypto.fetch_ca()
+            # TODO(vish): This will need to be replaced with something else
+            #             when we remove M2Crypto
+            signed_cert = X509.load_cert_string(cert_str)
+            project_cert = X509.load_cert_string(project_cert)
+            cloud_cert = X509.load_cert_string(cloud_cert)
+            self.assertTrue(signed_cert.verify(project_cert.get_pubkey()))
+
+            if not FLAGS.use_project_ca:
+                self.assertTrue(signed_cert.verify(cloud_cert.get_pubkey()))
+            else:
+                self.assertFalse(signed_cert.verify(cloud_cert.get_pubkey()))
+        finally:
+            shutil.rmtree(tmpdir)
+
+    def test_encrypt_decrypt_x509(self):
+        tmpdir = tempfile.mkdtemp()
+        self.flags(ca_path=tmpdir)
+        project_id = "fake"
+        try:
+            crypto.ensure_ca_filesystem()
+            cert = crypto.fetch_ca(project_id)
+            public_key = os.path.join(tmpdir, "public.pem")
+            with open(public_key, 'w') as keyfile:
+                keyfile.write(cert)
+            text = "some @#!%^* test text"
+            enc, _err = utils.execute('openssl',
+                                     'rsautl',
+                                     '-certin',
+                                     '-encrypt',
+                                     '-inkey', '%s' % public_key,
+                                     process_input=text)
+            dec = crypto.decrypt_text(project_id, enc)
+            self.assertEqual(text, dec)
+        finally:
+            shutil.rmtree(tmpdir)
 
-    def setUp(self):
-        super(RevokeCertsTest, self).setUp()
-        self.stubs = stubout.StubOutForTesting()
 
-    def tearDown(self):
-        self.stubs.UnsetAll()
-        super(RevokeCertsTest, self).tearDown()
+class RevokeCertsTest(test.TestCase):
 
     def test_revoke_certs_by_user_and_project(self):
         user_id = 'test_user'