puresnmp.plugins.mpm module

This module provides a plugin architecture for SNMP message-processing models (mpm).

Each mpm plugin can be distributed as separate package by providing modules inside the namespace-package “puresnmp.mpm”. Note that in order to be a valid namespace-package, such a package must not have a __init__.py file!

Example folder-structure for a mpm plugin:

my-security-plugin/
 +- setup.py (or pyproject.toml)
 +- puresnmp/
     +- mpm/
         +- mymodule.py
         +- myothermodule.py

Note that there is no __init__.py file!

In order for modules to be detected as plugin, they must follow the following rules:

  • Have a function create returning an instance of a class implementing the message-processing model. The function must take a network-transport handler and a “local configuration directory” as arguments. A reference implementation can be found in puresnmp.mpm.v3.V3MPM.

  • Contain a int-variable IDENTIFIER. This variable should map to the message-processing model identifiers as defined in the SNMPv3 standard. See https://www.iana.org/assignments/snmp-number-spaces/snmp-number-spaces.xhtml and RFC 3411

class puresnmp.plugins.mpm.AbstractEncodingResult(data: bytes, security_model: SecurityModel[Any, Any] | None = None)

Bases: NamedTuple

Generic type for the output of the conversion from messages to bytes by the message processing modules.

data: bytes

The bytes that are to be sent out to the network

security_model: SecurityModel[Any, Any] | None

The security model that was applied by the message-processing module

puresnmp.plugins.mpm.DISCOVERED_PLUGINS: Dict[int, TMPMPlugin] = {}

Global registry of detected plugins

class puresnmp.plugins.mpm.MessageProcessingModel(transport_handler: Callable[[bytes], Awaitable[bytes]], lcd: Dict[str, Any])

Bases: Generic[TEncodeResult, TSecurityModel]

Each Message Processing Model defines the format of a particular version of an SNMP message and coordinates the preparation and extraction of each such version-specific message format.

See https://tools.ietf.org/html/rfc3411#section-6.2

IDENTIFIER: int

The IANA registered identifier for thie mosule

decode(whole_msg: bytes, credentials: Credentials) → PDU

Convert bytes (as received raw from the network) into an SNMP PDU

Parameters:

  • whole_msg – The message as received from the network

  • credentials – Credentials which will be used if required by the model.

Returns:

A plain SNMP PDU

disco: Any | None

Discovery data (if required by the message-processing-module)

async encode(request_id: int, credentials: Credentials, engine_id: bytes, context_name: bytes, pdu: PDU) → TEncodeResult

Convert an SNMP PDU into raw bytes for the network.

Parameters:

  • request_id – A unique ID for the request.

  • credentials – Credentials which are used by the model if needed (encryption, authentication, …).

  • engine_id – The engine-id for use in SNMPv3. puresnmp.util contains helper methods to create those according to the RFC.

  • context_name – An identifier for the SNMPv3 context

  • pdu – The plain SNMP PDU to process

Returns:

Raw bytes that can be sent to the network

security_model: TSecurityModel | None

The security model applied by this MPM (if any)

class puresnmp.plugins.mpm.TMPMPlugin(*args, **kwargs)

Bases: Protocol

Protocol for the plugin factory function.

create(transport_handler: Callable[[bytes], Awaitable[bytes]], lcd: Dict[str, Any]) → MessageProcessingModel[TEncodeResult, TSecurityModel]

See create()

puresnmp.plugins.mpm.create(identifier: int, transport_handler: Callable[[bytes], Awaitable[bytes]], lcd: Dict[str, Any]) → MessageProcessingModel[Any, Any]

Creates a new instance of a message-processing-model

Parameters:

  • identifier – The IANA ID for the message-processing model

  • transport_handler – A callable that is responsible to send data to the network. It will be called with bytes to be sent, and should return bytes. It should by itself already be aware of where to send them (the only input it gets is the payload). This is used in situations where the message-processing model needs to communicate with the device outside of normal data-exchange message, like the initial discovery message in the SNMPv3 protocol.

  • lcd – A “local configuration directory” which is dynamically updated with “discovery” data if required.

Returns:

A new Message Processing Model instance

>>> async def packet_handler(data: bytes) -> bytes:
...     loop = get_event_loop()
...     _, protocol = await loop.create_datagram_endpoint(
...         lambda: SNMPClientProtocol(packet, loop),
...         remote_addr=("192.0.2.1", 161),
...     )
...     response = await protocol.get_data(timeout)
...     return response
>>> create(3, packet_handler, {})  
<puresnmp_plugins.mpm.v3.V3MPM object ...>
puresnmp.plugins.mpm.is_valid_mpm_plugin(mod: ModuleType) → bool

Return True if the module in mod is usable as MPM plugin