puresnmp.api.raw module

This module contains a high-level API to SNMP functions.

The arguments and return values of these functions have types which are internal to puresnmp (subclasses of x690.types.Type).

Alternatively, there is puresnmp.api.pythonic which converts these values into pure Python types. This makes day-to-day programming a bit easier but loses type information which may be useful in some edge-cases.

class puresnmp.api.raw.Client(ip: str, credentials: ~puresnmp.credentials.Credentials, port: int = 161, sender: ~puresnmp.transport.TSender = <function send_udp>, context_name: bytes = b'', engine_id: bytes = b'')

Bases: object

A client to execute SNMP commands on a remote device.

To run SNMP commands on a remote device, create an instance for that device, and then call the instance methods.

All functions are based on asyncio and must be used in an async context.

Credentials need to be instances of classes taken from puresnmp.credentials which are used to determine the appropriate communication model for this client instance.

>>> from puresnmp import Client, ObjectIdentifier, V2C
>>> import warnings
>>> warnings.simplefilter("ignore")
>>> client = Client("192.0.2.1", V2C("public"))
>>> client.get(ObjectIdentifier("1.3.6.1.2.1.1.2.0"))  
<coroutine ...>

Parameters:

ip – The IP-address of the remote SNMP device
credentials – User credentials for the request. These define the underlying protocol in use. See puresnmp.credentials for possible types.
port – The UDP port for the remote device
sender – A callable responsible to send out data to the remote device. The default implementation will use UDP using the IP and port given in the other arguments.
context_name – An optional context for SNMPv3 requests
engine_id – An optional Engine ID for SNMPv3 requests. Helper functions are provided in puresnmp.util to generate valid IDs.

async bulkget(scalar_oids: List[ObjectIdentifier], repeating_oids: List[ObjectIdentifier], max_list_size: int = 1) → BulkResult

Runs a “bulk” get operation and returns a BulkResult instance. This contains both a mapping for the scalar variables (the “non-repeaters”) and an OrderedDict instance containing the remaining list (the “repeaters”).

The OrderedDict is ordered the same way as the SNMP response (whatever the remote device returns).

This operation can retrieve both single/scalar values and lists of values (“repeating values”) in one single request. You can for example retrieve the hostname (a scalar value), the list of interfaces (a repeating value) and the list of physical entities (another repeating value) in one single request.

Note that this behaves like a getnext request for scalar values! So you will receive the value of the OID which is immediately following the OID you specified for both scalar and repeating values!

Parameters:

scalar_oids – contains the OIDs that should be fetched as single value.
repeating_oids – contains the OIDs that should be fetched as list.
max_list_size – defines the max length of each list.

>>> from puresnmp import Client, ObjectIdentifier as OID, V2C
>>> import warnings
>>> warnings.simplefilter("ignore")
>>> client = Client("192.0.2.1", V2C("private"), port=50009)
>>> result = client.bulkget(  
...     scalar_oids=[
...         OID('1.3.6.1.2.1.1.1'),
...         OID('1.3.6.1.2.1.1.2'),
...     ],
...     repeating_oids=[
...         OID('1.3.6.1.2.1.3.1'),
...         OID('1.3.6.1.2.1.5.1'),
...     ],
...     max_list_size=10
... )
BulkResult(
    scalars={
        ObjectIdentifier('1.3.6.1.2.1.1.1.0'): OctetString(
            b'Linux c8582f39c32b 4.15.0-115-generic #116-Ubuntu SMP '
            b'Wed Aug 26 14:04:49 UTC 2020 x86_64'
        ),
        ObjectIdentifier('1.3.6.1.2.1.1.2.0'): ObjectIdentifier(
            '1.3.6.1.4.1.8072.3.2.10'
        )
    },
    listing=OrderedDict([
        (
            ObjectIdentifier('1.3.6.1.2.1.3.1.1.1.8769.1.10.100.0.1'),
            Integer(8769),
        ),
        (ObjectIdentifier('1.3.6.1.2.1.5.1.0'), Counter(1)),
        (
            ObjectIdentifier('1.3.6.1.2.1.3.1.1.2.8769.1.10.100.0.1'),
            OctetString(b'B#>'),
        ),
        (ObjectIdentifier('1.3.6.1.2.1.5.2.0'), Counter(0)),
        (
            ObjectIdentifier('1.3.6.1.2.1.3.1.1.3.8769.1.10.100.0.1'),
            IpAddress(IPv4Address('10.100.0.1')),
        ),
        (ObjectIdentifier('1.3.6.1.2.1.5.3.0'), Counter(0)),
        (ObjectIdentifier('1.3.6.1.2.1.4.1.0'), Integer(1)),
        (ObjectIdentifier('1.3.6.1.2.1.5.4.0'), Counter(1)),
        (ObjectIdentifier('1.3.6.1.2.1.4.2.0'), Integer(64)),
        (ObjectIdentifier('1.3.6.1.2.1.5.5.0'), Counter(0)),
        (ObjectIdentifier('1.3.6.1.2.1.4.3.0'), Counter(4)),
        (ObjectIdentifier('1.3.6.1.2.1.5.6.0'), Counter(0)),
        (ObjectIdentifier('1.3.6.1.2.1.4.4.0'), Counter(0)),
        (ObjectIdentifier('1.3.6.1.2.1.5.7.0'), Counter(0)),
        (ObjectIdentifier('1.3.6.1.2.1.4.5.0'), Counter(0)),
        (ObjectIdentifier('1.3.6.1.2.1.5.8.0'), Counter(0)),
        (ObjectIdentifier('1.3.6.1.2.1.4.6.0'), Counter(0)),
        (ObjectIdentifier('1.3.6.1.2.1.5.9.0'), Counter(0)),
        (ObjectIdentifier('1.3.6.1.2.1.4.7.0'), Counter(0)),
        (ObjectIdentifier('1.3.6.1.2.1.5.10.0'), Counter(0))
    ])
)

