Merge "Add encryption method using an ssh public key."

nova/crypto.py

@@ -171,13 +171,44 @@ def decrypt_text(project_id, text):
         raise exception.ProjectNotFound(project_id=project_id)
     try:
         dec, _err = utils.execute('openssl',
-                                 'rsautl',
-                                 '-decrypt',
-                                 '-inkey', '%s' % private_key,
-                                 process_input=text)
+                                  'rsautl',
+                                  '-decrypt',
+                                  '-inkey', '%s' % private_key,
+                                  process_input=text)
         return dec
-    except exception.ProcessExecutionError:
-        raise exception.DecryptionFailure()
+    except exception.ProcessExecutionError as exc:
+        raise exception.DecryptionFailure(reason=exc.stderr)
+
+
+def ssh_encrypt_text(ssh_public_key, text):
+    """Encrypt text with an ssh public key.
+
+    Requires recent ssh-keygen binary in addition to openssl binary.
+    """
+    with utils.tempdir() as tmpdir:
+        sshkey = os.path.abspath(os.path.join(tmpdir, 'ssh.key'))
+        with open(sshkey, 'w') as f:
+            f.write(ssh_public_key)
+        sslkey = os.path.abspath(os.path.join(tmpdir, 'ssl.key'))
+        try:
+            # NOTE(vish): -P is to skip prompt on bad keys
+            out, _err = utils.execute('ssh-keygen',
+                                      '-P', '',
+                                      '-e',
+                                      '-f', sshkey,
+                                      '-m', 'PKCS8')
+            with open(sslkey, 'w') as f:
+                f.write(out)
+            enc, _err = utils.execute('openssl',
+                                      'rsautl',
+                                      '-encrypt',
+                                      '-pubin',
+                                      '-inkey', sslkey,
+                                      '-keyform', 'PEM',
+                                      process_input=text)
+            return enc
+        except exception.ProcessExecutionError as exc:
+            raise exception.EncryptionFailure(reason=exc.stderr)
 
 
 def revoke_cert(project_id, file_name):

nova/exception.py

@@ -179,8 +179,12 @@ class DBDuplicateEntry(DBError):
         super(DBDuplicateEntry, self).__init__(inner_exception)
 
 
+class EncryptionFailure(NovaException):
+    message = _("Failed to encrypt text: %(reason)s")
+
+
 class DecryptionFailure(NovaException):
-    message = _("Failed to decrypt text")
+    message = _("Failed to decrypt text: %(reason)s")
 
 
 class VirtualInterfaceCreateException(NovaException):

nova/tests/test_crypto.py

@@ -149,3 +149,66 @@ class CertExceptionTests(test.TestCase):
 
             self.assertRaises(exception.CryptoCRLFileNotFound,
                               crypto.fetch_crl, project_id='fake')
