Add support for --direct access for cim plugins

dissect/target/loaders/direct.py

@@ -1,31 +1,95 @@
 from __future__ import annotations
 
+import sys
+from itertools import chain
 from pathlib import Path
 from typing import TYPE_CHECKING
 
 from dissect.target.filesystem import VirtualFilesystem
+from dissect.target.helpers.logging import get_logger
 from dissect.target.loader import Loader
 from dissect.target.plugins.os.default._os import DefaultOSPlugin
 
 if TYPE_CHECKING:
+    from collections.abc import Iterator
+
     from dissect.target.target import Target
 
+log = get_logger(__name__)
+
 
 class DirectLoader(Loader):
-    def __init__(self, paths: list[str | Path]):
+    def __init__(self, paths: list[str | Path], case_sensitive: bool = False):
+        self.case_sensitive = case_sensitive
         self.paths = [(Path(path) if not isinstance(path, Path) else path).resolve() for path in paths]
 
     @staticmethod
     def detect(path: Path) -> bool:
         return False
 
     def map(self, target: Target) -> None:
-        vfs = VirtualFilesystem()
+        if not self.case_sensitive and self.check_case_insensitive_overlap():
+            log.warning(
+                "Direct mode used in case insensitive mode, but this will cause files overlap, "
+                "consider using --direct-sensitive"
+            )
+        vfs = VirtualFilesystem(case_sensitive=self.case_sensitive)
         for path in self.paths:
             if path.is_file():
-                vfs.map_file(str(path), str(path))
+                vfs.map_file(str(path), path)
             elif path.is_dir():
-                vfs.map_dir(str(path), str(path))
+                vfs.map_dir(str(path), path)
 
         target.filesystems.add(vfs)
         target._os_plugin = DefaultOSPlugin
+
+    def check_case_insensitive_overlap(self) -> bool:
+        """Verify if two differents files will have the same path in a case-insensitive fs"""
+        if sys.version_info >= (3, 12) or sys.platform != "win32":
+
+            def get_files(path: Path) -> Iterator[Path]:
+                """Return list of all files recursively,"""
+                if not path.exists():
+                    return
+                if path.is_file():
+                    yield path
+                # Recursively find all files in the directory
+                yield from path.rglob("*")
+        else:
+
+            def get_files(path: Path, max_depth: int = 7) -> Iterator[Path]:
+                """
+                rglob seems to have issue on windows with python <3.12 when working on a case sensitive FS. Thus
+                we use another implemntation without using rglob.
+                Probably related to https://github.com/python/cpython/issues/94537
+
+                :param path:
+                :param max_depth:
+                :return:
+                """
+                if max_depth == 0:
+                    return
+                if not path.exists():
+                    return
+                if path.is_file():
+                    yield path
+                for f in path.iterdir():
+                    if f.is_dir():
+                        yield from get_files(f, max_depth=max_depth - 1)
+                    else:
+                        yield f
+
+        # Create a flat list of all file paths from all input directories
+        all_paths = chain.from_iterable(get_files(p) for p in self.paths)
+        # Filter out directories, keeping only files
+        all_files = [p for p in all_paths if p.is_file()]
+        # Compare the count of all files with the count of unique, lowercased file paths
+
+        return len({str(p).lower() for p in all_files}) != len(all_files)
+
+    def __repr__(self) -> str:
+        """
+        As DirectLoader does not call super().__init__() self.path is not defined, we need to redefine the __repr__ func
+        :return:
+        """
+        return f"{self.__class__.__name__}({str(self.paths)!r})"

dissect/target/plugins/os/windows/cim.py

@@ -12,6 +12,7 @@
 
 if TYPE_CHECKING:
     from collections.abc import Iterator
+    from pathlib import Path
 
     from dissect.target.target import Target
 
@@ -105,20 +106,38 @@ class CimPlugin(Plugin):
     def __init__(self, target: Target):
         super().__init__(target)
         self._repo = None
