puresnmp.util module

Collection of utility functions for the puresnmp package.

class puresnmp.util.BulkResult(scalars: Dict[ObjectIdentifier, Any], listing: Dict[ObjectIdentifier, Any])

Bases: object

A representation for results of a “bulk” request.

These requests get both “non-repeating values” (scalars) and “repeating values” (lists). This wrapper makes these terms a bit friendlier to use.

listing: Dict[ObjectIdentifier, Any]: A mapping from object-identifiers to “walk” results below those OIDs

scalars: Dict[ObjectIdentifier, Any]: A mapping from object-identifiers to scalar (single) values for those OIDs

class puresnmp.util.TDigestable(*args, **kwargs)

Bases: Protocol

Typing protocol copied from https://github.com/python/typeshed/blob/569fcea637023b3df2fe45c6f08c26037bfa6a74/stdlib/hashlib.pyi#L5

block_size: int

copy() → TDigestable

digest() → bytes

digest_size: int

hexdigest() → str

name: str

update(_TDigestable__data: bytes) → None

class puresnmp.util.WalkRow(value: Any, unfinished: bool)

Bases: object

A wrapper around an SNMP Walk item.

This also keeps track whether this walk result should be considered the last row or not.

unfinished: bool: Whether there are still values following this row

value: Any: The value of the “current” row in a walk operation

puresnmp.util.generate_engine_id_ip(pen: int, ip: IPv4Address | IPv6Address) → bytes

Generates a valid SNMP Engine ID using a private enterprise number and an ip-address.

>>> from ipaddress import ip_address
>>> generate_engine_id_ip(696, ip_address("192.0.2.1"))
b'\x80\x00\x02\xb8\x01\xc0\x00\x02\x01'

See also

Engine ID structure: ASN.1 definition for engine-id encoding
Engine ID types: List of valid engine-id variants
PEN list: List of publicly registered private enterprise numbers

puresnmp.util.generate_engine_id_mac(pen: int, mac_address: str) → bytes

Generates a valid SNMP Engine ID using a private enterprise number and a mac-address.

>>> generate_engine_id_mac(696, "01:02:03:04:05:06")
b'\x80\x00\x02\xb8\x03\x01\x02\x03\x04\x05\x06'

See also

Engine ID structure: ASN.1 definition for engine-id encoding
Engine ID types: List of valid engine-id variants
PEN list: List of publicly registered private enterprise numbers

puresnmp.util.generate_engine_id_octets(pen: int, octets: bytes) → bytes

Generates a valid SNMP Engine ID using a private enterprise number and a custom byte-string (no longer than 27 bytes)

>>> generate_engine_id_octets(696, b"hello")
b'\x80\x00\x02\xb8\x05hello'

See also

Engine ID structure: ASN.1 definition for engine-id encoding
Engine ID types: List of valid engine-id variants
PEN list: List of publicly registered private enterprise numbers

puresnmp.util.generate_engine_id_text(pen: int, text: str) → bytes

Generates a valid SNMP Engine ID using a private enterprise number and a custom text (no longer than 27 characters).

>>> generate_engine_id_text(696, "hello")
b'\x80\x00\x02\xb8\x04hello'

See also

Engine ID structure: ASN.1 definition for engine-id encoding
Engine ID types: List of valid engine-id variants
PEN list: List of publicly registered private enterprise numbers

puresnmp.util.get_request_id() → int

Generates a SNMP request ID.

This returns a simple integer used to validate if a given response matches with the given request.

puresnmp.util.get_unfinished_walk_oids(grouped_oids: Dict[ObjectIdentifier, List[VarBind]]) → List[Tuple[ObjectIdentifier, WalkRow]]

Create a list of OIDs which still have subsequent values in a “walk” operation

Parameters:: grouped_oids – A dictionary containing VarBinds as values. The keys are the base OID of those VarBinds as requested by the user. We need to keep track of the base to be able to tell when a walk over OIDs is finished (that is, when we hit the first OID outside the base).

puresnmp.util.group_varbinds(varbinds: List[VarBind], effective_roots: List[ObjectIdentifier], user_roots: List[ObjectIdentifier] | None = None) → Dict[ObjectIdentifier, List[VarBind]]

Takes a list of varbinds and a list of base OIDs and returns a mapping from those base IDs to lists of varbinds.

Varbinds returned from a walk operation which targets multiple OIDs is returned as “interleaved” list. This functions extracts these interleaved items into a more usable dictionary.

>>> from x690.types import Integer
>>> result = group_varbinds(
...     [
...         VarBind(ObjectIdentifier("1.1.1"), Integer(1)),
...         VarBind(ObjectIdentifier("2.2.2"), Integer(1)),
...         VarBind(ObjectIdentifier("1.1.2"), Integer(1)),
...         VarBind(ObjectIdentifier("2.2.3"), Integer(1)),
...     ],
...     [
...         ObjectIdentifier("1.1"),
...         ObjectIdentifier("2.2"),
...     ],
... )
>>> sorted(result.keys())
[ObjectIdentifier('1.1'), ObjectIdentifier('2.2')]
>>> result[ObjectIdentifier("1.1")]
[VarBind(oid=ObjectIdentifier('1.1.1'), value=Integer(1)), VarBind(oid=ObjectIdentifier('1.1.2'), value=Integer(1))]
>>> result[ObjectIdentifier("2.2")]
[VarBind(oid=ObjectIdentifier('2.2.2'), value=Integer(1)), VarBind(oid=ObjectIdentifier('2.2.3'), value=Integer(1))]

