Validate transmit message types before native access

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<int __stdcall(void*, icsSpyMessage*, int, int)> icsneoTxMessages(lib, "icsneoTxMessages");
         const Py_ssize_t TUPLE_COUNT = PyTuple_Size(tuple);
         icsSpyMessage** msgs = new icsSpyMessage*[static_cast<size_t>(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<Py_ssize_t>(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) {

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)