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 Trap(PDU):
    """
    Represents an SNMP Trap
    """

    TAG = 7

    def __init__(self, *args, **kwargs):
        # type: (Any, Any) -> None
        super().__init__(*args, **kwargs)
        self.source = None  # type: Optional[SocketInfo] 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 Transport:
    """
    A simple UDP transport.

    Calling ``send`` will attempt to send a packet as many times as specified
    in *retries* (default=3). If it fails after those attemps, a
    py:exc:`puresnmp.Timeout` exception is raised.

    :param timeout: How long to wait on the socket before retrying
    :param retries: The number of retries attempted if a low-level
        socket-timeout occurs. If set to ``None`` it will use
        :py:data:`puresnmp.transport.RETRIES` as default.
    :param buffer_size: How much data to read from the socket. If this is too
        small, it may result in incomplete (corrupt) packages. This should be
        kept as low as possible (see ``man(2) recv``). If set to ``None`` it
        will use :py:data:`puresnmp.transport.BUFFER_SIZE` as default.
    """

    def __init__(self, timeout=2, retries=None, buffer_size=None):
        # type: (int, Optional[int], Optional[int]) -> None
        self.timeout = timeout
        self.retries = retries or RETRIES
        self.buffer_size = buffer_size or BUFFER_SIZE

    def send(self, ip, port, packet, timeout=2):  # pragma: no cover
        # type: ( str, int, bytes, int ) -> bytes
        """
        Opens a TCP connection to *ip:port*, sends a packet with *bytes* and
        returns the raw bytes as returned from the remote host.
        """
        checked_ip = ip_address(ip)
        if checked_ip.version == 4:
            address_family = socket.AF_INET
        else:
            address_family = socket.AF_INET6

        sock = socket.socket(address_family, socket.SOCK_DGRAM)
        sock.settimeout(timeout or self.timeout)

        for num_retry in range(self.retries):
            try:
                if LOG.isEnabledFor(logging.DEBUG):
                    hexdump = visible_octets(packet)
                    LOG.debug(
                        "Sending packet to %s:%s (attempt %d/%d)\n%s",
                        ip,
                        port,
                        (num_retry + 1),
                        self.retries,
                        hexdump,
                    )
                sock.sendto(packet, (ip, port))
                response = sock.recv(self.buffer_size)
                break
            except socket.timeout:
                LOG.debug(
                    "Timeout during attempt #%d", (num_retry + 1)
                )  # TODO add detail
                continue
        else:
            sock.close()
            raise Timeout("Max of %d retries reached" % self.retries)
        sock.close()

        if LOG.isEnabledFor(logging.DEBUG):
            hexdump = visible_octets(response)
            LOG.debug("Received packet:\n%s", hexdump)

        return response

    def listen(self, bind_address="0.0.0.0", port=162):  # pragma: no cover
        # type: (str, int) -> Generator[SocketResponse, None, None]
        """
        Sets up a listening UDP socket and returns a generator over recevied
        packets::

            >>> transport = Transport()
            >>> for seq, packet in enumerate(transport.listen()):
            ...     print(seq, repr(packet))
            0, b'...'
            1, b'...'
            2, b'...'

        .. note::

            This defaults to the standard SNMP Trap port 162. This is a
            privileged port so processes using this port must run as root!
        """
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
            sock.bind((bind_address, port))
            while True:
                request, addr = sock.recvfrom(self.buffer_size)
                if LOG.isEnabledFor(logging.DEBUG):
                    hexdump = visible_octets(request)
                    LOG.debug("Received packet:\n%s", hexdump)

                yield SocketResponse(request, SocketInfo(addr[0], addr[1]))

    def get_request_id(self):  # pragma: no cover
        # type: () -> int
        # pylint: disable=no-self-use
        """
        Generates a SNMP request ID. This value should be unique for each
        request.
        """
        # TODO check if this is good enough. My gut tells me "no"! Depends if
        # it has to be unique across all clients, or just one client. If it's
        # just one client it *may* be enough.

        return int(time()) 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())
import logging
from collections import OrderedDict
from typing import TYPE_CHECKING, Any, Callable, List, Tuple
from typing import Type as TType
from typing import TypeVar, 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,
    Trap,
)
from ..snmp import VarBind
from ..transport import Transport
from ..util import BulkResult  # NOQA (must be here for type detection)
from ..util import get_unfinished_walk_oids, group_varbinds, tablify
    from typing import Dict, Generator, Set, Union
    from puresnmp.typevars import PyType

    if any([not isinstance(v, Type) for k, v in mappings]):
        raise TypeError(
            "SNMP requires typing information. The value for a "
            '"set" request must be an instance of "Type"!'
        )

    binds = [VarBind(OID(k), v) for k, v in mappings]  # type: ignore

    request = SetRequest(transport.get_request_id(), binds)
    packet = Sequence(Integer(version), OctetString(community), request)
    response = transport.send(ip, port, to_bytes(packet))
    raw_response = cast(
        Tuple[Any, Any, GetResponse], Sequence.from_bytes(response)
    )
    output = {str(oid): value for oid, value in raw_response[2].varbinds}
    if len(output) != len(mappings):
        raise SnmpError(
            "Unexpected response. Expected %d varbinds, "
            "but got %d!" % (len(mappings), len(output))
        )
    return output  # type: ignore


