Add a windows drive container type to load Windows logical or physical drive from a windows host.

dissect/target/container.py

@@ -258,3 +258,4 @@ def open(item: list | str | BinaryIO | Path, *args, **kwargs) -> Container:
 register("hds", "HdsContainer")
 register("split", "SplitContainer")
 register("fortifw", "FortiFirmwareContainer")
+register("windows_drive", "WindowsDrive")

dissect/target/containers/windows_drive.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+import io
+import re
+from typing import TYPE_CHECKING, BinaryIO
+
+from dissect.util.stream import BufferedStream
+
+from dissect.target.container import Container
+from dissect.target.helpers.windows_ffi import _windows_get_disk_size, _windows_get_drive_size, run_on_windows
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def is_physical_drive_path(path: str) -> bool:
+    r"""Check if path match a logical drive path, E.g : \\.\PhysicalDrive1 ."""
+    return re.fullmatch(r"\\\\.\\+PhysicalDrive[0-9]+", path, re.IGNORECASE) is not None
+
+
+def is_logical_drive_path(path: str) -> bool:
+    r"""Check if path match a logical drive path, E.g : `\\.\C:` or `\\.\\\Z:` ."""
+    return re.fullmatch(r"\\\\.\\+[a-z]:", path, re.IGNORECASE) is not None
+
+
+class WindowsDrive(Container):
+    r"""Allows to load windows drive, such as `\\.\C:` or `\\.\PhysicalDrive1` directly from Windows.
+    This Container is needed as Windows Drive does not support seek end, and must be wrapped inside a bufferedStream.
+    Specific Windows api call must be used to retrieves drive length.
+    """
+
+    __type__ = "windows_drive"
+
+    def __init__(self, fh: BinaryIO | Path, *args, **kwargs):
+        if hasattr(fh, "read"):
+            raise TypeError("Windows Drive can only be opened by path")
+        if not run_on_windows():
+            raise TypeError("Windows Drive is only available on Windows platform.")
+        if is_physical_drive_path(path=str(fh)):
+            drive_size = _windows_get_disk_size(str(fh))
+        else:
+            drive_size = _windows_get_drive_size(str(fh))
+        self._raw_stream = fh.open("rb")
+        self.stream = BufferedStream(
+            self._raw_stream,
+            size=drive_size,
+        )
+        super().__init__(fh, drive_size, *args, **kwargs)
+
+    @staticmethod
+    def _detect_fh(fh: BinaryIO, original: list | BinaryIO) -> bool:
+        return False
+
+    @staticmethod
+    def detect_path(path: Path, original: list | BinaryIO) -> bool:
+        if not run_on_windows():
+            return False
+        return is_physical_drive_path(str(path)) or is_logical_drive_path(str(path))
+
+    def read(self, length: int = -1) -> bytes:
+        return self.stream.read(length)
+
+    def seek(self, offset: int, whence: int = io.SEEK_SET) -> int:
+        return self.stream.seek(offset, whence)
+
+    def tell(self) -> int:
+        return self.stream.tell()
+
+    def close(self) -> None:
+        self.stream.close()
+        self._raw_stream.close()

dissect/target/helpers/windows_ffi.py

