diff --git a/README.md b/README.md index fd60c1a..a376593 100644 --- a/README.md +++ b/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 diff --git a/sbom-vm.py b/sbom-vm.py index 6560d69..78c1d67 100755 --- a/sbom-vm.py +++ b/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,10 +94,25 @@ 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'): @@ -101,23 +120,24 @@ def _detect_image_format(self) -> str: 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.""" @@ -125,6 +145,7 @@ def _prepare_image(self) -> Path: 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 diff --git a/tests/test_image_format.py b/tests/test_image_format.py new file mode 100644 index 0000000..a9f159f --- /dev/null +++ b/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()