Fix script buffer leaks in coremini_load

src/methods.cpp

@@ -1547,7 +1547,7 @@ PyObject* meth_coremini_load(PyObject* self, PyObject* args)
         return NULL;
     }
     long fsize;
-    unsigned char* data = NULL;
+    std::unique_ptr<unsigned char, decltype(&free)> data(nullptr, &free);
     int data_size = 0;
 #if PY_MAJOR_VERSION >= 3
     if (PyUnicode_CheckExact(arg_data)) {
@@ -1575,23 +1575,38 @@ PyObject* meth_coremini_load(PyObject* self, PyObject* args)
             fclose(f);
             return set_ics_exception(exception_runtime_error(), "CoreMini script file size is invalid");
         }
-        data = (unsigned char*)malloc(sizeof(char) * fsize);
-        data_size = (int)fread(data, 1, static_cast<size_t>(fsize), f);
+        data.reset(static_cast<unsigned char*>(malloc(fsize ? static_cast<size_t>(fsize) : 1)));
+        if (!data) {
+            fclose(f);
+            return PyErr_NoMemory();
+        }
+        data_size = (int)fread(data.get(), 1, static_cast<size_t>(fsize), f);
         fclose(f);
         if (fsize != data_size) {
             return set_ics_exception(exception_runtime_error(), "CoreMini binary file size mismatch");
         }
     } else if (PyTuple_CheckExact(arg_data)) {
         Py_ssize_t tuple_size = PyTuple_Size(arg_data);
-        data = (unsigned char*)malloc(sizeof(char) * tuple_size);
+        if (tuple_size > INT_MAX) {
+            PyErr_SetString(PyExc_OverflowError, "Script tuple is too large");
+            return NULL;
+        }
+        data.reset(static_cast<unsigned char*>(malloc(tuple_size ? static_cast<size_t>(tuple_size) : 1)));
+        if (!data) {
+            return PyErr_NoMemory();
+        }
         // Move tuple data into array
         for (int i = 0; i < tuple_size; ++i) {
             PyObject* value = PyTuple_GET_ITEM(arg_data, i);
             if (!PyLong_CheckExact(value)) {
                 return set_ics_exception(exception_runtime_error(),
                                          "Failed to convert tuple data. Tuple data must be integer type");
             }
-            data[i] = (unsigned char)PyLong_AsLong(PyTuple_GET_ITEM(arg_data, i));
+            long byte = PyLong_AsLong(value);
+            if (byte == -1 && PyErr_Occurred()) {
+                return NULL;
+            }
+            data.get()[i] = static_cast<unsigned char>(byte);
         }
         fsize = static_cast<long>(tuple_size);
         data_size = fsize;
@@ -1607,7 +1622,7 @@ PyObject* meth_coremini_load(PyObject* self, PyObject* args)
         ice::Function<int __stdcall(void*, const unsigned char*, unsigned long, int)> icsneoScriptLoad(
             lib, "icsneoScriptLoad");
         auto gil = PyAllowThreads();
-        if (!icsneoScriptLoad(handle, data, static_cast<unsigned long>(data_size), location)) {
+        if (!icsneoScriptLoad(handle, data.get(), static_cast<unsigned long>(data_size), location)) {
             gil.restore();
             return set_ics_exception(exception_runtime_error(), "icsneoScriptLoad() Failed");
         }
@@ -3496,7 +3511,7 @@ PyObject* meth_load_readbin(PyObject* self, PyObject* args)
         return NULL;
     }
     long fsize;
-    unsigned char* data = NULL;
+    std::unique_ptr<unsigned char, decltype(&free)> data(nullptr, &free);
     int data_size = 0;
 #if PY_MAJOR_VERSION >= 3
     if (PyUnicode_CheckExact(arg_data)) {
@@ -3524,23 +3539,38 @@ PyObject* meth_load_readbin(PyObject* self, PyObject* args)
             fclose(f);
             return set_ics_exception(exception_runtime_error(), "Readbin file size is invalid");
         }
-        data = (unsigned char*)malloc(sizeof(char) * fsize);
-        data_size = (int)fread(data, 1, static_cast<size_t>(fsize), f);
+        data.reset(static_cast<unsigned char*>(malloc(fsize ? static_cast<size_t>(fsize) : 1)));
+        if (!data) {
+            fclose(f);
+            return PyErr_NoMemory();
+        }
+        data_size = (int)fread(data.get(), 1, static_cast<size_t>(fsize), f);
         fclose(f);
         if (fsize != data_size) {
             return set_ics_exception(exception_runtime_error(), "Readbin file size mismatch");
         }
     } else if (PyTuple_CheckExact(arg_data)) {
         Py_ssize_t tuple_size = PyTuple_Size(arg_data);
-        data = (unsigned char*)malloc(sizeof(char) * tuple_size);
+        if (tuple_size > INT_MAX) {
+            PyErr_SetString(PyExc_OverflowError, "Script tuple is too large");
+            return NULL;
+        }
+        data.reset(static_cast<unsigned char*>(malloc(tuple_size ? static_cast<size_t>(tuple_size) : 1)));
+        if (!data) {
+            return PyErr_NoMemory();
+        }
         // Move tuple data into array
         for (int i = 0; i < tuple_size; ++i) {
             PyObject* value = PyTuple_GET_ITEM(arg_data, i);
             if (!PyLong_CheckExact(value)) {
                 return set_ics_exception(exception_runtime_error(),
                                          "Failed to convert tuple data. Tuple data must be integer type");
             }
-            data[i] = (unsigned char)PyLong_AsLong(PyTuple_GET_ITEM(arg_data, i));
+            long byte = PyLong_AsLong(value);
+            if (byte == -1 && PyErr_Occurred()) {
+                return NULL;
+            }
+            data.get()[i] = static_cast<unsigned char>(byte);
         }
         fsize = static_cast<long>(tuple_size);
         data_size = fsize;
@@ -3556,7 +3586,7 @@ PyObject* meth_load_readbin(PyObject* self, PyObject* args)
         ice::Function<int __stdcall(void*, const unsigned char*, unsigned long, int)> icsneoScriptLoadReadBin(
             lib, "icsneoScriptLoadReadBin");
         auto gil = PyAllowThreads();
-        if (!icsneoScriptLoadReadBin(handle, data, static_cast<unsigned long>(data_size), location)) {
+        if (!icsneoScriptLoadReadBin(handle, data.get(), static_cast<unsigned long>(data_size), location)) {
             gil.restore();
             return set_ics_exception(exception_runtime_error(), "icsneoScriptLoadReadBin() Failed");
         }

tests/test_script_load.py

@@ -0,0 +1,141 @@
+"""Hardware-free script loading tests; each mock runs in a separate process."""
+
+import os
+from pathlib import Path
+import shutil
+import struct
+import subprocess
+import sys
+
+import ics
+import pytest
+
+
+@pytest.fixture(scope="module")
+def script_library(tmp_path_factory):
+    root = tmp_path_factory.mktemp("script-library")
+    source = root / "mock.c"
+    source.write_text(r"""
+#ifdef _WIN32
+#define API __declspec(dllexport)
+#define CALL __stdcall
+#else
+#define API __attribute__((visibility("default")))
+#define CALL
+#endif
+API int CALL icsneoScriptLoad(void* h, const unsigned char* data, unsigned long size, int location) {
+    unsigned long i;
+    if (location == -1) return 0;
+    for (i = 0; i < size; ++i) if (data[i] != (unsigned char)i) return 0;
+    return 1;
+}
+API int CALL icsneoScriptLoadReadBin(void* h, const unsigned char* data, unsigned long size, int location) {
+    return icsneoScriptLoad(h, data, size, location);
+}
+""")
+    if sys.platform == "win32":
+        compiler, linker = shutil.which("clang-cl"), shutil.which("lld-link")
+        if not compiler or not linker:
+            pytest.skip("LLVM is required for the mock DLL")
+        library = root / "mock.dll"
+        is_32bit = struct.calcsize("P") == 4
+        target = "i686-pc-windows-msvc" if is_32bit else "x86_64-pc-windows-msvc"
+        subprocess.run([compiler, f"--target={target}", "/nologo", "/c", "/GS-", "/Zl", str(source),
+                        f"/Fo{root / 'mock.obj'}"], check=True, capture_output=True, timeout=60)
+        aliases = (["/export:icsneoScriptLoad=_icsneoScriptLoad@16",
+                    "/export:icsneoScriptLoadReadBin=_icsneoScriptLoadReadBin@16"] if is_32bit else [])
+        subprocess.run([linker, "/dll", "/noentry", "/nodefaultlib", f"/out:{library}",
+                        str(root / "mock.obj"), *aliases], check=True, capture_output=True, timeout=60)
+    else:
+        compiler = shutil.which("cc")
+        if not compiler:
+            pytest.skip("C compiler is required for the mock library")
+        library = root / "mock.so"
+        subprocess.run([compiler, "-shared", "-fPIC", str(source), "-o", str(library)], check=True, timeout=60)
+    return library
+
+
+@pytest.mark.parametrize("loader", ["coremini_load", "load_readbin"])
+@pytest.mark.parametrize("kind", ["file", "tuple"])
+@pytest.mark.parametrize("failure", [False, True])
+def test_script_load(script_library, tmp_path, loader, kind, failure):
+    if not hasattr(ics, loader):
+        pytest.skip("load_readbin requires the internal header build")
+    # Preserve the exact extension under test without changing the parent process's DLL.
+    env = os.environ.copy()
+    env["PYTHONPATH"] = str(Path(ics.__file__).resolve().parent.parent)
+    script = r'''
+import ctypes
+import sys
+from pathlib import Path
+import ics
+
+library, loader, kind, failure, root = sys.argv[1:]
+ics.override_library_name(library)
+load = getattr(ics, loader)
+device = ics.PyNeoDeviceEx()
+device._auto_handle_close = False
+payload = bytes(range(256)) * 4096
+path = Path(root) / "payload.bin"
+path.write_bytes(payload)
+data = str(path) if kind == "file" else tuple(payload)
+
+def call():
+    if failure == "True":
+        try:
+            load(device, data, -1)
+        except ics.RuntimeError:
+            pass
+        else:
+            raise AssertionError("native failure was ignored")
+    else:
+        assert load(device, data, 0) is None
+
+def private_bytes():
+    # Windows private commit includes malloc storage even when it is not resident.
+    class Counters(ctypes.Structure):
+        _fields_ = [("cb", ctypes.c_ulong), ("PageFaultCount", ctypes.c_ulong)] + [
+            (name, ctypes.c_size_t) for name in (
+                "PeakWorkingSetSize", "WorkingSetSize", "QuotaPeakPagedPoolUsage",
+                "QuotaPagedPoolUsage", "QuotaPeakNonPagedPoolUsage", "QuotaNonPagedPoolUsage",
+                "PagefileUsage", "PeakPagefileUsage", "PrivateUsage")]
+    counters = Counters()
+    kernel = ctypes.WinDLL("kernel32", use_last_error=True)
+    kernel.GetCurrentProcess.restype = ctypes.c_void_p
+    psapi = ctypes.WinDLL("psapi", use_last_error=True)
+    psapi.GetProcessMemoryInfo.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_ulong]
+    assert psapi.GetProcessMemoryInfo(kernel.GetCurrentProcess(), ctypes.byref(counters), ctypes.sizeof(counters))
+    return counters.PrivateUsage
+
+for _ in range(4):
+    call()
+before = private_bytes() if sys.platform == "win32" else None
+for _ in range(32):
+    call()
+if before is not None:
+    growth = private_bytes() - before
+    print("private byte growth:", growth)
+    assert growth < 8 * 1024 * 1024, growth
+
+# Failed conversions must preserve their exception and never call the native API.
+for value, error in [(object(), ics.RuntimeError), (1 << 100, OverflowError)]:
+    try:
+        load(device, (0, value), 0)
+    except error:
+        pass
+    else:
+        raise AssertionError("invalid tuple accepted")
+empty = Path(root) / "empty.bin"
+empty.write_bytes(b"")
+assert load(device, str(empty), 0) is None
+assert load(device, (), 0) is None
+try:
+    load(device, str(Path(root) / "missing.bin"), 0)
+except ics.RuntimeError:
+    pass
+else:
+    raise AssertionError("missing file accepted")
+'''
+    result = subprocess.run([sys.executable, "-c", script, str(script_library), loader, kind,
+                             str(failure), str(tmp_path)], env=env, capture_output=True, text=True, timeout=60)
+    assert result.returncode == 0, result.stdout + result.stderr