Add anacronjobs plugins (unix)

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

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+import datetime
+import re
+import shlex
+from typing import TYPE_CHECKING
+
+from dissect.util import ts
+
+from dissect.target.exceptions import UnsupportedPluginError
+from dissect.target.helpers.record import (
+    TargetRecordDescriptor,
+)
+from dissect.target.plugin import Plugin, export
+from dissect.target.plugins.os.unix.cronjobs import EnvironmentVariableRecord
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from pathlib import Path
+
+    from dissect.target.target import Target
+
+AnacronjobRecord = TargetRecordDescriptor(
+    "unix/anacronjob",
+    [
+        ("string", "period_name"),
+        ("varint", "delay_in_minutes"),
+        ("string", "job_identify"),
+        ("string", "command"),
+        ("datetime", "ts_last_exec"),  # based on /var/spool/anacron/job_identify content and last modification time
+        ("path", "source"),
+    ],
+)
+
+RE_ANACRONJOB = re.compile(
+    r"""
+        ^
+        (?P<period_name>\S+)
+        \s+
+        (?P<delay>\S+)
+        \s+
+        (?P<job_identify>\S+)
+        \s+
+        (?P<command>.+)
+        $
+    """,
+    re.VERBOSE,
+)
+
+
+# From man :
+# ``Spaces around VAR are removed. No spaces around VALUE are allowed (unless you want them to be part of the value).``
+
+RE_ENVVAR = re.compile(r"^\s*(?P<key>[a-zA-Z_]+[a-zA-Z[0-9_])\s*=(?P<value>.*)")
+
+
+class AnacronjobPlugin(Plugin):
+    """Unix anacron plugin."""
+
+    ANACRONTAB_FILES = (
+        "/etc/anacrontab",  # Linux
+        "/usr/local/etc/anacrontab",  # FreeBSD
+    )
+
+    def __init__(self, target: Target):
+        super().__init__(target)
+        self.anacrontabs = list(self.get_paths())
+
+    def check_compatible(self) -> None:
+        if not self.anacrontabs:
+            raise UnsupportedPluginError("No anacrontab found on target")
+
+    def _get_paths(self) -> Iterator[Path]:
+        for anacrontab_file in self.ANACRONTAB_FILES:
+            if (file := self.target.fs.path(anacrontab_file)).exists():
+                yield file
+
+    @export(record=[AnacronjobRecord, EnvironmentVariableRecord])
+    def anacronjobs(self) -> Iterator[AnacronjobRecord | EnvironmentVariableRecord]:
+        """Yield anacron jobs, and their configured environment variables on a Unix system.
+
+        An anacron job is a scheduled task/command on a Unix based system. Adversaries may use anacronjobs to gain
+        persistence on the system. This plugins also iterate over files executed using run-parts
+
+        References:
+            - https://linux.die.net/man/5/anacrontab
+            - https://man.freebsd.org/cgi/man.cgi?anacron(8)
+            - https://linux.die.net/man/8/anacron
+        """
+        for file in self.anacrontabs:
+            for line in file.open("rt"):
+                line = line.strip()
+                ts_last_exec = None
+                if line.startswith("#") or not line:
+                    continue
+
+                if match := RE_ANACRONJOB.search(line):
+                    match = match.groupdict()
+                    job_identify = match.get("job_identify", None)
+                    command = match.get("command", None)
+                    if (ts_file := self.target.fs.path(f"/var/spool/anacron/{job_identify}")).exists():
+                        ts_file_stat = ts_file.stat()
+                        ts_last_exec = ts.from_unix(ts_file_stat.st_mtime)
+                        anacron_ts_value = ts_file.read_text(errors="backslashreplace").strip()
+                        if ts_last_exec.strftime("%Y%m%d") != anacron_ts_value:
+                            # incoherent value, maybe related to ts modification/data loss during collection
+                            ts_last_exec = datetime.datetime.strptime(anacron_ts_value, "%Y%m%d")  # noqa: DTZ007
+                    yield AnacronjobRecord(
+                        period_name=match.get("period_name", None),
+                        delay_in_minutes=match.get("delay", None),
+                        job_identify=job_identify,
+                        command=command,
+                        ts_last_exec=ts_last_exec,
+                        source=file,
+                        _target=self.target,
+                    )
+                    if command:
+                        splited = shlex.split(command)
+                        # Anacron often use run-parts or nice run-parts to run a list of script in a directory
+                        # Last part of the command is the name of the folder
+                        if (
+                            "run-parts" in splited[:2]
+                            and (run_part_dir := self.target.fs.path(splited[-1])).exists()
+                            and run_part_dir.is_dir()
+                        ):
+                            for f in run_part_dir.iterdir():
+                                yield AnacronjobRecord(
+                                    period_name=match.get("period_name", None),
+                                    delay_in_minutes=match.get("delay", None),
+                                    job_identify=job_identify,
+                                    command=f,
+                                    ts_last_exec=ts_last_exec,
+                                    source=file,
+                                    _target=self.target,
+                                )
+
+                # Anacron allows for Environment assignment
+                elif match := RE_ENVVAR.search(line):
+                    match = match.groupdict()
+                    yield EnvironmentVariableRecord(
+                        **match,
+                        source=file,
+                        _target=self.target,
+                    )
+
+                else:
+                    self.target.log.warning("Unable to match anacronjob line in %s: '%s'", file, line)

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

