Add mimetype detection to walkfs

dissect/target/helpers/magic/magic.py

@@ -91,7 +91,6 @@ def from_descriptor(fh: BinaryIO, suffix: str | None = None, *, mime: bool = Fal
         raise TypeError("Provided fh does not have a read or seek method")
 
     return Magic().detect(fh, suffix, mime=mime)
-    return from_buffer(fh, suffix, mime=mime)
 
 
 # Convenience alias, not present in python-magic.

dissect/target/plugins/filesystem/walkfs.py

@@ -7,6 +7,7 @@
 
 from dissect.target.exceptions import FileNotFoundError, UnsupportedPluginError
 from dissect.target.filesystem import FilesystemEntry, LayerFilesystemEntry
+from dissect.target.helpers.magic import magic
 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
@@ -29,6 +30,7 @@
         ("uint32", "mode"),
         ("uint32", "uid"),
         ("uint32", "gid"),
+        ("string", "mimetype"),
         ("boolean", "is_suid"),
         ("string[]", "attr"),
         ("string[]", "fs_types"),
@@ -144,6 +146,14 @@ def generate_record(target: Target, entry: FilesystemEntry, capability: bool) ->
         target.log.warning("Unable to expand xattr for entry %s: %s", entry.path, e)
         target.log.debug("", exc_info=e)
 
+    try:
+        fields["mimetype"] = magic.from_entry(entry, mime=True)
+    except (FileNotFoundError, IsADirectoryError):
+        pass
+    except Exception as e:
+        target.log.warning("Unable to determine mimetype for entry %s: %s", entry.path, e)
+        target.log.debug("", exc_info=e)
+
     yield FilesystemRecord(
         **fields,
         _target=target,

tests/plugins/filesystem/test_walkfs.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import stat
+from io import BytesIO
 from pathlib import Path
 from typing import TYPE_CHECKING
 from unittest.mock import Mock
@@ -20,12 +21,14 @@
 
 
 def test_walkfs_plugin(target_unix: Target, fs_unix: VirtualFilesystem) -> None:
-    fs_unix.map_file_entry("/path/to/some/file", VirtualFile(fs_unix, "file", None))
-    fs_unix.map_file_entry("/path/to/some/other/file.ext", VirtualFile(fs_unix, "file.ext", None))
-    fs_unix.map_file_entry("/root_file", VirtualFile(fs_unix, "root_file", None))
-    fs_unix.map_file_entry("/other_root_file.ext", VirtualFile(fs_unix, "other_root_file.ext", 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))
+    """Test basic walkfs plugin behavior."""
+
+    fs_unix.map_file_fh("/path/to/some/file", BytesIO(bytes.fromhex("89504e470d0a1a0a")))
+    fs_unix.map_file_fh("/path/to/some/other/file.ext", BytesIO(b""))
+    fs_unix.map_file_fh("/root_file", BytesIO(b""))
+    fs_unix.map_file_fh("/other_root_file.ext", BytesIO(b""))
+    fs_unix.map_file_fh("/.test/test.txt", BytesIO(b""))
+    fs_unix.map_file_fh("/.test/.more.test.txt", BytesIO(b""))
 
     target_unix.add_plugin(WalkFsPlugin)
 
@@ -48,6 +51,12 @@ def test_walkfs_plugin(target_unix: Target, fs_unix: VirtualFilesystem) -> None:
         "/var",
     ]
 
+    assert {str(r.path): r.mimetype for r in results if r.mimetype} == {
+        "/.test/.more.test.txt": "text/plain",  # inferred by txt extension
+        "/.test/test.txt": "text/plain",  # inferred by txt extension
+        "/path/to/some/file": "image/png",  # inferred by magic bytes
+    }
+
 
 @pytest.mark.benchmark
 def test_benchmark_walkfs(target_bare: Target, benchmark: BenchmarkFixture) -> None: