Read MFT records based on segment numbers

dissect/ntfs/mft.py

@@ -120,12 +120,20 @@ def get(self, ref: Union[int, str, Instance], root: Optional[MftRecord] = None)
         else:
             raise TypeError(f"Invalid MFT reference: {ref!r}")
 
-    def segments(self) -> Iterator[MftRecord]:
-        """Yield all valid MFT records, regardless if they're allocated or not."""
+    def segments(self, start: int = 0, end: int = -1) -> Iterator[MftRecord]:
+        """Yield all valid MFT records, regardless if they're allocated or not.
+
+        Args:
+            start: The starting segment number. Use ``-1`` to start from the last segment.
+            end: The ending segment number. Use ``-1`` to end with the last segment.
+        """
         record_size = self.ntfs._record_size if self.ntfs else DEFAULT_RECORD_SIZE
-        mft_size = self.get(FILE_NUMBER_MFT).size()
+        last_segment = self.get(FILE_NUMBER_MFT).size() // record_size
+        start = last_segment if start == -1 else start
+        end = last_segment if end == -1 else end
+        step = 1 if start <= end else -1
 
-        for segment in range(mft_size // record_size):
+        for segment in range(start, end + step, step):
             try:
                 yield self.get(segment)
             except Error:

tests/test_mft.py

@@ -79,3 +79,18 @@ def test_mft_record() -> None:
     record = mft.get(0)
     assert record.filename == "$MFT"
     assert record.filenames() == ["$MFT"]
+
+
+def test_mft_records_segment_number(mft_bin: BinaryIO) -> None:
+    fs = NTFS(mft=mft_bin)
+    assert fs.mft
+
+    records_forward = list(fs.mft.segments(0, 4))
+    assert len(records_forward) == 5
+    assert records_forward[0].filename == "$MFT"
+    assert records_forward[4].filename == "$AttrDef"
+
+    records_backwards = list(fs.mft.segments(4, 0))
+    assert len(records_backwards) == 5
+    assert records_backwards[0].filename == "$AttrDef"
+    assert records_backwards[4].filename == "$MFT"