-        repodir = self.target.resolve("%windir%/system32/wbem/repository")
+
+        repodirs = list(self.get_paths())
+        if len(repodirs) > 1:
+            raise UnsupportedPluginError(
+                "CIM plugins does not support multiple paths. "
+                "If using with --direct access use wbem/repository folder as input"
+            )
+        if len(repodirs) == 0:
+            raise UnsupportedPluginError("No CIM database found")
+
         self._filters: dict[str, EventFilter] = {}
-        if repodir.exists():
-            index = repodir.joinpath("index.btr")
-            objects = repodir.joinpath("objects.data")
-            mappings = [repodir.joinpath(f"mapping{i}.map") for i in range(1, 4)]
-
-            if all([index.exists(), objects.exists(), all(m.exists() for m in mappings)]):
-                try:
-                    self._repo = cim.CIM(index.open(), objects.open(), [m.open() for m in mappings])
-                except cim.Error as e:
-                    self.target.log.warning("Error opening CIM database")
-                    self.target.log.debug("", exc_info=e)
-            self._filters = self._get_filters()
+        repodir = repodirs[0]
+        index = repodir.joinpath("index.btr")
+        objects = repodir.joinpath("objects.data")
+        mappings = [repodir.joinpath(f"mapping{i}.map") for i in range(1, 4)]
+
+        if all([index.exists(), objects.exists(), all(m.exists() for m in mappings)]):
+            try:
+                self._repo = cim.CIM(index.open(), objects.open(), [m.open() for m in mappings])
+            except cim.Error as e:
+                self.target.log.warning("Error opening CIM database")
+                self.target.log.debug("", exc_info=e)
+        else:
+            missing_files = ",".join(str(f) for f in [index, objects, *mappings] if not f.exists())
+            raise UnsupportedPluginError(f"missing expected files : {missing_files}")
+
+        self._filters = self._get_filters()
+
+    def _get_paths(self) -> Iterator[Path]:
+        repodir = self.target.resolve("%windir%/system32/wbem/repository")
+        if repodir.exists:
+            yield repodir
 
     def check_compatible(self) -> None:
         if not self._repo:

dissect/target/target.py

@@ -446,12 +446,12 @@ def _open_all(spec: str | Path, include_children: bool = False, *, apply: bool =
             raise TargetError(f"Failed to find any loader for targets: {paths}")
 
     @classmethod
-    def open_direct(cls, paths: list[str | Path]) -> Self:
+    def open_direct(cls, paths: list[str | Path], case_sensitive: bool = False) -> Self:
         """Create a minimal target with a virtual root filesystem with all ``paths`` mapped into it.
 
         This is useful when running plugins on individual files.
         """
-        return cls._load("direct", DirectLoader(paths))
+        return cls._load("direct", DirectLoader(paths, case_sensitive))
 
     @property
     def is_direct(self) -> bool:

dissect/target/tools/query.py

@@ -99,6 +99,12 @@ def main() -> int:
         type=pathlib.Path,
         help="write the query report file to the given directory",
     )
+
+    advanced_group = parser.add_argument_group("Advanced options")
+    advanced_group.add_argument(
+        "--direct-sensitive", action="store_true", help="Same as --direct, but paths will be case sensitive"
+    )
+
     configure_generic_arguments(parser)
 
     args, rest = parser.parse_known_args()

dissect/target/tools/utils/cli.py

