Check swtpm version and start swtpm before qemu

gns3server/compute/qemu/__init__.py

@@ -223,7 +223,7 @@ class Qemu(BaseManager):
                     version = match.group(1)
                     return version
                 else:
-                    raise QemuError("Could not determine the Qemu version for {}".format(qemu_path))
+                    raise QemuError("Could not determine the Qemu version for '{}'".format(qemu_path))
             except (OSError, subprocess.SubprocessError) as e:
                 raise QemuError("Error while looking for the Qemu version: {}".format(e))
 
@@ -242,10 +242,29 @@ class Qemu(BaseManager):
                 version = match.group(1)
                 return version
             else:
-                raise QemuError("Could not determine the Qemu-img version for {}".format(qemu_img_path))
+                raise QemuError("Could not determine the Qemu-img version for '{}'".format(qemu_img_path))
         except (OSError, subprocess.SubprocessError) as e:
             raise QemuError("Error while looking for the Qemu-img version: {}".format(e))
 
+    @staticmethod
+    async def get_swtpm_version(swtpm_path):
+        """
+        Gets the swtpm version.
+
+        :param swtpm_path: path to swtpm executable.
+        """
+
+        try:
+            output = await subprocess_check_output(swtpm_path, "--version")
+            match = re.search(r"version\s+([\d.]+)", output)
+            if match:
+                version = match.group(1)
+                return version
+            else:
+                raise QemuError("Could not determine the swtpm version for '{}'".format(swtpm_path))
+        except (OSError, subprocess.SubprocessError) as e:
+            raise QemuError("Error while looking for the swtpm version: {}".format(e))
+
     @staticmethod
     def get_haxm_windows_version():
         """

gns3server/compute/qemu/qemu_vm.py

@@ -1079,6 +1079,10 @@ class QemuVM(BaseNode):
             # check if there is enough RAM to run
             self.check_available_ram(self.ram)
 
+            # start swtpm (TPM emulator) first if TPM is enabled
+            if self._tpm:
+                await self._start_swtpm()
+
             command = await self._build_command()
             command_string = " ".join(shlex_quote(s) for s in command)
             try:
@@ -1104,8 +1108,6 @@ class QemuVM(BaseNode):
             await self._set_process_priority()
             if self._cpu_throttling:
                 self._set_cpu_throttling()
-            if self._tpm:
-                self._start_swtpm()
             if "-enable-kvm" in command_string or "-enable-hax" in command_string:
                 self._hw_virtualization = True
 
@@ -2019,10 +2021,9 @@ class QemuVM(BaseNode):
             options.extend(["-kernel", self._kernel_image.replace(",", ",,")])
         if self._kernel_command_line:
             options.extend(["-append", self._kernel_command_line])
-
         return options
 
-    def _start_swtpm(self):
+    async def _start_swtpm(self):
         """
         Start swtpm (TPM emulator)
         """
@@ -2035,6 +2036,10 @@ class QemuVM(BaseNode):
         swtpm = shutil.which("swtpm")
         if not swtpm:
             raise QemuError("Could not find swtpm (TPM emulator)")
+        swtpm_version = await self.manager.get_swtpm_version(swtpm)
+        if swtpm_version and parse_version(swtpm_version) < parse_version("0.8.0"):
+            # swtpm >= version 0.8.0 is required
+            raise QemuError("swtpm version 0.8.0 or above must be installed (detected version is {})".format(swtpm_version))
         try:
             command = [
                 swtpm,
@@ -2066,6 +2071,8 @@ class QemuVM(BaseNode):
         """
 
         tpm_sock = os.path.join(self.temporary_directory, "swtpm.sock")
+        if not os.path.exists(tpm_sock):
+            raise QemuError("swtpm socket file '{}' does not exist".format(tpm_sock))
         options = [
             "-chardev",
             "socket,id=chrtpm,path={}".format(tpm_sock),

tests/compute/qemu/test_qemu_vm.py

@@ -409,7 +409,8 @@ async def test_tpm_option(vm, tmpdir, fake_qemu_img_binary):
     vm.manager.get_qemu_version = AsyncioMagicMock(return_value="3.1.0")
     vm._tpm = True
     tpm_sock = os.path.join(vm.temporary_directory, "swtpm.sock")
-    options = await vm._build_command()
+    with patch("os.path.exists", return_value=True) as os_path:
+        options = await vm._build_command()
     assert '-chardev socket,id=chrtpm,path={}'.format(tpm_sock) in ' '.join(options)
     assert '-tpmdev emulator,id=tpm0,chardev=chrtpm' in ' '.join(options)
     assert '-device tpm-tis,tpmdev=tpm0' in ' '.join(options)