Disallow unmanaging encrypted volumes

Unmanaging encrypted volumes is problematic because
unmanage assumes that you will be able to manage the
volume again for later use, but, we have no mechanism
currently to keep track of the encryption key which
would be required for using an encrypted volume again.

While this may work out ok when using the conf_key
manager, this patch does not distinguish between conf_key
and barbican deployments.

Closes-Bug: #1731518
Change-Id: I7506fa36962404c80f1cc9c6370693728e5393a7

cinder/message/defined_messages.py

@@ -27,6 +27,7 @@ class EventIds(object):
     UNABLE_TO_ALLOCATE = 'VOLUME_000002'
     ATTACH_READONLY_VOLUME = 'VOLUME_000003'
     IMAGE_FROM_VOLUME_OVER_QUOTA = 'VOLUME_000004'
+    UNMANAGE_ENCRYPTED_VOLUME_UNSUPPORTED = 'VOLUME_000005'
 
 
 event_id_message_map = {
@@ -40,6 +41,8 @@ event_id_message_map = {
     EventIds.IMAGE_FROM_VOLUME_OVER_QUOTA: _(
         "Failed to copy volume to image as image quota has been met. Please "
         "delete images or have your limit increased, then try again."),
+    EventIds.UNMANAGE_ENCRYPTED_VOLUME_UNSUPPORTED: _(
+        "Unmanaging encrypted volumes is not supported."),
 }
 
 

cinder/message/message_field.py

@@ -36,12 +36,15 @@ class Action(object):
     COPY_VOLUME_TO_IMAGE = ('003', _('copy volume to image'))
     UPDATE_ATTACHMENT = ('004', _('update attachment'))
     COPY_IMAGE_TO_VOLUME = ('005', _('copy image to volume'))
+    UNMANAGE_VOLUME = ('006', _('unmanage volume'))
 
     ALL = (SCHEDULE_ALLOCATE_VOLUME,
            ATTACH_VOLUME,
            COPY_VOLUME_TO_IMAGE,
            UPDATE_ATTACHMENT,
-           COPY_IMAGE_TO_VOLUME)
+           COPY_IMAGE_TO_VOLUME,
+           UNMANAGE_VOLUME
+           )
 
 
 class Detail(object):
@@ -62,6 +65,9 @@ class Detail(object):
     NOT_ENOUGH_SPACE_FOR_IMAGE = ('007',
                                   _("Image used for creating volume exceeds "
                                     "available space."))
+    UNMANAGE_ENC_NOT_SUPPORTED = (
+        '008',
+        _("Unmanaging encrypted volumes is not supported."))
 
     ALL = (UNKNOWN_ERROR,
            DRIVER_NOT_INITIALIZED,
@@ -69,7 +75,9 @@ class Detail(object):
            FAILED_TO_UPLOAD_VOLUME,
            VOLUME_ATTACH_MODE_INVALID,
            QUOTA_EXCEED,
-           NOT_ENOUGH_SPACE_FOR_IMAGE)
+           NOT_ENOUGH_SPACE_FOR_IMAGE,
+           UNMANAGE_ENC_NOT_SUPPORTED,
+           )
 
     # Exception and detail mappings
     EXCEPTION_DETAIL_MAPPINGS = {
@@ -79,7 +87,8 @@ class Detail(object):
         QUOTA_EXCEED: ['ImageLimitExceeded',
                        'BackupLimitExceeded',
                        'SnapshotLimitExceeded'],
-        NOT_ENOUGH_SPACE_FOR_IMAGE: ['ImageTooBig']
+        NOT_ENOUGH_SPACE_FOR_IMAGE: ['ImageTooBig'],
+        UNMANAGE_ENC_NOT_SUPPORTED: ['UnmanageEncVolNotSupported'],
     }
 
 

cinder/tests/unit/api/contrib/test_volume_unmanage.py

@@ -95,3 +95,11 @@ class VolumeUnmanageTest(test.TestCase):
         self.assertEqual(http_client.BAD_REQUEST, res.status_int, res)
         db.volume_destroy(self.ctxt, vol.id)
         db.snapshot_destroy(self.ctxt, snap.id)
+
+    def test_unmanage_encrypted_volume_denied(self):
+        vol = utils.create_volume(
+            self.ctxt,
+            encryption_key_id='7a98391f-6619-46af-bd00-5862a3f7f1bd')
+        res = self._get_resp(vol.id)
+        self.assertEqual(http_client.BAD_REQUEST, res.status_int, res)
+        db.volume_destroy(self.ctxt, vol.id)

cinder/tests/unit/volume/test_volume.py

@@ -796,6 +796,20 @@ class VolumeTestCase(base.BaseVolumeTestCase):
             self.assertEqual("available", volume_ref.status)
             mock_del_vol.assert_called_once_with(volume)
 
+    def test_unmanage_encrypted_volume_fails(self):
+        volume = tests_utils.create_volume(
+            self.context,
+            encryption_key_id=fake.ENCRYPTION_KEY_ID,
+            **self.volume_params)
+        self.volume.create_volume(self.context, volume)
+        manager = vol_manager.VolumeManager()
+        self.assertRaises(exception.Invalid,
+                          manager.delete_volume,
+                          self.context,
+                          volume,
+                          unmanage_only=True)
+        self.volume.delete_volume(self.context, volume)
+
     def test_get_volume_different_tenant(self):
         """Test can't get volume of another tenant when viewable_admin_meta."""
         volume = tests_utils.create_volume(self.context,

cinder/volume/api.py

@@ -39,6 +39,8 @@ from cinder import flow_utils
 from cinder.i18n import _
 from cinder.image import cache as image_cache
 from cinder.image import glance
+from cinder.message import api as message_api
+from cinder.message import message_field
 from cinder import objects
 from cinder.objects import base as objects_base
 from cinder.objects import fields
@@ -108,6 +110,7 @@ class API(base.Base):
         self.availability_zones = []
         self.availability_zones_last_fetched = None
         self.key_manager = key_manager.API(CONF)
+        self.message = message_api.API()
         super(API, self).__init__(db_driver)
 
     def list_availability_zones(self, enable_cache=False, refresh_cache=False):
@@ -405,6 +408,17 @@ class API(base.Base):
         if not unmanage_only:
             volume.assert_not_frozen()
 
+        if unmanage_only and volume.encryption_key_id is not None:
+            msg = _("Unmanaging encrypted volumes is not supported.")
+            e = exception.Invalid(reason=msg)
+            self.message.create(
+                context,
+                message_field.Action.UNMANAGE_VOLUME,
+                resource_uuid=volume.id,
+                detail=message_field.Detail.UNMANAGE_ENC_NOT_SUPPORTED,
+                exception=e)
+            raise e
+
         # Build required conditions for conditional update
         expected = {
             'attach_status': db.Not(fields.VolumeAttachStatus.ATTACHED),

cinder/volume/manager.py

@@ -732,6 +732,11 @@ class VolumeManager(manager.CleanableManager,
             raise exception.VolumeAttached(volume_id=volume.id)
         self._check_is_our_resource(volume)
 
+        if unmanage_only and volume.encryption_key_id is not None:
+            raise exception.Invalid(
+                reason=_("Unmanaging encrypted volumes is not "
+                         "supported."))
+
         if unmanage_only and cascade:
             # This could be done, but is ruled out for now just
             # for simplicity.