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..4af99c74 --- /dev/null +++ b/tests/test_transmit_messages.py @@ -0,0 +1,140 @@ +"""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() + 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" + +# 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() +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)