diff --git a/src/methods.cpp b/src/methods.cpp index 1fde87fd..2a13d08e 100644 --- a/src/methods.cpp +++ b/src/methods.cpp @@ -42,6 +42,20 @@ class PyAllowThreads } }; +// Own an already acquired export. Construct only after successful argument +// parsing: PyArg_ParseTuple releases any acquired buffers itself on failure. +// Declare before PyAllowThreads so unwinding restores the GIL before release. +class PyBufferRelease +{ + Py_buffer& buffer; + + public: + explicit PyBufferRelease(Py_buffer& value) : buffer(value) {} + ~PyBufferRelease() { PyBuffer_Release(&buffer); } + PyBufferRelease(const PyBufferRelease&) = delete; + PyBufferRelease& operator=(const PyBufferRelease&) = delete; +}; + extern PyTypeObject spy_message_object_type; // __func__, __FUNCTION__ and __PRETTY_FUNCTION__ are not preprocessor macros. // but MSVC doesn't follow c standard and treats __FUNCTION__ as a string literal macro... @@ -3153,17 +3167,20 @@ PyObject* meth_get_hw_firmware_info(PyObject* self, PyObject* args) if (!info) { return NULL; } + std::unique_ptr result(info, Py_DecRef); Py_buffer info_buffer = {}; - PyObject_GetBuffer(info, &info_buffer, PyBUF_CONTIG); + if (PyObject_GetBuffer(info, &info_buffer, PyBUF_CONTIG) < 0) { + return NULL; + } + PyBufferRelease buffer_release(info_buffer); auto gil = PyAllowThreads(); if (!icsneoGetHWFirmwareInfo(handle, (stAPIFirmwareInfo*)info_buffer.buf)) { gil.restore(); - PyBuffer_Release(&info_buffer); return set_ics_exception(exception_runtime_error(), "icsneoGetHWFirmwareInfo() Failed"); } gil.restore(); - return info; + return result.release(); } catch (ice::Exception& ex) { return set_ics_exception(exception_runtime_error(), (char*)ex.what()); } @@ -3359,15 +3376,19 @@ PyObject* meth_get_dll_firmware_info(PyObject* self, PyObject* args) if (!info) { return NULL; } + std::unique_ptr result(info, Py_DecRef); Py_buffer info_buffer = {}; - PyObject_GetBuffer(info, &info_buffer, PyBUF_CONTIG); + if (PyObject_GetBuffer(info, &info_buffer, PyBUF_CONTIG) < 0) { + return NULL; + } + PyBufferRelease buffer_release(info_buffer); auto gil = PyAllowThreads(); if (!icsneoGetDLLFirmwareInfo(handle, (stAPIFirmwareInfo*)info_buffer.buf)) { gil.restore(); return set_ics_exception(exception_runtime_error(), "icsneoGetDLLFirmwareInfo() Failed"); } gil.restore(); - return info; + return result.release(); } catch (ice::Exception& ex) { return set_ics_exception(exception_runtime_error(), (char*)ex.what()); } @@ -3977,8 +3998,12 @@ PyObject* meth_get_device_status(PyObject* self, PyObject* args) if (!device_status) { return NULL; } + std::unique_ptr result(device_status, Py_DecRef); Py_buffer device_status_buffer = {}; - PyObject_GetBuffer(device_status, &device_status_buffer, PyBUF_CONTIG); + if (PyObject_GetBuffer(device_status, &device_status_buffer, PyBUF_CONTIG) < 0) { + return NULL; + } + PyBufferRelease buffer_release(device_status_buffer); size_t device_status_size = static_cast(device_status_buffer.len); ice::Function icsneoGetDeviceStatus(lib, @@ -3986,18 +4011,16 @@ PyObject* meth_get_device_status(PyObject* self, PyObject* args) auto gil = PyAllowThreads(); if (!icsneoGetDeviceStatus(handle, (icsDeviceStatus*)device_status_buffer.buf, &device_status_size)) { gil.restore(); - PyBuffer_Release(&device_status_buffer); return set_ics_exception(exception_runtime_error(), "icsneoGetDeviceStatus() Failed"); } if (throw_exception_on_size_mismatch) { if (device_status_size != (size_t)device_status_buffer.len) { gil.restore(); - PyBuffer_Release(&device_status_buffer); return set_ics_exception(exception_runtime_error(), "icsneoGetDeviceStatus() API mismatch detected!"); } } gil.restore(); - return device_status; + return result.release(); } catch (ice::Exception& ex) { return set_ics_exception(exception_runtime_error(), (char*)ex.what()); } @@ -4276,13 +4299,17 @@ PyObject* meth_flash_accessory_firmware(PyObject* self, PyObject* args) lib, "icsneoFlashAccessoryFirmware"); Py_buffer parms_buffer = {}; - PyObject_GetBuffer(parms, &parms_buffer, PyBUF_CONTIG_RO); + if (PyObject_GetBuffer(parms, &parms_buffer, PyBUF_CONTIG_RO) < 0) { + return NULL; + } + PyBufferRelease parms_release(parms_buffer); auto gil = PyAllowThreads(); if (!icsneoFlashAccessoryFirmware(handle, (FlashAccessoryFirmwareParams*)parms_buffer.buf, &function_error)) { gil.restore(); return set_ics_exception(exception_runtime_error(), "icsneoFlashAccessoryFirmware() Failed"); } + gil.restore(); // check the return value to make sure we are good if (check_success && function_error != AccessoryOperationSuccess) { std::stringstream ss; @@ -5029,6 +5056,7 @@ PyObject* meth_uart_write(PyObject* self, PyObject* args) return NULL; } + PyBufferRelease data_release(data); // Get the device handle if (!PyNeoDeviceEx_CheckExact(obj)) { return set_ics_exception(exception_runtime_error(), "Argument must be of type " MODULE_NAME ".PyNeoDeviceEx"); @@ -5210,6 +5238,7 @@ PyObject* meth_generic_api_send_command(PyObject* self, PyObject* args) args, arg_parse("Obbby*:", __FUNCTION__), &obj, &apiIndex, &instanceIndex, &functionIndex, &data)) { return NULL; } + PyBufferRelease data_release(data); // Get the device handle if (!PyNeoDeviceEx_CheckExact(obj)) { return set_ics_exception(exception_runtime_error(), "Argument must be of type " MODULE_NAME ".PyNeoDeviceEx"); diff --git a/tests/buffer_exports_mock.c b/tests/buffer_exports_mock.c new file mode 100644 index 00000000..304d9d69 --- /dev/null +++ b/tests/buffer_exports_mock.c @@ -0,0 +1,51 @@ +#include + +#ifdef _WIN32 +#define API __declspec(dllexport) +#define CALL __stdcall +#else +#define API __attribute__((visibility("default"))) +#define CALL +#endif + +static int mode; +API void set_mode(int value) { mode = value; } + +#ifndef OMIT_BUFFER_APIS +/* The loader requests undecorated API names, including on 32-bit Windows. */ +#if defined(_WIN32) && defined(_M_IX86) +#pragma comment(linker, "/export:icsneoUartWrite=_icsneoUartWrite@24") +#pragma comment(linker, "/export:icsneoGenericAPISendCommand=_icsneoGenericAPISendCommand@28") +#pragma comment(linker, "/export:icsneoGetDeviceStatus=_icsneoGetDeviceStatus@12") +#pragma comment(linker, "/export:icsneoGetHWFirmwareInfo=_icsneoGetHWFirmwareInfo@8") +#pragma comment(linker, "/export:icsneoGetDLLFirmwareInfo=_icsneoGetDLLFirmwareInfo@8") +#pragma comment(linker, "/export:icsneoFlashAccessoryFirmware=_icsneoFlashAccessoryFirmware@12") +#endif + +/* No hardware calls or file access: mode 1 fails, mode 2 reports a mismatch. */ +API int CALL icsneoUartWrite(void* h, int port, const void* data, size_t len, + size_t* sent, unsigned char* flags) +{ + *sent = mode == 2 ? 0 : len; + return mode != 1; +} +API int CALL icsneoGenericAPISendCommand(void* h, unsigned char a, unsigned char i, + unsigned char f, void* data, unsigned int len, + unsigned char* error) +{ + *error = 7; + return mode != 1; +} +API int CALL icsneoGetDeviceStatus(void* h, void* status, size_t* size) +{ + if (mode == 2) *size = 0; + return mode != 1; +} +API int CALL icsneoGetHWFirmwareInfo(void* h, void* info) { return mode != 1; } +API int CALL icsneoGetDLLFirmwareInfo(void* h, void* info) { return mode != 1; } +API int CALL icsneoFlashAccessoryFirmware(void* h, void* params, int* error) +{ + *error = mode == 2 ? 0 : 1; + return mode != 1; +} +#endif diff --git a/tests/test_buffer_exports.py b/tests/test_buffer_exports.py new file mode 100644 index 00000000..a007e089 --- /dev/null +++ b/tests/test_buffer_exports.py @@ -0,0 +1,185 @@ +"""Exercise native buffer ownership in subprocesses using a hardware-free library.""" + +import os +from pathlib import Path +import shutil +import subprocess +import sys +import textwrap + +import pytest + + +@pytest.fixture(scope="module") +def mock_libraries(tmp_path_factory): + directory = tmp_path_factory.mktemp("buffer_exports") + source = Path(__file__).with_name("buffer_exports_mock.c") + libraries = [] + for missing in (False, True): + name = "missing" if missing else "mock" + defines = ["-DOMIT_BUFFER_APIS"] if missing else [] + if sys.platform == "win32": + compiler = shutil.which("clang-cl") + linker = shutil.which("lld-link") + if not compiler or not linker: + pytest.skip("LLVM clang-cl and lld-link are required for the mock DLL") + library = directory / (name + ".dll") + obj = directory / (name + ".obj") + target = "x86_64-pc-windows-msvc" if sys.maxsize > 2**32 else "i686-pc-windows-msvc" + subprocess.run([compiler, "--target=" + target, "/nologo", "/c", "/GS-", "/Zl", *defines, + str(source), "/Fo" + str(obj)], check=True, capture_output=True) + subprocess.run([linker, "/dll", "/noentry", "/nodefaultlib", + "/out:" + str(library), str(obj)], check=True, capture_output=True) + else: + compiler = shutil.which("cc") + if not compiler: + pytest.skip("A C compiler is required for the mock library") + library = directory / (name + (".dylib" if sys.platform == "darwin" else ".so")) + subprocess.run([compiler, "-shared", "-fPIC", *defines, str(source), "-o", str(library)], + check=True, capture_output=True) + libraries.append(library) + return libraries + + +def run_case(library, code): + # Keep the process-global library override out of other tests. Use the same + # import paths as pytest, including a locally built extension when selected. + env = dict(os.environ, PYTHONPATH=os.pathsep.join(map(str, sys.path))) + setup = f""" +import ctypes, gc, importlib, weakref +import ics +library = {str(library)!r} +mock = ctypes.CDLL(library) +mock.set_mode.argtypes = [ctypes.c_int] +mock.set_mode.restype = None +ics.override_library_name(library) +device = ics.PyNeoDeviceEx() +device._auto_handle_close = False +""" + result = subprocess.run([sys.executable, "-c", setup + textwrap.dedent(code)], + env=env, capture_output=True, text=True, timeout=30) + assert result.returncode == 0, result.stdout + result.stderr + + +@pytest.mark.parametrize("method,prefix,expected", [ + ("uart_write", "device, 0", 3), + ("generic_api_send_command", "device, 0, 0, 0", 7), + ("flash_accessory_firmware", "device", 1), +]) +@pytest.mark.parametrize("path", ["success", "failure", "missing", "invalid_device"]) +def test_input_exports_released(mock_libraries, method, prefix, expected, path): + library = mock_libraries[path == "missing"] + if path == "invalid_device": + prefix = prefix.replace("device", "None") + run_case(library, f""" + mock.set_mode({int(path == 'failure')}) + data = bytearray(b'abc') + try: + result = ics.{method}({prefix}, data) + except ics.RuntimeError: + assert {path != 'success'!r} + else: + assert {path == 'success'!r} + assert result == {expected} + data.extend(b'd') + assert data == b'abcd' + """) + + +@pytest.mark.parametrize("method,args", [ + ("uart_write", "device, 0, data"), + ("flash_accessory_firmware", "device, data"), +]) +def test_checked_error_releases_input(mock_libraries, method, args): + run_case(mock_libraries[0], f""" + mock.set_mode(2) + data = bytearray(b'abc') + try: + ics.{method}({args}) + except ics.RuntimeError: + pass + else: + raise AssertionError('expected checked error') + data.extend(b'd') + """) + + +@pytest.mark.parametrize("method,module_name", [ + ("get_device_status", "ics_device_status"), + ("get_hw_firmware_info", "st_api_firmware_info"), + ("get_dll_firmware_info", "st_api_firmware_info"), +]) +@pytest.mark.parametrize("path", ["success", "failure", "missing", "no_buffer"]) +def test_result_objects_collectable(mock_libraries, method, module_name, path): + run_case(mock_libraries[path == "missing"], f""" + module = importlib.import_module('ics.structures.{module_name}') + original = module.{module_name} + refs = [] + class NoBuffer: + pass + def create(): + obj = NoBuffer() if {path == 'no_buffer'!r} else original() + refs.append(weakref.ref(obj)) + return obj + module.{module_name} = create + mock.set_mode({int(path == 'failure')}) + try: + result = ics.{method}(device) + except (ics.RuntimeError, TypeError): + assert {path != 'success'!r} + else: + assert {path == 'success'!r} + assert isinstance(result, original) + del result + gc.collect() + assert all(ref() is None for ref in refs) + if {path != 'missing'!r}: + assert len(refs) == 1 + """) + + +def test_status_size_mismatch_collectable(mock_libraries): + run_case(mock_libraries[0], """ + module = importlib.import_module('ics.structures.ics_device_status') + original = module.ics_device_status + refs = [] + def create(): + obj = original() + refs.append(weakref.ref(obj)) + return obj + module.ics_device_status = create + mock.set_mode(2) + try: + ics.get_device_status(device, True) + except ics.RuntimeError: + pass + else: + raise AssertionError('expected size mismatch') + gc.collect() + assert refs[0]() is None + """) + + +def test_accessory_buffer_acquisition_failure(mock_libraries): + run_case(mock_libraries[0], """ + try: + ics.flash_accessory_firmware(device, object()) + except TypeError: + pass + else: + raise AssertionError('expected buffer acquisition failure') + """) + + +def test_uart_late_parse_failure_releases_once(mock_libraries): + run_case(mock_libraries[0], """ + data = bytearray(b'abc') + try: + ics.uart_write(device, 0, data, object()) + except TypeError: + pass + else: + raise AssertionError('expected argument conversion failure') + data.extend(b'd') + """) +