Transport tweaks¶

SNMPv2c, custom timeout¶

Send a SNMP GET request:

with SNMPv2c, community ‘public’
over IPv4/UDP
to an Agent at 104.236.166.95:161
wait 3 seconds for response, retry 5 times (plus one initial attempt)
for an OID in tuple form

This script performs similar to the following Net-SNMP command:

$ snmpget -v2c -c public -ObentU -r 5 -t 1 104.236.166.95 1.3.6.1.2.1.1.1.0

from pysnmp.entity import engine, config
from pysnmp.carrier.asyncore.dgram import udp
from pysnmp.entity.rfc3413 import cmdgen

# Create SNMP engine instance
snmpEngine = engine.SnmpEngine()

#
# SNMPv2c setup
#

# SecurityName <-> CommunityName mapping
config.addV1System(snmpEngine, 'my-area', 'public')

# Specify security settings per SecurityName (SNMPv1 - 0, SNMPv2c - 1)
config.addTargetParams(snmpEngine, 'my-creds', 'my-area', 'noAuthNoPriv', 1)

#
# Setup transport endpoint and bind it with security settings yielding
# a target name 
#

# UDP/IPv4
config.addTransport(
    snmpEngine,
    udp.DOMAIN_NAME,
    udp.UdpSocketTransport().openClientMode()
)

config.addTargetAddr(
    snmpEngine, 'my-router',
    udp.DOMAIN_NAME, ('104.236.166.95', 161),
    'my-creds',
    timeout=300,  # in 1/100 sec
    retryCount=5
)


# Error/response receiver
# noinspection PyUnusedLocal,PyUnusedLocal,PyUnusedLocal
def cbFun(snmpEngine, sendRequestHandle, errorIndication,
          errorStatus, errorIndex, varBinds, cbCtx):
    if errorIndication:
        print(errorIndication)

    elif errorStatus:
        print('%s at %s' % (errorStatus.prettyPrint(),
                            errorIndex and varBinds[int(errorIndex) - 1][0] or '?'))

    else:
        for oid, val in varBinds:
            print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))


# Prepare and send a request message
cmdgen.GetCommandGenerator().sendVarBinds(
    snmpEngine,
    'my-router',
    None, '',  # contextEngineId, contextName
    [((1, 3, 6, 1, 2, 1, 1, 1, 0), None)],
    cbFun
)

# Run I/O dispatcher which would send pending queries and process responses
snmpEngine.transportDispatcher.runDispatcher()

Download script.

Send packets from specific local interface¶

Send a series of SNMP GETNEXT requests with the following options:

with SNMPv2c, community ‘public’
over IPv4/UDP
to an Agent at 104.236.166.95:161
sending packets from primary local interface 0.0.0.0, local port 61024
for two OIDs in tuple form
stop on end-of-mib condition for both OIDs

This script performs similar to the following Net-SNMP command:

$ snmpwalk -v2c -c public -ObentU 104.236.166.95 1.3.6.1.2.1.1 1.3.6.1.4.1.1

from pysnmp.entity import engine, config
from pysnmp.carrier.asyncore.dgram import udp
from pysnmp.entity.rfc3413 import cmdgen

# Create SNMP engine instance
snmpEngine = engine.SnmpEngine()

#
# SNMPv1/2c setup (if you use SNMPv1 or v2c)
#

# SecurityName <-> CommunityName mapping
config.addV1System(snmpEngine, 'my-area', 'public')

# Specify security settings per SecurityName (SNMPv1 - 0, SNMPv2c - 1)
config.addTargetParams(snmpEngine, 'my-creds', 'my-area', 'noAuthNoPriv', 0)

#
# Setup transport endpoint and bind it with security settings yielding
# a target name
#

# UDP/IPv4
config.addTransport(
    snmpEngine,
    udp.DOMAIN_NAME,
    udp.UdpSocketTransport().openClientMode(('0.0.0.0', 61024))
)

config.addTargetAddr(
    snmpEngine, 'my-router',
    udp.DOMAIN_NAME, ('104.236.166.95', 161),
    'my-creds'
)


# Error/response receiver
# noinspection PyUnusedLocal,PyUnusedLocal,PyUnusedLocal
def cbFun(snmpEngine, sendRequestHandle, errorIndication,
          errorStatus, errorIndex, varBindTable, cbCtx):
    if errorIndication:
        print(errorIndication)
        return False

    if errorStatus:
        print('%s at %s' % (errorStatus.prettyPrint(),
                            errorIndex and varBindTable[-1][int(errorIndex) - 1][0] or '?'))
        return  # stop on error

    for varBindRow in varBindTable:
        for oid, val in varBindRow:
            print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))

    return True  # signal dispatcher to continue


# Prepare initial request to be sent
cmdgen.NextCommandGenerator().sendVarBinds(
    snmpEngine,
    'my-router',
    None, '',  # contextEngineId, contextName
    [((1, 3, 6, 1, 2, 1, 1), None),
     ((1, 3, 6, 1, 2, 1, 11), None)],
    cbFun
)

# Run I/O dispatcher which would send pending queries and process responses
snmpEngine.transportDispatcher.runDispatcher()

Download script.

Spoof source address¶

Send a SNMP GET request * with SNMPv2c, community ‘public’ * over IPv4/UDP * to an Agent at 104.236.166.95:161 * from a non-local, spoofed IP 1.2.3.4 (root and Python 3.3+ required) * for an OID in tuple form

This script performs similar to the following Net-SNMP command:

$ snmpget -v2c -c public -ObentU 104.236.166.95 1.3.6.1.2.1.1.1.0

But unlike the above command, this script issues SNMP request from a non-default, non-local IP address.

