FabledSteward/plugins/snmp/service.py

# plugins/snmp/service.py
"""SNMP devices: the form a person edits, the row it becomes, what the poller
asks for, and what a page says about the newest poll."""
from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime

from steward.core.counter_rates import CounterRates
from steward.core.crypto import decrypt_secret, encrypt_secret

from .models import SnmpDevice
from .poller import AUTH_PROTOCOLS, PRIV_PROTOCOLS, Credentials
from .readings import PRESETS, Reading, device_readings, parse_interfaces

VERSIONS = ("1", "2c", "3")
PRESET_IDS = set(PRESETS) | {"interfaces"}
SECRET_FIELDS = ("community", "v3_auth_key", "v3_priv_key")


class FormError(ValueError):
    """What is wrong with a submitted device form, said to the person."""


def readings_for(device: SnmpDevice) -> list[Reading]:
    """What the device is asked for. A row saved through the form always
    parses; one that somehow doesn't is asked for nothing rather than raising
    in the poll loop."""
    try:
        return device_readings(device.presets or [], parse_interfaces(device.interfaces),
                               device.custom, device.version)
    except ValueError:
        return []


def credentials(device: SnmpDevice) -> Credentials:
    return Credentials(
        version=device.version,
        community=decrypt_secret(device.community, context=f"snmp device {device.name} community"),
        user=device.v3_user, auth=device.v3_auth,
        auth_key=decrypt_secret(device.v3_auth_key, context=f"snmp device {device.name} auth key"),
        priv=device.v3_priv,
        priv_key=decrypt_secret(device.v3_priv_key, context=f"snmp device {device.name} privacy key"))


def form_values(device: SnmpDevice | None) -> dict:
    """The form's fields for an existing device. Secrets are never sent back
    to the browser; a blank secret field on save keeps the stored one."""
    if device is None:
        return {"port": 161, "version": "2c", "presets": ["system"], "v3_auth": "sha", "v3_priv": "aes128"}
    return {"name": device.name, "address": device.address, "port": device.port,
            "version": device.version, "v3_user": device.v3_user, "v3_auth": device.v3_auth,
            "v3_priv": device.v3_priv, "presets": list(device.presets or []),
            "interfaces": device.interfaces, "custom": device.custom, "host_id": device.host_id or "",
            "has_community": bool(device.community), "has_auth_key": bool(device.v3_auth_key),
            "has_priv_key": bool(device.v3_priv_key)}


