Fix handling of volumes at offset 0

dissect/target/target.py

@@ -723,7 +723,8 @@ def open(self, vol: volume.Volume) -> None:
             self.target.log.debug("", exc_info=e)
 
     def apply(self) -> None:
-        todo = self.entries
+        # We don't want later additions to modify the todo, so make a copy
+        todo = self.entries[:]
 
         while todo:
             new_volumes = []
@@ -738,6 +739,21 @@ def apply(self) -> None:
                 else:
                     # We could be getting "regular" volume systems out of LVM or encrypted volumes
                     # Try to open each volume as a regular volume system, or add as a filesystem if it fails
+                    # There are a few scenarios were we want to discard the opened volume, though
+                    #
+                    # If the current volume offset is 0 and originates from a "regular" volume system, we're likely
+                    # opening a volume system on the same disk again
+                    # Sometimes BSD systems are configured this way and an FFS volume "starts" at offset 0
+                    #
+                    # If we opened an empty volume system, it might also be the case that a filesystem actually
+                    # "starts" at offset 0
+
+                    if vol.offset == 0 and vol.vs and vol.vs.__type__ == "disk":
+                        # We are going to re-open a volume system on itself, bail out
+                        self.target.log.info("Found volume with offset 0, opening as raw volume instead")
+                        self.open(vol)
+                        continue
+
                     try:
                         vs = volume.open(vol)
                     except Exception:
@@ -746,15 +762,12 @@ def apply(self) -> None:
                         continue
 
                     if not len(vs.volumes):
+                        # We opened an empty volume system, discard
                         self.open(vol)
                         continue
 
-                    for new_vol in vs.volumes:
-                        if new_vol.offset == 0:
-                            self.target.log.info("Found volume with offset 0, opening as raw volume instead")
-                            self.open(new_vol)
-                            continue
-                        new_volumes.append(new_vol)
+                    self.entries.extend(vs.volumes)
+                    new_volumes.extend(vs.volumes)
 
             self.target.log.debug("LVM volumes found: %s", lvm_volumes)
             self.target.log.debug("Encrypted volumes found: %s", encrypted_volumes)

dissect/target/volume.py

@@ -57,6 +57,11 @@ class VolumeSystem:
         serial: Serial number of the volume system, if any.
     """
 
+    # Due to lazy importing we generally can't use isinstance(), so we add a short identifying string to each class
+    # This has the added benefit of having a readily available "pretty name" for each implementation
+    __type__: str = None
+    """A short string identifying the type of volume system."""
+
     def __init__(
         self, fh: Union[BinaryIO, list[BinaryIO]], dsk: Optional[Container] = None, serial: Optional[str] = None
     ):
@@ -65,6 +70,9 @@ def __init__(
         self.serial = serial
         self._volumes_list: list[Volume] = None
 
+        if self.__type__ is None:
+            raise NotImplementedError(f"{self.__class__.__name__} must define __type__")
+
     def __repr__(self) -> str:
         return f"<{self.__class__.__name__} serial={self.serial}>"
 
@@ -124,20 +132,13 @@ class EncryptedVolumeSystem(VolumeSystem):
     It adds helper functions for interacting with the :attr:`~dissect.target.helpers.keychain.KEYCHAIN`,
     so that subclasses don't have to manually interact with it.
 
-    Subclasses must set the ``PROVIDER`` class attribute to a unique string, e.g. ``bitlocker``.
-
     Args:
         fh: The file-like object on which to open the encrypted volume system.
     """
 
-    PROVIDER: str = None
-
     def __init__(self, fh: BinaryIO, *args, **kwargs):
         super().__init__(fh, *args, **kwargs)
