diff --git a/README.md b/README.md index c05d522..91f82e2 100644 --- a/README.md +++ b/README.md @@ -60,6 +60,25 @@ A new driver (FortiGate, pfSense, …) gets `diff_firewall_rules`/ `apply_firewall_ruleset` for free the moment it implements the three abstract methods — it never needs to reimplement the reconciliation logic itself. +**Second worked example — ping sweeps** (`PingSweepMixin`, mixed into +`DeviceTypeDriver`, so *every* device-type driver has it): + +```python +class PingSweepMixin: + # Concrete — the loop, the reply parsing, the target cap and the progress + # reporting are the same for every device that can ping at all. + def ping_sweep(self, destinations, *, count=1, timeout=1, …) -> PingSweepResultDict: + ... # calls NAPALM's standard ping() once per destination +``` + +A driver becomes a usable sweep source the moment it implements NAPALM's +`ping()` — nothing else is required, and `driver_supports_ping(cls)` reports +whether it did (introspection, not a hand-maintained list). A driver whose +device offers something genuinely faster overrides `ping_sweep` and keeps the +return shape: `napalm-opnsense` starts a batch of ping jobs over the +diagnostics API, waits once for all of them, and reads every result with a +single request — a per-host loop would be unusable there. + This mirrors a similar split already documented on the consumer side, in NetOrk's `docs/ARCHITECTURE.md` ("Device Warnings — Trennung von Erkennung und Präsentation"): drivers return raw signals, the higher layer gives them meaning. diff --git a/napalm_device_types/__init__.py b/napalm_device_types/__init__.py index c717fb7..9e148f1 100644 --- a/napalm_device_types/__init__.py +++ b/napalm_device_types/__init__.py @@ -36,6 +36,7 @@ from napalm_device_types.config_lifecycle import ConfigLifecycleMixin from napalm_device_types.firewall import FirewallDriver from napalm_device_types.hypervisor import HypervisorDriver from napalm_device_types.os import OSDriver +from napalm_device_types.ping_sweep import PingSweepMixin, driver_supports_ping from napalm_device_types.residential_gateway import ResidentialGatewayDriver from napalm_device_types.storage import StorageDriver from napalm_device_types.switch import SwitchDriver @@ -48,8 +49,10 @@ __all__ = [ "FirewallDriver", "HypervisorDriver", "OSDriver", + "PingSweepMixin", "PortSpec", "ResidentialGatewayDriver", "StorageDriver", "SwitchDriver", + "driver_supports_ping", ] diff --git a/napalm_device_types/base.py b/napalm_device_types/base.py index e3fb2c2..60297f0 100644 --- a/napalm_device_types/base.py +++ b/napalm_device_types/base.py @@ -12,6 +12,8 @@ from typing import NamedTuple from napalm.base import NetworkDriver +from napalm_device_types.ping_sweep import PingSweepMixin + class FingerprintRule(NamedTuple): """Single pattern-matching rule for device fingerprinting. @@ -45,13 +47,15 @@ class PortSpec(NamedTuple): mandatory: bool = False -class DeviceTypeDriver(NetworkDriver): +class DeviceTypeDriver(PingSweepMixin, NetworkDriver): """Common base for all netOrk device-type drivers. Sits between napalm.base.NetworkDriver and the type-specific abstract classes (FirewallDriver, SwitchDriver, …). Adds the fingerprinting - interface consumed by the discovery subsystem; does not implement any - NAPALM abstract methods. + interface consumed by the discovery subsystem plus the generic + ``ping_sweep()`` from :class:`~napalm_device_types.ping_sweep.PingSweepMixin` + (usable by every driver that implements NAPALM's ``ping()``); does not + implement any NAPALM abstract methods. Override these class attributes in each concrete driver: diff --git a/napalm_device_types/models.py b/napalm_device_types/models.py index 6097a38..bb2599a 100644 --- a/napalm_device_types/models.py +++ b/napalm_device_types/models.py @@ -801,3 +801,26 @@ class StorageTargetDict(TypedDict): type: str # Backend type: "dir", "lvmthin", "zfspool", "nfs", etc. total_gb: float # Total capacity in gigabytes available_gb: float # Free capacity in gigabytes + + +# --------------------------------------------------------------------------- +# Ping sweep (shared across device types) +# --------------------------------------------------------------------------- + + +class PingSweepEntryDict(TypedDict): + """Outcome of a single ``ping`` inside a sweep (see ``PingSweepMixin``).""" + + ip: str # destination that was probed + alive: bool # True if at least one probe was answered + rtt_ms: Optional[float] # average round-trip time in ms; None if unreachable + error: NotRequired[str] # driver/transport error for this destination + + +class PingSweepResultDict(TypedDict): + """Result of a ``ping_sweep()`` call.""" + + entries: List[PingSweepEntryDict] # one entry per probed destination, in input order + scanned: int # destinations actually probed + alive_count: int # entries with alive=True + truncated: bool # True if targets were dropped at the sweep cap diff --git a/napalm_device_types/ping_sweep.py b/napalm_device_types/ping_sweep.py new file mode 100644 index 0000000..10e19b3 --- /dev/null +++ b/napalm_device_types/ping_sweep.py @@ -0,0 +1,172 @@ +# -*- coding: utf-8 -*- +"""Generic ICMP sweep on top of the NAPALM-standard ``ping()``. + +Sweeping a range of addresses is orchestration, not device mechanics: the +only vendor-specific part is how a single ``ping`` is executed, and NAPALM +already standardises that. So the loop, the reply parsing, the target cap +and the progress reporting live here once, and a concrete driver only has +to implement ``ping()`` to become a usable sweep source. + +A driver whose device offers a *faster* sweep mechanism (a batch API, a +single shell command that pings many hosts in parallel, an ARP-assisted +scan) overrides :meth:`PingSweepMixin.ping_sweep` and keeps the same return +shape — see ``napalm-opnsense`` for an example. + +The generic implementation is deliberately **sequential**: a NAPALM +connection is a single session (SSH channel, HTTP client) and is not safe to +drive from several threads at once. Callers that need many addresses covered +quickly should either use a driver with its own parallel override or cap the +target list (see ``PING_SWEEP_MAX_TARGETS``). +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Callable, ClassVar, Dict, Iterable, List, Optional + +from napalm.base import NetworkDriver + +from napalm_device_types.models import PingSweepEntryDict, PingSweepResultDict + + +def driver_supports_ping(driver_cls: type) -> bool: + """Whether *driver_cls* can actually execute ``ping()``. + + True when the class provides its own ``ping`` implementation instead of + inheriting NAPALM's ``NotImplementedError`` stub. A driver that inherits + a working ``ping`` but cannot use it (unsupported firmware, disabled + service) opts out by setting ``SUPPORTS_PING = False``. + """ + if getattr(driver_cls, "SUPPORTS_PING", None) is False: + return False + ping_impl = getattr(driver_cls, "ping", None) + if ping_impl is None: + return False + return ping_impl is not getattr(NetworkDriver, "ping", None) + + +class PingSweepMixin: + """Adds :meth:`ping_sweep` to any driver that implements ``ping()``. + + Mixed into :class:`~napalm_device_types.base.DeviceTypeDriver`, so every + device-type driver inherits it; drivers without a ``ping()`` of their own + simply report ``supports_ping() is False`` and raise from ``ping_sweep``. + """ + + #: Upper bound on destinations probed in one sweep. The sequential + #: default implementation costs roughly ``timeout`` seconds per silent + #: host, so an uncapped /24 would keep a device session busy for minutes. + #: Drivers with a parallel mechanism raise this. + PING_SWEEP_MAX_TARGETS: ClassVar[int] = 256 + + #: ``False`` opts a driver out of ping sweeps even though it implements + #: ``ping()``. ``None`` (the default) means "decide by introspection". + SUPPORTS_PING: ClassVar[Optional[bool]] = None + + if TYPE_CHECKING: # pragma: no cover - declared for type checkers only + + def ping( + self, + destination: str, + source: str = "", + ttl: int = 255, + timeout: int = 2, + size: int = 100, + count: int = 5, + vrf: str = "", + ) -> Dict[str, Any]: ... + + @classmethod + def supports_ping(cls) -> bool: + """Whether this driver class can be used as a ping-sweep source.""" + return driver_supports_ping(cls) + + def ping_sweep( + self, + destinations: Iterable[str], + *, + count: int = 1, + timeout: int = 1, + max_targets: Optional[int] = None, + on_progress: Optional[Callable[[int, int], None]] = None, + should_stop: Optional[Callable[[], bool]] = None, + ) -> PingSweepResultDict: + """Ping every address in *destinations* and report who answered. + + :param destinations: IP addresses / hostnames to probe, in order. + :param count: probes per destination — 1 is enough for liveness. + :param timeout: seconds to wait for a reply per destination. + :param max_targets: cap for this call; defaults to + ``PING_SWEEP_MAX_TARGETS``. Excess destinations are dropped and + ``truncated`` is set in the result. + :param on_progress: called as ``(done, total)`` after each probe. + :param should_stop: polled before each probe; returning True ends the + sweep early (cancelled job, shutting-down worker). + :raises NotImplementedError: if the driver has no ``ping()``. + """ + if not self.supports_ping(): + raise NotImplementedError( + f"{type(self).__name__} does not implement ping(); cannot run a ping sweep" + ) + + limit = self.PING_SWEEP_MAX_TARGETS if max_targets is None else max_targets + targets = list(destinations) + truncated = len(targets) > limit + if truncated: + targets = targets[:limit] + + total = len(targets) + entries: List[PingSweepEntryDict] = [] + for done, destination in enumerate(targets, start=1): + if should_stop is not None and should_stop(): + break + entries.append(self._ping_once(destination, count=count, timeout=timeout)) + if on_progress is not None: + on_progress(done, total) + + return { + "entries": entries, + "scanned": len(entries), + "alive_count": sum(1 for entry in entries if entry["alive"]), + "truncated": truncated, + } + + # ── internals ──────────────────────────────────────────────────────────── + + def _ping_once(self, destination: str, *, count: int, timeout: int) -> PingSweepEntryDict: + """One probe, never raising — a dead session must not abort the sweep.""" + try: + reply = self.ping(destination, count=count, timeout=timeout) + except Exception as exc: # noqa: BLE001 - any driver error is just "no answer" + return {"ip": destination, "alive": False, "rtt_ms": None, "error": str(exc)} + return self._parse_ping_reply(destination, reply) + + @staticmethod + def _parse_ping_reply(destination: str, reply: Any) -> PingSweepEntryDict: + """Map a NAPALM ``ping()`` reply onto a sweep entry.""" + if not isinstance(reply, dict) or "success" not in reply: + error = "malformed ping reply" + if isinstance(reply, dict) and reply.get("error"): + error = str(reply["error"]) + return {"ip": destination, "alive": False, "rtt_ms": None, "error": error} + + success = reply.get("success") or {} + probes_sent = _as_int(success.get("probes_sent")) + packet_loss = _as_int(success.get("packet_loss"), default=probes_sent) + alive = bool(success.get("results")) or probes_sent > packet_loss + if not alive: + return {"ip": destination, "alive": False, "rtt_ms": None} + return {"ip": destination, "alive": True, "rtt_ms": _as_float(success.get("rtt_avg"))} + + +def _as_int(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _as_float(value: Any) -> Optional[float]: + try: + return float(value) + except (TypeError, ValueError): + return None diff --git a/tests/test_ping_sweep.py b/tests/test_ping_sweep.py new file mode 100644 index 0000000..db2e4d9 --- /dev/null +++ b/tests/test_ping_sweep.py @@ -0,0 +1,221 @@ +"""Tests for the generic ping sweep (PingSweepMixin + driver_supports_ping).""" + +import pytest +from napalm.base import NetworkDriver + +from napalm_device_types import DeviceTypeDriver, PingSweepMixin, driver_supports_ping + + +# ── Fakes ───────────────────────────────────────────────────────────────────── + + +def _napalm_ok(rtt: float, probes: int = 1) -> dict: + """A NAPALM-format ping() reply for a reachable destination.""" + return { + "success": { + "probes_sent": probes, + "packet_loss": 0, + "rtt_min": rtt, + "rtt_avg": rtt, + "rtt_max": rtt, + "rtt_stddev": 0.0, + "results": [{"ip_address": "10.0.0.1", "rtt": rtt}], + } + } + + +def _napalm_lost(probes: int = 1) -> dict: + """A NAPALM-format ping() reply where every probe was lost.""" + return { + "success": { + "probes_sent": probes, + "packet_loss": probes, + "rtt_min": 0.0, + "rtt_avg": 0.0, + "rtt_max": 0.0, + "rtt_stddev": 0.0, + "results": [], + } + } + + +class FakePingDriver(PingSweepMixin): + """Minimal driver exposing ping() — stands in for a real vendor driver.""" + + def __init__(self, replies=None, raises=None): + self.replies = replies or {} + self.raises = raises or {} + self.calls = [] + + def ping(self, destination, source="", ttl=255, timeout=2, size=100, count=5, vrf=""): + self.calls.append({"destination": destination, "timeout": timeout, "count": count}) + if destination in self.raises: + raise self.raises[destination] + return self.replies.get(destination, _napalm_lost()) + + +class NoPingDriver(PingSweepMixin): + """Driver without its own ping() — inherits NAPALM's NotImplementedError stub.""" + + ping = NetworkDriver.ping + + +class OptedOutDriver(FakePingDriver): + """Driver that implements ping() but declares it unusable for sweeps.""" + + SUPPORTS_PING = False + + +# ── driver_supports_ping ────────────────────────────────────────────────────── + + +def test_driver_supports_ping_false_for_unoverridden_ping(): + assert driver_supports_ping(NoPingDriver) is False + + +def test_driver_supports_ping_false_for_base_network_driver(): + assert driver_supports_ping(NetworkDriver) is False + + +def test_driver_supports_ping_true_when_overridden(): + assert driver_supports_ping(FakePingDriver) is True + + +def test_driver_supports_ping_honours_explicit_opt_out(): + assert driver_supports_ping(OptedOutDriver) is False + + +def test_driver_supports_ping_false_for_class_without_ping(): + class Bare: + pass + + assert driver_supports_ping(Bare) is False + + +def test_device_type_driver_exposes_supports_ping_classmethod(): + assert DeviceTypeDriver.supports_ping() is False + assert FakePingDriver.supports_ping() is True + + +# ── ping_sweep ──────────────────────────────────────────────────────────────── + + +def test_ping_sweep_marks_reachable_and_unreachable_hosts(): + driver = FakePingDriver(replies={"10.0.0.1": _napalm_ok(1.5)}) + + result = driver.ping_sweep(["10.0.0.1", "10.0.0.2"]) + + assert result["scanned"] == 2 + assert result["alive_count"] == 1 + assert result["truncated"] is False + assert result["entries"] == [ + {"ip": "10.0.0.1", "alive": True, "rtt_ms": 1.5}, + {"ip": "10.0.0.2", "alive": False, "rtt_ms": None}, + ] + + +def test_ping_sweep_uses_single_fast_probe_by_default(): + driver = FakePingDriver() + + driver.ping_sweep(["10.0.0.1"]) + + assert driver.calls == [{"destination": "10.0.0.1", "timeout": 1, "count": 1}] + + +def test_ping_sweep_forwards_count_and_timeout(): + driver = FakePingDriver() + + driver.ping_sweep(["10.0.0.1"], count=3, timeout=5) + + assert driver.calls == [{"destination": "10.0.0.1", "timeout": 5, "count": 3}] + + +def test_ping_sweep_treats_error_reply_as_unreachable(): + driver = FakePingDriver(replies={"10.0.0.9": {"error": "unknown host"}}) + + result = driver.ping_sweep(["10.0.0.9"]) + + assert result["entries"][0]["alive"] is False + assert result["entries"][0]["error"] == "unknown host" + assert result["alive_count"] == 0 + + +def test_ping_sweep_records_exception_and_continues(): + driver = FakePingDriver( + replies={"10.0.0.2": _napalm_ok(2.0)}, + raises={"10.0.0.1": RuntimeError("session closed")}, + ) + + result = driver.ping_sweep(["10.0.0.1", "10.0.0.2"]) + + assert result["entries"][0] == { + "ip": "10.0.0.1", + "alive": False, + "rtt_ms": None, + "error": "session closed", + } + assert result["entries"][1]["alive"] is True + assert result["scanned"] == 2 + + +def test_ping_sweep_truncates_at_max_targets(): + driver = FakePingDriver() + + result = driver.ping_sweep([f"10.0.0.{i}" for i in range(1, 11)], max_targets=4) + + assert result["scanned"] == 4 + assert result["truncated"] is True + assert len(driver.calls) == 4 + + +def test_ping_sweep_respects_class_level_max_targets(): + class SmallSweepDriver(FakePingDriver): + PING_SWEEP_MAX_TARGETS = 2 + + driver = SmallSweepDriver() + + result = driver.ping_sweep(["10.0.0.1", "10.0.0.2", "10.0.0.3"]) + + assert result["scanned"] == 2 + assert result["truncated"] is True + + +def test_ping_sweep_reports_progress_per_destination(): + driver = FakePingDriver(replies={"10.0.0.1": _napalm_ok(1.0)}) + seen = [] + + driver.ping_sweep( + ["10.0.0.1", "10.0.0.2"], + on_progress=lambda done, total: seen.append((done, total)), + ) + + assert seen == [(1, 2), (2, 2)] + + +def test_ping_sweep_on_empty_destination_list(): + driver = FakePingDriver() + + result = driver.ping_sweep([]) + + assert result == {"entries": [], "scanned": 0, "alive_count": 0, "truncated": False} + + +def test_ping_sweep_raises_when_driver_cannot_ping(): + driver = NoPingDriver() + + with pytest.raises(NotImplementedError): + driver.ping_sweep(["10.0.0.1"]) + + +def test_ping_sweep_stops_when_stop_requested(): + driver = FakePingDriver() + calls = {"n": 0} + + def _should_stop(): + calls["n"] += 1 + return calls["n"] > 1 + + result = driver.ping_sweep(["10.0.0.1", "10.0.0.2", "10.0.0.3"], should_stop=_should_stop) + + assert result["scanned"] == 1 + assert len(driver.calls) == 1