Add support for ESXI auth, hostd and shell logs.

.gitignore

@@ -9,3 +9,4 @@ __pycache__/
 tests/_docs/api
 tests/_docs/build
 .tox/
+uv.lock

dissect/target/plugins/os/unix/esxi/esxi_log/__init__.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+import re
+import typing
+from datetime import datetime
+from re import Pattern
+
+from dissect.target.helpers.fsutil import open_decompress
+from dissect.target.helpers.record import TargetRecordDescriptor
+
+if typing.TYPE_CHECKING:
+    from collections.abc import Iterator
+    from pathlib import Path
+
+    from dissect.target import Target
+
+ESXiLogRecord = TargetRecordDescriptor(
+    "esxi/log",
+    [
+        ("datetime", "ts"),
+        ("string", "type"),
+        ("string", "log_level"),
+        ("string", "application"),
+        ("varint", "pid"),
+        ("string", "op_id"),
+        ("string", "user"),
+        ("string", "event_metadata"),
+        ("string", "message"),
+        ("path", "source"),
+    ],
+)
+
+RE_LOG_FORMAT: Pattern = re.compile(
+    r"""
+    ((?P<ts>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d{3})?Z)\s)? # ts, moslty including milliseconds, but not always
+    (
+        ((?P<log_level>[\w()]+)\s)? # info, warning, of In(166), Wa(164), Er(163) in esxi8+, sometime missing
+        ((?P<application>(\w+|-))\[(?P<pid>(\d+))\]|-):?\s  # hostd[pid] < esxi8, Hostd[pid]: esxi8+
+
+    )?
+   (?P<newline_delimiter>--> ?)? # in Exi8+, newline marker is positionned after the ts loglevel application part
+   # but for some log this marker is missing...
+   (\[(?P<metadata>(.*?))\]\s)?
+   (?P<message>.*?)""",
+    re.VERBOSE,
+)
+
+
+def get_esxi_log_path(target: Target, logname: str) -> Iterator[Path]:
+    """
+    Get log location, looking in most usual location, as well as in the osdata partition
+
+    References:
+        - https://knowledge.broadcom.com/external/article/306962/location-of-esxi-log-files.html
+    :return:
+    """
+    # Esxi/loaders should ensure that logs are symlinked to /var/run/log, as on a live ESXi hosts.
+    if (var_run_log := target.fs.path("/var/run/log")).exists():
+        print("HERE")
+        for path in var_run_log.glob(f"{logname}.*"):
+            try:
+                yield path.resolve(strict=True)
+            except FilesystemError as e:  # noqa PERF203
+                target.info.warning("Fail to resolve path to %s : %s", path, str(e))
+    return
+
+
+def yield_log_records(
+    target: Target, log_paths: list[Path], re_log_format: re.Pattern, logname: str
+) -> Iterator[ESXiLogRecord]:
+    """Yield parsed log entries, iterate on identified log files"""
+    for path in log_paths:
+        try:
+            current_record = None
+            for line in open_decompress(path, "rt"):
+                if not line:
+                    continue
+                line = line.strip("\n")
+                if match := re_log_format.fullmatch(line):
+                    log = match.groupdict()
+                    # For multiline event, line start with --> Before Esxi8
+                    # For Esxi8+, --> is after the Date loglevel application[pid] block
+                    # but sometime --> is missing but it's still previous line continuation
+                    if log.get("newline_delimiter") == "-->" or log.get("ts") is None:
+                        if current_record:
+                            current_record.message = current_record.message + "\n" + log.get("message")
+                        else:
+                            target.log.warning("log file contains unrecognized format in %s", path)
+                            target.log.debug("log file contains unrecognized format in %s : %s", path, line)
+                    else:
+                        if current_record:
+                            yield current_record
+                        current_record = None
+                        if metadata := log.get("metadata"):
+                            user = re.search(r"user=(\S+)", metadata)
+                            op_id = re.search(r"opID=(\S+)", metadata)
+                        else:
+                            user = None
+                            op_id = None
+                        ts = log["ts"]
+                        current_record = ESXiLogRecord(
+                            _target=target,
+                            type=logname,
+                            message=log.get("message", ""),
+                            log_level=log.get("log_level", None),
+                            application=log.get("application", None),
+                            pid=log.get("pid", None),
+                            user=None if user is None else user.groups()[0],
+                            op_id=None if op_id is None else op_id.groups()[0],
+                            event_metadata=log.get("metadata", ""),
+                            ts=datetime.strptime(ts, "%Y-%m-%dT%H:%M:%S.%f%z")
+                            if "." in ts
+                            else datetime.strptime(ts, "%Y-%m-%dT%H:%M:%S%z"),
+                            source=path,
+                        )
+                else:
+                    target.log.warning("log file contains unrecognized format in %s", path)
+                    target.log.debug("log file contains unrecognized format in %s : %s", path, line)
+                    continue
+            if current_record:
+                yield current_record
+        except Exception as e:
+            target.log.warning("An error occurred parsing %s log file %s: %s", logname, path, str(e))
+            target.log.debug("", exc_info=e)

