diff --git a/dissect/target/plugins/child/vmware_vcenter.py b/dissect/target/plugins/child/vmware_vcenter.py new file mode 100644 index 0000000000..7f26d7d962 --- /dev/null +++ b/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, + ) diff --git a/dissect/target/plugins/os/unix/bsd/citrix/history.py b/dissect/target/plugins/os/unix/bsd/citrix/history.py index d0cb1c5a74..041534470f 100644 --- a/dissect/target/plugins/os/unix/bsd/citrix/history.py +++ b/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: diff --git a/dissect/target/plugins/os/unix/esxi/history.py b/dissect/target/plugins/os/unix/esxi/history.py new file mode 100644 index 0000000000..62739b5dbe --- /dev/null +++ b/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\S+)\s+" + r"(?P[^\[:]+)" + r"(?:\[(?P\d+)\])?" + r":\s+" + r"(?:\[(?P[^\]]+)\]:\s*)?" + r"(?P.+)$" +) + + +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 diff --git a/dissect/target/plugins/os/unix/history.py b/dissect/target/plugins/os/unix/history.py index 2acae047e4..352fc9fd35 100644 --- a/dissect/target/plugins/os/unix/history.py +++ b/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) diff --git a/dissect/target/plugins/os/unix/log/auth.py b/dissect/target/plugins/os/unix/log/auth.py index 607b718642..311a46647f 100644 --- a/dissect/target/plugins/os/unix/log/auth.py +++ b/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"])) diff --git a/dissect/target/plugins/os/unix/log/helpers.py b/dissect/target/plugins/os/unix/log/helpers.py index ec7189075b..6b7f0490fa 100644 --- a/dissect/target/plugins/os/unix/log/helpers.py +++ b/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 diff --git a/tests/_data/plugins/os/unix/log/auth/esxi_auth.log b/tests/_data/plugins/os/unix/log/auth/esxi_auth.log new file mode 100644 index 0000000000..0ec5acb7f2 --- /dev/null +++ b/tests/_data/plugins/os/unix/log/auth/esxi_auth.log @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:70fcdd5e6193d7ebf395249cb006b83ee9ffbd1d069614c7de379ffb9b09089c +size 99 diff --git a/tests/plugins/child/test_vmware_vcenter.py b/tests/plugins/child/test_vmware_vcenter.py new file mode 100644 index 0000000000..2d0b4b5e20 --- /dev/null +++ b/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" diff --git a/tests/plugins/os/unix/esxi/test_history.py b/tests/plugins/os/unix/esxi/test_history.py new file mode 100644 index 0000000000..7b89ba766f --- /dev/null +++ b/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 diff --git a/tests/plugins/os/unix/log/test_auth.py b/tests/plugins/os/unix/log/test_auth.py index 981952e2af..707ece0571 100644 --- a/tests/plugins/os/unix/log/test_auth.py +++ b/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"))