Fix: Path to sevctl was not from settings

Author: hoh

src/aleph/vm/conf.py

@@ -388,8 +388,7 @@ def check(self):
             assert (
                 check_system_module("kvm_amd/parameters/sev_es") == "Y"
             ), "SEV-ES feature isn't enabled, enable it in BIOS"
-            assert is_command_available("sevctl"), "Command `sevctl` not found, run `cargo install sevctl`"
-
+            assert self.SEV_CTL_PATH.is_file(), f"File not found {self.SEV_CTL_PATH}"
             assert self.ENABLE_QEMU_SUPPORT, "Qemu Support is needed for confidential computing and it's disabled, "
             "enable it setting the env variable `ENABLE_QEMU_SUPPORT=True` in configuration"
 

src/aleph/vm/orchestrator/supervisor.py

@@ -163,7 +163,7 @@ def run():
 
     # Store sevctl app singleton only if confidential feature is enabled
     if settings.ENABLE_CONFIDENTIAL_COMPUTING:
-        sev_client = SevClient(settings.CONFIDENTIAL_DIRECTORY)
+        sev_client = SevClient(settings.CONFIDENTIAL_DIRECTORY, settings.SEV_CTL_PATH)
         app["sev_client"] = sev_client
         # TODO: Review and check sevctl first initialization steps, like (sevctl generate and sevctl provision)
 

src/aleph/vm/sevclient.py

@@ -4,20 +4,27 @@
 
 
 class SevClient:
-    def __init__(self, sev_dir: Path):
+    sev_dir: Path
+    sev_ctl_executable: Path
+    certificates_dir: Path
+    certificates_archive: Path
+
+    def __init__(self, sev_dir: Path, sev_ctl_executable: Path):
         self.sev_dir = sev_dir
+        self.sev_ctl_executable = sev_ctl_executable
         self.certificates_dir = sev_dir / "platform"
         self.certificates_dir.mkdir(exist_ok=True, parents=True)
         self.certificates_archive = self.certificates_dir / "certs_export.cert"
 
-    async def sevctl_cmd(self, *args) -> bytes:
+    async def sev_ctl_cmd(self, *args) -> bytes:
+        """Run a command of the 'sevctl' tool."""
         return await run_in_subprocess(
-            ["sevctl", *args],
+            [self.sev_ctl_executable, *args],
             check=True,
         )
 
     async def get_certificates(self) -> Path:
         if not self.certificates_archive.is_file():
-            _ = await self.sevctl_cmd("export", str(self.certificates_archive))
+            _ = await self.sev_ctl_cmd("export", str(self.certificates_archive))
 
         return self.certificates_archive

tests/supervisor/test_views.py

@@ -135,7 +135,7 @@ async def test_about_certificates_missing_setting(aiohttp_client):
     settings.ENABLE_CONFIDENTIAL_COMPUTING = False
 
     app = setup_webapp()
-    app["sev_client"] = SevClient(Path().resolve())
+    app["sev_client"] = SevClient(Path().resolve(), Path("/opt/sevctl").resolve())
     client = await aiohttp_client(app)
     response: web.Response = await client.get("/about/certificates")
     assert response.status == 400
@@ -160,7 +160,7 @@ async def test_about_certificates(aiohttp_client):
         ) as export_mock:
             with tempfile.TemporaryDirectory() as tmp_dir:
                 app = setup_webapp()
-                sev_client = SevClient(Path(tmp_dir))
+                sev_client = SevClient(Path(tmp_dir), Path("/opt/sevctl"))
                 app["sev_client"] = sev_client
                 # Create mock file to return it
                 Path(sev_client.certificates_archive).touch(exist_ok=True)
@@ -170,4 +170,6 @@ async def test_about_certificates(aiohttp_client):
                 assert response.status == 200
                 is_file_mock.assert_has_calls([call(), call()])
                 certificates_expected_dir = sev_client.certificates_archive
-                export_mock.assert_called_once_with(["sevctl", "export", str(certificates_expected_dir)], check=True)
+                export_mock.assert_called_once_with(
+                    ["/opt/sevctl", "export", str(certificates_expected_dir)], check=True
+                )