Refactor unix cronjob plugin

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

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import re
-from typing import Iterator
+from pathlib import Path
+from typing import Iterator, get_args
 
+from dissect.target.exceptions import UnsupportedPluginError
 from dissect.target.helpers.record import TargetRecordDescriptor
 from dissect.target.plugin import Plugin, export
 
@@ -29,82 +31,116 @@
     ],
 )
 
+RE_CRONJOB = re.compile(
+    r"""
+        ^
+        (?P<minute>\S+)
+        \s+
+        (?P<hour>\S+)
+        \s+
+        (?P<day>\S+)
+        \s+
+        (?P<month>\S+)
+        \s+
+        (?P<weekday>\S+)
+        \s+
+        (?P<command>.+)
+        $
+    """,
+    re.VERBOSE,
+)
+RE_ENVVAR = re.compile(
+    r"""
+        ^
+        ([a-zA-Z_]+[a-zA-Z[0-9_])=(.*)
+    """,
+    re.VERBOSE,
+)
+
 
 class CronjobPlugin(Plugin):
     """Unix cronjob plugin."""
 
+    CRONTAB_DIRS = [
+        "/var/cron/tabs",
+        "/var/spool/cron",
+        "/var/spool/cron/crontabs",
+        "/etc/cron.d",
+        "/usr/local/etc/cron.d",  # FreeBSD
+    ]
+
+    CRONTAB_FILES = [
+        "/etc/crontab",
+        "/etc/anacrontab",
+    ]
+
+    def __init__(self, target):
+        super().__init__(target)
+        self.crontabs = list(self.find_crontabs())
+
     def check_compatible(self) -> None:
-        pass
+        if not self.crontabs:
+            raise UnsupportedPluginError("No crontab(s) found on target")
 
-    def parse_crontab(self, file_path) -> Iterator[CronjobRecord | EnvironmentVariableRecord]:
-        for line in file_path.open("rt"):
-            line = line.strip()
-            if line.startswith("#") or not len(line):
+    def find_crontabs(self) -> Iterator[Path]:
+        for crontab_dir in self.CRONTAB_DIRS:
+            if not (dir := self.target.fs.path(crontab_dir)).exists():
                 continue
 
-            match = re.search(r"^(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(.*)$", line)
-            if match:
-                usr = match.group(6)
-                cmd = match.group(7)
-                if not str(file_path).startswith("/etc/crontab") or not str(file_path).startswith("/etc/cron.d"):
-                    cmd = usr + " " + cmd
-                    usr = ""
-
-                yield CronjobRecord(
-                    minute=match.group(1),
-                    hour=match.group(2),
-                    day=match.group(3),
-                    month=match.group(4),
-                    weekday=match.group(5),
-                    user=usr,
-                    command=cmd,
-                    source=file_path,
-                    _target=self.target,
-                )
-
-            s = re.search(r"^([a-zA-Z_]+[a-zA-Z[0-9_])=(.*)", line)
-            if s:
-                yield EnvironmentVariableRecord(
-                    key=s.group(1),
-                    value=s.group(2),
-                    source=file_path,
-                    _target=self.target,
-                )
-
-    @export(record=[CronjobRecord, EnvironmentVariableRecord])
+            for file in dir.iterdir():
+                if file.resolve().is_file():
+                    yield file
+
+        for crontab_file in self.CRONTAB_FILES:
+            if (file := self.target.fs.path(crontab_file)).exists():
+                yield file
+
+    @export(record=get_args([CronjobRecord, EnvironmentVariableRecord]))
     def cronjobs(self) -> Iterator[CronjobRecord | EnvironmentVariableRecord]:
-        """Yield cronjobs on the unix system.
+        """Yield cronjobs on a unix system.
 
         A cronjob is a scheduled task/command on a Unix based system. Adversaries may use cronjobs to gain
         persistence on the system.
