DEFAULT_TIMEOUT = 6 ERRORS_STRICT = "strict" ERRORS_WARN = "warn" class Version:
    """
    The SNMP Version identifier. This is used in the SNMP :term:`PDU`.
    """

    V2C = 0x01
    V1 = 0x00 class FaultySNMPImplementation(SnmpError):
    """
    Exception which indicates an unexpected response from an SNMP agent.
    """ class NoSuchOID(ErrorResponse):
    """
    This error is returned in any of the following cases:

    * the targeted OID does not support "snmp-set" operations
    * the targeted OID does not exist.
    * the targeted OID is an SMI aggregate type.
    * the targeted OID does not precede a known name in the MIB view.
    """

    DEFAULT_MESSAGE = "No such name/oid"

    def __init__(self, offending_oid, message=""):
        # type: (Optional[ObjectIdentifier], str) -> None
        super().__init__(2, offending_oid, message) class SnmpError(Exception):
    """
    Generic exception originating from the puresnmp package. Every SNMP related
    error inherits from this class.
    """ END_OF_MIB_VIEW = EndOfMibView() class BulkGetRequest(Type):  # type: ignore
    """
    Represents a SNMP GetBulk request
    """

    # pylint: disable=abstract-method

    TYPECLASS = TypeInfo.CONTEXT
    TAG = 5

    def __init__(self, request_id, non_repeaters, max_repeaters, *oids):
        # type: (int, int, int, ObjectIdentifier) -> None
        if len(oids) > MAX_VARBINDS:
            raise TooManyVarbinds(len(oids))
        self.request_id = request_id
        self.non_repeaters = non_repeaters
        self.max_repeaters = max_repeaters
        self.varbinds = []  # type: List[VarBind]
        for oid in oids:
            self.varbinds.append(VarBind(oid, Null()))

    def __bytes__(self) -> bytes:
        wrapped_varbinds = [Sequence(vb.oid, vb.value) for vb in self.varbinds]
        data = [
            Integer(self.request_id),
            Integer(self.non_repeaters),
            Integer(self.max_repeaters),
            Sequence(*wrapped_varbinds),
        ]
        payload = b"".join([to_bytes(chunk) for chunk in data])

        tinfo = TypeInfo(TypeInfo.CONTEXT, TypeInfo.CONSTRUCTED, self.TAG)
        length = encode_length(len(payload))
        return to_bytes(tinfo) + length + payload  # type: ignore

    def __repr__(self):
        # type: () -> str
        oids = [repr(oid) for oid, _ in self.varbinds]
        return "%s(%r, %r, %r, %s)" % (
            self.__class__.__name__,
            self.request_id,
            self.non_repeaters,
            self.max_repeaters,
            ", ".join(oids),
        )

    def __eq__(self, other):
        # type: (Any) -> bool
        # pylint: disable=unidiomatic-typecheck
        return (
            type(other) == type(self)
            and self.request_id == other.request_id
            and self.non_repeaters == other.non_repeaters
            and self.max_repeaters == other.max_repeaters
            and self.varbinds == other.varbinds
        )

    def pretty(self):  # pragma: no cover
        # type: () -> str
        """
        Returns a "prettified" string representing the SNMP message.
        """
        lines = [
            self.__class__.__name__,
            "    Request ID: %s" % self.request_id,
            "    Non Repeaters: %s" % self.non_repeaters,
            "    Max Repeaters: %s" % self.max_repeaters,
            "    Varbinds: ",
        ]
        for bind in self.varbinds:
            lines.append("        %s: %s" % (bind.oid, bind.value))  # type: ignore

        return "\n".join(lines) class GetNextRequest(GetRequest):
    """
    Represents an SNMP GetNext Request.
    """

    TAG = 1 class GetRequest(PDU):
    """
    Represents an SNMP Get Request.
    """

    TAG = 0

    def __init__(
        self, request_id: int, *oids: Union[str, ObjectIdentifier]
    ) -> None:
        if len(oids) > MAX_VARBINDS:
            raise TooManyVarbinds(len(oids))
        wrapped_oids = []
        for oid in oids:
            if isinstance(oid, str):
                wrapped_oids.append(ObjectIdentifier.from_string(oid))
            else:
                wrapped_oids.append(oid)
        super().__init__(
            request_id, [VarBind(oid, Null()) for oid in wrapped_oids]
        ) class GetResponse(PDU):
    """
    Represents an SNMP basic response (this may be returned for other requests
    than GET as well).
    """

    TAG = 2

    @classmethod
    def decode(cls, data):
        # type: (bytes) -> PDU
        """
        Try decoding the response. If nothing was returned (the message was
        empty), raise a :py:exc:`~puresnmp.exc.NoSuchOID` exception.
        """
        # TODO This is not reslly clean, but it should work. A GetResponse has
        #      the same type identifier as a "endOfMibView" *value*. The way
        #      puresnmp is structured (recursively calling "pop_tlv") makes it
        #      difficult to distinguish between a valid GetResponse object and
        #      an endOfMibView value, except that the endOfMibView had no data.
        if not data:
            return END_OF_MIB_VIEW
        try:
            return super().decode(data)
        except EmptyMessage as exc:
            raise NoSuchOID(
                ObjectIdentifier(0), "Nothing found at the given OID (%s)" % exc
            ) from exc class SetRequest(PDU):
    """
    Represents an SNMP SET Request.
    """

    TAG = 3 class VarBind:
    """
    A "VarBind" is a 2-tuple containing an object-identifier and the
    corresponding value.
    """

    # TODO: This class should be split in two for both the raw and pythonic
    #       API, that would simplify the typing of both "oid" and "value"a lot
    #       and keep things explicit
    oid: ObjectIdentifier = ObjectIdentifier(0)
    value: Union[PyType, Type, None] = None

    def __init__(self, oid, value):
        # type: (Union[ObjectIdentifier, str], PyType) -> None
        if not isinstance(oid, (ObjectIdentifier, str)):
            raise TypeError(
                "OIDs for VarBinds must be ObjectIdentifier or str"
                " instances! Your value: %r" % oid
            )
        if isinstance(oid, str):
            oid = ObjectIdentifier.from_string(oid)
        self.oid = oid
        self.value = value

    def __iter__(self) -> Iterator[Union[ObjectIdentifier, PyType]]:
        return iter([self.oid, self.value])

    def __getitem__(self, idx: int) -> Union[PyType, Type, None]:
        return list(self)[idx]

    def __lt__(self, other):
        # type: (Any) -> bool
        return (self.oid, self.value) < (other.oid, other.value)

    def __eq__(self, other):
        # type: (Any) -> bool
        return (self.oid, self.value) == (other.oid, other.value)

    def __hash__(self):
        # type: () -> int
        return hash((self.oid, self.value))

    def __repr__(self):
        # type: () -> str
        return "VarBind(%r, %r)" % (self.oid, self.value) class SocketInfo:
