Add support for FATX and OG XBOX HDD

dissect/target/filesystem.py

@@ -1864,6 +1864,7 @@ def open_multi_volume(fhs: list[BinaryIO], *args, **kwargs) -> Iterator[Filesyst
 register("extfs", "ExtFilesystem")
 register("xfs", "XfsFilesystem")
 register("fat", "FatFilesystem")
+register("fatx", "FatxFilesystem")
 register("ffs", "FfsFilesystem")
 register("vmfs", "VmfsFilesystem")
 register("apfs", "ApfsFilesystem")

dissect/target/filesystems/fatx.py

@@ -0,0 +1,130 @@
+from __future__ import annotations
+
+import datetime
+import math
+import stat
+from functools import cached_property
+from typing import TYPE_CHECKING, BinaryIO
+
+import dissect.fat as fat
+from dissect.fat import fatx
+
+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
+
+
+class FatxFilesystem(Filesystem):
+    __type__ = "fatx"
+
+    def __init__(self, fh: BinaryIO, *args, **kwargs):
+        super().__init__(fh, *args, case_sensitive=False, alt_separator="\\", **kwargs)
+        self.fatx = fatx.FATX(fh)
+        # FATX 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.
+        self.tzinfo = datetime.timezone.utc
+
+    @staticmethod
+    def _detect(fh: BinaryIO) -> bool:
+        """Detect a FATX filesystem on a given file-like object."""
+        return fatx.is_fatx(fh)
+
+    def get(self, path: str) -> FilesystemEntry:
+        return FatxFilesystemEntry(self, path, self._get_entry(path))
+
+    def _get_entry(
+        self, path: str, entry: fatx.RootDirectory | fatx.DirectoryEntry | None = None
+    ) -> fatx.RootDirectory | fatx.DirectoryEntry:
+        """Returns an internal FATX entry for a given path and optional relative entry."""
+        try:
+            return self.fatx.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 int(self.fatx.volume_id, 16)
+
+
+class FatxDirEntry(DirEntry):
+    fs: FatxFilesystem
+    entry: fatx.RootDirectory | fatx.DirectoryEntry
+
+    def get(self) -> FatxFilesystemEntry:
+        return FatxFilesystemEntry(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 FatxFilesystemEntry(FilesystemEntry):
+    fs: FatxFilesystem
+    entry: fatx.RootDirectory | fatx.DirectoryEntry
+
+    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 FatxFilesystemEntry(self.fs, full_path, self.fs._get_entry(path, self.entry))
+
+    def open(self) -> BinaryIO:
+        """Returns file handle (file-like object)."""
+        if self.is_dir():
+            raise IsADirectoryError(self.path)
+        return self.entry.open()
+
+    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 entry in self.entry.iterdir():
+            if entry.name in (".", ".."):
+                continue
+
+            yield FatxDirEntry(self.fs, self.path, entry.name, entry)
+
+    def is_symlink(self) -> bool:
+        """Return whether this entry is a link."""
+        return False
+
+    def is_dir(self, follow_symlinks: bool = True) -> bool:
+        """Return whether this entry is a 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 the given path, without resolving links."""
+        # mode, ino, dev, nlink, uid, gid, size, atime, mtime, ctime
+        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.replace(tzinfo=self.fs.tzinfo).timestamp(),
+                self.entry.mtime.replace(tzinfo=self.fs.tzinfo).timestamp(),
+                self.entry.ctime.replace(tzinfo=self.fs.tzinfo).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

dissect/target/plugin.py

@@ -68,6 +68,7 @@ class OperatingSystem(StrEnum):
     UNIX = "unix"
     VYOS = "vyos"
     WINDOWS = "windows"
+    XBOX = "xbox"
 
 
 @dataclass(frozen=True, eq=True, slots=True)

dissect/target/plugins/os/xbox/_os.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from dissect.target.helpers.record import EmptyRecord
+from dissect.target.plugin import OperatingSystem, OSPlugin, export, internal
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+    from typing_extensions import Self
+
+    from dissect.target.filesystem import Filesystem
+    from dissect.target.target import Target
+
+
+class XboxPlugin(OSPlugin):
+    @classmethod
+    def detect(cls, target: Target) -> Filesystem | None:
+        for disk in target.disks:
+            if disk.vs.__type__ == "xbox":
+                for volume in disk.vs.volumes:
+                    if volume.name == "C":
+                        return volume.fs
+
+        return None
+
+    @classmethod
+    def create(cls, target: Target, sysvol: Filesystem) -> Self:
+        target.fs.case_sensitive = False
+        target.fs.alt_separator = "\\"
+
+        for disk in target.disks:
+            if disk.vs.__type__ == "xbox":
+                for volume in disk.vs.volumes:
+                    if volume.fs:
+                        target.fs.mount(volume.name.lower() + ":", volume.fs)
+
+        return cls(target)
+
+    @export(property=True)
+    def hostname(self) -> str | None:
+        return "XBOX"
+
+    @export(property=True)
+    def ips(self) -> list[str]:
+        return []
+
+    @export(property=True)
+    def version(self) -> str | None:
+        return None
+
+    @export(property=True)
+    def architecture(self) -> str | None:
+        return None
+
+    @export(record=EmptyRecord)
+    def users(self) -> Iterator[EmptyRecord]:
+        yield from ()
+
+    @internal
+    def misc_user_paths(self) -> Iterator[tuple[str, tuple[str, str] | None]]:
+        yield from ()
+
+    @export(property=True)
+    def os(self) -> str:
+        return OperatingSystem.XBOX

dissect/target/volume.py

@@ -15,10 +15,11 @@
     from typing_extensions import Self
 
     from dissect.target.filesystem import Filesystem
-    from dissect.target.volumes.disk import DissectVolumeSystem
 
 disk = import_lazy("dissect.target.volumes.disk")
 """A lazy import of :mod:`dissect.target.volumes.disk`."""
+xbox = import_lazy("dissect.target.volumes.xbox")
+"""A lazy import of :mod:`dissect.target.volumes.xbox`."""
 lvm = import_lazy("dissect.target.volumes.lvm")
 """A lazy import of :mod:`dissect.target.volumes.lvm`."""
 vmfs = import_lazy("dissect.target.volumes.vmfs")
@@ -36,14 +37,21 @@
 log = get_logger(__name__)
 """A logger instance for this module."""
 
+ALTERNATIVE_VOLUME_MANAGERS: list[type[VolumeSystem]] = [
+    xbox.XboxVolumeSystem,
+]
+"""All available :class:`VolumeSystem` classes that aren't :class:`~dissect.target.volumes.disk.DissectVolumeSystem`."""
 LOGICAL_VOLUME_MANAGERS: list[type[LogicalVolumeSystem]] = [
     lvm.LvmVolumeSystem,
     vmfs.VmfsVolumeSystem,
     md.MdVolumeSystem,
     ddf.DdfVolumeSystem,
 ]
 """All available :class:`LogicalVolumeSystem` classes."""
-ENCRYPTED_VOLUME_MANAGERS: list[type[EncryptedVolumeSystem]] = [bde.BitlockerVolumeSystem, luks.LUKSVolumeSystem]
+ENCRYPTED_VOLUME_MANAGERS: list[type[EncryptedVolumeSystem]] = [
+    bde.BitlockerVolumeSystem,
+    luks.LUKSVolumeSystem,
+]
 """All available :class:`EncryptedVolumeSystem` classes."""
 
 
@@ -320,25 +328,38 @@ def seekable(self) -> bool:
         return True
 
 
-def open(fh: BinaryIO, *args, **kwargs) -> DissectVolumeSystem:
-    """Open a :class:`~dissect.target.volumes.disk.DissectVolumeSystem` on the given file-like object.
+def open(fh: BinaryIO, *args, **kwargs) -> VolumeSystem:
+    """Open a supported :class:`~dissect.target.volume.VolumeSystem` on the given file-like object.
 
     Args:
-        fh: The file-like object to open a :class:`~dissect.target.volumes.disk.DissectVolumeSystem` on.
+        fh: The file-like object to open a :class:`~dissect.target.volume.VolumeSystem` on.
 
     Raises:
-        VolumeSystemError: If opening the :class:`~dissect.target.volumes.disk.DissectVolumeSystem` failed.
+        VolumeSystemError: If opening the :class:`~dissect.target.volume.VolumeSystem` failed.
 
     Returns:
-        An opened :class:`~dissect.target.volumes.disk.DissectVolumeSystem`.
+        An opened :class:`~dissect.target.volume.VolumeSystem`.
     """
     offset = fh.tell()
-    fh.seek(0)
-
     try:
+        fh.seek(0)
         return disk.DissectVolumeSystem(fh)
     except Exception as e:
-        raise VolumeSystemError(f"Failed to load volume system for {fh}") from e
+        for cls in ALTERNATIVE_VOLUME_MANAGERS:
+            try:
+                if cls.detect(fh):
+                    return cls(fh)
+            except ImportError as e:  # noqa: PERF203
+                log.info("Failed to import %s", cls)
+                log.debug("", exc_info=e)
+            except Exception as e:
+                log.error(  # noqa: TRY400
+                    "Failed to open %s volume manager for %s: %s", cls.__type__, fh, e
+                )
+                log.debug("", exc_info=e)
+        else:
+            # Otherwise raise the original exception
+            raise VolumeSystemError(f"Failed to load volume system for {fh}") from e
     finally:
         fh.seek(offset)
 
@@ -349,12 +370,12 @@ def is_lvm_volume(volume: BinaryIO) -> bool:
     Args:
         volume: A file-like object to test if it is part of any logical volume system.
     """
-    for logical_vs in LOGICAL_VOLUME_MANAGERS:
+    for cls in LOGICAL_VOLUME_MANAGERS:
         try:
-            if logical_vs.detect_volume(volume):
+            if cls.detect_volume(volume):
                 return True
         except ImportError as e:  # noqa: PERF203
-            log.info("Failed to import %s", logical_vs)
+            log.info("Failed to import %s", cls)
             log.debug("", exc_info=e)
         except Exception as e:
             raise VolumeSystemError(f"Failed to detect logical volume for {volume}") from e
@@ -368,12 +389,12 @@ def is_encrypted(volume: BinaryIO) -> bool:
     Args:
         volume: A file-like object to test if it is part of any encrypted volume system.
     """
-    for manager in ENCRYPTED_VOLUME_MANAGERS:
+    for cls in ENCRYPTED_VOLUME_MANAGERS:
         try:
-            if manager.detect(volume):
+            if cls.detect(volume):
                 return True
         except ImportError as e:  # noqa: PERF203
-            log.info("Failed to import %s", manager)
+            log.info("Failed to import %s", cls)
             log.debug("", exc_info=e)
         except Exception as e:
             raise VolumeSystemError(f"Failed to detect encrypted volume for {volume}") from e
@@ -394,17 +415,17 @@ def open_encrypted(volume: BinaryIO) -> Iterator[Volume]:
     Returns:
         An iterator of decrypted :class:`Volume` objects as opened by the encrypted volume manager.
     """
-    for manager_cls in ENCRYPTED_VOLUME_MANAGERS:
+    for cls in ENCRYPTED_VOLUME_MANAGERS:
         try:
-            if manager_cls.detect(volume):
-                volume_manager = manager_cls(volume)
+            if cls.detect(volume):
+                volume_manager = cls(volume)
                 yield from volume_manager.volumes
         except ImportError as e:  # noqa: PERF203
-            log.info("Failed to import %s", manager_cls)
+            log.info("Failed to import %s", cls)
             log.debug("", exc_info=e)
         except Exception as e:
             log.error(  # noqa: TRY400
-                "Failed to open an encrypted volume %s with volume manager %s: %s", volume, manager_cls.__type__, e
+                "Failed to open an encrypted volume %s with volume manager %s: %s", volume, cls.__type__, e
             )
             log.debug("", exc_info=e)
     return None
@@ -419,11 +440,11 @@ def open_lvm(volumes: list[BinaryIO], *args, **kwargs) -> Iterator[VolumeSystem]
     Returns:
         An iterator of all the :class:`Volume` objects opened by the logical volume system.
     """
-    for logical_vs in LOGICAL_VOLUME_MANAGERS:
+    for cls in LOGICAL_VOLUME_MANAGERS:
         try:
-            yield from logical_vs.open_all(volumes)
+            yield from cls.open_all(volumes)
         except ImportError as e:  # noqa: PERF203
-            log.info("Failed to import %s", logical_vs)
+            log.info("Failed to import %s", cls)
             log.debug("", exc_info=e)
         except Exception as e:
-            raise VolumeSystemError(f"Failed to load logical volume system for {volumes}") from e
+            raise VolumeSystemError(f"Failed to open a logical volume system for {volumes}") from e