Changes to expose data stream extents #597 (#634)

log2timeline · Jan 14, 2022 · db2f73d · db2f73d
1 parent 0ca8862
commit db2f73d
Show file tree

Hide file tree

Showing 7 changed files with 75 additions and 41 deletions.
diff --git a/config/dpkg/control b/config/dpkg/control
@@ -9,7 +9,7 @@ Homepage: https://github.com/log2timeline/dfvfs
 
 Package: python3-dfvfs
 Architecture: all
-Depends: libbde-python3 (>= 20140531), libewf-python3 (>= 20131210), libfsapfs-python3 (>= 20201107), libfsext-python3 (>= 20220112), libfshfs-python3 (>= 20220113), libfsntfs-python3 (>= 20211229), libfsxfs-python3 (>= 20210726), libfvde-python3 (>= 20160719), libfwnt-python3 (>= 20210717), libluksde-python3 (>= 20200101), libmodi-python3 (>= 20210405), libphdi-python3 (>= 20220110), libqcow-python3 (>= 20201213), libsigscan-python3 (>= 20191221), libsmdev-python3 (>= 20140529), libsmraw-python3 (>= 20140612), libvhdi-python3 (>= 20201014), libvmdk-python3 (>= 20140421), libvsgpt-python3 (>= 20211115), libvshadow-python3 (>= 20160109), libvslvm-python3 (>= 20160109), python3-cffi-backend (>= 1.9.1), python3-cryptography (>= 2.0.2), python3-dfdatetime (>= 20211113), python3-dtfabric (>= 20170524), python3-idna (>= 2.5), python3-pytsk3 (>= 20210419), python3-pyxattr (>= 0.7.2), python3-yaml (>= 3.10), ${misc:Depends}
+Depends: libbde-python3 (>= 20140531), libewf-python3 (>= 20131210), libfsapfs-python3 (>= 20201107), libfsext-python3 (>= 20220112), libfshfs-python3 (>= 20220113), libfsntfs-python3 (>= 20211229), libfsxfs-python3 (>= 20220113), libfvde-python3 (>= 20160719), libfwnt-python3 (>= 20210717), libluksde-python3 (>= 20200101), libmodi-python3 (>= 20210405), libphdi-python3 (>= 20220110), libqcow-python3 (>= 20201213), libsigscan-python3 (>= 20191221), libsmdev-python3 (>= 20140529), libsmraw-python3 (>= 20140612), libvhdi-python3 (>= 20201014), libvmdk-python3 (>= 20140421), libvsgpt-python3 (>= 20211115), libvshadow-python3 (>= 20160109), libvslvm-python3 (>= 20160109), python3-cffi-backend (>= 1.9.1), python3-cryptography (>= 2.0.2), python3-dfdatetime (>= 20211113), python3-dtfabric (>= 20170524), python3-idna (>= 2.5), python3-pytsk3 (>= 20210419), python3-pyxattr (>= 0.7.2), python3-yaml (>= 3.10), ${misc:Depends}
 Description: Python 3 module of dfVFS
  dfVFS, or Digital Forensics Virtual File System, provides read-only access to
  file-system objects from various storage media types and file formats. The goal

diff --git a/dependencies.ini b/dependencies.ini
@@ -82,7 +82,7 @@ version_property: get_version()
 [pyfsxfs]
 dpkg_name: libfsxfs-python3
 l2tbinaries_name: libfsxfs
-minimum_version: 20210726
+minimum_version: 20220113
 pypi_name: libfsxfs-python
 rpm_name: libfsxfs-python3
 version_property: get_version()

diff --git a/dfvfs/vfs/xfs_file_entry.py b/dfvfs/vfs/xfs_file_entry.py
@@ -7,6 +7,7 @@
 from dfvfs.lib import errors
 from dfvfs.path import xfs_path_spec
 from dfvfs.vfs import attribute
