diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f3c54bb..b5f8479 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -10,20 +10,27 @@ on: jobs: build: runs-on: ubuntu-latest + strategy: + max-parallel: 4 + matrix: + python-version: ["3.9", "3.10", "3.11", "3.12"] steps: - - uses: actions/checkout@v3 - - name: Set up Python 3.9 - uses: actions/setup-python@v4 + - uses: actions/checkout@v4 + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 with: - python-version: 3.9 + python-version: ${{ matrix.python-version }} - name: Install dependencies run: | python -m pip install --upgrade pip - pip install pytest pycodestyle + pip install pytest pycodestyle pyright if [ -f requirements.txt ]; then pip install -r requirements.txt; fi # - name: Run tests with pytest # run: | # pytest - name: Lint with PyCodeStyle run: | - find . -name \*.py -exec pycodestyle {} + \ No newline at end of file + find . -name \*.py -exec pycodestyle {} + + - name: Type checking with PyRight + run: | + pyright diff --git a/asyncsnmplib/asn1.py b/asyncsnmplib/asn1.py index 52d7cfa..2e4e20d 100644 --- a/asyncsnmplib/asn1.py +++ b/asyncsnmplib/asn1.py @@ -60,6 +60,8 @@ class Class(enum.IntEnum): TNumber = Union[Number, int] TType = Union[Type, int] TClass = Union[Class, int] +TOid = Tuple[int, ...] +TValue = Any class Tag(NamedTuple): @@ -546,7 +548,7 @@ def _decode_null(bytes_data: bytes) -> None: raise Error("ASN1 syntax error") @staticmethod - def _decode_object_identifier(bytes_data: bytes) -> tuple: + def _decode_object_identifier(bytes_data: bytes) -> TOid: result: List[int] = [] value: int = 0 for i in range(len(bytes_data)): diff --git a/asyncsnmplib/client.py b/asyncsnmplib/client.py index 67a24c9..353c999 100644 --- a/asyncsnmplib/client.py +++ b/asyncsnmplib/client.py @@ -1,15 +1,17 @@ import asyncio +from typing import Iterable, Optional, Tuple, List, Type from .exceptions import ( SnmpNoConnection, SnmpErrorNoSuchName, SnmpTooMuchRows, SnmpNoAuthParams, ) +from .asn1 import Tag, TOid, TValue from .package import SnmpMessage from .pdu import SnmpGet, SnmpGetNext, SnmpGetBulk from .protocol import SnmpProtocol -from .v3.auth import AUTH_PROTO -from .v3.encr import PRIV_PROTO +from .v3.auth import Auth +from .v3.encr import Priv from .v3.package import SnmpV3Message from .v3.protocol import SnmpV3Protocol @@ -17,8 +19,14 @@ class Snmp: version = 1 # = v2 - def __init__(self, host, port=161, community='public', max_rows=10000): - self._loop = asyncio.get_event_loop() + def __init__( + self, + host: str, + port: int = 161, + community: str = 'public', + max_rows: int = 10_000, + loop: Optional[asyncio.AbstractEventLoop] = None): + self._loop = loop if loop else asyncio.get_running_loop() self._protocol = None self._transport = None self.host = host @@ -28,7 +36,7 @@ def __init__(self, host, port=161, community='public', max_rows=10000): # On some systems it seems to be required to set the remote_addr argument # https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.loop.create_datagram_endpoint - async def connect(self, timeout=10): + async def connect(self, timeout: float = 10.0): try: infos = await self._loop.getaddrinfo(self.host, self.port) family, *_, addr = infos[0] @@ -44,7 +52,7 @@ async def connect(self, timeout=10): self._transport = transport def _get(self, oids, timeout=None): - if self._transport is None: + if self._protocol is None: raise SnmpNoConnection pdu = SnmpGet(0, oids) message = SnmpMessage.make(self.version, self.community, pdu) @@ -54,32 +62,34 @@ def _get(self, oids, timeout=None): return self._protocol.send(message) def _get_next(self, oids): - if self._transport is None: + if self._protocol is None: raise SnmpNoConnection pdu = SnmpGetNext(0, oids) message = SnmpMessage.make(self.version, self.community, pdu) return self._protocol.send(message) def _get_bulk(self, oids): - if self._transport is None: + if self._protocol is None: raise SnmpNoConnection pdu = SnmpGetBulk(0, oids) message = SnmpMessage.make(self.version, self.community, pdu) return self._protocol.send(message) - async def get(self, oid, timeout=None): + async def get(self, oid: TOid, timeout: Optional[float] = None + ) -> Tuple[TOid, Tag, TValue]: vbs = await self._get([oid], timeout) return vbs[0] - async def get_next(self, oid): + async def get_next(self, oid: TOid) -> Tuple[TOid, Tag, TValue]: vbs = await self._get_next([oid]) return vbs[0] - async def get_next_multi(self, oids): + async def get_next_multi(self, oids: Iterable[TOid] + ) -> List[Tuple[TOid, TValue]]: vbs = await self._get_next(oids) return [(oid, value) for oid, _, value in vbs if oid[:-1] in oids] - async def walk(self, oid): + async def walk(self, oid: TOid) -> List[Tuple[TOid, TValue]]: next_oid = oid prefixlen = len(oid) rows = [] @@ -115,7 +125,7 @@ def close(self): class SnmpV1(Snmp): version = 0 - async def walk(self, oid): + async def walk(self, oid: TOid) -> List[Tuple[TOid, TValue]]: next_oid = oid prefixlen = len(oid) rows = [] @@ -150,15 +160,14 @@ class SnmpV3(Snmp): def __init__( self, - host, - username, - auth_proto='USM_AUTH_NONE', - auth_passwd=None, - priv_proto='USM_PRIV_NONE', - priv_passwd=None, - port=161, - max_rows=10000): - self._loop = asyncio.get_event_loop() + host: str, + username: str, + auth: Optional[Tuple[Type[Auth], str]] = None, + priv: Optional[Tuple[Type[Priv], str]] = None, + port: int = 161, + max_rows: int = 10_000, + loop: Optional[asyncio.AbstractEventLoop] = None): + self._loop = loop if loop else asyncio.get_running_loop() self._protocol = None self._transport = None self.host = host @@ -170,28 +179,16 @@ def __init__( self._auth_hash_localized = None self._priv_hash = None self._priv_hash_localized = None - try: - self._auth_proto = AUTH_PROTO[auth_proto] - except KeyError: - raise Exception('Supply valid auth_proto') - try: - self._priv_proto = PRIV_PROTO[priv_proto] - except KeyError: - raise Exception('Supply valid priv_proto') - if self._priv_proto and not self._auth_proto: - raise Exception('Supply auth_proto') - if self._auth_proto: - if auth_passwd is None: - raise Exception('Supply auth_passwd') + if auth is not None: + self._auth_proto, auth_passwd = auth self._auth_hash = self._auth_proto.hash_passphrase(auth_passwd) - if self._priv_proto: - if priv_passwd is None: - raise Exception('Supply priv_passwd') - self._priv_hash = self._auth_proto.hash_passphrase(priv_passwd) + if priv is not None: + self._priv_proto, priv_passwd = priv + self._priv_hash = self._auth_proto.hash_passphrase(priv_passwd) # On some systems it seems to be required to set the remote_addr argument # https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.loop.create_datagram_endpoint - async def connect(self, timeout=10): + async def connect(self, timeout: float = 10.0): try: infos = await self._loop.getaddrinfo(self.host, self.port) family, *_, addr = infos[0] @@ -211,6 +208,9 @@ async def connect(self, timeout=10): raise SnmpNoAuthParams async def _get_auth_params(self, timeout=10): + # TODO for long requests this will need to be refreshed + # https://datatracker.ietf.org/doc/html/rfc3414#section-2.2.3 + assert self._protocol is not None pdu = SnmpGet(0, []) message = SnmpV3Message.make(pdu, [b'', 0, 0, b'', b'', b'']) # this request will not retry like the other requests @@ -225,7 +225,7 @@ async def _get_auth_params(self, timeout=10): if self._priv_proto else None def _get(self, oids, timeout=None): - if self._transport is None: + if self._protocol is None: raise SnmpNoConnection elif self._auth_params is None: raise SnmpNoAuthParams @@ -248,7 +248,7 @@ def _get(self, oids, timeout=None): self._priv_hash_localized) def _get_next(self, oids): - if self._transport is None: + if self._protocol is None: raise SnmpNoConnection elif self._auth_params is None: raise SnmpNoAuthParams @@ -262,7 +262,7 @@ def _get_next(self, oids): self._priv_hash_localized) def _get_bulk(self, oids): - if self._transport is None: + if self._protocol is None: raise SnmpNoConnection elif self._auth_params is None: raise SnmpNoAuthParams diff --git a/asyncsnmplib/exceptions.py b/asyncsnmplib/exceptions.py index 559c461..0fbad96 100644 --- a/asyncsnmplib/exceptions.py +++ b/asyncsnmplib/exceptions.py @@ -1,6 +1,5 @@ __all__ = ( "SnmpTimeoutError", - "SnmpUnsupportedValueType", "SnmpErrorTooBig", "SnmpErrorNoSuchName", "SnmpErrorBadValue", diff --git a/asyncsnmplib/mib/mib.py b/asyncsnmplib/mib/mib.py index 019f3c4..5b522d7 100644 --- a/asyncsnmplib/mib/mib.py +++ b/asyncsnmplib/mib/mib.py @@ -66,6 +66,8 @@ def on_mib(mi: dict, mibname: str, mib: dict, lk_definitions: dict): obj['syntax'] = obj['syntax']['syntax'] lk_definitions[name] = obj + elif obj['tp'] == 'TRAP-TYPE': + lk_definitions[name] = obj for name, obj in mib.items(): if 'value' in obj: @@ -90,6 +92,8 @@ def on_mib(mi: dict, mibname: str, mib: dict, lk_definitions: dict): names[name] = obj elif obj['tp'] == 'OBJECT-GROUP': names[name] = obj + elif obj['tp'] == 'NOTIFICATION-TYPE': + names[name] = obj for name, obj in names.items(): other_name = name diff --git a/asyncsnmplib/mib/utils.py b/asyncsnmplib/mib/utils.py index c5688b3..e0cc162 100644 --- a/asyncsnmplib/mib/utils.py +++ b/asyncsnmplib/mib/utils.py @@ -1,4 +1,5 @@ -from typing import Tuple, Union +from typing import Tuple, Union, List +from ..asn1 import TOid, TValue from .mib_index import MIB_INDEX from .syntax_funs import SYNTAX_FUNS @@ -8,7 +9,7 @@ FLAGS_SEPERATOR = ',' -def on_octet_string(value: bytes) -> str: +def on_octet_string(value: TValue) -> Union[str, None]: """ used as a fallback for OCTET STRING when no formatter is found/defined """ @@ -18,13 +19,13 @@ def on_octet_string(value: bytes) -> str: return -def on_integer(value: int) -> str: +def on_integer(value: TValue) -> Union[int, None]: if not isinstance(value, int): return return value -def on_oid_map(oid: Tuple[int]) -> str: +def on_oid_map(oid: TValue) -> Union[str, None]: if not isinstance(oid, tuple): # some devices don't follow mib's syntax # for example ipAddressTable.ipAddressPrefix returns an int in case of @@ -45,7 +46,7 @@ def on_value_map_b(value: bytes, map_: dict) -> str: v for k, v in map_.items() if value[k // 8] & (1 << k % 8)) -def on_syntax(syntax: dict, value: Union[int, str, bytes]): +def on_syntax(syntax: dict, value: TValue): """ this is point where bytes are converted to right datatype """ @@ -65,7 +66,10 @@ def on_syntax(syntax: dict, value: Union[int, str, bytes]): raise Exception(f'Invalid syntax {syntax}') -def on_result(base_oid: Tuple[int], result: dict) -> Tuple[str, list]: +def on_result( + base_oid: TOid, + result: List[Tuple[TOid, TValue]], +) -> Tuple[str, List[dict]]: """returns a more compat result (w/o prefixes) and groups formatted metrics by base_oid """ @@ -109,7 +113,10 @@ def on_result(base_oid: Tuple[int], result: dict) -> Tuple[str, list]: return result_name, list(table.values()) -def on_result_base(base_oid: Tuple[int], result: dict) -> Tuple[str, list]: +def on_result_base( + base_oid: TOid, + result: List[Tuple[TOid, TValue]], +) -> Tuple[str, List[dict]]: """returns formatted metrics grouped by base_oid """ base = MIB_INDEX[base_oid] diff --git a/asyncsnmplib/package.py b/asyncsnmplib/package.py index c7a76e2..6366581 100644 --- a/asyncsnmplib/package.py +++ b/asyncsnmplib/package.py @@ -1,4 +1,5 @@ -from .asn1 import Decoder, Encoder, Number +from typing import Optional, Tuple, List +from .asn1 import Decoder, Encoder, Number, Tag, TOid, TValue class Package: @@ -8,12 +9,13 @@ class Package: pdu = None def __init__(self): - self.request_id = None - self.error_status = None - self.error_index = None - self.variable_bindings = [] + self.request_id: Optional[int] = None + self.error_status: Optional[int] = None + self.error_index: Optional[int] = None + self.variable_bindings: List[Tuple[TOid, Tag, TValue]] = [] def encode(self): + assert self.pdu is not None encoder = Encoder() with encoder.enter(Number.Sequence): diff --git a/asyncsnmplib/protocol.py b/asyncsnmplib/protocol.py index bdd2f7d..92ed7b6 100644 --- a/asyncsnmplib/protocol.py +++ b/asyncsnmplib/protocol.py @@ -1,5 +1,6 @@ import asyncio import logging +from typing import Any from . import exceptions from .package import Package @@ -30,7 +31,7 @@ class SnmpProtocol(asyncio.DatagramProtocol): __slots__ = ('loop', 'target', 'transport', 'requests', '_request_id') def __init__(self, target): - self.loop = asyncio.get_event_loop() + self.loop = asyncio.get_running_loop() self.target = target self.requests = {} self._request_id = 0 @@ -38,7 +39,11 @@ def __init__(self, target): def connection_made(self, transport): self.transport = transport - def datagram_received(self, data: bytes, *args): + def datagram_received(self, data: bytes, addr: Any): + # NOTE on typing + # https://docs.python.org/3/library/asyncio-protocol.html + # addr is the address of the peer sending the data; + # the exact format depends on the transport. pkg = Package() try: pkg.decode(data) @@ -47,8 +52,11 @@ def datagram_received(self, data: bytes, *args): # before request_id is known we cannot do anything and the query # will time out pid = pkg.request_id - if pid is not None: + if pid in self.requests: self.requests[pid].set_exception(exceptions.SnmpDecodeError) + elif pid is not None: + logging.error( + self._log_with_suffix(f'Unknown package pid {pid}')) else: logging.error( self._log_with_suffix('Failed to decode package')) @@ -59,9 +67,9 @@ def datagram_received(self, data: bytes, *args): self._log_with_suffix(f'Unknown package pid {pid}')) else: exception = None - if pkg.error_status != 0: + if pkg.error_status: # also exclude None for trap-pdu oid = None - if pkg.error_index != 0: + if pkg.error_index: # also exclude None for trap-pdu oidtuple = \ pkg.variable_bindings[pkg.error_index - 1][0] oid = '.'.join(map(str, oidtuple)) diff --git a/asyncsnmplib/trapserver.py b/asyncsnmplib/trapserver.py new file mode 100644 index 0000000..cc3da93 --- /dev/null +++ b/asyncsnmplib/trapserver.py @@ -0,0 +1,103 @@ + +import asyncio +import logging +from typing import Optional, Any +from .protocol import SnmpProtocol, Package +from .asn1 import Decoder +from .mib.mib_index import MIB_INDEX + +# TODOK +# GENERIC_TRAP = { +# v['value']: {**v, 'name': k} +# for k, v in MIB_INDEX['RFC-1215'][None].items() +# } + + +def on_package(data): + decoder = Decoder(data) + with decoder.enter(): + decoder.read() # version + decoder.read() # community + + with decoder.enter(): + tag, value = decoder.read() + # print(value) + tag, value = decoder.read() + # print(value) + tag, value = decoder.read() + generic_trap_id = value + # print(value) + tag, value = decoder.read() + # print(value) + tag, value = decoder.read() + # print(value) + + variable_bindings = [] + with decoder.enter(): + while not decoder.eof(): + with decoder.enter(): + _, oid = decoder.read() + tag, value = decoder.read() + variable_bindings.append((oid, tag, value)) + + # print(GENERIC_TRAP[generic_trap_id]) + print(variable_bindings) + + +class SnmpTrapProtocol(SnmpProtocol): + + def datagram_received(self, data: bytes, addr: Any): + # NOTE on typing + # https://docs.python.org/3/library/asyncio-protocol.html + # addr is the address of the peer sending the data; + # the exact format depends on the transport. + pkg = Package() + try: + pkg.decode(data) + except Exception: + # TODO SnmpDecodeError? + logging.error( + self._log_with_suffix('Failed to decode package')) + else: + logging.debug('Trap message received') + for oid, tag, value in pkg.variable_bindings: + mib_object = MIB_INDEX.get(oid[:-1]) + if mib_object is None: + # only accept oids from loaded mibs + continue + logging.info( + f'oid: {oid} name: {mib_object["name"]} value: {value}' + ) + # TODO some values need oid lookup for the value, do here or in + # outside processor + + +class SnmpTrap: + def __init__( + self, + host: str = '0.0.0.0', + port: int = 162, + community: str = 'public', + max_rows: int = 10_000, + loop: Optional[asyncio.AbstractEventLoop] = None): + self._loop = loop if loop else asyncio.get_running_loop() + self._protocol = None + self._transport = None + self.host = host + self.port = port + self.community = community + self.max_rows = max_rows + + async def listen(self): + transport, protocol = await self._loop.create_datagram_endpoint( + lambda: SnmpTrapProtocol((None, None)), + local_addr=(self.host, self.port), + ) + self._protocol = protocol + self._transport = transport + + def close(self): + if self._transport is not None and not self._transport.is_closing(): + self._transport.close() + self._protocol = None + self._transport = None diff --git a/asyncsnmplib/utils.py b/asyncsnmplib/utils.py index 7af691f..b9b8ed3 100644 --- a/asyncsnmplib/utils.py +++ b/asyncsnmplib/utils.py @@ -1,4 +1,6 @@ import logging +from typing import Dict, List, Tuple +from .asn1 import TOid, TValue from .client import Snmp, SnmpV1, SnmpV3 from .exceptions import SnmpException, SnmpNoConnection, SnmpNoAuthParams from .mib.utils import on_result_base @@ -6,16 +8,10 @@ from .v3.encr import PRIV_PROTO -class InvalidCredentialsException(SnmpException): - message = 'Invalid SNMP v3 credentials.' - - -class InvalidClientConfigException(SnmpException): - message = 'Invalid SNMP v3 client configuration.' - - -class InvalidSnmpVersionException(SnmpException): - message = 'Invalid SNMP version.' +class InvalidConfigException(SnmpException): + def __init__(self, message: str): + super().__init__(message) + self.message = message class ParseResultException(SnmpException): @@ -24,58 +20,10 @@ def __init__(self, message: str): self.message = message -def snmpv3_credentials(config: dict): - try: - user_name = config['username'] - except KeyError: - raise Exception(f'missing `username`') - - auth = config.get('auth') - if auth is not None: - auth_type = auth.get('type', 'USM_AUTH_NONE') - if auth_type != 'USM_AUTH_NONE': - if auth_type not in AUTH_PROTO: - raise Exception(f'invalid `auth.type`') - - try: - auth_passwd = auth['password'] - except KeyError: - raise Exception(f'missing `auth.password`') - - priv = config.get('priv', {}) - priv_type = priv.get('type', 'USM_PRIV_NONE') - if priv_type != 'USM_PRIV_NONE': - if priv_type not in PRIV_PROTO: - raise Exception(f'invalid `priv.type`') - - try: - priv_passwd = priv['password'] - except KeyError: - raise Exception(f'missing `priv.password`') - - return { - 'username': user_name, - 'auth_proto': auth_type, - 'auth_passwd': auth_passwd, - 'priv_proto': priv_type, - 'priv_passwd': priv_passwd, - } - else: - return { - 'username': user_name, - 'auth_proto': auth_type, - 'auth_passwd': auth_passwd, - } - else: - return { - 'username': user_name, - } - - async def snmp_queries( address: str, config: dict, - queries: tuple): + queries: Tuple[TOid, ...]) -> Dict[str, List[Tuple[TOid, TValue]]]: version = config.get('version', '2c') @@ -84,38 +32,51 @@ async def snmp_queries( if isinstance(community, dict): community = community.get('secret') if not isinstance(community, str): - raise TypeError('SNMP community must be a string.') + raise InvalidConfigException('`community` must be a string.') cl = Snmp( host=address, community=community, ) elif version == '3': - try: - cred = snmpv3_credentials(config) - except Exception as e: - logging.warning(f'invalid snmpv3 credentials {address}: {e}') - raise InvalidCredentialsException - try: - cl = SnmpV3( - host=address, - **cred, - ) - except Exception as e: - logging.warning(f'invalid snmpv3 client config {address}: {e}') - raise InvalidClientConfigException + username = config.get('username') + if not isinstance(username, str): + raise InvalidConfigException('`username` must be a string.') + auth = config.get('auth') + if auth: + auth_proto = AUTH_PROTO.get(auth.get('type')) + auth_passwd = auth.get('password') + if auth_proto is None: + raise InvalidConfigException('`auth.type` invalid') + elif not isinstance(auth_passwd, str): + raise InvalidConfigException('`auth.password` must be string') + auth = (auth_proto, auth_passwd) + priv = auth and config.get('priv') + if priv: + priv_proto = PRIV_PROTO.get(priv.get('type')) + priv_passwd = priv.get('password') + if priv_proto is None: + raise InvalidConfigException('`priv.type` invalid') + elif not isinstance(priv_passwd, str): + raise InvalidConfigException('`priv.password` must be string') + priv = (priv, priv_passwd) + cl = SnmpV3( + host=address, + username=username, + auth=auth, + priv=priv, + ) elif version == '1': community = config.get('community', 'public') if isinstance(community, dict): community = community.get('secret') if not isinstance(community, str): - raise TypeError('SNMP community must be a string.') + raise InvalidConfigException('`community` must be a string.') cl = SnmpV1( host=address, community=community, ) else: - logging.warning(f'unsupported snmp version {address}: {version}') - raise InvalidSnmpVersionException + raise InvalidConfigException(f'unsupported snmp version {version}') try: await cl.connect() diff --git a/asyncsnmplib/v3/auth.py b/asyncsnmplib/v3/auth.py index 2fb8023..facba36 100644 --- a/asyncsnmplib/v3/auth.py +++ b/asyncsnmplib/v3/auth.py @@ -1,5 +1,6 @@ import struct from hashlib import md5, sha1 +from typing import Callable, Type, Dict def hash_passphrase(passphrase, hash_func): @@ -71,20 +72,25 @@ def authenticate_sha(auth_key, msg): return msg.replace(b'\x00' * 12, d2[:12], 1) -class USM_AUTH_HMAC96_MD5: +class Auth: + hash_passphrase: Callable + localize: Callable + auth: Callable + + +class USM_AUTH_HMAC96_MD5(Auth): hash_passphrase = hash_passphrase_md5 localize = localize_key_md5 auth = authenticate_md5 -class USM_AUTH_HMAC96_SHA: +class USM_AUTH_HMAC96_SHA(Auth): hash_passphrase = hash_passphrase_sha localize = localize_key_sha auth = authenticate_sha -AUTH_PROTO = { +AUTH_PROTO: Dict[str, Type[Auth]] = { 'USM_AUTH_HMAC96_MD5': USM_AUTH_HMAC96_MD5, 'USM_AUTH_HMAC96_SHA': USM_AUTH_HMAC96_SHA, - 'USM_AUTH_NONE': None, } diff --git a/asyncsnmplib/v3/encr.py b/asyncsnmplib/v3/encr.py index ba88191..8d2a014 100644 --- a/asyncsnmplib/v3/encr.py +++ b/asyncsnmplib/v3/encr.py @@ -3,6 +3,7 @@ from Crypto.Cipher import DES, AES from Crypto.Random import get_random_bytes from Crypto.Util.Padding import pad +from typing import Callable, Type, Dict def encrypt_data(key, data, msgsecurityparams): @@ -73,12 +74,17 @@ def decrypt_data_aes(key, data, msgsecurityparams): return obj.decrypt(pad(data, 16)) -class USM_PRIV_CBC56_DES: +class Priv: + encrypt: Callable + decrypt: Callable + + +class USM_PRIV_CBC56_DES(Priv): encrypt = encrypt_data decrypt = decrypt_data -class USM_PRIV_CFB128_AES: +class USM_PRIV_CFB128_AES(Priv): encrypt = encrypt_data_aes decrypt = decrypt_data_aes @@ -86,5 +92,4 @@ class USM_PRIV_CFB128_AES: PRIV_PROTO = { 'USM_PRIV_CBC56_DES': USM_PRIV_CBC56_DES, 'USM_PRIV_CFB128_AES': USM_PRIV_CFB128_AES, - 'USM_PRIV_NONE': None, } diff --git a/asyncsnmplib/v3/protocol.py b/asyncsnmplib/v3/protocol.py index 2294215..3227411 100644 --- a/asyncsnmplib/v3/protocol.py +++ b/asyncsnmplib/v3/protocol.py @@ -1,5 +1,6 @@ import asyncio import logging +from typing import Any from ..exceptions import SnmpTimeoutError from ..protocol import SnmpProtocol, _ERROR_STATUS_TO_EXCEPTION from .package import Package @@ -7,7 +8,11 @@ class SnmpV3Protocol(SnmpProtocol): - def datagram_received(self, data, *args): + def datagram_received(self, data: bytes, addr: Any): + # NOTE on typing + # https://docs.python.org/3/library/asyncio-protocol.html + # addr is the address of the peer sending the data; + # the exact format depends on the transport. pkg = Package() try: pkg.decode(data) diff --git a/asyncsnmplib/version.py b/asyncsnmplib/version.py index e6d0c4f..7fd229a 100644 --- a/asyncsnmplib/version.py +++ b/asyncsnmplib/version.py @@ -1 +1 @@ -__version__ = '0.1.12' +__version__ = '0.2.0' diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..381efb0 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,6 @@ +[tool.pyright] +pythonVersion = "3.12" +pythonPlatform = "Linux" +include = [ + "asyncsnmplib" +] \ No newline at end of file diff --git a/setup.py b/setup.py index 1fce998..6535f11 100644 --- a/setup.py +++ b/setup.py @@ -37,10 +37,10 @@ classifiers=[ 'Intended Audience :: Developers', 'Operating System :: OS Independent', - 'Programming Language :: Python :: 3.7', - 'Programming Language :: Python :: 3.8', 'Programming Language :: Python :: 3.9', 'Programming Language :: Python :: 3.10', + 'Programming Language :: Python :: 3.11', + 'Programming Language :: Python :: 3.12', 'Programming Language :: Python :: 3 :: Only', 'Topic :: Software Development :: Libraries :: Python Modules', ],