Gufo SNMP Example: SNMPv3 Get Request¶

In the previous example, we demonstrated how to request a single item using SNMP v2c in Gufo SNMP. Now, we'll show you how to achieve the same with SNMP v3, which offers a similar API with additional authentication options.

Despite SNMP v3's increased complexity, Gufo SNMP effectively handles all the intricacies, making SNMP v3 operations as straightforward as v2c. Let's modify our previous example to utilize SNMP v3.

get.py
import asyncio
import sys

from gufo.snmp import Aes128Key, DesKey, Md5Key, Sha1Key, User
from gufo.snmp.aio import SnmpSession

AUTH_ALG = {
    "md5": Md5Key,
    "sha": Sha1Key,
}

PRIV_ALG = {
    "des": DesKey,
    "aes128": Aes128Key,
}


def get_user() -> User:
    name = sys.argv[2]
    if len(sys.argv) > 4:
        auth_alg, key = sys.argv[4].split(":", 1)
        auth_key = AUTH_ALG[auth_alg](key.encode())
    else:
        auth_key = None
    if len(sys.argv) > 5:
        priv_alg, key = sys.argv[5].split(":", 1)
        priv_key = PRIV_ALG[priv_alg](key.encode())
    else:
        priv_key = None
    return User(name, auth_key=auth_key, priv_key=priv_key)


async def main(addr: str, user: User, oid: str) -> None:
    async with SnmpSession(addr=addr, user=user) as session:
        r = await session.get(oid)
        print(r)


asyncio.run(main(sys.argv[1], get_user(), sys.argv[3]))

Let's see the details.

get.py
import asyncio
import sys

from gufo.snmp import Aes128Key, DesKey, Md5Key, Sha1Key, User
from gufo.snmp.aio import SnmpSession

AUTH_ALG = {
    "md5": Md5Key,
    "sha": Sha1Key,
}

PRIV_ALG = {
    "des": DesKey,
    "aes128": Aes128Key,
}

Gufo SNMP is an async library. In our case we should run the client from our synchronous script, so we need to import asyncio to use asyncio.run().

get.py
import asyncio
import sys

from gufo.snmp import Aes128Key, DesKey, Md5Key, Sha1Key, User
from gufo.snmp.aio import SnmpSession

AUTH_ALG = {
    "md5": Md5Key,
    "sha": Sha1Key,
}

PRIV_ALG = {
    "des": DesKey,
    "aes128": Aes128Key,
}

Import sys module to parse the CLI argument.

Warning

We use sys.argv only for demonstration purposes. Use argparse or alternatives in real-world applications.

get.py
import asyncio
import sys

from gufo.snmp import Aes128Key, DesKey, Md5Key, Sha1Key, User
from gufo.snmp.aio import SnmpSession

AUTH_ALG = {
    "md5": Md5Key,
    "sha": Sha1Key,
}

PRIV_ALG = {
    "des": DesKey,
    "aes128": Aes128Key,
}

SnmpSession object holds all necessary API, so import it from gufo.snmp.aio. We also need to import User class and key algorithm helpers.

get.py
import asyncio
import sys

from gufo.snmp import Aes128Key, DesKey, Md5Key, Sha1Key, User
from gufo.snmp.aio import SnmpSession

AUTH_ALG = {
    "md5": Md5Key,
    "sha": Sha1Key,
}

PRIV_ALG = {
    "des": DesKey,
    "aes128": Aes128Key,
}

SNMPv3 offers various authentication options, so we define mappings between human-readable names and Gufo SNMP key wrappers to use later in the get_user function.

get.py
import asyncio
import sys

from gufo.snmp import Aes128Key, DesKey, Md5Key, Sha1Key, User
from gufo.snmp.aio import SnmpSession

AUTH_ALG = {
    "md5": Md5Key,
    "sha": Sha1Key,
}

PRIV_ALG = {
    "des": DesKey,
    "aes128": Aes128Key,
}

Similarly, SNMPv3 offers various privacy options, and we create mappings between human-readable names and key wrappers for these privacy options.

get.py
def get_user() -> User:
    name = sys.argv[2]
    if len(sys.argv) > 4:
        auth_alg, key = sys.argv[4].split(":", 1)
        auth_key = AUTH_ALG[auth_alg](key.encode())
    else:
        auth_key = None
    if len(sys.argv) > 5:
        priv_alg, key = sys.argv[5].split(":", 1)
        priv_key = PRIV_ALG[priv_alg](key.encode())
    else:
        priv_key = None
    return User(name, auth_key=auth_key, priv_key=priv_key)

While SNMP v2c relies on a simple community string for authentication, SNMPv3 introduces the more intricate User-Based Security Model (USM). In this model, a user typically consists of a username, along with optional authentication and privacy options. Gufo SNMP encapsulates these details within the User class.

To facilitate the configuration process, we define the get_user function. This function processes command-line arguments and returns an instance of the User class.

get.py
def get_user() -> User:
    name = sys.argv[2]
    if len(sys.argv) > 4:
        auth_alg, key = sys.argv[4].split(":", 1)
        auth_key = AUTH_ALG[auth_alg](key.encode())
    else:
        auth_key = None
    if len(sys.argv) > 5:
        priv_alg, key = sys.argv[5].split(":", 1)
        priv_key = PRIV_ALG[priv_alg](key.encode())
    else:
        priv_key = None
    return User(name, auth_key=auth_key, priv_key=priv_key)

