FabledSteward/plugins/snmp/poller.py

# plugins/snmp/poller.py
"""Ask an SNMP device for its readings, and say in words why it didn't answer.

pysnmp 7.x (what the image installs, #5470) and the deprecated
pysnmp-lextudio 6.x put the asyncio API in different modules with different
names; _api() finds whichever is installed, so a dependency change can't
silently break polling. Without pysnmp at all the poll reports that instead
of raising.
"""
from __future__ import annotations

import logging
from dataclasses import dataclass, field
from types import SimpleNamespace

logger = logging.getLogger(__name__)

TIMEOUT_SECONDS = 5
# OIDs per GET. A device's whole list usually fits in one request; this keeps
# a long custom list under the size an agent will answer.
OIDS_PER_REQUEST = 24

AUTH_PROTOCOLS = {"none": "usmNoAuthProtocol", "md5": "usmHMACMD5AuthProtocol",
                  "sha": "usmHMACSHAAuthProtocol", "sha224": "usmHMAC128SHA224AuthProtocol",
                  "sha256": "usmHMAC192SHA256AuthProtocol", "sha384": "usmHMAC256SHA384AuthProtocol",
                  "sha512": "usmHMAC384SHA512AuthProtocol"}
PRIV_PROTOCOLS = {"none": "usmNoPrivProtocol", "des": "usmDESPrivProtocol", "3des": "usm3DESEDEPrivProtocol",
                  "aes128": "usmAesCfb128Protocol", "aes192": "usmAesCfb192Protocol",
                  "aes256": "usmAesCfb256Protocol"}

# A varbind value of one of these types means the agent has no such object.
_ABSENT = {"NoSuchObject", "NoSuchInstance", "EndOfMibView"}


@dataclass
class Credentials:
    version: str = "2c"
    community: str = ""
    user: str = ""
    auth: str = "none"
    auth_key: str = ""
    priv: str = "none"
    priv_key: str = ""


@dataclass
class PollResult:
    readings: dict[str, float] = field(default_factory=dict)
    error: str | None = None          # the device didn't answer, and why
    missing: list[str] = field(default_factory=list)   # keys it answered without


def _pysnmp_available() -> bool:
    try:
        import pysnmp  # noqa: F401
        return True
    except ImportError:
        return False


def _api() -> SimpleNamespace:
    """The installed pysnmp's asyncio API under one set of names."""
    try:
        from pysnmp.hlapi.v3arch import asyncio as hl
        from pysnmp.hlapi.v3arch.asyncio import auth
        return SimpleNamespace(hl=hl, auth=auth, get=hl.get_cmd, walk=hl.walk_cmd, async_transport=True)
    except ImportError:
        from pysnmp.hlapi import asyncio as hl
        from pysnmp.hlapi.asyncio import auth
        return SimpleNamespace(hl=hl, auth=auth, get=hl.getCmd, walk=hl.walkCmd, async_transport=False)


def _auth_data(api, creds: Credentials):
    if creds.version == "3":
        return api.hl.UsmUserData(
            creds.user,
            authKey=creds.auth_key or None,
            privKey=creds.priv_key or None,
            authProtocol=getattr(api.auth, AUTH_PROTOCOLS.get(creds.auth, "usmNoAuthProtocol")),
            privProtocol=getattr(api.auth, PRIV_PROTOCOLS.get(creds.priv, "usmNoPrivProtocol")))
    return api.hl.CommunityData(creds.community, mpModel=0 if creds.version == "1" else 1)


def _close_engine(engine) -> None:
    """Release the engine's UDP transport socket.

    pysnmp opens a UDP socket per ``SnmpEngine`` and never closes it on its own.
    Since we build a fresh engine for every poll, an unclosed engine leaks one
    file descriptor each scheduler tick; over hours of polling that exhausts the
    process fd limit (``OSError: [Errno 24] Too many open files``), which also
    takes down the app's listening socket. So close it explicitly here.

    The close method/attribute names differ across pysnmp majors (6.2.x lextudio
    is camelCase, canonical 7.x is snake_case), so probe for whatever exists.
    All paths are best-effort — a failed close must never break the poll loop.
    """
    # 7.x exposes a convenience close directly on the engine.
    for meth in ("close_dispatcher", "closeDispatcher"):
        fn = getattr(engine, meth, None)
        if callable(fn):
            try:
                fn()
            except Exception:
                pass
            return
    # 6.2.x: go through the transport dispatcher.
    for attr in ("transport_dispatcher", "transportDispatcher"):
        dispatcher = getattr(engine, attr, None)
        if dispatcher is None:
            continue
        for meth in ("close_dispatcher", "closeDispatcher"):
            fn = getattr(dispatcher, meth, None)
            if callable(fn):
                try:
                    fn()
                except Exception:
                    pass
                return
        return


def explain(error_indication, address: str, port: int, creds: Credentials) -> str:
    """pysnmp's error indication, said as what to check."""
    text = str(error_indication)
    low = text.lower()
    if "timeout" in low or "no snmp response" in low:
        who = (f"the v3 user “{creds.user}”" if creds.version == "3"
               else "the community")
        return (f"No reply from {address}:{port} within {TIMEOUT_SECONDS} s. Check the address and port, "
                f"that SNMP is on, and {who}: a device ignores a request it doesn't accept.")
    if "unknown" in low and "user" in low:
        return f"The device doesn't know the v3 user “{creds.user}”."
    if "digest" in low or "authentication" in low:
        return "The device refused the v3 authentication: check the auth protocol and key."
    if "decrypt" in low or "privacy" in low:
        return "The device couldn't decrypt the request: check the privacy protocol and key."
    if "not in time window" in low:
        return "The device rejected the request as out of its time window; try again."
    return f"SNMP error from {address}:{port}: {text}"