+from dfvfs.vfs import extent
 from dfvfs.vfs import file_entry
 from dfvfs.vfs import xfs_attribute
 from dfvfs.vfs import xfs_directory
@@ -209,13 +210,29 @@ def size(self):
     """int: size of the file entry in bytes or None if not available."""
     return self._fsxfs_file_entry.size
 
-  def GetXFSFileEntry(self):
-    """Retrieves the XFS file entry.
+  def GetExtents(self, data_stream_name=''):
+    """Retrieves extents of a specific data stream.
 
     Returns:
-      pyfsxfs.file_entry: XFS file entry.
+      list[Extent]: extents of the data stream.
     """
-    return self._fsxfs_file_entry
+    extents = []
+    if (self.entry_type == definitions.FILE_ENTRY_TYPE_FILE and
+        not data_stream_name):
+      for extent_index in range(self._fsxfs_file_entry.number_of_extents):
+        extent_offset, extent_size, extent_flags = (
+            self._fsxfs_file_entry.get_extent(extent_index))
+
+        if extent_flags & 0x1:
+          extent_type = definitions.EXTENT_TYPE_SPARSE
+        else:
+          extent_type = definitions.EXTENT_TYPE_DATA
+
+        data_stream_extent = extent.Extent(
+            extent_type=extent_type, offset=extent_offset, size=extent_size)
+        extents.append(data_stream_extent)
+
+    return extents
 
   def GetLinkedFileEntry(self):
     """Retrieves the linked file entry, e.g. for a symbolic link.
@@ -258,3 +275,11 @@ def GetParentFileEntry(self):
 
     return XFSFileEntry(
         self._resolver_context, self._file_system, path_spec, is_root=is_root)
+
+  def GetXFSFileEntry(self):
+    """Retrieves the XFS file entry.
+
+    Returns:
+      pyfsxfs.file_entry: XFS file entry.
+    """
+    return self._fsxfs_file_entry
diff --git a/requirements.txt b/requirements.txt
@@ -10,7 +10,7 @@ libfsapfs-python >= 20201107
 libfsext-python >= 20220112
 libfshfs-python >= 20220113
 libfsntfs-python >= 20211229
-libfsxfs-python >= 20210726
+libfsxfs-python >= 20220113
 libfvde-python >= 20160719
 libfwnt-python >= 20210717
 libluksde-python >= 20200101

diff --git a/setup.cfg b/setup.cfg
@@ -23,7 +23,7 @@ requires = libbde-python3 >= 20140531
            libfsext-python3 >= 20220112
            libfshfs-python3 >= 20220113
            libfsntfs-python3 >= 20211229
-           libfsxfs-python3 >= 20210726
+           libfsxfs-python3 >= 20220113
            libfvde-python3 >= 20160719
            libfwnt-python3 >= 20210717
            libluksde-python3 >= 20200101

diff --git a/tests/vfs/ext_file_entry.py b/tests/vfs/ext_file_entry.py
@@ -97,7 +97,7 @@ def testGetAttributes(self):
     """Tests the _GetAttributes function."""
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_EXT, inode=self._INODE_A_FILE,
-        location='/a_directory/another_file', parent=self._raw_path_spec)
+        location='/a_directory/a_file', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
     self.assertIsNotNone(file_entry)
 
@@ -190,7 +190,7 @@ def testGetFileEntryByPathSpec(self):
     """Tests the GetFileEntryByPathSpec function."""
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_EXT, inode=self._INODE_A_FILE,
-        parent=self._raw_path_spec)
+        location='/a_directory/a_file', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
 
     self.assertIsNotNone(file_entry)
@@ -468,7 +468,7 @@ def testGetFileEntryByPathSpec(self):
     """Tests the GetFileEntryByPathSpec function."""
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_EXT, inode=self._INODE_A_FILE,
-        parent=self._raw_path_spec)
+        location='/a_directory/a_file', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
 
     self.assertIsNotNone(file_entry)

