Merge "TPM: fixups for live migration of host secret security"

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

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'])
 

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,

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)
 

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':