Add vmware vCenter support bundle and ESXi plugins

dissect/target/plugins/child/vmware_vcenter.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from dissect.target.exceptions import UnsupportedPluginError
+from dissect.target.helpers.record import ChildTargetRecord
+from dissect.target.plugin import ChildTargetPlugin
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+    from dissect.target.target import Target
+
+
+class VmwareVcenterChildTargetPlugin(ChildTargetPlugin):
+    """Child target plugin that yields from VMware vCenter support bundle."""
+
+    __type__ = "vmware_vcenter"
+
+    def __init__(self, target: Target):
+        super().__init__(target)
+        self.hypervisors = list(self.target.fs.glob("/*@*.tgz"))
+
+    def check_compatible(self) -> None:
+        if not self.hypervisors:
+            raise UnsupportedPluginError("No Vmware Vcenter childs found")
+
+    def list_children(self) -> Iterator[ChildTargetRecord]:
+        for hypervisor in self.hypervisors:
+            yield ChildTargetRecord(
+                type=self.__type__,
+                path=hypervisor,
+                _target=self.target,
+            )

dissect/target/plugins/os/unix/bsd/citrix/history.py

@@ -61,19 +61,6 @@ class CitrixCommandHistoryPlugin(CommandHistoryPlugin):
         ("citrix-netscaler-cli", ".nscli_history"),
     )
 
-    def _find_history_files(self) -> list[tuple[str, TargetPath, UnixUserRecord | None]]:
-        """Find history files on the target that this plugin can parse."""
-        history_files = []
-        for shell, history_absolute_path_glob in self.COMMAND_HISTORY_ABSOLUTE_PATHS:
-            history_files.extend(
-                (shell, path, None) for path in self.target.fs.path("/").glob(history_absolute_path_glob.lstrip("/"))
-            )
-
-        # Also utilize the _find_history_files function of the parent class for relative paths.
-        history_files.extend(super()._find_history_files())
-
-        return history_files
-
     def _find_user_by_name(self, username: str) -> UnixUserRecord | None:
         """Cached function to return the matching UnixUserRecord for a given username."""
         if username is None:

dissect/target/plugins/os/unix/esxi/history.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import re
+from typing import TYPE_CHECKING
+
+from dissect.target.plugin import export
+from dissect.target.plugins.os.unix.history import (
+    CommandHistoryPlugin,
+    CommandHistoryRecord,
+)
+from dissect.target.plugins.os.windows.defender._plugin import parse_iso_datetime
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+    from dissect.target.helpers.fsutil import TargetPath
+
+RE_ESXI_BASH_HISTORY_LINE = re.compile(
+    r"^(?P<date>\S+)\s+"
+    r"(?P<process>[^\[:]+)"
+    r"(?:\[(?P<pid>\d+)\])?"
+    r":\s+"
+    r"(?:\[(?P<user>[^\]]+)\]:\s*)?"
+    r"(?P<command>.+)$"
+)
+
+
+class ESXICommandHistoryPlugin(CommandHistoryPlugin):
+    """ESXI command history plugin."""
+
+    COMMAND_HISTORY_ABSOLUTE_PATHS = (("esxi-shell", "/var/run/log/shell.log*"),)
+    COMMAND_HISTORY_RELATIVE_PATHS = ()
+
+    @export(record=CommandHistoryRecord)
+    def commandhistory(self) -> Iterator[CommandHistoryRecord]:
+        """Return shell history."""
+        for shell, history_path, user in self._history_files:
+            if shell == "esxi-shell":
+                yield from self.parse_esxi_shell_history(history_path)
+            else:
+                yield from self.parse_generic_history(history_path, user, shell)
+
+    def parse_esxi_shell_history(self, path: TargetPath) -> Iterator[CommandHistoryRecord]:
+        """Parse shell.log* contents."""
+        i = 0
+        for line in path.open("rt", errors="replace"):
+            line_match = RE_ESXI_BASH_HISTORY_LINE.match(line)
+
+            if not line_match:
+                self.target.log.warning("Failed to parse line %s", line)
+                continue
+
+            record = CommandHistoryRecord(
+                ts=parse_iso_datetime(line_match.group("date")),
+                command=line_match.group("command"),
+                order=i,  # year_rollover_helper returns entries in reverse order.
+                shell="esxi-shell",
+                source=path,
+                _target=self.target,
+            )
+
+            record.username = line_match.group("user")
+            yield record

dissect/target/plugins/os/unix/history.py

@@ -37,6 +37,8 @@
 class CommandHistoryPlugin(Plugin):
     """UNIX command history plugin."""
 
