Create separate LayerFilesystem from RootFilesystem (#575)

dissect/target/loaders/vb.py

@@ -14,8 +14,8 @@ def detect(path):
         return (mft_exists or c_drive_exists) and config_exists
 
     def map(self, target):
-        ntfs_overlay = target.fs.add_layer()
-        remap_overlay = target.fs.add_layer()
+        remap_overlay = target.fs.append_layer()
+        ntfs_overlay = target.fs.append_layer()
         dfs = DirectoryFilesystem(self.path, case_sensitive=False)
         target.filesystems.add(dfs)
 

dissect/target/plugins/filesystem/walkfs.py

@@ -3,7 +3,7 @@
 from dissect.util.ts import from_unix
 
 from dissect.target.exceptions import FileNotFoundError, UnsupportedPluginError
-from dissect.target.filesystem import RootFilesystemEntry
+from dissect.target.filesystem import LayerFilesystemEntry
 from dissect.target.helpers.fsutil import TargetPath
 from dissect.target.helpers.record import TargetRecordDescriptor
 from dissect.target.plugin import Plugin, export
@@ -50,7 +50,7 @@ def generate_record(target: Target, path: TargetPath) -> FilesystemRecord:
     stat = path.lstat()
     btime = from_unix(stat.st_birthtime) if stat.st_birthtime else None
     entry = path.get()
-    if isinstance(entry, RootFilesystemEntry):
+    if isinstance(entry, LayerFilesystemEntry):
         fs_types = [sub_entry.fs.__type__ for sub_entry in entry.entries]
     else:
         fs_types = [entry.fs.__type__]

dissect/target/plugins/os/unix/esxi/_os.py

@@ -97,7 +97,7 @@ def create(cls, target: Target, sysvol: Filesystem) -> ESXiPlugin:
 
         # Create a root layer for the "local state" filesystem
         # This stores persistent configuration data
-        local_layer = target.fs.add_layer()
+        local_layer = target.fs.append_layer()
 
         # Mount all the visor tars in individual filesystem layers
         _mount_modules(target, sysvol, cfg)
@@ -209,7 +209,7 @@ def _mount_modules(target: Target, sysvol: Filesystem, cfg: dict[str, str]):
             tfs = tar.TarFilesystem(cfile, tarinfo=vmtar.VisorTarInfo)
 
         if tfs:
-            target.fs.add_layer().mount("/", tfs)
+            target.fs.append_layer().mount("/", tfs)
 
 
 def _mount_local(target: Target, local_layer: VirtualFilesystem):

dissect/target/plugins/os/unix/linux/debian/vyos/_os.py

@@ -24,7 +24,7 @@ def __init__(self, target: Target):
         self._version, rootpath = latest
 
         # VyOS does some additional magic with base system files
-        layer = target.fs.add_layer()
+        layer = target.fs.append_layer()
         layer.map_file_entry("/", target.fs.root.get(f"/boot/{self._version}/{rootpath}"))
         super().__init__(target)
 

dissect/target/plugins/os/unix/linux/fortios/_os.py

@@ -113,7 +113,7 @@ def create(cls, target: Target, sysvol: Filesystem) -> FortiOSPlugin:
 
         # FortiGate
         if (datafs_tar := sysvol.path("/datafs.tar.gz")).exists():
-            target.fs.add_layer().mount("/data", TarFilesystem(datafs_tar.open("rb")))
+            target.fs.append_layer().mount("/data", TarFilesystem(datafs_tar.open("rb")))
 
         # Additional FortiGate or FortiManager tars with corrupt XZ streams
         target.log.warning("Attempting to load XZ files, this can take a while.")
@@ -127,11 +127,11 @@ def create(cls, target: Target, sysvol: Filesystem) -> FortiOSPlugin:
         ):
             if (tar := target.fs.path(path)).exists() or (tar := sysvol.path(path)).exists():
                 fh = xz.repair_checksum(tar.open("rb"))