async bulktable(oid: ObjectIdentifier, bulk_size: int = 10, _rowtype: Type[TTableRow] = typing.Dict[str, typing.Any]) → List[TTableRow]

Identical to table() but uses “bulk” requests.

“Bulk” requests fetch more than one OID in one request, so they are more efficient, but large return-values may overflow the transport buffer.

Parameters:

oid – Delegated to table()
bulk_size – Number of values to fetch per request.

async bulkwalk(oids: List[ObjectIdentifier], bulk_size: int = 10) → AsyncGenerator[VarBind, None]

Identical to walk() but uses “bulk” requests instead.

“Bulk” requests fetch more than one OID in one request, so they are more efficient, but large return-values may overflow the transport buffer.

Parameters:

oids – Delegated to walk()
bulk_size – Number of values to fetch per request.

configure(**kwargs: Any) → None

Update the configuration of the client.

For temporary configuration changes see reconfigure().

The values that can be overridden are defined in ClientConfig. Any fields in that class can be overridden

>>> client = Client("192.0.2.1", V2C("public"))
>>> client.config.retries
10
>>> client.configure(retries=3)
>>> client.config.retries
3

property context: Context: Accessor to the SNMPv3 context

property credentials: Credentials: Accessor to the client credentials

async get(oid: ObjectIdentifier) → X690Type[Any]

Retrieve the value of a single OID

>>> from asyncio import run
>>> import warnings
>>> from puresnmp import Client, ObjectIdentifier as OID, V2C
>>> warnings.simplefilter("ignore")
>>> client = Client("127.0.0.1", V2C("private"), port=50009)
>>> coro = client.get(OID("1.3.6.1.2.1.1.2.0"))
>>> run(coro)  
ObjectIdentifier('1.3.6.1.4.1.8072.3.2.10')

async getnext(oid: ObjectIdentifier) → VarBind

Executes a single SNMP GETNEXT request (used inside walk).

>>> from puresnmp import Client, ObjectIdentifier as OID
>>> import warnings
>>> warnings.simplefilter("ignore")
>>> client = Client("192.0.2.1", V2C("private"))
>>> # The line below needs to be "awaited" to get the result.
>>> # This is not shown here to make it work with doctest
>>> client.getnext(OID('1.2.3.4'))
<coroutine object ...>

property ip: IPv4Address | IPv6Address: Accessor to the endpoint IP address

async multiget(oids: List[ObjectIdentifier]) → List[X690Type[Any]]

Retrieve (scalar) values from multiple OIDs in one request.

>>> from asyncio import run
>>> import warnings
>>> from puresnmp import Client, ObjectIdentifier as OID, V2C
>>> warnings.simplefilter("ignore")
>>> client = Client("127.0.0.1", V2C("private"), port=50009)
>>> coro = client.multiget(
...     [OID('1.3.6.1.2.1.1.2.0'), OID('1.3.6.1.2.1.1.1.0')]
... )
>>> run(coro)  
[ObjectIdentifier('1.3.6.1.4.1.8072.3.2.10'), OctetString(b'Linux c8582f39c32b 4.15.0-115-generic #116-Ubuntu SMP Wed Aug 26 14:04:49 UTC 2020 x86_64')]

async multigetnext(oids: List[ObjectIdentifier]) → List[VarBind]

Executes a single multi-oid GETNEXT request.

The request sends one packet to the remote host requesting the value of the OIDs following one or more given OIDs.

>>> from asyncio import run
>>> from puresnmp import Client, ObjectIdentifier as OID, V2C
>>> import warnings
>>> warnings.simplefilter("ignore")
>>> client = Client("127.0.0.1", V2C("private"), port=50009)
>>> # The line below needs to be "awaited" to get the result.
>>> # This is not shown here to make it work with doctest
>>> coro = client.multigetnext(
...     [OID('1.3.6.1.2.1.1.2.0'), OID('1.3.6.1.2.1.1.1.0')]
... )
>>> run(coro)  
[VarBind(oid=ObjectIdentifier('1.3.6.1.2.1.1.3.0'), value=TimeTicks(...)), VarBind(oid=ObjectIdentifier('1.3.6.1.2.1.1.2.0'), value=ObjectIdentifier('1.3.6.1.4.1.8072.3.2.10'))]

async multiset(mappings: Dict[ObjectIdentifier, X690Type[Any]]) → Dict[ObjectIdentifier, X690Type[Any]]

Executes an SNMP SET request on multiple OIDs. The result is returned as pure Python data structure.

>>> from asyncio import run
>>> from puresnmp import Client, ObjectIdentifier as OID, V2C
>>> import warnings
>>> warnings.simplefilter("ignore")
>>> from x690.types import OctetString
>>> client = Client("127.0.0.1", V2C("private"), port=50009)
>>> coro = client.multiset({
...     OID('1.3.6.1.2.1.1.4.0'): OctetString(b'new-contact'),
...     OID('1.3.6.1.2.1.1.6.0'): OctetString(b'new-location')
... })
>>> run(coro)  
{ObjectIdentifier('1.3.6.1.2.1.1.4.0'): OctetString(b'new-c...cation')}

async multiwalk(oids: List[ObjectIdentifier], fetcher: Callable[[List[ObjectIdentifier]], Awaitable[List[VarBind]]] | None = None, errors: str = 'strict') → AsyncGenerator[VarBind, None]

Retrieve all values “below” multiple OIDs with a single operation.

Note: This will send out as many “GetNext” requests as needed.

This is almost the same as walk() except that it is capable of iterating over multiple OIDs at the same time.

