Add filesystem allocation support

dissect/target/filesystem.py

@@ -7,8 +7,11 @@
 import stat
 from collections import defaultdict
 from functools import cache, cached_property
+from itertools import zip_longest
 from typing import TYPE_CHECKING, Any, BinaryIO, Final
 
+from dissect.util.stream import MappingStream
+
 from dissect.target.exceptions import (
     FileNotFoundError,
     FilesystemError,
@@ -174,6 +177,24 @@ def get(self, path: str) -> FilesystemEntry:
         """
         raise NotImplementedError
 
+    def allocated_space_iter(self) -> Iterator[BinaryIO]:
+        """Iteratively yield streams of allocated data on the filesystem."""
+        raise NotImplementedError
+
+    def unallocated_space_iter(self) -> Iterator[BinaryIO]:
+        """Iteratively yield streams of unallocated data on the filesystem."""
+        raise NotImplementedError
+
+    def space_allocation_iter(self) -> Iterator[tuple[BinaryIO, BinaryIO]]:
+        """Iteratively yield tuples of [unallocated, allocated] streams. Sometimes the same bytes need to be parsed to
+        determine both allocated and unallocated space, so this method allows filesystems to expose a method of parsing
+        those bytes once rather than seperately for unallocated and allocated.
+
+        This default implementation just yields from allocated_space_iter and unallocated_space_iter, padding with None
+        values if one is exhausted before the other.
+        """
+        yield from zip_longest(self.unallocated_space_iter(), self.allocated_space_iter())
+
     def open(self, path: str) -> BinaryIO:
         """Open a filesystem entry.
 
@@ -500,6 +521,34 @@ def hash(self, path: str, algos: list[str] | list[Callable] | None = None) -> tu
         """
         return self.get(path).hash(algos)
 
+    def allocated_space(self) -> BinaryIO:
+        """Expose the allocated space of this filesystem as one contiguous stream."""
+        return self._stream_list_to_contiguous(list(self.allocated_space_iter()))
+
+    def unallocated_space(self) -> BinaryIO:
+        """Expose the unallocated space on this filesystem as one contiguous stream."""
+        return self._stream_list_to_contiguous(list(self.unallocated_space_iter()))
+
+    def space_allocation(self) -> tuple[BinaryIO, BinaryIO]:
+        """Fully iterates the underlying filesystem's space allocation iterator function to convert individual 'runs' of
+        unallocated/allocated space into one, contiguous stream for each. Depending on the underlying filesystem, using
+        this function rather than allocated_space() and unallocated_space() seperately may be much faster as
+        space_allocation_iter is called once rather than twice (once for unallocated and once for allocated).
+        """
+        unallocated, allocated = zip(*self.space_allocation_iter(), strict=True)
+        return self._stream_list_to_contiguous(
+            [stream for stream in unallocated if stream is not None]
+        ), self._stream_list_to_contiguous([stream for stream in allocated if stream is not None])
+
+    def _stream_list_to_contiguous(self, streams: list[BinaryIO]) -> MappingStream:
+        """Convert a list of streams into one contiguous stream."""
+        merged_stream = MappingStream(sum(stream.size for stream in streams))
+        offset = 0
+        for stream in streams:
+            merged_stream.add(offset, stream.size, stream)
+            offset += stream.size
+        return merged_stream
+
     @cached_property
     def uuid(self) -> UUID | None:
         """Return the UUID of the filesystem if it has one."""

dissect/target/filesystems/extfs.py

@@ -52,6 +52,20 @@ def _get_node(self, path: str, node: extfs.INode | None = None) -> extfs.INode:
     def uuid(self) -> UUID | None:
         return self.extfs.uuid
 
+    def allocated_space_iter(self) -> Iterator[BinaryIO]:
+        for group_num in range(self.extfs.groups_count):
+            _, allocated = self.extfs.get_group_data_block_allocation(group_num)
+            yield allocated
+
+    def unallocated_space_iter(self) -> Iterator[BinaryIO]:
+        for group_num in range(self.extfs.groups_count):
+            unallocated, _ = self.extfs.get_group_data_block_allocation(group_num)
+            yield unallocated
+
+    def space_allocation_iter(self) -> Iterator[tuple[BinaryIO, BinaryIO]]:
+        for group_num in range(self.extfs.groups_count):
+            yield self.extfs.get_group_data_block_allocation(group_num)
+
 
 class ExtDirEntry(DirEntry):
     fs: ExtFilesystem

dissect/target/filesystems/ntfs.py

@@ -6,6 +6,7 @@
 from typing import TYPE_CHECKING, BinaryIO
 
 from dissect.ntfs import NTFS, NTFS_SIGNATURE
+from dissect.ntfs.bitmap import Bitmap
 from dissect.ntfs.exceptions import Error as NtfsError
 from dissect.ntfs.exceptions import FileNotFoundError as NtfsFileNotFoundError
 from dissect.ntfs.exceptions import NotADirectoryError as NtfsNotADirectoryError
@@ -67,6 +68,18 @@ def _get_record(self, path: str, root: MftRecord | None = None) -> MftRecord:
     def serial(self) -> int | str | None:
         return self.ntfs.serial
 
+    def allocated_space_iter(self) -> Iterator[BinaryIO]:
+        """Iteratively yield streams of allocated data on the filesystem."""
+        yield from (allocated_stream for _, allocated_stream in Bitmap(self.ntfs).iter())
+
+    def unallocated_space_iter(self) -> Iterator[BinaryIO]:
+        """Iteratively yield streams of unallocated data on the filesystem."""
+        yield from (unallocated_stream for unallocated_stream, _ in Bitmap(self.ntfs).iter())
+
+    def space_allocation_iter(self) -> Iterator[tuple[BinaryIO, BinaryIO]]:
+        """Iteratively yield tuples of [unallocated, allocated] data streams on the filesystem."""
+        yield from Bitmap(self.ntfs).iter()
+
 
 class NtfsDirEntry(DirEntry):
     entry: IndexEntry

dissect/target/tools/mount.py

@@ -80,6 +80,14 @@ def main() -> int:
         vnames.setdefault(basename, []).append(v)
         vfs.map_file_fh(f"volumes/{fname}", v)
 
+        if v.fs is not None and (allocation_implementation := getattr(v.fs, "space_allocation", None)):
+            try:
+                unallocated, allocated = allocation_implementation()
+                vfs.map_file_fh(f"volumes/{fname}_unallocated", unallocated)
+                vfs.map_file_fh(f"volumes/{fname}_allocated", allocated)
+            except NotImplementedError:
+                pass
+
     fake_ntfs = set()
     for i, fs in enumerate(t.filesystems):
         ntfs_obj = getattr(fs, "ntfs", None)