Files
napalm-device-types/README.md
T
christianmanivong b4c4ec0b0b
CI / test (3.10) (push) Successful in 28s
CI / test (3.11) (push) Successful in 27s
CI / test (3.12) (push) Successful in 29s
CI / test (3.13) (push) Successful in 30s
CI / test (3.10) (pull_request) Successful in 29s
CI / test (3.11) (pull_request) Successful in 26s
CI / test (3.12) (pull_request) Successful in 35s
CI / test (3.13) (pull_request) Successful in 33s
feat: SnmpHealthProfile, the OIDs a driver declares; deprecate HealthMetricsMixin (4.12.0)
Drivers no longer collect over SNMP (docs/STRUCTURE.md). A driver whose
device publishes its health values under OIDs of its own declares them
as SNMP_HEALTH_PROFILE: per value either scalars asked in order or a
table folded by first, average or max, optionally with a pattern for
text answers; memory as total with used or free, or a percentage, in
KiB or bytes; and how to read the interfaces. The application polls
and computes.

HealthMetricsMixin.get_health_metrics now warns that it is deprecated;
it goes in 5.0.

Closes #32
2026-10-16 15:29:27 +02:00

25 KiB
Raw Blame History

napalm-device-types

Abstract intermediate device-type base classes for NAPALM drivers.

Instead of inheriting directly from napalm.base.NetworkDriver, a driver can inherit from one of the device-type classes here to gain a richer, type-specific contract:

# without napalm-device-types
class OpenWrtDriver(NetworkDriver):
    ...

# with napalm-device-types
from napalm_device_types import AccessPointDriver

class OpenWrtDriver(AccessPointDriver):
    ...

Why?

NAPALM's NetworkDriver defines a common interface for all network devices. In practice, devices fall into distinct categories with very different capabilities. A switch exposes spanning-tree and PoE data; a firewall exposes NAT tables and VPN tunnels; a NAS exposes disk pools and shares. Writing these methods directly in a concrete driver mixes concerns and makes drivers harder to discover and compare.

napalm-device-types sits in between: it adds one well-typed layer of abstract methods per device category, so every driver for the same category exposes the same interface.

Roles: what a device is

A device is often several things at once. A QNAP NAS runs VMs on a Linux userland; an OpenMediaVault box is a NAS built on Debian. So a driver inherits one role base per role its device fills, and the order it lists them in is the ranking:

class QnapQtsDriver(StorageDriver, HypervisorDriver, LinuxDriver):
    ...   # primary_role_of(...) == "storage"

roles_of(cls), role_keys_of(cls) and primary_role_of(cls) read that back. Nothing restates the ranking: there is no precedence table and no attribute to override.

A server's baseboard management controller (Supermicro BMC, Dell iDRAC, HPE iLO, Lenovo XCC) is a role of its own, ManagementControllerDriver (management_controller), reached over Redfish or IPMI (USES_SSH = False). It is not the server it manages: reboot_host restarts the BMC (Redfish Manager.Reset), while set_power_state (On, GracefulShutdown, ForceOff, GracefulRestart, ForceRestart, PowerCycle) acts on the server. get_hardware_inventory carries the server's serial number, system UUID and NIC MACs -- what ties the BMC to the host it manages. Its sensors come through NAPALM's get_environment, with cpu and memory empty, since a BMC does not know the host OS's load; it has no HealthMetricsMixin (SNMP) and no command channel.

Role bases declare; they never implement

A role base must not contain a single runtime method — not even a NotImplementedError placeholder. Its methods are declared under if TYPE_CHECKING:

class StorageDriver(DeviceTypeDriver):
    """Contract, not code."""
    ROLE: str = "storage"
    TYPE_LABEL: str = "Storage"

    if TYPE_CHECKING:                      # nothing exists at runtime
        def get_disks(self) -> List[PhysicalDiskDict]: ...

This is not a style preference. A placeholder on a base class is not neutral under multiple inheritance: it wins the MRO against a sibling base's working implementation and silently replaces it. Adding one stub to a base is therefore a breaking change for every driver that mixes that base with another. It happened three times here before the rule existed, and each time the fix was hand-written forwarding methods in the driver.

Two things follow, and both are improvements:

  • hasattr is truthful again. A method exists on a driver class exactly when that driver implemented it, which is how netOrk asks "can this driver list disks".
  • A method that was never implemented raises AttributeError, not NotImplementedError. Ask before calling.

