From 718a4ffd310c7eccb0878f57306b3fc78a344b32 Mon Sep 17 00:00:00 2001 From: soundgoof Date: Wed, 22 Jul 2026 21:55:42 +0200 Subject: [PATCH] Fix scalar values in SNMP set requests --- netsnmp/client_intf.c | 36 ++++++++++++--- netsnmp/tests/system/run.sh | 3 ++ netsnmp/tests/system/set_test_agent.py | 35 +++++++++++++++ netsnmp/tests/system/test_set.py | 62 ++++++++++++++++++++++++++ 4 files changed, 130 insertions(+), 6 deletions(-) create mode 100644 netsnmp/tests/system/set_test_agent.py diff --git a/netsnmp/client_intf.c b/netsnmp/client_intf.c index c6e058f..de2a0c0 100644 --- a/netsnmp/client_intf.c +++ b/netsnmp/client_intf.c @@ -1045,16 +1045,36 @@ py_netsnmp_attr_set_bytes(PyObject *obj, char *attr_name, } static int -py_netsnmp_attr_bytes(PyObject *obj, char * attr_name, char **val, - Py_ssize_t *len) +py_netsnmp_attr_value(PyObject *obj, char *attr_name, char **val, + Py_ssize_t *len, PyObject **value_obj) { *val = NULL; + *value_obj = NULL; if (obj && attr_name && PyObject_HasAttrString(obj, attr_name)) { PyObject *attr = PyObject_GetAttrString(obj, attr_name); if (attr) { - int retval; - retval = PyBytes_AsStringAndSize(attr, val, len); - Py_DECREF(attr); + int retval = -1; + + if (PyBytes_Check(attr)) { + retval = PyBytes_AsStringAndSize(attr, val, len); + } else { + PyObject *str_attr = PyObject_Str(attr); + Py_DECREF(attr); + attr = str_attr; + if (attr) { + const char *str_val = PyUnicode_AsUTF8AndSize(attr, len); + if (str_val) { + *val = (char *)str_val; + retval = 0; + } + } + } + + if (retval == 0) { + *value_obj = attr; + } else { + Py_XDECREF(attr); + } return retval; } } @@ -2594,6 +2614,7 @@ netsnmp_set(PyObject *self, PyObject *args) char err_str[STR_BUF_SIZE]; char *tmpstr; Py_ssize_t tmplen; + PyObject *value_obj = NULL; oid_arr = calloc(MAX_OID_LEN, sizeof(oid)); @@ -2654,7 +2675,8 @@ netsnmp_set(PyObject *self, PyObject *args) } } - if (py_netsnmp_attr_bytes(varbind, "val", &val, &tmplen) < 0) { + if (py_netsnmp_attr_value(varbind, "val", &val, &tmplen, + &value_obj) < 0) { snmp_free_pdu(pdu); goto done; } @@ -2675,6 +2697,7 @@ netsnmp_set(PyObject *self, PyObject *args) len = (int)tmplen; status = __add_var_val_str(pdu, oid_arr, oid_arr_len, (char *) tmp_val_str, len, type); + Py_CLEAR(value_obj); if (verbose && status == FAILURE) printf("error: set: adding variable/value to PDU"); @@ -2706,6 +2729,7 @@ netsnmp_set(PyObject *self, PyObject *args) ret = Py_BuildValue("i",0); /* fail, return False */ } done: + Py_XDECREF(value_obj); Py_XDECREF(varbind); SAFE_FREE(oid_arr); if (PyErr_Occurred()) diff --git a/netsnmp/tests/system/run.sh b/netsnmp/tests/system/run.sh index 6a24c37..270764f 100755 --- a/netsnmp/tests/system/run.sh +++ b/netsnmp/tests/system/run.sh @@ -20,6 +20,9 @@ python3 setup.py build_ext --inplace SNMPD_CONFIG="$RUNTIME_DIR/snmpd.conf" cat "$ROOT/netsnmp/tests/system/snmpd.conf" >"$SNMPD_CONFIG" printf '\npersistentDir %s\n' "$RUNTIME_DIR" >>"$SNMPD_CONFIG" +printf 'pass_persist .1.3.6.1.4.1.8072.9999.9999 %s %s\n' \ + "$(command -v python3)" \ + "$ROOT/netsnmp/tests/system/set_test_agent.py" >>"$SNMPD_CONFIG" snmpd -f -Lo -C \ -c "$SNMPD_CONFIG" \ diff --git a/netsnmp/tests/system/set_test_agent.py b/netsnmp/tests/system/set_test_agent.py new file mode 100644 index 0000000..769a64b --- /dev/null +++ b/netsnmp/tests/system/set_test_agent.py @@ -0,0 +1,35 @@ +import sys + + +ROOT = '.1.3.6.1.4.1.8072.9999.9999' +INTEGER_OID = ROOT + '.1.0' +VALUES = { + INTEGER_OID: ('integer', '42'), +} + + +def respond(*lines): + print(*lines, sep='\n', flush=True) + + +for command in sys.stdin: + command = command.rstrip('\n') + if command == 'PING': + respond('PONG') + elif command == 'get': + oid = sys.stdin.readline().rstrip('\n') + value = VALUES.get(oid) + if value is None: + respond('NONE') + else: + respond(oid, value[0], value[1]) + elif command == 'set': + oid = sys.stdin.readline().rstrip('\n') + value_type, value = sys.stdin.readline().rstrip('\n').split(' ', 1) + if oid not in VALUES: + respond('not-writable') + else: + VALUES[oid] = (value_type, value) + respond('DONE') + else: + respond('NONE') diff --git a/netsnmp/tests/system/test_set.py b/netsnmp/tests/system/test_set.py index 2ff80b2..dd7db2f 100644 --- a/netsnmp/tests/system/test_set.py +++ b/netsnmp/tests/system/test_set.py @@ -1,3 +1,4 @@ +import socket import unittest import netsnmp @@ -9,7 +10,68 @@ ) +TEST_INTEGER = '.1.3.6.1.4.1.8072.9999.9999.1' + + class SetTests(unittest.TestCase): + def test_accepts_canonical_scalar_values(self): + read_session = netsnmp.Session(**READ_ARGS) + write_session = netsnmp.Session(**WRITE_ARGS) + values = ( + (73, b'73'), + ('74', b'74'), + (b'75', b'75'), + ) + + for value, expected in values: + with self.subTest(value=value): + varbind = netsnmp.Varbind( + TEST_INTEGER, '0', value, 'INTEGER') + self.assertEqual( + write_session.set(netsnmp.VarList(varbind)), 1) + self.assertEqual( + read_session.get(netsnmp.VarList( + netsnmp.Varbind(TEST_INTEGER, '0'))), + (expected,)) + + def test_accepts_text_octet_string_values(self): + read_session = netsnmp.Session(**READ_ARGS) + write_session = netsnmp.Session(**WRITE_ARGS) + + for value, expected in ( + ('text-value', b'text-value'), + (b'bytes-value', b'bytes-value')): + with self.subTest(value=value): + varbind = netsnmp.Varbind( + SYS_LOCATION, '0', value, 'OCTETSTR') + self.assertEqual( + write_session.set(netsnmp.VarList(varbind)), 1) + self.assertEqual( + read_session.get(netsnmp.VarList( + netsnmp.Varbind(SYS_LOCATION, '0'))), + (expected,)) + + def test_accepts_binary_octet_string(self): + value = b'\x00\xff' + + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as listener: + listener.bind(('127.0.0.1', 0)) + listener.settimeout(1) + port = listener.getsockname()[1] + session = netsnmp.Session( + Version=2, + DestHost='127.0.0.1:{}'.format(port), + Community='public', + Timeout=1000, + Retries=0, + ) + varbind = netsnmp.Varbind( + SYS_LOCATION, '0', value, 'OCTETSTR') + + self.assertEqual(session.set(netsnmp.VarList(varbind)), 0) + request, _ = listener.recvfrom(65535) + self.assertIn(value, request) + def test_convenience_function(self): value = b'convenience-api' result = netsnmp.snmpset(