feat: Add EB Linux support to QEMU ITF plugin

.bazelrc

@@ -15,6 +15,7 @@ build:x86_64-qnx --credential_helper=*.qnx.com=%workspace%/tools/qnx_credential_
 
 test:qemu-integration --config=x86_64-qnx
 test:qemu-integration --run_under=//scripts:run_under_qemu
+test:qemu-integration --build_tests_only
 
 build --java_language_version=17
 build --java_runtime_version=remotejdk_17

.github/workflows/itf.yml

@@ -25,6 +25,18 @@ jobs:
         uses: docker/setup-buildx-action@v3.8.0
       - name: Setup Bazel
         uses: bazel-contrib/setup-bazel@0.14.0
+      - name: Install qemu
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y qemu-system
+      - name: Enable KVM group perms
+        run: |
+          echo 'KERNEL=="kvm", GROUP="kvm", MODE="0666", OPTIONS+="static_node=kvm"' | sudo tee /etc/udev/rules.d/99-kvm4all.rules
+          sudo udevadm control --reload-rules
+          sudo udevadm trigger --name-match=kvm
+      - name: Allow unprivileged user namespaces
+        run: |
+          sudo sysctl kernel.apparmor_restrict_unprivileged_userns=0
       - name: Run build
         run: |
           bazel build //...

MODULE.bazel

@@ -224,3 +224,17 @@ imagefs.toolchain(
     type = "ifs",
 )
 use_repo(imagefs, "score_qnx_x86_64_ifs_toolchain")
+
+###############################################################################
+#
+# EBclfsa fastdev archive repository for Linux QEMU integration tests
+#
+###############################################################################
+http_file = use_repo_rule("@bazel_tools//tools/build_defs/repo:http.bzl", "http_file")
+
+http_file(
+    name = "eb_corbos_toolkit_fastdev_archive",
+    downloaded_file_path = "fastdev-ubuntu-ebclfsa-ebcl-qemuarm64.tar.gz",
+    sha256 = "9ca3891b27e4b7bbf6c519d8924283e89c03b62513a988f751a3ee3d10c293e4",
+    urls = ["https://github.com/elektrobit/eb_corbos_toolkit/releases/download/v2.0.0-beta1/fastdev-ubuntu-ebclfsa-ebcl-qemuarm64.tar.gz"],
+)

score/itf/plugins/qemu/__init__.py

@@ -11,7 +11,10 @@
 # SPDX-License-Identifier: Apache-2.0
 # *******************************************************************************
 import logging
+import os
+import subprocess
 import socket
+import tempfile
 import pytest
 
 from score.itf.plugins.qemu.qemu_target import qemu_target
@@ -23,14 +26,74 @@
 logger = logging.getLogger(__name__)
 
 
+def _get_image_format(path_to_image: str) -> str:
+    """Determine the disk image format based on the file extension."""
+    image_lower = path_to_image.lower()
+    if image_lower.endswith(".wic") or image_lower.endswith(".img"):
+        return "raw"
+    return "qcow2"
+
+
+def _create_overlay(base_image: str) -> str:
+    """Create a temporary qcow2 overlay backed by *base_image*.
+
+    All writes go to the ephemeral overlay so the original image is never
+    modified.  The caller is responsible for cleaning up the returned path.
+    """
+    overlay_fd, overlay_path = tempfile.mkstemp(suffix=".qcow2", prefix="qemu_overlay_")
+    os.close(overlay_fd)
+    subprocess.run(
+        [
+            "qemu-img",
+            "create",
+            "-f",
+            "qcow2",
+            "-b",
+            base_image,
+            "-F",
+            _get_image_format(base_image),
+            overlay_path,
+        ],
+        check=True,
+        capture_output=True,
+    )
+    logger.info(f"Created qcow2 overlay: {overlay_path} (backing: {base_image})")
+    return overlay_path
+
+
 def pytest_addoption(parser):
     parser.addoption(
         "--qemu-config",
         action="store",
         required=True,
         help="Path to json file with target configurations.",
     )
-    parser.addoption("--qemu-image", action="store", help="Path to a QEMU image")
+    parser.addoption(
+        "--qemu-image",
+        action="store",
+        default=None,
+        help="Path to a QEMU kernel image",
+    )
+    parser.addoption(
+        "--qemu-disk-image",
+        action="store",
+        default=None,
+        help="Path to a QEMU disk image (qcow2, wic, or img). "
+        "An ephemeral overlay is created so the original image is not modified.",
+    )
+    parser.addoption(
+        "--qemu-architecture",
+        action="store",
+        choices=("x86_64", "aarch64"),
+        default="x86_64",
+        help="Target CPU architecture used to select the QEMU binary and CPU model.",
+    )
+    parser.addoption(
+        "--qemu-kernel-cmdline-file",
+        action="store",
+        default=None,
+        help="Path to a file containing the Linux kernel command line.",
+    )
 
 
 @pytest.fixture(scope="session")