dissect/target/plugins/os/unix/esxi/esxi_log/auth.py

@@ -0,0 +1,37 @@
+from collections.abc import Iterator
+from pathlib import Path
+
+from dissect.target import Target
+from dissect.target.exceptions import UnsupportedPluginError
+from dissect.target.plugin import OperatingSystem, Plugin, export
+from dissect.target.plugins.os.unix.esxi.esxi_log import (
+    RE_LOG_FORMAT,
+    ESXiLogRecord,
+    get_esxi_log_path,
+    yield_log_records,
+)
+
+
+class EsxiAuthPlugin(Plugin):
+    """ESXi auth.log plugins"""
+
+    def __init__(self, target: Target):
+        super().__init__(target)
+        self.log_paths: list[Path] = list(self.get_paths())
+
+    def check_compatible(self) -> None:
+        # Log path as the same as on other unix target, so we fail fast
+        if not self.target.os == OperatingSystem.ESXI:
+            raise UnsupportedPluginError("Not an ESXi host")
+        if not len(self.log_paths):
+            raise UnsupportedPluginError("No auth logs found")
+
+    def _get_paths(self) -> Iterator[Path]:
+        yield from get_esxi_log_path(self.target, "auth")
+
+    @export(record=ESXiLogRecord)
+    def auth(self) -> Iterator[ESXiLogRecord]:
+        """
+        Records for auth log file (ESXi Shell authentication success and failure.) Seems to be empty in ESXi8+
+        """
+        yield from yield_log_records(self.target, self.log_paths, RE_LOG_FORMAT, "auth")

dissect/target/plugins/os/unix/esxi/esxi_log/hostd.py

@@ -0,0 +1,37 @@
+from collections.abc import Iterator
+from pathlib import Path
+
+from dissect.target import Target
+from dissect.target.exceptions import UnsupportedPluginError
+from dissect.target.plugin import OperatingSystem, Plugin, export
+from dissect.target.plugins.os.unix.esxi.esxi_log import (
+    RE_LOG_FORMAT,
+    ESXiLogRecord,
+    get_esxi_log_path,
+    yield_log_records,
+)
+
+
+class HostdPlugin(Plugin):
+    """ESXi hostd logs plugins"""
+
+    def __init__(self, target: Target):
+        super().__init__(target)
+        self.log_paths: list[Path] = list(self.get_paths())
+
+    def check_compatible(self) -> None:
+        # Log path as the same as on other unix target, so we fail fast
+        if not self.target.os == OperatingSystem.ESXI:
+            raise UnsupportedPluginError("Not an ESXi host")
+        if not len(self.log_paths):
+            raise UnsupportedPluginError("No hostd logs found")
+
+    def _get_paths(self) -> Iterator[Path]:
+        yield from get_esxi_log_path(self.target, "hostd")
+
+    @export(record=ESXiLogRecord)
+    def hostd(self) -> Iterator[ESXiLogRecord]:
+        """
+        Records for hostd log file (Host management service logs, including virtual machine and host Task and Events)
+        """
+        yield from yield_log_records(self.target, self.log_paths, RE_LOG_FORMAT, "hostd")

dissect/target/plugins/os/unix/esxi/esxi_log/shell_log.py

