From bfe5b7fd1425ce385a89c4afc0b51ffdb3b4484c Mon Sep 17 00:00:00 2001 From: melanie witt Date: Tue, 10 Feb 2026 14:18:29 -0800 Subject: [PATCH] TPM: fixups for live migration of `host` secret security Addressing review comments from the previous patch: https://review.opendev.org/c/openstack/nova/+/941483 Change-Id: Iad53e7bd9ef5c50c491016e98e257fafc1424272 Signed-off-by: melanie witt --- nova/objects/migrate_data.py | 35 +++++++++++++++++++++ nova/tests/functional/libvirt/test_vtpm.py | 14 +++++++++ nova/tests/unit/virt/libvirt/test_driver.py | 2 +- nova/virt/libvirt/driver.py | 8 ++--- nova/virt/libvirt/host.py | 2 +- 5 files changed, 53 insertions(+), 8 deletions(-) diff --git a/nova/objects/migrate_data.py b/nova/objects/migrate_data.py index 161f6352a59a..7e60f800178c 100644 --- a/nova/objects/migrate_data.py +++ b/nova/objects/migrate_data.py @@ -375,6 +375,41 @@ class LibvirtLiveMigrateData(LiveMigrateData): return ( self.has_vtpm and self.vtpm_secret_uuid and self.vtpm_secret_value) + @property + def vtpm_secret_value_bytes(self): + """Get the vTPM secret value as bytes after transport over RPC. + + A vTPM secret is a Barbican secret of type "passphrase", which are + used for storing plain text secrets. A Barbican passphrase is an + unencrypted bytestring of data type: bytes. + + The secret value is generated in nova/crypto.py as a random bytestring + that is subsequently base64 encoded using the standard Base64 alphabet. + It is then stored in Barbican as a passphrase. + + The caller expects to receive bytes from here so we can convert the + value to the original data type: bytes with 'ascii' encoding. + """ + return self.vtpm_secret_value.encode(encoding='ascii') + + @vtpm_secret_value_bytes.setter + def vtpm_secret_value_bytes(self, value): + """Store the vTPM secret value as str for transport over RPC. + + A vTPM secret is a Barbican secret of type "passphrase", which are + used for storing plain text secrets. A Barbican passphrase is an + unencrypted bytestring of data type: bytes. + + The secret value is generated in nova/crypto.py as a random bytestring + that is subsequently base64 encoded using the standard Base64 alphabet. + It is then stored in Barbican as a passphrase. + + We expect to receive bytes here and we can convert the value to a str + with 'ascii' encoding because we know it was base64 encoded using the + standard Base64 alphabet. + """ + self.vtpm_secret_value = value.decode(encoding='ascii') + # TODO(gmann): HyperV virt driver has been removed in Nova 29.0.0 (OpenStack # 2024.1) release but we kept this object for a couple of cycle. This can be diff --git a/nova/tests/functional/libvirt/test_vtpm.py b/nova/tests/functional/libvirt/test_vtpm.py index 41bad05925e0..a9ebaebba619 100644 --- a/nova/tests/functional/libvirt/test_vtpm.py +++ b/nova/tests/functional/libvirt/test_vtpm.py @@ -563,6 +563,9 @@ class VTPMServersTest(base.LibvirtMigrationMixin, base.ServersTestBase): # Try to recover the instance by hard-rebooting it. self._reboot_server(self.server, hard=True) + # The libvirt secret should have been re-created. + self._assert_libvirt_has_secret(self.src, self.server['id']) + # This time the live migration should work because the libvirt secret # should have been re-created by the hard reboot. self._live_migrate(self.server, migration_expected_state='completed', @@ -616,8 +619,19 @@ class VTPMServersTest(base.LibvirtMigrationMixin, base.ServersTestBase): # After the live migration fails, we should still have a secret in the # key manager service. self.assertInstanceHasSecret(self.server) + + # The instance should be on the source host. + instances = self.src.driver._host.list_instance_domains() + self.assertEqual(1, len(instances)) + self.assertEqual(self.server['id'], instances[0].UUIDString()) + # We should have a libvirt secret on the source host. self._assert_libvirt_has_secret(self.src, self.server['id']) + + # There should be no instance on the destination host. + instances = self.dest.driver._host.list_instance_domains() + self.assertEqual(0, len(instances)) + # And no libvirt secret on the destination host. self._assert_libvirt_secret_missing(self.dest, self.server['id']) diff --git a/nova/tests/unit/virt/libvirt/test_driver.py b/nova/tests/unit/virt/libvirt/test_driver.py index f2847b52030c..0107d29f7b2c 100644 --- a/nova/tests/unit/virt/libvirt/test_driver.py +++ b/nova/tests/unit/virt/libvirt/test_driver.py @@ -13516,7 +13516,7 @@ class LibvirtConnTestCase(test.NoDBTestCase, } dest_check_data = objects.LibvirtLiveMigrateData(filename='file') mock_find.return_value.UUIDString.return_value = uuids.secret - mock_find.return_value.value.return_value.decode.return_value = 'foo' + mock_find.return_value.value.return_value = b'foo' drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False) drvr.check_can_live_migrate_source(self.context, instance, diff --git a/nova/virt/libvirt/driver.py b/nova/virt/libvirt/driver.py index b6adc9be93d8..cd7b10726aeb 100644 --- a/nova/virt/libvirt/driver.py +++ b/nova/virt/libvirt/driver.py @@ -10954,9 +10954,7 @@ class LibvirtDriver(driver.ComputeDriver): raise exception.VTPMSecretNotFound(msg) dest_check_data.vtpm_secret_uuid = secret.UUIDString() - # Have to decode the bytes type to conform to the object's - # SensitiveStringField type. - dest_check_data.vtpm_secret_value = secret.value().decode() + dest_check_data.vtpm_secret_value_bytes = secret.value() else: # If the instance has a vTPM, set the relevant fields to None in # order to convey that we are actively choosing not to pass any @@ -12065,9 +12063,7 @@ class LibvirtDriver(driver.ComputeDriver): if migrate_data.has_vtpm_secret_data: self._host.create_secret( 'vtpm', instance.uuid, - # Convert the SensitiveStringField back to bytes when creating - # the libvirt secret. - password=migrate_data.vtpm_secret_value.encode(), + password=migrate_data.vtpm_secret_value_bytes, uuid=migrate_data.vtpm_secret_uuid, ephemeral=False, private=False) diff --git a/nova/virt/libvirt/host.py b/nova/virt/libvirt/host.py index ea8ec0725ed9..89ada9319817 100644 --- a/nova/virt/libvirt/host.py +++ b/nova/virt/libvirt/host.py @@ -1144,7 +1144,7 @@ class Host(object): def find_secret(self, usage_type, usage_id): """Find a secret. - usage_type: one of 'iscsi', 'ceph', 'rbd' or 'volume' + usage_type: one of 'iscsi', 'ceph', 'rbd', 'volume' or 'vtpm' usage_id: name of resource in secret """ if usage_type == 'iscsi':