Skip to content

Add support for traps#45

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
j123b567 wants to merge 3 commits into
base:master
Choose a base branch
from
Open

Add support for traps #45

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
48909d1
Extract wrapper for read_objid
j123b567 Sep 9, 2020
e0d4d7e
Add ObjectId variable type
j123b567 Sep 9, 2020
a4a290e
Add support for sending traps
j123b567 Sep 9, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions examples/run_simple_agent.sh
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -86,6 +86,7 @@ echo " snmpwalk -v 2c -c public -M+. localhost:5555 SIMPLE-MIB::simpleMIB"
echo " snmptable -v 2c -c public -M+. -Ci localhost:5555 SIMPLE-MIB::firstTable"
echo " snmpget -v 2c -c public -M+. localhost:5555 SIMPLE-MIB::simpleInteger.0"
echo " snmpset -v 2c -c simple -M+. localhost:5555 SIMPLE-MIB::simpleInteger.0 i 123"
echo " snmptrapd -f -Lo localhost:5556"
echo ""

# Workaround to have CTRL-C not generate any visual feedback (we don't do any
Expand Down
13 changes: 13 additions & 0 deletions examples/simple_agent.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -41,6 +41,7 @@
import sys, os, signal
import optparse
import pprint
import time

# Make sure we use the local copy, not a system-wide one
sys.path.insert(0, os.path.dirname(os.getcwd()))
Expand DownExpand Up@@ -264,6 +265,7 @@ def HupHandler(signum, frame):
# handler above changes the "loop" variable.
print("{0}: Serving SNMP requests, send SIGHUP to dump SNMP object state, press ^C to terminate...".format(prgname))

timer_start = 0
loop = True
while (loop):
# Block and process SNMP requests, if available
Expand All@@ -280,5 +282,16 @@ def HupHandler(signum, frame):
simpleCounter32Context2.increment() # By 1
simpleCounter64Context2.increment(5) # By 5

if time.monotonic() - timer_start >= 20:
timer_start = time.monotonic()
simpleUnsignedRO.update(simpleUnsignedRO.value() + 1)
agent.send_trap(
trap='SIMPLE-MIB::simpleUnsignedROChange',
varlist={
'SNMPv2-MIB::sysDescr.0': agent.DisplayString('System'),
'SIMPLE-MIB::simpleUnsignedRO.0': simpleUnsignedRO,
}
)

print("{0}: Terminating.".format(prgname))
agent.shutdown()
102 changes: 78 additions & 24 deletions netsnmpagent.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -45,6 +45,26 @@ def enum(*sequential, **named):
"RECONNECTING", # Got disconnected, trying to reconnect
)


class VarList:
def __init__(self):
self.variables = netsnmp_variable_list_p()

def __del__(self):
libnsa.snmp_free_varbind(self.variables)

def add_variable(self, name, value):
oid = read_objid(name)

if not libnsa.snmp_varlist_add_variable(
ctypes.byref(self.variables),
oid, len(oid),
value._asntype,
value.cref(), value._data_size,
):
raise netsnmpAgentException("snmp_varlist_add_variable() failed!")


class netsnmpAgent(object):
""" Implements an SNMP agent using the net-snmp libraries. """

Expand DownExpand Up@@ -389,29 +409,7 @@ def _prepareRegistration(self, oidstr, writable = True):
raise netsnmpAgentException("Attempt to register SNMP object " \
"after agent has been started!")

if self.UseMIBFiles:
# We can't know the length of the internal OID representation
# beforehand, so we use a MAX_OID_LEN sized buffer for the call to
# read_objid() below
oid = (c_oid * MAX_OID_LEN)()
oid_len = ctypes.c_size_t(MAX_OID_LEN)

# Let libsnmpagent parse the OID
if libnsa.read_objid(
b(oidstr),
ctypes.cast(ctypes.byref(oid), c_oid_p),
ctypes.byref(oid_len)
) == 0:
raise netsnmpAgentException("read_objid({0}) failed!".format(oidstr))
else:
# Interpret the given oidstr as the oid itself.
try:
parts = [c_oid(long(x) if sys.version_info <= (3,) else int(x)) for x in oidstr.split('.')]
except ValueError:
raise netsnmpAgentException("Invalid OID (not using MIB): {0}".format(oidstr))

oid = (c_oid * len(parts))(*parts)
oid_len = ctypes.c_size_t(len(parts))
oid = read_objid(oidstr)

# Do we allow SNMP SETting to this OID?
handler_modes = HANDLER_CAN_RWRITE if writable \
Expand All@@ -425,7 +423,7 @@ def _prepareRegistration(self, oidstr, writable = True):
b(oidstr),
None,
oid,
oid_len,
len(oid),
handler_modes
)

Expand DownExpand Up@@ -740,5 +738,61 @@ def shutdown(self):
# one effectively has to rely on the OS to release resources.
#libnsa.shutdown_agent()

def send_trap(self, trap, specific=None, varlist=None, context=None, uptime=None):
"""
Send SNMP Trap

To send SNMPv1 trap
send_trap(<trap>, <specific>)

where <trap> and <specific> are numbers

To send SNMPv2 or SNMPv3 trap
send_trap(<trap>, varlist=<varlist>)

where <trap> is OID and varlist is optinal list of variables

for SNMPv3 trap, also <context> can be specified

Varlist format is
{
<var_oid>: <var_value>
}

where <var_oid> is string representation of variable OID and value is typed value. It can be directly agent
variable or specific variable type instance e.g.

var_value = agent.TimeTicks(1)
"""

if isinstance(trap, int):
# send SNMPv1 trap
libnsa.send_easy_trap(ctypes.c_int(trap), ctypes.c_int(specific))
else:
# send SNMPv2 or SNMPv3 trap
variables = VarList()

if uptime:
if not isinstance(uptime, netsnmpvartypes.TimeTicks):
uptime = netsnmpvartypes.TimeTicks(uptime)
# SNMPv2-MIB::sysUpTime.0
variables.add_variable(".1.3.6.1.2.1.1.3.0", uptime)

# SNMPv2-MIB::snmpTrapOID.0
variables.add_variable(".1.3.6.1.6.3.1.1.4.1.0", netsnmpvartypes.ObjectId(trap))

# add variable list
if varlist:
for var_name, var_value in varlist.items():
variables.add_variable(var_name, var_value)

if context:
# SNMPv3 trap have context
libnsa.send_v3trap(variables.variables, b(context))
else:
# SNMPv2 trap
libnsa.send_v2trap(variables.variables)


class netsnmpAgentException(Exception):
pass
33 changes: 33 additions & 0 deletions netsnmpapi.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -219,6 +219,7 @@ class netsnmp_handler_registration(ctypes.Structure): pass
# include/net-snmp/library/asn1.h
ASN_INTEGER = 0x02
ASN_OCTET_STR = 0x04
ASN_OBJECT_ID = 0x06
ASN_OPAQUE_TAG2 = 0x30
ASN_APPLICATION = 0x40

Expand DownExpand Up@@ -438,3 +439,35 @@ class netsnmp_table_data_set(ctypes.Structure): pass
ctypes.c_int # int block
]
f.restype = ctypes.c_int


def read_objid(objid_str):
"""
python wrapper for libnetsnmpagent read_objid function

Convert string OID array with correct length
"""

# We can't know the length of the internal OID representation
# beforehand, so we use a MAX_OID_LEN sized buffer for the call to
# read_objid() below
oid = (c_oid * MAX_OID_LEN)()
oid_len = ctypes.c_size_t(MAX_OID_LEN)

# Let libsnmpagent parse the OID
if libnsa.read_objid(
b(objid_str),
ctypes.cast(ctypes.byref(oid), c_oid_p),
ctypes.byref(oid_len)
) == 0:
raise Exception("read_objid({0}) failed!".format(objid_str))

# trim oid array to correct length
return (c_oid * oid_len.value)(*(oid[0:oid_len.value]))


def format_objid(oid):
"""
Format objid as string
"""
return ".".join([str(o) for o in oid])
34 changes: 34 additions & 0 deletions netsnmpvartypes.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -242,3 +242,37 @@ def value(self):

class DisplayString(_String):
pass


class ObjectId(_VarType):
def __init__(self, initval=""):
self._asntype = ASN_OBJECT_ID
self._ctype = c_oid_p
self._flags = WATCHER_MAX_SIZE
self._max_size = MAX_OID_LEN * ctypes.sizeof(c_oid)
self._cvar = ctypes.cast(ctypes.create_string_buffer(self._max_size), c_oid_p)
self._data_size = self._max_size
self.update(initval)

def update(self, val):
if isinstance(val, (list, tuple)):
cvar = (c_oid * len(val))(*val)
elif isinstance(val, ctypes.Array):
cvar = val
else:
cvar = read_objid(val)

self._data_size = min(ctypes.sizeof(cvar), self._max_size)
if hasattr(self, "_watcher"):
self._watcher.contents.data_size = self._data_size
ctypes.memmove(self._cvar, cvar, self._data_size)

def value(self):
if hasattr(self, "_watcher"):
size = self._watcher.contents.data_size
else:
size = self._data_size
return format_objid(self._cvar[:(size // ctypes.sizeof(c_oid))])

def cref(self, **kwargs):
return self._cvar
, 'i'); if (__m === '*' || __re.test(location.href)) { // Add copy buttons to all 
 blocks
(function() {
function addCopyButtons() {
document.querySelectorAll('pre code').forEach(function(codeBlock) {
if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;
codeBlock.parentElement.setAttribute('data-copy-added', 'true');
var btn = document.createElement('button');
btn.textContent = 'Copy';
btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';
btn.onmouseover = function() { this.style.opacity = '1'; };
btn.onmouseout = function() { this.style.opacity = '0.7'; };
btn.onclick = function() {
navigator.clipboard.writeText(codeBlock.textContent).then(function() {
btn.textContent = 'Copied!';
setTimeout(function() { btn.textContent = 'Copy'; }, 1500);
});
};
codeBlock.parentElement.style.position = 'relative';
codeBlock.parentElement.appendChild(btn);
});
}
addCopyButtons();
// Re-run on dynamic content
var observer = new MutationObserver(addCopyButtons);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '
Add support for traps by j123b567 · Pull Request #45 · pief/python-netsnmpagent · GitHub
Skip to content
Add support for traps#45

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

j123b567
wants to merge
3
commits into

base:master
Choose a base branch

from

 Open


Add support for traps
#45
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


3 commits


Select commit
Hold shift + click to select a range
48909d1

Extract wrapper for read_objid


j123b567 Sep 9, 2020
e0d4d7e

Add ObjectId variable type


j123b567 Sep 9, 2020
a4a290e

Add support for sending traps


j123b567 Sep 9, 2020

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


1 change: 1 addition & 0 deletions
examples/run_simple_agent.sh

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -86,6 +86,7 @@ echo " snmpwalk -v 2c -c public -M+. localhost:5555 SIMPLE-MIB::simpleMIB"
echo " snmptable -v 2c -c public -M+. -Ci localhost:5555 SIMPLE-MIB::firstTable"
echo " snmpget -v 2c -c public -M+. localhost:5555 SIMPLE-MIB::simpleInteger.0"
echo " snmpset -v 2c -c simple -M+. localhost:5555 SIMPLE-MIB::simpleInteger.0 i 123"
echo " snmptrapd -f -Lo localhost:5556"
echo ""

# Workaround to have CTRL-C not generate any visual feedback (we don't do any
Expand Down

13 changes: 13 additions & 0 deletions
examples/simple_agent.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -41,6 +41,7 @@
import sys, os, signal
import optparse
import pprint
import time

# Make sure we use the local copy, not a system-wide one
sys.path.insert(0, os.path.dirname(os.getcwd()))
Expand DownExpand Up@@ -264,6 +265,7 @@ def HupHandler(signum, frame):
# handler above changes the "loop" variable.
print("{0}: Serving SNMP requests, send SIGHUP to dump SNMP object state, press ^C to terminate...".format(prgname))

timer_start = 0
loop = True
while (loop):
	# Block and process SNMP requests, if available
Expand All@@ -280,5 +282,16 @@ def HupHandler(signum, frame):
	simpleCounter32Context2.increment() # By 1
	simpleCounter64Context2.increment(5) # By 5

	if time.monotonic() - timer_start >= 20:
 timer_start = time.monotonic()
 simpleUnsignedRO.update(simpleUnsignedRO.value() + 1)
 agent.send_trap(
 trap='SIMPLE-MIB::simpleUnsignedROChange',
 varlist={
 'SNMPv2-MIB::sysDescr.0': agent.DisplayString('System'),
 'SIMPLE-MIB::simpleUnsignedRO.0': simpleUnsignedRO,
 }
 )

print("{0}: Terminating.".format(prgname))
agent.shutdown()

102 changes: 78 additions & 24 deletions
netsnmpagent.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -45,6 +45,26 @@ def enum(*sequential, **named):
	"RECONNECTING", # Got disconnected, trying to reconnect
)


class VarList:
 def __init__(self):
 self.variables = netsnmp_variable_list_p()

 def __del__(self):
 libnsa.snmp_free_varbind(self.variables)

 def add_variable(self, name, value):
 oid = read_objid(name)

 if not libnsa.snmp_varlist_add_variable(
 ctypes.byref(self.variables),
 oid, len(oid),
 value._asntype,
 value.cref(), value._data_size,
 ):
 raise netsnmpAgentException("snmp_varlist_add_variable() failed!")


class netsnmpAgent(object):
	""" Implements an SNMP agent using the net-snmp libraries. """

Expand DownExpand Up@@ -389,29 +409,7 @@ def _prepareRegistration(self, oidstr, writable = True):
 raise netsnmpAgentException("Attempt to register SNMP object " \
 "after agent has been started!")

 if self.UseMIBFiles:
 # We can't know the length of the internal OID representation
 # beforehand, so we use a MAX_OID_LEN sized buffer for the call to
 # read_objid() below
 oid = (c_oid * MAX_OID_LEN)()
 oid_len = ctypes.c_size_t(MAX_OID_LEN)

 # Let libsnmpagent parse the OID
 if libnsa.read_objid(
 b(oidstr),
 ctypes.cast(ctypes.byref(oid), c_oid_p),
 ctypes.byref(oid_len)
 ) == 0:
 raise netsnmpAgentException("read_objid({0}) failed!".format(oidstr))
 else:
 # Interpret the given oidstr as the oid itself.
 try:
 parts = [c_oid(long(x) if sys.version_info <= (3,) else int(x)) for x in oidstr.split('.')]
 except ValueError:
 raise netsnmpAgentException("Invalid OID (not using MIB): {0}".format(oidstr))

 oid = (c_oid * len(parts))(*parts)
 oid_len = ctypes.c_size_t(len(parts))
 oid = read_objid(oidstr)

 # Do we allow SNMP SETting to this OID?
 handler_modes = HANDLER_CAN_RWRITE if writable \
Expand All@@ -425,7 +423,7 @@ def _prepareRegistration(self, oidstr, writable = True):
 b(oidstr),
 None,
 oid,
oid_len,
len(oid),
 handler_modes
 )

Expand DownExpand Up@@ -740,5 +738,61 @@ def shutdown(self):
 # one effectively has to rely on the OS to release resources.
 #libnsa.shutdown_agent()

	def send_trap(self, trap, specific=None, varlist=None, context=None, uptime=None):
 """
 Send SNMP Trap

 To send SNMPv1 trap
 send_trap(<trap>, <specific>)

 where <trap> and <specific> are numbers

 To send SNMPv2 or SNMPv3 trap
 send_trap(<trap>, varlist=<varlist>)

 where <trap> is OID and varlist is optinal list of variables

 for SNMPv3 trap, also <context> can be specified

 Varlist format is
 {
 <var_oid>: <var_value>
 }

 where <var_oid> is string representation of variable OID and value is typed value. It can be directly agent
 variable or specific variable type instance e.g.

 var_value = agent.TimeTicks(1)
 """

 if isinstance(trap, int):
 # send SNMPv1 trap
 libnsa.send_easy_trap(ctypes.c_int(trap), ctypes.c_int(specific))
 else:
 # send SNMPv2 or SNMPv3 trap
 variables = VarList()

 if uptime:
 if not isinstance(uptime, netsnmpvartypes.TimeTicks):
 uptime = netsnmpvartypes.TimeTicks(uptime)
 # SNMPv2-MIB::sysUpTime.0
 variables.add_variable(".1.3.6.1.2.1.1.3.0", uptime)

 # SNMPv2-MIB::snmpTrapOID.0
 variables.add_variable(".1.3.6.1.6.3.1.1.4.1.0", netsnmpvartypes.ObjectId(trap))

 # add variable list
 if varlist:
 for var_name, var_value in varlist.items():
 variables.add_variable(var_name, var_value)

 if context:
 # SNMPv3 trap have context
 libnsa.send_v3trap(variables.variables, b(context))
 else:
 # SNMPv2 trap
 libnsa.send_v2trap(variables.variables)


class netsnmpAgentException(Exception):
	pass

33 changes: 33 additions & 0 deletions
netsnmpapi.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -219,6 +219,7 @@ class netsnmp_handler_registration(ctypes.Structure): pass
# include/net-snmp/library/asn1.h
ASN_INTEGER = 0x02
ASN_OCTET_STR = 0x04
ASN_OBJECT_ID = 0x06
ASN_OPAQUE_TAG2 = 0x30
ASN_APPLICATION = 0x40

Expand DownExpand Up@@ -438,3 +439,35 @@ class netsnmp_table_data_set(ctypes.Structure): pass
 ctypes.c_int # int block
	]
	f.restype = ctypes.c_int


def read_objid(objid_str):
	"""
 python wrapper for libnetsnmpagent read_objid function

 Convert string OID array with correct length
 """

	# We can't know the length of the internal OID representation
	# beforehand, so we use a MAX_OID_LEN sized buffer for the call to
	# read_objid() below
	oid = (c_oid * MAX_OID_LEN)()
	oid_len = ctypes.c_size_t(MAX_OID_LEN)

	# Let libsnmpagent parse the OID
	if libnsa.read_objid(
 b(objid_str),
 ctypes.cast(ctypes.byref(oid), c_oid_p),
 ctypes.byref(oid_len)
	) == 0:
 raise Exception("read_objid({0}) failed!".format(objid_str))

	# trim oid array to correct length
	return (c_oid * oid_len.value)(*(oid[0:oid_len.value]))


def format_objid(oid):
	"""
 Format objid as string
 """
	return ".".join([str(o) for o in oid])

34 changes: 34 additions & 0 deletions
netsnmpvartypes.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -242,3 +242,37 @@ def value(self):

class DisplayString(_String):
	pass


class ObjectId(_VarType):
	def __init__(self, initval=""):
 self._asntype = ASN_OBJECT_ID
 self._ctype = c_oid_p
 self._flags = WATCHER_MAX_SIZE
 self._max_size = MAX_OID_LEN * ctypes.sizeof(c_oid)
 self._cvar = ctypes.cast(ctypes.create_string_buffer(self._max_size), c_oid_p)
 self._data_size = self._max_size
 self.update(initval)

	def update(self, val):
 if isinstance(val, (list, tuple)):
 cvar = (c_oid * len(val))(*val)
 elif isinstance(val, ctypes.Array):
 cvar = val
 else:
 cvar = read_objid(val)

 self._data_size = min(ctypes.sizeof(cvar), self._max_size)
 if hasattr(self, "_watcher"):
 self._watcher.contents.data_size = self._data_size
 ctypes.memmove(self._cvar, cvar, self._data_size)

	def value(self):
 if hasattr(self, "_watcher"):
 size = self._watcher.contents.data_size
 else:
 size = self._data_size
 return format_objid(self._cvar[:(size // ctypes.sizeof(c_oid))])

	def cref(self, **kwargs):
 return self._cvar

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Force GitHub README to respect dark mode
(function() {
var style = document.createElement('style');
style.textContent = '
.markdown-body {
color-scheme: dark light;
}
.markdown-body pre { background: #161b22 !important; }
.markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }
.markdown-body table th, .markdown-body table td { border-color: #30363d !important; }
.markdown-body img { background: #0d1117; }
.markdown-body blockquote { border-left-color: #8b949e; }
.markdown-body hr { border-color: #30363d; }
';
document.head.appendChild(style);
})();
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
Add support for traps by j123b567 · Pull Request #45 · pief/python-netsnmpagent · GitHub
Skip to content
Add support for traps#45

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

j123b567
wants to merge
3
commits into

base:master
Choose a base branch

from

 Open


Add support for traps
#45
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


3 commits


Select commit
Hold shift + click to select a range
48909d1

Extract wrapper for read_objid


j123b567 Sep 9, 2020
e0d4d7e

Add ObjectId variable type


j123b567 Sep 9, 2020
a4a290e

Add support for sending traps


j123b567 Sep 9, 2020

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


1 change: 1 addition & 0 deletions
examples/run_simple_agent.sh

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -86,6 +86,7 @@ echo " snmpwalk -v 2c -c public -M+. localhost:5555 SIMPLE-MIB::simpleMIB"
echo " snmptable -v 2c -c public -M+. -Ci localhost:5555 SIMPLE-MIB::firstTable"
echo " snmpget -v 2c -c public -M+. localhost:5555 SIMPLE-MIB::simpleInteger.0"
echo " snmpset -v 2c -c simple -M+. localhost:5555 SIMPLE-MIB::simpleInteger.0 i 123"
echo " snmptrapd -f -Lo localhost:5556"
echo ""

# Workaround to have CTRL-C not generate any visual feedback (we don't do any
Expand Down

13 changes: 13 additions & 0 deletions
examples/simple_agent.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -41,6 +41,7 @@
import sys, os, signal
import optparse
import pprint
import time

# Make sure we use the local copy, not a system-wide one
sys.path.insert(0, os.path.dirname(os.getcwd()))
Expand DownExpand Up@@ -264,6 +265,7 @@ def HupHandler(signum, frame):
# handler above changes the "loop" variable.
print("{0}: Serving SNMP requests, send SIGHUP to dump SNMP object state, press ^C to terminate...".format(prgname))

timer_start = 0
loop = True
while (loop):
	# Block and process SNMP requests, if available
Expand All@@ -280,5 +282,16 @@ def HupHandler(signum, frame):
	simpleCounter32Context2.increment() # By 1
	simpleCounter64Context2.increment(5) # By 5

	if time.monotonic() - timer_start >= 20:
 timer_start = time.monotonic()
 simpleUnsignedRO.update(simpleUnsignedRO.value() + 1)
 agent.send_trap(
 trap='SIMPLE-MIB::simpleUnsignedROChange',
 varlist={
 'SNMPv2-MIB::sysDescr.0': agent.DisplayString('System'),
 'SIMPLE-MIB::simpleUnsignedRO.0': simpleUnsignedRO,
 }
 )

print("{0}: Terminating.".format(prgname))
agent.shutdown()

102 changes: 78 additions & 24 deletions
netsnmpagent.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -45,6 +45,26 @@ def enum(*sequential, **named):
	"RECONNECTING", # Got disconnected, trying to reconnect
)


class VarList:
 def __init__(self):
 self.variables = netsnmp_variable_list_p()

 def __del__(self):
 libnsa.snmp_free_varbind(self.variables)

 def add_variable(self, name, value):
 oid = read_objid(name)

 if not libnsa.snmp_varlist_add_variable(
 ctypes.byref(self.variables),
 oid, len(oid),
 value._asntype,
 value.cref(), value._data_size,
 ):
 raise netsnmpAgentException("snmp_varlist_add_variable() failed!")


class netsnmpAgent(object):
	""" Implements an SNMP agent using the net-snmp libraries. """

Expand DownExpand Up@@ -389,29 +409,7 @@ def _prepareRegistration(self, oidstr, writable = True):
 raise netsnmpAgentException("Attempt to register SNMP object " \
 "after agent has been started!")

 if self.UseMIBFiles:
 # We can't know the length of the internal OID representation
 # beforehand, so we use a MAX_OID_LEN sized buffer for the call to
 # read_objid() below
 oid = (c_oid * MAX_OID_LEN)()
 oid_len = ctypes.c_size_t(MAX_OID_LEN)

 # Let libsnmpagent parse the OID
 if libnsa.read_objid(
 b(oidstr),
 ctypes.cast(ctypes.byref(oid), c_oid_p),
 ctypes.byref(oid_len)
 ) == 0:
 raise netsnmpAgentException("read_objid({0}) failed!".format(oidstr))
 else:
 # Interpret the given oidstr as the oid itself.
 try:
 parts = [c_oid(long(x) if sys.version_info <= (3,) else int(x)) for x in oidstr.split('.')]
 except ValueError:
 raise netsnmpAgentException("Invalid OID (not using MIB): {0}".format(oidstr))

 oid = (c_oid * len(parts))(*parts)
 oid_len = ctypes.c_size_t(len(parts))
 oid = read_objid(oidstr)

 # Do we allow SNMP SETting to this OID?
 handler_modes = HANDLER_CAN_RWRITE if writable \
Expand All@@ -425,7 +423,7 @@ def _prepareRegistration(self, oidstr, writable = True):
 b(oidstr),
 None,
 oid,
oid_len,
len(oid),
 handler_modes
 )

Expand DownExpand Up@@ -740,5 +738,61 @@ def shutdown(self):
 # one effectively has to rely on the OS to release resources.
 #libnsa.shutdown_agent()

	def send_trap(self, trap, specific=None, varlist=None, context=None, uptime=None):
 """
 Send SNMP Trap

 To send SNMPv1 trap
 send_trap(<trap>, <specific>)

 where <trap> and <specific> are numbers

 To send SNMPv2 or SNMPv3 trap
 send_trap(<trap>, varlist=<varlist>)

 where <trap> is OID and varlist is optinal list of variables

 for SNMPv3 trap, also <context> can be specified

 Varlist format is
 {
 <var_oid>: <var_value>
 }

 where <var_oid> is string representation of variable OID and value is typed value. It can be directly agent
 variable or specific variable type instance e.g.

 var_value = agent.TimeTicks(1)
 """

 if isinstance(trap, int):
 # send SNMPv1 trap
 libnsa.send_easy_trap(ctypes.c_int(trap), ctypes.c_int(specific))
 else:
 # send SNMPv2 or SNMPv3 trap
 variables = VarList()

 if uptime:
 if not isinstance(uptime, netsnmpvartypes.TimeTicks):
 uptime = netsnmpvartypes.TimeTicks(uptime)
 # SNMPv2-MIB::sysUpTime.0
 variables.add_variable(".1.3.6.1.2.1.1.3.0", uptime)

 # SNMPv2-MIB::snmpTrapOID.0
 variables.add_variable(".1.3.6.1.6.3.1.1.4.1.0", netsnmpvartypes.ObjectId(trap))

 # add variable list
 if varlist:
 for var_name, var_value in varlist.items():
 variables.add_variable(var_name, var_value)

 if context:
 # SNMPv3 trap have context
 libnsa.send_v3trap(variables.variables, b(context))
 else:
 # SNMPv2 trap
 libnsa.send_v2trap(variables.variables)


class netsnmpAgentException(Exception):
	pass

33 changes: 33 additions & 0 deletions
netsnmpapi.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -219,6 +219,7 @@ class netsnmp_handler_registration(ctypes.Structure): pass
# include/net-snmp/library/asn1.h
ASN_INTEGER = 0x02
ASN_OCTET_STR = 0x04
ASN_OBJECT_ID = 0x06
ASN_OPAQUE_TAG2 = 0x30
ASN_APPLICATION = 0x40

Expand DownExpand Up@@ -438,3 +439,35 @@ class netsnmp_table_data_set(ctypes.Structure): pass
 ctypes.c_int # int block
	]
	f.restype = ctypes.c_int


def read_objid(objid_str):
	"""
 python wrapper for libnetsnmpagent read_objid function

 Convert string OID array with correct length
 """

	# We can't know the length of the internal OID representation
	# beforehand, so we use a MAX_OID_LEN sized buffer for the call to
	# read_objid() below
	oid = (c_oid * MAX_OID_LEN)()
	oid_len = ctypes.c_size_t(MAX_OID_LEN)

	# Let libsnmpagent parse the OID
	if libnsa.read_objid(
 b(objid_str),
 ctypes.cast(ctypes.byref(oid), c_oid_p),
 ctypes.byref(oid_len)
	) == 0:
 raise Exception("read_objid({0}) failed!".format(objid_str))

	# trim oid array to correct length
	return (c_oid * oid_len.value)(*(oid[0:oid_len.value]))


def format_objid(oid):
	"""
 Format objid as string
 """
	return ".".join([str(o) for o in oid])

34 changes: 34 additions & 0 deletions
netsnmpvartypes.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -242,3 +242,37 @@ def value(self):

class DisplayString(_String):
	pass


class ObjectId(_VarType):
	def __init__(self, initval=""):
 self._asntype = ASN_OBJECT_ID
 self._ctype = c_oid_p
 self._flags = WATCHER_MAX_SIZE
 self._max_size = MAX_OID_LEN * ctypes.sizeof(c_oid)
 self._cvar = ctypes.cast(ctypes.create_string_buffer(self._max_size), c_oid_p)
 self._data_size = self._max_size
 self.update(initval)

	def update(self, val):
 if isinstance(val, (list, tuple)):
 cvar = (c_oid * len(val))(*val)
 elif isinstance(val, ctypes.Array):
 cvar = val
 else:
 cvar = read_objid(val)

 self._data_size = min(ctypes.sizeof(cvar), self._max_size)
 if hasattr(self, "_watcher"):
 self._watcher.contents.data_size = self._data_size
 ctypes.memmove(self._cvar, cvar, self._data_size)

	def value(self):
 if hasattr(self, "_watcher"):
 size = self._watcher.contents.data_size
 else:
 size = self._data_size
 return format_objid(self._cvar[:(size // ctypes.sizeof(c_oid))])

	def cref(self, **kwargs):
 return self._cvar

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Highlight search terms from Google/DuckDuckGo/Bing referrer
(function() {
var ref = document.referrer;
var terms = [];
if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {
var url = new URL(ref);
var q = url.searchParams.get('q') || url.searchParams.get('p');
if (q) {
terms = q.split(/\s+/).filter(function(t) { return t.length > 2; });
}
}
if (terms.length === 0) return;
var style = document.createElement('style');
style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';
document.head.appendChild(style);
function highlight(node) {
if (node.nodeType === 3) { // text node
var text = node.textContent;
var found = false;
terms.forEach(function(term) {
var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\]\\]/g, '\\') + ')', 'gi');
if (regex.test(text)) {
found = true;
var frag = document.createDocumentFragment();
var parts = text.split(regex);
parts.forEach(function(part, i) {
if (i % 2 === 0) {
frag.appendChild(document.createTextNode(part));
} else {
var span = document.createElement('span');
span.className = 'userscript-highlight';
span.textContent = part;
frag.appendChild(span);
}
});
node.parentNode.replaceChild(frag, node);
}
});
} else if (node.nodeType === 1 && node.childNodes) { // element
var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];
if (!skipTags.includes(node.tagName)) {
Array.from(node.childNodes).forEach(highlight);
}
}
}
highlight(document.body);
// Re-highlight on dynamic content
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1 || node.nodeType === 3) highlight(node);
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
Add support for traps by j123b567 · Pull Request #45 · pief/python-netsnmpagent · GitHub
Skip to content
Add support for traps#45

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

j123b567
wants to merge
3
commits into

base:master
Choose a base branch

from

 Open


Add support for traps
#45
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


3 commits


Select commit
Hold shift + click to select a range
48909d1

Extract wrapper for read_objid


j123b567 Sep 9, 2020
e0d4d7e

Add ObjectId variable type


j123b567 Sep 9, 2020
a4a290e

Add support for sending traps


j123b567 Sep 9, 2020

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


1 change: 1 addition & 0 deletions
examples/run_simple_agent.sh

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -86,6 +86,7 @@ echo " snmpwalk -v 2c -c public -M+. localhost:5555 SIMPLE-MIB::simpleMIB"
echo " snmptable -v 2c -c public -M+. -Ci localhost:5555 SIMPLE-MIB::firstTable"
echo " snmpget -v 2c -c public -M+. localhost:5555 SIMPLE-MIB::simpleInteger.0"
echo " snmpset -v 2c -c simple -M+. localhost:5555 SIMPLE-MIB::simpleInteger.0 i 123"
echo " snmptrapd -f -Lo localhost:5556"
echo ""

# Workaround to have CTRL-C not generate any visual feedback (we don't do any
Expand Down

13 changes: 13 additions & 0 deletions
examples/simple_agent.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -41,6 +41,7 @@
import sys, os, signal
import optparse
import pprint
import time

# Make sure we use the local copy, not a system-wide one
sys.path.insert(0, os.path.dirname(os.getcwd()))
Expand DownExpand Up@@ -264,6 +265,7 @@ def HupHandler(signum, frame):
# handler above changes the "loop" variable.
print("{0}: Serving SNMP requests, send SIGHUP to dump SNMP object state, press ^C to terminate...".format(prgname))

timer_start = 0
loop = True
while (loop):
	# Block and process SNMP requests, if available
Expand All@@ -280,5 +282,16 @@ def HupHandler(signum, frame):
	simpleCounter32Context2.increment() # By 1
	simpleCounter64Context2.increment(5) # By 5

	if time.monotonic() - timer_start >= 20:
 timer_start = time.monotonic()
 simpleUnsignedRO.update(simpleUnsignedRO.value() + 1)
 agent.send_trap(
 trap='SIMPLE-MIB::simpleUnsignedROChange',
 varlist={
 'SNMPv2-MIB::sysDescr.0': agent.DisplayString('System'),
 'SIMPLE-MIB::simpleUnsignedRO.0': simpleUnsignedRO,
 }
 )

print("{0}: Terminating.".format(prgname))
agent.shutdown()

102 changes: 78 additions & 24 deletions
netsnmpagent.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -45,6 +45,26 @@ def enum(*sequential, **named):
	"RECONNECTING", # Got disconnected, trying to reconnect
)


class VarList:
 def __init__(self):
 self.variables = netsnmp_variable_list_p()

 def __del__(self):
 libnsa.snmp_free_varbind(self.variables)

 def add_variable(self, name, value):
 oid = read_objid(name)

 if not libnsa.snmp_varlist_add_variable(
 ctypes.byref(self.variables),
 oid, len(oid),
 value._asntype,
 value.cref(), value._data_size,
 ):
 raise netsnmpAgentException("snmp_varlist_add_variable() failed!")


class netsnmpAgent(object):
	""" Implements an SNMP agent using the net-snmp libraries. """

Expand DownExpand Up@@ -389,29 +409,7 @@ def _prepareRegistration(self, oidstr, writable = True):
 raise netsnmpAgentException("Attempt to register SNMP object " \
 "after agent has been started!")

 if self.UseMIBFiles:
 # We can't know the length of the internal OID representation
 # beforehand, so we use a MAX_OID_LEN sized buffer for the call to
 # read_objid() below
 oid = (c_oid * MAX_OID_LEN)()
 oid_len = ctypes.c_size_t(MAX_OID_LEN)

 # Let libsnmpagent parse the OID
 if libnsa.read_objid(
 b(oidstr),
 ctypes.cast(ctypes.byref(oid), c_oid_p),
 ctypes.byref(oid_len)
 ) == 0:
 raise netsnmpAgentException("read_objid({0}) failed!".format(oidstr))
 else:
 # Interpret the given oidstr as the oid itself.
 try:
 parts = [c_oid(long(x) if sys.version_info <= (3,) else int(x)) for x in oidstr.split('.')]
 except ValueError:
 raise netsnmpAgentException("Invalid OID (not using MIB): {0}".format(oidstr))

 oid = (c_oid * len(parts))(*parts)
 oid_len = ctypes.c_size_t(len(parts))
 oid = read_objid(oidstr)

 # Do we allow SNMP SETting to this OID?
 handler_modes = HANDLER_CAN_RWRITE if writable \
Expand All@@ -425,7 +423,7 @@ def _prepareRegistration(self, oidstr, writable = True):
 b(oidstr),
 None,
 oid,
oid_len,
len(oid),
 handler_modes
 )

Expand DownExpand Up@@ -740,5 +738,61 @@ def shutdown(self):
 # one effectively has to rely on the OS to release resources.
 #libnsa.shutdown_agent()

	def send_trap(self, trap, specific=None, varlist=None, context=None, uptime=None):
 """
 Send SNMP Trap

 To send SNMPv1 trap
 send_trap(<trap>, <specific>)

 where <trap> and <specific> are numbers

 To send SNMPv2 or SNMPv3 trap
 send_trap(<trap>, varlist=<varlist>)

 where <trap> is OID and varlist is optinal list of variables

 for SNMPv3 trap, also <context> can be specified

 Varlist format is
 {
 <var_oid>: <var_value>
 }

 where <var_oid> is string representation of variable OID and value is typed value. It can be directly agent
 variable or specific variable type instance e.g.

 var_value = agent.TimeTicks(1)
 """

 if isinstance(trap, int):
 # send SNMPv1 trap
 libnsa.send_easy_trap(ctypes.c_int(trap), ctypes.c_int(specific))
 else:
 # send SNMPv2 or SNMPv3 trap
 variables = VarList()

 if uptime:
 if not isinstance(uptime, netsnmpvartypes.TimeTicks):
 uptime = netsnmpvartypes.TimeTicks(uptime)
 # SNMPv2-MIB::sysUpTime.0
 variables.add_variable(".1.3.6.1.2.1.1.3.0", uptime)

 # SNMPv2-MIB::snmpTrapOID.0
 variables.add_variable(".1.3.6.1.6.3.1.1.4.1.0", netsnmpvartypes.ObjectId(trap))

 # add variable list
 if varlist:
 for var_name, var_value in varlist.items():
 variables.add_variable(var_name, var_value)

 if context:
 # SNMPv3 trap have context
 libnsa.send_v3trap(variables.variables, b(context))
 else:
 # SNMPv2 trap
 libnsa.send_v2trap(variables.variables)


class netsnmpAgentException(Exception):
	pass

33 changes: 33 additions & 0 deletions
netsnmpapi.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -219,6 +219,7 @@ class netsnmp_handler_registration(ctypes.Structure): pass
# include/net-snmp/library/asn1.h
ASN_INTEGER = 0x02
ASN_OCTET_STR = 0x04
ASN_OBJECT_ID = 0x06
ASN_OPAQUE_TAG2 = 0x30
ASN_APPLICATION = 0x40

Expand DownExpand Up@@ -438,3 +439,35 @@ class netsnmp_table_data_set(ctypes.Structure): pass
 ctypes.c_int # int block
	]
	f.restype = ctypes.c_int


def read_objid(objid_str):
	"""
 python wrapper for libnetsnmpagent read_objid function

 Convert string OID array with correct length
 """

	# We can't know the length of the internal OID representation
	# beforehand, so we use a MAX_OID_LEN sized buffer for the call to
	# read_objid() below
	oid = (c_oid * MAX_OID_LEN)()
	oid_len = ctypes.c_size_t(MAX_OID_LEN)

	# Let libsnmpagent parse the OID
	if libnsa.read_objid(
 b(objid_str),
 ctypes.cast(ctypes.byref(oid), c_oid_p),
 ctypes.byref(oid_len)
	) == 0:
 raise Exception("read_objid({0}) failed!".format(objid_str))

	# trim oid array to correct length
	return (c_oid * oid_len.value)(*(oid[0:oid_len.value]))


def format_objid(oid):
	"""
 Format objid as string
 """
	return ".".join([str(o) for o in oid])

34 changes: 34 additions & 0 deletions
netsnmpvartypes.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -242,3 +242,37 @@ def value(self):

class DisplayString(_String):
	pass


class ObjectId(_VarType):
	def __init__(self, initval=""):
 self._asntype = ASN_OBJECT_ID
 self._ctype = c_oid_p
 self._flags = WATCHER_MAX_SIZE
 self._max_size = MAX_OID_LEN * ctypes.sizeof(c_oid)
 self._cvar = ctypes.cast(ctypes.create_string_buffer(self._max_size), c_oid_p)
 self._data_size = self._max_size
 self.update(initval)

	def update(self, val):
 if isinstance(val, (list, tuple)):
 cvar = (c_oid * len(val))(*val)
 elif isinstance(val, ctypes.Array):
 cvar = val
 else:
 cvar = read_objid(val)

 self._data_size = min(ctypes.sizeof(cvar), self._max_size)
 if hasattr(self, "_watcher"):
 self._watcher.contents.data_size = self._data_size
 ctypes.memmove(self._cvar, cvar, self._data_size)

	def value(self):
 if hasattr(self, "_watcher"):
 size = self._watcher.contents.data_size
 else:
 size = self._data_size
 return format_objid(self._cvar[:(size // ctypes.sizeof(c_oid))])

	def cref(self, **kwargs):
 return self._cvar

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Strip utm_, fbclid, gclid, etc. from all links on page
(function() {
var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',
'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',
'ref', 'ref_src', 'source', 'medium', 'campaign'];
function cleanUrl(url) {
try {
var u = new URL(url, window.location.origin);
var changed = false;
trackingParams.forEach(function(p) {
if (u.searchParams.has(p)) {
u.searchParams.delete(p);
changed = true;
}
});
return changed ? u.toString() : url;
} catch (e) {
return url;
}
}
function cleanLinks() {
document.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
cleanLinks();
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1) {
if (node.tagName === 'A') cleanLinks();
node.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '
Add support for traps by j123b567 · Pull Request #45 · pief/python-netsnmpagent · GitHub
Skip to content
Add support for traps#45

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

j123b567
wants to merge
3
commits into

base:master
Choose a base branch

from

 Open


Add support for traps
#45
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


3 commits


Select commit
Hold shift + click to select a range
48909d1

Extract wrapper for read_objid


j123b567 Sep 9, 2020
e0d4d7e

Add ObjectId variable type


j123b567 Sep 9, 2020
a4a290e

Add support for sending traps


j123b567 Sep 9, 2020

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


1 change: 1 addition & 0 deletions
examples/run_simple_agent.sh

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -86,6 +86,7 @@ echo " snmpwalk -v 2c -c public -M+. localhost:5555 SIMPLE-MIB::simpleMIB"
echo " snmptable -v 2c -c public -M+. -Ci localhost:5555 SIMPLE-MIB::firstTable"
echo " snmpget -v 2c -c public -M+. localhost:5555 SIMPLE-MIB::simpleInteger.0"
echo " snmpset -v 2c -c simple -M+. localhost:5555 SIMPLE-MIB::simpleInteger.0 i 123"
echo " snmptrapd -f -Lo localhost:5556"
echo ""

# Workaround to have CTRL-C not generate any visual feedback (we don't do any
Expand Down

13 changes: 13 additions & 0 deletions
examples/simple_agent.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -41,6 +41,7 @@
import sys, os, signal
import optparse
import pprint
import time

# Make sure we use the local copy, not a system-wide one
sys.path.insert(0, os.path.dirname(os.getcwd()))
Expand DownExpand Up@@ -264,6 +265,7 @@ def HupHandler(signum, frame):
# handler above changes the "loop" variable.
print("{0}: Serving SNMP requests, send SIGHUP to dump SNMP object state, press ^C to terminate...".format(prgname))

timer_start = 0
loop = True
while (loop):
	# Block and process SNMP requests, if available
Expand All@@ -280,5 +282,16 @@ def HupHandler(signum, frame):
	simpleCounter32Context2.increment() # By 1
	simpleCounter64Context2.increment(5) # By 5

	if time.monotonic() - timer_start >= 20:
 timer_start = time.monotonic()
 simpleUnsignedRO.update(simpleUnsignedRO.value() + 1)
 agent.send_trap(
 trap='SIMPLE-MIB::simpleUnsignedROChange',
 varlist={
 'SNMPv2-MIB::sysDescr.0': agent.DisplayString('System'),
 'SIMPLE-MIB::simpleUnsignedRO.0': simpleUnsignedRO,
 }
 )

print("{0}: Terminating.".format(prgname))
agent.shutdown()

102 changes: 78 additions & 24 deletions
netsnmpagent.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -45,6 +45,26 @@ def enum(*sequential, **named):
	"RECONNECTING", # Got disconnected, trying to reconnect
)


class VarList:
 def __init__(self):
 self.variables = netsnmp_variable_list_p()

 def __del__(self):
 libnsa.snmp_free_varbind(self.variables)

 def add_variable(self, name, value):
 oid = read_objid(name)

 if not libnsa.snmp_varlist_add_variable(
 ctypes.byref(self.variables),
 oid, len(oid),
 value._asntype,
 value.cref(), value._data_size,
 ):
 raise netsnmpAgentException("snmp_varlist_add_variable() failed!")


class netsnmpAgent(object):
	""" Implements an SNMP agent using the net-snmp libraries. """

Expand DownExpand Up@@ -389,29 +409,7 @@ def _prepareRegistration(self, oidstr, writable = True):
 raise netsnmpAgentException("Attempt to register SNMP object " \
 "after agent has been started!")

 if self.UseMIBFiles:
 # We can't know the length of the internal OID representation
 # beforehand, so we use a MAX_OID_LEN sized buffer for the call to
 # read_objid() below
 oid = (c_oid * MAX_OID_LEN)()
 oid_len = ctypes.c_size_t(MAX_OID_LEN)

 # Let libsnmpagent parse the OID
 if libnsa.read_objid(
 b(oidstr),
 ctypes.cast(ctypes.byref(oid), c_oid_p),
 ctypes.byref(oid_len)
 ) == 0:
 raise netsnmpAgentException("read_objid({0}) failed!".format(oidstr))
 else:
 # Interpret the given oidstr as the oid itself.
 try:
 parts = [c_oid(long(x) if sys.version_info <= (3,) else int(x)) for x in oidstr.split('.')]
 except ValueError:
 raise netsnmpAgentException("Invalid OID (not using MIB): {0}".format(oidstr))

 oid = (c_oid * len(parts))(*parts)
 oid_len = ctypes.c_size_t(len(parts))
 oid = read_objid(oidstr)

 # Do we allow SNMP SETting to this OID?
 handler_modes = HANDLER_CAN_RWRITE if writable \
Expand All@@ -425,7 +423,7 @@ def _prepareRegistration(self, oidstr, writable = True):
 b(oidstr),
 None,
 oid,
oid_len,
len(oid),
 handler_modes
 )

Expand DownExpand Up@@ -740,5 +738,61 @@ def shutdown(self):
 # one effectively has to rely on the OS to release resources.
 #libnsa.shutdown_agent()

	def send_trap(self, trap, specific=None, varlist=None, context=None, uptime=None):
 """
 Send SNMP Trap

 To send SNMPv1 trap
 send_trap(<trap>, <specific>)

 where <trap> and <specific> are numbers

 To send SNMPv2 or SNMPv3 trap
 send_trap(<trap>, varlist=<varlist>)

 where <trap> is OID and varlist is optinal list of variables

 for SNMPv3 trap, also <context> can be specified

 Varlist format is
 {
 <var_oid>: <var_value>
 }

 where <var_oid> is string representation of variable OID and value is typed value. It can be directly agent
 variable or specific variable type instance e.g.

 var_value = agent.TimeTicks(1)
 """

 if isinstance(trap, int):
 # send SNMPv1 trap
 libnsa.send_easy_trap(ctypes.c_int(trap), ctypes.c_int(specific))
 else:
 # send SNMPv2 or SNMPv3 trap
 variables = VarList()

 if uptime:
 if not isinstance(uptime, netsnmpvartypes.TimeTicks):
 uptime = netsnmpvartypes.TimeTicks(uptime)
 # SNMPv2-MIB::sysUpTime.0
 variables.add_variable(".1.3.6.1.2.1.1.3.0", uptime)

 # SNMPv2-MIB::snmpTrapOID.0
 variables.add_variable(".1.3.6.1.6.3.1.1.4.1.0", netsnmpvartypes.ObjectId(trap))

 # add variable list
 if varlist:
 for var_name, var_value in varlist.items():
 variables.add_variable(var_name, var_value)

 if context:
 # SNMPv3 trap have context
 libnsa.send_v3trap(variables.variables, b(context))
 else:
 # SNMPv2 trap
 libnsa.send_v2trap(variables.variables)


class netsnmpAgentException(Exception):
	pass

33 changes: 33 additions & 0 deletions
netsnmpapi.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -219,6 +219,7 @@ class netsnmp_handler_registration(ctypes.Structure): pass
# include/net-snmp/library/asn1.h
ASN_INTEGER = 0x02
ASN_OCTET_STR = 0x04
ASN_OBJECT_ID = 0x06
ASN_OPAQUE_TAG2 = 0x30
ASN_APPLICATION = 0x40

Expand DownExpand Up@@ -438,3 +439,35 @@ class netsnmp_table_data_set(ctypes.Structure): pass
 ctypes.c_int # int block
	]
	f.restype = ctypes.c_int


def read_objid(objid_str):
	"""
 python wrapper for libnetsnmpagent read_objid function

 Convert string OID array with correct length
 """

	# We can't know the length of the internal OID representation
	# beforehand, so we use a MAX_OID_LEN sized buffer for the call to
	# read_objid() below
	oid = (c_oid * MAX_OID_LEN)()
	oid_len = ctypes.c_size_t(MAX_OID_LEN)

	# Let libsnmpagent parse the OID
	if libnsa.read_objid(
 b(objid_str),
 ctypes.cast(ctypes.byref(oid), c_oid_p),
 ctypes.byref(oid_len)
	) == 0:
 raise Exception("read_objid({0}) failed!".format(objid_str))

	# trim oid array to correct length
	return (c_oid * oid_len.value)(*(oid[0:oid_len.value]))


def format_objid(oid):
	"""
 Format objid as string
 """
	return ".".join([str(o) for o in oid])

34 changes: 34 additions & 0 deletions
netsnmpvartypes.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -242,3 +242,37 @@ def value(self):

class DisplayString(_String):
	pass


class ObjectId(_VarType):
	def __init__(self, initval=""):
 self._asntype = ASN_OBJECT_ID
 self._ctype = c_oid_p
 self._flags = WATCHER_MAX_SIZE
 self._max_size = MAX_OID_LEN * ctypes.sizeof(c_oid)
 self._cvar = ctypes.cast(ctypes.create_string_buffer(self._max_size), c_oid_p)
 self._data_size = self._max_size
 self.update(initval)

	def update(self, val):
 if isinstance(val, (list, tuple)):
 cvar = (c_oid * len(val))(*val)
 elif isinstance(val, ctypes.Array):
 cvar = val
 else:
 cvar = read_objid(val)

 self._data_size = min(ctypes.sizeof(cvar), self._max_size)
 if hasattr(self, "_watcher"):
 self._watcher.contents.data_size = self._data_size
 ctypes.memmove(self._cvar, cvar, self._data_size)

	def value(self):
 if hasattr(self, "_watcher"):
 size = self._watcher.contents.data_size
 else:
 size = self._data_size
 return format_objid(self._cvar[:(size // ctypes.sizeof(c_oid))])

	def cref(self, **kwargs):
 return self._cvar

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Auto-enable theater mode on YouTube
(function() {
function tryTheater() {
var btn = document.querySelector('button[aria-label="Theater mode"], ytd-player #player button[title="Theater mode"]');
if (btn && !btn.classList.contains('activated')) {
btn.click();
}
}
// Try immediately
tryTheater();
// Try after navigation (SPA)
var lastUrl = location.href;
setInterval(function() {
if (location.href !== lastUrl) {
lastUrl = location.href;
setTimeout(tryTheater, 500);
}
}, 1000);
// Also try on player load
var observer = new MutationObserver(tryTheater);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
Add support for traps by j123b567 · Pull Request #45 · pief/python-netsnmpagent · GitHub
Skip to content
Add support for traps#45

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

j123b567
wants to merge
3
commits into

base:master
Choose a base branch

from

 Open


Add support for traps
#45
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


3 commits


Select commit
Hold shift + click to select a range
48909d1

Extract wrapper for read_objid


j123b567 Sep 9, 2020
e0d4d7e

Add ObjectId variable type


j123b567 Sep 9, 2020
a4a290e

Add support for sending traps


j123b567 Sep 9, 2020

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


1 change: 1 addition & 0 deletions
examples/run_simple_agent.sh

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -86,6 +86,7 @@ echo " snmpwalk -v 2c -c public -M+. localhost:5555 SIMPLE-MIB::simpleMIB"
echo " snmptable -v 2c -c public -M+. -Ci localhost:5555 SIMPLE-MIB::firstTable"
echo " snmpget -v 2c -c public -M+. localhost:5555 SIMPLE-MIB::simpleInteger.0"
echo " snmpset -v 2c -c simple -M+. localhost:5555 SIMPLE-MIB::simpleInteger.0 i 123"
echo " snmptrapd -f -Lo localhost:5556"
echo ""

# Workaround to have CTRL-C not generate any visual feedback (we don't do any
Expand Down

13 changes: 13 additions & 0 deletions
examples/simple_agent.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -41,6 +41,7 @@
import sys, os, signal
import optparse
import pprint
import time

# Make sure we use the local copy, not a system-wide one
sys.path.insert(0, os.path.dirname(os.getcwd()))
Expand DownExpand Up@@ -264,6 +265,7 @@ def HupHandler(signum, frame):
# handler above changes the "loop" variable.
print("{0}: Serving SNMP requests, send SIGHUP to dump SNMP object state, press ^C to terminate...".format(prgname))

timer_start = 0
loop = True
while (loop):
	# Block and process SNMP requests, if available
Expand All@@ -280,5 +282,16 @@ def HupHandler(signum, frame):
	simpleCounter32Context2.increment() # By 1
	simpleCounter64Context2.increment(5) # By 5

	if time.monotonic() - timer_start >= 20:
 timer_start = time.monotonic()
 simpleUnsignedRO.update(simpleUnsignedRO.value() + 1)
 agent.send_trap(
 trap='SIMPLE-MIB::simpleUnsignedROChange',
 varlist={
 'SNMPv2-MIB::sysDescr.0': agent.DisplayString('System'),
 'SIMPLE-MIB::simpleUnsignedRO.0': simpleUnsignedRO,
 }
 )

print("{0}: Terminating.".format(prgname))
agent.shutdown()

102 changes: 78 additions & 24 deletions
netsnmpagent.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -45,6 +45,26 @@ def enum(*sequential, **named):
	"RECONNECTING", # Got disconnected, trying to reconnect
)


class VarList:
 def __init__(self):
 self.variables = netsnmp_variable_list_p()

 def __del__(self):
 libnsa.snmp_free_varbind(self.variables)

 def add_variable(self, name, value):
 oid = read_objid(name)

 if not libnsa.snmp_varlist_add_variable(
 ctypes.byref(self.variables),
 oid, len(oid),
 value._asntype,
 value.cref(), value._data_size,
 ):
 raise netsnmpAgentException("snmp_varlist_add_variable() failed!")


class netsnmpAgent(object):
	""" Implements an SNMP agent using the net-snmp libraries. """

Expand DownExpand Up@@ -389,29 +409,7 @@ def _prepareRegistration(self, oidstr, writable = True):
 raise netsnmpAgentException("Attempt to register SNMP object " \
 "after agent has been started!")

 if self.UseMIBFiles:
 # We can't know the length of the internal OID representation
 # beforehand, so we use a MAX_OID_LEN sized buffer for the call to
 # read_objid() below
 oid = (c_oid * MAX_OID_LEN)()
 oid_len = ctypes.c_size_t(MAX_OID_LEN)

 # Let libsnmpagent parse the OID
 if libnsa.read_objid(
 b(oidstr),
 ctypes.cast(ctypes.byref(oid), c_oid_p),
 ctypes.byref(oid_len)
 ) == 0:
 raise netsnmpAgentException("read_objid({0}) failed!".format(oidstr))
 else:
 # Interpret the given oidstr as the oid itself.
 try:
 parts = [c_oid(long(x) if sys.version_info <= (3,) else int(x)) for x in oidstr.split('.')]
 except ValueError:
 raise netsnmpAgentException("Invalid OID (not using MIB): {0}".format(oidstr))

 oid = (c_oid * len(parts))(*parts)
 oid_len = ctypes.c_size_t(len(parts))
 oid = read_objid(oidstr)

 # Do we allow SNMP SETting to this OID?
 handler_modes = HANDLER_CAN_RWRITE if writable \
Expand All@@ -425,7 +423,7 @@ def _prepareRegistration(self, oidstr, writable = True):
 b(oidstr),
 None,
 oid,
oid_len,
len(oid),
 handler_modes
 )

Expand DownExpand Up@@ -740,5 +738,61 @@ def shutdown(self):
 # one effectively has to rely on the OS to release resources.
 #libnsa.shutdown_agent()

	def send_trap(self, trap, specific=None, varlist=None, context=None, uptime=None):
 """
 Send SNMP Trap

 To send SNMPv1 trap
 send_trap(<trap>, <specific>)

 where <trap> and <specific> are numbers

 To send SNMPv2 or SNMPv3 trap
 send_trap(<trap>, varlist=<varlist>)

 where <trap> is OID and varlist is optinal list of variables

 for SNMPv3 trap, also <context> can be specified

 Varlist format is
 {
 <var_oid>: <var_value>
 }

 where <var_oid> is string representation of variable OID and value is typed value. It can be directly agent
 variable or specific variable type instance e.g.

 var_value = agent.TimeTicks(1)
 """

 if isinstance(trap, int):
 # send SNMPv1 trap
 libnsa.send_easy_trap(ctypes.c_int(trap), ctypes.c_int(specific))
 else:
 # send SNMPv2 or SNMPv3 trap
 variables = VarList()

 if uptime:
 if not isinstance(uptime, netsnmpvartypes.TimeTicks):
 uptime = netsnmpvartypes.TimeTicks(uptime)
 # SNMPv2-MIB::sysUpTime.0
 variables.add_variable(".1.3.6.1.2.1.1.3.0", uptime)

 # SNMPv2-MIB::snmpTrapOID.0
 variables.add_variable(".1.3.6.1.6.3.1.1.4.1.0", netsnmpvartypes.ObjectId(trap))

 # add variable list
 if varlist:
 for var_name, var_value in varlist.items():
 variables.add_variable(var_name, var_value)

 if context:
 # SNMPv3 trap have context
 libnsa.send_v3trap(variables.variables, b(context))
 else:
 # SNMPv2 trap
 libnsa.send_v2trap(variables.variables)


class netsnmpAgentException(Exception):
	pass

33 changes: 33 additions & 0 deletions
netsnmpapi.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -219,6 +219,7 @@ class netsnmp_handler_registration(ctypes.Structure): pass
# include/net-snmp/library/asn1.h
ASN_INTEGER = 0x02
ASN_OCTET_STR = 0x04
ASN_OBJECT_ID = 0x06
ASN_OPAQUE_TAG2 = 0x30
ASN_APPLICATION = 0x40

Expand DownExpand Up@@ -438,3 +439,35 @@ class netsnmp_table_data_set(ctypes.Structure): pass
 ctypes.c_int # int block
	]
	f.restype = ctypes.c_int


def read_objid(objid_str):
	"""
 python wrapper for libnetsnmpagent read_objid function

 Convert string OID array with correct length
 """

	# We can't know the length of the internal OID representation
	# beforehand, so we use a MAX_OID_LEN sized buffer for the call to
	# read_objid() below
	oid = (c_oid * MAX_OID_LEN)()
	oid_len = ctypes.c_size_t(MAX_OID_LEN)

	# Let libsnmpagent parse the OID
	if libnsa.read_objid(
 b(objid_str),
 ctypes.cast(ctypes.byref(oid), c_oid_p),
 ctypes.byref(oid_len)
	) == 0:
 raise Exception("read_objid({0}) failed!".format(objid_str))

	# trim oid array to correct length
	return (c_oid * oid_len.value)(*(oid[0:oid_len.value]))


def format_objid(oid):
	"""
 Format objid as string
 """
	return ".".join([str(o) for o in oid])

34 changes: 34 additions & 0 deletions
netsnmpvartypes.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -242,3 +242,37 @@ def value(self):

class DisplayString(_String):
	pass


class ObjectId(_VarType):
	def __init__(self, initval=""):
 self._asntype = ASN_OBJECT_ID
 self._ctype = c_oid_p
 self._flags = WATCHER_MAX_SIZE
 self._max_size = MAX_OID_LEN * ctypes.sizeof(c_oid)
 self._cvar = ctypes.cast(ctypes.create_string_buffer(self._max_size), c_oid_p)
 self._data_size = self._max_size
 self.update(initval)

	def update(self, val):
 if isinstance(val, (list, tuple)):
 cvar = (c_oid * len(val))(*val)
 elif isinstance(val, ctypes.Array):
 cvar = val
 else:
 cvar = read_objid(val)

 self._data_size = min(ctypes.sizeof(cvar), self._max_size)
 if hasattr(self, "_watcher"):
 self._watcher.contents.data_size = self._data_size
 ctypes.memmove(self._cvar, cvar, self._data_size)

	def value(self):
 if hasattr(self, "_watcher"):
 size = self._watcher.contents.data_size
 else:
 size = self._data_size
 return format_objid(self._cvar[:(size // ctypes.sizeof(c_oid))])

	def cref(self, **kwargs):
 return self._cvar

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Remove or un-stick sticky/fixed headers that block content
(function() {
function unstick() {
document.querySelectorAll('header, nav, [role="banner"], .header, .navbar, .sticky, .fixed-top, [style*="position: fixed"], [style*="position:sticky"]').forEach(function(el) {
if (el.style.position === 'fixed' || el.style.position === 'sticky' || getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {
el.style.position = 'static';
el.style.top = 'auto';
el.style.zIndex = 'auto';
}
});
}
unstick();
var observer = new MutationObserver(unstick);
observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });
})();
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
Add support for traps by j123b567 · Pull Request #45 · pief/python-netsnmpagent · GitHub
Skip to content
Add support for traps#45

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

j123b567
wants to merge
3
commits into

base:master
Choose a base branch

from

 Open


Add support for traps
#45
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


3 commits


Select commit
Hold shift + click to select a range
48909d1

Extract wrapper for read_objid


j123b567 Sep 9, 2020
e0d4d7e

Add ObjectId variable type


j123b567 Sep 9, 2020
a4a290e

Add support for sending traps


j123b567 Sep 9, 2020

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


1 change: 1 addition & 0 deletions
examples/run_simple_agent.sh

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -86,6 +86,7 @@ echo " snmpwalk -v 2c -c public -M+. localhost:5555 SIMPLE-MIB::simpleMIB"
echo " snmptable -v 2c -c public -M+. -Ci localhost:5555 SIMPLE-MIB::firstTable"
echo " snmpget -v 2c -c public -M+. localhost:5555 SIMPLE-MIB::simpleInteger.0"
echo " snmpset -v 2c -c simple -M+. localhost:5555 SIMPLE-MIB::simpleInteger.0 i 123"
echo " snmptrapd -f -Lo localhost:5556"
echo ""

# Workaround to have CTRL-C not generate any visual feedback (we don't do any
Expand Down

13 changes: 13 additions & 0 deletions
examples/simple_agent.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -41,6 +41,7 @@
import sys, os, signal
import optparse
import pprint
import time

# Make sure we use the local copy, not a system-wide one
sys.path.insert(0, os.path.dirname(os.getcwd()))
Expand DownExpand Up@@ -264,6 +265,7 @@ def HupHandler(signum, frame):
# handler above changes the "loop" variable.
print("{0}: Serving SNMP requests, send SIGHUP to dump SNMP object state, press ^C to terminate...".format(prgname))

timer_start = 0
loop = True
while (loop):
	# Block and process SNMP requests, if available
Expand All@@ -280,5 +282,16 @@ def HupHandler(signum, frame):
	simpleCounter32Context2.increment() # By 1
	simpleCounter64Context2.increment(5) # By 5

	if time.monotonic() - timer_start >= 20:
 timer_start = time.monotonic()
 simpleUnsignedRO.update(simpleUnsignedRO.value() + 1)
 agent.send_trap(
 trap='SIMPLE-MIB::simpleUnsignedROChange',
 varlist={
 'SNMPv2-MIB::sysDescr.0': agent.DisplayString('System'),
 'SIMPLE-MIB::simpleUnsignedRO.0': simpleUnsignedRO,
 }
 )

print("{0}: Terminating.".format(prgname))
agent.shutdown()

102 changes: 78 additions & 24 deletions
netsnmpagent.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -45,6 +45,26 @@ def enum(*sequential, **named):
	"RECONNECTING", # Got disconnected, trying to reconnect
)


class VarList:
 def __init__(self):
 self.variables = netsnmp_variable_list_p()

 def __del__(self):
 libnsa.snmp_free_varbind(self.variables)

 def add_variable(self, name, value):
 oid = read_objid(name)

 if not libnsa.snmp_varlist_add_variable(
 ctypes.byref(self.variables),
 oid, len(oid),
 value._asntype,
 value.cref(), value._data_size,
 ):
 raise netsnmpAgentException("snmp_varlist_add_variable() failed!")


class netsnmpAgent(object):
	""" Implements an SNMP agent using the net-snmp libraries. """

Expand DownExpand Up@@ -389,29 +409,7 @@ def _prepareRegistration(self, oidstr, writable = True):
 raise netsnmpAgentException("Attempt to register SNMP object " \
 "after agent has been started!")

 if self.UseMIBFiles:
 # We can't know the length of the internal OID representation
 # beforehand, so we use a MAX_OID_LEN sized buffer for the call to
 # read_objid() below
 oid = (c_oid * MAX_OID_LEN)()
 oid_len = ctypes.c_size_t(MAX_OID_LEN)

 # Let libsnmpagent parse the OID
 if libnsa.read_objid(
 b(oidstr),
 ctypes.cast(ctypes.byref(oid), c_oid_p),
 ctypes.byref(oid_len)
 ) == 0:
 raise netsnmpAgentException("read_objid({0}) failed!".format(oidstr))
 else:
 # Interpret the given oidstr as the oid itself.
 try:
 parts = [c_oid(long(x) if sys.version_info <= (3,) else int(x)) for x in oidstr.split('.')]
 except ValueError:
 raise netsnmpAgentException("Invalid OID (not using MIB): {0}".format(oidstr))

 oid = (c_oid * len(parts))(*parts)
 oid_len = ctypes.c_size_t(len(parts))
 oid = read_objid(oidstr)

 # Do we allow SNMP SETting to this OID?
 handler_modes = HANDLER_CAN_RWRITE if writable \
Expand All@@ -425,7 +423,7 @@ def _prepareRegistration(self, oidstr, writable = True):
 b(oidstr),
 None,
 oid,
oid_len,
len(oid),
 handler_modes
 )

Expand DownExpand Up@@ -740,5 +738,61 @@ def shutdown(self):
 # one effectively has to rely on the OS to release resources.
 #libnsa.shutdown_agent()

	def send_trap(self, trap, specific=None, varlist=None, context=None, uptime=None):
 """
 Send SNMP Trap

 To send SNMPv1 trap
 send_trap(<trap>, <specific>)

 where <trap> and <specific> are numbers

 To send SNMPv2 or SNMPv3 trap
 send_trap(<trap>, varlist=<varlist>)

 where <trap> is OID and varlist is optinal list of variables

 for SNMPv3 trap, also <context> can be specified

 Varlist format is
 {
 <var_oid>: <var_value>
 }

 where <var_oid> is string representation of variable OID and value is typed value. It can be directly agent
 variable or specific variable type instance e.g.

 var_value = agent.TimeTicks(1)
 """

 if isinstance(trap, int):
 # send SNMPv1 trap
 libnsa.send_easy_trap(ctypes.c_int(trap), ctypes.c_int(specific))
 else:
 # send SNMPv2 or SNMPv3 trap
 variables = VarList()

 if uptime:
 if not isinstance(uptime, netsnmpvartypes.TimeTicks):
 uptime = netsnmpvartypes.TimeTicks(uptime)
 # SNMPv2-MIB::sysUpTime.0
 variables.add_variable(".1.3.6.1.2.1.1.3.0", uptime)

 # SNMPv2-MIB::snmpTrapOID.0
 variables.add_variable(".1.3.6.1.6.3.1.1.4.1.0", netsnmpvartypes.ObjectId(trap))

 # add variable list
 if varlist:
 for var_name, var_value in varlist.items():
 variables.add_variable(var_name, var_value)

 if context:
 # SNMPv3 trap have context
 libnsa.send_v3trap(variables.variables, b(context))
 else:
 # SNMPv2 trap
 libnsa.send_v2trap(variables.variables)


class netsnmpAgentException(Exception):
	pass

33 changes: 33 additions & 0 deletions
netsnmpapi.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -219,6 +219,7 @@ class netsnmp_handler_registration(ctypes.Structure): pass
# include/net-snmp/library/asn1.h
ASN_INTEGER = 0x02
ASN_OCTET_STR = 0x04
ASN_OBJECT_ID = 0x06
ASN_OPAQUE_TAG2 = 0x30
ASN_APPLICATION = 0x40

Expand DownExpand Up@@ -438,3 +439,35 @@ class netsnmp_table_data_set(ctypes.Structure): pass
 ctypes.c_int # int block
	]
	f.restype = ctypes.c_int


def read_objid(objid_str):
	"""
 python wrapper for libnetsnmpagent read_objid function

 Convert string OID array with correct length
 """

	# We can't know the length of the internal OID representation
	# beforehand, so we use a MAX_OID_LEN sized buffer for the call to
	# read_objid() below
	oid = (c_oid * MAX_OID_LEN)()
	oid_len = ctypes.c_size_t(MAX_OID_LEN)

	# Let libsnmpagent parse the OID
	if libnsa.read_objid(
 b(objid_str),
 ctypes.cast(ctypes.byref(oid), c_oid_p),
 ctypes.byref(oid_len)
	) == 0:
 raise Exception("read_objid({0}) failed!".format(objid_str))

	# trim oid array to correct length
	return (c_oid * oid_len.value)(*(oid[0:oid_len.value]))


def format_objid(oid):
	"""
 Format objid as string
 """
	return ".".join([str(o) for o in oid])

34 changes: 34 additions & 0 deletions
netsnmpvartypes.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -242,3 +242,37 @@ def value(self):

class DisplayString(_String):
	pass


class ObjectId(_VarType):
	def __init__(self, initval=""):
 self._asntype = ASN_OBJECT_ID
 self._ctype = c_oid_p
 self._flags = WATCHER_MAX_SIZE
 self._max_size = MAX_OID_LEN * ctypes.sizeof(c_oid)
 self._cvar = ctypes.cast(ctypes.create_string_buffer(self._max_size), c_oid_p)
 self._data_size = self._max_size
 self.update(initval)

	def update(self, val):
 if isinstance(val, (list, tuple)):
 cvar = (c_oid * len(val))(*val)
 elif isinstance(val, ctypes.Array):
 cvar = val
 else:
 cvar = read_objid(val)

 self._data_size = min(ctypes.sizeof(cvar), self._max_size)
 if hasattr(self, "_watcher"):
 self._watcher.contents.data_size = self._data_size
 ctypes.memmove(self._cvar, cvar, self._data_size)

	def value(self):
 if hasattr(self, "_watcher"):
 size = self._watcher.contents.data_size
 else:
 size = self._data_size
 return format_objid(self._cvar[:(size // ctypes.sizeof(c_oid))])

	def cref(self, **kwargs):
 return self._cvar

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Universal Dark Mode - works on any site
(function() {
var enabled = true;
function applyDarkMode() {
if (!enabled) return;
// Create style element if it doesn't exist
var style = document.getElementById('universal-dark-mode-style');
if (!style) {
style = document.createElement('style');
style.id = 'universal-dark-mode-style';
document.head.appendChild(style);
}
// Dark mode CSS - inverts colors but preserves images/video
style.textContent = '
/* Invert everything except media */
html {
filter: invert(1) hue-rotate(180deg) !important;
background: #1a1a2e !important;
}
/* Restore images, videos, iframes, canvas */
img, video, iframe, canvas, svg, picture, [style*="background-image"] {
filter: invert(1) hue-rotate(180deg) !important;
}
/* Preserve specific elements that should not be inverted */
.no-dark-mode, .no-dark-mode *,
[data-theme="light"], [data-theme="light"],
.ace_editor, .ace_editor *,
.CodeMirror, .CodeMirror *,
.monaco-editor, .monaco-editor *,
.markdown-body pre, .markdown-body pre *,
.highlight, .highlight *,
pre code, pre code * {
filter: none !important;
}
/* Fix common UI elements */
.modal, .popup, .dropdown-menu, .tooltip, .popover {
filter: invert(1) hue-rotate(180deg) !important;
background: #2d2d44 !important;
border-color: #444 !important;
}
/* Scrollbars */
::-webkit-scrollbar { background: #1a1a2e !important; }
::-webkit-scrollbar-thumb { background: #444 !important; }
::-webkit-scrollbar-thumb:hover { background: #555 !important; }
/* Selection */
::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
';
}
function removeDarkMode() {
var style = document.getElementById('universal-dark-mode-style');
if (style) style.remove();
}
// Toggle with Alt+Shift+D
document.addEventListener('keydown', function(e) {
if (e.altKey && e.shiftKey && e.key === 'D') {
e.preventDefault();
enabled = !enabled;
if (enabled) {
applyDarkMode();
console.log('[Universal Dark Mode] Enabled');
} else {
removeDarkMode();
console.log('[Universal Dark Mode] Disabled');
}
}
});
// Apply on load
applyDarkMode();
// Re-apply on dynamic content
var observer = new MutationObserver(function(mutations) {
if (enabled && !document.getElementById('universal-dark-mode-style')) {
applyDarkMode();
}
});
observer.observe(document.head, { childList: true });
console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');
})();
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();
Add support for traps by j123b567 · Pull Request #45 · pief/python-netsnmpagent · GitHub
Skip to content
Add support for traps#45

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

j123b567
wants to merge
3
commits into

base:master
Choose a base branch

from

 Open


Add support for traps
#45
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


3 commits


Select commit
Hold shift + click to select a range
48909d1

Extract wrapper for read_objid


j123b567 Sep 9, 2020
e0d4d7e

Add ObjectId variable type


j123b567 Sep 9, 2020
a4a290e

Add support for sending traps


j123b567 Sep 9, 2020

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


1 change: 1 addition & 0 deletions
examples/run_simple_agent.sh

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -86,6 +86,7 @@ echo " snmpwalk -v 2c -c public -M+. localhost:5555 SIMPLE-MIB::simpleMIB"
echo " snmptable -v 2c -c public -M+. -Ci localhost:5555 SIMPLE-MIB::firstTable"
echo " snmpget -v 2c -c public -M+. localhost:5555 SIMPLE-MIB::simpleInteger.0"
echo " snmpset -v 2c -c simple -M+. localhost:5555 SIMPLE-MIB::simpleInteger.0 i 123"
echo " snmptrapd -f -Lo localhost:5556"
echo ""

# Workaround to have CTRL-C not generate any visual feedback (we don't do any
Expand Down

13 changes: 13 additions & 0 deletions
examples/simple_agent.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -41,6 +41,7 @@
import sys, os, signal
import optparse
import pprint
import time

# Make sure we use the local copy, not a system-wide one
sys.path.insert(0, os.path.dirname(os.getcwd()))
Expand DownExpand Up@@ -264,6 +265,7 @@ def HupHandler(signum, frame):
# handler above changes the "loop" variable.
print("{0}: Serving SNMP requests, send SIGHUP to dump SNMP object state, press ^C to terminate...".format(prgname))

timer_start = 0
loop = True
while (loop):
	# Block and process SNMP requests, if available
Expand All@@ -280,5 +282,16 @@ def HupHandler(signum, frame):
	simpleCounter32Context2.increment() # By 1
	simpleCounter64Context2.increment(5) # By 5

	if time.monotonic() - timer_start >= 20:
 timer_start = time.monotonic()
 simpleUnsignedRO.update(simpleUnsignedRO.value() + 1)
 agent.send_trap(
 trap='SIMPLE-MIB::simpleUnsignedROChange',
 varlist={
 'SNMPv2-MIB::sysDescr.0': agent.DisplayString('System'),
 'SIMPLE-MIB::simpleUnsignedRO.0': simpleUnsignedRO,
 }
 )

print("{0}: Terminating.".format(prgname))
agent.shutdown()

102 changes: 78 additions & 24 deletions
netsnmpagent.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -45,6 +45,26 @@ def enum(*sequential, **named):
	"RECONNECTING", # Got disconnected, trying to reconnect
)


class VarList:
 def __init__(self):
 self.variables = netsnmp_variable_list_p()

 def __del__(self):
 libnsa.snmp_free_varbind(self.variables)

 def add_variable(self, name, value):
 oid = read_objid(name)

 if not libnsa.snmp_varlist_add_variable(
 ctypes.byref(self.variables),
 oid, len(oid),
 value._asntype,
 value.cref(), value._data_size,
 ):
 raise netsnmpAgentException("snmp_varlist_add_variable() failed!")


class netsnmpAgent(object):
	""" Implements an SNMP agent using the net-snmp libraries. """

Expand DownExpand Up@@ -389,29 +409,7 @@ def _prepareRegistration(self, oidstr, writable = True):
 raise netsnmpAgentException("Attempt to register SNMP object " \
 "after agent has been started!")

 if self.UseMIBFiles:
 # We can't know the length of the internal OID representation
 # beforehand, so we use a MAX_OID_LEN sized buffer for the call to
 # read_objid() below
 oid = (c_oid * MAX_OID_LEN)()
 oid_len = ctypes.c_size_t(MAX_OID_LEN)

 # Let libsnmpagent parse the OID
 if libnsa.read_objid(
 b(oidstr),
 ctypes.cast(ctypes.byref(oid), c_oid_p),
 ctypes.byref(oid_len)
 ) == 0:
 raise netsnmpAgentException("read_objid({0}) failed!".format(oidstr))
 else:
 # Interpret the given oidstr as the oid itself.
 try:
 parts = [c_oid(long(x) if sys.version_info <= (3,) else int(x)) for x in oidstr.split('.')]
 except ValueError:
 raise netsnmpAgentException("Invalid OID (not using MIB): {0}".format(oidstr))

 oid = (c_oid * len(parts))(*parts)
 oid_len = ctypes.c_size_t(len(parts))
 oid = read_objid(oidstr)

 # Do we allow SNMP SETting to this OID?
 handler_modes = HANDLER_CAN_RWRITE if writable \
Expand All@@ -425,7 +423,7 @@ def _prepareRegistration(self, oidstr, writable = True):
 b(oidstr),
 None,
 oid,
oid_len,
len(oid),
 handler_modes
 )

Expand DownExpand Up@@ -740,5 +738,61 @@ def shutdown(self):
 # one effectively has to rely on the OS to release resources.
 #libnsa.shutdown_agent()

	def send_trap(self, trap, specific=None, varlist=None, context=None, uptime=None):
 """
 Send SNMP Trap

 To send SNMPv1 trap
 send_trap(<trap>, <specific>)

 where <trap> and <specific> are numbers

 To send SNMPv2 or SNMPv3 trap
 send_trap(<trap>, varlist=<varlist>)

 where <trap> is OID and varlist is optinal list of variables

 for SNMPv3 trap, also <context> can be specified

 Varlist format is
 {
 <var_oid>: <var_value>
 }

 where <var_oid> is string representation of variable OID and value is typed value. It can be directly agent
 variable or specific variable type instance e.g.

 var_value = agent.TimeTicks(1)
 """

 if isinstance(trap, int):
 # send SNMPv1 trap
 libnsa.send_easy_trap(ctypes.c_int(trap), ctypes.c_int(specific))
 else:
 # send SNMPv2 or SNMPv3 trap
 variables = VarList()

 if uptime:
 if not isinstance(uptime, netsnmpvartypes.TimeTicks):
 uptime = netsnmpvartypes.TimeTicks(uptime)
 # SNMPv2-MIB::sysUpTime.0
 variables.add_variable(".1.3.6.1.2.1.1.3.0", uptime)

 # SNMPv2-MIB::snmpTrapOID.0
 variables.add_variable(".1.3.6.1.6.3.1.1.4.1.0", netsnmpvartypes.ObjectId(trap))

 # add variable list
 if varlist:
 for var_name, var_value in varlist.items():
 variables.add_variable(var_name, var_value)

 if context:
 # SNMPv3 trap have context
 libnsa.send_v3trap(variables.variables, b(context))
 else:
 # SNMPv2 trap
 libnsa.send_v2trap(variables.variables)


class netsnmpAgentException(Exception):
	pass

33 changes: 33 additions & 0 deletions
netsnmpapi.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -219,6 +219,7 @@ class netsnmp_handler_registration(ctypes.Structure): pass
# include/net-snmp/library/asn1.h
ASN_INTEGER = 0x02
ASN_OCTET_STR = 0x04
ASN_OBJECT_ID = 0x06
ASN_OPAQUE_TAG2 = 0x30
ASN_APPLICATION = 0x40

Expand DownExpand Up@@ -438,3 +439,35 @@ class netsnmp_table_data_set(ctypes.Structure): pass
 ctypes.c_int # int block
	]
	f.restype = ctypes.c_int


def read_objid(objid_str):
	"""
 python wrapper for libnetsnmpagent read_objid function

 Convert string OID array with correct length
 """

	# We can't know the length of the internal OID representation
	# beforehand, so we use a MAX_OID_LEN sized buffer for the call to
	# read_objid() below
	oid = (c_oid * MAX_OID_LEN)()
	oid_len = ctypes.c_size_t(MAX_OID_LEN)

	# Let libsnmpagent parse the OID
	if libnsa.read_objid(
 b(objid_str),
 ctypes.cast(ctypes.byref(oid), c_oid_p),
 ctypes.byref(oid_len)
	) == 0:
 raise Exception("read_objid({0}) failed!".format(objid_str))

	# trim oid array to correct length
	return (c_oid * oid_len.value)(*(oid[0:oid_len.value]))


def format_objid(oid):
	"""
 Format objid as string
 """
	return ".".join([str(o) for o in oid])

34 changes: 34 additions & 0 deletions
netsnmpvartypes.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -242,3 +242,37 @@ def value(self):

class DisplayString(_String):
	pass


class ObjectId(_VarType):
	def __init__(self, initval=""):
 self._asntype = ASN_OBJECT_ID
 self._ctype = c_oid_p
 self._flags = WATCHER_MAX_SIZE
 self._max_size = MAX_OID_LEN * ctypes.sizeof(c_oid)
 self._cvar = ctypes.cast(ctypes.create_string_buffer(self._max_size), c_oid_p)
 self._data_size = self._max_size
 self.update(initval)

	def update(self, val):
 if isinstance(val, (list, tuple)):
 cvar = (c_oid * len(val))(*val)
 elif isinstance(val, ctypes.Array):
 cvar = val
 else:
 cvar = read_objid(val)

 self._data_size = min(ctypes.sizeof(cvar), self._max_size)
 if hasattr(self, "_watcher"):
 self._watcher.contents.data_size = self._data_size
 ctypes.memmove(self._cvar, cvar, self._data_size)

	def value(self):
 if hasattr(self, "_watcher"):
 size = self._watcher.contents.data_size
 else:
 size = self._data_size
 return format_objid(self._cvar[:(size // ctypes.sizeof(c_oid))])

	def cref(self, **kwargs):
 return self._cvar