Add explicit img disk image support

README.md

@@ -7,7 +7,7 @@ In its current state, this script leverages common Linux utilities (gdisk, qemu-
 ## Features
 
 * Read-only mounting of VM disk images via qemu-nbd
-* Supports multiple VM disk formats (qcow2, vmdk)
+* Supports raw disk images (including `.img` and `.ami` suffix fallbacks), qcow2, and vmdk
 * Automatic detection and mounting of common filesystems:
   * Windows (NTFS)
   * Linux (ext3, ext4, xfs, ZFS, BTRFS)
@@ -35,7 +35,7 @@ Here's what I envisage for an MVP (Minimum Viable Product), to generate SBOMs fr
 
 ### MVP
 
-* Examine common disk image formats (raw, ami, qcow2, vmdk)
+* Examine common disk image formats (raw, including `.img` and `.ami` suffix fallbacks; qcow2; vmdk)
 * Mount common partition types (ntfs, hfsplus, apfs, ext4, vfat, zfs)
 * Launch Syft with appropriate options to generate an SBOM
 
@@ -79,6 +79,18 @@ $ sudo apt install qemu-utils gdisk fdisk parted util-linux ntfs-3g hfsprogs apf
 
 * Run the python script with a disk image as the only parameter
 
+Files ending in `.img` are treated as raw disk images when `qemu-img` cannot
+identify a more specific format. Content detection always takes precedence, so
+a qcow2 or vmdk image named with an `.img` suffix is still handled using its
+actual format. The image must contain a partition table and filesystem supported
+by `sbom-vm`; package discovery then depends on what Syft can identify in the
+mounted filesystem. Compressed `.img` inputs are rejected; decompress them
+explicitly before scanning so their expanded size can be managed safely.
+
+```bash
+sudo ./sbom-vm.py firmware.img
+```
+
 ## Example output
 
 ### Ubuntu 24.04 qcow2 image with ext4

sbom-vm.py

@@ -37,6 +37,9 @@ def setup_logging(image_path: Path) -> logging.Logger:
 
 logger = None  # Will be initialized in main()
 
+RAW_FALLBACK_SUFFIXES = {'.ami', '.img', '.raw'}
+GZIP_FALLBACK_SUFFIXES = {'.ami', '.raw'}
+
 class ImageMounter:
     def __init__(self, image_path: str, mount_point: str = None):
         self.image_path = Path(image_path)
@@ -45,6 +48,7 @@ def __init__(self, image_path: str, mount_point: str = None):
         self.mounted_partition = None
         self.temp_dir = None
         self.temp_image = None
+        self.prepared_image_format = None
         self.imported_zfs_pool = None
 
     def parse_size(self, size_str):
@@ -90,41 +94,58 @@ def _run_command(self, command: list, check: bool = True, **kwargs) -> subproces
                 logger.error(f"Error output: {e.stderr}")
             raise
 
+    def _has_gzip_magic(self) -> bool:
+        try:
+            with open(self.image_path, 'rb') as image_file:
+                return image_file.read(2) == b'\x1f\x8b'
+        except OSError as e:
+            logger.warning(f"Failed to check if file is gzipped: {e}")
+            return False
+
     def _detect_image_format(self) -> str:
         """Detect the format of the input image."""
         logger.info(f"Detecting format of {self.image_path}")
-
+
+        suffix = self.image_path.suffix.lower()
+        if suffix == '.img' and self._has_gzip_magic():
+            raise RuntimeError(
+                "Compressed .img files are not supported; decompress the image "
+                "before scanning it"
+            )
+
         try:
             result = self._run_command(["qemu-img", "info", str(self.image_path)])
             for line in result.stdout.split('\n'):
                 if line.startswith('file format:'):
                     fmt = line.split(':')[1].strip()
                     logger.info(f"qemu-img detected format: {fmt}")
                     return fmt
-        except subprocess.CalledProcessError as e:
+        except (subprocess.CalledProcessError, OSError) as e:
             logger.warning(f"qemu-img info failed: {e}")
 
         # Fallback to extension-based detection
-        suffix = self.image_path.suffix.lower()
         if suffix == '.vmdk':
             return 'vmdk'
-        elif suffix in ['.ami', '.raw']:
-            # Check if gzipped
-            try:
-                with open(self.image_path, 'rb') as f:
-                    if f.read(2).startswith(b'\x1f\x8b'):
-                        return 'gzip'
-            except Exception as e:
-                logger.warning(f"Failed to check if file is gzipped: {e}")
+        elif suffix in RAW_FALLBACK_SUFFIXES:
+            if suffix in GZIP_FALLBACK_SUFFIXES and self._has_gzip_magic():
+                return 'gzip'
+
+            logger.info(f"Using raw format fallback for {suffix} image")
+            return 'raw'
 
-        return 'raw'  # Default to raw format
+        logger.warning(
+            f"Unable to identify image format from content or suffix {suffix!r}; "
+            "assuming raw"
+        )
+        return 'raw'
 
     def _prepare_image(self) -> Path:
         """Prepare image for mounting, converting if necessary."""
         self.temp_dir = tempfile.mkdtemp(prefix='sbomvm_')
         if self.mount_point is None:
             self.mount_point = Path(tempfile.mkdtemp(prefix='sbomvm_mount_', dir=self.temp_dir))
         image_format = self._detect_image_format()
+        self.prepared_image_format = image_format
 
         if image_format == 'gzip':
             logger.info("Decompressing gzipped image")
@@ -134,6 +155,7 @@ def _prepare_image(self) -> Path:
                     ["gunzip", "-c", str(self.image_path)],
                     stdout=output,
                 )
+            self.prepared_image_format = 'raw'
             return self.temp_image
 
         elif image_format in ['vmdk', 'vhd', 'vpc']:
