Simplify GenericTarSubLoader

dissect/target/filesystems/tar.py

@@ -40,7 +40,7 @@ class TarFilesystem(Filesystem):
 
     def __init__(
         self,
-        fh: BinaryIO,
+        fh: BinaryIO | None,
         base: str | None = None,
         *,
         tarinfo: tf.TarInfo | None = None,
@@ -51,9 +51,11 @@ def __init__(
 
         if tarfile:
             self.tar = tarfile
-        else:
+        elif fh:
             fh.seek(0)
             self.tar = tf.open(mode="r", fileobj=fh, tarinfo=tarinfo)  # noqa: SIM115
+        else:
+            raise ValueError("fh or tarfile required")
 
         self.base = base or ""
 

dissect/target/loaders/tar.py

@@ -4,12 +4,7 @@
 from io import BytesIO
 from typing import TYPE_CHECKING
 
-from dissect.target import filesystem
-from dissect.target.filesystems.tar import (
-    TarFilesystemDirectoryEntry,
-    TarFilesystemEntry,
-)
-from dissect.target.helpers import fsutil, loaderutil
+from dissect.target.filesystems.tar import TarFilesystem
 from dissect.target.helpers.lazy import import_lazy
 from dissect.target.helpers.logging import get_logger
 from dissect.target.loader import Loader, SubLoader
@@ -92,47 +87,11 @@ def detect(path: Path, tarfile: tf.TarFile) -> bool:
         return True
 
     def map(self, target: target.Target) -> None:
-        volumes = {}
-        windows_found = False
-
-        for member in self.tar.getmembers():
-            if member.name == ".":
-                continue
-
-            if member.name.lower().startswith(WINDOWS_MEMBERS):
-                windows_found = True
-                if "/" in volumes:
-                    # Root filesystem was already added
-                    volumes["/"].case_sensitive = False
-
-            if "/" not in volumes:
-                vol = filesystem.VirtualFilesystem(case_sensitive=not windows_found)
-                vol.tar = self.tar
-                volumes["/"] = vol
-                target.filesystems.add(vol)
-
-            volume = volumes["/"]
-            mname = member.name
-
-            entry_cls = TarFilesystemDirectoryEntry if member.isdir() else TarFilesystemEntry
-            entry = entry_cls(volume, fsutil.normpath(mname), member)
-
-            try:
-                volume.map_file_entry(entry.path, entry)
-            except KeyError as e:
-                log.debug("Skipping directory member %r in tar as %r is already mapped: %s", member, entry.path, e)
-
-        for vol_name, vol in volumes.items():
-            loaderutil.add_virtual_ntfs_filesystem(
-                target,
-                vol,
-                usnjrnl_path=[
-                    "$Extend/$Usnjrnl:$J",
-                    "$Extend/$Usnjrnl:J",  # Old versions of acquire used $Usnjrnl:J
-                ],
-            )
+        """Map the given :class`TarFile` to the :class:`Target` as a :class:`TarFilesystem`."""
+        windows_found = any(name for name in self.tar.getnames() if name.lower().startswith(WINDOWS_MEMBERS))
 
-            target.fs.mount(vol_name, vol)
+        fs = TarFilesystem(fh=None, tarfile=self.tar, case_sensitive=not windows_found)
+        target.filesystems.add(fs)
 
 
 class TarLoader(Loader):

tests/loaders/test_tar.py

@@ -12,7 +12,7 @@
 from dissect.target.plugins.os.windows._os import WindowsPlugin
 from dissect.target.target import Target
 from tests._utils import absolute_path
-from tests.filesystems.test_tar import _mkdir
+from tests.filesystems.test_tar import _mkdir, _mkfile
 
 if TYPE_CHECKING:
     from collections.abc import Callable
@@ -47,11 +47,10 @@ def test_compressed_tar_file(caplog: pytest.LogCaptureFixture) -> None:
     t = Target()
     loader.map(t)
     assert isinstance(loader.subloader, GenericTarSubLoader)
-
     assert len(t.filesystems) == 1
 
+    t.fs.mount("/", t.filesystems[0])
     test_file = t.fs.path("test-data/test-file.txt")
-
     assert test_file.exists()
     assert test_file.open().read() == b"test-value\n"
 
@@ -65,6 +64,7 @@ def test_compressed_tar_file_with_empty_dir() -> None:
 
     t = Target()
     loader.map(t)
+    t.fs.mount("/", t.filesystems[0])
     assert isinstance(loader.subloader, GenericTarSubLoader)
 
     assert len(t.filesystems) == 1
@@ -176,6 +176,23 @@ def test_skip_folder_member_if_previously_mapped() -> None:
     loader = loader_open(path)
     target = Target()
     loader.map(target)
+    target.fs.mount("/", target.filesystems[0])
 
     assert list(target.fs.get("/").iterdir()) == ["folder"]
     assert list(target.fs.get("/folder").iterdir()) == ["file"]
+
+
+def test_windows_tar(tmp_path: Path) -> None:
+    """Test if we can parse a simple Windows tar image without explicit directory entries."""
+    path = tmp_path / "test.tar"
+    tf = tarfile.TarFile(path, mode="w")
+    _mkfile(tf, "Users/John/Documents/hello.txt", b"Hello world!")
+    _mkfile(tf, "Windows/System32/drivers/etc/hosts", b"127.0.0.1\tlocalhost\t#example")
+    tf.close()
+
+    target = Target.open(path)
+    assert target.os == "windows"
+    assert list(map(str, target.fs.path("/").iterdir())) == ["sysvol", "c:"]
+    assert target.fs.path("c:/Users").is_dir()
+    assert target.fs.path("c:/Windows").is_dir()
+    assert target.fs.path("c:/Users/John/Documents/hello.txt").read_text() == "Hello world!"

tests/tools/test_info.py

@@ -27,7 +27,7 @@ def test_target_info(
             assert json.loads(stdout) == {
                 "disks": [],
                 "volumes": [],
-                "mounts": [{"fs": "virtual", "path": "/"}],
+                "mounts": [{"fs": "tar", "path": "/"}],
                 "children": [],
                 "hostname": "ubuntu",
                 "domain": None,
@@ -47,7 +47,7 @@ def test_target_info(
                 "<target/info hostname='ubuntu' domain=None last_activity=None install_date=2024-07-02 12:00:56+00:00 "
                 "ips=[net.ipaddress('1.2.3.4')] os_family='linux' os_version='Ubuntu 22.04.4 LTS (Jammy Jellyfish)' "
                 "architecture=None device=None language=[] timezone='Europe/Amsterdam' disks=[] volumes=[] "
-                "mounts=[\"{'fs': 'virtual', 'path': '/'}\"] children=[]>\n"
+                "mounts=[\"{'fs': 'tar', 'path': '/'}\"] children=[]>\n"
             )
 
         else: