diff --git a/.gitignore b/.gitignore index 0f22a4e..2461d0d 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,7 @@ build/ dist/ +__pycache__/ +*.py[co] .*.swp .*.swo *.egg-info/ diff --git a/netsnmp/client_intf.c b/netsnmp/client_intf.c index 14f943d..7d58302 100644 --- a/netsnmp/client_intf.c +++ b/netsnmp/client_intf.c @@ -1346,6 +1346,12 @@ netsnmp_create_session_v3(PyObject *self, PyObject *args) end: free (session.securityEngineID); free (session.contextEngineID); + /* snmp_sess_open() duplicates the security protocol OIDs into the session + it returns, so the copies made above are ours to release. Freeing them + here also covers the early "goto end" paths, where the privacy protocol + is rejected after the authentication protocol was already duplicated. */ + free (session.securityAuthProto); + free (session.securityPrivProto); if (PyErr_Occurred()) { return NULL; diff --git a/netsnmp/tests/system/common.py b/netsnmp/tests/system/common.py index 2c68326..934ba72 100644 --- a/netsnmp/tests/system/common.py +++ b/netsnmp/tests/system/common.py @@ -9,6 +9,21 @@ WRITE_ARGS = READ_ARGS.copy() WRITE_ARGS['Community'] = 'private' +# SNMPv3 authPriv credentials. Keep in sync with the createUser/rwuser +# directives in netsnmp/tests/system/snmpd.conf. +V3_ARGS = { + 'Version': 3, + 'DestHost': HOST, + 'SecLevel': 'authPriv', + 'SecName': 'testuser', + 'AuthProto': 'SHA', + 'AuthPass': 'auth_pass', + 'PrivProto': 'AES', + 'PrivPass': 'priv_pass', + 'Timeout': 1000000, + 'Retries': 0, +} + SYS_DESCR = '.1.3.6.1.2.1.1.1' SYS_UPTIME = '.1.3.6.1.2.1.1.3' SYS_LOCATION = '.1.3.6.1.2.1.1.6' diff --git a/netsnmp/tests/system/snmpd.conf b/netsnmp/tests/system/snmpd.conf index e986783..c93d84c 100644 --- a/netsnmp/tests/system/snmpd.conf +++ b/netsnmp/tests/system/snmpd.conf @@ -1,3 +1,17 @@ agentAddress udp:127.0.0.1:1161 rocommunity public 127.0.0.1 rwcommunity private 127.0.0.1 + +# SNMPv3 USM user used by test_v3.py. Passphrases must be at least 8 +# characters for net-snmp to accept them. Keep in sync with V3_ARGS in +# netsnmp/tests/system/common.py. +createUser testuser SHA "auth_pass" AES "priv_pass" +rwuser testuser authPriv + +# Only ever contacted with a deliberately wrong passphrase, by +# test_wrong_auth_password_is_rejected. It must not be used by any passing +# test: net-snmp caches localized USM keys per (engineID, username) for the +# life of the process, so a successful auth would prime that cache and mask +# the rejection we are asserting. +createUser baduser SHA "correct_auth_pass" AES "correct_priv_pass" +rouser baduser authPriv diff --git a/netsnmp/tests/system/test_v3.py b/netsnmp/tests/system/test_v3.py new file mode 100644 index 0000000..7d15319 --- /dev/null +++ b/netsnmp/tests/system/test_v3.py @@ -0,0 +1,73 @@ +import unittest + +import netsnmp + +from netsnmp.tests.system.common import ( + SYS_DESCR, + SYS_LOCATION, + SYSTEM, + V3_ARGS, + assert_value, +) + + +class SnmpV3Tests(unittest.TestCase): + """SNMPv3 authPriv coverage (USM authentication and privacy).""" + + def test_get(self): + values = netsnmp.snmpget(netsnmp.Varbind(SYS_DESCR, '0'), **V3_ARGS) + assert_value(self, values) + + def test_getnext(self): + values = netsnmp.snmpgetnext(netsnmp.Varbind(SYSTEM), **V3_ARGS) + assert_value(self, values) + + def test_getbulk(self): + session = netsnmp.Session(**V3_ARGS) + varlist = netsnmp.VarList(netsnmp.Varbind(SYSTEM)) + assert_value(self, session.getbulk(0, 4, varlist)) + self.assertGreater(len(varlist), 1) + + def test_set(self): + value = b'snmp-v3' + result = netsnmp.snmpset( + netsnmp.Varbind(SYS_LOCATION, '0', value, 'OCTETSTR'), **V3_ARGS) + self.assertEqual(result, 1) + self.assertEqual(netsnmp.snmpget( + netsnmp.Varbind(SYS_LOCATION, '0'), **V3_ARGS)[0], value) + + def test_walk(self): + varlist = netsnmp.VarList(netsnmp.Varbind(SYSTEM)) + assert_value(self, netsnmp.snmpwalk(varlist, **V3_ARGS)) + self.assertGreater(len(varlist), 1) + + def test_session_get(self): + session = netsnmp.Session(**V3_ARGS) + varlist = netsnmp.VarList(netsnmp.Varbind(SYS_DESCR, '0')) + assert_value(self, session.get(varlist)) + + def test_unknown_user_is_rejected(self): + session = netsnmp.Session(**dict(V3_ARGS, SecName='nosuchuser')) + varlist = netsnmp.VarList(netsnmp.Varbind(SYS_DESCR, '0')) + values = session.get(varlist) + self.assertFalse(values and values[0]) + + def test_wrong_auth_password_is_rejected(self): + """Proves the agent really enforces authPriv, so the tests above are + not silently passing over an unauthenticated session. + + Uses a dedicated user rather than the one above: net-snmp caches + localized USM keys per (engineID, username) process-wide, so reusing + a name that already authenticated successfully would reuse the good + key and the wrong passphrase would never reach the agent. + """ + session = netsnmp.Session( + **dict(V3_ARGS, SecName='baduser', AuthPass='wrong_auth_pass', + PrivPass='wrong_priv_pass')) + varlist = netsnmp.VarList(netsnmp.Varbind(SYS_DESCR, '0')) + values = session.get(varlist) + self.assertFalse(values and values[0]) + + +if __name__ == '__main__': + unittest.main() diff --git a/netsnmp/tests/test.py b/netsnmp/tests/test.py deleted file mode 100644 index 4787e9f..0000000 --- a/netsnmp/tests/test.py +++ /dev/null @@ -1,321 +0,0 @@ -""" Runs all unit tests for the netsnmp package. """ -# Copyright (c) 2006 Andy Gross. See LICENSE.txt for details. - -import sys -import unittest -import netsnmp -import time - -class BasicTests(unittest.TestCase): - def testFuncs(self): - print("") - var = netsnmp.Varbind('sysDescr.0') - var = netsnmp.Varbind('sysDescr','0') - var = netsnmp.Varbind( - '.iso.org.dod.internet.mgmt.mib-2.system.sysDescr','0') - var = netsnmp.Varbind( - '.iso.org.dod.internet.mgmt.mib-2.system.sysDescr.0') - var = netsnmp.Varbind('.1.3.6.1.2.1.1.1.0') - - var = netsnmp.Varbind('.1.3.6.1.2.1.1.1','0') - - print("---v1 GET tests -------------------------------------\n") - res = netsnmp.snmpget(var, - Version = 1, - DestHost='localhost', - Community='public') - - print("v1 snmpget result: ", res, "\n") - - print("v1 get var: ", var.tag, var.iid, "=", var.val, '(',var.type,')') - - print("---v1 GETNEXT tests-------------------------------------\n") - res = netsnmp.snmpgetnext(var, - Version = 1, - DestHost='localhost', - Community='public') - - print("v1 snmpgetnext result: ", res, "\n") - - print("v1 getnext var: ", var.tag, var.iid, "=", var.val, '(',var.type,')') - - print("---v1 SET tests-------------------------------------\n") - var = netsnmp.Varbind('sysLocation','0', 'my new location') - res = netsnmp.snmpset(var, - Version = 1, - DestHost='localhost', - Community='public') - - print("v1 snmpset result: ", res, "\n") - - print("v1 set var: ", var.tag, var.iid, "=", var.val, '(',var.type,')') - - print("---v1 walk tests-------------------------------------\n") - vars = netsnmp.VarList(netsnmp.Varbind('system')) - - print("v1 varlist walk in: ") - for var in vars: - print(" ",var.tag, var.iid, "=", var.val, '(',var.type,')') - - res = netsnmp.snmpwalk(vars, - Version = 1, - DestHost='localhost', - Community='public') - print("v1 snmpwalk result: ", res, "\n") - - for var in vars: - print(var.tag, var.iid, "=", var.val, '(',var.type,')') - - - print("---v1 walk 2-------------------------------------\n") - - print("v1 varbind walk in: ") - var = netsnmp.Varbind('system') - res = netsnmp.snmpwalk(var, - Version = 1, - DestHost='localhost', - Community='public') - print("v1 snmpwalk result (should be = orig): ", res, "\n") - - print(var.tag, var.iid, "=", var.val, '(',var.type,')') - - print("---v1 multi-varbind test-------------------------------------\n") - sess = netsnmp.Session(Version=1, - DestHost='localhost', - Community='public') - - vars = netsnmp.VarList(netsnmp.Varbind('sysUpTime', 0), - netsnmp.Varbind('sysContact', 0), - netsnmp.Varbind('sysLocation', 0)) - vals = sess.get(vars) - print("v1 sess.get result: ", vals, "\n") - - for var in vars: - print(var.tag, var.iid, "=", var.val, '(',var.type,')') - - vals = sess.getnext(vars) - print("v1 sess.getnext result: ", vals, "\n") - - for var in vars: - print(var.tag, var.iid, "=", var.val, '(',var.type,')') - - vars = netsnmp.VarList(netsnmp.Varbind('sysUpTime'), - netsnmp.Varbind('sysORLastChange'), - netsnmp.Varbind('sysORID'), - netsnmp.Varbind('sysORDescr'), - netsnmp.Varbind('sysORUpTime')) - - vals = sess.getbulk(2, 8, vars) - print("v1 sess.getbulk result: ", vals, "\n") - - for var in vars: - print(var.tag, var.iid, "=", var.val, '(',var.type,')') - - print("---v1 set2-------------------------------------\n") - - vars = netsnmp.VarList( - netsnmp.Varbind('sysLocation', '0', 'my newer location')) - res = sess.set(vars) - print("v1 sess.set result: ", res, "\n") - - print("---v1 walk3-------------------------------------\n") - vars = netsnmp.VarList(netsnmp.Varbind('system')) - - vals = sess.walk(vars) - print("v1 sess.walk result: ", vals, "\n") - - for var in vars: - print(" ",var.tag, var.iid, "=", var.val, '(',var.type,')') - - print("---v2c get-------------------------------------\n") - - sess = netsnmp.Session(Version=2, - DestHost='localhost', - Community='public') - - sess.UseEnums = 1 - sess.UseLongNames = 1 - - vars = netsnmp.VarList(netsnmp.Varbind('sysUpTime', 0), - netsnmp.Varbind('sysContact', 0), - netsnmp.Varbind('sysLocation', 0)) - vals = sess.get(vars) - print("v2 sess.get result: ", vals, "\n") - - print("---v2c getnext-------------------------------------\n") - - for var in vars: - print(var.tag, var.iid, "=", var.val, '(',var.type,')') - print("\n") - - vals = sess.getnext(vars) - print("v2 sess.getnext result: ", vals, "\n") - - for var in vars: - print(var.tag, var.iid, "=", var.val, '(',var.type,')') - print("\n") - - print("---v2c getbulk-------------------------------------\n") - - vars = netsnmp.VarList(netsnmp.Varbind('sysUpTime'), - netsnmp.Varbind('sysORLastChange'), - netsnmp.Varbind('sysORID'), - netsnmp.Varbind('sysORDescr'), - netsnmp.Varbind('sysORUpTime')) - - vals = sess.getbulk(2, 8, vars) - print("v2 sess.getbulk result: ", vals, "\n") - - for var in vars: - print(var.tag, var.iid, "=", var.val, '(',var.type,')') - print("\n") - - print("---v2c set-------------------------------------\n") - - vars = netsnmp.VarList( - netsnmp.Varbind('sysLocation','0','my even newer location')) - - res = sess.set(vars) - print("v2 sess.set result: ", res, "\n") - - print("---v2c walk-------------------------------------\n") - vars = netsnmp.VarList(netsnmp.Varbind('system')) - - vals = sess.walk(vars) - print("v2 sess.walk result: ", vals, "\n") - - for var in vars: - print(" ",var.tag, var.iid, "=", var.val, '(',var.type,')') - - print("---v3 setup-------------------------------------\n") - sess = netsnmp.Session(Version=3, - DestHost='localhost', - SecLevel='authPriv', - SecName='initial', - PrivPass='priv_pass', - AuthPass='auth_pass') - - sess.UseSprintValue = 1 - - vars = netsnmp.VarList(netsnmp.Varbind('sysUpTime', 0), - netsnmp.Varbind('sysContact', 0), - netsnmp.Varbind('sysLocation', 0)) - print("---v3 get-------------------------------------\n") - vals = sess.get(vars) - print("v3 sess.get result: ", vals, "\n") - - for var in vars: - print(var.tag, var.iid, "=", var.val, '(',var.type,')') - print("\n") - - print("---v3 getnext-------------------------------------\n") - - vals = sess.getnext(vars) - print("v3 sess.getnext result: ", vals, "\n") - - for var in vars: - print(var.tag, var.iid, "=", var.val, '(',var.type,')') - print("\n") - - vars = netsnmp.VarList(netsnmp.Varbind('sysUpTime'), - netsnmp.Varbind('sysORLastChange'), - netsnmp.Varbind('sysORID'), - netsnmp.Varbind('sysORDescr'), - netsnmp.Varbind('sysORUpTime')) - - vals = sess.getbulk(2, 8, vars) - print("v3 sess.getbulk result: ", vals, "\n") - - for var in vars: - print(var.tag, var.iid, "=", var.val, '(',var.type,')') - print("\n") - - print("---v3 set-------------------------------------\n") - - vars = netsnmp.VarList( - netsnmp.Varbind('sysLocation','0', 'my final destination')) - res = sess.set(vars) - print("v3 sess.set result: ", res, "\n") - - print("---v3 walk-------------------------------------\n") - vars = netsnmp.VarList(netsnmp.Varbind('system')) - - vals = sess.walk(vars) - print("v3 sess.walk result: ", vals, "\n") - - for var in vars: - print(" ",var.tag, var.iid, "=", var.val, '(',var.type,')') - - -class SetTests(unittest.TestCase): - def testFuncs(self): - print("\n-------------- SET Test Start ----------------------------\n") - - var = netsnmp.Varbind('sysUpTime','0') - res = netsnmp.snmpget(var, Version = 1, DestHost='localhost', - Community='public') - print("uptime = ", res[0]) - - - var = netsnmp.Varbind('versionRestartAgent','0', 1) - res = netsnmp.snmpset(var, Version = 1, DestHost='localhost', - Community='public') - - var = netsnmp.Varbind('sysUpTime','0') - res = netsnmp.snmpget(var, Version = 1, DestHost='localhost', - Community='public') - print("uptime = ", res[0]) - - var = netsnmp.Varbind('nsCacheEntry') - res = netsnmp.snmpgetnext(var, Version = 1, DestHost='localhost', - Community='public') - print("var = ", var.tag, var.iid, "=", var.val, '(',var.type,')') - - var.val = 65 - res = netsnmp.snmpset(var, Version = 1, DestHost='localhost', - Community='public') - res = netsnmp.snmpget(var, Version = 1, DestHost='localhost', - Community='public') - print("var = ", var.tag, var.iid, "=", var.val, '(',var.type,')') - - sess = netsnmp.Session(Version = 1, DestHost='localhost', - Community='public') - - vars = netsnmp.VarList(netsnmp.Varbind('.1.3.6.1.6.3.12.1.2.1.2.116.101.115.116','','.1.3.6.1.6.1.1'), - netsnmp.Varbind('.1.3.6.1.6.3.12.1.2.1.3.116.101.115.116','','1234'), - netsnmp.Varbind('.1.3.6.1.6.3.12.1.2.1.9.116.101.115.116','', 4)) - res = sess.set(vars) - - print("res = ", res) - - vars = netsnmp.VarList(netsnmp.Varbind('snmpTargetAddrTDomain'), - netsnmp.Varbind('snmpTargetAddrTAddress'), - netsnmp.Varbind('snmpTargetAddrRowStatus')) - - res = sess.getnext(vars) - - for var in vars: - print(var.tag, var.iid, "=", var.val, '(',var.type,')') - print("\n") - - vars = netsnmp.VarList(netsnmp.Varbind('.1.3.6.1.6.3.12.1.2.1.9.116.101.115.116','', 6)) - - res = sess.set(vars) - - print("res = ", res) - - vars = netsnmp.VarList(netsnmp.Varbind('snmpTargetAddrTDomain'), - netsnmp.Varbind('snmpTargetAddrTAddress'), - netsnmp.Varbind('snmpTargetAddrRowStatus')) - - res = sess.getnext(vars) - - for var in vars: - print(var.tag, var.iid, "=", var.val, '(',var.type,')') - print("\n") - - print("\n-------------- SET Test End ----------------------------\n") - - -if __name__=='__main__': - unittest.main() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..b66d069 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,32 @@ +[build-system] +requires = ["setuptools>=61"] +build-backend = "setuptools.build_meta" + +[project] +name = "python3-netsnmp" +version = "1.1a2" +description = "The Net-SNMP Python Interface" +readme = { file = "README", content-type = "text/plain" } +requires-python = ">=3.7" +license = { text = "BSD" } +authors = [ + { name = "G. S. Marzot", email = "giovanni.marzot@sparta.com" }, +] +maintainers = [ + { name = "Christian Svensson", email = "blue@cmd.nu" }, + { name = "Markus Häll", email = "soundgoof@gmail.com" }, +] +classifiers = [ + "License :: OSI Approved :: BSD License", + "Programming Language :: Python :: 3", + "Topic :: System :: Networking :: Monitoring", +] + +[project.urls] +Homepage = "http://www.net-snmp.org" +Source = "https://github.com/bluecmd/python3-netsnmp" + +# The C extension (net-snmp client bindings) is built from setup.py, which +# discovers the net-snmp libraries dynamically via net-snmp-config. +[tool.setuptools.packages.find] +include = ["netsnmp*"] diff --git a/setup.py b/setup.py index e3a65f6..46b5ddd 100644 --- a/setup.py +++ b/setup.py @@ -1,15 +1,15 @@ -from distutils.core import setup, Extension -from setuptools import setup, Extension, find_packages +"""Build the netsnmp C extension. + +Static project metadata lives in pyproject.toml. This file only remains for +the parts that can't be expressed declaratively: the net-snmp client library +is discovered at build time via net-snmp-config. +""" import os import re -import string import sys -args = sys.argv[:] -for arg in args: - if '--basedir' in arg: - basedir = string.split(arg,'=')[1] - sys.argv.remove(arg) +from setuptools import Extension, setup + def netsnmp_config(flag): cmd = 'net-snmp-config ' + flag @@ -20,6 +20,7 @@ def netsnmp_config(flag): cmd = 'sh -c "net-snmp-config %s"' % flag return os.popen(cmd).read() + netsnmp_libs = netsnmp_config('--libs') libdirs = re.findall(r" -L(\S+)", netsnmp_libs) incdirs = [] @@ -29,26 +30,13 @@ def netsnmp_config(flag): libs.append('ws2_32') setup( - name="python3-netsnmp", version="1.1a2", - description = 'The Net-SNMP Python Interface', - long_description = ''' -Python3 port of the official Net-SNMP Python bindings. - -Maintainer: Christian Svensson - -Source: https://github.com/bluecmd/python3-netsnmp -''', - author = 'G. S. Marzot', - author_email = 'giovanni.marzot@sparta.com', - url = 'http://www.net-snmp.org', - license="BSD", - packages=find_packages(), - test_suite = "netsnmp.tests.test", - - ext_modules = [ - Extension("netsnmp.client_intf", ["netsnmp/client_intf.c"], - library_dirs=libdirs, - include_dirs=incdirs, - libraries=libs ) - ] - ) + ext_modules=[ + Extension( + "netsnmp.client_intf", + ["netsnmp/client_intf.c"], + library_dirs=libdirs, + include_dirs=incdirs, + libraries=libs, + ) + ], +)