The rule covers this package's methods, not NAPALM's. NetworkDriver still carries a runtime NotImplementedError stub for each of its getters (get_interfaces, get_environment, get_users, …), and every role base inherits it. So a NAPALM getter must come from the driver class itself or from a mixin listed before the first role base; a mixin listed after it loses the MRO to NAPALM's stub. For the same reason hasattr says nothing about NAPALM's getters, and the OS role's account listing is get_os_users, not get_users -- NAPALM's get_users has a different shape.

Function classes: what a device can do

Behaviour shared across roles lives in a function class, exactly once, and a role base is a thin bundle over them — PackageManagementMixin, HealthMetricsMixin, ServiceControlMixin, UpdateMixin, NatVpnMixin, MacAclMixin, FirewallRuleMixin, DhcpServerMixin, PingSweepMixin, ConfigLifecycleMixin, InterfaceFilterMixin. HostRebootMixin (reboot_host) is mixed into DeviceTypeDriver itself, since any device may be restartable; like the others it only declares.

SNMP is the application's (docs/STRUCTURE.md). HealthMetricsMixin is deprecated since 4.12 and goes in 5.0. A driver does not collect over SNMP; a driver whose device publishes its health values under OIDs of its own declares them as data:

class MySwitchDriver(SwitchDriver):
    SNMP_HEALTH_PROFILE: ClassVar[SnmpHealthProfile] = {
        "uptime": {"oids": ["1.3.6.1.2.1.1.3.0"]},
        "cpu": {"oids": ["<vendor cpu oid>", "<older firmware's>"]},
        "memory": {"total": {"oids": ["<oid>"]}, "free": {"oids": ["<oid>"]}, "unit": "KiB"},
    }

A value is oids (the first that answers wins) or a table folded by first, average or max, with an optional pattern for an agent that answers with text. What the profile does not name, the application reads from the standard UCD-MIB and IF-MIB.

KernelFactsMixin (get_kernel_facts) is the exception that is mixed in by a driver rather than by a role base: what a Linux kernel has built and loaded is read the same way everywhere, so the command and its parse are concrete here and a driver supplies only _run_kernel_facts_command. OSDriver does not carry it — a Windows host is an OS driver too, and hasattr(driver, "get_kernel_facts") has to stay truthful.

HostStatusMixin (get_host_status) is mixed in the same way: whether a Linux host needs a reboot to finish an update (/var/run/reboot-required, needs-restarting -r, or a newer kernel of the running flavour installed), whether it patches itself (unattended-upgrades, dnf-automatic) and when that last succeeded (apt's upgrade-stamp), and whether dpkg was left halfway (a journal in /var/lib/dpkg/updates or a half-configured package; unknown while a package manager is running), and how full /, /boot and /var are (df -Pk, each filesystem once, in bytes; how full is too full is the caller's decision). package_updates holds the shared apt and dnf parsers: apt's suites become an update's origin, a -security suite makes it a security update, and dnf's security advisories do the same.

ListeningSocketsMixin (get_listening_sockets) is mixed in the same way: every listening TCP and bound UDP socket from ss -lntup, with the systemd service or container behind it from /proc/<pid>/cgroup, in one round trip. A driver supplies _run_listening_sockets_command(command, privileged=); the command arrives as one sh -c argument, so a sudo -n prefix covers all of it. Without root ss names only the login user's processes, and the reading says so (attributed: false) instead of failing. A host without ss is read with netstat -lntup (OpenWrt's busybox, old net-tools); on OpenWrt the cgroup names the procd service (/services/<name>/<instance>). A host with neither raises ListeningSocketsUnavailable.

VirtualIpsMixin (get_virtual_ips) reads the addresses a host's HA configuration lets float between machines: vip-manager (5.x's --config YAML, 1.x's -ip=${VIP_IP} resolved through the unit's environment) and keepalived (virtual_ipaddress of every vrrp_instance, include followed four levels deep). It reports what is declared, not what the host holds right now -- that is in its interface addresses. The same files hold the etcd password and auth_pass, so an awk program filters on the host and prints allowlisted fields only; a secret never reaches the caller. Root first, then without it. A mechanism it could not read completely is None ("did not look"), a host without it []. The command is about 4 kB, so a driver sends it on an exec channel (_run_virtual_ips_command(command, privileged=)), not typed into a shell.

