diff --git a/dissect/target/container.py b/dissect/target/container.py index 6931b2d5b9..187132edf2 100644 --- a/dissect/target/container.py +++ b/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") diff --git a/dissect/target/containers/windows_drive.py b/dissect/target/containers/windows_drive.py new file mode 100644 index 0000000000..64e9c1da43 --- /dev/null +++ b/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() diff --git a/dissect/target/helpers/windows_ffi.py b/dissect/target/helpers/windows_ffi.py new file mode 100644 index 0000000000..33f0f19292 --- /dev/null +++ b/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 diff --git a/dissect/target/loader.py b/dissect/target/loader.py index c57cd1bdb4..fea7cfbc05 100644 --- a/dissect/target/loader.py +++ b/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 diff --git a/dissect/target/loaders/local.py b/dissect/target/loaders/local.py index bb5da87ced..4501fa3832 100644 --- a/dissect/target/loaders/local.py +++ b/dissect/target/loaders/local.py @@ -6,7 +6,7 @@ import urllib.parse from functools import cache from pathlib import Path -from typing import TYPE_CHECKING, TypeVar +from typing import TYPE_CHECKING from dissect.util.stream import BufferedStream @@ -14,6 +14,10 @@ from dissect.target.containers.raw import RawContainer from dissect.target.exceptions import LoaderError from dissect.target.filesystems.dir import DirectoryFilesystem +from dissect.target.helpers.windows_ffi import ( + _windows_get_disk_size, + _windows_get_volume_disk_extents, +) from dissect.target.loader import Loader if TYPE_CHECKING: @@ -380,125 +384,5 @@ def _is_physical_drive(path: str) -> bool: return path in _windows_get_devices() -def _windows_get_disk_size(path: str) -> int: - geometry_ex = _windows_get_disk_geometry_ex(path) - return geometry_ex.DiskSize - - -def _windows_get_disk_geometry_ex(path: str) -> ctypes.Structure: - from ctypes import wintypes - - 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}") - - return res - - -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, error: 0x{err:08x}") - - return res - - -def _windows_createfile(path: str) -> int: - OPEN_EXISTING = 3 - FILE_ATTRIBUTE_NORMAL = 0x00000080 - - handle = ctypes.windll.kernel32.CreateFileW( - path, - 0, - 0, - 0, - OPEN_EXISTING, - FILE_ATTRIBUTE_NORMAL, - 0, - ) - - if handle == -1: - err = ctypes.windll.kernel32.GetLastError() - raise OSError(f"unable to open handle to {path}, error: 0x{err:08x}") - - return handle - - -def _windows_closehandle(handle: int) -> None: - ctypes.windll.kernel32.CloseHandle(handle) - - -T = TypeVar("T", bound=ctypes.Structure) - - -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 _get_os_name() -> str: return platform.system().lower() diff --git a/tests/_data/containers/raw/small.txt b/tests/_data/containers/raw/small.txt new file mode 100644 index 0000000000..912866553a --- /dev/null +++ b/tests/_data/containers/raw/small.txt @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:ee8c1db79fe4c2e3e0712b3304a1f524ad7928685421dd0e66ffa607676e3261 +size 15 diff --git a/tests/containers/test_raw.py b/tests/containers/test_raw.py new file mode 100644 index 0000000000..eed3744303 --- /dev/null +++ b/tests/containers/test_raw.py @@ -0,0 +1,27 @@ +from __future__ import annotations + +import io + +from dissect.target import container +from dissect.target.containers.raw import RawContainer +from tests._utils import absolute_path + + +def test_raw_container() -> None: + """Test that a raw containers are properly opened. + + Generated with:: + + echo "TestDissectRaw" > small.txt + """ + path = absolute_path("_data/containers/raw/small.txt") + + fh = container.open(path) + assert isinstance(fh, RawContainer) + a = fh.read(20) + # trailing \x00 are expected. This is the same when using qemu-nbd + assert a == b"TestDissectRaw\n" + assert fh.tell() == 15 + fh.seek(0, whence=io.SEEK_SET) + assert fh.read(20) == b"TestDissectRaw\n" + fh.close() diff --git a/tests/containers/test_windows_drive.py b/tests/containers/test_windows_drive.py new file mode 100644 index 0000000000..80f0fb2507 --- /dev/null +++ b/tests/containers/test_windows_drive.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +import io +from io import BytesIO +from unittest import mock + +from dissect.target import container +from dissect.target.containers.windows_drive import WindowsDrive +from dissect.target.filesystem import VirtualFilesystem + + +def test_windows_drive_detect_path() -> None: + """Test that Windows drive containers are properly opened, when using the path based matching. + + Windows API call are mocked, and thus not tested. + """ + vfs = VirtualFilesystem() + with ( + mock.patch("dissect.target.containers.windows_drive.run_on_windows", return_value=True), + mock.patch( + "dissect.target.containers.windows_drive._windows_get_disk_size", return_value=18 + ) as mock_windows_get_disk_size, + mock.patch( + "dissect.target.containers.windows_drive._windows_get_drive_size", return_value=18 + ) as mock_windows_get_drive_size, + ): + vfs.map_file_fh("\\\\.\\PhysicalDrive5", BytesIO(b'echo "hello world"')) + vfs.map_file_fh("\\\\.\\C:", BytesIO(b'echo "hello world"')) + + fh = container.open(vfs.path("\\\\.\\PhysicalDrive5")) + assert isinstance(fh, WindowsDrive) + assert mock_windows_get_disk_size.call_count == 1 + assert mock_windows_get_drive_size.call_count == 0 + a = fh.read(20) + assert a == b'echo "hello world"' + assert fh.tell() == 18 + fh.seek(0, whence=io.SEEK_END) + assert fh.tell() == 18 + fh.close() + + fh = container.open(vfs.path("\\\\.\\C:")) + assert isinstance(fh, WindowsDrive) + assert mock_windows_get_disk_size.call_count == 1 + assert mock_windows_get_drive_size.call_count == 1 + a = fh.read(20) + assert a == b'echo "hello world"' + assert fh.tell() == 18 + fh.seek(0, whence=io.SEEK_END) + assert fh.tell() == 18 + fh.close()