Implement get platform certificates endpoint

src/aleph/vm/conf.py

@@ -267,6 +267,11 @@ class Settings(BaseSettings):
         "with SEV and SEV-ES",
     )
 
+    CONFIDENTIAL_DIRECTORY: Path = Field(
+        None,
+        description="Confidential Computing default directory. Default to EXECUTION_ROOT/confidential",
+    )
+
     # Tests on programs
 
     FAKE_DATA_PROGRAM: Optional[Path] = None
@@ -382,6 +387,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.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"
@@ -409,6 +415,7 @@ def setup(self):
 
         os.makedirs(self.EXECUTION_LOG_DIRECTORY, exist_ok=True)
         os.makedirs(self.PERSISTENT_VOLUMES_DIR, exist_ok=True)
+        os.makedirs(self.CONFIDENTIAL_DIRECTORY, exist_ok=True)
 
         self.API_SERVER = self.API_SERVER.rstrip("/")
 
@@ -467,6 +474,8 @@ def __init__(
             self.EXECUTION_LOG_DIRECTORY = self.EXECUTION_ROOT / "executions"
         if not self.JAILER_BASE_DIR:
             self.JAILER_BASE_DIR = self.EXECUTION_ROOT / "jailer"
+        if not self.CONFIDENTIAL_DIRECTORY:
+            self.CONFIDENTIAL_DIRECTORY = self.EXECUTION_ROOT / "confidential"
 
     class Config:
         env_prefix = "ALEPH_VM_"

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,21 @@ 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"]
+
+    if not sev_client.certificates_archive.is_file():
+        await sev_client.export_certificates()
+
+    return web.FileResponse(sev_client.certificates_archive)
+
+
 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 @@ def setup_webapp():
         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 @@ def run():
     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,22 @@
+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:
+        result = await run_in_subprocess(
+            ["sevctl", *args],
+            check=True,
+        )
+
+        return result
+
+    async def export_certificates(self):
+        _ = await self.sevctl_cmd("export", str(self.certificates_archive))

tests/supervisor/test_views.py

@@ -1,8 +1,13 @@
+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 +126,50 @@ 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.CONFIDENTIAL_DIRECTORY = Path().resolve()
+    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:
+            app = setup_webapp()
+            sev_client = SevClient(settings.CONFIDENTIAL_DIRECTORY)
+            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)
+
+            # Remove file mock
+            Path(sev_client.certificates_archive).unlink()