def parse_form(form, existing: SnmpDevice | None = None) -> dict:
    """The submitted form as the columns to store, secrets encrypted.
    Raises FormError saying what to fix."""
    name = (form.get("name") or "").strip()
    address = (form.get("address") or "").strip()
    if not name:
        raise FormError("Give the device a name.")
    if len(name) > 100 or "/" in name:
        raise FormError("A name is up to 100 characters, without a slash.")
    if not address:
        raise FormError("Give the device's address: an IP or a hostname.")
    try:
        port = int(form.get("port") or 161)
    except ValueError:
        raise FormError("The port is a number, usually 161.") from None
    if not 1 <= port <= 65535:
        raise FormError("The port is between 1 and 65535, usually 161.")
    version = form.get("version") or "2c"
    if version not in VERSIONS:
        raise FormError("Choose SNMP version 1, 2c or 3.")

    presets = [p for p in form.getlist("presets") if p in PRESET_IDS]
    interfaces = (form.get("interfaces") or "").strip()
    custom = (form.get("custom") or "").strip()
    try:
        numbers = parse_interfaces(interfaces)
        if "interfaces" in presets and not numbers:
            raise FormError("Interface traffic needs the interface numbers to read, e.g. 1, 2 or 1-4.")
        if not device_readings(presets, numbers, custom, version):
            raise FormError("Choose at least one preset or add a custom reading.")
    except ValueError as exc:
        raise FormError(str(exc)) from None

    def secret(field: str) -> str:
        typed = form.get(field) or ""
        if typed:
            return encrypt_secret(typed)
        return getattr(existing, field) if existing is not None else ""

    values = {"name": name, "address": address, "port": port, "version": version,
              "presets": presets, "interfaces": interfaces, "custom": custom,
              "host_id": (form.get("host_id") or "").strip() or None,
              "community": "", "v3_user": "", "v3_auth": "none", "v3_auth_key": "",
              "v3_priv": "none", "v3_priv_key": ""}
    if version == "3":
        user = (form.get("v3_user") or "").strip()
        auth = form.get("v3_auth") or "none"
        priv = form.get("v3_priv") or "none"
        if not user:
            raise FormError("SNMPv3 needs a user name.")
        if auth not in AUTH_PROTOCOLS or priv not in PRIV_PROTOCOLS:
            raise FormError("Choose an authentication and a privacy protocol from the lists.")
        if priv != "none" and auth == "none":
            raise FormError("Privacy needs authentication too: choose an authentication protocol.")
        values.update(v3_user=user, v3_auth=auth, v3_priv=priv,
                      v3_auth_key=secret("v3_auth_key") if auth != "none" else "",
                      v3_priv_key=secret("v3_priv_key") if priv != "none" else "")
        for field, proto in (("v3_auth_key", auth), ("v3_priv_key", priv)):
            if proto != "none" and not values[field]:
                raise FormError("SNMPv3 with " + ("authentication" if field == "v3_auth_key" else "privacy")
                                + " needs its key (at least 8 characters).")
            if proto != "none" and form.get(field) and len(form.get(field)) < 8:
                raise FormError("An SNMPv3 key is at least 8 characters.")
    else:
        values["community"] = secret("community") or encrypt_secret("public")
    return values


def form_credentials(form, existing: SnmpDevice | None = None) -> tuple[str, int, Credentials]:
    """Address, port and credentials from an unsaved form, for Test. A blank
    secret falls back to the stored one when testing an existing device."""
    def secret(field: str, default: str = "") -> str:
        typed = form.get(field) or ""
        if typed:
            return typed
        if existing is not None and getattr(existing, field):
            return decrypt_secret(getattr(existing, field))
        return default

    version = form.get("version") if form.get("version") in VERSIONS else "2c"
    try:
        port = int(form.get("port") or 161)
    except ValueError:
        port = 161
    creds = Credentials(version=version, community=secret("community", "public"),
                        user=(form.get("v3_user") or "").strip(), auth=form.get("v3_auth") or "none",
                        auth_key=secret("v3_auth_key"), priv=form.get("v3_priv") or "none",
                        priv_key=secret("v3_priv_key"))
    return (form.get("address") or "").strip(), port, creds


# A device not polled for this many intervals has stopped being polled
# (plugin turned off, scheduler stuck), whatever its last outcome said.
STALE_POLLS = 3


@dataclass
class DeviceState:
    reach: str                 # up | down | stale | pending
    says: str                  # one line, in words


def device_state(device: SnmpDevice, poll_seconds: int, now: datetime) -> DeviceState:
    """What the newest poll says about the device."""
    if device.last_polled_at is None:
        return DeviceState("pending", "Not polled yet.")
    if (now - device.last_polled_at).total_seconds() > STALE_POLLS * max(int(poll_seconds), 1):
        return DeviceState("stale", "Not polled recently: the SNMP plugin may be turned off.")
    if device.last_error:
        return DeviceState("down", device.last_error)
    return DeviceState("up", "Answering.")


class RateKeeper(CounterRates):
    """A counter reading's previous sample per (device, key), its rate scaled
    to the reading's unit (steward.core.counter_rates has the rules)."""

    def rate(self, device_id: str, reading: Reading, value: float, at: datetime) -> float | None:
        from .readings import COUNTER_UNITS
        per_sec = super().rate((device_id, reading.key), value, at)
        return None if per_sec is None else per_sec * COUNTER_UNITS[reading.unit][1]