@@ -214,10 +214,14 @@ def process_plugin_arguments(parser: argparse.ArgumentParser, args: argparse.Nam
 
 
 def open_target(args: argparse.Namespace, *, apply: bool = True) -> Target:
-    direct: bool = getattr(args, "direct", False)
+    direct: bool = getattr(args, "direct", False) or getattr(args, "direct_sensitive", False)
     child: str | None = getattr(args, "child", None)
 
-    target = Target.open_direct(args.target) if direct else Target.open(args.target, apply=apply)
+    target = (
+        Target.open_direct(args.targets, getattr(args, "direct_sensitive", False))
+        if direct
+        else Target.open(args.target, apply=apply)
+    )
 
     if child:
         try:
@@ -234,12 +238,12 @@ def open_target(args: argparse.Namespace, *, apply: bool = True) -> Target:
 
 
 def open_targets(args: argparse.Namespace, *, apply: bool = True) -> Iterator[Target]:
-    direct: bool = getattr(args, "direct", False)
+    direct: bool = getattr(args, "direct", False) or getattr(args, "direct_sensitive", False)
     children: bool = getattr(args, "children", False)
     child: str | None = getattr(args, "child", None)
 
     targets: Iterable[Target] = (
-        [Target.open_direct(args.targets)]
+        [Target.open_direct(args.targets, getattr(args, "direct_sensitive", False))]
         if direct
         else Target.open_all(args.targets, include_children=children, apply=apply)
     )

tests/loaders/test_direct.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+import io
+import logging
+import typing
+
+from dissect.target import Target
+from dissect.target.filesystem import VirtualFilesystem
+
+if typing.TYPE_CHECKING:
+    from pathlib import Path
+
+    import pytest
+
+
+def test_direct_overlap_warning(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Assert direct raise warning in case sensitive mode if some files overlap
+    We must uncompress files in a temporary directory as having two files with same name
+    would cause issue with git on case insensitive fs.
+    """
+    source_vfs = VirtualFilesystem(case_sensitive=True)
+    source_vfs.map_file_fh("file.txt", io.BytesIO(b"a"))
+    source_vfs.map_file_fh("File.txt", io.BytesIO(b"b"))
+    with caplog.at_level(logging.WARNING):
+        _ = Target.open_direct([source_vfs.path("/")], case_sensitive=True)
+        assert (
+            "Direct mode used in case insensitive mode, but this will cause files overlap, "
+            "consider using --direct-sensitive" not in caplog.text
+        )
+    with caplog.at_level(logging.WARNING):
+        _ = Target.open_direct([source_vfs.path("/")], case_sensitive=False)
+        assert (
+            "Direct mode used in case insensitive mode, but this will cause files overlap, "
+            "consider using --direct-sensitive" in caplog.text
+        )
+    with caplog.at_level(logging.WARNING):
+        _ = Target.open_direct([source_vfs.path("/file.txt"), source_vfs.path("/File.txt")], case_sensitive=False)
+        assert (
+            "Direct mode used in case insensitive mode, but this will cause files overlap, "
+            "consider using --direct-sensitive" in caplog.text
+        )

tests/plugins/os/windows/test_cim.py

@@ -7,11 +7,11 @@
     ActiveScriptEventConsumerRecord,
     CommandLineEventConsumerRecord,
 )
+from dissect.target.target import Target
 from tests._utils import absolute_path
 
 if TYPE_CHECKING:
     from dissect.target.filesystem import VirtualFilesystem
-    from dissect.target.target import Target
 
 
 def test_cim_plugin(target_win: Target, fs_win: VirtualFilesystem) -> None:
@@ -28,6 +28,14 @@ def test_cim_plugin(target_win: Target, fs_win: VirtualFilesystem) -> None:
     assert len([record for record in target_win.cim() if record.filter_query]) == 3
 
 
+def test_cim_direct_mode() -> None:
+    data_path = absolute_path("_data/plugins/os/windows/cim/default-namespace")
+    target = Target.open_direct([data_path])
+    records = list(target.cim.consumerbindings())
+
+    assert len(records) == 3
+
+
 r"""
 Result of the WMI query on the system used to generate the test data
 

tests/tools/test_query.py

@@ -495,3 +495,23 @@ def test_mixed_namespace_and_regular_regression(capsys: pytest.CaptureFixture, m
         "<example/descriptor hostname=None domain=None field_a='example' field_b='record'>\n"
         "<example/descriptor hostname=None domain=None field_a='namespace_example' field_b='record'>\n"
     ) in out
+
+
+def test_direct_mode(capsys: pytest.CaptureFixture, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Asset cim plugin works in direct insensitive mode"""
+    with monkeypatch.context() as m:
+        m.setattr(
+            "sys.argv",
+            [
+                "target-query",
+                "-f",
+                "cim",
+                str(absolute_path("_data/plugins/os/windows/cim/default-namespace")),
+                "--direct",
+                "-s",
+            ],
+        )
+
+        target_query()
+        out, _ = capsys.readouterr()
+    assert len(out.splitlines()) == 3