forked from etingof/pysnmp
-
-
Notifications
You must be signed in to change notification settings - Fork 26
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
11 changed files
with
848 additions
and
454 deletions.
There are no files selected for viewing
Large diffs are not rendered by default.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,126 @@ | ||
from pysnmp.carrier.asyncio.dgram import udp | ||
from pysnmp.entity import config, engine | ||
from pysnmp.entity.rfc3413 import cmdrsp, context | ||
from pysnmp.hlapi.asyncio import SnmpEngine | ||
from pysnmp.proto.api import v2c | ||
|
||
|
||
import asyncio | ||
import time | ||
|
||
# Set the port to 1611 instead of 161, because 161 is a | ||
# privileged port and requires root access | ||
AGENT_PORT = 1611 | ||
|
||
|
||
async def start_agent() -> SnmpEngine: | ||
# Create SNMP engine | ||
snmpEngine = engine.SnmpEngine() | ||
|
||
# Set up transport endpoint | ||
config.addTransport( | ||
snmpEngine, | ||
udp.domainName, | ||
udp.UdpTransport().openServerMode(("localhost", AGENT_PORT)), | ||
) | ||
|
||
# Set up community data | ||
config.addV1System(snmpEngine, "public", "public") | ||
# Add SNMP v3 user | ||
config.addV3User( | ||
snmpEngine, "usr-none-none", config.usmNoAuthProtocol, config.usmNoPrivProtocol | ||
) | ||
|
||
config.addV3User( | ||
snmpEngine, | ||
"usr-sha-aes", | ||
config.usmHMACSHAAuthProtocol, | ||
"authkey1", | ||
config.usmAesCfb128Protocol, | ||
"privkey1", | ||
) | ||
|
||
config.addV3User( | ||
snmpEngine, "usr-sha-none", config.usmHMACSHAAuthProtocol, "authkey1" | ||
) | ||
|
||
# Allow read MIB access for this user / securityModels at VACM | ||
config.addVacmUser(snmpEngine, 1, "public", "noAuthNoPriv", (1, 3, 6), (1, 3, 6)) | ||
config.addVacmUser(snmpEngine, 2, "public", "noAuthNoPriv", (1, 3, 6), (1, 3, 6)) | ||
config.addVacmUser(snmpEngine, 3, "usr-none-none", "noAuthNoPriv", (1, 3, 6)) | ||
config.addVacmUser(snmpEngine, 3, "usr-sha-none", "authNoPriv", (1, 3, 6)) | ||
config.addVacmUser(snmpEngine, 3, "usr-sha-aes", "authPriv", (1, 3, 6)) | ||
|
||
# Configure SNMP context | ||
snmpContext = context.SnmpContext(snmpEngine) | ||
|
||
# --- create custom Managed Object Instance --- | ||
mibBuilder = snmpContext.getMibInstrum().getMibBuilder() | ||
|
||
MibScalar, MibScalarInstance = mibBuilder.importSymbols( | ||
"SNMPv2-SMI", "MibScalar", "MibScalarInstance" | ||
) | ||
|
||
class MyStaticMibScalarInstance(MibScalarInstance): | ||
# noinspection PyUnusedLocal,PyUnusedLocal | ||
def getValue(self, name, idx): | ||
time.sleep(3) # Add a 2-second sleep | ||
return self.getSyntax().clone(f"Test agent") | ||
|
||
def setValue(self, name, idx, value): | ||
# Here you can handle the SET operation. | ||
# `value` is the new value for the scalar instance. | ||
# You should store this value somewhere, and return it in the `getValue` method. | ||
# For this example, let's just print it. | ||
print(f"SET operation received. New value: {value}") | ||
return self.getSyntax().clone(value) | ||
|
||
mibBuilder.exportSymbols( | ||
"__MY_MIB", | ||
MibScalar((1, 3, 6, 1, 4, 1, 60069, 9, 1), v2c.OctetString()), | ||
MyStaticMibScalarInstance( | ||
(1, 3, 6, 1, 4, 1, 60069, 9, 1), (0,), v2c.OctetString() | ||
), | ||
) | ||
|
||
# --- end of Managed Object Instance initialization ---- | ||
|
||
# Register SNMP Applications at the SNMP engine for particular SNMP context | ||
cmdrsp.GetCommandResponder(snmpEngine, snmpContext) | ||
cmdrsp.NextCommandResponder(snmpEngine, snmpContext) | ||
cmdrsp.BulkCommandResponder(snmpEngine, snmpContext) | ||
cmdrsp.SetCommandResponder(snmpEngine, snmpContext) | ||
|
||
# Start the event loop | ||
snmpEngine.transportDispatcher.jobStarted(1) | ||
|
||
snmpEngine.transportDispatcher.runDispatcher() | ||
|
||
# Wait for the agent to start | ||
await asyncio.sleep(1) | ||
|
||
# return the engine | ||
return snmpEngine | ||
|
||
|
||
class AgentContextManager: | ||
""" | ||
A context manager for managing the lifecycle of an SNMP test agent. | ||
Usage: | ||
async with AgentContextManager() as agent: | ||
# Perform operations with the agent | ||
When the context is entered, the agent is started using the `start_agent()` function. | ||
When the context is exited, the agent's transport dispatcher is stopped and closed. | ||
Note: The `start_agent()` function and the `transportDispatcher` attribute are not defined in this code snippet. | ||
""" | ||
|
||
async def __aenter__(self): | ||
self.agent = await start_agent() | ||
return self.agent | ||
|
||
async def __aexit__(self, exc_type, exc_val, exc_tb): | ||
self.agent.transportDispatcher.jobFinished(1) | ||
self.agent.transportDispatcher.closeDispatcher() |
34 changes: 20 additions & 14 deletions
34
tests/hlapi/asyncio/manager/cmdgen/test_custom_asn1_mib_search_path.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,21 +1,27 @@ | ||
import pytest | ||
from pysnmp.hlapi.asyncio import * | ||
from tests.agent_context import AGENT_PORT, AgentContextManager | ||
|
||
|
||
@pytest.mark.asyncio | ||
async def test_custom_asn1_mib_search_path(): | ||
snmpEngine = SnmpEngine() | ||
errorIndication, errorStatus, errorIndex, varBinds = await getCmd( | ||
snmpEngine, | ||
CommunityData('public'), | ||
UdpTransportTarget(('demo.pysnmp.com', 161)), | ||
ContextData(), | ||
ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets', 1).addAsn1MibSource('file:///usr/share/snmp', | ||
'https://mibs.pysnmp.com/asn1/@mib@')) | ||
) | ||
async with AgentContextManager(): | ||
snmpEngine = SnmpEngine() | ||
errorIndication, errorStatus, errorIndex, varBinds = await getCmd( | ||
snmpEngine, | ||
CommunityData("public"), | ||
UdpTransportTarget(("localhost", AGENT_PORT)), | ||
ContextData(), | ||
ObjectType( | ||
ObjectIdentity("IF-MIB", "ifInOctets", 1).addAsn1MibSource( | ||
"file:///usr/share/snmp", "https://mibs.pysnmp.com/asn1/@mib@" | ||
) | ||
), | ||
) | ||
|
||
assert errorIndication is None | ||
assert errorStatus == 0 | ||
assert len(varBinds) == 1 | ||
assert varBinds[0][0].prettyPrint() == 'IF-MIB::ifInOctets.1' | ||
assert errorIndication is None | ||
assert errorStatus == 0 | ||
assert len(varBinds) == 1 | ||
assert varBinds[0][0].prettyPrint() == "IF-MIB::ifInOctets.1" | ||
|
||
snmpEngine.transportDispatcher.closeDispatcher() | ||
snmpEngine.transportDispatcher.closeDispatcher() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,41 +1,43 @@ | ||
import pytest | ||
from pysnmp.hlapi.asyncio import * | ||
from pysnmp.proto.errind import UnknownUserName | ||
from tests.agent_context import AGENT_PORT, AgentContextManager | ||
|
||
|
||
@pytest.mark.asyncio | ||
async def test_usm_no_auth_no_priv(): | ||
snmpEngine = SnmpEngine() | ||
authData = UsmUserData( | ||
"usr-none-none" | ||
) | ||
errorIndication, errorStatus, errorIndex, varBinds = await getCmd( | ||
snmpEngine, | ||
authData, | ||
UdpTransportTarget(("demo.pysnmp.com", 161), retries=0), | ||
ContextData(), | ||
ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), | ||
) | ||
async with AgentContextManager(): | ||
snmpEngine = SnmpEngine() | ||
authData = UsmUserData("usr-none-none") | ||
errorIndication, errorStatus, errorIndex, varBinds = await getCmd( | ||
snmpEngine, | ||
authData, | ||
UdpTransportTarget(("localhost", AGENT_PORT), retries=0), | ||
ContextData(), | ||
ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), | ||
) | ||
|
||
assert errorIndication is None | ||
assert errorStatus == 0 | ||
assert len(varBinds) == 1 | ||
assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysDescr.0" | ||
assert varBinds[0][1].prettyPrint().startswith("PySNMP engine version") | ||
isinstance(varBinds[0][1], OctetString) | ||
|
||
assert errorIndication is None | ||
assert errorStatus == 0 | ||
assert len(varBinds) == 1 | ||
assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysDescr.0" | ||
isinstance(varBinds[0][1], OctetString) | ||
|
||
@pytest.mark.asyncio | ||
async def test_usm_no_auth_no_priv_wrong_user(): | ||
snmpEngine = SnmpEngine() | ||
authData = UsmUserData( | ||
"usr-none-none-not-exist" | ||
) | ||
errorIndication, errorStatus, errorIndex, varBinds = await getCmd( | ||
snmpEngine, | ||
authData, | ||
UdpTransportTarget(("demo.pysnmp.com", 161), retries=0), | ||
ContextData(), | ||
ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), | ||
) | ||
async with AgentContextManager(): | ||
snmpEngine = SnmpEngine() | ||
authData = UsmUserData("usr-none-none-not-exist") | ||
errorIndication, errorStatus, errorIndex, varBinds = await getCmd( | ||
snmpEngine, | ||
authData, | ||
UdpTransportTarget(("localhost", AGENT_PORT), retries=0), | ||
ContextData(), | ||
ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), | ||
) | ||
|
||
assert isinstance(errorIndication, UnknownUserName) | ||
assert str(errorIndication) == 'Unknown USM user' | ||
snmpEngine.transportDispatcher.closeDispatcher() | ||
assert isinstance(errorIndication, UnknownUserName) | ||
assert str(errorIndication) == "Unknown USM user" | ||
snmpEngine.transportDispatcher.closeDispatcher() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,21 +1,26 @@ | ||
import pytest | ||
from pysnmp.hlapi.asyncio import * | ||
|
||
from tests.agent_context import AgentContextManager | ||
|
||
|
||
@pytest.mark.asyncio | ||
async def test_v1_get(): | ||
snmpEngine = SnmpEngine() | ||
errorIndication, errorStatus, errorIndex, varBinds = await getCmd( | ||
snmpEngine, | ||
CommunityData('public', mpModel=0), | ||
UdpTransportTarget(('demo.pysnmp.com', 161)), | ||
ContextData(), | ||
ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)) | ||
) | ||
async with AgentContextManager(): | ||
snmpEngine = SnmpEngine() | ||
errorIndication, errorStatus, errorIndex, varBinds = await getCmd( | ||
snmpEngine, | ||
CommunityData("public", mpModel=0), | ||
UdpTransportTarget(("localhost", 1611)), | ||
ContextData(), | ||
ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), | ||
) | ||
|
||
assert errorIndication is None | ||
assert errorStatus == 0 | ||
assert len(varBinds) == 1 | ||
assert varBinds[0][0].prettyPrint() == 'SNMPv2-MIB::sysDescr.0' | ||
assert isinstance(varBinds[0][1], OctetString) | ||
assert errorIndication is None | ||
assert errorStatus == 0 | ||
assert len(varBinds) == 1 | ||
assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysDescr.0" | ||
assert varBinds[0][1].prettyPrint().startswith("PySNMP engine version") | ||
assert isinstance(varBinds[0][1], OctetString) | ||
|
||
snmpEngine.transportDispatcher.closeDispatcher() | ||
snmpEngine.transportDispatcher.closeDispatcher() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,21 +1,24 @@ | ||
import pytest | ||
from pysnmp.hlapi.asyncio.slim import Slim | ||
from pysnmp.smi.rfc1902 import ObjectIdentity, ObjectType | ||
from tests.agent_context import AGENT_PORT, AgentContextManager | ||
|
||
|
||
@pytest.mark.asyncio | ||
async def test_v1_set(): | ||
slim = Slim(1) | ||
errorIndication, errorStatus, errorIndex, varBinds = await slim.set( | ||
'public', | ||
'demo.pysnmp.com', | ||
161, | ||
ObjectType(ObjectIdentity("SNMPv2-MIB", "sysLocation", 0), "Shanghai") | ||
) | ||
async with AgentContextManager(): | ||
slim = Slim(1) | ||
errorIndication, errorStatus, errorIndex, varBinds = await slim.set( | ||
"public", | ||
"localhost", | ||
AGENT_PORT, | ||
ObjectType(ObjectIdentity("SNMPv2-MIB", "sysLocation", 0), "Shanghai"), | ||
) | ||
|
||
assert errorIndication is None | ||
assert errorStatus == 0 | ||
assert len(varBinds) == 1 | ||
assert varBinds[0][0].prettyPrint() == 'SNMPv2-MIB::sysLocation.0' | ||
assert varBinds[0][1].prettyPrint() == 'Shanghai' | ||
assert errorIndication is None | ||
assert errorStatus == 0 | ||
assert len(varBinds) == 1 | ||
assert varBinds[0][0].prettyPrint() == "SNMPv2-MIB::sysLocation.0" | ||
assert varBinds[0][1].prettyPrint() == "Shanghai" | ||
|
||
slim.close() | ||
slim.close() |
Oops, something went wrong.