+
+
+class EncryptionTests(test.TestCase):
+    pubkey = ("ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDArtgrfBu/g2o28o+H2ng/crv"
+              "zgES91i/NNPPFTOutXelrJ9QiPTPTm+B8yspLsXifmbsmXztNOlBQgQXs6usxb4"
+              "fnJKNUZ84Vkp5esbqK/L7eyRqwPvqo7btKBMoAMVX/kUyojMpxb7Ssh6M6Y8cpi"
+              "goi+MSDPD7+5yRJ9z4mH9h7MCY6Ejv8KTcNYmVHvRhsFUcVhWcIISlNWUGiG7rf"
+              "oki060F5myQN3AXcL8gHG5/Qb1RVkQFUKZ5geQ39/wSyYA1Q65QTba/5G2QNbl2"
+              "0eAIBTyKZhN6g88ak+yARa6BLLDkrlP7L4WctHQMLsuXHohQsUO9AcOlVMARgrg"
+              "uF test@test")
+    prikey = """-----BEGIN RSA PRIVATE KEY-----
+MIIEpQIBAAKCAQEAwK7YK3wbv4NqNvKPh9p4P3K784BEvdYvzTTzxUzrrV3payfU
+Ij0z05vgfMrKS7F4n5m7Jl87TTpQUIEF7OrrMW+H5ySjVGfOFZKeXrG6ivy+3ska
+sD76qO27SgTKADFV/5FMqIzKcW+0rIejOmPHKYoKIvjEgzw+/uckSfc+Jh/YezAm
+OhI7/Ck3DWJlR70YbBVHFYVnCCEpTVlBohu636JItOtBeZskDdwF3C/IBxuf0G9U
+VZEBVCmeYHkN/f8EsmANUOuUE22v+RtkDW5dtHgCAU8imYTeoPPGpPsgEWugSyw5
+K5T+y+FnLR0DC7Llx6IULFDvQHDpVTAEYK4LhQIDAQABAoIBAF9ibrrgHnBpItx+
+qVUMbriiGK8LUXxUmqdQTljeolDZi6KzPc2RVKWtpazBSvG7skX3+XCediHd+0JP
+DNri1HlNiA6B0aUIGjoNsf6YpwsE4YwyK9cR5k5YGX4j7se3pKX2jOdngxQyw1Mh
+dkmCeWZz4l67nbSFz32qeQlwrsB56THJjgHB7elDoGCXTX/9VJyjFlCbfxVCsIng
+inrNgT0uMSYMNpAjTNOjguJt/DtXpwzei5eVpsERe0TRRVH23ycS0fuq/ancYwI/
+MDr9KSB8r+OVGeVGj3popCxECxYLBxhqS1dAQyJjhQXKwajJdHFzidjXO09hLBBz
+FiutpYUCgYEA6OFikTrPlCMGMJjSj+R9woDAOPfvCDbVZWfNo8iupiECvei88W28
+RYFnvUQRjSC0pHe//mfUSmiEaE+SjkNCdnNR+vsq9q+htfrADm84jl1mfeWatg/g
+zuGz2hAcZnux3kQMI7ufOwZNNpM2bf5B4yKamvG8tZRRxSkkAL1NV48CgYEA08/Z
+Ty9g9XPKoLnUWStDh1zwG+c0q14l2giegxzaUAG5DOgOXbXcw0VQ++uOWD5ARELG
+g9wZcbBsXxJrRpUqx+GAlv2Y1bkgiPQS1JIyhsWEUtwfAC/G+uZhCX53aI3Pbsjh
+QmkPCSp5DuOuW2PybMaw+wVe+CaI/gwAWMYDAasCgYEA4Fzkvc7PVoU33XIeywr0
+LoQkrb4QyPUrOvt7H6SkvuFm5thn0KJMlRpLfAksb69m2l2U1+HooZd4mZawN+eN
+DNmlzgxWJDypq83dYwq8jkxmBj1DhMxfZnIE+L403nelseIVYAfPLOqxUTcbZXVk
+vRQFp+nmSXqQHUe5rAy1ivkCgYEAqLu7cclchCxqDv/6mc5NTVhMLu5QlvO5U6fq
+HqitgW7d69oxF5X499YQXZ+ZFdMBf19ypTiBTIAu1M3nh6LtIa4SsjXzus5vjKpj
+FdQhTBus/hU83Pkymk1MoDOPDEtsI+UDDdSDldmv9pyKGWPVi7H86vusXCLWnwsQ
+e6fCXWECgYEAqgpGvva5kJ1ISgNwnJbwiNw0sOT9BMOsdNZBElf0kJIIy6FMPvap
+6S1ziw+XWfdQ83VIUOCL5DrwmcYzLIogS0agmnx/monfDx0Nl9+OZRxy6+AI9vkK
+86A1+DXdo+IgX3grFK1l1gPhAZPRWJZ+anrEkyR4iLq6ZoPZ3BQn97U=
+-----END RSA PRIVATE KEY-----"""
+    text = "Some text! %$*"
+
+    def _ssh_decrypt_text(self, ssh_private_key, text):
+        with utils.tempdir() as tmpdir:
+            sshkey = os.path.abspath(os.path.join(tmpdir, 'ssh.key'))
+            with open(sshkey, 'w') as f:
+                f.write(ssh_private_key)
+            try:
+                dec, _err = utils.execute('openssl',
+                                          'rsautl',
+                                          '-decrypt',
+                                          '-inkey', sshkey,
+                                          process_input=text)
+                return dec
+            except exception.ProcessExecutionError as exc:
+                raise exception.DecryptionFailure(reason=exc.stderr)
+
+    def test_ssh_encrypt_decrypt_text(self):
+        enc = crypto.ssh_encrypt_text(self.pubkey, self.text)
+        self.assertNotEqual(enc, self.text)
+        result = self._ssh_decrypt_text(self.prikey, enc)
+        self.assertEqual(result, self.text)
+
+    def test_ssh_encrypt_failure(self):
+        self.assertRaises(exception.EncryptionFailure,
+                          crypto.ssh_encrypt_text, '', self.text)