async def _transport(api, address: str, port: int):
    if api.async_transport:
        return await api.hl.UdpTransportTarget.create((address, port), timeout=TIMEOUT_SECONDS, retries=1)
    return api.hl.UdpTransportTarget((address, port), timeout=TIMEOUT_SECONDS, retries=1)


async def poll_device(address: str, port: int, creds: Credentials,
                      oids: list[tuple[str, str]]) -> PollResult:
    """GET every (oid, key) of a device, a few dozen to a request.

    Returns the numeric answers by key, the keys the device answered without
    (no such object, or not a number), and an error in words when it didn't
    answer at all. Never raises: a broken device must not stop the others.
    """
    result = PollResult()
    if not oids:
        return result
    if not _pysnmp_available():
        result.error = "pysnmp isn't installed in this image, so SNMP polling is off."
        return result

    api = _api()
    engine = api.hl.SnmpEngine()
    try:
        try:
            transport = await _transport(api, address, port)
        except Exception as exc:
            result.error = f"Can't reach {address}:{port}: {exc}"
            return result
        auth = _auth_data(api, creds)

        pending = list(oids)
        while pending:
            chunk, pending = pending[:OIDS_PER_REQUEST], pending[OIDS_PER_REQUEST:]
            ei, es, eidx, var_binds = await api.get(
                engine, auth, transport, api.hl.ContextData(),
                *[api.hl.ObjectType(api.hl.ObjectIdentity(oid)) for oid, _ in chunk],
                lookupMib=False)
            if ei:
                result.error = explain(ei, address, port, creds)
                return result
            if es:
                # v1 fails the whole request on the first unknown OID: note it,
                # and ask again for the rest.
                bad = int(eidx) - 1 if eidx else -1
                if 0 <= bad < len(chunk):
                    result.missing.append(chunk[bad][1])
                    pending = chunk[:bad] + chunk[bad + 1:] + pending
                    continue
                result.error = f"The device answered with an error: {es.prettyPrint()}"
                return result
            # A GET answers in the order it asked.
            for (_, key), (_, val) in zip(chunk, var_binds):
                if type(val).__name__ in _ABSENT:
                    result.missing.append(key)
                    continue
                try:
                    result.readings[key] = float(val)
                except (TypeError, ValueError):
                    result.missing.append(key)
        if not result.readings and result.missing:
            result.error = "The device answered but has none of these readings."
        return result
    except Exception as exc:
        logger.warning("SNMP poll of %s:%s failed: %s", address, port, exc)
        result.error = f"SNMP poll failed: {exc}"
        return result
    finally:
        _close_engine(engine)


SYS_DESCR, SYS_UPTIME, SYS_NAME = "1.3.6.1.2.1.1.1.0", "1.3.6.1.2.1.1.3.0", "1.3.6.1.2.1.1.5.0"
IF_DESCR, IF_NAME = "1.3.6.1.2.1.2.2.1.2", "1.3.6.1.2.1.31.1.1.1.1"
INTERFACES_LISTED = 128


async def describe_device(address: str, port: int, creds: Credentials) -> dict:
    """What Test shows: {"error"} or {"name", "description", "uptime_cs",
    "interfaces": [(number, name)]}. Asks the device directly, so only an
    operator's explicit Test calls it, never a page load."""
    if not _pysnmp_available():
        return {"error": "pysnmp isn't installed in this image, so SNMP is off."}
    api = _api()
    engine = api.hl.SnmpEngine()
    try:
        try:
            transport = await _transport(api, address, port)
        except Exception as exc:
            return {"error": f"Can't reach {address}:{port}: {exc}"}
        auth = _auth_data(api, creds)
        ei, es, _, var_binds = await api.get(
            engine, auth, transport, api.hl.ContextData(),
            *[api.hl.ObjectType(api.hl.ObjectIdentity(o)) for o in (SYS_NAME, SYS_DESCR, SYS_UPTIME)],
            lookupMib=False)
        if ei:
            return {"error": explain(ei, address, port, creds)}
        if es:
            return {"error": f"The device answered with an error: {es.prettyPrint()}"}
        values = [None if type(v).__name__ in _ABSENT else v for _, v in var_binds]
        out = {"name": str(values[0]) if values[0] is not None else "",
               "description": str(values[1]) if values[1] is not None else "",
               "uptime_cs": float(values[2]) if values[2] is not None else None,
               "interfaces": []}
        # ifName where the device has it (IF-MIB), else ifDescr.
        for column in (IF_NAME, IF_DESCR):
            found: list[tuple[int, str]] = []
            async for ei, es, _, rows in api.walk(engine, auth, transport, api.hl.ContextData(),
                                                  api.hl.ObjectType(api.hl.ObjectIdentity(column)),
                                                  lexicographicMode=False, lookupMib=False):
                if ei or es:
                    break
                for name, val in rows:
                    if type(val).__name__ in _ABSENT:
                        continue
                    found.append((int(str(name).rsplit(".", 1)[-1]), str(val)))
                if len(found) >= INTERFACES_LISTED:
                    break
            if found:
                out["interfaces"] = found
                break
        return out
    except Exception as exc:
        logger.warning("SNMP test of %s:%s failed: %s", address, port, exc)
        return {"error": f"SNMP test failed: {exc}"}
    finally:
        _close_engine(engine)