>>> from asyncio import run
>>> from puresnmp import Client, ObjectIdentifier as OID, V2C
>>> import warnings
>>> warnings.simplefilter("ignore")
>>> async def example():
...     client = Client("127.0.0.1", V2C("private"), port=50009)
...     result = client.multiwalk(
...         [OID('1.3.6.1.2.1.1'), OID('1.3.6.1.4.1.1')]
...     )
...     output = []
...     async for row in result:
...         output.append(row)
...     return output
>>> run(example())  
[VarBind(oid=ObjectIdentifier('1.3.6.1.2.1.1.1.0'), value=Oct...]

property port: int: Accessor to the endpoint port

reconfigure(**kwargs: Any) → Generator[None, None, None]

Temporarily reconfigure the client.

Some values may need to be modified during the lifetime of the client for some requests. A typical example would be using different credentials for “set” commands, or different socket timeouts for some targeted requests.

This is provided via this context-manager. When the context-manager exits, the previous config is restored

The values that can be overridden are defined in ClientConfig. Any fields in that class can be overridden

>>> client = Client("192.0.2.1", V2C("public"))
>>> client.config.timeout
6
>>> with client.reconfigure(timeout=10):
...     client.config.timeout
10
>>> client.config.timeout
6

async set(oid: ObjectIdentifier, value: T) → T

Update a value on the remote host

Values must be a subclass of x690.types.Type. See x690.types for a predefined collection of types.

>>> from asyncio import run
>>> from puresnmp import Client, ObjectIdentifier as OID, V2C
>>> import warnings
>>> warnings.simplefilter("ignore")
>>> from x690.types import OctetString
>>> client = Client("127.0.0.1", V2C("private"), port=50009)
>>> coro = client.set(
...     OID("1.3.6.1.2.1.1.4.0"), OctetString(b'new contact value')
... )
>>> run(coro)  
OctetString(b'new contact value')

async table(oid: ObjectIdentifier, _rowtype: Type[TTableRow] = typing.Dict[str, typing.Any]) → List[TTableRow]

Fetch an SNMP table

The resulting output will be a list of dictionaries where each dictionary corresponds to a row of the table.

SNMP Tables are indexed as follows:

<base-oid>.<column-id>.<row-id>

A “row-id” can be either a single numerical value, or a partial OID. The row-id will be contained in key '0' of each row (as a string) representing that partial OID (often a suffix which can be used in other tables). This key '0' is automatically injected by puresnmp. This ensures that the row-index is available even for tables that don’t include that value themselves.

SNMP-Tables are fetched first by column, then by row (by the nature of the defined MIB structure). This means that this method has to consume the complete table before being able to return anything.

Example output:

>>> from asyncio import run
>>> from puresnmp import Client, ObjectIdentifier as OID, V2C
>>> import warnings
>>> warnings.simplefilter("ignore")
>>> client = Client("127.0.0.1", V2C("private"), port=50009)
>>> coro = client.table(OID("1.3.6.1.2.1.2.2.1"))
>>> run(coro)  
[{'0': '1', '1': Integer(1), ... '22': ObjectIdentifier('0.0')}]

async walk(oid: ObjectIdentifier, errors: str = 'strict') → AsyncGenerator[VarBind, None]: A convenience method delegating to multiwalk() with exactly one OID

class puresnmp.api.raw.ClientConfig(credentials: Credentials, context: Context, lcd: Dict[str, Any], timeout: int = 6, retries: int = 10)

Bases: object

Overridable configuration for SNMP clients

These settings can be overridden via Client.reconfigure()

context: Context: The SNMPv3 Context. For SNMPv1 or SNMPv2 this value is ignored

credentials: Credentials: The credentials used to apply to SNMP requests

lcd: Dict[str, Any]: The SNMPv3 “local config cache”. For SNMPv1 or SNMPv2 this value is ignored

retries: int = 10: The number of retries we attempt when sending packets to the remote device before giving up. Note that some devices may refuse connections attempts if too many requests were made with incorrect credentials. Defaults to 10 retries

timeout: int = 6: The socket timeout for network requests. This value is passed through to the client’s “sender” callable

class puresnmp.api.raw.Context(engine_id: bytes, name: bytes)

Bases: object

Information about the current SNMP context

engine_id: bytes

name: bytes

puresnmp.api.raw.TFetcher

Type alias for a callable that is responsible to fetch a collection of OIDs from the remote device

alias of Callable[[List[ObjectIdentifier]], Awaitable[List[VarBind]]]

puresnmp.api.raw.deduped_varbinds(requested_oids: List[ObjectIdentifier], grouped_oids: Dict[ObjectIdentifier, List[VarBind]], yielded: Set[ObjectIdentifier]) → Generator[VarBind, None, None]

Generate grouped OIDs by ensuring they are contained in the original request and have no duplicates.

>>> OID = ObjectIdentifier
>>> list(deduped_varbinds(
...     [OID("1.2"), OID("2.3")],
...     {
...         OID("1.2"): [VarBind(OID("1.2.3.4"), 1)],
...         OID("1.2"): [VarBind(OID("1.2.3.4"), 1)],
...         OID("5.6"): [VarBind(OID("5.6.7.8"), 1)],
...     },
...     set()
... ))
[VarBind(oid=ObjectIdentifier('1.2.3.4'), value=1)]

Parameters:

requested_oids – A list of OIDs which were originally requested. If any value from the grouped varbinds are not children of any of these OIDs, the issue is logged and the value is skipped.
grouped_oids – The OIDs that need to be verified & deduped.
yielded – A set containing all OIDs that have already been generated by this function. This set will be updated by this function whenever a value is returned to detect duplicates.

puresnmp.api.raw.register_trap_callback(callback: ~typing.Callable[[~puresnmp.pdu.PDU], ~typing.Any], listen_address: str = '0.0.0.0', port: int = 162, credentials: ~puresnmp.credentials.Credentials = <puresnmp.credentials.V2C>, loop: ~asyncio.events.AbstractEventLoop | None = None) → AbstractEventLoop

Registers a callback function for for SNMP traps.

Every time a trap is received, the callback is called with the PDU contained in that trap.

As per RFC 3416#section-4.2.6, the first two varbinds are the system uptime and the trap OID. The following varbinds are the body of the trap

The callback will be called on the current asyncio loop. Alternatively, a loop can be passed into this function in which case, the traps will be handler on that loop instead.