-                target.fs.add_layer().mount("/", TarFilesystem(fh))
+                target.fs.append_layer().mount("/", TarFilesystem(fh))
 
         # FortiAnalyzer and FortiManager
         if (rootfs_ext_tar := sysvol.path("rootfs-ext.tar.xz")).exists():
-            target.fs.add_layer().mount("/", TarFilesystem(rootfs_ext_tar.open("rb")))
+            target.fs.append_layer().mount("/", TarFilesystem(rootfs_ext_tar.open("rb")))
 
         # Filesystem mounts can be discovered in the FortiCare debug report
         # or using ``fnsysctl ls`` and ``fnsysctl df`` in the cli.

dissect/target/tools/shell.py

@@ -31,7 +31,7 @@
     RegistryValueNotFoundError,
     TargetError,
 )
-from dissect.target.filesystem import FilesystemEntry, RootFilesystemEntry
+from dissect.target.filesystem import FilesystemEntry, LayerFilesystemEntry
 from dissect.target.helpers import cyber, fsutil, regutil
 from dissect.target.plugin import arg
 from dissect.target.target import Target
@@ -469,7 +469,7 @@ def scandir(self, path: str, color: bool = False) -> list[tuple[fsutil.TargetPat
                 # If we happen to scan an NTFS filesystem see if any of the
                 # entries has an alternative data stream and also list them.
                 entry = file_.get()
-                if isinstance(entry, RootFilesystemEntry):
+                if isinstance(entry, LayerFilesystemEntry):
                     if entry.entries.fs.__type__ == "ntfs":
                         attrs = entry.lattr()
                         for data_stream in attrs.DATA:

tests/test_filesystem.py

@@ -20,6 +20,7 @@
 )
 from dissect.target.filesystem import (
     FilesystemEntry,
+    LayerFilesystem,
     MappedFile,
     NotASymlinkError,
     RootFilesystem,
@@ -98,10 +99,10 @@ def test_symlink_across_layers(target_bare: Target) -> None:
     vfs2 = VirtualFilesystem()
     target_dir = vfs2.makedirs("/path/to/target")
 
-    layer1 = target_bare.fs.add_layer()
+    layer1 = target_bare.fs.append_layer()
     layer1.mount("/", vfs1)
 
-    layer2 = target_bare.fs.add_layer()
+    layer2 = target_bare.fs.append_layer()
     layer2.mount("/", vfs2)
 
     target_entry = target_bare.fs.get("/path/to/symlink/target").readlink_ext()
@@ -117,10 +118,10 @@ def test_symlink_files_across_layers(target_bare: Target) -> None:
     vfs2 = VirtualFilesystem()
     target_dir = vfs2.makedirs("/path/to/target/derp")
 
-    layer1 = target_bare.fs.add_layer()
+    layer1 = target_bare.fs.append_layer()
     layer1.mount("/", vfs1)
 
-    layer2 = target_bare.fs.add_layer()
+    layer2 = target_bare.fs.append_layer()
     layer2.mount("/", vfs2)
 
     target_entry = target_bare.fs.get("/path/to/symlink/target/derp")
@@ -138,10 +139,10 @@ def test_symlink_to_symlink_across_layers(target_bare: Target) -> None:
     vfs2 = VirtualFilesystem()
     vfs2.symlink("../target", "/path/to/target")
 
-    layer1 = target_bare.fs.add_layer()
+    layer1 = target_bare.fs.append_layer()
     layer1.mount("/", vfs1)
 
-    layer2 = target_bare.fs.add_layer()
+    layer2 = target_bare.fs.append_layer()
     layer2.mount("/", vfs2)
 
     target_entry = target_bare.fs.get("/path/to/symlink/target/").readlink_ext()
@@ -157,10 +158,10 @@ def test_recursive_symlink_across_layers(target_bare: Target) -> None:
     vfs2 = VirtualFilesystem()
     vfs2.symlink("symlink/target", "/path/to/target")
 
-    layer1 = target_bare.fs.add_layer()
+    layer1 = target_bare.fs.append_layer()
     layer1.mount("/", vfs1)
 
-    layer2 = target_bare.fs.add_layer()
+    layer2 = target_bare.fs.append_layer()
     layer2.mount("/", vfs2)
 
     with pytest.raises(SymlinkRecursionError):
@@ -178,13 +179,13 @@ def test_symlink_across_3_layers(target_bare: Target) -> None:
     vfs3 = VirtualFilesystem()
     target_dir = vfs3.makedirs("/path/target")
 
-    layer1 = target_bare.fs.add_layer()
+    layer1 = target_bare.fs.append_layer()
     layer1.mount("/", vfs1)
 
-    layer2 = target_bare.fs.add_layer()
+    layer2 = target_bare.fs.append_layer()
     layer2.mount("/", vfs2)
 
-    layer3 = target_bare.fs.add_layer()
+    layer3 = target_bare.fs.append_layer()
     layer3.mount("/", vfs3)
 
     target_entry = target_bare.fs.get("/path/to/symlink/target/").readlink_ext()
@@ -203,10 +204,10 @@ def test_recursive_symlink_open_across_layers(target_bare: Target) -> None:
     vfs2 = VirtualFilesystem()
     vfs2.symlink("symlink/target", "/path/to/target")
 
-    layer1 = target_bare.fs.add_layer()
+    layer1 = target_bare.fs.append_layer()
     layer1.mount("/", vfs1)
 
-    layer2 = target_bare.fs.add_layer()
+    layer2 = target_bare.fs.append_layer()
     layer2.mount("/", vfs2)
 
     with pytest.raises(SymlinkRecursionError):
@@ -797,7 +798,7 @@ def top_virt_dir() -> VirtualDirectory:
     return VirtualDirectory(Mock(), "")
 
 
-def test_virutal_directory_stat(virt_dir: VirtualDirectory, top_virt_dir: VirtualDirectory) -> None:
+def test_virtual_directory_stat(virt_dir: VirtualDirectory, top_virt_dir: VirtualDirectory) -> None:
     assert virt_dir.stat(follow_symlinks=False) == virt_dir._stat()
     assert virt_dir.stat(follow_symlinks=True) == virt_dir._stat()
 
@@ -806,7 +807,7 @@ def test_virutal_directory_stat(virt_dir: VirtualDirectory, top_virt_dir: Virtua
     assert virt_dir.stat(follow_symlinks=True) == top_virt_dir.stat(follow_symlinks=True)
 
 
-def test_virutal_directory_lstat(virt_dir: VirtualDirectory, top_virt_dir: VirtualDirectory) -> None:
+def test_virtual_directory_lstat(virt_dir: VirtualDirectory, top_virt_dir: VirtualDirectory) -> None:
     assert virt_dir.lstat() == virt_dir._stat()
     assert virt_dir.lstat() == virt_dir.stat(follow_symlinks=False)
     assert virt_dir.lstat().st_mode == stat.S_IFDIR
@@ -815,12 +816,12 @@ def test_virutal_directory_lstat(virt_dir: VirtualDirectory, top_virt_dir: Virtu
     assert virt_dir.lstat() == top_virt_dir.lstat()
 
 
-def test_virutal_directory_is_dir(virt_dir: VirtualDirectory) -> None:
+def test_virtual_directory_is_dir(virt_dir: VirtualDirectory) -> None:
     assert virt_dir.is_dir(follow_symlinks=True)
     assert virt_dir.is_dir(follow_symlinks=False)
 
 
-def test_virutal_directory_is_file(virt_dir: VirtualDirectory) -> None:
+def test_virtual_directory_is_file(virt_dir: VirtualDirectory) -> None:
     assert not virt_dir.is_file(follow_symlinks=True)
     assert not virt_dir.is_file(follow_symlinks=False)
 
@@ -830,21 +831,21 @@ def virt_file() -> VirtualFile:
     return VirtualFile(Mock(), "", Mock())
 
 
-def test_virutal_file_stat(virt_file: VirtualFile) -> None:
+def test_virtual_file_stat(virt_file: VirtualFile) -> None:
     assert virt_file.stat(follow_symlinks=False) == virt_file.lstat()
     assert virt_file.stat(follow_symlinks=True) == virt_file.lstat()
 
 
-def test_virutal_file_lstat(virt_file: VirtualFile) -> None:
+def test_virtual_file_lstat(virt_file: VirtualFile) -> None:
     assert virt_file.lstat().st_mode == stat.S_IFREG
 
 
-def test_virutal_file_is_dir(virt_file: VirtualFile) -> None:
+def test_virtual_file_is_dir(virt_file: VirtualFile) -> None:
     assert not virt_file.is_dir(follow_symlinks=True)
     assert not virt_file.is_dir(follow_symlinks=False)
 
 
-def test_virutal_file_is_file(virt_file: VirtualFile) -> None:
+def test_virtual_file_is_file(virt_file: VirtualFile) -> None:
     assert virt_file.is_file(follow_symlinks=True)
     assert virt_file.is_file(follow_symlinks=False)
 
@@ -1164,3 +1165,62 @@ def test_virtual_filesystem_map_file_from_tar() -> None:
 
     stat = mock_fs.path("/var/example/test.txt").stat()
     assert ts.from_unix(stat.st_mtime) == datetime(2021, 12, 6, 9, 51, 40, tzinfo=timezone.utc)
+
+
+def test_layer_filesystem() -> None:
+    lfs = LayerFilesystem()
+
+    vfs1 = VirtualFilesystem()
+    vfs1.map_file_fh("file1", BytesIO(b"value1"))
+
+    vfs2 = VirtualFilesystem()
+    vfs2.map_file_fh("file2", BytesIO(b"value2"))
+
+    vfs3 = VirtualFilesystem()
+    vfs3.map_file_fh("file3", BytesIO(b"value3"))
+
+    vfs4 = VirtualFilesystem()
+    vfs4.map_file_fh("file1", BytesIO(b"value4"))
+
+    lfs.append_fs_layer(vfs1)
+    assert lfs.path("file1").read_text() == "value1"
+
+    lfs.append_fs_layer(vfs2)
+    assert lfs.path("file1").read_text() == "value1"
+    assert lfs.path("file2").read_text() == "value2"
+
+    lfs.append_fs_layer(vfs3)
+    assert lfs.path("file1").read_text() == "value1"
+    assert lfs.path("file2").read_text() == "value2"
+    assert lfs.path("file3").read_text() == "value3"
+
+    lfs.append_fs_layer(vfs4)
+    assert lfs.path("file1").read_text() == "value4"
+    lfs.remove_fs_layer(vfs4)
+    lfs.prepend_fs_layer(vfs4)
+    assert lfs.path("file1").read_text() == "value1"
+
+
+def test_layer_filesystem_mount() -> None:
+    lfs = LayerFilesystem()
+
+    vfs1 = VirtualFilesystem()
+    vfs1.map_file_fh("file1", BytesIO(b"value1"))
+
+    vfs2 = VirtualFilesystem()
+    vfs2.map_file_fh("file2", BytesIO(b"value2"))
+
+    lfs.mount("/vfs", vfs1)
+    lfs.mount("/vfs", vfs2, ignore_existing=True)
+
+    assert lfs.listdir("/vfs") == ["file2"]
+
+    lfs.mount("/vfs", vfs1, ignore_existing=True)
+
+    assert lfs.listdir("/vfs") == ["file1"]
+
+    lfs.mount("/vfs", vfs2, ignore_existing=False)
+
+    assert sorted(lfs.listdir("/vfs")) == ["file1", "file2"]
+    assert lfs.path("/vfs/file1").read_text() == "value1"
+    assert lfs.path("/vfs/file2").read_text() == "value2"