Honor RemotePort for SNMP sessions

netsnmp/client.py

@@ -140,6 +140,9 @@ def __init__(self, **args):
         # check for transports that may be tunneled
         transportCheck = re.compile('^(tls|dtls|ssh)');
         match = transportCheck.match(sess_args['DestHost'])
+        peer = sess_args['DestHost']
+        if not match and sess_args['RemotePort'] != 161:
+            peer = '{}:{}'.format(peer, sess_args['RemotePort'])
 
         err = None
 
@@ -163,7 +166,7 @@ def __init__(self, **args):
             elif sess_args['Version'] == 3:
                 self.sess_ptr = client_intf.session_v3(
                     sess_args['Version'],
-                    sess_args['DestHost'],
+                    peer,
                     sess_args['LocalPort'],
                     sess_args['Retries'],
                     sess_args['Timeout'],
@@ -182,7 +185,7 @@ def __init__(self, **args):
                 self.sess_ptr = client_intf.session(
                     sess_args['Version'],
                     sess_args['Community'],
-                    sess_args['DestHost'],
+                    peer,
                     sess_args['LocalPort'],
                     sess_args['Retries'],
                     sess_args['Timeout'])

netsnmp/tests/system/test_remote_port.py

@@ -0,0 +1,45 @@
+import socket
+import unittest
+
+import netsnmp
+
+
+class RemotePortTests(unittest.TestCase):
+    def assert_request_received(
+            self, version, dest_host, use_remote_port=False):
+        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_args = {
+                'Version': version,
+                'DestHost': dest_host.format(port=port),
+                'Community': 'public',
+                'Timeout': 1000,
+                'Retries': 0,
+            }
+            if use_remote_port:
+                session_args['RemotePort'] = port
+            session = netsnmp.Session(**session_args)
+            varlist = netsnmp.VarList(
+                netsnmp.Varbind('.1.3.6.1.2.1.1.1', '0'))
+
+            session.get(varlist)
+            request, _ = listener.recvfrom(65535)
+            self.assertTrue(request)
+
+    def test_remote_port_selects_destination_port(self):
+        for version in (1, 2, 3):
+            with self.subTest(version=version):
+                self.assert_request_received(
+                    version, '127.0.0.1', use_remote_port=True)
+
+    def test_dest_host_port_remains_supported(self):
+        for version in (1, 2, 3):
+            with self.subTest(version=version):
+                self.assert_request_received(
+                    version, '127.0.0.1:{port}')
+
+
+if __name__ == '__main__':
+    unittest.main()