@@ -146,6 +168,7 @@ def _prepare_image(self) -> Path:
                 str(self.image_path),
                 str(self.temp_image)
             ])
+            self.prepared_image_format = 'qcow2'
             return self.temp_image
 
         return self.image_path
@@ -166,8 +189,17 @@ def _find_free_nbd_device(self) -> str:
     def connect_image(self):
         prepared_image = self._prepare_image()
         self.nbd_device = self._find_free_nbd_device()
-        logger.info(f"Connecting image {prepared_image} to NBD device {self.nbd_device}")
-        self._run_command(["qemu-nbd", "--connect", self.nbd_device, str(prepared_image)])
+        logger.info(
+            f"Connecting {self.prepared_image_format} image {prepared_image} "
+            f"to NBD device {self.nbd_device} (read-only)"
+        )
+        self._run_command([
+            "qemu-nbd",
+            "--read-only",
+            f"--format={self.prepared_image_format}",
+            "--connect", self.nbd_device,
+            str(prepared_image),
+        ])
         # Increase delay to allow NBD device to stabilize
         time.sleep(2)
         # Trigger partition rescanning

tests/test_image_format.py

@@ -0,0 +1,126 @@
+import importlib.util
+import logging
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+from unittest.mock import Mock
+
+
+MODULE_PATH = Path(__file__).resolve().parents[1] / "sbom-vm.py"
+SPEC = importlib.util.spec_from_file_location("sbom_vm", MODULE_PATH)
+assert SPEC is not None and SPEC.loader is not None
+sbom_vm = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(sbom_vm)
+
+
+class ImageFormatTest(unittest.TestCase):
+    def setUp(self):
+        setattr(sbom_vm, "logger", logging.getLogger("sbom-vm-test"))
+
+    def test_qemu_detected_format_takes_precedence_over_img_suffix(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            image_path = Path(temp_dir) / "firmware.img"
+            image_path.write_bytes(b"qcow2 image content")
+            mounter = sbom_vm.ImageMounter(str(image_path))
+            mounter._run_command = Mock(
+                return_value=subprocess.CompletedProcess(
+                    args=["qemu-img"],
+                    returncode=0,
+                    stdout="image: firmware.img\nfile format: qcow2\n",
+                )
+            )
+
+            self.assertEqual(mounter._detect_image_format(), "qcow2")
+
+    def test_img_falls_back_to_raw_when_qemu_cannot_identify_it(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            image_path = Path(temp_dir) / "firmware.img"
+            image_path.write_bytes(b"raw image content")
+            mounter = sbom_vm.ImageMounter(str(image_path))
+            mounter._run_command = Mock(
+                side_effect=subprocess.CalledProcessError(
+                    returncode=1,
+                    cmd=["qemu-img", "info", str(image_path)],
+                )
+            )
+
+            self.assertEqual(mounter._detect_image_format(), "raw")
+
+    def test_img_falls_back_to_raw_when_qemu_is_unavailable(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            image_path = Path(temp_dir) / "firmware.IMG"
+            image_path.write_bytes(b"raw image content")
+            mounter = sbom_vm.ImageMounter(str(image_path))
+            mounter._run_command = Mock(side_effect=FileNotFoundError("qemu-img"))
+
+            self.assertEqual(mounter._detect_image_format(), "raw")
+
+    def test_img_with_gzip_magic_is_rejected(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            image_path = Path(temp_dir) / "firmware.img"
+            image_path.write_bytes(b"\x1f\x8bcompressed image")
+            mounter = sbom_vm.ImageMounter(str(image_path))
+            mounter._run_command = Mock()
+
+            with self.assertRaisesRegex(
+                RuntimeError,
+                r"Compressed \.img files are not supported",
+            ):
+                mounter._detect_image_format()
+
+            mounter._run_command.assert_not_called()
+
+    def test_gzipped_raw_image_keeps_existing_gzip_handling(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            image_path = Path(temp_dir) / "firmware.raw"
+            image_path.write_bytes(b"\x1f\x8bcompressed image")
+            mounter = sbom_vm.ImageMounter(str(image_path))
+            mounter._run_command = Mock(
+                side_effect=subprocess.CalledProcessError(
+                    returncode=1,
+                    cmd=["qemu-img", "info", str(image_path)],
+                )
+            )
+
+            self.assertEqual(mounter._detect_image_format(), "gzip")
+
+    def test_raw_content_detection_precedes_gzip_fallback(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            image_path = Path(temp_dir) / "firmware.raw"
+            image_path.write_bytes(b"\x1f\x8braw image content")
+            mounter = sbom_vm.ImageMounter(str(image_path))
+            mounter._run_command = Mock(
+                return_value=subprocess.CompletedProcess(
+                    args=["qemu-img"],
+                    returncode=0,
+                    stdout="image: firmware.raw\nfile format: raw\n",
+                )
+            )
+
+            self.assertEqual(mounter._detect_image_format(), "raw")
+
+    def test_connect_passes_prepared_format_and_read_only_to_qemu_nbd(self):
+        image_path = Path("firmware.img")
+        mounter = sbom_vm.ImageMounter(str(image_path))
+        mounter._prepare_image = Mock(return_value=image_path)
+        mounter._find_free_nbd_device = Mock(return_value="/dev/nbd0")
+        mounter._run_command = Mock()
+        mounter.prepared_image_format = "raw"
+
+        mounter.connect_image()
+
+        self.assertEqual(
+            mounter._run_command.call_args_list[0].args[0],
+            [
+                "qemu-nbd",
+                "--read-only",
+                "--format=raw",
+                "--connect", "/dev/nbd0",
+                "firmware.img",
+            ],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()