diff --git a/dissect/target/filesystem.py b/dissect/target/filesystem.py index bbdeb37157..ade6bac155 100644 --- a/dissect/target/filesystem.py +++ b/dissect/target/filesystem.py @@ -1868,7 +1868,7 @@ def open_multi_volume(fhs: list[BinaryIO], *args, **kwargs) -> Iterator[Filesyst register("vmfs", "VmfsFilesystem") register("apfs", "ApfsFilesystem") register("btrfs", "BtrfsFilesystem") -register("exfat", "ExfatFilesystem") +register("exfat", "ExFatFilesystem") register("squashfs", "SquashFSFilesystem") register("cramfs", "CramfsFilesystem") register("jffs", "JffsFilesystem") diff --git a/dissect/target/filesystems/exfat.py b/dissect/target/filesystems/exfat.py index e50959941f..da1fd6c71b 100644 --- a/dissect/target/filesystems/exfat.py +++ b/dissect/target/filesystems/exfat.py @@ -1,112 +1,95 @@ from __future__ import annotations +import math import stat -from datetime import timedelta, timezone from functools import cached_property -from typing import TYPE_CHECKING, BinaryIO, Optional +from typing import TYPE_CHECKING, BinaryIO +import dissect.fat as fat from dissect.fat import exfat -from dissect.util.stream import RunlistStream -from dissect.util.ts import dostimestamp -from dissect.target.exceptions import FileNotFoundError, NotADirectoryError +from dissect.target.exceptions import FileNotFoundError, IsADirectoryError, NotADirectoryError from dissect.target.filesystem import DirEntry, Filesystem, FilesystemEntry from dissect.target.helpers import fsutil if TYPE_CHECKING: from collections.abc import Iterator -ExfatFileTree = tuple[exfat.c_exfat.FILE, dict[str, Optional["ExfatFileTree"]]] - -class ExfatFilesystem(Filesystem): +class ExFatFilesystem(Filesystem): __type__ = "exfat" def __init__(self, fh: BinaryIO, *args, **kwargs): super().__init__(fh, *args, case_sensitive=False, alt_separator="\\", **kwargs) self.exfat = exfat.ExFAT(fh) - self.cluster_size = self.exfat.cluster_size @staticmethod def _detect(fh: BinaryIO) -> bool: - return fh.read(11)[3:] == b"EXFAT " - - def get(self, path: str) -> ExfatFilesystemEntry: - return ExfatFilesystemEntry(self, path, self._get_entry(path)) - - def _get_entry(self, path: str, root: ExfatFileTree | None = None) -> ExfatFileTree: - dirent = root if root is not None else self.exfat.files["/"] - - # Programmatically we will often use the `/` separator, so replace it - # with the native path separator of exFAT `/` is an illegal character - # in exFAT filenames, so it's safe to replace - parts = path.replace("/", "\\").split("\\") - - for part in parts: - if not part: - continue - - file_tree = dirent[1] - if file_tree is None: - raise NotADirectoryError(f"Not a directory: {path}") - - for entry_name, entry_file_tree in file_tree.items(): - if entry_name.upper() == part.upper(): - dirent = entry_file_tree - break - else: - raise FileNotFoundError(f"File not found: {path}") - - return dirent + """Detect an exFAT filesystem on a given file-like object.""" + try: + exfat.validate_boot_sector(fh.read(512)) + except fat.InvalidBootSector: + return False + else: + return True + + def get(self, path: str) -> FilesystemEntry: + return ExFatFilesystemEntry(self, path, self._get_entry(path)) + + def _get_entry( + self, path: str, entry: exfat.RootDirectory | exfat.DirectoryEntry | None = None + ) -> exfat.RootDirectory | exfat.DirectoryEntry: + """Returns an internal exFAT entry for a given path and optional relative entry.""" + try: + return self.exfat.get(path, dirent=entry) + except fat.FileNotFoundError as e: + raise FileNotFoundError(path) from e + except fat.NotADirectoryError as e: + raise NotADirectoryError(path) from e + except fat.Error as e: + raise FileNotFoundError(path) from e @cached_property def serial(self) -> int | str | None: - return self.exfat.vbr.volume_serial + return int(self.exfat.volume_id, 16) -class ExfatDirEntry(DirEntry): - fs: ExfatFilesystem - entry: ExfatFileTree +class ExFatDirEntry(DirEntry): + fs: ExFatFilesystem + entry: exfat.RootDirectory | exfat.DirectoryEntry - def get(self) -> ExfatFilesystemEntry: - return ExfatFilesystemEntry(self.fs, self.path, self.entry) + def get(self) -> ExFatFilesystemEntry: + return ExFatFilesystemEntry(self.fs, self.path, self.entry) def stat(self, *, follow_symlinks: bool = True) -> fsutil.stat_result: return self.get().stat(follow_symlinks=follow_symlinks) -class ExfatFilesystemEntry(FilesystemEntry): - def __init__( - self, - fs: ExfatFilesystem, - path: str, - entry: ExfatFileTree, - ): - super().__init__(fs, path, entry) - self.size = self.entry[0].stream.data_length - self.cluster = self.entry[0].stream.location +class ExFatFilesystemEntry(FilesystemEntry): + fs: ExFatFilesystem + entry: exfat.RootDirectory | exfat.DirectoryEntry - def get(self, path: str) -> ExfatFilesystemEntry: + def get(self, path: str) -> FilesystemEntry: """Get a filesystem entry relative from the current one.""" full_path = fsutil.join(self.path, path, alt_separator=self.fs.alt_separator) - return ExfatFilesystemEntry(self.fs, full_path, self.fs._get_entry(path, self.entry)) + return ExFatFilesystemEntry(self.fs, full_path, self.fs._get_entry(path, self.entry)) def open(self) -> BinaryIO: - if self.entry[0].stream.flags.not_fragmented: - runlist = self.fs.exfat.runlist(self.cluster, True, self.size) - else: - runlist = self.fs.exfat.runlist(self.cluster, False) - return RunlistStream(self.fs.exfat.filesystem, runlist, self.size, self.fs.cluster_size) + """Returns file handle (file-like object).""" + if self.is_dir(): + raise IsADirectoryError(self.path) + return self.entry.open() - def scandir(self) -> Iterator[ExfatDirEntry]: + def scandir(self) -> Iterator[FilesystemEntry]: + """List the directory contents of this directory. Returns a generator of filesystem entries.""" if not self.is_dir(): raise NotADirectoryError(self.path) - for name, file_tree in self.entry[1].items(): - if name in (".", ".."): + for entry in self.entry.iterdir(): + if entry.name in (".", ".."): continue - yield ExfatDirEntry(self.fs, self.path, name, file_tree) + yield ExFatDirEntry(self.fs, self.path, entry.name, entry) def is_symlink(self) -> bool: """Return whether this entry is a link.""" @@ -114,48 +97,34 @@ def is_symlink(self) -> bool: def is_dir(self, follow_symlinks: bool = True) -> bool: """Return whether this entry is a directory.""" - return bool(self.entry[0].metadata.attributes.directory) + return self.entry.is_directory() def is_file(self, follow_symlinks: bool = True) -> bool: """Return whether this entry is a file.""" return not self.is_dir() def stat(self, follow_symlinks: bool = True) -> fsutil.stat_result: + """Return the stat information of this entry.""" return self.lstat() def lstat(self) -> fsutil.stat_result: - """Return the stat information of this entry.""" - fe = self.entry[0] - size = fe.stream.data_length - addr = fe.stream.location - - # exfat stores additional offsets for ctime and mtime to get 10 millisecond precision - fe = fe.metadata - - # all timestamps are recorded in local time. the utc offset (of the system generating the timestamp in question) - # is recorded in the associated tz byte - c_tz = self.fs.exfat._utc_timezone(fe.create_timezone) - c_tzinfo = timezone(timedelta(minutes=c_tz["offset"]), c_tz["name"]) - m_tz = self.fs.exfat._utc_timezone(fe.modified_timezone) - m_tzinfo = timezone(timedelta(minutes=m_tz["offset"]), m_tz["name"]) - a_tz = self.fs.exfat._utc_timezone(fe.access_timezone) - a_tzinfo = timezone(timedelta(minutes=a_tz["offset"]), a_tz["name"]) - - ctime = dostimestamp(fe.create_time, fe.create_offset).replace(tzinfo=c_tzinfo) - mtime = dostimestamp(fe.modified_time, fe.modified_offset).replace(tzinfo=m_tzinfo) - atime = dostimestamp(fe.access_time).replace(tzinfo=a_tzinfo) - + """Return the stat information of the given path, without resolving links.""" # mode, ino, dev, nlink, uid, gid, size, atime, mtime, ctime - st_info = [ - (stat.S_IFDIR if self.is_dir() else stat.S_IFREG) | 0o777, - addr, - id(self.fs), - 0, - 0, - 0, - size, - atime.timestamp(), - mtime.timestamp(), - ctime.timestamp(), - ] - return fsutil.stat_result(st_info) + st_info = fsutil.stat_result( + [ + (stat.S_IFDIR if self.is_dir() else stat.S_IFREG) | 0o777, + self.entry.cluster, + id(self.fs), + 1, + 0, + 0, + self.entry.size, + self.entry.atime.timestamp(), + self.entry.mtime.timestamp(), + self.entry.ctime.timestamp(), + ] + ) + + st_info.st_blocks = math.ceil(self.entry.size / self.entry.fs.cluster_size) + st_info.st_blksize = self.entry.fs.cluster_size + return st_info diff --git a/dissect/target/filesystems/fat.py b/dissect/target/filesystems/fat.py index 3abd1cc137..2a97b28a33 100644 --- a/dissect/target/filesystems/fat.py +++ b/dissect/target/filesystems/fat.py @@ -6,10 +6,10 @@ from functools import cached_property from typing import TYPE_CHECKING, BinaryIO -from dissect.fat import exceptions as fat_exc -from dissect.fat import fat +import dissect.fat as fat +from dissect.fat import fat as fatfs -from dissect.target.exceptions import FileNotFoundError, NotADirectoryError +from dissect.target.exceptions import FileNotFoundError, IsADirectoryError, NotADirectoryError from dissect.target.filesystem import DirEntry, Filesystem, FilesystemEntry from dissect.target.helpers import fsutil @@ -22,7 +22,7 @@ class FatFilesystem(Filesystem): def __init__(self, fh: BinaryIO, *args, **kwargs): super().__init__(fh, *args, case_sensitive=False, alt_separator="\\", **kwargs) - self.fatfs = fat.FATFS(fh) + self.fatfs = fatfs.FATFS(fh) # FAT timestamps are in local time, so to prevent skewing them even more, we specify UTC by default. # However, it should be noted that they are not actual UTC timestamps! # Implementers can optionally set the tzinfo attribute of this class to get correct UTC timestamps. @@ -32,8 +32,8 @@ def __init__(self, fh: BinaryIO, *args, **kwargs): def _detect(fh: BinaryIO) -> bool: """Detect a FAT filesystem on a given file-like object.""" try: - fat.validate_bpb(fh.read(512)) - except fat_exc.InvalidBPB: + fatfs.validate_boot_sector(fh.read(512)) + except fat.InvalidBootSector: return False else: return True @@ -42,16 +42,16 @@ def get(self, path: str) -> FilesystemEntry: return FatFilesystemEntry(self, path, self._get_entry(path)) def _get_entry( - self, path: str, entry: fat.RootDirectory | fat.DirectoryEntry | None = None - ) -> fat.RootDirectory | fat.DirectoryEntry: + self, path: str, entry: fatfs.RootDirectory | fatfs.DirectoryEntry | None = None + ) -> fatfs.RootDirectory | fatfs.DirectoryEntry: """Returns an internal FAT entry for a given path and optional relative entry.""" try: return self.fatfs.get(path, dirent=entry) - except fat_exc.FileNotFoundError as e: + except fat.FileNotFoundError as e: raise FileNotFoundError(path) from e - except fat_exc.NotADirectoryError as e: + except fat.NotADirectoryError as e: raise NotADirectoryError(path) from e - except fat_exc.Error as e: + except fat.Error as e: raise FileNotFoundError(path) from e @cached_property @@ -72,7 +72,7 @@ def stat(self, *, follow_symlinks: bool = True) -> fsutil.stat_result: class FatFilesystemEntry(FilesystemEntry): fs: FatFilesystem - entry: fat.RootDirectory | fat.DirectoryEntry + entry: fatfs.RootDirectory | fatfs.DirectoryEntry def get(self, path: str) -> FilesystemEntry: """Get a filesystem entry relative from the current one.""" diff --git a/pyproject.toml b/pyproject.toml index 13751a5263..af6cb4c6b5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -57,7 +57,7 @@ full = [ "dissect.etl>=3,<4", "dissect.executable>=1,<2", "dissect.extfs>=3,<4", - "dissect.fat>=3,<4", + "dissect.fat>=3.14.dev,<4", # TODO: update on release! "dissect.ffs>=3,<4", "dissect.fve>=4.5,<5", "dissect.jffs>=1.5,<2", @@ -90,7 +90,7 @@ dev = [ "dissect.evidence[dev]>=3.13,<4.0.dev", "dissect.executable[dev]>=1.0.dev,<2.0.dev", "dissect.extfs[dev]>=3.0.dev,<4.0.dev", - "dissect.fat[dev]>=3.0.dev,<4.0.dev", + "dissect.fat[dev]>=3.14.dev,<4.0.dev", "dissect.ffs[dev]>=3.0.dev,<4.0.dev", "dissect.fve[dev]>=4.5,<5.0.dev", "dissect.hypervisor[dev]>=3.21,<4.0.dev", diff --git a/tests/_data/filesystems/exfat/exfat.bin b/tests/_data/filesystems/exfat/exfat.bin new file mode 100644 index 0000000000..fed35dd5ff --- /dev/null +++ b/tests/_data/filesystems/exfat/exfat.bin @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:f246c09038c702a627b34b288b04c1a6253cc2e43dd7b07dc7e5b5bc867e3c20 +size 1048576 diff --git a/tests/filesystems/test_exfat.py b/tests/filesystems/test_exfat.py index 3dcce5c54f..ebc889ef2f 100644 --- a/tests/filesystems/test_exfat.py +++ b/tests/filesystems/test_exfat.py @@ -4,190 +4,47 @@ from typing import TYPE_CHECKING from unittest.mock import Mock, patch -import pytest +from dissect.target.filesystems.exfat import ExFatFilesystem +from tests._utils import absolute_path -from dissect.target.exceptions import FileNotFoundError, NotADirectoryError -from dissect.target.filesystems.exfat import ExfatFilesystem, ExfatFilesystemEntry -if TYPE_CHECKING: - from collections.abc import Iterator +def test_exfat() -> None: + """Test that we can correctly detect and load an exFAT filesystem.""" + with absolute_path("_data/filesystems/exfat/exfat.bin").open("rb") as fh: + assert ExFatFilesystem.detect(fh) + fs = ExFatFilesystem(fh) + assert fs.serial == 0x6859A296 + assert fs.get("/").is_dir() + assert list(fs.get("/").iterdir()) == [ + "$ALLOC_BITMAP", + "$UPCASE_TABLE", + "System Volume Information", + "find_me.txt", + "cat.jpg", + "directory", + ] -def mock_file(name: str, is_dir: bool = False) -> Mock: - file = Mock(name=name) + file = fs.get("find_me.txt") + assert not file.is_dir() + assert file.is_file() + assert file.stat().st_size == 9 - file.metadata = Mock() - file.metadata.attributes = Mock() - file.metadata.attributes.directory = is_dir + with file.open() as fh: + assert fh.read() == b"found me!" - file.stream = Mock() - file.stream.data_length = 0 - - return file - - -MOCK_EXFAT_FILE_TREE = { - "/": ( - mock_file("Mock_", is_dir=True), - { - "some_path": ( - mock_file("Mock_some_path", is_dir=True), - { - "some_file": ( - mock_file("Mock_some_path_some_file"), - None, - ), - "other_file": ( - mock_file("Mock_some_path_other_file"), - None, - ), - }, - ), - "other_file": ( - mock_file("Mock_other_file"), - None, - ), - }, - ), -} - - -@pytest.fixture -def exfat_fs() -> Iterator[ExfatFilesystem]: - with patch("dissect.fat.exfat.ExFAT"): - exfat_fs = ExfatFilesystem(Mock()) - exfat_fs.exfat.files = MOCK_EXFAT_FILE_TREE - - yield exfat_fs - - -@pytest.fixture -def other_file(exfat_fs: ExfatFilesystem) -> ExfatFilesystemEntry: - return ExfatFilesystemEntry(exfat_fs, "/other_file", MOCK_EXFAT_FILE_TREE["/"][1]["other_file"]) - - -@pytest.fixture -def some_path(exfat_fs: ExfatFilesystem) -> ExfatFilesystemEntry: - return ExfatFilesystemEntry(exfat_fs, "/some_path", MOCK_EXFAT_FILE_TREE["/"][1]["some_path"]) - - -def test_filesystem_get(exfat_fs: ExfatFilesystem) -> None: - path = "/some_path/SOME_FILE" - some_file = exfat_fs.get(path) - - assert isinstance(some_file, ExfatFilesystemEntry) - assert some_file.path == path - assert some_file.entry[0]._mock_name == "Mock_some_path_some_file" - - -def test_filesystem__get_entry(exfat_fs: ExfatFilesystem) -> None: - root = exfat_fs._get_entry("/") - assert root[0]._mock_name == "Mock_" - assert isinstance(root[1], dict) - - some_path = exfat_fs._get_entry("/some_path") - - assert some_path[0]._mock_name == "Mock_some_path" - assert exfat_fs._get_entry("/SoMe_PaTh") == some_path - - other_file = exfat_fs._get_entry("/some_path/OTHER_FILE") - - assert other_file[0]._mock_name == "Mock_some_path_other_file" - assert other_file[1] is None - assert exfat_fs._get_entry("OTHER_FILE", root=some_path) == other_file - - -def test_filesystem__get_entry_not_a_directory_error(exfat_fs: ExfatFilesystem) -> None: - with pytest.raises(NotADirectoryError): - exfat_fs._get_entry("/other_file/non-exisiting_file") - - -def test_filesystem__get_entry_file_not_found_error(exfat_fs: ExfatFilesystem) -> None: - with pytest.raises(FileNotFoundError): - exfat_fs._get_entry("/non-exisiting_file") - - -def test_filesystem_entry_get(exfat_fs: ExfatFilesystem, some_path: ExfatFilesystemEntry) -> None: - some_file = some_path.get("some_file") - - assert some_file.path == "/some_path/some_file" - assert some_file.entry[0]._mock_name == "Mock_some_path_some_file" - - -def test_filesystem_entry_iterdir( - exfat_fs: ExfatFilesystem, - some_path: ExfatFilesystemEntry, -) -> None: - file_names = list(some_path.iterdir()) - - assert len(file_names) == 2 - assert "some_file" in file_names - assert "other_file" in file_names - - -def test_filesystem_entry_iterdir_raises( - exfat_fs: ExfatFilesystem, - other_file: ExfatFilesystemEntry, -) -> None: - with pytest.raises(NotADirectoryError): - list(other_file.iterdir()) - - -def test_filesystem_entry_scandir( - exfat_fs: ExfatFilesystem, - some_path: ExfatFilesystemEntry, -) -> None: - file_names = [entry.name for entry in some_path.scandir()] - - assert len(file_names) == 2 - assert "some_file" in file_names - assert "other_file" in file_names - - -def test_filesystem_entry_scandir_raises( - exfat_fs: ExfatFilesystem, - other_file: ExfatFilesystemEntry, -) -> None: - with pytest.raises(NotADirectoryError): - list(other_file.scandir()) - - -def test_filesystem_entry_is_symlink( - exfat_fs: ExfatFilesystem, - other_file: ExfatFilesystemEntry, - some_path: ExfatFilesystemEntry, -) -> None: - assert not other_file.is_symlink() - assert not some_path.is_symlink() - - -def test_filesystem_entry_is_dir( - exfat_fs: ExfatFilesystem, - other_file: ExfatFilesystemEntry, - some_path: ExfatFilesystemEntry, -) -> None: - assert not other_file.is_dir() - assert some_path.is_dir() - assert some_path.is_dir(follow_symlinks=False) - - -def test_filesystem_entry_is_file( - exfat_fs: ExfatFilesystem, - other_file: ExfatFilesystemEntry, - some_path: ExfatFilesystemEntry, -) -> None: - assert other_file.is_file() - assert other_file.is_file(follow_symlinks=False) - assert not some_path.is_file() + dir = fs.get("directory") + assert dir.is_dir() + assert list(dir.iterdir()) == ["putty.exe"] def test_exfat_identifier_no_guid() -> None: """ExFAT.identifier fallback using exfat.vbr.volume_serial when volume.guid is None.""" dummy_fh = BytesIO(b"") # empty in-memory file handle - with patch("dissect.target.filesystems.exfat.ExfatFilesystem.__init__", lambda self, fh: None): - fs = ExfatFilesystem(fh=dummy_fh) + with patch("dissect.target.filesystems.exfat.ExFatFilesystem.__init__", lambda self, fh: None): + fs = ExFatFilesystem(fh=dummy_fh) fs.volume = Mock(guid=None) - fs.exfat = Mock(vbr=Mock(volume_serial=987654321)) + fs.exfat = Mock(volume_id="3ade68b1") expected_uuid = "987654321" assert fs.identifier == expected_uuid diff --git a/tests/filesystems/test_fat.py b/tests/filesystems/test_fat.py index 34db315045..3f7f0aeba2 100644 --- a/tests/filesystems/test_fat.py +++ b/tests/filesystems/test_fat.py @@ -3,22 +3,18 @@ from io import BytesIO from unittest.mock import Mock, create_autospec, patch +import dissect.fat as fat import pytest -from dissect.target.filesystems.fat import ( - FatFilesystem, - FileNotFoundError, - NotADirectoryError, - fat_exc, -) +from dissect.target.filesystems.fat import FatFilesystem @pytest.mark.parametrize( ("raised_exception", "expected_exception"), [ - (fat_exc.FileNotFoundError, FileNotFoundError), - (fat_exc.Error, FileNotFoundError), - (fat_exc.NotADirectoryError, NotADirectoryError), + (fat.FileNotFoundError, FileNotFoundError), + (fat.Error, FileNotFoundError), + (fat.NotADirectoryError, NotADirectoryError), ], ) def test_get_entry(raised_exception: Exception, expected_exception: Exception) -> None: