From e56291f008009fc57362d3063bb986d3519915c8 Mon Sep 17 00:00:00 2001 From: David Rebbe Date: Tue, 15 Sep 2026 23:32:21 -0400 Subject: [PATCH 1/2] fix: validate transmit message types Reject unsupported objects before reading native message fields or transmitting any part of a batch. Preserve both supported message layouts and cover validation with a hardware-free native stub. --- src/methods.cpp | 23 +++--- tests/test_transmit_messages.py | 121 ++++++++++++++++++++++++++++++++ 2 files changed, 134 insertions(+), 10 deletions(-) create mode 100644 tests/test_transmit_messages.py diff --git a/src/methods.cpp b/src/methods.cpp index 1fde87fd..1b913270 100644 --- a/src/methods.cpp +++ b/src/methods.cpp @@ -1771,6 +1771,17 @@ PyObject* meth_transmit_messages(PyObject* self, PyObject* args) if (!PyNeoDeviceEx_GetHandle(obj, &handle)) { return NULL; } + // Validate the entire batch before taking native pointers or transmitting. + // A non-tuple argument represents a single message. + const Py_ssize_t message_count = PyTuple_CheckExact(temp) ? PyTuple_Size(temp) : 1; + for (Py_ssize_t i = 0; i < message_count; ++i) { + PyObject* message = PyTuple_CheckExact(temp) ? PyTuple_GetItem(temp, i) : temp; + if (!PySpyMessage_CheckExact(message) && !PySpyMessageJ1850_CheckExact(message)) { + return set_ics_exception(PyExc_TypeError, + "Message must be of type " MODULE_NAME "." SPY_MESSAGE_OBJECT_NAME " or " + MODULE_NAME "." SPY_MESSAGE_J1850_OBJECT_NAME); + } + } PyObject* tuple = temp; if (!PyTuple_CheckExact(temp)) { tuple = Py_BuildValue("(O)", temp); @@ -1791,20 +1802,12 @@ PyObject* meth_transmit_messages(PyObject* self, PyObject* args) ice::Function icsneoTxMessages(lib, "icsneoTxMessages"); const Py_ssize_t TUPLE_COUNT = PyTuple_Size(tuple); icsSpyMessage** msgs = new icsSpyMessage*[static_cast(TUPLE_COUNT)](); - for (int i = 0; i < TUPLE_COUNT; ++i) { + for (Py_ssize_t i = 0; i < TUPLE_COUNT; ++i) { spy_message_object* _obj = (spy_message_object*)PyTuple_GetItem(tuple, static_cast(i)); - if (!_obj) { - if (created_tuple) { - Py_XDECREF(tuple); - } - delete[] msgs; - return set_ics_exception(exception_runtime_error(), - "Tuple item must be of " MODULE_NAME "." SPY_MESSAGE_OBJECT_NAME); - } msgs[i] = &(_obj->msg); } auto gil = PyAllowThreads(); - for (int i = 0; i < TUPLE_COUNT; ++i) { + for (Py_ssize_t i = 0; i < TUPLE_COUNT; ++i) { if (!icsneoTxMessages(handle, msgs[i], (msgs[i]->NetworkID2 << 8) | msgs[i]->NetworkID, 1)) { gil.restore(); if (created_tuple) { diff --git a/tests/test_transmit_messages.py b/tests/test_transmit_messages.py new file mode 100644 index 00000000..6b498f9e --- /dev/null +++ b/tests/test_transmit_messages.py @@ -0,0 +1,121 @@ +"""Hardware-free transmit validation using a native call-counting stub.""" + +import os +from pathlib import Path +import shutil +import subprocess +import sys + +import pytest + + +@pytest.fixture(scope="module") +def transmit_library(tmp_path_factory): + directory = tmp_path_factory.mktemp("transmit-library") + source = directory / "mock.c" + source.write_text( + """ +#ifdef _WIN32 +#define API __declspec(dllexport) +#define CALL __stdcall +#else +#define API +#define CALL +#endif +static int calls; +static int network; +static int result = 1; +API int CALL icsneoTxMessages(void* handle, void* message, int net, int count) { + ++calls; + network = net; + return result; +} +API int review_calls(void) { return calls; } +API int review_network(void) { return network; } +API void review_result(int value) { result = value; } +""" + ) + if sys.platform == "win32": + compiler = shutil.which("clang-cl") + linker = shutil.which("lld-link") + if not compiler or not linker: + pytest.skip("Native transmit tests require clang-cl and lld-link") + library = directory / "mock.dll" + obj = directory / "mock.obj" + bits = 64 if sys.maxsize > 2**32 else 32 + subprocess.run( + [compiler, f"-m{bits}", "/nologo", "/c", "/GS-", "/Zl", str(source), f"/Fo{obj}"], check=True + ) + command = [linker, "/dll", "/noentry", "/nodefaultlib", f"/out:{library}", str(obj)] + if bits == 32: + command.append("/export:icsneoTxMessages=_icsneoTxMessages@16") + subprocess.run(command, check=True) + else: + compiler = shutil.which("cc") + if not compiler: + pytest.skip("Native transmit tests require a C compiler") + library = directory / ("mock.dylib" if sys.platform == "darwin" else "mock.so") + subprocess.run([compiler, "-shared", "-fPIC", str(source), "-o", str(library)], check=True) + return library + + +def test_transmit_message_types(transmit_library): + # Isolate the process-global library override from other tests. No real + # device is opened and the mock never dereferences a message pointer. + script = r''' +import ctypes +import sys +import ics + +ics.override_library_name(sys.argv[1]) +library = ctypes.CDLL(sys.argv[1]) +device = ics.PyNeoDeviceEx() +device._auto_handle_close = False +invalid_device = ics.PyNeoDeviceEx() +invalid_device._auto_handle_close = False + +for invalid in (None, 42, object(), "message", [], {}, invalid_device): + for argument in (invalid, (invalid,), (ics.SpyMessage(), invalid), + (invalid, ics.SpyMessageJ1850()), + (ics.SpyMessage(), invalid, ics.SpyMessageJ1850())): + before = library.review_calls() + refs = sys.getrefcount(invalid) + try: + ics.transmit_messages(device, argument) + except TypeError as error: + assert "SpyMessage" in str(error) + else: + raise AssertionError(f"accepted invalid input: {argument!r}") + assert library.review_calls() == before, "partially transmitted invalid batch" + assert sys.getrefcount(invalid) == refs, "invalid input leaked a reference" + +standard = ics.SpyMessage() +j1850 = ics.SpyMessageJ1850() +standard.NetworkID = 0x34 +standard.NetworkID2 = 0x12 +j1850.NetworkID = 0x78 +j1850.NetworkID2 = 0x56 +for argument, count, network in ((standard, 1, 0x1234), (j1850, 1, 0x5678), + ((standard, j1850), 2, 0x5678), ((), 0, 0x5678)): + before = library.review_calls() + assert ics.transmit_messages(device, argument) is None + assert library.review_calls() == before + count + assert library.review_network() == network + +library.review_result(0) +before = library.review_calls() +try: + ics.transmit_messages(device, (standard, j1850)) +except ics.RuntimeError: + pass +else: + raise AssertionError("native transmit failure was ignored") +assert library.review_calls() == before + 1 +''' + # Pass the actual imported package location even when pytest was launched + # with an adjusted sys.path instead of an installed wheel. + import ics + + environment = os.environ.copy() + environment["PYTHONPATH"] = os.pathsep.join([str(Path(ics.__file__).resolve().parent.parent), *sys.path]) + subprocess.run([sys.executable, "-c", script, str(transmit_library)], env=environment, check=True) From 599f264eb4f5b4f3cbc08a735b0b0bf46af442ef Mon Sep 17 00:00:00 2001 From: David Rebbe Date: Wed, 16 Sep 2026 10:58:27 -0400 Subject: [PATCH 2/2] test: isolate transmit input ownership checks Shared None counts include device handle leaks and interpreter bookkeeping. Measure private rejected inputs, batch containers, and valid neighbors over repeated calls while retaining the full type-rejection matrix. --- tests/test_transmit_messages.py | 23 +++++++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/tests/test_transmit_messages.py b/tests/test_transmit_messages.py index 6b498f9e..4af99c74 100644 --- a/tests/test_transmit_messages.py +++ b/tests/test_transmit_messages.py @@ -79,7 +79,6 @@ def test_transmit_message_types(transmit_library): (invalid, ics.SpyMessageJ1850()), (ics.SpyMessage(), invalid, ics.SpyMessageJ1850())): before = library.review_calls() - refs = sys.getrefcount(invalid) try: ics.transmit_messages(device, argument) except TypeError as error: @@ -87,7 +86,27 @@ def test_transmit_message_types(transmit_library): else: raise AssertionError(f"accepted invalid input: {argument!r}") assert library.review_calls() == before, "partially transmitted invalid batch" - assert sys.getrefcount(invalid) == refs, "invalid input leaked a reference" + +# Measure ownership separately with private objects. Shared singleton counts +# include unrelated interpreter activity and device._handle lookups (#243). +# Check the rejected scalar, batch container, and valid neighbors so neither +# a direct reference leak nor a retained temporary tuple can go unnoticed. +sentinel = object() +neighbors = (ics.SpyMessage(), ics.SpyMessageJ1850()) +for argument in (sentinel, (sentinel,), (neighbors[0], sentinel), + (sentinel, neighbors[1]), (*neighbors, sentinel)): + tracked = (sentinel, argument, *neighbors) + refs = tuple(sys.getrefcount(value) for value in tracked) + before = library.review_calls() + for _ in range(100): + try: + ics.transmit_messages(device, argument) + except TypeError: + pass + else: + raise AssertionError("accepted invalid owned input") + assert library.review_calls() == before, "partially transmitted invalid batch" + assert tuple(sys.getrefcount(value) for value in tracked) == refs, "rejected input retained references" standard = ics.SpyMessage() j1850 = ics.SpyMessageJ1850()