CVM: add attestation report test for Azure Linux using snpguest tool

Signed-off-by: Thien Trung Vuong <[email protected]>

lisa/tools/cargo.py

@@ -3,7 +3,7 @@
 import os
 import re
 from pathlib import PurePath
-from typing import Any, List, Optional, Type, cast
+from typing import Any, List, Optional, Type, Union, cast
 
 from lisa.executable import Tool
 from lisa.operating_system import CBLMariner, Posix, Ubuntu
@@ -117,6 +117,7 @@ def __install_dependencies(self) -> None:
 
     def build(
         self,
+        options: str = "",
         sudo: bool = False,
         cwd: Optional[PurePath] = None,
     ) -> ExecutableResult:
@@ -132,8 +133,9 @@ def build(
         if os.path.dirname(self._command) not in path:
             path = f"{os.path.dirname(self._command)}:{path}"
 
+        command = f"build {options}"
         result = self.run(
-            "build",
+            command,
             expected_exit_code=0,
             expected_exit_code_failure_message=err_msg,
             sudo=sudo,

microsoft/testsuites/cvm/cvm_attestation.py

@@ -3,6 +3,8 @@
 from pathlib import Path
 from typing import Any, Dict
 
+from assertpy import assert_that
+
 from lisa import (
     Environment,
     Logger,
@@ -13,14 +15,15 @@
     features,
 )
 from lisa.features.security_profile import CvmEnabled
-from lisa.operating_system import Ubuntu
+from lisa.operating_system import CBLMariner, Ubuntu
 from lisa.sut_orchestrator import AZURE
 from lisa.testsuite import TestResult, simple_requirement
 from lisa.tools import Ls
 from lisa.util import SkippedException, UnsupportedDistroException
 from microsoft.testsuites.cvm.cvm_attestation_tool import (
     AzureCVMAttestationTests,
     NestedCVMAttestationTests,
+    SnpGuest,
 )
 
 
@@ -34,10 +37,11 @@
 class AzureCVMAttestationTestSuite(TestSuite):
     def before_case(self, log: Logger, **kwargs: Any) -> None:
         node: Node = kwargs["node"]
-        if not isinstance(node.os, Ubuntu):
+        if not isinstance(node.os, Ubuntu) and not isinstance(node.os, CBLMariner):
             raise SkippedException(
                 UnsupportedDistroException(
-                    node.os, "CVM attestation report supports only Ubuntu."
+                    node.os,
+                    "CVM attestation report supports only Ubuntu and Azure Linux.",
                 )
             )
 
@@ -61,11 +65,14 @@ def verify_azure_cvm_attestation_report(
         result: TestResult,
         variables: Dict[str, Any],
     ) -> None:
-        node.tools[AzureCVMAttestationTests].run_cvm_attestation(
-            result,
-            environment,
-            log_path,
-        )
+        if isinstance(node.os, Ubuntu):
+            node.tools[AzureCVMAttestationTests].run_cvm_attestation(
+                result,
+                environment,
+                log_path,
+            )
+        elif isinstance(node.os, CBLMariner):
+            assert_that(node.tools[SnpGuest].run_cvm_attestation()).is_true()
 
 
 @TestSuiteMetadata(

microsoft/testsuites/cvm/cvm_attestation_tool.py

@@ -10,10 +10,11 @@
 from lisa import Environment
 from lisa.executable import Tool
 from lisa.features import SerialConsole
-from lisa.operating_system import Posix, Ubuntu
+from lisa.operating_system import CBLMariner, Posix, Ubuntu
 from lisa.testsuite import TestResult
-from lisa.tools import Dmesg, Echo, Git, Make
+from lisa.tools import Cargo, Dmesg, Echo, Git, Make, Mkdir
 from lisa.util import UnsupportedDistroException
+from lisa.util.process import ExecutableResult
 
 
 class AzureCVMAttestationTests(Tool):
@@ -121,6 +122,151 @@ def _save_attestation_report(self, output: str, log_path: Path) -> None:
             f.write(output)
 
 
+class SnpGuest(Tool):
+    _snpguest_repo = "https://github.com/virtee/snpguest"
+    cmd_path: PurePath
+    repo_root: PurePath
+
+    @property
+    def command(self) -> str:
+        return str(self.cmd_path)
+
+    @property
+    def can_install(self) -> bool:
+        return True
+
+    @property
+    def dependencies(self) -> List[Type[Tool]]:
+        return [Git, Cargo, Mkdir]
+
+    def _initialize(self, *args: Any, **kwargs: Any) -> None:
+        tool_path = self.get_tool_path(use_global=True)
+
+        self.repo_root = tool_path / "snpguest"
+        self.cmd_path = self.repo_root / "target" / "release" / "snpguest"
+
+    def _install(self) -> bool:
+        if isinstance(self.node.os, CBLMariner):
+            self.node.os.install_packages(["perl", "tpm2-tss-devel"])
+        tool_path = self.get_tool_path(use_global=True)
+        git = self.node.tools[Git]
+        git.clone(self._snpguest_repo, tool_path)
+
+        cargo = self.node.tools[Cargo]
+        cargo.build(
+            options="--release --features=hyperv", sudo=False, cwd=self.repo_root
+        )
+
+        return self._check_exists()
+
+    def _fetch_ca(
+        self,
+        certs_dir: str,
+        encoding: str = "der",
+        processor_model: str = "milan",
+        endorser: str = "vcek",
+    ) -> ExecutableResult:
+        failure_msg = "failed to request CA chain from the KDS"
+        return self.run(
+            f"fetch ca {encoding} {processor_model} {certs_dir} --endorser {endorser}",
+            expected_exit_code=0,
+            expected_exit_code_failure_message=failure_msg,
+            shell=True,
+            sudo=False,
+            force_run=True,
+        )
+
+    def _fetch_vcek(
+        self,
+        certs_dir: str,
+        attestation_report_path: str,
+        encoding: str = "der",
+        processor_model: str = "milan",
+    ) -> ExecutableResult:
+        failure_msg = "failed to request VCEK from the KDS"
+        return self.run(
+            f"fetch vcek {encoding} {processor_model} {certs_dir} "
+            f"{attestation_report_path}",
+            expected_exit_code=0,
+            expected_exit_code_failure_message=failure_msg,
+            shell=True,
+            sudo=False,
+            force_run=True,
+        )
+
+    def _request_attestation_report(
+        self, attestation_report_path: str, request_file_path: str
+    ) -> ExecutableResult:
+        failure_msg = "failed to request attestation report from the host"
+        return self.run(
+            f"report {attestation_report_path} {request_file_path} --platform --vmpl 0",
+            expected_exit_code=0,
+            expected_exit_code_failure_message=failure_msg,
+            shell=True,
+            sudo=True,
+            force_run=True,
+        )
+
+    def _verify_certs(self, certs_dir: str) -> ExecutableResult:
+        failure_msg = "failed to verify certificates"
+        return self.run(
+            f"verify certs {certs_dir}",
+            expected_exit_code=0,
+            expected_exit_code_failure_message=failure_msg,
+            shell=True,
+            sudo=False,
+            force_run=True,
+        )
+
+    def _verify_attestation(
+        self, certs_dir: str, attestation_report_path: str
+    ) -> ExecutableResult:
+        failure_msg = "failed to verify attestation report"
+        return self.run(
+            f"verify attestation {certs_dir} {attestation_report_path}",
+            expected_exit_code=0,
+            expected_exit_code_failure_message=failure_msg,
+            shell=True,
+            sudo=False,
+            force_run=True,
+        )
+
+    def run_cvm_attestation(self, processor_model: str = "milan") -> bool:
+        """Regular attestation workflow
+
+        1. Request attestation report
+        2. Request AMD Root Key (ARK) and AMD SEV Key (ASK) from AMD Key Distribution
+           Service (KDS)
+        3. Request the Versioned Chip Endorsement Key (VCEK) from AMD KDS
+        4. Verify the certificates obtained
+        5. Verify the attestation report
+        """
+        data_dir = self.repo_root / "data"
+        certs_dir = data_dir / "certs"
+        attestation_report_path = data_dir / "attestation-report.bin"
+        request_file_path = data_dir / "request-file.txt"
+
+        mkdir = self.node.tools[Mkdir]
+        mkdir.create_directory(certs_dir.as_posix())
+
+        self._request_attestation_report(
+            attestation_report_path.as_posix(), request_file_path.as_posix()
+        )
+        self._fetch_ca(certs_dir.as_posix(), processor_model=processor_model)
+        self._fetch_vcek(
+            certs_dir.as_posix(),
+            attestation_report_path.as_posix(),
+            processor_model=processor_model,
+        )
+
+        self._verify_certs(certs_dir.as_posix())
+        self._verify_attestation(
+            certs_dir.as_posix(), attestation_report_path.as_posix()
+        )
+
+        return True
+
+
 class NestedCVMAttestationTests(Tool):
     repo = "https://github.com/microsoft/confidential-sidecar-containers.git"
     cmd_path: str