diff --git a/dissect/target/plugins/os/unix/anacronjobs.py b/dissect/target/plugins/os/unix/anacronjobs.py new file mode 100644 index 0000000000..85efdacc2d --- /dev/null +++ b/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\S+) + \s+ + (?P\S+) + \s+ + (?P\S+) + \s+ + (?P.+) + $ + """, + 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[a-zA-Z_]+[a-zA-Z[0-9_])\s*=(?P.*)") + + +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) diff --git a/dissect/target/plugins/os/unix/cronjobs.py b/dissect/target/plugins/os/unix/cronjobs.py index 6f946d149b..3e40c0fe57 100644 --- a/dissect/target/plugins/os/unix/cronjobs.py +++ b/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 """ diff --git a/tests/plugins/os/unix/test_anacronjobs.py b/tests/plugins/os/unix/test_anacronjobs.py new file mode 100644 index 0000000000..39237f595e --- /dev/null +++ b/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