Add ProcFdPlugin for Linux process FD analysis

dissect/target/plugins/os/unix/linux/fd.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from dissect.target.exceptions import UnsupportedPluginError
+from dissect.target.helpers.record import TargetRecordDescriptor
+from dissect.target.plugin import Plugin, export
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+FileDescriptorRecord = TargetRecordDescriptor(
+    "linux/proc/fd",
+    [
+        ("datetime", "ts"),
+        ("varint", "pid"),
+        ("string", "name"),
+        ("varint", "fd"),
+        ("string", "link"),
+        ("varint", "pos"),
+        ("string", "flags"),
+    ],
+)
+
+class ProcFdPlugin(Plugin):
+    """Linux process file descriptor plugin."""
+
+    def check_compatible(self) -> None:
+        if not self.target.has_function("proc"):
+            raise UnsupportedPluginError("proc filesystem not available")
+
+    @export(record=FileDescriptorRecord)
+    def fd(self) -> Iterator[FileDescriptorRecord]:
+        """Return information about open file descriptors for all processes.
+
+        This plugin identifies files, sockets, pipes, and other artifacts
+        currently in use by processes by parsing /proc/[pid]/fd and fdinfo.
+
+        Yields FileDescriptorRecord with the following fields:
+
+        .. code-block:: text
+
+            ts (datetime): The modification timestamp of the fd directory.
+            pid (varint): The process id (pid) of the process.
+            name (string): The name associated to the pid.
+            fd (varint): The file descriptor number.
+            path (string): The resolved path or resource link.
+            pos (uint64): The current file offset from fdinfo.
+            flags (string): The access flags from fdinfo.
+        """
+        for process in self.target.proc.processes():
+            for fd_obj in process.fd():
+                try:
+                    ts = fd_obj.path.stat().st_mtime
+                except Exception:
+                    ts = None
+                yield FileDescriptorRecord(
+                    ts=ts,
+                    pid=process.pid,
+                    name=process.name,
+                    fd=fd_obj.number,
+                    link=fd_obj.link,
+                    pos=int(fd_obj.info.get("pos", 0)),
+                    flags=fd_obj.info.get("flags", "0"),
+                    _target=self.target,
+                )

dissect/target/plugins/os/unix/linux/proc.py

@@ -154,6 +154,48 @@ class Environ:
     variable: str
     contents: str
 
+class FileDescriptor:
+    def __init__(self, proc: Path, fd_num: str):
+        self.proc = proc
+        self.number = fd_num
+
+    @property
+    def path(self) -> Path:
+        """The path to the file descriptor symlink."""
+        return self.proc.joinpath(self.number)
+
+    @property
+    def info_path(self) -> Path:
+        """The path to the fdinfo entry."""
+        return self.proc.parent.joinpath("fdinfo", self.number)
+
+    @cached_property
+    def link(self) -> str:
+        """Returns the resolved symlink."""
+        try:
+            return  str(self.path.readlink())
+        except Exception:
+            return "unknown"
+
+    @cached_property
+    def info(self)-> dict[str, str]:
+        """Parsed key-value pairs from fdinfo."""
+        data = {}
+        if not self.info_path.exists():
+            return data
+
+        for line in self.info_path.read_text().split("\n"):
+            line = line.strip()
+            if not line:
+                continue
+
+            parts = line.split(None, 1)
+            if len(parts) != 2:
+                continue
+            key, value = parts
+            data[key] = value
+
+        return data
 
 class ProcessStateEnum(StrEnum):
     R = "Running"  # Running
@@ -488,6 +530,15 @@ def _parse_environ(self) -> Iterator[Environ]:
 
             yield Environ(variable, contents)
 
