Fix scalar values in SNMP set requests

Author: SoundGoof

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())

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" \

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')

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(