Add support for --direct access for cim plugins

dissect/target/loaders/direct.py

@@ -1,26 +1,39 @@
 from __future__ import annotations
 
+import functools
+import operator
 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))
@@ -29,3 +42,31 @@ def map(self, target: Target) -> None:
 
         target.filesystems.add(vfs)
         target._os_plugin = DefaultOSPlugin
+
+    def yield_all_file_recursively(self, base_path: Path, max_depth: int = 7) -> Iterator[Path]:
+        """
+        Return list of all files recursively, as rglob is not case-sensitive until python 3.12
+
+        :param base_path:
+        :param max_depth: max depth, prevent infinite recursion
+        :return:
+        """
+        if max_depth == 0:
+            return
+        if not base_path.exists():
+            return
+        if base_path.is_file():
+            yield base_path
+            return
+        for f in base_path.iterdir():
+            if f.is_dir():
+                yield from self.yield_all_file_recursively(f, max_depth=max_depth - 1)
+            else:
+                yield f
+
+    def check_case_insensitive_overlap(self) -> bool:
+        """Verify if two differents files will have the same path in a case-insensitive fs"""
+        all_files_list = list(
+            functools.reduce(operator.iadd, (list(self.yield_all_file_recursively(p)) for p in self.paths), [])
+        )
+        return len({str(p).lower() for p in all_files_list}) != len(all_files_list)

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,22 +106,40 @@ 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 auth log files found")
+
         self._subscription_ns = None
         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._subscription_ns = self._repo.root.namespace("subscription")
-            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._subscription_ns = self._repo.root.namespace("subscription")
+        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)
     )

pyproject.toml

@@ -38,7 +38,7 @@ dependencies = [
     "flow.record~=3.21.0",
     "structlog",
 ]
-dynamic = ["version"]
+version = "2.30.1"
 
 [project.urls]
 homepage = "https://dissect.tools"

tests/loaders/test_direct.py

@@ -0,0 +1,31 @@
+import logging
+from pathlib import Path
+from zipfile import ZipFile
+
+import pytest
+
+from dissect.target import Target
+from tests._utils import absolute_path
+
+
+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.
+    """
+    ZipFile(absolute_path("_data/loaders/direct/overlap.zip")).extractall(tmp_path)
+    if len(list((tmp_path / "overlap").iterdir())) < 2:
+        pytest.skip("Test running on an insensitive fs")
+    with caplog.at_level(logging.WARNING):
+        _ = Target.open_direct([tmp_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
+        )
+        _ = Target.open_direct([tmp_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
+        )

tests/plugins/os/windows/test_cim.py

@@ -4,11 +4,11 @@
 
 from dissect.target.plugins.os.windows import cim
 from dissect.target.plugins.os.windows.cim import 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:
@@ -24,6 +24,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")
+    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,16 @@ 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/")), "--direct", "-s"],
+        )
+
+        target_query()
+        out, _ = capsys.readouterr()
+    assert len(out.splitlines()) == 3