diff --git a/tests/vfs/xfs_file_entry.py b/tests/vfs/xfs_file_entry.py
@@ -55,65 +55,59 @@ def testInitialize(self):
 
   def testAccessTime(self):
     """Test the access_time property."""
-    test_location = '/a_directory/another_file'
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_XFS, inode=self._INODE_ANOTHER_FILE,
-        location=test_location, parent=self._raw_path_spec)
+        location='/a_directory/another_file', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
 
     self.assertIsNotNone(file_entry)
     self.assertIsNotNone(file_entry.access_time)
 
   def testChangeTime(self):
     """Test the change_time property."""
-    test_location = '/a_directory/another_file'
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_XFS, inode=self._INODE_ANOTHER_FILE,
-        location=test_location, parent=self._raw_path_spec)
+        location='/a_directory/another_file', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
 
     self.assertIsNotNone(file_entry)
     self.assertIsNotNone(file_entry.change_time)
 
   def testCreationTime(self):
     """Test the creation_time property."""
-    test_location = '/a_directory/another_file'
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_XFS, inode=self._INODE_ANOTHER_FILE,
-        location=test_location, parent=self._raw_path_spec)
+        location='/a_directory/another_file', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
 
     self.assertIsNotNone(file_entry)
     self.assertIsNotNone(file_entry.creation_time)
 
   def testModificationTime(self):
     """Test the modification_time property."""
-    test_location = '/a_directory/another_file'
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_XFS, inode=self._INODE_ANOTHER_FILE,
-        location=test_location, parent=self._raw_path_spec)
+        location='/a_directory/another_file', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
 
     self.assertIsNotNone(file_entry)
     self.assertIsNotNone(file_entry.modification_time)
 
   def testSize(self):
     """Test the size property."""
-    test_location = '/a_directory/another_file'
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_XFS, inode=self._INODE_ANOTHER_FILE,
-        location=test_location, parent=self._raw_path_spec)
+        location='/a_directory/another_file', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
 
     self.assertIsNotNone(file_entry)
     self.assertEqual(file_entry.size, 22)
 
   def testGetAttributes(self):
     """Tests the _GetAttributes function."""
-    test_location = '/a_directory/a_file'
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_XFS, inode=self._INODE_A_FILE,
-        location=test_location, parent=self._raw_path_spec)
+        location='/a_directory/a_file', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
     self.assertIsNotNone(file_entry)
 
@@ -132,10 +126,9 @@ def testGetAttributes(self):
 
   def testGetStat(self):
     """Tests the _GetStat function."""
-    test_location = '/a_directory/another_file'
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_XFS, inode=self._INODE_ANOTHER_FILE,
-        location=test_location, parent=self._raw_path_spec)
+        location='/a_directory/another_file', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
     self.assertIsNotNone(file_entry)
 
@@ -163,10 +156,9 @@ def testGetStat(self):
 
   def testGetStatAttribute(self):
     """Tests the _GetStatAttribute function."""
-    test_location = '/a_directory/another_file'
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_XFS, inode=self._INODE_ANOTHER_FILE,
-        location=test_location, parent=self._raw_path_spec)
+        location='/a_directory/another_file', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
     self.assertIsNotNone(file_entry)
 
@@ -181,6 +173,30 @@ def testGetStatAttribute(self):
     self.assertEqual(stat_attribute.size, 22)
     self.assertEqual(stat_attribute.type, stat_attribute.TYPE_FILE)
 
