NAPALM/napalm-device-types
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial release (v0.1.0)

Browse Source 
Add abstract device-type base classes for NAPALM drivers:
- AccessPointDriver  (wireless APs)
- SwitchDriver       (Ethernet switches)
- FirewallDriver     (firewalls / UTM)
- HypervisorDriver   (Proxmox VE, ESXi, KVM, Hyper-V)
- StorageDriver      (NAS/SAN appliances)

All return types modelled as TypedDicts in napalm_device_types.models.
This commit is contained in:
christianmanivong
2026-05-11 21:31:52 +02:00
commit b03e4355c9
11 changed files with 3182 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
napalm_device_types/__init__.py
+37
View File
@@ -0,0 +1,37 @@
"""
napalm-device-types
~~~~~~~~~~~~~~~~~~~
Abstract 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 defined here to gain
type-specific abstract methods and a clearer contract::
from napalm_device_types import AccessPointDriver
class OpenWrtDriver(AccessPointDriver):
...
Available base classes:
* :class:`~napalm_device_types.access_point.AccessPointDriver`
* :class:`~napalm_device_types.switch.SwitchDriver`
* :class:`~napalm_device_types.firewall.FirewallDriver`
* :class:`~napalm_device_types.hypervisor.HypervisorDriver`
* :class:`~napalm_device_types.storage.StorageDriver`
"""
from napalm_device_types.access_point import AccessPointDriver
from napalm_device_types.firewall import FirewallDriver
from napalm_device_types.hypervisor import HypervisorDriver
from napalm_device_types.storage import StorageDriver
from napalm_device_types.switch import SwitchDriver
__all__ = [
"AccessPointDriver",
"FirewallDriver",
"HypervisorDriver",
"StorageDriver",
"SwitchDriver",
]
napalm_device_types/access_point.py
+530
View File
@@ -0,0 +1,530 @@
"""
Abstract base class for wireless access point drivers.
Usage::
from napalm_device_types import AccessPointDriver
class OpenWrtDriver(AccessPointDriver):
def get_wireless_clients(self):
...
"""
from typing import Any, Dict, List
from napalm.base import NetworkDriver
from napalm_device_types.models import (
Dot1XConfigDict,
FastTransitionConfigDict,
MACACLDict,
MeshConfigDict,
MeshPeerDict,
PackageDict,
RadioStatusDict,
SSIDBridgeDict,
SSIDDict,
WirelessClientDict,
WirelessConfigDict,
)
class AccessPointDriver(NetworkDriver):
"""
Abstract intermediate driver for wireless access points.
Inherits all standard NAPALM NetworkDriver methods and adds
access-point-specific operations that concrete drivers must implement.
"""
def get_wireless_clients(self) -> List[WirelessClientDict]:
"""
Returns a list of wireless clients currently associated with this
access point.
Each entry contains:
* mac (string) - client MAC address
* ssid (string) - SSID the client is connected to
* radio (string) - radio identifier (e.g. ``"radio0"``, ``"5GHz"``)
* signal (int) - received signal strength in dBm
* noise (int) - noise floor in dBm
* tx_rate (float) - TX bitrate in Mbit/s
* rx_rate (float) - RX bitrate in Mbit/s
* uptime (int) - association duration in seconds
Example::
[
{
"mac": "AA:BB:CC:DD:EE:FF",
"ssid": "MyNetwork",
"radio": "radio1",
"signal": -65,
"noise": -95,
"tx_rate": 300.0,
"rx_rate": 144.0,
"uptime": 3600,
}
]
"""
raise NotImplementedError
def get_ssids(self) -> Dict[str, SSIDDict]:
"""
Returns the configured SSIDs (VAPs) on this access point.
Keys are SSID names. Each value contains:
* enabled (bool) - whether the SSID is currently broadcasting
* radio (string) - radio the SSID is bound to
* bssid (string) - BSSID (MAC) of the VAP
* encryption (string) - e.g. ``"WPA2-PSK"``, ``"WPA3-SAE"``, ``"open"``
* hidden (bool) - whether the SSID is hidden
* clients (int) - number of currently associated clients
Example::
{
"MyNetwork": {
"enabled": True,
"radio": "radio1",
"bssid": "AA:BB:CC:DD:EE:F0",
"encryption": "WPA2-PSK",
"hidden": False,
"clients": 3,
},
"GuestNet": {
"enabled": True,
"radio": "radio0",
"bssid": "AA:BB:CC:DD:EE:F1",
"encryption": "WPA2-PSK",
"hidden": False,
"clients": 1,
},
}
"""
raise NotImplementedError
def get_radio_status(self) -> Dict[str, RadioStatusDict]:
"""
Returns the status of each radio interface.
Keys are radio identifiers (e.g. ``"radio0"``, ``"radio1"``).
Each value contains:
* enabled (bool) - whether the radio is active
* band (string) - frequency band, e.g. ``"2.4GHz"``, ``"5GHz"``, ``"6GHz"``
* channel (int) - operating channel number
* channel_width (int) - channel width in MHz (e.g. 20, 40, 80, 160)
* tx_power (int) - transmit power in dBm
* frequency (float) - center frequency in MHz
Example::
{
"radio0": {
"enabled": True,
"band": "2.4GHz",
"channel": 6,
"channel_width": 20,
"tx_power": 20,
"frequency": 2437.0,
},
"radio1": {
"enabled": True,
"band": "5GHz",
"channel": 36,
"channel_width": 80,
"tx_power": 23,
"frequency": 5180.0,
},
}
"""
raise NotImplementedError
def get_wireless_config(self) -> WirelessConfigDict:
"""
Returns global wireless configuration parameters that apply across
all radios and SSIDs.
* country_code (string) - ISO 3166-1 alpha-2 country code (e.g. ``"DE"``)
* regulatory_domain (string) - regulatory domain string (e.g. ``"ETSI"``)
* beacon_interval (int) - beacon interval in TUs (default 100)
* dtim_period (int) - DTIM period (default 2)
* rts_threshold (int) - RTS/CTS threshold in bytes (2347 = disabled)
* fragmentation_threshold (int) - fragmentation threshold in bytes
* short_preamble (bool) - whether short preamble is enabled
* wmm_enabled (bool) - whether WMM/QoS is enabled
Example::
{
"country_code": "DE",
"regulatory_domain": "ETSI",
"beacon_interval": 100,
"dtim_period": 2,
"rts_threshold": 2347,
"fragmentation_threshold": 2346,
"short_preamble": True,
"wmm_enabled": True,
}
"""
raise NotImplementedError
def get_fast_transition_config(self) -> Dict[str, FastTransitionConfigDict]:
"""
Returns the 802.11r Fast BSS Transition (FT) configuration per SSID.
Keys are SSID names. Each value contains:
* enabled (bool) - whether FT is active on this SSID
* ssid (string) - SSID name (repeated for convenience)
* mobility_domain (string) - 4-hex-digit Mobility Domain ID (MDID)
* reassociation_deadline (int) - FT reassociation deadline in TUs
* r0_key_lifetime (int) - PMK-R0 key lifetime in minutes
* r1_key_holder (string) - R1 Key Holder identifier (MAC-like string)
* pmk_r1_push (bool) - whether PMK-R1 is proactively pushed to neighbours
* over_ds (bool) - whether FT over DS (instead of FT over air) is used
Example::
{
"CorpWiFi": {
"enabled": True,
"ssid": "CorpWiFi",
"mobility_domain": "a1b2",
"reassociation_deadline": 1000,
"r0_key_lifetime": 10000,
"r1_key_holder": "00:11:22:33:44:55",
"pmk_r1_push": True,
"over_ds": False,
}
}
"""
raise NotImplementedError
def get_mesh_config(self) -> Dict[str, MeshConfigDict]:
"""
Returns the 802.11s mesh configuration per mesh interface.
Keys are mesh interface names (e.g. ``"mesh0"``). Each value contains:
* enabled (bool) - whether the mesh interface is active
* radio (string) - underlying radio (e.g. ``"radio0"``)
* mesh_id (string) - 802.11s Mesh ID (analogous to SSID)
* path_metric (string) - path selection metric, e.g. ``"airtime"`` or ``"hopcount"``
* gate_announcements (bool) - whether gate announcements (GANN) are sent
* is_gate (bool) - whether this node acts as a mesh gate to the DS
* encryption (string) - e.g. ``"SAE"``, ``"open"``
Example::
{
"mesh0": {
"enabled": True,
"radio": "radio1",
"mesh_id": "office-mesh",
"path_metric": "airtime",
"gate_announcements": True,
"is_gate": True,
"encryption": "SAE",
}
}
"""
raise NotImplementedError
def get_mesh_peers(self) -> List[MeshPeerDict]:
"""
Returns a list of currently active 802.11s mesh peers.
Each entry contains:
* mac (string) - peer MAC address
* radio (string) - radio on which the peering was established
* signal (int) - received signal strength in dBm
* tx_rate (float) - TX bitrate to peer in Mbit/s
* rx_rate (float) - RX bitrate from peer in Mbit/s
* uptime (int) - peering duration in seconds
* hop_count (int) - number of hops to the mesh gate (0 = this node is the gate)
Example::
[
{
"mac": "AA:BB:CC:DD:EE:01",
"radio": "radio1",
"signal": -58,
"tx_rate": 300.0,
"rx_rate": 270.0,
"uptime": 7200,
"hop_count": 1,
}
]
"""
raise NotImplementedError
def get_ssid_bridge_config(self) -> Dict[str, SSIDBridgeDict]:
"""
Returns the Layer-2 bridging configuration for each SSID, i.e. which
bridge interface and VLAN each SSID is mapped to.
Keys are SSID names. Each value contains:
* ssid (string) - SSID name (repeated for convenience)
* bridge (string) - bridge interface the VAP is attached to (e.g. ``"br-lan"``, ``"br-guest"``)
* vlan_id (int) - 802.1Q VLAN ID (0 = untagged / no VLAN separation)
* tagged (bool) - whether traffic is 802.1Q-tagged on the uplink port
* client_isolation (bool) - whether clients on this SSID are isolated from each other
Example::
{
"CorpWiFi": {
"ssid": "CorpWiFi",
"bridge": "br-corp",
"vlan_id": 10,
"tagged": True,
"client_isolation": False,
},
"GuestNet": {
"ssid": "GuestNet",
"bridge": "br-guest",
"vlan_id": 20,
"tagged": True,
"client_isolation": True,
},
"IoT": {
"ssid": "IoT",
"bridge": "br-iot",
"vlan_id": 30,
"tagged": True,
"client_isolation": True,
},
}
"""
raise NotImplementedError
def get_mac_acl(self) -> Dict[str, MACACLDict]:
"""
Returns the MAC-address-based access control lists configured per SSID.
Keys are SSID names. Each value contains:
* ssid (string) - SSID name (repeated for convenience)
* policy (string) - ACL mode:
* ``"allow"`` – whitelist: only listed MACs may associate
* ``"deny"`` – blacklist: listed MACs are blocked
* ``"disabled"`` – no MAC filtering active
* entries (list) - ACL entries, each with:
* mac (string) - MAC address (normalised, colon-separated)
* action (string) - ``"allow"`` or ``"deny"``
* description (string) - optional human-readable label
Example::
{
"CorpWiFi": {
"name": "CorpWiFi",
"policy": "allow",
"entries": [
{"mac": "AA:BB:CC:DD:EE:01", "action": "allow", "description": "CEO-Laptop"},
{"mac": "AA:BB:CC:DD:EE:02", "action": "allow", "description": "CFO-Laptop"},
],
},
"GuestNet": {
"name": "GuestNet",
"policy": "deny",
"entries": [
{"mac": "DE:AD:BE:EF:00:01", "action": "deny", "description": "blocked device"},
],
},
}
"""
raise NotImplementedError
def get_dot1x_config(self) -> Dict[str, Dot1XConfigDict]:
"""
Returns the 802.1X / WPA-Enterprise (RADIUS) configuration per SSID.
Keys are SSID names. Each value contains:
* enabled (bool) - whether 802.1X authentication is active on this SSID
* ssid (string) - SSID name (repeated for convenience)
* auth_server (dict) - RADIUS authentication server:
* host (string) - IP or FQDN of the RADIUS server
* port (int) - UDP port (default 1812)
* timeout (int) - request timeout in seconds
* retries (int) - number of retransmissions
* acct_server (dict or None) - RADIUS accounting server (same keys as auth_server,
``None`` if accounting is not configured)
* reauth_interval (int) - re-authentication interval in seconds (0 = disabled)
* pmksa_caching (bool) - whether PMKSA caching (opportunistic key caching) is enabled
Note: The RADIUS shared secret is intentionally omitted from the return
value for security reasons.
Example::
{
"CorpWiFi": {
"enabled": True,
"ssid": "CorpWiFi",
"auth_server": {
"host": "radius.corp.example",
"port": 1812,
"timeout": 5,
"retries": 3,
},
"acct_server": {
"host": "radius.corp.example",
"port": 1813,
"timeout": 5,
"retries": 3,
},
"reauth_interval": 3600,
"pmksa_caching": True,
}
}
"""
raise NotImplementedError
def get_packages(self) -> List[PackageDict]:
"""
Returns all packages currently known to the device's package manager
(e.g. ``opkg`` on OpenWrt, ``apk`` on Alpine-based APs).
Each entry contains:
* name (string) - package name
* version (string) - installed or available version string
* installed (bool) - ``True`` if the package is currently installed
* description (string) - short package description
* size (int) - package size in bytes (0 if unknown)
* source (string) - repository / feed the package comes from
Example::
[
{
"name": "luci-app-statistics",
"version": "git-24.001.00000-1",
"installed": True,
"description": "LuCI Statistics application",
"size": 20480,
"source": "openwrt/packages",
},
{
"name": "collectd-mod-wireless",
"version": "5.12.0-24",
"installed": False,
"description": "Wireless statistics plugin for collectd",
"size": 8192,
"source": "openwrt/packages",
},
]
"""
raise NotImplementedError
def install_package(self, name: str, version: str = "") -> None:
"""
Installs a package on the device.
The method blocks until the installation is complete. After it returns
successfully the package is available for use without a reboot
(where the underlying package manager supports this).
:param name: Package name as known to the package manager.
:param version: Exact version to install. An empty string (default)
installs the latest available version.
:raises NotImplementedError: If the driver does not support package management.
:raises ValueError: If the package name is unknown or the version does
not exist in any configured feed.
:raises RuntimeError: If the installation fails on the device side
(e.g. dependency conflict, disk full).
Example::
driver.install_package("luci-app-statistics")
driver.install_package("collectd", version="5.12.0-24")
"""
raise NotImplementedError
def remove_package(self, name: str) -> None:
"""
Removes an installed package from the device.
The method blocks until the removal is complete.
:param name: Package name to remove.
:raises NotImplementedError: If the driver does not support package management.
:raises ValueError: If the package is not currently installed.
:raises RuntimeError: If the removal fails on the device side
(e.g. other packages depend on it).
Example::
driver.remove_package("luci-app-statistics")
"""
raise NotImplementedError
def get_package_config(self, name: str) -> Dict[str, Any]:
"""
Returns the current configuration of an installed package as a
dictionary. The structure is package-specific.
:param name: Package name.
:raises NotImplementedError: If the driver does not support package management.
:raises ValueError: If the package is not installed.
Example::
driver.get_package_config("luci-app-statistics")
# →
{
"collectd": {
"enabled": True,
"interval": 30,
},
"rrdtool": {
"datadir": "/tmp/rrd",
"stepsize": 30,
"heartbeat": 60,
},
}
"""
raise NotImplementedError
def set_package_config(self, name: str, config: Dict[str, Any]) -> None:
"""
Writes a new configuration for an installed package.
The ``config`` dict must match the structure returned by
:meth:`get_package_config`. Unknown keys are ignored or raise a
``ValueError`` depending on the driver implementation.
Changes take effect immediately where the package supports live
reload; otherwise a package restart or device reboot may be
required – behaviour is driver-specific.
:param name: Package name.
:param config: New configuration as a nested dictionary.
:raises NotImplementedError: If the driver does not support package management.
:raises ValueError: If the package is not installed or the configuration
contains invalid values.
:raises RuntimeError: If the device rejects the configuration.
Example::
driver.set_package_config(
"luci-app-statistics",
{
"collectd": {"enabled": True, "interval": 60},
"rrdtool": {"datadir": "/tmp/rrd", "stepsize": 60, "heartbeat": 120},
},
)
"""
raise NotImplementedError
napalm_device_types/firewall.py
+285
View File
@@ -0,0 +1,285 @@
"""
Abstract base class for firewall drivers.
Usage::
from napalm_device_types import FirewallDriver
class FortiGateDriver(FirewallDriver):
def get_security_zones(self):
...
"""
from typing import Any, Dict, List
from napalm.base import NetworkDriver
from napalm_device_types.models import (
NATTranslationDict,
PackageDict,
SecurityZoneDict,
SessionDict,
VPNTunnelDict,
)
class FirewallDriver(NetworkDriver):
"""
Abstract intermediate driver for firewall/security devices.
Inherits all standard NAPALM NetworkDriver methods (including
``get_firewall_policies()``) and adds firewall-specific operations
that concrete drivers must implement.
"""
def get_nat_translations(self) -> List[NATTranslationDict]:
"""
Returns a list of active NAT translation entries.
Each entry contains:
* protocol (string) - ``"tcp"``, ``"udp"``, ``"icmp"``
* inside_local (string) - original source address (IP or IP:port)
* inside_global (string) - translated source address (IP or IP:port)
* outside_local (string) - destination as seen from inside
* outside_global (string) - actual destination address
* age (float) - translation entry age in seconds
Example::
[
{
"protocol": "tcp",
"inside_local": "192.168.1.10:54321",
"inside_global": "203.0.113.1:54321",
"outside_local": "1.1.1.1:443",
"outside_global": "1.1.1.1:443",
"age": 120.5,
}
]
"""
raise NotImplementedError
def get_security_zones(self) -> Dict[str, SecurityZoneDict]:
"""
Returns the security zone configuration.
Keys are zone names. Each value contains:
* interfaces (list of strings) - interfaces assigned to this zone
* policy (string) - name of the security policy applied to this zone
* description (string) - zone description
Example::
{
"LAN": {
"interfaces": ["eth0", "eth1"],
"policy": "LAN-policy",
"description": "Internal LAN zone",
},
"WAN": {
"interfaces": ["eth2"],
"policy": "WAN-policy",
"description": "Uplink to internet",
},
}
"""
raise NotImplementedError
def get_sessions(self) -> List[SessionDict]:
"""
Returns a list of active connection sessions (stateful flows).
Each entry contains:
* protocol (string) - ``"tcp"``, ``"udp"``, ``"icmp"``
* src_ip (string) - source IP address
* src_port (int) - source port (0 for ICMP)
* dst_ip (string) - destination IP address
* dst_port (int) - destination port (0 for ICMP)
* state (string) - session state, e.g. ``"established"``, ``"syn_sent"``
* age (float) - session age in seconds
Example::
[
{
"protocol": "tcp",
"src_ip": "192.168.1.10",
"src_port": 54321,
"dst_ip": "1.1.1.1",
"dst_port": 443,
"state": "established",
"age": 30.2,
}
]
"""
raise NotImplementedError
def get_vpn_tunnels(self) -> Dict[str, VPNTunnelDict]:
"""
Returns the status of VPN tunnels.
Keys are tunnel names or identifiers. Each value contains:
* type (string) - tunnel type: ``"IPsec"``, ``"SSL"``, ``"GRE"``, ``"WireGuard"``
* local_endpoint (string) - local tunnel endpoint IP
* remote_endpoint (string) - remote tunnel endpoint IP
* is_up (bool) - whether the tunnel is operationally up
* uptime (int) - tunnel uptime in seconds (0 if down)
* bytes_in (int) - total bytes received through the tunnel
* bytes_out (int) - total bytes sent through the tunnel
Example::
{
"vpn-to-branch": {
"type": "IPsec",
"local_endpoint": "203.0.113.1",
"remote_endpoint": "198.51.100.1",
"is_up": True,
"uptime": 86400,
"bytes_in": 104857600,
"bytes_out": 52428800,
}
}
"""
raise NotImplementedError
def get_packages(self) -> List[PackageDict]:
"""
Returns all packages / plugins currently known to the firewall's
package manager (e.g. ``pkg`` on pfSense/OPNsense, ``FortiGate
License`` add-ons, ``apt`` on Debian-based firewalls).
Each entry contains:
* name (string) - package name
* version (string) - installed or available version string
* installed (bool) - ``True`` if the package is currently installed
* description (string) - short package description
* size (int) - package size in bytes (0 if unknown)
* source (string) - repository / channel the package comes from
Example::
[
{
"name": "pfBlockerNG",
"version": "3.2.0_4",
"installed": True,
"description": "IP and DNS blocking for pfSense",
"size": 2097152,
"source": "pfSense-pkg",
},
{
"name": "suricata",
"version": "7.0.3_1",
"installed": False,
"description": "High-performance Network IDS/IPS",
"size": 51380224,
"source": "pfSense-pkg",
},
]
"""
raise NotImplementedError
def install_package(self, name: str, version: str = "") -> None:
"""
Installs a package or plugin on the firewall.
The method blocks until the installation is complete. Whether a
reboot is required afterwards depends on the device; check the vendor
documentation.
:param name: Package name as known to the package manager.
:param version: Exact version to install. An empty string (default)
installs the latest available version.
:raises NotImplementedError: If the driver does not support package management.
:raises ValueError: If the package name is unknown or the requested
version is not available.
:raises RuntimeError: If the installation fails on the device side
(e.g. license missing, dependency conflict, disk full).
Example::
driver.install_package("pfBlockerNG")
driver.install_package("suricata", version="7.0.3_1")
"""
raise NotImplementedError
def remove_package(self, name: str) -> None:
"""
Removes an installed package or plugin from the firewall.
The method blocks until the removal is complete.
:param name: Package name to remove.
:raises NotImplementedError: If the driver does not support package management.
:raises ValueError: If the package is not currently installed.
:raises RuntimeError: If the removal fails on the device side
(e.g. the package is a system dependency).
Example::
driver.remove_package("pfBlockerNG")
"""
raise NotImplementedError
def get_package_config(self, name: str) -> Dict[str, Any]:
"""
Returns the current configuration of an installed package or plugin
as a dictionary. The structure is package-specific.
:param name: Package name.
:raises NotImplementedError: If the driver does not support package management.
:raises ValueError: If the package is not installed.
Example::
driver.get_package_config("pfBlockerNG")
# →
{
"enable": True,
"maxmind_key": "",
"blocklists": [
{"name": "PRI1", "action": "Deny_Both", "enabled": True},
{"name": "DNSBL_ADs", "action": "Unbound", "enabled": True},
],
"update_interval": "Once a day",
}
"""
raise NotImplementedError
def set_package_config(self, name: str, config: Dict[str, Any]) -> None:
"""
Writes a new configuration for an installed package or plugin.
The ``config`` dict must match the structure returned by
:meth:`get_package_config`. Unknown keys are ignored or raise a
``ValueError`` depending on the driver implementation.
Changes take effect immediately where the package supports live
reload; otherwise a package restart or device reboot may be
required – behaviour is driver-specific.
:param name: Package name.
:param config: New configuration as a nested dictionary.
:raises NotImplementedError: If the driver does not support package management.
:raises ValueError: If the package is not installed or the configuration
contains invalid values.
:raises RuntimeError: If the device rejects the configuration.
Example::
driver.set_package_config(
"pfBlockerNG",
{
"enable": True,
"blocklists": [
{"name": "PRI1", "action": "Deny_Both", "enabled": True},
],
"update_interval": "Twice a day",
},
)
"""
raise NotImplementedError
napalm_device_types/hypervisor.py
+523
View File
@@ -0,0 +1,523 @@
"""
Abstract base class for hypervisor drivers.
Usage::
from napalm_device_types import HypervisorDriver
class ProxmoxDriver(HypervisorDriver):
def get_vms(self):
...
"""
from typing import Any, Dict, List
from napalm.base import NetworkDriver
from napalm_device_types.models import (
PackageDict,
SnapshotDict,
StorageVolumeDict,
VMConfigDict,
VMDict,
VirtualNetworkDict,
)
class HypervisorDriver(NetworkDriver):
"""
Abstract intermediate driver for hypervisors and virtualisation platforms
(e.g. Proxmox VE, VMware ESXi, KVM/libvirt, Hyper-V).
Inherits all standard NAPALM NetworkDriver methods and adds
hypervisor-specific operations that concrete drivers must implement.
"""
# ------------------------------------------------------------------
# Virtual machines – read
# ------------------------------------------------------------------
def get_vms(self) -> List[VMDict]:
"""
Returns a list of all virtual machines known to this hypervisor,
including their runtime status.
Each entry contains:
* name (string) - VM display name
* vmid (int) - hypervisor-internal numeric ID
* status (string) - ``"running"``, ``"stopped"``, ``"paused"``, ``"suspended"``
* vcpus (int) - number of virtual CPUs assigned
* memory (int) - configured RAM in megabytes
* cpu_usage (float) - current CPU utilisation 0.0–1.0
* memory_usage (int) - current RAM usage in megabytes
* uptime (int) - uptime in seconds (0 if not running)
* node (string) - cluster node this VM lives on (empty string for standalone)
Example::
[
{
"name": "web01",
"vmid": 100,
"status": "running",
"vcpus": 4,
"memory": 8192,
"cpu_usage": 0.12,
"memory_usage": 3200,
"uptime": 864000,
"node": "pve1",
},
{
"name": "db-backup",
"vmid": 101,
"status": "stopped",
"vcpus": 2,
"memory": 4096,
"cpu_usage": 0.0,
"memory_usage": 0,
"uptime": 0,
"node": "pve1",
},
]
"""
raise NotImplementedError
def get_vm_config(self, name: str) -> VMConfigDict:
"""
Returns the full hardware configuration of a virtual machine.
:param name: VM name or numeric VMID as a string.
:raises ValueError: If no VM with the given name/ID exists.
The returned dictionary contains:
* name (string) - VM display name
* vmid (int) - hypervisor-internal numeric ID
* vcpus (int) - number of virtual CPUs
* memory (int) - RAM in megabytes
* os_type (string) - guest OS type hint (e.g. ``"l26"``, ``"win11"``, ``"other"``)
* boot_order (list of strings) - boot device sequence (e.g. ``["scsi0", "net0"]``)
* disks (list) - attached virtual disks, each with:
* device (string) - device ID (e.g. ``"scsi0"``)
* storage (string) - backing storage pool
* size (int) - disk size in gigabytes
* format (string) - image format: ``"qcow2"``, ``"raw"``, ``"vmdk"``
* bootable (bool) - whether this disk is in the boot order
* nics (list) - virtual network interfaces, each with:
* device (string) - device ID (e.g. ``"net0"``)
* mac (string) - MAC address
* model (string) - NIC model (e.g. ``"virtio"``, ``"e1000"``)
* bridge (string) - host bridge the NIC is connected to
* vlan_id (int) - VLAN tag (0 = untagged)
* description (string) - free-text notes / description
* tags (list of strings) - organisational tags
Example::
{
"name": "web01",
"vmid": 100,
"vcpus": 4,
"memory": 8192,
"os_type": "l26",
"boot_order": ["scsi0"],
"disks": [
{
"device": "scsi0",
"storage": "local-lvm",
"size": 32,
"format": "raw",
"bootable": True,
}
],
"nics": [
{
"device": "net0",
"mac": "BC:24:11:AA:BB:CC",
"model": "virtio",
"bridge": "vmbr0",
"vlan_id": 10,
}
],
"description": "Production web server",
"tags": ["prod", "web"],
}
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Virtual machines – power actions
# ------------------------------------------------------------------
def start_vm(self, name: str) -> None:
"""
Powers on a stopped or suspended virtual machine.
The method blocks until the hypervisor reports the VM as running.
:param name: VM name or numeric VMID as a string.
:raises ValueError: If no VM with the given name/ID exists.
:raises RuntimeError: If the VM cannot be started (e.g. resource limit).
Example::
driver.start_vm("web01")
"""
raise NotImplementedError
def stop_vm(self, name: str, force: bool = False) -> None:
"""
Shuts down a virtual machine.
With ``force=False`` (default) a graceful ACPI shutdown is requested
and the method blocks until the VM is stopped. With ``force=True``
the VM is immediately powered off (equivalent to pulling the plug).
:param name: VM name or numeric VMID as a string.
:param force: ``True`` for immediate power-off, ``False`` for graceful shutdown.
:raises ValueError: If no VM with the given name/ID exists.
:raises RuntimeError: If the VM is already stopped.
Example::
driver.stop_vm("web01") # graceful
driver.stop_vm("web01", force=True) # hard off
"""
raise NotImplementedError
def reboot_vm(self, name: str, force: bool = False) -> None:
"""
Reboots a virtual machine.
With ``force=False`` (default) a graceful ACPI reboot is requested.
With ``force=True`` the VM is reset immediately without OS shutdown.
:param name: VM name or numeric VMID as a string.
:param force: ``True`` for an immediate reset, ``False`` for graceful reboot.
:raises ValueError: If no VM with the given name/ID exists.
:raises RuntimeError: If the VM is not currently running.
Example::
driver.reboot_vm("web01")
driver.reboot_vm("web01", force=True)
"""
raise NotImplementedError
def suspend_vm(self, name: str) -> None:
"""
Suspends (pauses) a running virtual machine, preserving its in-memory
state. The VM can be resumed with :meth:`start_vm`.
:param name: VM name or numeric VMID as a string.
:raises ValueError: If no VM with the given name/ID exists.
:raises RuntimeError: If the VM is not currently running.
Example::
driver.suspend_vm("web01")
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Snapshots
# ------------------------------------------------------------------
def get_snapshots(self, name: str) -> List[SnapshotDict]:
"""
Returns all snapshots of a virtual machine.
:param name: VM name or numeric VMID as a string.
:raises ValueError: If no VM with the given name/ID exists.
Each entry contains:
* name (string) - snapshot name
* vm (string) - VM name this snapshot belongs to
* created (float) - creation timestamp (Unix epoch)
* description (string) - optional snapshot description
* has_memory (bool) - whether the snapshot includes RAM state
* parent (string) - name of the parent snapshot (empty string for root)
Example::
[
{
"name": "before-upgrade",
"vm": "web01",
"created": 1746921600.0,
"description": "Clean state before kernel upgrade",
"has_memory": False,
"parent": "",
},
{
"name": "post-upgrade",
"vm": "web01",
"created": 1746925200.0,
"description": "",
"has_memory": False,
"parent": "before-upgrade",
},
]
"""
raise NotImplementedError
def snapshot_create(self, name: str, snapshot: str,
description: str = "", include_memory: bool = False) -> None:
"""
Creates a snapshot of a virtual machine.
:param name: VM name or numeric VMID as a string.
:param snapshot: Name for the new snapshot.
:param description: Optional human-readable description.
:param include_memory: Whether to include the current RAM state
(only possible while the VM is running).
:raises ValueError: If no VM with the given name/ID exists, or a
snapshot with that name already exists.
:raises RuntimeError: If snapshot creation fails.
Example::
driver.snapshot_create("web01", "before-upgrade",
description="Clean state before kernel upgrade")
"""
raise NotImplementedError
def snapshot_delete(self, name: str, snapshot: str) -> None:
"""
Deletes a snapshot of a virtual machine.
:param name: VM name or numeric VMID as a string.
:param snapshot: Name of the snapshot to delete.
:raises ValueError: If the VM or snapshot does not exist.
:raises RuntimeError: If other snapshots depend on this one (must delete children first).
Example::
driver.snapshot_delete("web01", "before-upgrade")
"""
raise NotImplementedError
def snapshot_rollback(self, name: str, snapshot: str) -> None:
"""
Reverts a virtual machine to a previously created snapshot.
The VM is stopped (if running), reverted, and then left in the state
the snapshot recorded (running or stopped depending on ``has_memory``).
:param name: VM name or numeric VMID as a string.
:param snapshot: Name of the snapshot to roll back to.
:raises ValueError: If the VM or snapshot does not exist.
:raises RuntimeError: If the rollback fails.
Example::
driver.snapshot_rollback("web01", "before-upgrade")
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Storage
# ------------------------------------------------------------------
def get_storage(self) -> Dict[str, StorageVolumeDict]:
"""
Returns the storage pools / datastores configured on this hypervisor.
Keys are storage pool names. Each value contains:
* name (string) - pool name (repeated for convenience)
* type (string) - backend type: ``"dir"``, ``"lvm"``, ``"zfs"``,
``"nfs"``, ``"ceph"``, ``"iscsi"`` etc.
* total (int) - total capacity in bytes
* used (int) - used space in bytes
* available (int) - free space in bytes
* enabled (bool) - whether the pool is administratively enabled
* shared (bool) - whether the pool is accessible from multiple cluster nodes
Example::
{
"local-lvm": {
"name": "local-lvm",
"type": "lvm",
"total": 107374182400,
"used": 53687091200,
"available": 53687091200,
"enabled": True,
"shared": False,
},
"ceph-pool": {
"name": "ceph-pool",
"type": "ceph",
"total": 1099511627776,
"used": 274877906944,
"available": 824633720832,
"enabled": True,
"shared": True,
},
}
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Virtual networking
# ------------------------------------------------------------------
def get_virtual_networks(self) -> Dict[str, VirtualNetworkDict]:
"""
Returns the virtual networks / bridges defined on this hypervisor.
Keys are network names. Each value contains:
* name (string) - network name (repeated for convenience)
* type (string) - network type: ``"bridge"``, ``"ovs"``, ``"nat"``, ``"vxlan"``
* bridge (string) - underlying host bridge interface
* vlan_id (int) - associated VLAN tag (0 = untagged / all VLANs)
* autostart (bool) - whether the network starts automatically at boot
* active (bool) - whether the network is currently active
Example::
{
"vmbr0": {
"name": "vmbr0",
"type": "bridge",
"bridge": "vmbr0",
"vlan_id": 0,
"autostart": True,
"active": True,
},
"vmbr10": {
"name": "vmbr10",
"type": "bridge",
"bridge": "vmbr10",
"vlan_id": 10,
"autostart": True,
"active": True,
},
}
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Package management (hypervisor extensions / plugins)
# ------------------------------------------------------------------
def get_packages(self) -> List[PackageDict]:
"""
Returns all packages known to the hypervisor's package manager
(e.g. ``apt`` on Proxmox VE, vendor extension bundles on ESXi).
Each entry contains:
* name (string) - package name
* version (string) - installed or available version string
* installed (bool) - ``True`` if the package is currently installed
* description (string) - short package description
* size (int) - package size in bytes (0 if unknown)
* source (string) - repository the package comes from
Example::
[
{
"name": "proxmox-backup-client",
"version": "3.2.4-1",
"installed": True,
"description": "Proxmox Backup Client tools",
"size": 8388608,
"source": "pve-no-subscription",
},
{
"name": "ifupdown2",
"version": "3.2.0-1+pmx4",
"installed": True,
"description": "Network interface management daemon",
"size": 524288,
"source": "pve-no-subscription",
},
]
"""
raise NotImplementedError
def install_package(self, name: str, version: str = "") -> None:
"""
Installs a package on the hypervisor host.
:param name: Package name as known to the package manager.
:param version: Exact version to install. Empty string installs latest.
:raises NotImplementedError: If the driver does not support package management.
:raises ValueError: If the package name is unknown or the version unavailable.
:raises RuntimeError: If the installation fails on the host side.
Example::
driver.install_package("proxmox-backup-client")
"""
raise NotImplementedError
def remove_package(self, name: str) -> None:
"""
Removes an installed package from the hypervisor host.
:param name: Package name to remove.
:raises NotImplementedError: If the driver does not support package management.
:raises ValueError: If the package is not currently installed.
:raises RuntimeError: If removal fails (e.g. required dependency).
Example::
driver.remove_package("proxmox-backup-client")
"""
raise NotImplementedError
def get_package_config(self, name: str) -> Dict[str, Any]:
"""
Returns the current configuration of an installed hypervisor package
as a dictionary. The structure is package-specific.
:param name: Package name.
:raises NotImplementedError: If the driver does not support package management.
:raises ValueError: If the package is not installed.
Example::
driver.get_package_config("proxmox-backup-client")
# →
{
"server": "backup.corp.example",
"datastore": "vm-backups",
"fingerprint": "AB:CD:EF:...",
"schedule": "daily",
"retention": {"keep_last": 7, "keep_weekly": 4},
}
"""
raise NotImplementedError
def set_package_config(self, name: str, config: Dict[str, Any]) -> None:
"""
Writes a new configuration for an installed hypervisor package.
:param name: Package name.
:param config: New configuration as a nested dictionary.
:raises NotImplementedError: If the driver does not support package management.
:raises ValueError: If the package is not installed or config is invalid.
:raises RuntimeError: If the device rejects the configuration.
Example::
driver.set_package_config(
"proxmox-backup-client",
{
"server": "backup.corp.example",
"datastore": "vm-backups",
"schedule": "daily",
"retention": {"keep_last": 14, "keep_weekly": 4},
},
)
"""
raise NotImplementedError
napalm_device_types/models.py
+430
View File
@@ -0,0 +1,430 @@
"""
TypedDicts for NAPALM device-type-specific return values.
These supplement the types defined in napalm.base.models and are used by the
abstract device-type driver classes in this package.
"""
from typing import Dict, List, Optional
from typing_extensions import TypedDict
# ---------------------------------------------------------------------------
# Common (shared across device types)
# ---------------------------------------------------------------------------
class RadiusServerDict(TypedDict):
host: str
port: int
timeout: int
retries: int
class MACACLEntryDict(TypedDict):
mac: str
action: str
description: str
class MACACLDict(TypedDict):
"""MAC access-control list for a single binding point.
``name`` is the SSID name on an access point, or the interface name on a
switch.
"""
name: str
policy: str
entries: List[MACACLEntryDict]
class PackageDict(TypedDict):
"""A software package / plugin installed on the device."""
name: str
version: str
installed: bool
description: str
size: int
source: str
# ---------------------------------------------------------------------------
# Access Point
# ---------------------------------------------------------------------------
class WirelessClientDict(TypedDict):
mac: str
ssid: str
radio: str
signal: int
noise: int
tx_rate: float
rx_rate: float
uptime: int
class SSIDDict(TypedDict):
enabled: bool
radio: str
bssid: str
encryption: str
hidden: bool
clients: int
class RadioStatusDict(TypedDict):
enabled: bool
band: str
channel: int
channel_width: int
tx_power: int
frequency: float
class WirelessConfigDict(TypedDict):
country_code: str
regulatory_domain: str
beacon_interval: int
dtim_period: int
rts_threshold: int
fragmentation_threshold: int
short_preamble: bool
wmm_enabled: bool
class FastTransitionConfigDict(TypedDict):
enabled: bool
ssid: str
mobility_domain: str
reassociation_deadline: int
r0_key_lifetime: int
r1_key_holder: str
pmk_r1_push: bool
over_ds: bool
class MeshPeerDict(TypedDict):
mac: str
radio: str
signal: int
tx_rate: float
rx_rate: float
uptime: int
hop_count: int
class MeshConfigDict(TypedDict):
enabled: bool
radio: str
mesh_id: str
path_metric: str
gate_announcements: bool
is_gate: bool
encryption: str
class SSIDBridgeDict(TypedDict):
ssid: str
bridge: str
vlan_id: int
tagged: bool
client_isolation: bool
class Dot1XConfigDict(TypedDict):
"""802.1X / WPA-Enterprise config for an access-point SSID."""
enabled: bool
ssid: str
auth_server: RadiusServerDict
acct_server: Optional[RadiusServerDict]
reauth_interval: int
pmksa_caching: bool
# ---------------------------------------------------------------------------
# Switch
# ---------------------------------------------------------------------------
class STPInterfaceDict(TypedDict):
role: str
state: str
cost: int
port_priority: int
class SpanningTreeDict(TypedDict):
mode: str
root_bridge: bool
root_id: str
root_priority: int
bridge_id: str
bridge_priority: int
interfaces: Dict[str, STPInterfaceDict]
class PortChannelDict(TypedDict):
members: List[str]
protocol: str
min_links: int
is_up: bool
class Dot1XPortDict(TypedDict):
"""802.1X / NAC configuration for a single switch port."""
enabled: bool
port_control: str
host_mode: str
auth_server: RadiusServerDict
acct_server: Optional[RadiusServerDict]
reauthentication: bool
reauth_interval: int
guest_vlan: int
auth_fail_vlan: int
class PoEPortDict(TypedDict):
enabled: bool
status: str
poe_class: str
power_draw: float
power_budget: float
voltage: float
current: float
class PoESummaryDict(TypedDict):
total_power_budget: float
total_power_draw: float
ports: Dict[str, PoEPortDict]
# ---------------------------------------------------------------------------
# Firewall
# ---------------------------------------------------------------------------
class NATTranslationDict(TypedDict):
protocol: str
inside_local: str
inside_global: str
outside_local: str
outside_global: str
age: float
class SecurityZoneDict(TypedDict):
interfaces: List[str]
policy: str
description: str
class SessionDict(TypedDict):
protocol: str
src_ip: str
src_port: int
dst_ip: str
dst_port: int
state: str
age: float
class VPNTunnelDict(TypedDict):
type: str
local_endpoint: str
remote_endpoint: str
is_up: bool
uptime: int
bytes_in: int
bytes_out: int
# ---------------------------------------------------------------------------
# Hypervisor
# ---------------------------------------------------------------------------
class VMDiskDict(TypedDict):
device: str
storage: str
size: int
format: str
bootable: bool
class VMNICDict(TypedDict):
device: str
mac: str
model: str
bridge: str
vlan_id: int
class VMDict(TypedDict):
name: str
vmid: int
status: str
vcpus: int
memory: int
cpu_usage: float
memory_usage: int
uptime: int
node: str
class VMConfigDict(TypedDict):
name: str
vmid: int
vcpus: int
memory: int
os_type: str
boot_order: List[str]
disks: List[VMDiskDict]
nics: List[VMNICDict]
description: str
tags: List[str]
class StorageVolumeDict(TypedDict):
name: str
type: str
total: int
used: int
available: int
enabled: bool
shared: bool
class VirtualNetworkDict(TypedDict):
name: str
type: str
bridge: str
vlan_id: int
autostart: bool
active: bool
class SnapshotDict(TypedDict):
name: str
vm: str
created: float
description: str
has_memory: bool
parent: str
# ---------------------------------------------------------------------------
# Storage / NAS
# ---------------------------------------------------------------------------
class PhysicalDiskDict(TypedDict):
"""A physical drive installed in the storage device."""
slot: str
model: str
serial: str
vendor: str
type: str # "hdd", "ssd", "nvme"
size: int # bytes
rpm: int # 0 for SSD/NVMe
temperature: int # Celsius; -1 if unavailable
health: str # "healthy", "warning", "failed", "unknown"
pool: str # name of the containing pool; empty string if spare/unassigned
class DiskPoolDict(TypedDict):
"""A RAID array, ZFS pool, or volume group."""
name: str
type: str # "zfs", "lvm", "md", "hardware-raid", "btrfs"
level: str # "stripe", "mirror", "raidz1", "raidz2", "raidz3",
# "raid0" … "raid60", "single", etc.
status: str # "online", "degraded", "faulted", "offline", "unknown"
total: int # bytes
used: int # bytes
available: int # bytes
disks: List[str] # slot identifiers of member disks
auto_expand: bool
dedup: bool
compression: str # "off", "lz4", "gzip", "zstd", etc.
class LogicalVolumeDict(TypedDict):
"""A logical volume, ZFS dataset, or LUN exposed to clients."""
name: str
pool: str
type: str # "filesystem", "volume" (block device / LUN), "zvol"
total: int # bytes
used: int # bytes
available: int # bytes
mountpoint: str # empty string for block volumes
compression: str # "off", "lz4", etc.
dedup: bool
readonly: bool
snapshots: int # number of snapshots currently held
class NASShareDict(TypedDict):
"""A network share or iSCSI target exported by the storage device."""
name: str
protocol: str # "nfs", "smb", "afp", "ftp", "sftp", "iscsi", "webdav"
path: str # local filesystem path or iSCSI target IQN
volume: str # logical volume or dataset this share is backed by
enabled: bool
readonly: bool
description: str
clients: List[str] # IP/subnet allow-list; empty list = all hosts allowed
class VolumeSnapshotDict(TypedDict):
"""A point-in-time snapshot of a logical volume or dataset."""
name: str
volume: str # parent volume / dataset
created: float # Unix epoch
size: int # bytes of unique data referenced by this snapshot
description: str
clones: List[str] # volumes cloned from this snapshot
class StorageQuotaDict(TypedDict):
"""A filesystem quota applied to a user, group, or dataset."""
target: str # username, group name, or dataset path
target_type: str # "user", "group", "dataset"
volume: str # volume / dataset the quota applies to
used: int # bytes currently used
quota: int # hard limit in bytes; 0 = no limit
ref_quota: int # referenced-data limit in bytes; 0 = no limit
class StorageServiceDict(TypedDict):
"""Status of a file-sharing or access service running on the device."""
name: str # "nfs", "smb", "ftp", "ssh", "iscsi", "webdav", etc.
enabled: bool # administratively enabled
running: bool # currently active / listening
port: int # primary listening port; 0 if N/A
version: str # protocol version string (e.g. "4.1" for NFSv4.1)
class ReplicationJobDict(TypedDict):
"""A scheduled replication or sync task."""
name: str
source: str # source path / dataset
target: str # destination path / dataset (may be remote: host:path)
direction: str # "push" or "pull"
schedule: str # cron expression or human label ("daily", "hourly")
enabled: bool
last_run: float # Unix epoch; 0.0 if never run
last_status: str # "success", "failed", "running", "pending"
bytes_sent: int # bytes transferred in the last run
napalm_device_types/storage.py
+599
View File
@@ -0,0 +1,599 @@
"""
Abstract base class for storage and NAS/SAN device drivers.
Usage::
from napalm_device_types import StorageDriver
class TrueNASDriver(StorageDriver):
def get_disks(self):
...
"""
from typing import Any, Dict, List
from napalm.base import NetworkDriver
from napalm_device_types.models import (
DiskPoolDict,
LogicalVolumeDict,
NASShareDict,
PackageDict,
PhysicalDiskDict,
ReplicationJobDict,
StorageQuotaDict,
StorageServiceDict,
VolumeSnapshotDict,
)
class StorageDriver(NetworkDriver):
"""
Abstract intermediate driver for storage appliances and NAS/SAN devices
(e.g. TrueNAS SCALE/CORE, Synology DSM, QNAP QTS, NetApp ONTAP,
OpenMediaVault, Pure Storage, IBM Storwize).
Inherits all standard NAPALM NetworkDriver methods and adds
storage-specific operations that concrete drivers must implement.
"""
# ------------------------------------------------------------------
# Physical hardware
# ------------------------------------------------------------------
def get_disks(self) -> List[PhysicalDiskDict]:
"""
Returns all physical drives detected by the storage device.
Each entry contains:
* slot (string) - bay or device identifier (e.g. ``"bay1"``, ``"sda"``, ``"nvme0n1"``)
* model (string) - drive model string
* serial (string) - drive serial number
* vendor (string) - drive manufacturer
* type (string) - ``"hdd"``, ``"ssd"``, or ``"nvme"``
* size (int) - raw capacity in bytes
* rpm (int) - rotational speed; ``0`` for SSD/NVMe
* temperature (int) - current temperature in Celsius; ``-1`` if unavailable
* health (string) - ``"healthy"``, ``"warning"``, ``"failed"``, or ``"unknown"``
* pool (string) - name of the pool this disk belongs to; empty string if unassigned/spare
Example::
[
{
"slot": "bay1",
"model": "HGST HUS726T6TALE6L4",
"serial": "K3GXXXXX",
"vendor": "HGST",
"type": "hdd",
"size": 6001175126016,
"rpm": 7200,
"temperature": 34,
"health": "healthy",
"pool": "tank",
},
{
"slot": "bay5",
"model": "Samsung SSD 870 EVO 1TB",
"serial": "S5XXXXXXX",
"vendor": "Samsung",
"type": "ssd",
"size": 1000204886016,
"rpm": 0,
"temperature": 28,
"health": "healthy",
"pool": "fast-pool",
},
]
"""
raise NotImplementedError
def get_disk_pools(self) -> Dict[str, DiskPoolDict]:
"""
Returns the disk pools (RAID arrays, ZFS pools, LVM volume groups, etc.)
configured on the device.
Keys are pool names. Each value contains:
* name (string) - pool name (repeated for convenience)
* type (string) - ``"zfs"``, ``"lvm"``, ``"md"``, ``"hardware-raid"``, ``"btrfs"``
* level (string) - RAID/redundancy level: ``"mirror"``, ``"raidz1"``, ``"raidz2"``,
``"raidz3"``, ``"stripe"``, ``"raid0"`` … ``"raid60"``, ``"single"``
* status (string) - ``"online"``, ``"degraded"``, ``"faulted"``, ``"offline"``, ``"unknown"``
* total (int) - usable capacity in bytes
* used (int) - used space in bytes
* available (int) - free space in bytes
* disks (list of strings) - slot identifiers of member drives
* auto_expand (bool) - whether the pool grows automatically when disks are replaced with larger ones
* dedup (bool) - whether deduplication is enabled
* compression (string) - pool-level compression algorithm: ``"off"``, ``"lz4"``,
``"gzip"``, ``"zstd"``, etc.
Example::
{
"tank": {
"name": "tank",
"type": "zfs",
"level": "raidz2",
"status": "online",
"total": 21990232555520,
"used": 8796093022208,
"available": 13194139533312,
"disks": ["bay1", "bay2", "bay3", "bay4", "bay5", "bay6"],
"auto_expand": True,
"dedup": False,
"compression": "lz4",
},
}
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Logical volumes / datasets
# ------------------------------------------------------------------
def get_volumes(self) -> Dict[str, LogicalVolumeDict]:
"""
Returns all logical volumes, ZFS datasets, or LUNs on the device.
Keys are volume paths (e.g. ``"tank/data"``, ``"tank/media"``).
Each value contains:
* name (string) - volume name (leaf component or full path)
* pool (string) - parent pool
* type (string) - ``"filesystem"``, ``"volume"`` (block device / LUN), or ``"zvol"``
* total (int) - quota or provisioned size in bytes; ``0`` means unlimited
* used (int) - space currently used in bytes
* available (int) - space available in bytes
* mountpoint (string) - local mount path; empty string for block volumes / LUNs
* compression (string) - active compression algorithm (``"off"``, ``"lz4"``, ``"zstd"``, etc.)
* dedup (bool) - whether deduplication is active on this volume
* readonly (bool) - whether the volume is mounted read-only
* snapshots (int) - number of snapshots currently held
Example::
{
"tank/media": {
"name": "media",
"pool": "tank",
"type": "filesystem",
"total": 0,
"used": 4398046511104,
"available": 13194139533312,
"mountpoint": "/mnt/tank/media",
"compression": "lz4",
"dedup": False,
"readonly": False,
"snapshots": 7,
},
"tank/backups": {
"name": "backups",
"pool": "tank",
"type": "filesystem",
"total": 5497558138880,
"used": 1099511627776,
"available": 4398046511104,
"mountpoint": "/mnt/tank/backups",
"compression": "zstd",
"dedup": False,
"readonly": False,
"snapshots": 14,
},
}
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Shares
# ------------------------------------------------------------------
def get_shares(self) -> Dict[str, NASShareDict]:
"""
Returns all network shares and iSCSI targets exported by the device.
Keys are share names. Each value contains:
* name (string) - share name (repeated for convenience)
* protocol (string) - ``"nfs"``, ``"smb"``, ``"afp"``, ``"ftp"``, ``"sftp"``,
``"iscsi"``, or ``"webdav"``
* path (string) - local filesystem path; for iSCSI the target IQN
* volume (string) - logical volume or dataset this share is backed by
* enabled (bool) - whether the share is currently exported
* readonly (bool) - whether the share is exported read-only
* description (string) - optional human-readable description
* clients (list of strings) - IP address or subnet allow-list;
empty list means all hosts are permitted
Example::
{
"media": {
"name": "media",
"protocol": "nfs",
"path": "/mnt/tank/media",
"volume": "tank/media",
"enabled": True,
"readonly": False,
"description": "Media library",
"clients": ["192.168.1.0/24"],
},
"homes": {
"name": "homes",
"protocol": "smb",
"path": "/mnt/tank/homes",
"volume": "tank/homes",
"enabled": True,
"readonly": False,
"description": "User home directories",
"clients": [],
},
}
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Snapshots
# ------------------------------------------------------------------
def get_volume_snapshots(self, volume: str = "") -> List[VolumeSnapshotDict]:
"""
Returns volume / dataset snapshots.
:param volume: Restrict results to this volume path (e.g. ``"tank/media"``).
Pass an empty string (default) to list snapshots for all volumes.
Each entry contains:
* name (string) - snapshot name (e.g. ``"auto-2026-05-11"`` or full ``"tank/media@auto-2026-05-11"``)
* volume (string) - parent volume / dataset path
* created (float) - creation timestamp (Unix epoch)
* size (int) - bytes of unique data referenced only by this snapshot
* description (string) - optional description
* clones (list of strings) - volumes that were cloned from this snapshot
Example::
[
{
"name": "auto-2026-05-11",
"volume": "tank/media",
"created": 1746921600.0,
"size": 2097152,
"description": "Automatic daily snapshot",
"clones": [],
},
{
"name": "before-migration",
"volume": "tank/backups",
"created": 1746835200.0,
"size": 1073741824,
"description": "Snapshot before storage migration",
"clones": ["tank/backups-clone"],
},
]
"""
raise NotImplementedError
def snapshot_create(self, volume: str, name: str, description: str = "") -> None:
"""
Creates a snapshot of a logical volume or dataset.
:param volume: Volume / dataset path (e.g. ``"tank/media"``).
:param name: Name for the new snapshot.
:param description: Optional human-readable description.
:raises ValueError: If the volume does not exist, or a snapshot with that
name already exists.
:raises RuntimeError: If snapshot creation fails on the device side.
Example::
driver.snapshot_create("tank/media", "before-migration",
description="Snapshot before storage migration")
"""
raise NotImplementedError
def snapshot_delete(self, volume: str, name: str) -> None:
"""
Deletes a snapshot of a logical volume or dataset.
:param volume: Volume / dataset path.
:param name: Name of the snapshot to delete.
:raises ValueError: If the volume or snapshot does not exist.
:raises RuntimeError: If other clones depend on this snapshot
(delete or promote clones first).
Example::
driver.snapshot_delete("tank/media", "auto-2026-05-01")
"""
raise NotImplementedError
def snapshot_rollback(self, volume: str, name: str) -> None:
"""
Reverts a volume / dataset to a previously created snapshot.
All data written after the snapshot was taken is permanently discarded.
Any snapshots created after the target snapshot are also deleted.
:param volume: Volume / dataset path.
:param name: Name of the snapshot to roll back to.
:raises ValueError: If the volume or snapshot does not exist.
:raises RuntimeError: If the rollback fails (e.g. active clones block it).
Example::
driver.snapshot_rollback("tank/media", "before-migration")
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Quotas
# ------------------------------------------------------------------
def get_quotas(self) -> List[StorageQuotaDict]:
"""
Returns all filesystem quotas configured on the device.
Each entry contains:
* target (string) - username, group name, or dataset path depending on ``target_type``
* target_type (string) - ``"user"``, ``"group"``, or ``"dataset"``
* volume (string) - volume / dataset the quota applies to
* used (int) - bytes currently consumed by this target
* quota (int) - hard storage limit in bytes; ``0`` means no limit
* ref_quota (int) - referenced-data limit in bytes (excludes snapshots);
``0`` means no limit
Example::
[
{
"target": "alice",
"target_type": "user",
"volume": "tank/homes",
"used": 53687091200,
"quota": 107374182400,
"ref_quota": 0,
},
{
"target": "tank/backups",
"target_type": "dataset",
"volume": "tank/backups",
"used": 1099511627776,
"quota": 5497558138880,
"ref_quota": 0,
},
]
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Services
# ------------------------------------------------------------------
def get_services(self) -> Dict[str, StorageServiceDict]:
"""
Returns the file-sharing and access services available on the device.
Keys are service names (e.g. ``"nfs"``, ``"smb"``). Each value contains:
* name (string) - service name (repeated for convenience)
* enabled (bool) - administratively enabled (will start on next boot)
* running (bool) - currently active and listening
* port (int) - primary listening port; ``0`` if not applicable
* version (string) - protocol version string (e.g. ``"4.1"`` for NFSv4.1,
``"3.1.1"`` for SMB3); empty string if unknown
Example::
{
"nfs": {
"name": "nfs",
"enabled": True,
"running": True,
"port": 2049,
"version": "4.2",
},
"smb": {
"name": "smb",
"enabled": True,
"running": True,
"port": 445,
"version": "3.1.1",
},
"ftp": {
"name": "ftp",
"enabled": False,
"running": False,
"port": 21,
"version": "",
},
}
"""
raise NotImplementedError
def set_service_enabled(self, service: str, enabled: bool) -> None:
"""
Administratively enables or disables a file-sharing service.
Disabling stops the service immediately; enabling starts it immediately.
:param service: Service name (e.g. ``"nfs"``, ``"smb"``, ``"ftp"``).
:param enabled: ``True`` to start and enable; ``False`` to stop and disable.
:raises ValueError: If the service name is not recognised.
:raises RuntimeError: If the operation fails on the device side.
Example::
driver.set_service_enabled("ftp", False) # disable FTP
driver.set_service_enabled("nfs", True) # enable NFS
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Replication
# ------------------------------------------------------------------
def get_replication_jobs(self) -> List[ReplicationJobDict]:
"""
Returns all replication / sync jobs configured on the device.
Each entry contains:
* name (string) - job name
* source (string) - source path or dataset
* target (string) - destination path or dataset; may include a remote
host prefix (e.g. ``"backup-server:tank/replica"``)
* direction (string) - ``"push"`` (local → remote) or ``"pull"`` (remote → local)
* schedule (string) - cron expression or descriptive label (``"daily"``,
``"hourly"``, etc.)
* enabled (bool) - whether the job is scheduled to run
* last_run (float) - Unix epoch of the last run; ``0.0`` if never run
* last_status (string) - ``"success"``, ``"failed"``, ``"running"``, or ``"pending"``
* bytes_sent (int) - bytes transferred in the most recent run; ``0`` if never run
Example::
[
{
"name": "media-offsite",
"source": "tank/media",
"target": "backup-nas:tank/media-replica",
"direction": "push",
"schedule": "0 2 * * *",
"enabled": True,
"last_run": 1746921600.0,
"last_status": "success",
"bytes_sent": 1073741824,
},
{
"name": "backups-local",
"source": "tank/backups",
"target": "tank/backups-mirror",
"direction": "push",
"schedule": "hourly",
"enabled": True,
"last_run": 1746918000.0,
"last_status": "success",
"bytes_sent": 104857600,
},
]
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Package management (plugins / extensions)
# ------------------------------------------------------------------
def get_packages(self) -> List[PackageDict]:
"""
Returns all packages / plugins installed on the storage appliance
(e.g. TrueNAS SCALE Apps, Synology packages, OpenMediaVault plugins).
Each entry contains:
* name (string) - package name
* version (string) - installed or available version string
* installed (bool) - ``True`` if the package is currently installed
* description (string) - short package description
* size (int) - package size in bytes; ``0`` if unknown
* source (string) - repository or catalogue the package comes from
Example::
[
{
"name": "plex-media-server",
"version": "1.40.0",
"installed": True,
"description": "Plex Media Server",
"size": 134217728,
"source": "TrueNAS Community",
},
{
"name": "nextcloud",
"version": "28.0.3",
"installed": True,
"description": "Nextcloud – self-hosted file sync and share",
"size": 268435456,
"source": "TrueNAS Community",
},
]
"""
raise NotImplementedError
def install_package(self, name: str, version: str = "") -> None:
"""
Installs a package or plugin on the storage appliance.
:param name: Package name as known to the package catalogue.
:param version: Exact version to install. Empty string installs latest.
:raises NotImplementedError: If the driver does not support package management.
:raises ValueError: If the package name is unknown or the version unavailable.
:raises RuntimeError: If the installation fails on the device side.
Example::
driver.install_package("nextcloud")
"""
raise NotImplementedError
def remove_package(self, name: str) -> None:
"""
Removes an installed package from the storage appliance.
:param name: Package name to remove.
:raises NotImplementedError: If the driver does not support package management.
:raises ValueError: If the package is not currently installed.
:raises RuntimeError: If removal fails (e.g. required dependency).
Example::
driver.remove_package("plex-media-server")
"""
raise NotImplementedError
def get_package_config(self, name: str) -> Dict[str, Any]:
"""
Returns the current configuration of an installed package as a dictionary.
The structure is package-specific.
:param name: Package name.
:raises NotImplementedError: If the driver does not support package management.
:raises ValueError: If the package is not installed.
Example::
driver.get_package_config("nextcloud")
# →
{
"admin_user": "admin",
"trusted_domains": ["nas.corp.example"],
"mail_smtphost": "smtp.corp.example",
"maintenance_window_start": 2,
}
"""
raise NotImplementedError
def set_package_config(self, name: str, config: Dict[str, Any]) -> None:
"""
Writes a new configuration for an installed package.
:param name: Package name.
:param config: New configuration as a nested dictionary.
:raises NotImplementedError: If the driver does not support package management.
:raises ValueError: If the package is not installed or config is invalid.
:raises RuntimeError: If the device rejects the configuration.
Example::
driver.set_package_config(
"nextcloud",
{
"trusted_domains": ["nas.corp.example", "192.168.1.10"],
"maintenance_window_start": 3,
},
)
"""
raise NotImplementedError
napalm_device_types/switch.py
+298
View File
@@ -0,0 +1,298 @@
"""
Abstract base class for switch drivers.
Usage::
from napalm_device_types import SwitchDriver
class CiscoSGDriver(SwitchDriver):
def get_spanning_tree(self):
...
"""
from typing import Dict
from napalm.base import NetworkDriver
from napalm_device_types.models import (
Dot1XPortDict,
MACACLDict,
PoESummaryDict,
PortChannelDict,
SpanningTreeDict,
)
class SwitchDriver(NetworkDriver):
"""
Abstract intermediate driver for Ethernet switches.
Inherits all standard NAPALM NetworkDriver methods (including
``get_vlans()``, ``get_mac_address_table()``) and adds switch-specific
operations that concrete drivers must implement.
"""
def get_spanning_tree(self) -> Dict[str, SpanningTreeDict]:
"""
Returns spanning tree status for each STP instance.
Keys are STP instance identifiers (e.g. VLAN IDs for PVST,
``"MST0"`` for MSTP, or ``"0"`` for a single instance).
Each value contains:
* mode (string) - STP variant: ``"STP"``, ``"RSTP"``, ``"MSTP"``, ``"PVST"``
* root_bridge (bool) - whether this device is the root bridge
* root_id (string) - root bridge MAC address
* root_priority (int) - root bridge priority
* bridge_id (string) - this bridge's MAC address
* bridge_priority (int) - this bridge's priority
* interfaces (dict) - per-interface STP state:
* role (string) - ``"root"``, ``"designated"``, ``"alternate"``, ``"backup"``
* state (string) - ``"forwarding"``, ``"blocking"``, ``"learning"``, ``"listening"``
* cost (int) - port path cost
* port_priority (int) - port priority
Example::
{
"1": {
"mode": "RSTP",
"root_bridge": False,
"root_id": "00:11:22:33:44:55",
"root_priority": 4096,
"bridge_id": "AA:BB:CC:DD:EE:FF",
"bridge_priority": 32768,
"interfaces": {
"GigabitEthernet0/1": {
"role": "root",
"state": "forwarding",
"cost": 4,
"port_priority": 128,
},
"GigabitEthernet0/2": {
"role": "designated",
"state": "forwarding",
"cost": 4,
"port_priority": 128,
},
},
}
}
"""
raise NotImplementedError
def get_port_channels(self) -> Dict[str, PortChannelDict]:
"""
Returns port-channel (LAG) configuration and status.
Keys are port-channel interface names (e.g. ``"Port-Channel1"``).
Each value contains:
* members (list of strings) - names of member interfaces
* protocol (string) - aggregation protocol: ``"LACP"``, ``"PAgP"``, ``"static"``
* min_links (int) - minimum number of active members required
* is_up (bool) - whether the LAG is operationally up
Example::
{
"Port-Channel1": {
"members": ["GigabitEthernet0/1", "GigabitEthernet0/2"],
"protocol": "LACP",
"min_links": 1,
"is_up": True,
}
}
"""
raise NotImplementedError
def get_mac_acl(self) -> Dict[str, MACACLDict]:
"""
Returns the MAC-address-based access control lists configured per port.
Keys are interface names. Each value contains:
* name (string) - interface name (repeated for convenience)
* policy (string) - ACL mode:
* ``"allow"`` – whitelist: only listed MACs may use this port
* ``"deny"`` – blacklist: listed MACs are blocked
* ``"disabled"`` – no MAC filtering active
* entries (list) - ACL entries, each with:
* mac (string) - MAC address (normalised, colon-separated)
* action (string) - ``"allow"`` or ``"deny"``
* description (string) - optional human-readable label
Example::
{
"GigabitEthernet0/1": {
"name": "GigabitEthernet0/1",
"policy": "allow",
"entries": [
{"mac": "AA:BB:CC:DD:EE:01", "action": "allow", "description": "printer"},
],
},
"GigabitEthernet0/2": {
"name": "GigabitEthernet0/2",
"policy": "disabled",
"entries": [],
},
}
"""
raise NotImplementedError
def get_dot1x_config(self) -> Dict[str, Dot1XPortDict]:
"""
Returns the 802.1X / NAC configuration per switch port.
Keys are interface names. Each value contains:
* enabled (bool) - whether 802.1X is active on this port
* port_control (string) - authentication mode:
* ``"auto"`` – port authenticates normally
* ``"force-authorized"`` – port always passes traffic (bypass)
* ``"force-unauthorized"`` – port always blocks traffic
* host_mode (string) - how many identities are authenticated per port:
* ``"single-host"`` – one device, then port is locked
* ``"multi-host"`` – first auth unlocks port for all devices
* ``"multi-domain"`` – one data + one voice device (IP phone scenario)
* ``"multi-auth"`` – each device authenticates individually
* auth_server (dict) - RADIUS authentication server (host, port, timeout, retries)
* acct_server (dict or None) - RADIUS accounting server, ``None`` if unused
* reauthentication (bool) - whether periodic re-authentication is enabled
* reauth_interval (int) - re-authentication interval in seconds (0 = disabled)
* guest_vlan (int) - VLAN ID for unauthenticated clients (0 = disabled)
* auth_fail_vlan (int) - VLAN ID for clients that fail authentication (0 = disabled)
Note: RADIUS shared secrets are intentionally omitted.
Example::
{
"GigabitEthernet0/1": {
"enabled": True,
"port_control": "auto",
"host_mode": "multi-domain",
"auth_server": {
"host": "radius.corp.example",
"port": 1812,
"timeout": 5,
"retries": 3,
},
"acct_server": None,
"reauthentication": True,
"reauth_interval": 3600,
"guest_vlan": 99,
"auth_fail_vlan": 999,
},
}
"""
raise NotImplementedError
def get_poe_status(self) -> PoESummaryDict:
"""
Returns the PoE status of the switch as a whole and per port.
The returned dictionary contains:
* total_power_budget (float) - total PoE power available in watts
* total_power_draw (float) - total PoE power currently consumed in watts
* ports (dict) - per-interface PoE state, keyed by interface name:
* enabled (bool) - whether PoE is configured on this port
* status (string) - operational state:
* ``"delivering"`` – power is being delivered to a PD
* ``"searching"`` – port is looking for a powered device
* ``"fault"`` – an error condition was detected
* ``"disabled"`` – PoE is administratively off
* ``"denied"`` – PD detected but power budget exceeded
* poe_class (string) - IEEE 802.3 class: ``"Class 0"`` … ``"Class 8"``
(``"unknown"`` if not yet negotiated)
* power_draw (float) - current power consumption in watts
* power_budget (float) - per-port power limit in watts
* voltage (float) - measured port voltage in volts
* current (float) - measured port current in milliamps
Example::
{
"total_power_budget": 740.0,
"total_power_draw": 43.2,
"ports": {
"GigabitEthernet0/1": {
"enabled": True,
"status": "delivering",
"poe_class": "Class 3",
"power_draw": 12.4,
"power_budget": 30.0,
"voltage": 53.5,
"current": 231.0,
},
"GigabitEthernet0/2": {
"enabled": True,
"status": "searching",
"poe_class": "unknown",
"power_draw": 0.0,
"power_budget": 30.0,
"voltage": 0.0,
"current": 0.0,
},
"GigabitEthernet0/3": {
"enabled": False,
"status": "disabled",
"poe_class": "unknown",
"power_draw": 0.0,
"power_budget": 0.0,
"voltage": 0.0,
"current": 0.0,
},
},
}
"""
raise NotImplementedError
def set_poe_enabled(self, interface: str, enabled: bool) -> None:
"""
Administratively enables or disables PoE on a single port.
:param interface: Interface name (e.g. ``"GigabitEthernet0/1"``).
:param enabled: ``True`` to enable PoE, ``False`` to disable it.
:raises NotImplementedError: If the driver does not support PoE control.
:raises ValueError: If the interface does not exist or does not support PoE.
Example::
driver.set_poe_enabled("GigabitEthernet0/1", False) # cut power
driver.set_poe_enabled("GigabitEthernet0/1", True) # restore
"""
raise NotImplementedError
def power_cycle_port(self, interface: str, delay: int = 5) -> None:
"""
Power-cycles the PoE port: cuts power, waits ``delay`` seconds, then
restores power. Useful for rebooting a hung IP camera, AP, or IP phone
without physical access.
The method blocks until the full cycle (off → wait → on) is complete.
After it returns the port is back in the delivering/searching state.
:param interface: Interface name (e.g. ``"GigabitEthernet0/1"``).
:param delay: Seconds to keep the port powered off (default: 5).
:raises NotImplementedError: If the driver does not support PoE control.
:raises ValueError: If the interface does not exist, does not support PoE,
or PoE is administratively disabled on the port.
Example::
driver.power_cycle_port("GigabitEthernet0/1") # 5 s off
driver.power_cycle_port("GigabitEthernet0/1", delay=15) # 15 s off
"""
raise NotImplementedError