get_dhcp_ha (declared on DhcpServerMixin, optional) is what a DHCP server's own HA configuration says: whether HA is on, which peer this server is, each peer's name, role -- as the server words it, not a shared vocabulary -- and, where the server has one, the URL its HA listener answers on. It is configuration, not state: whether the pair is in hot standby right now is not in it. ConfigSyncMixin (get_config_sync) is where a host pushes its configuration (peer, None for no one) and for which services, by the names get_services reports; sync items the driver has no service name for stay visible under unmapped. A driver mixes it in -- no role promises it. Both getters raise when they could not read; neither answers "off" for "could not ask". What the facts mean for a pair (hot standby, which side to write to) is the caller's to work out.

A firewall's CARP addresses are a third mechanism, carp: a firewall driver reads them from its own API and reports them under the reading's carp key (vip_manager and keepalived then None). What a CARP state means is protocol knowledge and stays in that driver. The contract only carries the result: unlike the other two, a CARP entry says who holds the address (holds), because a firewall's interface list need not show it. holds None leaves the holder to the interfaces.

SystemIdentityMixin (get_system_identity) reads a host's SMBIOS identity -- system UUID, product serial and board serial from /sys/class/dmi/id -- which is what ties a server to its BMC (ManagementControllerDriver.get_hardware_inventory reports the same values from the other side). Vendor placeholders (0123456789 on Supermicro boards, To be filled by O.E.M., Default string, all-zero or the common 03000200-…-000700080009 UUID) read as empty: matching on them would join unrelated machines. is_placeholder_identity is the same filter for a BMC driver's side. product_uuid and product_serial are root's, so a driver supplies _run_system_identity_command(command, privileged=).

OpenSshServerMixin (harden_ssh_server, get_ssh_server_algorithms) takes algorithms out of what an OpenSSH server offers -- the ones ssh-audit flags. It reads the lists sshd really uses (sshd -T), takes out what was asked and writes the rest to /etc/ssh/sshd_config.d/00-netork-ssh-hardening.conf; only where sshd_config has no such Include is sshd_config itself rewritten, after a copy. It never removes the last algorithm of a class, nor -- given the managing client's lists -- the last one that client speaks. sshd -t must accept the result and sshd -T must show it; after the reload the caller's verify opens a new session, and if that fails the still-open one puts the old state back. A driver mixes it in and supplies run_command; a platform that regenerates sshd_config itself (an appliance with its own database) must not, or opts out with harden_ssh_server = None.

Update readers raise when they cannot read. get_available_updates returns an empty list only when nothing is pending; netOrk keeps "pending since" per package, and an empty list for "don't know" would reset it.

SystemdServicesMixin (get_services, manage_service) is mixed in the same way, by the drivers whose host runs systemd. Listing the services, checking a unit name and reading an action's exit status are the same on every such host, so they are concrete here, and a driver supplies only _run_service_command(command, *, privileged, timeout) — how a command reaches its host and how it gains root there. The listing is one round trip (list-unit-files plus one systemctl show over every loaded unit) instead of an is-enabled and a show per unit. A host without systemd raises SystemdUnavailable, a NotImplementedError, so a driver can fall back to another init system.

Access channels: why there is no container model here

CommandChannelMixin declares a driver's public channel to its host:

  • run_command(command, *, privileged=False, timeout=60, stdin=None) returns a CommandResult(stdout, stderr, exit_code);
  • open_stream(command, *, privileged=False) returns a ByteStream to the command's stdin and stdout.

The mixin sits in DeviceTypeDriver and only declares, so hasattr(driver, "run_command") is true exactly where a driver implements it. For SSH drivers, channel.run_on_transport and channel.open_stream_on_transport are the implementation over a paramiko exec channel: no PTY, stderr kept apart, a real exit code. How a command gains root stays with the driver.

ContainerEngineMixin is deliberately different from SystemdServicesMixin. The systemd mixin owns its command and its parse; this one owns neither.

  • container_engines() says which engines the host offers ([{"engine": "docker", "api": "docker-engine"}]).
  • open_container_engine(engine) returns a ContainerEngineConnection:
    • open_api() is a stream to the engine's API (docker system dial-stdio);
    • run_cli(args) and stream_cli(args) run the engine's CLI with arguments the caller chooses. With privileged=True the driver runs it the way run_command gains root, for an engine that refuses the login user.
  • The driver decides only how the engine is reached. Its hook _container_engine_binary(engine) returns, for QNAP, the Container Station path, and the caller never sees it.

What runs on the engine, and what to do with it, is netOrk's (NetOrk/netork#765): the container model, the Engine API requests, compose, updates. Above the connection everything is specific to the service, so this package abstracts the connection and nothing more.