class SocketResponse: class BulkResult:
    """
    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.
    """

    scalars: Dict[str, Any]
    listing: Dict[str, Any] def get_unfinished_walk_oids(grouped_oids):
    # type: (Dict[ObjectIdentifier, List[VarBind]]) -> List[Tuple[ObjectIdentifier, WalkRow]]
    """
    :param 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).
    """

    # grouped_oids contains a list of values for each requested OID. We need to
    # determine if we need to continue fetching: Inspect the last item of each
    # list if those OIDs are still children of the requested IDs we need to
    # continue fetching using *those* IDs (as we're using GetNext behaviour).
    # If they are *not* children of the requested OIDs, we went too far (in the
    # case of a bulk operation) and need to remove all outliers.
    #
    # The above behaviour is the same for both bulk and simple operations. For
    # simple operations we simply have a list of 1 element per OID, but the
    # behaviour is identical

    # Build a mapping from the originally requested OID to the last fetched OID
    # from that tree.
    last_received_oids = {
        k: WalkRow(v[-1], v[-1].oid in k) for k, v in grouped_oids.items() if v
    }

    output = [
        item
        for item in sorted(last_received_oids.items())
        if item[1].unfinished
    ]
    return output def group_varbinds(varbinds, effective_roots, user_roots=None):
    # type: (List[VarBind], List[ObjectIdentifier], Optional[List[ObjectIdentifier]]) -> 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.

    :param varbinds: A list of VarBind instances.
    :param effective_roots: The list of OIDs that were requested from the SNMP
        agent.
    :param 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.
    """
    user_roots = user_roots or []
    n = len(effective_roots)

    results = {}
    for i in range(n):
        results[effective_roots[i]] = varbinds[i::n]

    if user_roots:
        new_results = {}
        for key, value in results.items():
            containment = [base for base in user_roots if key in base]
            if len(containment) > 1:
                raise RuntimeError(
                    "Unexpected OID result. A value was "
                    "contained in more than one base than "
                    "should be possible!"
                )
            if not containment:
                continue
            new_results[containment[0]] = value
            results = new_results

    return results def tablify(varbinds, num_base_nodes=0, base_oid=""):
    # type: ( Iterable[Tuple[Any, Any]], int, str ) -> List[Dict[str, Any]]
    """
    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.from_string('1.2.1.1'), 'row 1 col 1'),
        >>>     (ObjectIdentifier.from_string('1.2.1.2'), 'row 2 col 1'),
        >>>     (ObjectIdentifier.from_string('1.2.2.1'), 'row 1 col 2'),
        >>>     (ObjectIdentifier.from_string('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.from_string('1.2.1.5.10'), 'row 5.10 col 1'),
        >>>     (ObjectIdentifier.from_string('1.2.1.6.10'), 'row 6.10 col 1'),
        >>>     (ObjectIdentifier.from_string('1.2.2.5.10'), 'row 5.10 col 2'),
        >>>     (ObjectIdentifier.from_string('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'},
        ]
    """

    if isinstance(base_oid, str) and base_oid:

        base_oid_parsed = ObjectIdentifier.from_string(base_oid)
        # Each table has a sub-index for the table "entry" so the number of
        # base-nodes needs to be incremented by 1
        num_base_nodes = len(base_oid_parsed)

    rows = {}  # type: Dict[str, Dict[str, Type[PyType]]]
    for oid, value in varbinds:
        if num_base_nodes:
            tail = oid.identifiers[num_base_nodes:]
            col_id, row_id = tail[0], tail[1:]
            row_id = ".".join([str(node) for node in row_id])
        else:
            col_id = str(oid.identifiers[-2])
            row_id = str(oid.identifiers[-1])
        tmp = {
            "0": row_id,
        }
        row = rows.setdefault(row_id, tmp)
        row[str(col_id)] = value
    return list(rows.values()) class Transport(SyncTransport):
    """
    An async variant of the synchronous TCP transport
    """

    # pylint: disable=invalid-overridden-method

    async def send(  # type: ignore
        self, ip, port, packet, timeout=6, loop=None
    ):  # pragma: no cover
        # type: ( str, int, bytes, int, Optional[AbstractEventLoop] ) -> bytes
        # pylint: disable=arguments-differ
        """
        A coroutine that opens a UDP socket to *ip:port*, sends a packet with
        *bytes* and returns the raw bytes as returned from the remote host.

        If the connection fails due to a timeout, a Timeout exception is
        raised.
        """
        if loop is None:
            loop = asyncio.get_event_loop()

        # family could be specified here (and is in the sync implementation),
        # is it needed? are retries necessary for async implementation?
        _, protocol = await loop.create_datagram_endpoint(
            lambda: SNMPClientProtocol(packet, loop),  # type: ignore
            remote_addr=(ip, port),
        )

        response = await protocol.get_data(timeout)  # type: ignore

        return response  # type: ignore
import logging
from collections import OrderedDict
from typing import TYPE_CHECKING, Any, Dict, List, Tuple, cast
from warnings import warn
from x690.types import (  # type: ignore
    Integer,
    ObjectIdentifier,
    OctetString,
    Sequence,
    Type,
)
from x690.util import to_bytes  # type: ignore
from ...const import DEFAULT_TIMEOUT, ERRORS_STRICT, ERRORS_WARN, Version
from ...exc import FaultySNMPImplementation, NoSuchOID, SnmpError
from ...pdu import (
    END_OF_MIB_VIEW,
    BulkGetRequest,
    GetNextRequest,
    GetRequest,
    GetResponse,
    SetRequest,
)
from ...snmp import VarBind
from ...typevars import PyType, TWrappedPyType
from ...util import BulkResult  # NOQA (must be here for type detection)
from ...util import get_unfinished_walk_oids, group_varbinds, tablify
from ..transport import Transport
    from typing import AsyncGenerator, Callable, Coroutine, Set
    from typing import Type as TType
    from typing import TypeVar, Union
"""
This module contains a high-level API to SNMP functions.

The arguments and return values of these functions have types which are
internal to ``puresnmp`` (subclasses of :py:class:`x690.Type`).

Alternatively, there is :py:mod:`puresnmp.aio.api.pythonic` which converts
these values into pure Python types. This makes day-to-day programming a bit
easier but loses type information which may be useful in some edge-cases. In
such a case it's recommended to use :py:mod:`puresnmp.aio.api.raw`.
"""




if TYPE_CHECKING:  # pragma: no cover
    # pylint: disable=unused-import, invalid-name, ungrouped-imports

    TWalkResponse = AsyncGenerator[VarBind, None]
    TFetcher = Callable[
        [str, str, List[str], int, int], Coroutine[Any, Any, List[VarBind]]
    ]

_set = set

LOG = logging.getLogger(__name__)
OID = ObjectIdentifier.from_string


async def get(ip, community, oid, port=161, timeout=DEFAULT_TIMEOUT):
    # type: ( str, str, str, int, int ) -> Type[PyType]
    """
    Executes a simple SNMP GET request and returns a pure Python data
    structure.

    Example::

        >>> await get('192.168.1.1', 'private', '1.2.3.4')
        'non-functional example'
    """
    result = await multiget(ip, community, [oid], port, timeout=timeout)
    return result[0]


async def multiget(ip, community, oids, port=161, timeout=DEFAULT_TIMEOUT):
    # type: ( str, str, List[str], int, int ) -> List[Type[PyType]]
    """
    Executes an SNMP GET request with multiple OIDs and returns a list of pure
    Python objects. The order of the output items is the same order as the OIDs
    given as arguments.

    Example::

        >>> await multiget('192.168.1.1', 'private', ['1.2.3.4', '1.2.3.5'])
        ['non-functional example', 'second value']
    """
    transport = Transport(timeout=timeout)

    parsed_oids = [OID(oid) for oid in oids]

    packet = Sequence(
        Integer(Version.V2C),
        OctetString(community),
        GetRequest(transport.get_request_id(), *parsed_oids),
    )

    response = await transport.send(ip, port, to_bytes(packet))
    raw_response = cast(

        Tuple[Any, Any, GetResponse], Sequence.from_bytes(response)