@@ -0,0 +1,59 @@
+import re
+from collections.abc import Iterator
+from pathlib import Path
+
+from dissect.target import Target
+from dissect.target.exceptions import UnsupportedPluginError
+from dissect.target.plugin import OperatingSystem, Plugin, export
+from dissect.target.plugins.os.unix.esxi.esxi_log import (
+    ESXiLogRecord,
+    get_esxi_log_path,
+    yield_log_records,
+)
+
+
+class ShellLogPlugin(Plugin):
+    """ESXi shell.log plugins"""
+
+    # Mostly equal to EsxiLogBasePlugin.RE_LOG_FORMAT, but some difference in metadata part
+    RE_LOG_FORMAT: re.Pattern = re.compile(
+        r"""
+        ((?P<ts>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d{3})?Z)\s)? # ts, moslty including milliseconds, but not always
+        (
+            ((?P<log_level>[\w()]+)\s)? # info, warning, of In(166), Wa(164), Er(163) in esxi8+, sometime missing
+            ((?P<application>(\w+))\[(?P<pid>(\d+))\]):?\s  # hostd[pid] < esxi8, Hostd[pid]: esxi8+
+
+        )?
+       (?P<newline_delimiter>--> ?)? # in Exi8+, newline marker is positionned after the ts loglevel application part
+       # but for some log this marker is missing...
+       (\[(?P<metadata>(.+?))\]:\s)? # Metadata = user. Instead of \s, metadata is followed by a ":"
+       (?P<message>.*?)""",
+        re.VERBOSE,
+    )
+
+    def __init__(self, target: Target):
+        super().__init__(target)
+        self.log_paths: list[Path] = list(self.get_paths())
+
+    def check_compatible(self) -> None:
+        # Log path as the same as on other unix target, so we fail fast
+        if not self.target.os == OperatingSystem.ESXI:
+            raise UnsupportedPluginError("Not an ESXi host")
+        if not len(self.log_paths):
+            raise UnsupportedPluginError("No log found")
+
+    def _get_paths(self) -> Iterator[Path]:
+        yield from get_esxi_log_path(self.target, "shell")
+
+    @export(record=ESXiLogRecord)
+    def shell_log(self) -> Iterator[ESXiLogRecord]:
+        """
+        Records for shell.log files (ESXi Shell usage logs, including enable/disable and every command entered).
+
+        References:
+            - https://knowledge.broadcom.com/external/article/321910
+        """
+        for record in yield_log_records(self.target, self.log_paths, self.RE_LOG_FORMAT, "shell"):
+            record.user = record.event_metadata
+            record.event_metadata = None
+            yield record

tests/conftest.py

@@ -23,6 +23,7 @@
 from dissect.target.plugins.os.unix.bsd.citrix._os import CitrixPlugin
 from dissect.target.plugins.os.unix.bsd.darwin.ios._os import IOSPlugin
 from dissect.target.plugins.os.unix.bsd.darwin.macos._os import MacOSPlugin
+from dissect.target.plugins.os.unix.esxi._os import ESXiPlugin
 from dissect.target.plugins.os.unix.linux._os import LinuxPlugin
 from dissect.target.plugins.os.unix.linux.android._os import AndroidPlugin
 from dissect.target.plugins.os.unix.linux.debian._os import DebianPlugin
@@ -38,7 +39,6 @@
 
     from dissect.target.plugin import OSPlugin
 
-
 HAS_BENCHMARK = importlib.util.find_spec("pytest_benchmark") is not None
 
 
@@ -447,6 +447,11 @@ def target_android(tmp_path: pathlib.Path, fs_android: Filesystem) -> Target:
     return make_os_target(tmp_path, AndroidPlugin, root_fs=fs_android)
 
 
+@pytest.fixture
+def target_esxi(tmp_path: pathlib.Path, fs_esxi: Filesystem) -> Target:
+    return make_os_target(tmp_path, ESXiPlugin, root_fs=fs_esxi)
+
+
 def add_win_user(hive_hklm: VirtualHive, hive_hku: VirtualHive, target_win: Target, sid: str, home: str) -> None:
     """Add a user to the provided Windows target."""