We get the user name from the second command-line positional parameter.

get.py
def get_user() -> User:
    name = sys.argv[2]
    if len(sys.argv) > 4:
        auth_alg, key = sys.argv[4].split(":", 1)
        auth_key = AUTH_ALG[auth_alg](key.encode())
    else:
        auth_key = None
    if len(sys.argv) > 5:
        priv_alg, key = sys.argv[5].split(":", 1)
        priv_key = PRIV_ALG[priv_alg](key.encode())
    else:
        priv_key = None
    return User(name, auth_key=auth_key, priv_key=priv_key)

Authentication options are optional, so we check the fourth command-line parameter.

get.py
def get_user() -> User:
    name = sys.argv[2]
    if len(sys.argv) > 4:
        auth_alg, key = sys.argv[4].split(":", 1)
        auth_key = AUTH_ALG[auth_alg](key.encode())
    else:
        auth_key = None
    if len(sys.argv) > 5:
        priv_alg, key = sys.argv[5].split(":", 1)
        priv_key = PRIV_ALG[priv_alg](key.encode())
    else:
        priv_key = None
    return User(name, auth_key=auth_key, priv_key=priv_key)

If an authentication option is set, it must have the format <alg>:<key>, where:

<alg> - authentication algorithm, which must be one of AUTH_ALG keys.
<key> - an authentication key.

Note

SNMPv3 introduces three forms of keys:

* Password
* Master key
* Localized key

Choose the form deliberately. Gufo SNMP supports all three forms, which can be selected through optional parameters of the *Key classes. This example uses the default form: a password.

Then we find a proper key class via AUTH_ALG mapping and pass a key.

Note

All keys in Gufo SNMP are passed as bytes, so we use .encode() method to convert from str.

get.py
def get_user() -> User:
    name = sys.argv[2]
    if len(sys.argv) > 4:
        auth_alg, key = sys.argv[4].split(":", 1)
        auth_key = AUTH_ALG[auth_alg](key.encode())
    else:
        auth_key = None
    if len(sys.argv) > 5:
        priv_alg, key = sys.argv[5].split(":", 1)
        priv_key = PRIV_ALG[priv_alg](key.encode())
    else:
        priv_key = None
    return User(name, auth_key=auth_key, priv_key=priv_key)

If no authentication key is provided, set it to None to disable authentication.

get.py
def get_user() -> User:
    name = sys.argv[2]
    if len(sys.argv) > 4:
        auth_alg, key = sys.argv[4].split(":", 1)
        auth_key = AUTH_ALG[auth_alg](key.encode())
    else:
        auth_key = None
    if len(sys.argv) > 5:
        priv_alg, key = sys.argv[5].split(":", 1)
        priv_key = PRIV_ALG[priv_alg](key.encode())
    else:
        priv_key = None
    return User(name, auth_key=auth_key, priv_key=priv_key)

Privacy settings are handled just like authentication settings. We expect them in the fifth command-line parameter, and then use PRIV_ALG mapping to get a proper algorithm.

As with authentication, None means no encryption.

get.py
def get_user() -> User:
    name = sys.argv[2]
    if len(sys.argv) > 4:
        auth_alg, key = sys.argv[4].split(":", 1)
        auth_key = AUTH_ALG[auth_alg](key.encode())
    else:
        auth_key = None
    if len(sys.argv) > 5:
        priv_alg, key = sys.argv[5].split(":", 1)
        priv_key = PRIV_ALG[priv_alg](key.encode())
    else:
        priv_key = None
    return User(name, auth_key=auth_key, priv_key=priv_key)

Then we construct and return a User instance.

get.py
async def main(addr: str, user: User, oid: str) -> None:
    async with SnmpSession(addr=addr, user=user) as session:
        r = await session.get(oid)
        print(r)

Asynchronous code must be executed in the asynchronous functions or coroutines. So we define our function as async. We expect the following arguments:

Address of the agent.
User instance.
OID to query.

get.py
async def main(addr: str, user: User, oid: str) -> None:
    async with SnmpSession(addr=addr, user=user) as session:
        r = await session.get(oid)
        print(r)

First, we need to create SnmpSession object which wraps the client's session. The SnmpSession may be used as an instance directly or operated as async context manager with the async with clause. When used as a context manager, the client automatically closes all connections on the exit of context, so its lifetime is defined explicitly.

SnmpSession constructor offers lots of configuration variables for fine-tuning. Refer to the SnmpSession reference for further details. In our example, we set the agent's address and User instance.

get.py
async def main(addr: str, user: User, oid: str) -> None:
    async with SnmpSession(addr=addr, user=user) as session:
        r = await session.get(oid)
        print(r)

We use SnmpSession.get() function to query OID. The function is asynchronous and must be awaited. See SnmpSession.get() reference for further details.

get.py
async def main(addr: str, user: User, oid: str) -> None:
    async with SnmpSession(addr=addr, user=user) as session:
        r = await session.get(oid)
        print(r)

It is up to the application how to deal with the result. In our example we just print it.

get.py
asyncio.run(main(sys.argv[1], get_user(), sys.argv[3]))

Let's run our asynchronous main() function via asyncio.run. Pass the address as the first command-line parameter, construct the user with get_user(), and pass the OID as the third parameter.

Running¶

Let's check our script. Run example as:

$ python3 examples/aio/get-v3.py 127.0.0.1 public 1.3.6.1.2.1.1.6.0 sha:12345678 aes128:87654321
Gufo SNMP Test