A driver mixes ContainerEngineMixin in itself; OSDriver does not carry it. A Windows host is an OS driver too, and has no dial-stdio to offer over WinRM.

Never half-close early. ByteStream.write does not close anything; only close_write does. dial-stdio hands the daemon a half-close as "client gone", and the daemon then answers an unfinished request with HTTP 499.

A function class may use the template form — public method concrete, the device-specific part a _hook declared under if TYPE_CHECKING — when the base genuinely does work on the result: normalising, sorting, validating, or orchestrating several hooks. ConfigLifecycleMixin.compare_config over _get_running_config is the model. Where the base would only pass the call through, declare the method directly; two names for one pass-through is ceremony, not design.

NAPALM's own getters (get_facts, get_interfaces, ping, get_config) are never wrapped in a template — they belong to NAPALM, and code outside this repo relies on their contract.

Design principle: generic vs. device-specific logic

When adding behavior to a device-type base class, split it along one line: would this exact logic work unchanged for a different vendor's driver of the same device-type, if that driver only implemented the same abstract methods?

  • If yes, it's generic — implement it once as a concrete method on the device-type base class (here, in this repo).
  • If no — it talks to the device itself (a specific REST endpoint, a CLI command, a vendor-specific payload format) — it belongs in the concrete driver as the implementation of an abstract method the base class declares.

Concretely: matching/comparison/reconciliation algorithms, orchestration flows, and generic data shapes belong here. Only the actual device communication belongs in vendor/napalm-<name>.

Worked example — firewall rule diff/apply (FirewallRuleMixin, in FirewallDriver):

class FirewallRuleMixin:            # mixed into FirewallDriver
    # Declared only — every driver implements its own device communication.
    if TYPE_CHECKING:
        def get_firewall_rules(self) -> List[FirewallRuleDict]: ...
        def apply_firewall_rule(self, rule: FirewallRuleDict, *, uuid: Optional[str] = None) -> Dict[str, Any]: ...
        def commit_firewall_rules(self) -> Dict[str, Any]: ...

    # Concrete — the matching/comparison/orchestration algorithm is identical
    # for every firewall vendor, so it lives here once.
    def diff_firewall_rules(self, desired: List[FirewallRuleDict]) -> FirewallRuleDiffDict:
        ...  # matches self.get_firewall_rules() against `desired` by description

    def apply_firewall_ruleset(self, desired: List[FirewallRuleDict]):
        ...  # computes the diff, calls apply_firewall_rule() per change, commits

The same split applies to DhcpServerMixin: get_dhcp_reservations/ apply_dhcp_reservation/commit_dhcp_reservations are abstract (Kea REST on OPNsense, dnsmasq/odhcpd UCI on OpenWrt), while diff_dhcp_reservations and apply_dhcp_reservationset are concrete — matching by normalised MAC and the apply-then-commit orchestration are identical for every DHCP server.

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):

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. Same shape of separation, different axis — device-specific vs. generic here, detection vs. presentation there.

Installation

pip install napalm-device-types

Requires Python ≥ 3.10 (tested up to 3.13) and NAPALM ≥ 4.0.

Available base classes

Class Target devices Example implementations
AccessPointDriver Wireless access points OpenWrt, Ubiquiti UniFi, Cisco Meraki AP
SwitchDriver Ethernet switches Cisco IOS, Arista EOS, Juniper EX
FirewallDriver Firewalls & UTM appliances pfSense, Fortinet FortiOS, Cisco ASA
HypervisorDriver Hypervisors & virtualisation platforms Proxmox VE, VMware ESXi, KVM/libvirt
OSDriver General-purpose operating systems Linux, BSD, macOS
StorageDriver Storage appliances & NAS/SAN TrueNAS, Synology DSM, QNAP QTS
ResidentialGatewayDriver Router + firewall + AP in one box OpenWrt, FritzBox
PhoneDriver IP telephony endpoints Yealink
MediaDriver Speakers, streamers, media renderers Sonos, Chromecast
ManagementControllerDriver Server baseboard management controllers Supermicro BMC, Dell iDRAC, HPE iLO, Lenovo XCC

Mixins that come with the classes above: PingSweepMixin (subnet sweeps, in every driver through DeviceTypeDriver) and DhcpServerMixin (static DHCP reservations — mixed into FirewallDriver and ResidentialGatewayDriver, since both commonly run the DHCP server for their networks). ConfigLifecycleMixin (config load/compare/commit/rollback) is mixed in by a driver itself.

Fingerprints

