napalm-device-types/README.md

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.

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.

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.

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) and whether it patches itself (unattended-upgrades, dnf-automatic). 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.

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.
• 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 (FirewallDriver):

class FirewallDriver(DeviceTypeDriver):
    # Abstract — every driver implements its own device communication.
    def get_firewall_rules(self) -> List[FirewallRuleDict]: raise NotImplementedError
    def apply_firewall_rule(self, rule: FirewallRuleDict, *, uuid: Optional[str] = None) -> Dict[str, Any]: raise NotImplementedError
    def commit_firewall_rules(self) -> Dict[str, Any]: raise NotImplementedError

    # 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.9 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

Mixins mixed into the classes above rather than used on their own: ConfigLifecycleMixin (config load/compare/commit/rollback), PingSweepMixin (subnet sweeps), and DhcpServerMixin (static DHCP reservations — mixed into FirewallDriver and ResidentialGatewayDriver, since both commonly run the DHCP server for their networks).

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=...)
        ...

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_users(self):
        # return List[UserDict]
        ...

    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.