Fix URL-encoded filesystem entries in the Velociraptor loader

dissect/target/filesystems/dir.py

@@ -27,12 +27,7 @@ def __repr__(self) -> str:
     def _detect(fh: BinaryIO) -> bool:
         raise TypeError("Detect is not allowed on DirectoryFilesystem class")
 
-    def get(self, path: str) -> FilesystemEntry:
-        path = path.strip("/")
-
-        if not path:
-            return DirectoryFilesystemEntry(self, "/", self.base_path)
-
+    def _resolve_path(self, path: str) -> Path:
         if not self.case_sensitive:
             searchpath = self.base_path
 
@@ -48,6 +43,16 @@ def get(self, path: str) -> FilesystemEntry:
         else:
             entry = self.base_path.joinpath(path.strip("/"))
 
+        return entry
+
+    def get(self, path: str) -> FilesystemEntry:
+        path = path.strip("/")
+
+        if not (path := path.strip("/")):
+            return DirectoryFilesystemEntry(self, "/", self.base_path)
+
+        entry = self._resolve_path(path)
+
         try:
             entry.lstat()
             return DirectoryFilesystemEntry(self, path, entry)

dissect/target/filesystems/zip.py

@@ -56,14 +56,17 @@ def __init__(
             if not mname.startswith(self.base) or mname == ".":
                 continue
 
-            rel_name = fsutil.normpath(mname[len(self.base) :], alt_separator=self.alt_separator)
+            rel_name = self._resolve_path(mname)
             self._fs.map_file_entry(rel_name, ZipFilesystemEntry(self, rel_name, member))
 
     @staticmethod
     def _detect(fh: BinaryIO) -> bool:
         """Detect a zip file on a given file-like object."""
         return zipfile.is_zipfile(fh)
 
+    def _resolve_path(self, path: str) -> str:
+        return fsutil.normpath(path[len(self.base) :], alt_separator=self.alt_separator)
+
     def get(self, path: str, relentry: FilesystemEntry = None) -> FilesystemEntry:
         """Returns a ZipFilesystemEntry object corresponding to the given path."""
         return self._fs.get(path, relentry=relentry)

dissect/target/loaders/dir.py

@@ -36,13 +36,23 @@ def find_entry_path(path: Path) -> str | None:
             return prefix
 
 
-def map_dirs(target: Target, dirs: list[Path | tuple[str, Path]], os_type: str, **kwargs) -> None:
+def map_dirs(
+    target: Target,
+    dirs: list[Path | tuple[str, Path]],
+    os_type: str,
+    *,
+    dirfs: type[DirectoryFilesystem] = DirectoryFilesystem,
+    zipfs: type[ZipFilesystem] = ZipFilesystem,
+    **kwargs,
+) -> None:
     """Map directories as filesystems into the given target.
 
     Args:
         target: The target to map into.
         dirs: The directories to map as filesystems. If a list member is a tuple, the first element is the drive letter.
         os_type: The operating system type, used to determine how the filesystem should be mounted.
+        dirfs: The filesystem class to use for directory filesystems.
+        zipfs: The filesystem class to use for ZIP filesystems.
     """
     alt_separator = ""
     case_sensitive = True
@@ -59,9 +69,9 @@ def map_dirs(target: Target, dirs: list[Path | tuple[str, Path]], os_type: str,
             drive_letter = path.name[0]
 
         if isinstance(path, zipfile.Path):
-            dfs = ZipFilesystem(path.root.fp, path.at, alt_separator=alt_separator, case_sensitive=case_sensitive)
+            dfs = zipfs(path.root.fp, path.at, alt_separator=alt_separator, case_sensitive=case_sensitive)
         else:
-            dfs = DirectoryFilesystem(path, alt_separator=alt_separator, case_sensitive=case_sensitive)
+            dfs = dirfs(path, alt_separator=alt_separator, case_sensitive=case_sensitive)
 
         drive_letter_map[drive_letter].append(dfs)
 

dissect/target/loaders/velociraptor.py

@@ -4,7 +4,11 @@
 import zipfile
 from pathlib import Path
 from typing import TYPE_CHECKING
+from urllib.parse import quote, unquote
 
+from dissect.target.filesystems.dir import DirectoryFilesystem
+from dissect.target.filesystems.zip import ZipFilesystem
+from dissect.target.helpers.fsutil import basename, dirname, join
 from dissect.target.loaders.dir import DirLoader, find_dirs, map_dirs
 from dissect.target.plugin import OperatingSystem
 
@@ -87,11 +91,13 @@ def __init__(self, path: Path, **kwargs):
         super().__init__(path)
 
         if path.suffix == ".zip":
-            log.warning(
-                f"Velociraptor target {path!r} is compressed, which will slightly affect performance. "
-                "Consider uncompressing the archive and passing the uncompressed folder to Dissect."
-            )
             self.root = zipfile.Path(path.open("rb"))
+            if self.root.root.getinfo("uploads.json").compress_type > 0:
+                log.warning(
+                    "Velociraptor target '%s' is compressed, which will slightly affect performance. "
+                    "Consider uncompressing the archive and passing the uncompressed folder to Dissect.",
+                    path,
+                )
         else:
             self.root = path
 
@@ -116,14 +122,28 @@ def detect(path: Path) -> bool:
 
     def map(self, target: Target) -> None:
         os_type, dirs = find_fs_directories(self.root)
-        if os_type == OperatingSystem.WINDOWS:
-            # Velociraptor doesn't have the correct filenames for the paths "$J" and "$Secure:$SDS"
-            map_dirs(
-                target,
-                dirs,
-                os_type,
-                usnjrnl_path="$Extend/$UsnJrnl%3A$J",
-                sds_path="$Secure%3A$SDS",
-            )
-        else:
-            map_dirs(target, dirs, os_type)
+
+        # Velociraptor URL encodes paths before storing these in a collection, this leads plugins not being able to find
+        # these paths. To circumvent this issue, for a zip file the path names are URL decoded before mapping into the
+        # VFS and for a directory the paths are URL encoded at lookup time.
+        map_dirs(
+            target,
+            dirs,
+            os_type,
+            dirfs=VelociraptorDirectoryFilesystem,
+            zipfs=VelociraptorZipFilesystem,
+        )
+
+
+class VelociraptorDirectoryFilesystem(DirectoryFilesystem):
+    def _resolve_path(self, path: str) -> Path:
+        path = quote(path, safe="$/% ")
+        path = (
+            join(dirname(path), str(basename(path)).replace(".", "%2E", 1)) if basename(path).startswith(".") else path
+        )
+        return super()._resolve_path(path)
+
+
+class VelociraptorZipFilesystem(ZipFilesystem):
+    def _resolve_path(self, path: str) -> str:
+        return unquote(super()._resolve_path(path))

tests/loaders/test_velociraptor.py

@@ -17,12 +17,15 @@ def create_root(sub_dir: str, tmp_path: Path) -> Path:
         f"uploads/{sub_dir}/%5C%5C%3F%5CGLOBALROOT%5CDevice%5CHarddiskVolumeShadowCopy1/",
         f"uploads/{sub_dir}/%5C%5C%3F%5CGLOBALROOT%5CDevice%5CHarddiskVolumeShadowCopy1/$Extend",
         f"uploads/{sub_dir}/%5C%5C%3F%5CGLOBALROOT%5CDevice%5CHarddiskVolumeShadowCopy1/windows/system32",
+        f"uploads/{sub_dir}/%5C%5C.%5CC%3A/%2ETEST",
     ]
     root = tmp_path
     mkdirs(root, paths)
 
     (root / "uploads.json").write_bytes(b"{}")
     (root / f"uploads/{sub_dir}/%5C%5C.%5CC%3A/C-DRIVE.txt").write_bytes(b"{}")
+    (root / f"uploads/{sub_dir}/%5C%5C.%5CC%3A/Microsoft-Windows-Windows Defender%254WHC.evtx").write_bytes(b"{}")
+    (root / f"uploads/{sub_dir}/%5C%5C.%5CC%3A/other.txt").write_text("my first file")
 
     with open(absolute_path("_data/plugins/filesystem/ntfs/mft/mft.raw"), "rb") as fh:
         mft = fh.read(10 * 1025)
@@ -54,8 +57,6 @@ def create_root(sub_dir: str, tmp_path: Path) -> Path:
 )
 def test_windows_ntfs(sub_dir: str, other_dir: str, target_bare: Target, tmp_path: Path) -> None:
     root = create_root(sub_dir, tmp_path)