+    def _parse_fd(self) -> Iterator[FileDescriptor]:
+        """Internal function to parse entries in ``/proc/[pid]/fd/[fd_num]`` and ``/proc/[pid]/fdinfo/[fd_num]``."""
+        if not (fd_path := self.get("fd")).exists():
+            return
+
+        for fd_link in fd_path.iterdir():
+            yield FileDescriptor(fd_path, fd_link.name)
+
+
     @property
     def _boottime(self) -> int | None:
         """Returns the boot time of the system.
@@ -583,6 +634,10 @@ def environ(self) -> Iterator[Environ]:
         """Yields the content of the environ file associated with the process."""
         yield from self._parse_environ()
 
+    def fd(self) -> Iterator[FileDescriptor]:
+        """Yields the file descriptors associated with the process."""
+        yield from self._parse_fd()
+
     @property
     def uptime(self) -> timedelta:
         """Returns the uptime of the system from the moment it was acquired."""

tests/conftest.py

@@ -239,6 +239,12 @@ def fs_linux_proc(fs_linux: VirtualFilesystem) -> VirtualFilesystem:
             "",
         ),
     )
+    fdinfo_files_data = (
+        "pos\t0\nflags\t00000002\nsocket\t1337\n",
+        "pos\t0\nflags\t00000002\nsocket\t1338\n",
+        "pos\t0\nflags\t00000002\nsocket\t1339\n",
+        "pos\t0\nflags\t00000002\nsocket\t1337\n"
+    )
     stat_files_data = (
         "1 (systemd) S 0 1 1 0 -1 4194560 53787 457726 166 4255 112 260 761 548 20 0 1 0 30 184877056 2658 18446744073709551615 93937510957056 93937511789581 140726499200496 0 0 0 671173123 4096 1260 0 0 0 17 0 0 0 11 0 0 93937512175824 93937512476912 93937519890432 140726499204941 140726499204952 140726499204952 140726499205101 0\n",  # noqa
         "2 (kthread) K 1 1 1 0 -1 4194560 53787 457726 166 4255 112 260 761 548 20 0 1 0 30 184877056 2658 18446744073709551615 93937510957056 93937511789581 140726499200496 0 0 0 671173123 4096 1260 0 0 0 17 0 0 0 11 0 0 93937512175824 93937512476912 93937519890432 140726499204941 140726499204952 140726499204952 140726499205101 0\n",  # noqa
@@ -251,6 +257,7 @@ def fs_linux_proc(fs_linux: VirtualFilesystem) -> VirtualFilesystem:
         fs.makedirs(dir)
         fs.map_file_entry(fd.path, fd)
 
+        fs.map_file_fh(dir + "/fdinfo/" + fd.name, BytesIO(fdinfo_files_data[idx].encode()))
         fs.map_file_fh(dir + "/stat", BytesIO(stat_files_data[idx].encode()))
         fs.map_file_fh(dir + "/cmdline", BytesIO(cmdline.encode()))
         if environ:
@@ -259,7 +266,7 @@ def fs_linux_proc(fs_linux: VirtualFilesystem) -> VirtualFilesystem:
     # symlink acquire process to self
     fs.link("/proc/1337", "/proc/self")
 
-    # boottime and uptime are needed for for time tests
+    # boottime and uptime are needed for time tests
     fs.map_file_fh("/proc/uptime", BytesIO(b"134368.27 132695.52\n"))
     fs.map_file_fh("/proc/stat", BytesIO(b"btime 1680559854"))
 

tests/plugins/os/unix/linux/test_fd.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from dissect.target.plugins.os.unix.linux.proc import ProcPlugin
+
+if TYPE_CHECKING:
+    from dissect.target.filesystem import VirtualFilesystem
+    from dissect.target.target import Target
+
+
+def test_fd(target_linux_users: Target, fs_linux_proc: VirtualFilesystem) -> None:
+    target_linux_users.add_plugin(ProcPlugin)
+    results = list(target_linux_users.fd())
+    assert len(results) == 4

tests/plugins/os/unix/linux/test_proc.py

@@ -30,6 +30,11 @@ def test_process(target_linux_users: Target, fs_linux_proc: VirtualFilesystem) -
     assert environ[0].variable == "VAR"
     assert environ[0].contents == "1"
 
+    fd = list(process.fd())
+    assert "socket" in fd[0].link.name
+    assert fd[0].info["pos"] == "0"
+    assert fd[0].info["flags"] == "00000002"
+
 
 def test_process_not_found(target_linux_users: Target, fs_linux_proc: VirtualFilesystem) -> None:
     target_linux_users.add_plugin(ProcPlugin)
@@ -52,6 +57,11 @@ def test_processes(target_linux_users: Target, fs_linux_proc: VirtualFilesystem)
             assert env.variable == "VAR"
             assert env.contents == "1"
 
+        for fd in process.fd():
+            assert "socket" in fd.link.name
+            assert fd.info["pos"] == "0"
+            assert fd.info["flags"] == "00000002"
+
 
 def test_processes_without_boottime(target_linux_users: Target, fs_linux_proc: VirtualFilesystem) -> None:
     target_linux_users.add_plugin(ProcPlugin)
@@ -70,6 +80,11 @@ def test_processes_without_boottime(target_linux_users: Target, fs_linux_proc: V
                 assert env.variable == "VAR"
                 assert env.contents == "1"
 
+            for fd in process.fd():
+                assert "socket" in fd.link.name
+                assert fd.info["pos"] == "0"
+                assert fd.info["flags"] == "00000002"
+
 
 def test_proc_plugin_incompatible(target_linux_users: Target, fs_linux: VirtualFilesystem) -> None:
     with pytest.raises(UnsupportedPluginError, match="No /proc directory found"):