+  def testGetExtents(self):
+    """Tests the GetExtents function."""
+    path_spec = path_spec_factory.Factory.NewPathSpec(
+        definitions.TYPE_INDICATOR_XFS, inode=self._INODE_ANOTHER_FILE,
+        location='/a_directory/another_file', parent=self._raw_path_spec)
+    file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
+    self.assertIsNotNone(file_entry)
+
+    extents = file_entry.GetExtents()
+    self.assertEqual(len(extents), 1)
+
+    self.assertEqual(extents[0].extent_type, definitions.EXTENT_TYPE_DATA)
+    self.assertEqual(extents[0].offset, 5652480)
+    self.assertEqual(extents[0].size, 4096)
+
+    path_spec = path_spec_factory.Factory.NewPathSpec(
+        definitions.TYPE_INDICATOR_XFS, inode=self._INODE_A_DIRECTORY,
+        location='/a_directory', parent=self._raw_path_spec)
+    file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
+    self.assertIsNotNone(file_entry)
+
+    extents = file_entry.GetExtents()
+    self.assertEqual(len(extents), 0)
+
   def testGetFileEntryByPathSpec(self):
     """Tests the GetFileEntryByPathSpec function."""
     path_spec = path_spec_factory.Factory.NewPathSpec(
@@ -192,10 +208,9 @@ def testGetFileEntryByPathSpec(self):
 
   def testGetLinkedFileEntry(self):
     """Tests the GetLinkedFileEntry function."""
-    test_location = '/a_link'
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_XFS, inode=self._INODE_A_LINK,
-        location=test_location, parent=self._raw_path_spec)
+        location='/a_link', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
     self.assertIsNotNone(file_entry)
 
@@ -207,10 +222,9 @@ def testGetLinkedFileEntry(self):
 
   def testGetParentFileEntry(self):
     """Tests the GetParentFileEntry function."""
-    test_location = '/a_directory/another_file'
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_XFS, inode=self._INODE_ANOTHER_FILE,
-        location=test_location, parent=self._raw_path_spec)
+        location='/a_directory/another_file', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
     self.assertIsNotNone(file_entry)
 
@@ -222,10 +236,9 @@ def testGetParentFileEntry(self):
 
   def testIsFunctions(self):
     """Tests the Is? functions."""
-    test_location = '/a_directory/another_file'
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_XFS, inode=self._INODE_ANOTHER_FILE,
-        location=test_location, parent=self._raw_path_spec)
+        location='/a_directory/another_file', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
     self.assertIsNotNone(file_entry)
 
@@ -240,10 +253,9 @@ def testIsFunctions(self):
     self.assertFalse(file_entry.IsPipe())
     self.assertFalse(file_entry.IsSocket())
 
-    test_location = '/a_directory'
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_XFS, inode=self._INODE_A_DIRECTORY,
-        location=test_location, parent=self._raw_path_spec)
+        location='/a_directory', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
     self.assertIsNotNone(file_entry)
 
@@ -310,10 +322,9 @@ def testSubFileEntries(self):
 
   def testDataStreams(self):
     """Tests the data streams functionality."""
-    test_location = '/a_directory/another_file'
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_XFS, inode=self._INODE_ANOTHER_FILE,
-        location=test_location, parent=self._raw_path_spec)
+        location='/a_directory/another_file', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
     self.assertIsNotNone(file_entry)
 
@@ -325,10 +336,9 @@ def testDataStreams(self):
 
     self.assertEqual(data_stream_names, [''])
 
-    test_location = '/a_directory'
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_XFS, inode=self._INODE_A_DIRECTORY,
-        location=test_location, parent=self._raw_path_spec)
+        location='/a_directory', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
     self.assertIsNotNone(file_entry)
 
@@ -342,10 +352,9 @@ def testDataStreams(self):
 
   def testGetDataStream(self):
     """Tests the GetDataStream function."""
-    test_location = '/a_directory/another_file'
     path_spec = path_spec_factory.Factory.NewPathSpec(
         definitions.TYPE_INDICATOR_XFS, inode=self._INODE_ANOTHER_FILE,
-        location=test_location, parent=self._raw_path_spec)
+        location='/a_directory/another_file', parent=self._raw_path_spec)
     file_entry = self._file_system.GetFileEntryByPathSpec(path_spec)
     self.assertIsNotNone(file_entry)