puresnmp package

This module contains the high-level functions to access the library. Care is taken to make this as pythonic as possible and hide as many of the gory implementations as possible.

class puresnmp.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: Union[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: Optional[Callable[[List[ObjectIdentifier]], Awaitable[List[VarBind]]]] = 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.ObjectIdentifier(value: ~typing.Union[str, ~x690.types._SENTINEL_UNINITIALISED] = <x690.types._SENTINEL_UNINITIALISED object>)

Bases: X690Type[str]

Represents an OID.

Instances of this class support containment checks to determine if one OID is a sub-item of another:

>>> ObjectIdentifier("1.2.3.4.5") in ObjectIdentifier("1.2.3")
True

>>> ObjectIdentifier("1.2.4.5.6") in ObjectIdentifier("1.2.3")
False

NATURE = [<TypeNature.PRIMITIVE: 'primitive'>]: The x690 “private/constructed” information

TAG: int = 6: The x690 identifier for the type

childof(other: ObjectIdentifier) → bool: Convenience method to check whether this OID is a child of another OID

collapse_identifiers() → Tuple[int, ...]

Meld the first two octets into one octet as defined by x.690

In x.690 ObjectIdentifiers are a sequence of numbers. In the byte-representation the first two of those numbers are stored in the first byte.

This function takes a “human-readable” OID tuple and returns a new tuple with the first two elements merged (collapsed) together.

>>> ObjectIdentifier("1.3.6.1.4.1").collapse_identifiers()
(43, 6, 1, 4, 1)
>>> ObjectIdentifier().collapse_identifiers()
()

static decode_large_value(current_char: int, stream: Iterator[int]) → int

If we encounter a value larger than 127, we have to consume from the stram until we encounter a value below 127 and recombine them.

See: https://msdn.microsoft.com/en-us/library/bb540809(v=vs.85).aspx

static decode_raw(data: bytes, slc: slice = slice(None, None, None)) → str

Converts the raw byte-value (without type & length header) into a pure Python type

Overrides decode_raw()

static encode_large_value(value: int) → List[int]: Inverse function of decode_large_value()

encode_raw() → bytes: Overrides X690Type.encode_raw()

property nodes: Tuple[int, ...]

Returns the numerical nodes for this instance as tuple

>>> ObjectIdentifier("1.2.3").nodes
(1, 2, 3)
>>> ObjectIdentifier().nodes
()

parentof(other: ObjectIdentifier) → bool: Convenience method to check whether this OID is a parent of another OID

pyvalue: Union[TWrappedPyType, _SENTINEL_UNINITIALISED]: The decoded (or to-be encoded) Python value

class puresnmp.PyWrapper(client: Client)

Bases: object

A wrapper around a puresnmp.api.raw.Client instance.

The wrapper ensures conversion of internal API data-types to and from Python-native types.

Using Python native types shields from internal changes internally in puresnmp at the cost of flexibility. Most applications should mostly benefit from this.

In cases internal data-types are still wanted, one can access the .client attribute of PyWrapper instances which exposes the same API but with internally used data-types.

async bulkget(scalar_oids: List[str], repeating_oids: List[str], max_list_size: int = 10) → BulkResult: Delegates to puresnmp.api.raw.Client.bulkget() but converts internal types to simple Python types.

async bulktable(oid: str, bulk_size: int = 10, _rowtype: TTableRow = typing.Dict[str, typing.Any]) → List[TTableRow]: Delegates to puresnmp.api.raw.Client.bulktable() but converts internal types to simple Python types.

async bulkwalk(oids: List[str], bulk_size: int = 10) → AsyncGenerator[PyVarBind, None]: Delegates to puresnmp.api.raw.Client.bulkwalk() but converts internal types to simple Python types.

client: Client

async get(oid: str) → Any: Delegates to puresnmp.api.raw.Client.get() but converts internal types to simple Python types.

async getnext(oid: str) → PyVarBind: Delegates to puresnmp.api.raw.Client.getnext() but converts internal types to simple Python types.

async multiget(oids: List[str]) → List[Any]: Delegates to puresnmp.api.raw.Client.multiget() but converts internal types to simple Python types.

async multiset(mappings: Dict[str, X690Type[Any]]) → Dict[str, Any]: Delegates to puresnmp.api.raw.Client.multiset() but converts internal types to simple Python types.

async multiwalk(oids: List[str]) → AsyncGenerator[PyVarBind, None]: Delegates to puresnmp.api.raw.Client.multiwalk() but converts internal types to simple Python types.

async set(oid: str, value: X690Type[Any]) → Dict[str, Any]: Delegates to puresnmp.api.raw.Client.set() but converts internal types to simple Python types.

async table(oid: str, _rowtype: TTableRow = typing.Dict[str, typing.Any]) → List[Dict[str, Any]]: Delegates to puresnmp.api.raw.Client.table() but converts internal types to simple Python types.

async walk(oid: str, errors: str = 'strict') → AsyncGenerator[PyVarBind, None]: Delegates to puresnmp.api.raw.Client.walk() but converts internal types to simple Python types.

class puresnmp.V1(community: str)

Bases: Credentials

Credentials for SNMPv1 exchanges

community: str: The community-string

class puresnmp.V2C(community: str)

Bases: V1

Credentials for community-based SNMPv2 exchanges

community: str: The community-string

class puresnmp.V3(username: str, auth: Optional[Auth] = None, priv: Optional[Priv] = None)

Bases: Credentials

Credentials for SNMPv3 exchanges

auth: Optional[Auth]: Authentication details. If left at “None”, authentication is disabled.

priv: Optional[Priv]: Encryption details. If left at “None”, encryption is disabled. Note that for encryption to be enabled, authentication must also be enabled.

username: str: The local username for SNMP exchanges

puresnmp package

Subpackages

Submodules