-    root.joinpath(f"uploads/{other_dir}/C%3A").mkdir(parents=True, exist_ok=True)
-    root.joinpath(f"uploads/{other_dir}/C%3A/other.txt").write_text("my first file")
 
     assert VelociraptorLoader.detect(root) is True
 
@@ -72,9 +73,21 @@ def test_windows_ntfs(sub_dir: str, other_dir: str, target_bare: Target, tmp_pat
             usnjrnl_records += len(list(fs.ntfs.usnjrnl.records()))
     assert usnjrnl_records == 2
     assert len(target_bare.filesystems) == 4
+
     assert target_bare.fs.path("sysvol/C-DRIVE.txt").exists()
     assert target_bare.fs.path("sysvol/other.txt").read_text() == "my first file"
 
+    paths = list(target_bare.fs.path("sysvol").iterdir())
+
+    assert paths[2].exists()
+    assert paths[2] == target_bare.fs.path("sysvol/C-DRIVE.txt")
+
+    assert paths[5].exists()
+    assert paths[5] == target_bare.fs.path("sysvol/other.txt")
+
+    assert target_bare.fs.path("sysvol/.TEST").exists()
+    assert target_bare.fs.path("Microsoft-Windows-Windows Defender%4WHC.evtx")
+
 
 @pytest.mark.parametrize(
     "sub_dir",
@@ -102,17 +115,19 @@ def test_windows_ntfs_zip(sub_dir: str, target_bare: Target, tmp_path: Path) ->
     assert usnjrnl_records == 2
     assert len(target_bare.filesystems) == 4
     assert target_bare.fs.path("sysvol/C-DRIVE.txt").exists()
+    assert target_bare.fs.path("sysvol/.TEST").exists()
+    assert target_bare.fs.path("Microsoft-Windows-Windows Defender%4WHC.evtx")
 
 
 @pytest.mark.parametrize(
     "paths",
     [
-        (["uploads/file/etc", "uploads/file/var"]),
-        (["uploads/auto/etc", "uploads/auto/var"]),
-        (["uploads/file/etc", "uploads/file/var", "uploads/file/opt"]),
-        (["uploads/auto/etc", "uploads/auto/var", "uploads/auto/opt"]),
-        (["uploads/file/Library", "uploads/file/Applications"]),
-        (["uploads/auto/Library", "uploads/auto/Applications"]),
+        (["uploads/file/etc", "uploads/file/var", "uploads/file/%2ETEST"]),
+        (["uploads/auto/etc", "uploads/auto/var", "uploads/auto/%2ETEST"]),
+        (["uploads/file/etc", "uploads/file/var", "uploads/file/opt", "uploads/file/%2ETEST"]),
+        (["uploads/auto/etc", "uploads/auto/var", "uploads/auto/opt", "uploads/auto/%2ETEST"]),
+        (["uploads/file/Library", "uploads/file/Applications", "uploads/file/%2ETEST"]),
+        (["uploads/auto/Library", "uploads/auto/Applications", "uploads/auto/%2ETEST"]),
     ],
 )
 def test_unix(paths: list[str], target_bare: Target, tmp_path: Path) -> None:
@@ -125,19 +140,22 @@ def test_unix(paths: list[str], target_bare: Target, tmp_path: Path) -> None:
 
     loader = VelociraptorLoader(root)
     loader.map(target_bare)
+    target_bare.apply()
 
     assert len(target_bare.filesystems) == 1
+    assert target_bare.fs.path("/.TEST").exists()
 
 
 @pytest.mark.parametrize(
     "paths",
     [
-        (["uploads/file/etc", "uploads/file/var"]),
-        (["uploads/auto/etc", "uploads/auto/var"]),
-        (["uploads/file/etc", "uploads/file/var", "uploads/file/opt"]),
-        (["uploads/auto/etc", "uploads/auto/var", "uploads/auto/opt"]),
-        (["uploads/file/Library", "uploads/file/Applications"]),
-        (["uploads/auto/Library", "uploads/auto/Applications"]),
+        (["uploads/file/etc", "uploads/file/var", "uploads/file/%2ETEST"]),
+        (["uploads/file/etc", "uploads/file/var", "uploads/file/%2ETEST"]),
+        (["uploads/auto/etc", "uploads/auto/var", "uploads/auto/%2ETEST"]),
+        (["uploads/file/etc", "uploads/file/var", "uploads/file/opt", "uploads/file/%2ETEST"]),
+        (["uploads/auto/etc", "uploads/auto/var", "uploads/auto/opt", "uploads/auto/%2ETEST"]),
+        (["uploads/file/Library", "uploads/file/Applications", "uploads/file/%2ETEST"]),
+        (["uploads/auto/Library", "uploads/auto/Applications", "uploads/auto/%2ETEST"]),
     ],
 )
 def test_unix_zip(paths: list[str], target_bare: Target, tmp_path: Path) -> None:
@@ -153,5 +171,7 @@ def test_unix_zip(paths: list[str], target_bare: Target, tmp_path: Path) -> None
 
     loader = VelociraptorLoader(zip_path)
     loader.map(target_bare)
+    target_bare.apply()
 
     assert len(target_bare.filesystems) == 1
+    assert target_bare.fs.path("/.TEST").exists()