def bulkget(
    ip,
    community,
    scalar_oids,
    repeating_oids,
    max_list_size=1,
    port=161,
    timeout=DEFAULT_TIMEOUT,
    version=Version.V2C,
):
    # type: (str, str, List[str], List[str], int, int, int, int) -> BulkResult
    # pylint: disable=unused-argument, too-many-locals
    """
    Runs a "bulk" get operation and returns a :py:class:`~.BulkResult`
    instance.  This contains both a mapping for the scalar variables (the
    "non-repeaters") and an OrderedDict instance containing the remaining list
    (the "repeaters").

    The OrderedDict is ordered the same way as the SNMP response (whatever the
    remote device returns).

    This operation can retrieve both single/scalar values *and* lists of values
    ("repeating values") in one single request. You can for example retrieve
    the hostname (a scalar value), the list of interfaces (a repeating value)
    and the list of physical entities (another repeating value) in one single
    request.

    Note that this behaves like a **getnext** request for scalar values! So you
    will receive the value of the OID which is *immediately following* the OID
    you specified for both scalar and repeating values!

    :param scalar_oids: contains the OIDs that should be fetched as single
        value.
    :param repeating_oids: contains the OIDs that should be fetched as list.
    :param max_list_size: defines the max length of each list.

    Example::

        >>> ip = '192.168.1.1'
        >>> community = 'private'
        >>> result = bulkget(
        ...     ip,
        ...     community,
        ...     scalar_oids=['1.3.6.1.2.1.1.1',
        ...                  '1.3.6.1.2.1.1.2'],
        ...     repeating_oids=['1.3.6.1.2.1.3.1',
        ...                     '1.3.6.1.2.1.5.1'],
        ...     max_list_size=10)
        BulkResult(
            scalars={'1.3.6.1.2.1.1.2.0': '1.3.6.1.4.1.8072.3.2.10',
                     '1.3.6.1.2.1.1.1.0': b'Linux aafa4dce0ad4 4.4.0-28-'
                                          b'generic #47-Ubuntu SMP Fri Jun 24 '
                                          b'10:09:13 UTC 2016 x86_64'},
            listing=OrderedDict([
                ('1.3.6.1.2.1.3.1.1.1.10.1.172.17.0.1', 10),
                ('1.3.6.1.2.1.5.1.0', b'\x01'),
                ('1.3.6.1.2.1.3.1.1.2.10.1.172.17.0.1', b'\x02B\x8e>\x9ee'),
                ('1.3.6.1.2.1.5.2.0', b'\x00'),
                ('1.3.6.1.2.1.3.1.1.3.10.1.172.17.0.1', b'\xac\x11\x00\x01'),
                ('1.3.6.1.2.1.5.3.0', b'\x00'),
                ('1.3.6.1.2.1.4.1.0', 1),
                ('1.3.6.1.2.1.5.4.0', b'\x01'),
                ('1.3.6.1.2.1.4.3.0', b'\x00\xb1'),
                ('1.3.6.1.2.1.5.5.0', b'\x00'),
                ('1.3.6.1.2.1.4.4.0', b'\x00'),
                ('1.3.6.1.2.1.5.6.0', b'\x00'),
                ('1.3.6.1.2.1.4.5.0', b'\x00'),
                ('1.3.6.1.2.1.5.7.0', b'\x00'),
                ('1.3.6.1.2.1.4.6.0', b'\x00'),
                ('1.3.6.1.2.1.5.8.0', b'\x00'),
                ('1.3.6.1.2.1.4.7.0', b'\x00'),
                ('1.3.6.1.2.1.5.9.0', b'\x00'),
                ('1.3.6.1.2.1.4.8.0', b'\x00'),
                ('1.3.6.1.2.1.5.10.0', b'\x00')]))
    """
    transport = Transport(timeout=timeout)

    scalar_oids = scalar_oids or []  # protect against empty values
    repeating_oids = repeating_oids or []  # protect against empty values

    oids = [OID(oid) for oid in scalar_oids] + [
        OID(oid) for oid in repeating_oids
    ]

    non_repeaters = len(scalar_oids)

    packet = Sequence(
        Integer(version),
        OctetString(community),
        BulkGetRequest(
            transport.get_request_id(), non_repeaters, max_list_size, *oids
        ),
    )

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

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