Add encryptor attach/detach in utils

To create encrypted volume from image, or retype a volume
to different encryptions, Cinder needs to attach/detch encrypted
volume.

This patch is the first patch which adds encryptor attach/detach
functions in utils so that later it is called to resolve above problems.

Change-Id: I5b6eba7e5da54cb386bf597bf100db106c7a1b06
Implements: blueprint improve-encrypted-volume

cinder/tests/unit/test_utils.py

@@ -28,6 +28,7 @@ import webob.exc
 import cinder
 from cinder import exception
 from cinder import test
+from cinder.tests.unit import fake_constants as fake
 from cinder import utils
 
 
@@ -762,6 +763,52 @@ class BrickUtils(test.TestCase):
             'protocol', mock_helper.return_value, driver=None,
             use_multipath=False, device_scan_attempts=3)
 
+    @mock.patch('os_brick.encryptors.get_volume_encryptor')
+    @mock.patch('cinder.utils.get_root_helper')
+    def test_brick_attach_volume_encryptor(self, mock_helper,
+                                           mock_get_encryptor):
+        attach_info = {'device': {'path': 'dev/sda'},
+                       'conn': {'driver_volume_type': 'iscsi',
+                                'data': {}, }}
+        encryption = {'encryption_key_id': fake.ENCRYPTION_KEY_ID}
+        ctxt = mock.Mock(name='context')
+        mock_encryptor = mock.Mock()
+        mock_get_encryptor.return_value = mock_encryptor
+        utils.brick_attach_volume_encryptor(ctxt, attach_info, encryption)
+
+        connection_info = attach_info['conn']
+        connection_info['data']['device_path'] = attach_info['device']['path']
+        mock_helper.assert_called_once_with()
+        mock_get_encryptor.assert_called_once_with(
+            root_helper=mock_helper.return_value,
+            connection_info=connection_info,
+            keymgr=mock.ANY,
+            **encryption)
+        mock_encryptor.attach_volume.assert_called_once_with(
+            ctxt, **encryption)
+
+    @mock.patch('os_brick.encryptors.get_volume_encryptor')
+    @mock.patch('cinder.utils.get_root_helper')
+    def test_brick_detach_volume_encryptor(self,
+                                           mock_helper, mock_get_encryptor):
+        attach_info = {'device': {'path': 'dev/sda'},
+                       'conn': {'driver_volume_type': 'iscsi',
+                                'data': {}, }}
+        encryption = {'encryption_key_id': fake.ENCRYPTION_KEY_ID}
+        mock_encryptor = mock.Mock()
+        mock_get_encryptor.return_value = mock_encryptor
+        utils.brick_detach_volume_encryptor(attach_info, encryption)
+
+        mock_helper.assert_called_once_with()
+        connection_info = attach_info['conn']
+        connection_info['data']['device_path'] = attach_info['device']['path']
+        mock_get_encryptor.assert_called_once_with(
+            root_helper=mock_helper.return_value,
+            connection_info=connection_info,
+            keymgr=mock.ANY,
+            **encryption)
+        mock_encryptor.detach_volume.assert_called_once_with(**encryption)
+
 
 class StringLengthTestCase(test.TestCase):
     def test_check_string_length(self):

cinder/utils.py

@@ -37,6 +37,7 @@ import tempfile
 import time
 import types
 
+from os_brick import encryptors
 from os_brick.initiator import connector
 from oslo_concurrency import lockutils
 from oslo_concurrency import processutils
@@ -53,6 +54,7 @@ import webob.exc
 
 from cinder import exception
 from cinder.i18n import _, _LE, _LW
+from cinder import keymgr
 
 
 CONF = cfg.CONF
@@ -507,6 +509,36 @@ def brick_get_connector(protocol, driver=None,
                                                 *args, **kwargs)
 
 
+def brick_get_encryptor(connection_info, *args, **kwargs):
+    """Wrapper to get a brick encryptor object."""
+
+    root_helper = get_root_helper()
+    key_manager = keymgr.API()
+    return encryptors.get_volume_encryptor(root_helper=root_helper,
+                                           connection_info=connection_info,
+                                           keymgr=key_manager,
+                                           *args, **kwargs)
+
+
+def brick_attach_volume_encryptor(context, attach_info, encryption):
+    """Attach encryption layer."""
+    connection_info = attach_info['conn']
+    connection_info['data']['device_path'] = attach_info['device']['path']
+    encryptor = brick_get_encryptor(connection_info,
+                                    **encryption)
+    encryptor.attach_volume(context, **encryption)
+
+
+def brick_detach_volume_encryptor(attach_info, encryption):
+    """Detach encryption layer."""
+    connection_info = attach_info['conn']
+    connection_info['data']['device_path'] = attach_info['device']['path']
+
+    encryptor = brick_get_encryptor(connection_info,
+                                    **encryption)
+    encryptor.detach_volume(**encryption)
+
+
 def require_driver_initialized(driver):
     """Verifies if `driver` is initialized