Parameters:

varbinds – A list of VarBind instances.
effective_roots – The list of OIDs that were requested from the SNMP agent.
user_roots – The set of VarBind instances that were requested by the user. This is used internally for walk requests. On each request following the first, the requested OIDs will differ from the OIDs requested by the user. This list will keep track of the original OIDs to determine when the walk needs to terminate.

puresnmp.util.iter_namespace(ns_pkg: ModuleType) → Generator[ModuleInfo, None, None]: Iterates over modules inside the given namespace

puresnmp.util.localise_key(credentials: V3, engine_id: bytes) → bytes

Derive a localised key from the user-credentials.

This follows the logic as dictated by RFC 3414 melding the recipient engine-id with the user-credentials into a kay used for de-/en-cryption of packets.

>>> from puresnmp.credentials import V3, Auth, Priv
>>> engine_id = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02"
>>> localised = localise_key(
...     V3(b"user", Auth(b"maplesyrup", "md5"), Priv(b"privkey", "des")),
...     engine_id
... )
>>> localised.hex()
'4924c679907476d038b258097995a15c'

puresnmp.util.password_to_key(hash_implementation: Callable[[bytes], TDigestable], padding_length: int) → Callable[[bytes, bytes], bytes]

Create a helper function to convert passwords to SNMP compliant keys according to RFC 3414.

>>> hasher = password_to_key(hashlib.sha1, 20)
>>> key = hasher(b"mypasswd", b"target-engine-id")
>>> key.hex()
'999ec23ca66b9d3f187ab5208840c30b0450b452'

Parameters:

hash_implementation – A callable that creates an object with a “.digest()” method from a bytes-object. Usable examples are hashlib.md5 and hashlib.sha1
padding_length – The padding length to be used during hashing (as defined in the SNMP rfc)

Returns:

A callable which can be used to derive an SNMP compliant key from a password.

puresnmp.util.sync(coro: Awaitable[T]) → T

Execute an asyncio corouting in the current event-loop and return the result

>>> async def foo():
...     return "hello"
>>> sync(foo())
'hello'

Warning

This is intended for debugging and testing. If possible you should use your own async environment.

puresnmp.util.tablify(varbinds: Iterable[VarBind], num_base_nodes: int = 0, base_oid: str = '', _rowtype: Type[TTableRow] = typing.Dict[str, typing.Any]) → List[TTableRow]

Converts a list of varbinds into a table-like structure. num_base_nodes can be used for table which row-ids consist of multiple OID tree nodes. By default, the last node is considered the row ID, and the second-last is the column ID. Example:

By default, for the table-cell at OID 1.2.3.4.5, 4 is the column index and 5 is the row index.

Using num_base_nodes=2 will only use the first two nodes (1.2) as table-identifier, so 3 becomes the column index, and 4.5 becomes the row index.

The output should not be considered ordered in any way. If you need it sorted, you must sort it after retrieving the table from this function!

Each element of the output is a dictionary where each key is the column index. By default the index 0 will be added, representing the row ID.

Example:

>>> data = [
...     (ObjectIdentifier('1.2.1.1'), 'row 1 col 1'),
...     (ObjectIdentifier('1.2.1.2'), 'row 2 col 1'),
...     (ObjectIdentifier('1.2.2.1'), 'row 1 col 2'),
...     (ObjectIdentifier('1.2.2.2'), 'row 2 col 2'),
... ]
>>> tablify(data)
[{'0': '1', '1': 'row 1 col 1', '2': 'row 1 col 2'}, {'0': '2', '1': 'row 2 col 1', '2': 'row 2 col 2'}]

Example with longer row ids (using the first two as table identifiers):

>>> data = [
...     (ObjectIdentifier('1.2.1.5.10'), 'row 5.10 col 1'),
...     (ObjectIdentifier('1.2.1.6.10'), 'row 6.10 col 1'),
...     (ObjectIdentifier('1.2.2.5.10'), 'row 5.10 col 2'),
...     (ObjectIdentifier('1.2.2.6.10'), 'row 6.10 col 2'),
... ]
>>> tablify(data, num_base_nodes=2)
[{'0': '5.10', '1': 'row 5.10 col 1', '2': 'row 5.10 col 2'}, {'0': '6.10', '1': 'row 6.10 col 1', '2': 'row 6.10 col 2'}]

puresnmp.util.validate_response_id(request_id: int, response_id: int) → None

Compare request and response IDs and raise an appropriate error.

Raises an appropriate error if the IDs differ. Otherwise returns

This helper method ensures we’re always returning the same exception type on invalid response IDs.