DeviceTypeDriver carries what discovery matches a device against: PORT_SPECS, SNMP_OBJECT_ID_PREFIX, SNMP_FINGERPRINT, SSH_FINGERPRINT, HTTP_FINGERPRINT and OUI_PREFIXES. A FingerprintRule is a lowercase substring with a weight; mandatory disqualifies the driver when it is missing, negative subtracts.

Discovery compares drivers by score divided by the most a driver could score. Rules that can never match together, such as one vendor's landing page or another's, share a group: only the strongest rule of a group counts towards that maximum (rules_max_weight). Without it, a driver that recognises three vendors' pages would score a third as well as one that recognises a single page:

HTTP_FINGERPRINT = [
    FingerprintRule("aten international", 6.0, group="landing_page"),  # Supermicro
    FingerprintRule("idrac", 6.0, group="landing_page"),
    FingerprintRule("integrated lights-out", 6.0, group="landing_page"),  # HPE iLO
]

Usage

Access Point

from napalm_device_types import AccessPointDriver

class OpenWrtDriver(AccessPointDriver):

    def get_wireless_clients(self):
        # return List[WirelessClientDict]
        ...

    def get_ssids(self):
        # return Dict[str, SSIDDict]
        ...

Switch

from napalm_device_types import SwitchDriver

class CiscoIOSDriver(SwitchDriver):

    def get_spanning_tree(self):
        # return Dict[str, SpanningTreeDict]
        ...

    def get_poe_status(self):
        # return PoESummaryDict
        ...

Firewall

from napalm_device_types import FirewallDriver

class PfSenseDriver(FirewallDriver):

    def get_nat_translations(self):
        # return List[NATTranslationDict]
        ...

    def get_vpn_tunnels(self):
        # return Dict[str, VPNTunnelDict]
        ...

    def get_port_forwards(self):
        # return List[PortForwardDict] — forwards from the WAN only, never a
        # redirect between internal networks (shared with home gateways)
        ...

Hypervisor

from napalm_device_types import HypervisorDriver

class ProxmoxDriver(HypervisorDriver):

    def get_vms(self):
        # return List[VMDict]
        ...

    def create_vm_snapshot(self, name, snapshot, description="", include_memory=False):
        ...

    def get_vm_cpu_types(self):
        # optional — return List[VMCpuTypeDict]: the CPU models a new VM may get
        # on this node, each with its cpuinfo flags and whether the node can run
        # it; the name goes to create_vm_from_cloud_init(cpu_type=...)
        ...

A hypervisor that provisions VMs from cloud images declares, per guest OS, the agent cloud-init installs so the hypervisor can read the new VM's IP: GUEST_AGENTS = {guest_os: (packages, runcmd)}. Its keys are the guests the driver can provision, and create_vm_from_cloud_init(guest_os=...) refuses any other. A QEMU-based driver takes provisioning.QEMU_GUEST_AGENTS (Linux, FreeBSD, OpenBSD). provisioning.network_config() writes the network-config v2 every cloud-init flavour reads (FreeBSD's nuageinit reads no other), and provisioning.split_compression() says how a packed image is unpacked; both are the same for every hypervisor.

OS / Linux

from napalm_device_types import OSDriver

class LinuxDriver(OSDriver):

    def get_packages(self):
        # return List[PackageDict]
        ...

    def get_services(self):
        # return List[ServiceDict]
        ...

    def get_os_users(self):
        # return List[UserDict] -- not NAPALM's get_users, whose shape differs
        ...

    def get_processes(self):
        # return List[ProcessDict]
        ...

    def get_cron_jobs(self):
        # return List[CronJobDict]
        ...

Storage / NAS

from napalm_device_types import StorageDriver

class TrueNASDriver(StorageDriver):

    def get_disks(self):
        # return List[PhysicalDiskDict]
        ...

    def get_shares(self):
        # return Dict[str, NASShareDict]
        ...

Return types

All return types are TypedDict classes defined in napalm_device_types.models. Import them directly for type annotations in your driver:

from napalm_device_types.models import (
    PhysicalDiskDict,
    DiskPoolDict,
    NASShareDict,
    VolumeSnapshotDict,
)

Development

git clone https://github.com/chrismanivong/napalm-device-types.git
cd napalm-device-types
pip install -e ".[dev]"

Run type-checking:

mypy napalm_device_types

Contributing

  1. Fork the repository
  2. Create a feature branch (git checkout -b feature/router-driver)
  3. Implement your changes
  4. Open a Pull Request

License

Apache-2.0 – see LICENSE for details.