Add the option to acquire private openssh keys

Additionally, it adds the options to filter out
certain paths if the filter function evaluates to `True`

(DIS-1869)

acquire/acquire.py

@@ -1354,6 +1354,11 @@ class Boot(Module):
     ]
 
 
+def private_key_filter(path: fsutil.TargetPath) -> bool:
+    with path.open("rt") as file:
+        return "PRIVATE KEY" in file.readline()
+
+
 @register_module("--home")
 class Home(Module):
     SPEC = [
@@ -1371,33 +1376,25 @@ class Home(Module):
 
 
 @register_module("--ssh")
+@module_arg("--private-keys", action="store_true", help="Add any private keys", default=False)
 class SSH(Module):
-    @classmethod
-    def _run(cls, target: Target, collector):
-        user_pattern = ".ssh/*"
-
-        # Gather user paths
-        # TODO: Use from_user_home if supported for osx
-        if target._os.os == "osx":
-            iterator = [f"/Users/*/{user_pattern}"]
-        else:
-            iterator = list(from_user_home(target, user_pattern))
+    SPEC = [
+        ("glob", ".ssh/*", from_user_home),
+        ("glob", "/etc/ssh/*"),
+        ("glob", "sysvol/ProgramData/ssh/*"),
+    ]
 
+    @classmethod
+    def run(cls, target: Target, cli_args: argparse.Namespace, collector: Collector):
         # Acquire SSH configuration in sshd directories
-        iterator += ["/etc/ssh/*", "sysvol/ProgramData/ssh/*"]
 
-        globbed_path = (path for pattern in iterator for path in target.fs.glob(pattern))
-        for path in globbed_path:
-            if target.fs.path(path).is_dir():
-                collector.collect_dir(path)
-                continue
+        filter = None if cli_args.private_keys else private_key_filter
 
-            with target.fs.path(path).open("rt") as file:
-                if "PRIVATE KEY" in file.readline():
-                    # Detected a private key, skipping.
-                    continue
+        if filter:
+            log.info("Executing SSH without --private-keys, skipping private keys.")
 
-            collector.collect_file(path, outpath=path)
+        with collector.file_filter(filter):
+            super().run(target, cli_args, collector)
 
 
 @register_module("--var")

acquire/collector.py

@@ -10,7 +10,16 @@
 from dataclasses import dataclass
 from itertools import groupby
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Iterable, Optional, Sequence, Type, Union
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Iterable,
+    Optional,
+    Sequence,
+    Type,
+    Union,
+)
 
 from dissect.target import Target
 from dissect.target.exceptions import (
@@ -185,6 +194,7 @@ def __init__(self, target: Target, output: Output, base: str = "fs", skip_list:
 
         self.report = CollectionReport()
         self.bound_module_name = None
+        self.filter = lambda _: False
 
         self.output.init(self.target)
 
@@ -202,6 +212,15 @@ def bind_module(self, module: Type) -> Collector:
         finally:
             self.unbind()
 
+    @contextmanager
+    def file_filter(self, filter: Optional[Callable[[fsutil.TargetPath], bool]]) -> Collector:
+        try:
+            if filter:
+                self.filter = filter
+            yield self
+        finally:
+            self.filter = lambda _: False
+
     def bind(self, module: Type) -> None:
         self.bound_module_name = module.__name__
 
@@ -272,6 +291,10 @@ def collect_file(
         if not isinstance(path, fsutil.TargetPath):
             path = self.target.fs.path(path)
 
+        if self.filter(path) is True:
+            log.info("- Collecting file %s: Skipped (filtered out)", path)
+            return
+
         if self.report.was_path_seen(path):
             log.info("- Collecting file %s: Skipped (DEDUP)", path)
             return