@@ -0,0 +1,251 @@
+"""Function related to low level Windows API call (using ctypes.win*).
+
+Mainly used for Physical/logical drive information gathering on live windows systems.
+"""
+
+from __future__ import annotations
+
+import ctypes
+import sys
+from enum import IntFlag
+from typing import TypeVar
+
+T = TypeVar("T", bound=ctypes.Structure)
+
+# Some commons err code that we may encounter, to provide more readable error message.
+ERR_CODE_MAP = {
+    0x4: "ERROR_TOO_MANY_OPEN_FILES",
+    0x5: "ERROR_ACCESS_DENIED",
+    0x6: "ERROR_INVALID_HANDLE",
+    0xE: "ERROR_INVALID_DRIVE",
+    0x20: "ERROR_SHARING_VIOLATION",
+}
+
+
+class GenericAccessRight(IntFlag):
+    """CreateFileW dwDesiredAccess (some IOCTL required specific access, such as READ instead of ZERO).
+
+    References:
+        - https://learn.microsoft.com/en-us/windows/win32/secauthz/generic-access-rights
+    """
+
+    GENERIC_ALL = 0x10000000
+    GENERIC_EXECUTE = 0x20000000
+    GENERIC_WRITE = 0x40000000
+    GENERIC_READ = 0x80000000
+    GENERIC_ZERO = 0x0
+
+
+class FileShareMode(IntFlag):
+    """CreateFileW file share mode.
+
+    References:
+    - https://learn.microsoft.com/en-us/windows/win32/api/fileapi/nf-fileapi-createfilew
+    """
+
+    NONE = 0x0
+    READ = 0x1
+    WRITE = 0x2
+    DELETE = 0x4
+
+
+def run_on_windows() -> bool:
+    """Simple wrapper around sys.plateform check.
+
+    Allows to ease mock.
+    """
+    return sys.platform == "win32"
+
+
+def _windows_get_disk_size(path: str) -> int:
+    """Get disk size from a Drive path. Must be used only on a Windows platform.
+
+    Args:
+        path: Drive path, E.g. `\\\\.\\PhysicalDrive0`
+    """
+    geometry_ex = _windows_get_disk_geometry_ex(path)
+    return geometry_ex.DiskSize
+
+
+def _windows_get_drive_size(path: str) -> int:
+    """Retrieves the length of the specified disk, volume, or partition. Must be used only on a Windows platform.
+
+    Unlike _windows_get_disk_size, also works on volume and partition but require GENERIC_READ permission
+        on file handle.
+
+    Args:
+        path: Drive path, E.g `\\\\.\\PhysicalDrive0`, `\\\\.\\D:`
+    """
+    return _windows_disk_get_length_info(path)
+
+
+def _windows_disk_get_length_info(path: str) -> int:
+    """Call IOCTL_DISK_GET_LENGTH_INFO.
+
+    References:
+        - https://learn.microsoft.com/en-us/windows/win32/api/winioctl/ni-winioctl-ioctl_disk_get_length_info
+    """
+    # IOCTL_DISK_GET_LENGTH_INFO = CTL_CODE(IOCTL_DISK_BASE, 0x0017, METHOD_BUFFERED, FILE_READ_ACCESS)
+    IOCTL_DISK_GET_LENGTH_INFO = 0x7405C
+    from ctypes import wintypes
+
+    class GET_LENGTH_INFORMATION(ctypes.Structure):
+        _fields_ = (("Length", wintypes.LARGE_INTEGER),)
+
+    # File share mode is not the desired access, but controls how other processes are
+    # allowed to access the file/device while you have it open.
+    # E.g. FileShareMode.DELETE means that file can be open,
+    # even if another process has a handle open with delete access.
+    # Being permissive allows to prevent ERROR_SHARING_VIOLATION.
+    # ERROR_SHARING_VIOLATION can still occur if an handle is already open without FileShareMode.READ, as we need
+    # GENERIC_READ access.
+    handle = _windows_createfile(
+        path,
+        desired_access=GenericAccessRight.GENERIC_READ,
+        file_share_mode=FileShareMode.READ | FileShareMode.DELETE | FileShareMode.WRITE,
+    )
+    try:
+        status, res = _windows_ioctl(handle, IOCTL_DISK_GET_LENGTH_INFO, GET_LENGTH_INFORMATION)
+    finally:
+        _windows_closehandle(handle)
+
+    if status == 0:
+        err = ctypes.windll.kernel32.GetLastError()
+        raise OSError(f"unable to execute IOCTL_DISK_GET_LENGTH_INFO, error: 0x{err:08x}. {ERR_CODE_MAP.get(err, '')}")
+
+    return res.Length
+
+
+def _windows_get_disk_geometry_ex(path: str) -> ctypes.Structure:
+    """Call IOCTL_DISK_GET_DRIVE_GEOMETRY_EX to retrieve size from the physical disk's geometry.
+
+    References:
+        - https://learn.microsoft.com/en-us/windows/win32/api/winioctl/ni-winioctl-ioctl_disk_get_drive_geometry_ex
+    """
+    from ctypes import wintypes
+
+    # IOCTL_DISK_GET_DRIVE_GEOMETRY_EX = CTL_CODE(IOCTL_DISK_BASE, 0x0028, METHOD_BUFFERED, FILE_ANY_ACCESS)
+    IOCTL_DISK_GET_DRIVE_GEOMETRY_EX = 0x700A0
+
+    class DISK_GEOMETRY(ctypes.Structure):
+        _fields_ = (
+            ("Cylinders", wintypes.LARGE_INTEGER),
+            ("MediaType", wintypes.BYTE),
+            ("TracksPerCylinder", wintypes.DWORD),
+            ("SectorsPerTrack", wintypes.DWORD),
+            ("BytesPerSector", wintypes.DWORD),
+        )
+
+    class DISK_GEOMETRY_EX(ctypes.Structure):
+        _fields_ = (
+            ("Geometry", DISK_GEOMETRY),
+            ("DiskSize", wintypes.LARGE_INTEGER),
+            ("Data", wintypes.BYTE),
+        )
+
+    handle = _windows_createfile(path)
+    try:
+        status, res = _windows_ioctl(handle, IOCTL_DISK_GET_DRIVE_GEOMETRY_EX, DISK_GEOMETRY_EX)
+    finally:
+        _windows_closehandle(handle)
+
+    if status == 0:
+        err = ctypes.windll.kernel32.GetLastError()
+        raise OSError(
+            f"unable to execute IOCTL_DISK_GET_DRIVE_GEOMETRY_EX, error: 0x{err:08x} {ERR_CODE_MAP.get(err, '')}"
+        )
+
+    return res
+
+
+def _windows_createfile(
+    path: str,
+    desired_access: GenericAccessRight = GenericAccessRight.GENERIC_ZERO,
+    file_share_mode: FileShareMode = FileShareMode.NONE,
+) -> int:
+    """Open a file using the windows CreateFileW API."""
+    OPEN_EXISTING = 3
+    FILE_ATTRIBUTE_NORMAL = 0x00000080
+
+    handle = ctypes.windll.kernel32.CreateFileW(
+        path,
+        int(desired_access),
+        int(file_share_mode),
+        0,
+        OPEN_EXISTING,
+        FILE_ATTRIBUTE_NORMAL,
+        0,
+    )
+
+    if handle == -1:
+        err = ctypes.windll.kernel32.GetLastError()
+        raise OSError(
+            f"unable to open handle to {path} using CreateFileW, error: 0x{err:08x} {ERR_CODE_MAP.get(err, '')}"
+        )
+
+    return handle
+
+
+def _windows_closehandle(handle: int) -> None:
+    """Close an Handle using windows CloseHandle API.
+
+    Args:
+        handle: file handle create dusing _windows_createfile.
+    """
+    ctypes.windll.kernel32.CloseHandle(handle)
+
+
+def _windows_ioctl(handle: int, ioctl: int, out_struct: type[T]) -> tuple[int, T]:
+    # http://www.ioctls.net/
+    from ctypes import wintypes
+
+    out_inst = out_struct()
+    bytes_returned = wintypes.DWORD(0)
+    status = ctypes.windll.kernel32.DeviceIoControl(
+        handle,
+        ioctl,
+        None,
+        0,
+        ctypes.pointer(out_inst),
+        ctypes.sizeof(out_struct),
+        ctypes.byref(bytes_returned),
+        None,
+    )
+
+    return status, out_inst
+
+
+def _windows_get_volume_disk_extents(path: str) -> ctypes.Structure:
+    from ctypes import wintypes
+
+    ERROR_MORE_DATA = 234
+    IOCTL_VOLUME_GET_VOLUME_DISK_EXTENTS = 0x560000
+
+    class DISK_EXTENT(ctypes.Structure):
+        _fields_ = (
+            ("DiskNumber", wintypes.DWORD),
+            ("StartingOffset", wintypes.LARGE_INTEGER),
+            ("ExtentLength", wintypes.LARGE_INTEGER),
+        )
+
+    class VOLUME_DISK_EXTENTS(ctypes.Structure):
+        _fields_ = (
+            ("NumberOfDiskExtents", wintypes.DWORD),
+            ("Extents", DISK_EXTENT * 1),
+        )
+
+    handle = _windows_createfile(path)
+    try:
+        status, res = _windows_ioctl(handle, IOCTL_VOLUME_GET_VOLUME_DISK_EXTENTS, VOLUME_DISK_EXTENTS)
+    finally:
+        _windows_closehandle(handle)
+
+    if status == 0:
+        err = ctypes.windll.kernel32.GetLastError()
+        if err != ERROR_MORE_DATA:
+            raise OSError(
+                f"unable to execute IOCTL_VOLUME_GET_VOLUME_DISK_EXTENTS, "
+                f"error: 0x{err:08x} {ERR_CODE_MAP.get(err, '')}"
+            )
+
+    return res

dissect/target/loader.py

@@ -203,7 +203,9 @@ def find_loader(
         except ImportError as e:  # noqa: PERF203
             log.info("Failed to import %s", loader)
             log.debug("", exc_info=e)
-
+        except OSError as e:
+            log.info("Loader %s fail with exception %s", loader, e)
+            log.debug("", exc_info=e)
     return None