Add xattr, suid and capability functionality to WalkFsPlugin (#1144)

JSCU-CNI · web-flow · commit 69c908a42bf3 · 2025-10-30T16:07:31.000+01:00
diff --git a/dissect/target/plugins/filesystem/unix/capability.py b/dissect/target/plugins/filesystem/unix/capability.py
@@ -11,10 +11,13 @@
 if TYPE_CHECKING:
     from collections.abc import Iterator
 
+    from dissect.target.filesystem import FilesystemEntry
+    from dissect.target.target import Target
+
 CapabilityRecord = TargetRecordDescriptor(
     "filesystem/unix/capability",
     [
-        ("datetime", "ts_mtime"),
+        ("datetime", "mtime"),
         ("path", "path"),
         ("string[]", "permitted"),
         ("string[]", "inheritable"),
@@ -104,30 +107,33 @@ def capability_binaries(self) -> Iterator[CapabilityRecord]:
         for entry in self.target.fs.recurse("/"):
             if not entry.is_file() or entry.is_symlink():
                 continue
-
-            try:
-                attrs = [attr for attr in entry.lattr() if attr.name == "security.capability"]
-            except Exception as e:
-                self.target.log.warning("Failed to get attrs for entry %s", entry)
-                self.target.log.debug("", exc_info=e)
-                continue
-
-            for attr in attrs:
-                try:
-                    permitted, inheritable, effective, root_id = parse_attr(attr.value)
-                except ValueError as e:
-                    self.target.log.warning("Could not parse attributes for entry %s: %s", entry, str(e.value))
-                    self.target.log.debug("", exc_info=e)
-
-                yield CapabilityRecord(
-                    ts_mtime=entry.lstat().st_mtime,
-                    path=self.target.fs.path(entry.path),
-                    permitted=permitted,
-                    inheritable=inheritable,
-                    effective=effective,
-                    root_id=root_id,
-                    _target=self.target,
-                )
+            yield from parse_entry(entry, self.target)
+
+
+def parse_entry(entry: FilesystemEntry, target: Target) -> Iterator[CapabilityRecord]:
+    try:
+        attrs = [attr for attr in entry.lattr() if attr.name == "security.capability"]
+    except Exception as e:
+        target.log.warning("Failed to get attrs for entry %s", entry)
+        target.log.debug("", exc_info=e)
+        return
+
+    for attr in attrs:
+        try:
+            permitted, inheritable, effective, root_id = parse_attr(attr.value)
+        except ValueError as e:
+            target.log.warning("Could not parse attributes for entry %s: %s", entry, str(e.value))
+            target.log.debug("", exc_info=e)
+
+        yield CapabilityRecord(
+            mtime=entry.lstat().st_mtime,
+            path=target.fs.path(entry.path),
+            permitted=permitted,
+            inheritable=inheritable,
+            effective=effective,
+            root_id=root_id,
+            _target=target,
+        )
 
 
 def parse_attr(attr: bytes) -> tuple[list[str], list[str], bool, int]:
diff --git a/dissect/target/plugins/filesystem/walkfs.py b/dissect/target/plugins/filesystem/walkfs.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import stat
 from typing import TYPE_CHECKING
 
 from dissect.util.ts import from_unix
@@ -8,6 +9,7 @@
 from dissect.target.filesystem import FilesystemEntry, LayerFilesystemEntry
 from dissect.target.helpers.record import TargetRecordDescriptor
 from dissect.target.plugin import Plugin, arg, export
+from dissect.target.plugins.filesystem.unix.capability import parse_entry as parse_capability_entry
 
 if TYPE_CHECKING:
     from collections.abc import Iterator
@@ -27,12 +29,14 @@
         ("uint32", "mode"),
         ("uint32", "uid"),
         ("uint32", "gid"),
-        ("string[]", "fstypes"),
+        ("boolean", "is_suid"),
+        ("string[]", "attr"),
+        ("string[]", "fs_types"),
     ],
 )
 
 
-class WalkFSPlugin(Plugin):
+class WalkFsPlugin(Plugin):
     """Filesystem agnostic walkfs plugin."""
 
     def check_compatible(self) -> None:
@@ -41,8 +45,39 @@ def check_compatible(self) -> None:
 
     @export(record=FilesystemRecord)
     @arg("--walkfs-path", default="/", help="path to recursively walk")
-    def walkfs(self, walkfs_path: str = "/") -> Iterator[FilesystemRecord]:
-        """Walk a target's filesystem and return all filesystem entries."""
+    @arg("--capability", action="store_true", help="output capability records")
+    def walkfs(self, walkfs_path: str = "/", capability: bool = False) -> Iterator[FilesystemRecord]:
+        """Walk a target's filesystem and return all filesystem entries.
+
+        References:
+            - https://man7.org/linux/man-pages/man2/lstat.2.html
+            - https://man7.org/linux/man-pages/man7/inode.7.html
+            - https://man7.org/linux/man-pages/man7/xattr.7.html
+            - https://man7.org/linux/man-pages/man2/execve.2.html
+            - https://steflan-security.com/linux-privilege-escalation-suid-binaries
+            - https://github.com/torvalds/linux/blob/master/include/uapi/linux/capability.h
+
+        Yields FilesystemRecords for every filesystem entry and CapabilityRecords if ``xattr`` security
+        attributes were found in the filesystem entry and the ``--capability`` flag is set.
+
+        .. code-block:: text
+
+            hostname (string): The target hostname.
+            domain (string): The target domain.
+            mtime (datetime): modified timestamp indicates the last time the contents of a file were modified.
+            atime (datetime): access timestamp indicates the last time a file was accessed.
+            ctime (datetime): changed timestamp indicates the last time metadata of a file was modified.
+            btime (datetime): birth timestamp indicates the time when a file was created.
+            ino (varint): number of the corresponding underlying filesystem inode.
+            path (path): path location of the entry.
+            size (filesize): size of the file in bytes on the filesystem.
+            mode (uint32): contains the file type and mode.
+            uid (uint32): the user id of the owner of the entry.
+            gid (uint32): the group id of the owner of the entry.
+            is_suid (boolean): denotes if the entry has the set-user-id bit set.
+            attr (string[]): list of key-value pair attributes separated by '='.
+            fs_types (string[]): list of filesystem type(s) of the entry.
+        """
 
         path = self.target.fs.path(walkfs_path)
 
@@ -56,45 +91,64 @@ def walkfs(self, walkfs_path: str = "/") -> Iterator[FilesystemRecord]:
 
         for entry in self.target.fs.recurse(walkfs_path):
             try:
-                yield generate_record(self.target, entry)
+                yield from generate_record(self.target, entry, capability)
 
             except FileNotFoundError as e:  # noqa: PERF203
                 self.target.log.warning("File not found: %s", entry)
                 self.target.log.debug("", exc_info=e)
             except Exception as e:
-                self.target.log.warning("Exception generating record for: %s", entry)
+                self.target.log.warning("Exception generating walkfs record for %s: %s", entry, e)
                 self.target.log.debug("", exc_info=e)
                 continue
 
 
-def generate_record(target: Target, entry: FilesystemEntry) -> FilesystemRecord:
-    """Generate a :class:`FilesystemRecord` from the given :class:`FilesystemEntry`.
+def generate_record(target: Target, entry: FilesystemEntry, capability: bool) -> Iterator[FilesystemRecord]:
+    """Generate a :class:`WalkFsRecord` from the given :class:`FilesystemEntry`.
 
     Args:
         target: :class:`Target` instance
         entry: :class:`FilesystemEntry` instance
 
     Returns:
-        Generated :class:`FilesystemRecord` for the given :class:`FilesystemEntry`.
+        Generator of :class:`FilesystemRecord` for the given :class:`FilesystemEntry`.
     """
-    stat = entry.lstat()
+    entry_stat = entry.lstat()
 
     if isinstance(entry, LayerFilesystemEntry):
         fs_types = [sub_entry.fs.__type__ for sub_entry in entry.entries]
     else:
         fs_types = [entry.fs.__type__]
 
-    return FilesystemRecord(
-        atime=from_unix(stat.st_atime),
-        mtime=from_unix(stat.st_mtime),
-        ctime=from_unix(stat.st_ctime),
-        btime=from_unix(stat.st_birthtime) if stat.st_birthtime else None,
-        ino=stat.st_ino,
-        path=entry.path,
-        size=stat.st_size,
-        mode=stat.st_mode,
-        uid=stat.st_uid,
-        gid=stat.st_gid,
-        fstypes=fs_types,
+    fields = {
+        "atime": from_unix(entry_stat.st_atime),
+        "mtime": from_unix(entry_stat.st_mtime),
+        "ctime": from_unix(entry_stat.st_ctime),
+        "btime": from_unix(entry_stat.st_birthtime) if entry_stat.st_birthtime else None,
+        "ino": entry_stat.st_ino,
+        "path": entry.path,
+        "size": entry_stat.st_size,
+        "mode": entry_stat.st_mode,
+        "uid": entry_stat.st_uid,
+        "gid": entry_stat.st_gid,
+        "is_suid": bool(entry_stat.st_mode & stat.S_ISUID),
+        "fs_types": fs_types,
+    }
+
+    try:
+        fields["attr"] = [f"{attr.name}={attr.value.hex()}" for attr in entry.lattr()]
+
+    except (TypeError, AttributeError, NotImplementedError):
+        # Suppress lattr calls on VirtualDirectory entries, filesystems without implemented attr's and NTFS attr's.
+        pass
+
+    except Exception as e:
+        target.log.warning("Unable to expand xattr for entry %s: %s", entry.path, e)
+        target.log.debug("", exc_info=e)
+
+    yield FilesystemRecord(
+        **fields,
         _target=target,
     )
+
+    if capability and fields.get("attr"):
+        yield from parse_capability_entry(entry, target)
diff --git a/tests/plugins/filesystem/test_walkfs.py b/tests/plugins/filesystem/test_walkfs.py
@@ -1,13 +1,16 @@
 from __future__ import annotations
 
+import stat
 from pathlib import Path
 from typing import TYPE_CHECKING
+from unittest.mock import Mock
 
 import pytest
 
 from dissect.target.filesystem import VirtualFile, VirtualFilesystem
+from dissect.target.helpers import fsutil
 from dissect.target.loaders.tar import TarLoader
-from dissect.target.plugins.filesystem.walkfs import WalkFSPlugin
+from dissect.target.plugins.filesystem.walkfs import WalkFsPlugin
 from tests._utils import absolute_path
 
 if TYPE_CHECKING:
@@ -24,11 +27,11 @@ def test_walkfs_plugin(target_unix: Target, fs_unix: VirtualFilesystem) -> None:
     fs_unix.map_file_entry("/.test/test.txt", VirtualFile(fs_unix, "test.txt", None))
     fs_unix.map_file_entry("/.test/.more.test.txt", VirtualFile(fs_unix, ".more.test.txt", None))
 
-    target_unix.add_plugin(WalkFSPlugin)
+    target_unix.add_plugin(WalkFsPlugin)
 
-    results = list(target_unix.walkfs())
+    results = sorted(target_unix.walkfs(), key=lambda r: r.path)
     assert len(results) == 14
-    assert sorted([r.path for r in results]) == [
+    assert [r.path for r in results] == [
         "/",
         "/.test",
         "/.test/.more.test.txt",
@@ -54,6 +57,71 @@ def test_benchmark_walkfs(target_bare: Target, benchmark: BenchmarkFixture) -> N
     loader.map(target_bare)
     target_bare.apply()
 
-    result = benchmark(lambda: next(WalkFSPlugin(target_bare).walkfs()))
+    result = benchmark(lambda: next(WalkFsPlugin(target_bare).walkfs()))
 
     assert result.path == "/"
+
+
+def test_walkfs_suid(target_unix: Target, fs_unix: VirtualFilesystem) -> None:
+    """Test if we detect a SUID binary correctly in the WalkFS plugin."""
+
+    vfile = VirtualFile(fs_unix, "binary", None)
+    vfile.lstat = Mock()
+    vfile.lstat.return_value = fsutil.stat_result([stat.S_IFREG | stat.S_ISUID, 0, 0, 0, 0, 0, 0, 0, 0, 0])
+    fs_unix.map_file_entry("/path/to/suid/binary", vfile)
+
+    target_unix.add_plugin(WalkFsPlugin)
+
+    results = list(target_unix.walkfs())
+    assert len(results) == 7
+
+    assert results[-1].path == "/path/to/suid/binary"
+    assert results[-1].fs_types == ["virtual"]
+    assert results[-1].mode == 34816
+    assert results[-1].is_suid
+
+
+def test_walkfs_xattr(target_unix: Target, fs_unix: VirtualFilesystem) -> None:
+    """Test if we parse xattrs correctly in the WalkFS plugin."""
+
+    xattr1 = Mock()
+    xattr1.name = "security.capability"
+    xattr1.value = bytes.fromhex("010000010020f00000f00f0f")
+
+    xattr2 = Mock()
+    xattr2.name = "example.attr"
+    xattr2.value = b"some value"
+
+    vfile1 = VirtualFile(fs_unix, "file", None)
+    vfile1.lattr = Mock()
+    vfile1.lattr.return_value = [xattr1, xattr2]
+    vfile1.fs.__type__ = "extfs"
+    fs_unix.map_file_entry("/path/to/xattr1/file", vfile1)
+
+    target_unix.add_plugin(WalkFsPlugin)
+
+    results = list(target_unix.walkfs(capability=True))
+    assert len(results) == 8
+
+    assert results[-2].path == "/path/to/xattr1/file"
+    assert results[-2].fs_types == ["extfs"]
+    assert results[-2].attr == ["security.capability=010000010020f00000f00f0f", "example.attr=736f6d652076616c7565"]
+
+    assert results[-1].mtime
+    assert results[-1].permitted == ["CAP_NET_RAW", "CAP_SYS_PACCT", "CAP_SYS_ADMIN", "CAP_SYS_BOOT", "CAP_SYS_NICE"]
+    assert results[-1].inheritable == [
+        "CAP_NET_ADMIN",
+        "CAP_NET_RAW",
+        "CAP_IPC_LOCK",
+        "CAP_IPC_OWNER",
+        "CAP_SYS_MODULE",
+        "CAP_SYS_RAWIO",
+        "CAP_SYS_CHROOT",
+        "CAP_SYS_PTRACE",
+        "CAP_SYS_RESOURCE",
+        "CAP_SYS_TIME",
+        "CAP_SYS_TTY_CONFIG",
+        "CAP_MKNOD",
+    ]
+    assert results[-1].effective
+    assert results[-1].root_id is None
diff --git a/tests/tools/test_query.py b/tests/tools/test_query.py
@@ -426,7 +426,7 @@ def test_record_stream_write_exception_handling(
         with patch("dissect.target.tools.query.record_output", return_value=None):
             target_query()
 
-    assert "Exception occurred while processing output of WalkFSPlugin.walkfs:" in caplog.text
+    assert "Exception occurred while processing output of WalkFsPlugin.walkfs:" in caplog.text
 
 
 @patch("dissect.target.plugin.PLUGINS", new_callable=PluginRegistry)
