feat(instances): a shut-off VM with a TPM migrates with its TPM state

A shut-off VM with TPM state, as every Windows 11 VM, was refused for
migration: the state is a file of the source host, and an offline move
would start it there with a new TPM. Such a VM now migrates as a shut-off
VM with disks to copy does: its UEFI variables are copied, it is started
paused, its CPUs never run, migrates live, which carries the TPM state,
and is shut off on the destination. A shut-off VM with a saved state is
refused before any of that, for every way it moves: started paused, it
would resume and then lose the saved memory.

The clone tab says that a clone gets a new, empty TPM, and the destroy
page that the TPM state goes with the VM.

instances/tests/test_destroy.py

@@ -175,6 +175,14 @@ class DestroyOrderTestCase(TestCase):
         self.assertEqual(response.status_code, 200)
         self.assertEqual(calls, [])
 
+    def test_the_confirmation_page_says_the_tpm_state_goes(self):
+        for tpm, shown in (("emulator", True), (None, False), ("custom", False)):
+            with self.subTest(tpm=tpm), patch("instances.models.wvmInstance") as wvm:
+                wvm.return_value.get_status.return_value = 5
+                wvm.return_value.get_tpm.return_value = tpm
+                response = self.client.get(reverse("instances:destroy", args=[self.instance.id]))
+                self.assertEqual(b"Its TPM state is deleted" in response.content, shown)
+
     def test_the_confirmation_page_of_a_paused_vm_offers_the_form(self):
         with patch("instances.models.wvmInstance") as wvm:
             wvm.return_value.get_status.return_value = 3  # paused

instances/tests/test_live_jobs.py

@@ -6,6 +6,7 @@ migration address. The ISO upload runs on both hosts: TCP writes through a
 libvirt stream, SSH through SFTP."""
 
 import os
+import subprocess
 import time
 import unittest
 
@@ -258,6 +259,32 @@ class LiveJobsTestCase(TransactionTestCase):
             self.assertEqual(nvram(target), content)
             self.assertIsNone(nvram(source))
 
+    def test_a_shut_off_vm_with_tpm_state_migrates_with_it_there_and_back(self):
+        dest, dest_conn = self.second_host()
+        tpm = "<tpm model='tpm-crb'><backend type='emulator' version='2.0'/></tpm>"
+        # UEFI with a TPM, as a Windows 11 VM: its NVRAM file is copied, its TPM state migrated
+        dom = livetest.define_vm(self.conn, P + "tpm", [], uefi=True, extra_devices=tpm)
+        dom.createWithFlags(libvirt.VIR_DOMAIN_START_PAUSED)  # swtpm makes the state
+        dom.destroy()
+        uuid = dom.UUIDString()
+
+        def state_on_dest():
+            """The checksum of the TPM state file on the destination (an SSH host, as root)."""
+            path = f"/var/lib/libvirt/swtpm/{uuid}/tpm2/tpm2-00.permall"
+            return subprocess.run(["ssh", "-o", "BatchMode=yes", "-l", dest.login, dest.hostname,
+                                   f"sha256sum {path} 2>/dev/null"], capture_output=True, text=True).stdout.split()[:1]
+
+        instance = Instance.objects.create(compute=self.compute, name=dom.name(), uuid=uuid)
+        sums = []
+        for n, target in enumerate((dest, self.compute, dest)):
+            offline = {"instance": instance.pk, "target": target.pk, "offline": True}
+            self.run_job(jobs.submit("instance.migrate", self.admin, offline, f"key-tpm{n:05d}"))
+            sums.append(state_on_dest())
+        self.assertFalse(dest_conn.lookupByUUIDString(uuid).isActive())  # shut off again after the move
+        # the same state went there, back and there again; none stays behind
+        self.assertTrue(sums[0])
+        self.assertEqual((sums[1], sums[2]), ([], sums[0]))
+
     def upload(self, compute, key, chunks):
         """Post chunks of a two-chunk ISO upload to the storage view; returns the job."""
         self.client.force_login(self.admin)

instances/tests/test_pending_edits.py

@@ -235,6 +235,11 @@ class PendingEditsTestCase(TestCase):
         doc = self.page(get_tpm=("emulator", "emulator"), tpm_model=None)
         self.assertEqual(doc.xpath(f"{tpm}/@disabled"), [])
 
+    def test_the_clone_tab_says_a_clone_gets_a_new_tpm(self):
+        self.assertIn("new, empty TPM", self.page(get_tpm=("emulator", "emulator")).xpath("string(//p[@id='clone-tpm'])"))
+        for tpm in (None, "custom"):  # a passthrough TPM or one with its own state path comes along
+            self.assertEqual(self.page(get_tpm=(tpm, tpm)).xpath("//p[@id='clone-tpm']"), [])
+
     def test_the_users_tab_shows_each_owner_and_offers_only_new_owners(self):
         from accounts.models import UserInstance