@@ -72,20 +72,17 @@ class CronjobPlugin(Plugin):
         "/usr/local/etc/cron.d",  # FreeBSD
     )
 
-    CRONTAB_FILES = (
-        "/etc/crontab",
-        "/etc/anacrontab",
-    )
+    CRONTAB_FILES = ("/etc/crontab",)
 
     def __init__(self, target: Target):
         super().__init__(target)
-        self.crontabs = list(self.find_crontabs())
+        self.crontabs = list(self.get_paths())
 
     def check_compatible(self) -> None:
         if not self.crontabs:
             raise UnsupportedPluginError("No crontab(s) found on target")
 
-    def find_crontabs(self) -> Iterator[Path]:
+    def _get_paths(self) -> Iterator[Path]:
         for crontab_dir in self.CRONTAB_DIRS:
             if not (dir := self.target.fs.path(crontab_dir)).exists():
                 continue
@@ -110,7 +107,6 @@ def cronjobs(self) -> Iterator[CronjobRecord | EnvironmentVariableRecord]:
             - https://linux.die.net/man/1/crontab
             - https://linux.die.net/man/5/crontab
             - https://en.wikipedia.org/wiki/Cron
-            - https://linux.die.net/man/8/anacron
             - https://manpages.ubuntu.com/manpages/oracular/en/man5/crontab.5.html
             - https://www.gnu.org/software/mcron/manual/mcron.html#Guile-Syntax
         """

tests/plugins/os/unix/test_anacronjobs.py

@@ -0,0 +1,122 @@
+from __future__ import annotations
+
+import stat
+import textwrap
+from io import BytesIO
+from typing import TYPE_CHECKING
+from unittest.mock import Mock
+
+from flow.record.fieldtypes import datetime as dt
+
+from dissect.target.filesystem import VirtualFile
+from dissect.target.helpers import fsutil
+from dissect.target.plugins.os.unix.anacronjobs import AnacronjobPlugin, AnacronjobRecord, EnvironmentVariableRecord
+
+if TYPE_CHECKING:
+    from dissect.target.filesystem import VirtualFilesystem
+    from dissect.target.target import Target
+
+
+ANACRONTAB_DEFAULT = """
+ # environment variables
+ SHELL=/bin/sh
+ PATH=/sbin:/bin:/usr/sbin:/usr/bin
+ MAILTO=root
+ RANDOM_DELAY=30
+ # the jobs will be started during the following hours only
+ START_HOURS_RANGE=3-22
+ # delay will be 5 minutes + RANDOM_DELAY for cron.daily
+ 1         5    cron.daily          nice run-parts /etc/cron.daily
+ 7         0    cron.weekly         nice run-parts /etc/cron.weekly
+ @monthly  0    cron.monthly        nice run-parts /etc/cron.monthly
+"""
+
+
+def test_unix_anacrontab(target_unix_users: Target, fs_unix: VirtualFilesystem) -> None:
+    fs_unix.map_file_fh("/etc/anacrontab", BytesIO(textwrap.dedent(ANACRONTAB_DEFAULT).encode()))
+    fs_unix.map_file_fh("/var/spool/anacron/cron.daily", BytesIO(b"20260206\n"))
+    # File to tests enumeration of files references in run-parts commands
+    fs_unix.map_file_fh("/etc/cron.daily/logrotate", BytesIO(b"#!/bin/sh\n/usr/sbin/logrotate\n"))
+
+    # Virtual file
+    # is set according to file birth and modification time
+    weekly_virtual_file = VirtualFile(fs_unix, "/var/spool/anacron/cron.weekly", BytesIO(b"20260119\n"))
+    weekly_virtual_file.lstat = Mock()
+    weekly_virtual_file.lstat.return_value = fsutil.stat_result(
+        [
+            stat.S_IFREG,
+            0,
+            0,
+            1,
+            0,
+            0,
+            9,
+            int(dt("2026-01-19 03:01:01+00:00").timestamp()),
+            int(dt("2026-01-19 02:12:17+00:00").timestamp()),
+            int(dt("2026-01-19 02:12:17+00:00").timestamp()),
+        ]
+    )
+    fs_unix.map_file_entry("/var/spool/anacron/cron.weekly", weekly_virtual_file)
+
+    target_unix_users.add_plugin(AnacronjobPlugin)
+
+    results = list(target_unix_users.anacronjobs())
+
+    assert len(results) == 9
+
+    anacronjob_records = [r for r in results if isinstance(r, type(AnacronjobRecord()))]
+    environmentvariable_records = [r for r in results if isinstance(r, type(EnvironmentVariableRecord()))]
+
+    assert len(anacronjob_records) == 4
+    assert len(environmentvariable_records) == 5
+
+    assert anacronjob_records[0].period_name == "1"
+    assert anacronjob_records[0].delay_in_minutes == 5
+    assert anacronjob_records[0].job_identify == "cron.daily"
+    assert anacronjob_records[0].command == "nice run-parts /etc/cron.daily"
+    assert anacronjob_records[0].ts_last_exec == dt("2026-02-06 00:00:00+00:00")
+    assert anacronjob_records[0].source == "/etc/anacrontab"
+    assert anacronjob_records[0].hostname == "localhost"
+
+    assert anacronjob_records[1].period_name == "1"
+    assert anacronjob_records[1].delay_in_minutes == 5
+    assert anacronjob_records[1].job_identify == "cron.daily"
+    assert anacronjob_records[1].command == "/etc/cron.daily/logrotate"
+    # If ts associated to timestamps files are not consistent with date in file, fallback to file content date
+    assert anacronjob_records[1].ts_last_exec == dt("2026-02-06 00:00:00+00:00")
+    assert anacronjob_records[1].source == "/etc/anacrontab"
+    assert anacronjob_records[1].hostname == "localhost"
+
+    assert anacronjob_records[2].period_name == "7"
+    assert anacronjob_records[2].delay_in_minutes == 0
+    assert anacronjob_records[2].job_identify == "cron.weekly"
+    assert anacronjob_records[2].command == "nice run-parts /etc/cron.weekly"
+    assert anacronjob_records[2].ts_last_exec == dt("2026-01-19 02:12:17+00:00")
+
+    assert anacronjob_records[3].period_name == "@monthly"
+    assert anacronjob_records[3].delay_in_minutes == 0
+    assert anacronjob_records[3].job_identify == "cron.monthly"
+    assert anacronjob_records[3].command == "nice run-parts /etc/cron.monthly"
+
+    assert anacronjob_records[3].ts_last_exec is None
+
+    # Test env variable
+    assert environmentvariable_records[0].key == "SHELL"
+    assert environmentvariable_records[0].value == "/bin/sh"
+    assert environmentvariable_records[0].source == "/etc/anacrontab"
+    assert environmentvariable_records[0].hostname == "localhost"
+
+    assert environmentvariable_records[1].key == "PATH"
+    assert environmentvariable_records[1].value == "/sbin:/bin:/usr/sbin:/usr/bin"
+
+    assert environmentvariable_records[4].key == "START_HOURS_RANGE"
+    assert environmentvariable_records[4].value == "3-22"
+
+
+def test_freebsd_anacrontab(target_unix_users: Target, fs_unix: VirtualFilesystem) -> None:
+    """Test that anacrontab path on freebsd is also identified."""
+    fs_unix.map_file_fh("/usr/local/etc/anacrontab", BytesIO(textwrap.dedent(ANACRONTAB_DEFAULT).encode()))
+    target_unix_users.add_plugin(AnacronjobPlugin)
+
+    results = list(target_unix_users.anacronjobs())
+    assert len(results) == 8