+    COMMAND_HISTORY_ABSOLUTE_PATHS = ()
+
     COMMAND_HISTORY_RELATIVE_PATHS = (
         ("bash", ".bash_history"),
         ("fish", ".local/share/fish/fish_history"),
@@ -61,6 +63,11 @@ def check_compatible(self) -> None:
     def _find_history_files(self) -> list[tuple[str, TargetPath, UnixUserRecord]]:
         """Find existing history files."""
         history_files = []
+        for shell, history_absolute_path_glob in self.COMMAND_HISTORY_ABSOLUTE_PATHS:
+            history_files.extend(
+                (shell, path, None) for path in self.target.fs.path("/").glob(history_absolute_path_glob.lstrip("/"))
+            )
+
         for user_details in self.target.user_details.all_with_home():
             for shell, history_relative_path in self.COMMAND_HISTORY_RELATIVE_PATHS:
                 history_path = user_details.home_path.joinpath(history_relative_path)

dissect/target/plugins/os/unix/log/auth.py

@@ -344,7 +344,8 @@ def check_compatible(self) -> None:
 
     def _get_paths(self) -> Iterator[Path]:
         var_log = self.target.fs.path("/var/log")
-        return chain(var_log.glob("auth.log*"), var_log.glob("secure*"))
+        var_run_log = self.target.fs.path("/var/run/log")
+        return chain(var_log.glob("auth.log*"), var_log.glob("secure*"), var_run_log.glob("auth.log*"))
 
     @alias("securelog")
     @export(record=DynamicDescriptor(["datetime", "path", "string"]))

dissect/target/plugins/os/unix/log/helpers.py

@@ -15,7 +15,7 @@
 log = get_logger(__name__)
 
 RE_TS = re.compile(r"^[A-Za-z]{3}\s*\d{1,2}\s\d{1,2}:\d{2}:\d{2}")
-RE_TS_ISO = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}\+\d{2}:\d{2}")
+RE_TS_ISO = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{2,6})?((?:\+\d{2}:\d{2})|Z)")
 RE_LINE = re.compile(
     r"""
     \d{2}:\d{2}\s                           # First match on the similar ending of the different timestamps
@@ -41,11 +41,16 @@ def iso_readlines(file: Path, max_lines: int | None = None) -> Iterator[tuple[da
                 log.debug("Skipping line: %s", line)
                 continue
 
-            try:
-                ts = datetime.strptime(match[0], "%Y-%m-%dT%H:%M:%S.%f%z")
-            except ValueError as e:
-                log.warning("Unable to parse ISO timestamp in line: %s", line)
-                log.debug("", exc_info=e)
+            ts = None
+            for format in ["%Y-%m-%dT%H:%M:%S.%f%z", "%Y-%m-%dT%H:%M:%S%z"]:
+                try:
+                    ts = datetime.strptime(match[0], format)
+                    break
+                except ValueError:
+                    continue
+
+            if not ts:
+                log.warning("Unable to parse ISO timestamp in line: %s TS:%s", line, match[0])
                 continue
 
             yield ts, line

tests/plugins/child/test_vmware_vcenter.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, BinaryIO
+
+from dissect.target.plugins.child.vmware_vcenter import (
+    VmwareVcenterChildTargetPlugin,
+)
+
+if TYPE_CHECKING:
+    from dissect.target.filesystem import VirtualFilesystem
+    from dissect.target.target import Target
+
+
+def test_child_vmware_vcenter(target_unix: Target, fs_unix: VirtualFilesystem) -> None:
+    """Test if we detect VMware vCenter children"""
+    fs_unix.map_file_fh("/10.10.10.10-vm2026-01-07@00-00-00.tgz", BinaryIO())
+
+    target_unix.add_plugin(VmwareVcenterChildTargetPlugin)
+    children = [child for _, child in target_unix.list_children()]
+
+    assert len(children) == 1
+    assert children[0].type == "vmware_vcenter"
+    assert children[0].path == "/10.10.10.10-vm2026-01-07@00-00-00.tgz"

tests/plugins/os/unix/esxi/test_history.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import datetime
+import gzip
+import textwrap
+from io import BytesIO
+from typing import TYPE_CHECKING
+
+import pytest
+
+from dissect.target.plugins.os.unix.esxi.history import ESXICommandHistoryPlugin
+
+if TYPE_CHECKING:
+    from dissect.target.filesystem import VirtualFilesystem
+    from dissect.target.target import Target
+
+COMMANDHISTORY_DATA = """\
+2026-01-01T02:03:04.891Z In(14) shell[37790689]: Interactive shell session started
+2026-01-01T02:03:04.891Z In(14) shell[37790689]: [root]: ls -lah
+2026-01-01T03:04:04.891Z ESXShell: ESXi Shell unavailable
+
+"""
+
+
+def test_esxi_history(target_unix: Target, fs_unix: VirtualFilesystem) -> None:
+    fs_unix.map_file_fh("/var/run/log/shell.log", BytesIO(textwrap.dedent(COMMANDHISTORY_DATA).encode()))
+    target_unix.add_plugin(ESXICommandHistoryPlugin)
+
+    results = list(target_unix.commandhistory())
+    assert len(results) == 3

tests/plugins/os/unix/log/test_auth.py

@@ -39,6 +39,19 @@ def test_auth(target_unix: Target, fs_unix: VirtualFilesystem) -> None:
         assert results[-1].message == "pam_unix(cron:session): session opened for user root by (uid=0)"
 
 
+def test_auth_esxi(target_unix: Target, fs_unix: VirtualFilesystem) -> None:
+    fs_unix.map_file_fh("/etc/timezone", BytesIO(b"Europe/Amsterdam"))
+
+    data_path = "_data/plugins/os/unix/log/auth/esxi_auth.log"
+    data_file = absolute_path(data_path)
+    fs_unix.map_file("var/run/log/auth.log", data_file)
+
+    target_unix.add_plugin(AuthPlugin)
+    results = list(target_unix.authlog())
+
+    assert len(results) == 1
+
+
 def test_auth_with_gz(target_unix: Target, fs_unix: VirtualFilesystem) -> None:
     fs_unix.map_file_fh("/etc/timezone", BytesIO(b"Pacific/Honolulu"))