Prevent leaking encryption key when deleting volume

Fail requests to delete an encrypted volume when the associated
encryption key cannot be deleted. This avoids leaking an encryption
key when the volume is deleted but the key cannot be deleted.

This patch addresses situations where Barbican is the key manager,
and Barbican's security policy does not allow the user to delete
the key. Barbican has stricter rules on deleting secrets than Cinder
does on deleting volumes. For example, given two non-admin users X
and Y, Cinder allows user X to delete user Y's volume, but Barbican
does not allow user X to delete user Y's secrets.

If the encryption key cannot be deleted then the volume's status is
set to "error_deleting". To delete the volume, the volume's status
can be reset, and a user with suitable privileges (i.e. an admin or
the volume's owner) can reissue the delete request.

Closes-Bug: #1731528
Change-Id: If0462c18345a4bfbfbcecb6b8caa33d38b86d7f0

cinder/tests/unit/volume/test_volume.py

@@ -744,6 +744,42 @@ def test_create_delete_volume_with_encrypted_volume_type(self):
                           self.context,
                           volume['id'])
 
+    @mock.patch.object(key_manager, 'API', fake_keymgr.fake_api)
+    def test_delete_encrypted_volume_fail_deleting_key(self):
+        cipher = 'aes-xts-plain64'
+        key_size = 256
+        db.volume_type_create(self.context,
+                              {'id': fake.VOLUME_TYPE_ID, 'name': 'LUKS'})
+        db.volume_type_encryption_create(
+            self.context, fake.VOLUME_TYPE_ID,
+            {'control_location': 'front-end', 'provider': ENCRYPTION_PROVIDER,
+             'cipher': cipher, 'key_size': key_size})
+
+        db_vol_type = db.volume_type_get_by_name(self.context, 'LUKS')
+
+        volume = self.volume_api.create(self.context,
+                                        1,
+                                        'name',
+                                        'description',
+                                        volume_type=db_vol_type)
+
+        volume_id = volume['id']
+        volume['host'] = 'fake_host'
+        volume['status'] = 'available'
+        db.volume_update(self.context, volume_id, {'status': 'available'})
+
+        with mock.patch.object(
+                self.volume_api.key_manager,
+                'delete',
+                side_effect=Exception):
+            self.assertRaises(exception.InvalidVolume,
+                              self.volume_api.delete,
+                              self.context,
+                              volume)
+        volume = objects.Volume.get_by_id(self.context, volume_id)
+        self.assertEqual("error_deleting", volume.status)
+        volume.destroy()
+
     def test_delete_busy_volume(self):
         """Test volume survives deletion if driver reports it as busy."""
         volume = tests_utils.create_volume(self.context, **self.volume_params)

cinder/volume/api.py

@@ -480,12 +480,16 @@ def delete(self, context, volume,
         if encryption_key_id is not None:
             try:
                 self.key_manager.delete(context, encryption_key_id)
-            except exception.CinderException as e:
-                LOG.warning("Unable to delete encryption key for "
-                            "volume: %s.", e.msg, resource=volume)
-            except Exception:
-                LOG.exception("Unable to delete encryption key for "
-                              "volume.")
+            except Exception as e:
+                volume.update({'status': 'error_deleting'})
+                volume.save()
+                if hasattr(e, 'msg'):
+                    msg = _("Unable to delete encryption key for "
+                            "volume: %s") % (e.msg)
+                else:
+                    msg = _("Unable to delete encryption key for volume.")
+                LOG.error(msg)
+                raise exception.InvalidVolume(reason=msg)
 
         self.volume_rpcapi.delete_volume(context,
                                          volume,