diff --git a/cinder/backup/driver.py b/cinder/backup/driver.py index b085cdbe7d6..b4158e4f5e7 100644 --- a/cinder/backup/driver.py +++ b/cinder/backup/driver.py @@ -17,7 +17,6 @@ import abc -from castellan import key_manager from oslo_config import cfg from oslo_log import log as logging from oslo_serialization import jsonutils @@ -26,7 +25,6 @@ import six from cinder.db import base from cinder import exception from cinder.i18n import _ -from cinder.volume import utils as vol_utils service_opts = [ cfg.IntOpt('backup_metadata_version', default=2, @@ -60,13 +58,6 @@ class BackupMetadataAPI(base.Base): self.context = context self._key_mgr = None - @property - def _key_manager(self): - # Allows for lazy initialization of the key manager - if self._key_mgr is None: - self._key_mgr = key_manager.API(CONF) - return self._key_mgr - @staticmethod def _is_serializable(value): """Returns True if value is serializable.""" @@ -96,12 +87,13 @@ class BackupMetadataAPI(base.Base): LOG.info("Unable to serialize field '%s' - excluding " "from backup", key) continue - # Copy the encryption key UUID for backup - if key is 'encryption_key_id' and value is not None: - value = vol_utils.clone_encryption_key(self.context, - self._key_manager, - value) - LOG.debug("Copying encryption key UUID for backup.") + # NOTE(abishop): The backup manager is now responsible for + # ensuring a copy of the volume's encryption key ID is + # retained in case the volume is deleted. Yes, this means + # the backup's volume base metadata now stores the volume's + # original encryption key ID, which affects how things are + # handled when backups are restored. The backup manager + # handles this, too. container[type_tag][key] = value LOG.debug("Completed fetching metadata type '%s'", type_tag) diff --git a/cinder/backup/manager.py b/cinder/backup/manager.py index 0cb3900b65b..f8c22afce86 100644 --- a/cinder/backup/manager.py +++ b/cinder/backup/manager.py @@ -33,6 +33,7 @@ Volume backups can be created, restored, deleted and listed. import os +from castellan import key_manager from eventlet import tpool from oslo_config import cfg from oslo_log import log as logging @@ -235,15 +236,26 @@ class BackupManager(manager.ThreadPoolManager): backup.status = fields.BackupStatus.AVAILABLE backup.save() elif backup['status'] == fields.BackupStatus.DELETING: - LOG.info('Resuming delete on backup: %s.', backup['id']) - if CONF.backup_service_inithost_offload: - # Offload all the pending backup delete operations to the - # threadpool to prevent the main backup service thread - # from being blocked. - self._add_to_threadpool(self.delete_backup, ctxt, backup) + # Don't resume deleting the backup of an encrypted volume. The + # admin context won't be sufficient to delete the backup's copy + # of the encryption key ID (a real user context is required). + if backup.encryption_key_id is None: + LOG.info('Resuming delete on backup: %s.', backup.id) + if CONF.backup_service_inithost_offload: + # Offload all the pending backup delete operations to the + # threadpool to prevent the main backup service thread + # from being blocked. + self._add_to_threadpool(self.delete_backup, ctxt, backup) + else: + # Delete backups sequentially + self.delete_backup(ctxt, backup) else: - # Delete backups sequentially - self.delete_backup(ctxt, backup) + LOG.info('Unable to resume deleting backup of an encrypted ' + 'volume, resetting backup %s to error_deleting ' + '(was deleting).', + backup.id) + backup.status = fields.BackupStatus.ERROR_DELETING + backup.save() def _detach_all_attachments(self, ctxt, volume): attachments = volume['volume_attachment'] or [] @@ -412,6 +424,15 @@ class BackupManager(manager.ThreadPoolManager): self._notify_about_backup_usage(context, backup, "create.end") def _run_backup(self, context, backup, volume): + # Save a copy of the encryption key ID in case the volume is deleted. + if (volume.encryption_key_id is not None and + backup.encryption_key_id is None): + backup.encryption_key_id = volume_utils.clone_encryption_key( + context, + key_manager.API(CONF), + volume.encryption_key_id) + backup.save() + backup_service = self.get_backup_driver(context) properties = utils.brick_get_connector_properties() @@ -538,6 +559,7 @@ class BackupManager(manager.ThreadPoolManager): self._notify_about_backup_usage(context, backup, "restore.end") def _run_restore(self, context, backup, volume): + orig_key_id = volume.encryption_key_id backup_service = self.get_backup_driver(context) properties = utils.brick_get_connector_properties() @@ -570,6 +592,48 @@ class BackupManager(manager.ThreadPoolManager): self._detach_device(context, attach_info, volume, properties, force=True) + # Regardless of whether the restore was successful, do some + # housekeeping to ensure the restored volume's encryption key ID is + # unique, and any previous key ID is deleted. Start by fetching fresh + # info on the restored volume. + restored_volume = objects.Volume.get_by_id(context, volume.id) + restored_key_id = restored_volume.encryption_key_id + if restored_key_id != orig_key_id: + LOG.info('Updating encryption key ID for volume %(volume_id)s ' + 'from backup %(backup_id)s.', + {'volume_id': volume.id, 'backup_id': backup.id}) + + key_mgr = key_manager.API(CONF) + if orig_key_id is not None: + LOG.debug('Deleting original volume encryption key ID.') + volume_utils.delete_encryption_key(context, + key_mgr, + orig_key_id) + + if backup.encryption_key_id is None: + # This backup predates the current code that stores the cloned + # key ID in the backup database. Fortunately, the key ID + # restored from the backup data _is_ a clone of the original + # volume's key ID, so grab it. + LOG.debug('Gleaning backup encryption key ID from metadata.') + backup.encryption_key_id = restored_key_id + backup.save() + + # Clone the key ID again to ensure every restored volume has + # a unique key ID. The volume's key ID should not be the same + # as the backup.encryption_key_id (the copy made when the backup + # was first created). + new_key_id = volume_utils.clone_encryption_key( + context, + key_mgr, + backup.encryption_key_id) + restored_volume.encryption_key_id = new_key_id + restored_volume.save() + else: + LOG.debug('Encryption key ID for volume %(volume_id)s already ' + 'matches encryption key ID in backup %(backup_id)s.', + {'volume_id': volume.id, 'backup_id': backup.id}) + def delete_backup(self, context, backup): """Delete volume backup from configured backup service.""" LOG.info('Delete backup started, backup: %s.', backup.id) @@ -631,6 +695,13 @@ class BackupManager(manager.ThreadPoolManager): reservations = None LOG.exception("Failed to update usages deleting backup") + if backup.encryption_key_id is not None: + volume_utils.delete_encryption_key(context, + key_manager.API(CONF), + backup.encryption_key_id) + backup.encryption_key_id = None + backup.save() + backup.destroy() # If this backup is incremental backup, handle the # num_dependent_backups of parent backup diff --git a/cinder/db/sqlalchemy/migrate_repo/versions/117_add_encryption_key_id_to_backups.py b/cinder/db/sqlalchemy/migrate_repo/versions/117_add_encryption_key_id_to_backups.py new file mode 100644 index 00000000000..126a3918c4c --- /dev/null +++ b/cinder/db/sqlalchemy/migrate_repo/versions/117_add_encryption_key_id_to_backups.py @@ -0,0 +1,23 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +from sqlalchemy import Column, MetaData, String, Table + + +def upgrade(migrate_engine): + """Add encryption_key_id column to Backups.""" + meta = MetaData() + meta.bind = migrate_engine + backups = Table('backups', meta, autoload=True) + + encryption_key_id = Column('encryption_key_id', String(length=36)) + backups.create_column(encryption_key_id) diff --git a/cinder/db/sqlalchemy/models.py b/cinder/db/sqlalchemy/models.py index 6936d3dff44..35bf7918347 100644 --- a/cinder/db/sqlalchemy/models.py +++ b/cinder/db/sqlalchemy/models.py @@ -775,6 +775,7 @@ class Backup(BASE, CinderBase): snapshot_id = Column(String(36)) data_timestamp = Column(DateTime) restore_volume_id = Column(String(36)) + encryption_key_id = Column(String(36)) @validates('fail_reason') def validate_fail_reason(self, key, fail_reason): diff --git a/cinder/objects/backup.py b/cinder/objects/backup.py index acc68904559..add4b8c49ee 100644 --- a/cinder/objects/backup.py +++ b/cinder/objects/backup.py @@ -39,7 +39,8 @@ class Backup(base.CinderPersistentObject, base.CinderObject, # Version 1.3: Changed 'status' field to use BackupStatusField # Version 1.4: Add restore_volume_id # Version 1.5: Add metadata - VERSION = '1.5' + # Version 1.6: Add encryption_key_id + VERSION = '1.6' OPTIONAL_FIELDS = ('metadata',) @@ -75,6 +76,7 @@ class Backup(base.CinderPersistentObject, base.CinderObject, 'data_timestamp': fields.DateTimeField(nullable=True), 'restore_volume_id': fields.StringField(nullable=True), 'metadata': fields.DictOfStringsField(nullable=True), + 'encryption_key_id': fields.StringField(nullable=True), } obj_extra_fields = ['name', 'is_incremental', 'has_dependent_backups'] diff --git a/cinder/objects/base.py b/cinder/objects/base.py index 8ae5ff4d9b2..428129a1613 100644 --- a/cinder/objects/base.py +++ b/cinder/objects/base.py @@ -142,6 +142,7 @@ OBJ_VERSIONS.add('1.31', {'Volume': '1.7'}) OBJ_VERSIONS.add('1.32', {'RequestSpec': '1.3'}) OBJ_VERSIONS.add('1.33', {'Volume': '1.8'}) OBJ_VERSIONS.add('1.34', {'VolumeAttachment': '1.3'}) +OBJ_VERSIONS.add('1.35', {'Backup': '1.6', 'BackupImport': '1.6'}) class CinderObjectRegistry(base.VersionedObjectRegistry): diff --git a/cinder/tests/unit/backup/drivers/test_backup_driver_base.py b/cinder/tests/unit/backup/drivers/test_backup_driver_base.py index ad55977b48c..cc955b3cd83 100644 --- a/cinder/tests/unit/backup/drivers/test_backup_driver_base.py +++ b/cinder/tests/unit/backup/drivers/test_backup_driver_base.py @@ -286,8 +286,7 @@ class BackupMetadataAPITestCase(test.TestCase): def _create_encrypted_volume_db_entry(self, id, type_id, encrypted): if encrypted: - key_id = self.bak_meta_api._key_manager.create_key( - 'context', algorithm='AES', length=256) + key_id = str(uuid.uuid4()) vol = {'id': id, 'size': 1, 'status': 'available', 'volume_type_id': type_id, 'encryption_key_id': key_id} else: diff --git a/cinder/tests/unit/backup/test_backup.py b/cinder/tests/unit/backup/test_backup.py index 21ac6cdd5ee..f7137df1d05 100644 --- a/cinder/tests/unit/backup/test_backup.py +++ b/cinder/tests/unit/backup/test_backup.py @@ -20,7 +20,7 @@ import os import uuid import mock -from os_brick.initiator.connectors import fake +from os_brick.initiator.connectors import fake as fake_connectors from oslo_config import cfg from oslo_db import exception as db_exc from oslo_utils import importutils @@ -36,6 +36,7 @@ from cinder import objects from cinder.objects import fields from cinder import test from cinder.tests import fake_driver +from cinder.tests.unit import fake_constants as fake from cinder.tests.unit import utils from cinder.volume import rpcapi as volume_rpcapi @@ -82,7 +83,8 @@ class BaseBackupTest(test.TestCase): temp_snapshot_id=None, snapshot_id=None, metadata=None, - parent_id=None): + parent_id=None, + encryption_key_id=None): """Create a backup entry in the DB. Return the entry ID @@ -107,6 +109,7 @@ class BaseBackupTest(test.TestCase): kwargs['temp_volume_id'] = temp_volume_id kwargs['temp_snapshot_id'] = temp_snapshot_id kwargs['metadata'] = metadata or {} + kwargs['encryption_key_id'] = encryption_key_id backup = objects.Backup(context=self.ctxt, **kwargs) backup.create() return backup @@ -116,7 +119,8 @@ class BaseBackupTest(test.TestCase): status='backing-up', previous_status='available', size=1, - host='testhost'): + host='testhost', + encryption_key_id=None): """Create a volume entry in the DB. Return the entry ID @@ -132,6 +136,7 @@ class BaseBackupTest(test.TestCase): vol['attach_status'] = fields.VolumeAttachStatus.DETACHED vol['availability_zone'] = '1' vol['previous_status'] = previous_status + vol['encryption_key_id'] = encryption_key_id volume = objects.Volume(context=self.ctxt, **vol) volume.create() return volume.id @@ -430,7 +435,7 @@ class BackupTestCase(BaseBackupTest): self.assertEqual('error_restoring', volume.status) def test_cleanup_one_deleting_backup(self): - """Test cleanup_one_backup for volume status 'deleting'.""" + """Test cleanup_one_backup for backup status 'deleting'.""" self.override_config('backup_service_inithost_offload', False) backup = self._create_backup_db_entry( @@ -443,6 +448,21 @@ class BackupTestCase(BaseBackupTest): self.ctxt, backup.id) + def test_cleanup_one_deleting_encrypted_backup(self): + """Test cleanup of backup status 'deleting' (encrypted).""" + self.override_config('backup_service_inithost_offload', False) + + backup = self._create_backup_db_entry( + status=fields.BackupStatus.DELETING, + encryption_key_id=fake.ENCRYPTION_KEY_ID) + + self.backup_mgr._cleanup_one_backup(self.ctxt, backup) + + backup = db.backup_get(self.ctxt, backup.id) + self.assertIsNotNone(backup) + self.assertEqual(fields.BackupStatus.ERROR_DELETING, + backup.status) + def test_detach_all_attachments_handles_exceptions(self): """Test detach_all_attachments with exceptions.""" @@ -636,6 +656,7 @@ class BackupTestCase(BaseBackupTest): backup = db.backup_get(self.ctxt, backup.id) self.assertEqual(fields.BackupStatus.AVAILABLE, backup['status']) self.assertEqual(vol_size, backup['size']) + self.assertIsNone(backup.encryption_key_id) @mock.patch('cinder.utils.brick_get_connector_properties') @mock.patch('cinder.volume.rpcapi.VolumeAPI.get_backup_device') @@ -831,7 +852,7 @@ class BackupTestCase(BaseBackupTest): attach_info = { 'device': {'path': '/dev/null'}, 'conn': {'data': {}}, - 'connector': fake.FakeConnector(None)} + 'connector': fake_connectors.FakeConnector(None)} mock_terminate_connection_snapshot = self.mock_object( volume_rpcapi.VolumeAPI, 'terminate_connection_snapshot') @@ -926,6 +947,58 @@ class BackupTestCase(BaseBackupTest): self.backup_mgr.create_backup(self.ctxt, backup) self.assertEqual(2, notify.call_count) + @mock.patch('cinder.volume.rpcapi.VolumeAPI.get_backup_device') + @mock.patch('cinder.volume.utils.clone_encryption_key') + @mock.patch('cinder.utils.brick_get_connector_properties') + def test_create_backup_encrypted_volume(self, + mock_connector_properties, + mock_clone_encryption_key, + mock_get_backup_device): + """Test backup of encrypted volume. + + Test whether the volume's encryption key ID is cloned and + saved in the backup. + """ + vol_id = self._create_volume_db_entry(encryption_key_id=fake.UUID1) + backup = self._create_backup_db_entry(volume_id=vol_id) + + self.mock_object(self.backup_mgr, '_detach_device') + mock_attach_device = self.mock_object(self.backup_mgr, + '_attach_device') + mock_attach_device.return_value = {'device': {'path': '/dev/null'}} + mock_clone_encryption_key.return_value = fake.UUID2 + + self.backup_mgr.create_backup(self.ctxt, backup) + mock_clone_encryption_key.assert_called_once_with(self.ctxt, + mock.ANY, + fake.UUID1) + backup = db.backup_get(self.ctxt, backup.id) + self.assertEqual(fake.UUID2, backup.encryption_key_id) + + @mock.patch('cinder.volume.rpcapi.VolumeAPI.get_backup_device') + @mock.patch('cinder.volume.utils.clone_encryption_key') + @mock.patch('cinder.utils.brick_get_connector_properties') + def test_create_backup_encrypted_volume_again(self, + mock_connector_properties, + mock_clone_encryption_key, + mock_get_backup_device): + """Test backup of encrypted volume. + + Test when the backup already has a clone of the volume's encryption + key ID. + """ + vol_id = self._create_volume_db_entry(encryption_key_id=fake.UUID1) + backup = self._create_backup_db_entry(volume_id=vol_id, + encryption_key_id=fake.UUID2) + + self.mock_object(self.backup_mgr, '_detach_device') + mock_attach_device = self.mock_object(self.backup_mgr, + '_attach_device') + mock_attach_device.return_value = {'device': {'path': '/dev/null'}} + + self.backup_mgr.create_backup(self.ctxt, backup) + mock_clone_encryption_key.assert_not_called() + def test_restore_backup_with_bad_volume_status(self): """Test error handling. @@ -1064,6 +1137,159 @@ class BackupTestCase(BaseBackupTest): self.backup_mgr.restore_backup(self.ctxt, backup, vol_id) self.assertEqual(2, notify.call_count) + @mock.patch('cinder.volume.utils.clone_encryption_key') + @mock.patch('cinder.volume.utils.delete_encryption_key') + @mock.patch( + 'cinder.tests.unit.backup.fake_service.FakeBackupService.restore') + @mock.patch('cinder.utils.brick_get_connector_properties') + def test_restore_backup_encrypted_volume(self, + mock_connector_properties, + mock_backup_driver_restore, + mock_delete_encryption_key, + mock_clone_encryption_key): + """Test restore of encrypted volume. + + Test restoring a volume from its own backup. In this situation, + the volume's encryption key ID shouldn't change. + """ + vol_id = self._create_volume_db_entry(status='restoring-backup', + encryption_key_id=fake.UUID1) + backup = self._create_backup_db_entry( + volume_id=vol_id, + status=fields.BackupStatus.RESTORING, + encryption_key_id=fake.UUID2) + + self.mock_object(self.backup_mgr, '_detach_device') + mock_attach_device = self.mock_object(self.backup_mgr, + '_attach_device') + mock_attach_device.return_value = {'device': {'path': '/dev/null'}} + + self.backup_mgr.restore_backup(self.ctxt, backup, vol_id) + volume = db.volume_get(self.ctxt, vol_id) + self.assertEqual(fake.UUID1, volume.encryption_key_id) + mock_clone_encryption_key.assert_not_called() + mock_delete_encryption_key.assert_not_called() + + @mock.patch('cinder.volume.utils.clone_encryption_key') + @mock.patch('cinder.volume.utils.delete_encryption_key') + @mock.patch( + 'cinder.tests.unit.backup.fake_service.FakeBackupService.restore') + @mock.patch('cinder.utils.brick_get_connector_properties') + def test_restore_backup_new_encrypted_volume(self, + mock_connector_properties, + mock_backup_driver_restore, + mock_delete_encryption_key, + mock_clone_encryption_key): + """Test restore of encrypted volume. + + Test handling of encryption key IDs when retoring to another + encrypted volume, i.e. a volume whose key ID is different from + the volume originally backed up. + - The volume's prior encryption key ID is deleted. + - The volume is assigned a fresh clone of the backup's encryption + key ID. + """ + vol_id = self._create_volume_db_entry(status='restoring-backup', + encryption_key_id=fake.UUID1) + backup = self._create_backup_db_entry( + volume_id=vol_id, + status=fields.BackupStatus.RESTORING, + encryption_key_id=fake.UUID2) + + self.mock_object(self.backup_mgr, '_detach_device') + mock_attach_device = self.mock_object(self.backup_mgr, + '_attach_device') + mock_attach_device.return_value = {'device': {'path': '/dev/null'}} + mock_clone_encryption_key.return_value = fake.UUID3 + + # Mimic the driver's side effect where it updates the volume's + # metadata. For backups of encrypted volumes, this will essentially + # overwrite the volume's encryption key ID prior to the restore. + def restore_side_effect(backup, volume_id, volume_file): + db.volume_update(self.ctxt, + volume_id, + {'encryption_key_id': fake.UUID4}) + mock_backup_driver_restore.side_effect = restore_side_effect + + self.backup_mgr.restore_backup(self.ctxt, backup, vol_id) + + # Volume's original encryption key ID should be deleted + mock_delete_encryption_key.assert_called_once_with(self.ctxt, + mock.ANY, + fake.UUID1) + + # Backup's encryption key ID should have been cloned + mock_clone_encryption_key.assert_called_once_with(self.ctxt, + mock.ANY, + fake.UUID2) + + # Volume should have the cloned backup key ID + volume = db.volume_get(self.ctxt, vol_id) + self.assertEqual(fake.UUID3, volume.encryption_key_id) + + # Backup's key ID should not have changed + backup = db.backup_get(self.ctxt, backup.id) + self.assertEqual(fake.UUID2, backup.encryption_key_id) + + @mock.patch('cinder.volume.utils.clone_encryption_key') + @mock.patch('cinder.volume.utils.delete_encryption_key') + @mock.patch( + 'cinder.tests.unit.backup.fake_service.FakeBackupService.restore') + @mock.patch('cinder.utils.brick_get_connector_properties') + def test_restore_backup_glean_key_id(self, + mock_connector_properties, + mock_backup_driver_restore, + mock_delete_encryption_key, + mock_clone_encryption_key): + """Test restore of encrypted volume. + + Test restoring a backup that was created prior to when the encryption + key ID is saved in the backup DB. The backup encryption key ID is + gleaned from the restored volume. + """ + vol_id = self._create_volume_db_entry(status='restoring-backup', + encryption_key_id=fake.UUID1) + backup = self._create_backup_db_entry( + volume_id=vol_id, + status=fields.BackupStatus.RESTORING) + + self.mock_object(self.backup_mgr, '_detach_device') + mock_attach_device = self.mock_object(self.backup_mgr, + '_attach_device') + mock_attach_device.return_value = {'device': {'path': '/dev/null'}} + mock_clone_encryption_key.return_value = fake.UUID3 + + # Mimic the driver's side effect where it updates the volume's + # metadata. For backups of encrypted volumes, this will essentially + # overwrite the volume's encryption key ID prior to the restore. + def restore_side_effect(backup, volume_id, volume_file): + db.volume_update(self.ctxt, + volume_id, + {'encryption_key_id': fake.UUID4}) + mock_backup_driver_restore.side_effect = restore_side_effect + + self.backup_mgr.restore_backup(self.ctxt, backup, vol_id) + + # Volume's original encryption key ID should be deleted + mock_delete_encryption_key.assert_called_once_with(self.ctxt, + mock.ANY, + fake.UUID1) + + # Backup's encryption key ID should have been cloned from + # the value restored from the metadata. + mock_clone_encryption_key.assert_called_once_with(self.ctxt, + mock.ANY, + fake.UUID4) + + # Volume should have the cloned backup key ID + volume = db.volume_get(self.ctxt, vol_id) + self.assertEqual(fake.UUID3, volume.encryption_key_id) + + # Backup's key ID should have been gleaned from value restored + # from the backup's metadata + backup = db.backup_get(self.ctxt, backup.id) + self.assertEqual(fake.UUID4, backup.encryption_key_id) + def test_delete_backup_with_bad_backup_status(self): """Test error handling. @@ -1144,6 +1370,25 @@ class BackupTestCase(BaseBackupTest): self.assertGreaterEqual(timeutils.utcnow(), backup.deleted_at) self.assertEqual(fields.BackupStatus.DELETED, backup.status) + @mock.patch('cinder.volume.utils.delete_encryption_key') + def test_delete_backup_of_encrypted_volume(self, + mock_delete_encryption_key): + """Test deletion of backup of encrypted volume""" + vol_id = self._create_volume_db_entry( + encryption_key_id=fake.UUID1) + backup = self._create_backup_db_entry( + volume_id=vol_id, + status=fields.BackupStatus.DELETING, + encryption_key_id=fake.UUID2) + self.backup_mgr.delete_backup(self.ctxt, backup) + mock_delete_encryption_key.assert_called_once_with(self.ctxt, + mock.ANY, + fake.UUID2) + ctxt_read_deleted = context.get_admin_context('yes') + backup = db.backup_get(ctxt_read_deleted, backup.id) + self.assertTrue(backup.deleted) + self.assertIsNone(backup.encryption_key_id) + @mock.patch('cinder.volume.utils.notify_about_backup_usage') def test_delete_backup_with_notify(self, notify): """Test normal backup deletion with notifications.""" diff --git a/cinder/tests/unit/objects/test_objects.py b/cinder/tests/unit/objects/test_objects.py index 7935e4cbad5..4c915e2ecbd 100644 --- a/cinder/tests/unit/objects/test_objects.py +++ b/cinder/tests/unit/objects/test_objects.py @@ -23,9 +23,9 @@ from cinder import test # NOTE: The hashes in this list should only be changed if they come with a # corresponding version bump in the affected objects. object_data = { - 'Backup': '1.5-3ab4b305bd43ec0cff6701fe2a849194', + 'Backup': '1.6-c7ede487ba6fbcdd2a4711343cd972be', 'BackupDeviceInfo': '1.0-74b3950676c690538f4bc6796bd0042e', - 'BackupImport': '1.5-3ab4b305bd43ec0cff6701fe2a849194', + 'BackupImport': '1.6-c7ede487ba6fbcdd2a4711343cd972be', 'BackupList': '1.0-15ecf022a68ddbb8c2a6739cfc9f8f5e', 'CleanupRequest': '1.0-e7c688b893e1d5537ccf65cc3eb10a28', 'Cluster': '1.1-e2c533eb8cdd8d229b6c45c6cf3a9e2c', diff --git a/cinder/tests/unit/test_db_api.py b/cinder/tests/unit/test_db_api.py index 97242da8849..9dea45baf43 100644 --- a/cinder/tests/unit/test_db_api.py +++ b/cinder/tests/unit/test_db_api.py @@ -2735,6 +2735,7 @@ class DBAPIBackupTestCase(BaseTest): 'temp_snapshot_id': 'temp_snapshot_id', 'num_dependent_backups': 0, 'snapshot_id': 'snapshot_id', + 'encryption_key_id': 'encryption_key_id', 'restore_volume_id': 'restore_volume_id'} if one: return base_values diff --git a/releasenotes/notes/fix-backup-handling-of-encryption-key-id-f2fa56cadd80d582.yaml b/releasenotes/notes/fix-backup-handling-of-encryption-key-id-f2fa56cadd80d582.yaml new file mode 100644 index 00000000000..91390fc236e --- /dev/null +++ b/releasenotes/notes/fix-backup-handling-of-encryption-key-id-f2fa56cadd80d582.yaml @@ -0,0 +1,13 @@ +--- +fixes: + - | + Fix the way encryption key IDs are managed for encrypted volume backups. + When creating a backup, the volume's encryption key is cloned and assigned + a new key ID. The backup's cloned key ID is now stored in the backup + database so that it can be deleted whenever the backup is deleted. + + When restoring the backup of an encrypted volume, the destination volume + is assigned a clone of the backup's encryption key ID. This ensures every + restored backup has a unique encryption key ID, even when multiple volumes + have been restored from the same backup. +