Add file descriptor and fdinfo parsing to Linux

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_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"):