Implement get platform certificates endpoint

src/aleph/vm/conf.py

@@ -387,6 +387,7 @@
             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.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/resources.py

@@ -11,6 +11,7 @@
 from pydantic import BaseModel, Field
 
 from aleph.vm.conf import settings
+from aleph.vm.sevclient import SevClient
 from aleph.vm.utils import cors_allow_all
 
 
@@ -122,6 +123,18 @@ async def about_system_usage(_: web.Request):
     return web.json_response(text=usage.json(exclude_none=True))
 
 
+@cors_allow_all
+async def about_certificates(request: web.Request):
+    """Public endpoint to expose platform certificates for confidential computing."""
+
+    if not settings.ENABLE_CONFIDENTIAL_COMPUTING:
+        return web.HTTPBadRequest(reason="Confidential computing setting not enabled on that server")
+
+    sev_client: SevClient = request.app["sev_client"]
+
+    return web.FileResponse(await sev_client.get_certificates())
+
+
 class Allocation(BaseModel):
     """An allocation is the set of resources that are currently allocated on this orchestrator.
     It contains the item_hashes of all persistent VMs, instances, on-demand VMs and jobs.

src/aleph/vm/orchestrator/supervisor.py

@@ -18,10 +18,11 @@
 
 from aleph.vm.conf import settings
 from aleph.vm.pool import VmPool
+from aleph.vm.sevclient import SevClient
 from aleph.vm.version import __version__
 
 from .metrics import create_tables, setup_engine
-from .resources import about_system_usage
+from .resources import about_certificates, about_system_usage
 from .tasks import (
     start_payment_monitoring_task,
     start_watch_for_messages_task,
@@ -95,6 +96,7 @@
         web.get("/about/executions/details", about_executions),
         web.get("/about/executions/records", about_execution_records),
         web.get("/about/usage/system", about_system_usage),
+        web.get("/about/certificates", about_certificates),
         web.get("/about/config", about_config),
         # /control APIs are used to control the VMs and access their logs
         web.post("/control/allocation/notify", notify_allocation),
@@ -159,6 +161,12 @@
     app["secret_token"] = secret_token
     app["vm_pool"] = pool
 
+    # Store sevctl app singleton only if confidential feature is enabled
+    if settings.ENABLE_CONFIDENTIAL_COMPUTING:
+        sev_client = SevClient(settings.CONFIDENTIAL_DIRECTORY)
+        app["sev_client"] = sev_client
+        # TODO: Review and check sevctl first initialization steps, like (sevctl generate and sevctl provision)
+
     logger.debug(f"Login to /about pages {protocol}://{hostname}/about/login?token={secret_token}")
 
     try:

src/aleph/vm/sevclient.py

@@ -0,0 +1,23 @@
+from pathlib import Path
+
+from aleph.vm.utils import run_in_subprocess
+
+
+class SevClient:
+    def __init__(self, sev_dir: Path):
+        self.sev_dir = sev_dir
+        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:
+        return await run_in_subprocess(
+            ["sevctl", *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))
+
+        return self.certificates_archive

tests/supervisor/test_views.py

@@ -1,8 +1,14 @@
+import tempfile
+from pathlib import Path
+from unittest import mock
+from unittest.mock import call
+
 import pytest
 from aiohttp import web
 
 from aleph.vm.conf import settings
 from aleph.vm.orchestrator.supervisor import setup_webapp
+from aleph.vm.sevclient import SevClient
 
 
 @pytest.mark.asyncio
@@ -121,3 +127,47 @@ def get_persistent_executions(self):
     )
     assert response.status == 200
     assert await response.json() == {"success": True, "successful": [], "failing": [], "errors": {}}
+
+
+@pytest.mark.asyncio
+async def test_about_certificates_missing_setting(aiohttp_client):
+    """Test that the certificates system endpoint returns an error if the setting isn't enabled"""
+    settings.ENABLE_CONFIDENTIAL_COMPUTING = False
+
+    app = setup_webapp()
+    app["sev_client"] = SevClient(Path().resolve())
+    client = await aiohttp_client(app)
+    response: web.Response = await client.get("/about/certificates")
+    assert response.status == 400
+    assert await response.text() == "400: Confidential computing setting not enabled on that server"
+
+
+@pytest.mark.asyncio
+async def test_about_certificates(aiohttp_client):
+    """Test that the certificates system endpoint responds. No auth needed"""
+
+    settings.ENABLE_QEMU_SUPPORT = True
+    settings.ENABLE_CONFIDENTIAL_COMPUTING = True
+    settings.setup()
+
+    with mock.patch(
+        "pathlib.Path.is_file",
+        return_value=False,
+    ) as is_file_mock:
+        with mock.patch(
+            "aleph.vm.sevclient.run_in_subprocess",
+            return_value=True,
+        ) as export_mock:
+            with tempfile.TemporaryDirectory() as tmp_dir:
+                app = setup_webapp()
+                sev_client = SevClient(Path(tmp_dir))
+                app["sev_client"] = sev_client
+                # Create mock file to return it
+                Path(sev_client.certificates_archive).touch(exist_ok=True)
+
+                client = await aiohttp_client(app)
+                response: web.Response = await client.get("/about/certificates")
+                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)