Stop Win32Exception from reporting success, and keep cleanup failures inspectable

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)

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",

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

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)