-
-        if not self.PROVIDER:
-            raise ValueError("Provider identifier is not set")
-        self.keys = keychain.get_keys_for_provider(self.PROVIDER) + keychain.get_keys_without_provider()
+        self.keys = keychain.get_keys_for_provider(self.__type__) + keychain.get_keys_without_provider()
 
     def get_keys_for_identifier(self, identifier: str) -> list[keychain.Key]:
         """Retrieves a list of keys that match a single ``identifier``.
@@ -399,7 +400,7 @@ def open_encrypted(volume: BinaryIO) -> Iterator[Volume]:
             log.debug("", exc_info=e)
         except Exception as e:
             log.error(
-                "Failed to open an encrypted volume %s with volume manager %s: %s", volume, manager_cls.PROVIDER, e
+                "Failed to open an encrypted volume %s with volume manager %s: %s", volume, manager_cls.__type__, e
             )
             log.debug("", exc_info=e)
     return None

dissect/target/volumes/bde.py

@@ -17,7 +17,7 @@ class BitlockerVolumeSystemError(VolumeSystemError):
 
 
 class BitlockerVolumeSystem(EncryptedVolumeSystem):
-    PROVIDER = "bitlocker"
+    __type__ = "bitlocker"
 
     def __init__(self, fh: Union[BinaryIO, list[BinaryIO]], *args, **kwargs):
         super().__init__(fh, *args, **kwargs)

dissect/target/volumes/ddf.py

@@ -8,6 +8,8 @@
 
 
 class DdfVolumeSystem(LogicalVolumeSystem):
+    __type__ = "ddf"
+
     def __init__(self, fh: Union[BinaryIO, list[BinaryIO]], *args, **kwargs):
         self.ddf = DDF(fh)
         super().__init__(fh, *args, **kwargs)

dissect/target/volumes/disk.py

@@ -6,6 +6,8 @@
 
 
 class DissectVolumeSystem(VolumeSystem):
+    __type__ = "disk"
+
     def __init__(self, fh: Union[BinaryIO, list[BinaryIO]], *args, **kwargs):
         self._disk = disk.Disk(fh)
         super().__init__(fh, serial=self._disk.serial, *args, **kwargs)

dissect/target/volumes/luks.py

@@ -17,7 +17,7 @@ class LUKSVolumeSystemError(VolumeSystemError):
 
 
 class LUKSVolumeSystem(EncryptedVolumeSystem):
-    PROVIDER = "luks"
+    __type__ = "luks"
 
     def __init__(self, fh: Union[BinaryIO, list[BinaryIO]], *args, **kwargs):
         super().__init__(fh, *args, **kwargs)

dissect/target/volumes/lvm.py

@@ -22,6 +22,8 @@
 
 
 class LvmVolumeSystem(LogicalVolumeSystem):
+    __type__ = "lvm"
+
     def __init__(self, fh: Union[BinaryIO, list[BinaryIO]], *args, **kwargs):
         self.lvm = lvm.LVM2(fh)
         super().__init__(fh, *args, **kwargs)

dissect/target/volumes/md.py

@@ -7,6 +7,8 @@
 
 
 class MdVolumeSystem(LogicalVolumeSystem):
+    __type__ = "md"
+
     def __init__(self, fh: Union[BinaryIO, list[BinaryIO]], *args, **kwargs):
         self.md = MD(fh)
         super().__init__(fh, *args, **kwargs)

dissect/target/volumes/vmfs.py

@@ -11,6 +11,8 @@
 
 
 class VmfsVolumeSystem(LogicalVolumeSystem):
+    __type__ = "vmfs"
+
     def __init__(self, fh: Union[BinaryIO, list[BinaryIO]], *args, **kwargs):
         self.lvm = lvm.LVM(fh)
         super().__init__(fh, *args, **kwargs)

tests/test_target.py

@@ -450,10 +450,14 @@ def test_empty_vs(target_bare: Target) -> None:
 
 def test_nested_vs(target_bare: Target) -> None:
     mock_base_volume = MagicMock()
+    mock_base_volume.offset = 0
+    mock_base_volume.vs = None
+    mock_base_volume.fs = None
     target_bare.volumes.add(mock_base_volume)
 
     mock_volume = MagicMock()
     mock_volume.offset = 0
+    mock_volume.vs.__type__ = "disk"
     mock_volume.fs = None
 
     mock_volume_system = MagicMock()
@@ -464,4 +468,33 @@ def test_nested_vs(target_bare: Target) -> None:
 
         target_bare.volumes.apply()
         filesystem_open.assert_called_once_with(mock_volume)
+        assert len(target_bare.volumes) == 2
+        assert len(target_bare.filesystems) == 1
+
+
+def test_vs_offset_0(target_bare: Target) -> None:
+    mock_disk = MagicMock()
+    mock_disk.vs = None
+    target_bare.disks.add(mock_disk)
+
+    mock_volume = MagicMock()
+    mock_volume.offset = 0
+    mock_volume.vs.__type__ = "disk"
+    mock_volume.fs = None
+
+    mock_volume_system = MagicMock()
+    mock_volume_system.volumes = [mock_volume]
+
+    with patch("dissect.target.volume.open") as volume_open, patch("dissect.target.filesystem.open") as filesystem_open:
+        volume_open.return_value = mock_volume_system
+
+        target_bare.disks.apply()
+        volume_open.assert_called_once_with(mock_disk)
+        assert len(target_bare.volumes) == 1
+
+        target_bare.volumes.apply()
+        # volume.open must still only be called once
+        volume_open.assert_called_once_with(mock_disk)
+        filesystem_open.assert_called_once_with(mock_volume)
+        assert len(target_bare.volumes) == 1
         assert len(target_bare.filesystems) == 1

tests/volumes/test_bde.py

@@ -39,7 +39,7 @@ def test_bde_volume_with_recovery_key(target_win, encrypted_volume):
         keychain.KeyType.RECOVERY_KEY,
         recovery_key,
         identifier=None,
-        provider=BitlockerVolumeSystem.PROVIDER,
+        provider=BitlockerVolumeSystem.__type__,
     )
 
     enc_vol = volume.Volume(encrypted_volume, 1, 0, None, None, None, disk=encrypted_volume)
@@ -67,7 +67,7 @@ def test_bde_volume_with_passphrase(target_win, encrypted_volume):
         keychain.KeyType.PASSPHRASE,
         passphrase,
         identifier=identifier,
-        provider=BitlockerVolumeSystem.PROVIDER,
+        provider=BitlockerVolumeSystem.__type__,
     )
 
     enc_vol = volume.Volume(encrypted_volume, 1, 0, None, None, None, disk=encrypted_volume)