+
+        Resources:
+            - https://linux.die.net/man/8/cron
+            - https://linux.die.net/man/1/crontab
+            - https://linux.die.net/man/5/crontab
+            - https://en.wikipedia.org/wiki/Cron
         """
-        tabs = []
-        crontab_dirs = [
-            "/var/cron/tabs",
-            "/var/spool/cron",
-            "/var/spool/cron/crontabs",
-            "/etc/cron.d",
-            "/usr/local/etc/cron.d",  # FreeBSD
-        ]
-        for path in crontab_dirs:
-            fspath = self.target.fs.path(path)
-            if not fspath.exists():
-                continue
 
-            for f in fspath.iterdir():
-                if not f.exists():
+        for file in self.crontabs:
+            for line in file.open("rt"):
+                line = line.strip()
+                if line.startswith("#") or not len(line):
                     continue
-                if f.is_file():
-                    tabs.append(f)
-
-        crontab_file = self.target.fs.path("/etc/crontab")
-        if crontab_file.exists():
-            tabs.append(crontab_file)
-
-        crontab_file = self.target.fs.path("/etc/anacrontab")
-        if crontab_file.exists():
-            tabs.append(crontab_file)
 
-        for f in tabs:
-            for record in self.parse_crontab(f):
-                yield record
+                if match := RE_CRONJOB.search(line):
+                    match = match.groupdict()
+
+                    # Cronjobs in user crontab files do not have a user field specified.
+                    user = None
+                    if file.is_relative_to("/var/spool/cron/crontabs"):
+                        user = file.name
+                    else:
+                        user, match["command"] = re.split(r"\s", match["command"], maxsplit=1)
+
+                    match["command"] = match["command"].strip()
+
+                    yield CronjobRecord(
+                        **match,
+                        user=user,
+                        source=file,
+                        _target=self.target,
+                    )
+
+                elif match := RE_ENVVAR.search(line):
+                    yield EnvironmentVariableRecord(
+                        key=match.group(1),
+                        value=match.group(2),
+                        source=file,
+                        _target=self.target,
+                    )
+
+                else:
+                    self.target.log.warning("Unable to match cronjob line in %s: '%s'", file, line)

tests/plugins/os/unix/test_cronjobs.py

@@ -0,0 +1,39 @@
+from io import BytesIO
+
+from dissect.target.filesystem import VirtualFilesystem
+from dissect.target.plugins.os.unix.cronjobs import CronjobPlugin
+from dissect.target.target import Target
+
+
+def test_unix_cronjobs_system(target_unix_users: Target, fs_unix: VirtualFilesystem) -> None:
+    """test if we correctly infer the username of the cronjob from the command."""
+
+    fs_unix.map_file_fh("/etc/crontab", BytesIO(b"17 *	* * *	root	cd / && run-parts --report /etc/cron.hourly"))
+    target_unix_users.add_plugin(CronjobPlugin)
+
+    results = list(target_unix_users.cronjobs())
+    assert len(results) == 1
+    assert results[0].minute == "17"
+    assert results[0].hour == "*"
+    assert results[0].day == "*"
+    assert results[0].month == "*"
+    assert results[0].weekday == "*"
+    assert results[0].user == "root"
+    assert results[0].command == "cd / && run-parts --report /etc/cron.hourly"
+
+
+def test_unix_cronjobs_user(target_unix_users: Target, fs_unix: VirtualFilesystem) -> None:
+    """test if we correctly infer the username of the crontab from the file path."""
+
+    fs_unix.map_file_fh("/var/spool/cron/crontabs/user", BytesIO(b"0 0 * * * /path/to/example.sh\n"))
+    target_unix_users.add_plugin(CronjobPlugin)
+
+    results = list(target_unix_users.cronjobs())
+    assert len(results) == 1
+    assert results[0].minute == "0"
+    assert results[0].hour == "0"
+    assert results[0].day == "*"
+    assert results[0].month == "*"
+    assert results[0].weekday == "*"
+    assert results[0].user == "user"
+    assert results[0].command == "/path/to/example.sh"