It is indeed possible to originate SNMP traffic from any valid local IP addresses. It could be a secondary IP interface, for instance. Superuser privileges are only required to send spoofed packets. Alternatively, sending from local interface could also be achieved by binding to it (via openClientMode() parameter).

from pysnmp.entity import engine, config
from pysnmp.carrier.asyncore.dgram import udp
from pysnmp.entity.rfc3413 import cmdgen

# Create SNMP engine instance
snmpEngine = engine.SnmpEngine()

#
# SNMPv1 setup
#

# SecurityName <-> CommunityName mapping
config.addV1System(snmpEngine, 'my-area', 'public')

# Specify security settings per SecurityName (SNMPv1 - 0, SNMPv2c - 1)
config.addTargetParams(snmpEngine, 'my-creds', 'my-area', 'noAuthNoPriv', 0)

#
# Setup transport endpoint and bind it with security settings yielding
# a target name
#

# Initialize asyncore-based UDP/IPv4 transport
udpSocketTransport = udp.UdpSocketTransport().openClientMode()

# Use sendmsg()/recvmsg() for socket communication (required for
# IP source spoofing functionality)
udpSocketTransport.enablePktInfo()

# Enable IP source spoofing (requires root privileges)
udpSocketTransport.enableTransparent()

# Register this transport at SNMP Engine
config.addTransport(
    snmpEngine,
    udp.DOMAIN_NAME,
    udpSocketTransport
)

# Configure destination IPv4 address as well as source IPv4 address
config.addTargetAddr(
    snmpEngine, 'my-router',
    udp.DOMAIN_NAME, ('104.236.166.95', 161),
    'my-creds',
    sourceAddress=('1.2.3.4', 0)
)


# Error/response receiver
# noinspection PyUnusedLocal,PyUnusedLocal,PyUnusedLocal
def cbFun(snmpEngine, sendRequestHandle, errorIndication,
          errorStatus, errorIndex, varBinds, cbCtx):
    if errorIndication:
        print(errorIndication)

    # SNMPv1 response may contain noSuchName error *and* SNMPv2c exception,
    # so we ignore noSuchName error here
    elif errorStatus and errorStatus != 2:
        print('%s at %s' % (errorStatus.prettyPrint(),
                            errorIndex and varBinds[int(errorIndex) - 1][0] or '?'))

    else:
        for oid, val in varBinds:
            print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))


# Prepare and send a request message
cmdgen.GetCommandGenerator().sendVarBinds(
    snmpEngine,
    'my-router',
    None, '',  # contextEngineId, contextName
    [((1, 3, 6, 1, 2, 1, 1, 1, 0), None)],
    cbFun
)

# Run I/O dispatcher which would send pending queries and process responses
snmpEngine.transportDispatcher.runDispatcher()

Download script.

Walk Agent over IPv6¶

Send a series of SNMP GETNEXT requests with the following options:

with SNMPv3 with user ‘usr-md5-none’, MD5 auth and no privacy protocols
over IPv6/UDP
to an Agent at [::1]:161
for two OIDs in tuple form
stop on end-of-mib condition for both OIDs

This script performs similar to the following Net-SNMP command:

$ snmpwalk -v3 -l authNoPriv -u usr-md5-none -A authkey1 -ObentU udp6:[::1]:161 1.3.6.1.2.1.1 1.3.6.1.4.1.1

from pysnmp.entity import engine, config
from pysnmp.carrier.asyncore.dgram import udp6
from pysnmp.entity.rfc3413 import cmdgen

# Create SNMP engine instance
snmpEngine = engine.SnmpEngine()

#
# SNMPv3/USM setup
#

# user: usr-md5-des, auth: MD5, priv NONE
config.addV3User(
    snmpEngine, 'usr-md5-none',
    config.USM_AUTH_HMAC96_MD5, 'authkey1'
)
config.addTargetParams(snmpEngine, 'my-creds', 'usr-md5-none', 'authNoPriv')

#
# Setup transport endpoint and bind it with security settings yielding
# a target name
#

# UDP/IPv6
config.addTransport(
    snmpEngine,
    udp6.DOMAIN_NAME,
    udp6.Udp6SocketTransport().openClientMode()
)

config.addTargetAddr(
    snmpEngine, 'my-router',
    udp6.DOMAIN_NAME, ('::1', 161),
    'my-creds'
)


# Error/response receiver
# noinspection PyUnusedLocal,PyUnusedLocal,PyUnusedLocal
def cbFun(snmpEngine, sendRequestHandle, errorIndication,
          errorStatus, errorIndex, varBindTable, cbCtx):
    if errorIndication:
        print(errorIndication)
        return

    if errorStatus:
        print('%s at %s' % (errorStatus.prettyPrint(),
                            errorIndex and varBindTable[-1][int(errorIndex) - 1][0] or '?'))
        return  # stop on error

    for varBindRow in varBindTable:

        for oid, val in varBindRow:
            print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))

    return True  # signal dispatcher to continue


# Prepare initial request to be sent
cmdgen.NextCommandGenerator().sendVarBinds(
    snmpEngine,
    'my-router',
    None, '',  # contextEngineId, contextName
    [((1, 3, 6, 1, 2, 1, 1), None),
     ((1, 3, 6, 1, 4, 1, 1), None)],
    cbFun
)

# Run I/O dispatcher which would send pending queries and process responses
snmpEngine.transportDispatcher.runDispatcher()

Download script.

Navigation

Related Topics

Transport tweaks¶

SNMPv2c, custom timeout¶

Send packets from specific local interface¶

Spoof source address¶

Walk Agent over IPv6¶