@@ -43,15 +106,39 @@ def dlt():
 
 @pytest.fixture(scope="session")
 def config(request):
+    qemu_kernel_cmdline = None
+    kernel_cmdline_file = request.config.getoption("qemu_kernel_cmdline_file")
+    if kernel_cmdline_file:
+        with open(os.path.abspath(kernel_cmdline_file), encoding="utf-8") as handle:
+            qemu_kernel_cmdline = handle.read().strip()
+
     return Bunch(
         qemu_config=load_configuration(request.config.getoption("qemu_config")),
         qemu_image=request.config.getoption("qemu_image"),
+        qemu_disk_image=request.config.getoption("qemu_disk_image"),
+        qemu_architecture=request.config.getoption("qemu_architecture"),
+        qemu_kernel_cmdline=qemu_kernel_cmdline,
     )
 
 
 @pytest.fixture(scope="session")
 def target_init(config, request, dlt):
     logger.info(f"Starting tests on host: {socket.gethostname()}")
-    with qemu_target(config) as qemu:
-        pre_tests_phase(qemu)
-        yield qemu
+    overlay_path = None
+    if config.qemu_disk_image:
+        overlay_path = _create_overlay(os.path.abspath(config.qemu_disk_image))
+    try:
+        with qemu_target(
+            Bunch(
+                qemu_config=config.qemu_config,
+                qemu_image=config.qemu_image,
+                qemu_disk_image=overlay_path,
+                qemu_architecture=config.qemu_architecture,
+                qemu_kernel_cmdline=config.qemu_kernel_cmdline,
+            )
+        ) as qemu:
+            pre_tests_phase(qemu)
+            yield qemu
+    finally:
+        if overlay_path and os.path.exists(overlay_path):
+            os.unlink(overlay_path)

score/itf/plugins/qemu/qemu.py

@@ -10,15 +10,32 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 # *******************************************************************************
+import logging
 import os
-import shlex
 import subprocess
 import sys
-import logging
 
 logger = logging.getLogger(__name__)
 
 
+_SUPPORTED_ARCHITECTURES = {
+    "x86_64": {
+        "qemu_path": "/usr/bin/qemu-system-x86_64",
+        "cpu": "Cascadelake-Server-v5",
+        "network_device": "virtio-net-pci",
+        "machine": "pc",
+        "block_device": "virtio-blk-pci",
+    },
+    "aarch64": {
+        "qemu_path": "/usr/bin/qemu-system-aarch64",
+        "cpu": "cortex-a53",
+        "network_device": "virtio-net-device",
+        "machine": "virt,virtualization=true,gic-version=3",
+        "block_device": "virtio-blk-device",
+    },
+}
+
+
 class Qemu:
     """
     This class shall be used to start an qemu instance based on pre-configured Qemu parameters.
@@ -27,32 +44,37 @@ class Qemu:
     def __init__(
         self,
         path_to_image,
-        ram="1G",
-        cores="2",
-        cpu="Cascadelake-Server-v5",
-        network_adapters=[],
-        port_forwarding=[],
+        ram,
+        cores,
+        architecture,
+        network_adapters,
+        port_forwarding,
+        disk_image,
+        kernel_cmdline,
     ):
         """Create a QEMU instance with the specified parameters.
 
         :param str path_to_image: The path to the Qemu image file.
         :param str ram: The amount of RAM to allocate to the QEMU instance.
         :param str cores: The number of CPU cores to allocate to the QEMU instance.
