diff --git a/src/methods.cpp b/src/methods.cpp index 1fde87fd..e5bb3cd2 100644 --- a/src/methods.cpp +++ b/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 data(nullptr, &free); int data_size = 0; #if PY_MAJOR_VERSION >= 3 if (PyUnicode_CheckExact(arg_data)) { @@ -1575,15 +1575,26 @@ 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(fsize), f); + data.reset(static_cast(malloc(fsize ? static_cast(fsize) : 1))); + if (!data) { + fclose(f); + return PyErr_NoMemory(); + } + data_size = (int)fread(data.get(), 1, static_cast(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(malloc(tuple_size ? static_cast(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); @@ -1591,7 +1602,11 @@ PyObject* meth_coremini_load(PyObject* self, PyObject* args) 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(byte); } fsize = static_cast(tuple_size); data_size = fsize; @@ -1607,7 +1622,7 @@ PyObject* meth_coremini_load(PyObject* self, PyObject* args) ice::Function icsneoScriptLoad( lib, "icsneoScriptLoad"); auto gil = PyAllowThreads(); - if (!icsneoScriptLoad(handle, data, static_cast(data_size), location)) { + if (!icsneoScriptLoad(handle, data.get(), static_cast(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 data(nullptr, &free); int data_size = 0; #if PY_MAJOR_VERSION >= 3 if (PyUnicode_CheckExact(arg_data)) { @@ -3524,15 +3539,26 @@ 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(fsize), f); + data.reset(static_cast(malloc(fsize ? static_cast(fsize) : 1))); + if (!data) { + fclose(f); + return PyErr_NoMemory(); + } + data_size = (int)fread(data.get(), 1, static_cast(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(malloc(tuple_size ? static_cast(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); @@ -3540,7 +3566,11 @@ PyObject* meth_load_readbin(PyObject* self, PyObject* args) 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(byte); } fsize = static_cast(tuple_size); data_size = fsize; @@ -3556,7 +3586,7 @@ PyObject* meth_load_readbin(PyObject* self, PyObject* args) ice::Function icsneoScriptLoadReadBin( lib, "icsneoScriptLoadReadBin"); auto gil = PyAllowThreads(); - if (!icsneoScriptLoadReadBin(handle, data, static_cast(data_size), location)) { + if (!icsneoScriptLoadReadBin(handle, data.get(), static_cast(data_size), location)) { gil.restore(); return set_ics_exception(exception_runtime_error(), "icsneoScriptLoadReadBin() Failed"); } diff --git a/tests/test_script_load.py b/tests/test_script_load.py new file mode 100644 index 00000000..4f145bbc --- /dev/null +++ b/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