diff --git a/MemLib/SharedMemory.py b/MemLib/SharedMemory.py index 6cc49b7..7aa9555 100644 --- a/MemLib/SharedMemory.py +++ b/MemLib/SharedMemory.py @@ -81,6 +81,31 @@ def is_valid(self): return True +class SharedMemoryCleanupError(Exception): + """Raised when one or more shared-memory teardown steps fail. + + The individual failures stay inspectable via :attr:`errors` so callers can react to + a specific code (for example ignoring ERROR_ACCESS_DENIED from an already-dead + target) instead of pattern-matching the formatted message. + + Attributes: + errors (list[Win32Exception]): The failures collected during cleanup. + """ + + def __init__(self, errors: List[Win32Exception]) -> None: + self.errors: List[Win32Exception] = list(errors) + + fmt_error: list[str] = [f'[Error {i + 1}] -> ' + str(error) for i, error in enumerate(self.errors)] + message: str = f'Caught {len(self.errors)} Win32Exception:\n' + '\n-> '.join(fmt_error) + + super().__init__(message) + + @property + def codes(self) -> list[int]: + """Returns the Windows error codes of every collected failure.""" + return [error.code for error in self.errors] + + def close_shared_memory_connection(handle: int, base_addr: int) -> None: """ Disconnects and cleans up resources for a shared memory region. @@ -90,7 +115,8 @@ def close_shared_memory_connection(handle: int, base_addr: int) -> None: base_addr (int): Base address of the mapped view. Raises: - Exception: Aggregated Win32Exception(s) if cleanup fails. + SharedMemoryCleanupError: If cleanup fails. Inspect ``.errors`` for the + individual `Win32Exception` instances. """ errors: List[Win32Exception] = list() if base_addr and not UnmapViewOfFile(base_addr): @@ -100,8 +126,7 @@ def close_shared_memory_connection(handle: int, base_addr: int) -> None: errors.append(Win32Exception()) if len(errors): - fmt_error: list[str] = [f'[Error {i + 1}] -> ' + str(error) for i, error in enumerate(errors)] - raise Exception(f'Caught {len(errors)} Win32Exception:\n' + '\n-> '.join(fmt_error)) + raise SharedMemoryCleanupError(errors) class SharedMemory: """ @@ -264,7 +289,8 @@ def destroy(self) -> None: Disconnects and releases all resources associated with the shared memory. Raises: - Exception: Aggregated Win32Exception(s) if cleanup fails. + SharedMemoryCleanupError: If cleanup fails. Inspect ``.errors`` for the + individual `Win32Exception` instances. """ errors: List[Win32Exception] = list() if not self._owns_remote_resources: @@ -291,8 +317,7 @@ def destroy(self) -> None: errors.append(Win32Exception()) if len(errors): - fmt_error: list[str] = [f'[Error {i + 1}] -> ' + str(error) for i, error in enumerate(errors)] - raise Exception(f'Caught {len(errors)} Win32Exception:\n' + '\n-> '.join(fmt_error)) + raise SharedMemoryCleanupError(errors) mapping.handle = HANDLE(0) mapping.handle_ex = HANDLE(0) diff --git a/MemLib/__init__.py b/MemLib/__init__.py index dd033d3..248dfe5 100644 --- a/MemLib/__init__.py +++ b/MemLib/__init__.py @@ -23,7 +23,12 @@ from MemLib.Process import Process from MemLib.Registry import get_registry_value, set_registry_value from MemLib.Scanner import BinaryScanner, Pattern, generate_assembly_payload -from MemLib.SharedMemory import SharedMemory, SharedMemoryBuffer, close_shared_memory_connection +from MemLib.SharedMemory import ( + SharedMemory, + SharedMemoryBuffer, + SharedMemoryCleanupError, + close_shared_memory_connection, +) from MemLib.Stopwatch import Stopwatch from MemLib.Struct import Struct from MemLib.Thread import Priority, Thread @@ -57,6 +62,7 @@ "Process", "SharedMemory", "SharedMemoryBuffer", + "SharedMemoryCleanupError", "Stopwatch", "Struct", "Thread", diff --git a/MemLib/windows.py b/MemLib/windows.py index d7208a8..15f4037 100644 --- a/MemLib/windows.py +++ b/MemLib/windows.py @@ -1405,12 +1405,20 @@ def __init__(self, error_code: int = None, custom_message: str = None): error_code (int, optional): The Windows error code. If None, the result of GetLastError() will be used. custom_message (str, optional): A custom message. If None, a message will be retrieved using FormatMessageW. """ - self._error_code: int = GetLastError() if (error_code is None) else error_code + self._error_code: int = GetLastError() if (error_code is None) else int(error_code) self._message: str = custom_message if custom_message is None: self.__format_message() + # Populate RuntimeError.args so the exception survives pickling, logging and + # str()/repr() round-trips through code that only looks at `args`. + super().__init__(self._message, self._error_code) + + def __reduce__(self): + """Supports pickling without re-reading the (by then unrelated) thread-local last error.""" + return self.__class__, (self._error_code, self._message) + @property def code(self) -> int: """ @@ -1457,6 +1465,15 @@ def __format_message(self) -> None: or a maximum buffer size is reached. If the message cannot be retrieved, sets the message to 'Unknown Error'. """ + if self._error_code == 0: + # FormatMessageW(0) yields "The operation completed successfully.", which + # is actively misleading inside a raised exception. Callers that construct + # Win32Exception() after a falsy return without checking GetLastError() + # land here; say so plainly instead. + self._message = 'No error reported by Windows (GetLastError() == 0); ' \ + 'the call failed without setting an error code' + return + size: int = 256 while size < 0x10000: # Found 0x10000 in C# std lib diff --git a/tests/test_win32_exception.py b/tests/test_win32_exception.py new file mode 100644 index 0000000..dd3c4f9 --- /dev/null +++ b/tests/test_win32_exception.py @@ -0,0 +1,107 @@ +import ctypes +import pickle + +import pytest + +from MemLib.SharedMemory import SharedMemoryCleanupError, close_shared_memory_connection +from MemLib.windows import Win32Exception + + +def _set_last_error(code: int) -> None: + ctypes.windll.kernel32.SetLastError(code) + + +def test_zero_error_code_does_not_claim_success(): + error = Win32Exception(0) + + assert error.code == 0 + # FormatMessageW(0) would yield "The operation completed successfully." + assert "completed successfully" not in error.message + assert "completed successfully" not in str(error) + assert "GetLastError() == 0" in error.message + + +def test_implicit_zero_last_error_does_not_claim_success(): + _set_last_error(0) + error = Win32Exception() + + assert error.code == 0 + assert "completed successfully" not in str(error) + + +def test_real_error_code_still_formats_the_windows_message(): + error = Win32Exception(5) # ERROR_ACCESS_DENIED + + assert error.code == 5 + # FormatMessageW is localized, so compare against what this system reports for 5 + # rather than the English string. + assert ctypes.FormatError(5).strip() in error.message + + +def test_custom_message_is_preserved(): + error = Win32Exception(5, "custom text") + + assert error.code == 5 + assert error.message == "custom text" + assert "custom text" in str(error) + + +def test_args_are_populated_for_logging_and_reraise(): + error = Win32Exception(5) + + # Previously empty, which made the exception opaque to code inspecting `args`. + assert error.args == (error.message, 5) + + +def test_exception_survives_pickling_without_rereading_last_error(): + _set_last_error(5) + error = Win32Exception() + assert error.code == 5 + + # A different last-error value must not leak into the unpickled copy. + _set_last_error(2) + restored = pickle.loads(pickle.dumps(error)) + + assert restored.code == 5 + assert restored.message == error.message + + +def test_is_still_a_runtime_error(): + assert isinstance(Win32Exception(5), RuntimeError) + + +def test_cleanup_error_keeps_individual_failures_inspectable(): + errors = [Win32Exception(5), Win32Exception(2)] + aggregate = SharedMemoryCleanupError(errors) + + assert aggregate.codes == [5, 2] + assert [error.code for error in aggregate.errors] == [5, 2] + # Callers can now branch on a code instead of grepping the message. + assert 5 in aggregate.codes + assert "Caught 2 Win32Exception" in str(aggregate) + + +def test_cleanup_error_is_catchable_as_exception(): + # Existing `except Exception` handlers must keep working. + assert issubclass(SharedMemoryCleanupError, Exception) + + with pytest.raises(Exception) as caught: + raise SharedMemoryCleanupError([Win32Exception(5)]) + + assert isinstance(caught.value, SharedMemoryCleanupError) + + +def test_close_connection_raises_cleanup_error_with_codes(): + # 0x1 fails both cleanup steps: CloseHandle sets ERROR_INVALID_HANDLE and + # UnmapViewOfFile sets ERROR_INVALID_ADDRESS. Not the pseudo-handle -1, which means + # "current process" and makes CloseHandle *succeed*. + with pytest.raises(SharedMemoryCleanupError) as caught: + close_shared_memory_connection(handle=0x1, base_addr=0x1) + + assert caught.value.errors + assert all(isinstance(error, Win32Exception) for error in caught.value.errors) + assert all("completed successfully" not in str(error) for error in caught.value.errors) + + +def test_close_connection_is_a_noop_for_empty_arguments(): + close_shared_memory_connection(handle=0, base_addr=0)