-        :param str cpu: The CPU model to emulate.
-         Default is Cascadelake-Server-v5 used to emulate modern Intel CPU features.
-         For older Ubuntu versions change that to host in case of errors.
+        :param str architecture: The CPU architecture to emulate (x86_64 or aarch64).
+        :param list network_adapters: List of network adapter names.
+        :param list port_forwarding: List of port forwarding configurations.
+        :param str disk_image: Optional path to a qcow2 disk image.
+        :param str kernel_cmdline: Optional kernel command line string.
         """
-        self.__qemu_path = "/usr/bin/qemu-system-x86_64"
+        if architecture not in _SUPPORTED_ARCHITECTURES:
+            raise ValueError("architecture must be one of: " + ", ".join(sorted(_SUPPORTED_ARCHITECTURES)))
+        self.__arch_config = _SUPPORTED_ARCHITECTURES[architecture]
         self.__path_to_image = path_to_image
         self.__ram = ram
         self.__cores = cores
-        self.__cpu = cpu
         self.__network_adapters = network_adapters
         self.__port_forwarding = port_forwarding
+        self.__disk_image = disk_image
+        self.__kernel_cmdline = kernel_cmdline
 
         self.__check_qemu_is_installed()
-        self.__find_available_kvm_support()
-        self.__check_kvm_readable_when_necessary()
 
         self._subprocess = None
 
@@ -82,46 +104,28 @@ def stop(self):
             raise Exception(f"QEMU process returned: {ret}")
 
     def __check_qemu_is_installed(self):
-        if not os.path.isfile(self.__qemu_path):
-            logger.fatal(f"Qemu is not installed under {self.__qemu_path}")
+        qemu_path = self.__arch_config["qemu_path"]
+        if not os.path.isfile(qemu_path):
+            logger.fatal(f"Qemu is not installed under {qemu_path}")
             sys.exit(-1)
 
-    def __find_available_kvm_support(self):
-        self._accelerator_support = "kvm"
-        with open("/proc/cpuinfo") as cpuinfo:
-            cpu_options = str(cpuinfo.read())
-            if "vmx" not in cpu_options and "svm" not in cpu_options:
-                logger.error("No virtual capability on machine. We're using standard TCG accel on QEMU")
-                self._accelerator_support = "tcg"
-
-            if not os.path.exists("/dev/kvm"):
-                logger.error("No KVM available. We're using standard TCG accel on QEMU")
-                self._accelerator_support = "tcg"
-
-    def __check_kvm_readable_when_necessary(self):
-        if self._accelerator_support == "kvm":
-            if not os.access("/dev/kvm", os.R_OK):
-                logger.fatal(
-                    "You dont have access rights to /dev/kvm. Consider adding yourself to kvm group. Aborting."
-                )
-                sys.exit(-1)
-
     def __build_qemu_command(self):
-        # Use hardware virtualization if available
-        accel = ["-enable-kvm"] if self._accelerator_support == "kvm" else ["-accel", "tcg"]
-
         return (
-            [f"{self.__qemu_path}"]
-            + accel
-            + [
+            [
+                self.__arch_config["qemu_path"],
+                # Use hardware virtualization if available
+                "-accel",
+                "kvm",
+                "-accel",
+                "tcg",
                 "-smp",
                 f"{self.__cores},maxcpus={self.__cores},cores={self.__cores}",
+                "-machine",
+                self.__arch_config["machine"],
                 "-cpu",
-                f"{self.__cpu}",  # Specify CPU to emulate
+                self.__arch_config["cpu"],  # Specify CPU to emulate
                 "-m",
                 f"{self.__ram}",  # Specify RAM size
-                "-kernel",
-                f"{self.__path_to_image}",  # Specify kernel image
                 "-nographic",  # Disable graphical display (console-only)
                 "-serial",
                 "mon:stdio",  # Redirect serial output to console
@@ -132,15 +136,35 @@ def __build_qemu_command(self):
             ]
             + self.__network_devices_args()
             + self.__port_forwarding_args()
+            + self.__kernel_args()
+            + self.__disk_image_args()
         )
 
+    def __kernel_args(self):
+        if not self.__path_to_image:
+            return []
+        args = ["-kernel", self.__path_to_image]
+        if self.__kernel_cmdline:
+            args.extend(["-append", self.__kernel_cmdline])
+        return args
+
+    def __disk_image_args(self):
+        if not self.__disk_image:
+            return []
+        return [
+            "-device",
+            f"{self.__arch_config['block_device']},drive=vd0",
+            "-drive",
+            f"if=none,format=qcow2,file={self.__disk_image},id=vd0",
+        ]
+
     def __network_devices_args(self):
         def get_netdev_args(adapter, id):
             return [
                 "-netdev",
                 f"tap,id=t{id},ifname={adapter},script=no,downscript=no",
                 "-device",
-                f"virtio-net-pci,netdev=t{id},id=nic{id},guest_csum=off",
+                f"{self.__arch_config['network_device']},netdev=t{id},id=nic{id},guest_csum=off",
             ]
 
         result = []
@@ -157,7 +181,7 @@ def __port_forwarding_args(self):
                     "-netdev",
                     f"user,id=net{id},hostfwd=tcp::{forwarding.host_port}-:{forwarding.guest_port}",
                     "-device",
-                    f"virtio-net-pci,netdev=net{id}",
+                    f"{self.__arch_config['network_device']},netdev=net{id}",
                 ]
             )
         return result