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

feat!: role bases declare their methods instead of stubbing them

Browse Source 
A role base used to fill its methods with `raise NotImplementedError`. That is
not neutral under multiple inheritance: the placeholder wins the MRO against a
sibling base's working implementation and silently replaces it. Adding one stub
to a base was therefore a breaking change for every driver mixing that base with
another, and it broke three of them — OpenWrt grew seven forwarding methods,
QNAP one, and OpenMediaVault avoided inheriting StorageDriver at all.

Role bases now declare their surface under `if TYPE_CHECKING` and implement
nothing. There is no longer anything to shadow, so a device can finally say what
it is:

    class QnapQtsDriver(StorageDriver, HypervisorDriver, LinuxDriver):

The order of those bases is the ranking, read back by roles_of(),
role_keys_of() and primary_role_of() in the new roles module. Nothing restates
it: no precedence table, no attribute to override.

Two consequences, both wanted. `hasattr` is a truthful capability probe again,
because a method exists exactly when a driver provided it. And a method that was
never implemented now raises AttributeError rather than NotImplementedError, so
callers should ask before calling.

Shared behaviour moves out of the roles and into function classes, each holding
it once: PackageManagementMixin (was five byte-identical copies),
HealthMetricsMixin (five), ServiceControlMixin, UpdateMixin, NatVpnMixin,
MacAclMixin, FirewallRuleMixin, InterfaceFilterMixin.

BREAKING CHANGE: methods whose contract genuinely differed were renamed apart —
StorageDriver.get_services -> get_storage_services, the storage and hypervisor
snapshot writers -> create/delete/rollback_{volume,vm}_snapshot,
HypervisorDriver.get_storage -> get_vm_storage_pools, get_snapshots ->
get_vm_snapshots, SwitchDriver.get_dot1x_config -> get_dot1x_ports. Two
duplicate names collapsed onto the one already in use: get_pending_updates ->
get_available_updates and remove_package -> uninstall_package.

Also fixes __doc__ being None on all seven role bases: TYPE_LABEL was assigned
above the triple-quoted string, which made it a bare expression rather than a
docstring.
This commit is contained in:
christianmanivong
2026-08-21 12:49:45 +07:00
parent ec0612b300
commit d8dbc7a442
26 changed files with 3119 additions and 3084 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+60
View File
@@ -22,6 +22,66 @@ NAPALM's `NetworkDriver` defines a common interface for all network devices. In
`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**:
```python
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`:
```python
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`.
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
napalm_device_types/__init__.py
+50 -14
View File
@@ -4,16 +4,22 @@ 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::
A driver inherits one base per role its device fills, and **the order of those
bases is the ranking** -- the first one is what netOrk shows as the device's
class::
from napalm_device_types import AccessPointDriver
from napalm_device_types import StorageDriver, HypervisorDriver
class OpenWrtDriver(AccessPointDriver):
...
class QnapQtsDriver(StorageDriver, HypervisorDriver, LinuxDriver):
... # a NAS that also runs VMs on a Linux userland
Available base classes:
That works because a role base *declares* its methods (under ``if
TYPE_CHECKING``) and implements none of them. Nothing exists at runtime until a
concrete driver provides it, so no base can shadow a working implementation
inherited from a sibling, and ``hasattr`` is a truthful answer to "can this
driver do X".
Role bases -- what a device *is*:
* :class:`~napalm_device_types.access_point.AccessPointDriver`
* :class:`~napalm_device_types.switch.SwitchDriver`
@@ -23,14 +29,24 @@ Available base classes:
* :class:`~napalm_device_types.storage.StorageDriver`
* :class:`~napalm_device_types.residential_gateway.ResidentialGatewayDriver`
Also provided:
Function classes -- what a device *can do*. Shared behaviour lives here once
instead of being restated on every role that happens to need it:
* :class:`~napalm_device_types.config_lifecycle.ConfigLifecycleMixin` --
stand-alone mixin to reduce duplication of config lifecycle methods across
drivers.
* :class:`~napalm_device_types.dhcp.DhcpServerMixin` -- static DHCP
reservation read/diff/apply, mixed into the firewall and gateway base
classes.
* :class:`~napalm_device_types.config_lifecycle.ConfigLifecycleMixin`
* :class:`~napalm_device_types.dhcp.DhcpServerMixin`
* :class:`~napalm_device_types.firewall_rules.FirewallRuleMixin`
* :class:`~napalm_device_types.health_metrics.HealthMetricsMixin`
* :class:`~napalm_device_types.interface_filter.InterfaceFilterMixin`
* :class:`~napalm_device_types.mac_acl.MacAclMixin`
* :class:`~napalm_device_types.nat_vpn.NatVpnMixin`
* :class:`~napalm_device_types.packages.PackageManagementMixin`
* :class:`~napalm_device_types.ping_sweep.PingSweepMixin`
* :class:`~napalm_device_types.services.ServiceControlMixin`
* :class:`~napalm_device_types.updates.UpdateMixin`
Introspection -- :func:`~napalm_device_types.roles.roles_of`,
:func:`~napalm_device_types.roles.role_keys_of` and
:func:`~napalm_device_types.roles.primary_role_of`.
"""
from napalm_device_types.base import DeviceTypeDriver, FingerprintRule, PortSpec
@@ -40,7 +56,16 @@ from napalm_device_types.dhcp import DhcpServerMixin, normalize_cidr, normalize_
from napalm_device_types.firewall import FirewallDriver
from napalm_device_types.hypervisor import HypervisorDriver
from napalm_device_types.os import OSDriver
from napalm_device_types.firewall_rules import FirewallRuleMixin
from napalm_device_types.health_metrics import HealthMetricsMixin
from napalm_device_types.interface_filter import InterfaceFilterMixin
from napalm_device_types.mac_acl import MacAclMixin
from napalm_device_types.nat_vpn import NatVpnMixin
from napalm_device_types.packages import PackageManagementMixin
from napalm_device_types.ping_sweep import PingSweepMixin, driver_supports_ping
from napalm_device_types.roles import primary_role_of, role_keys_of, roles_of
from napalm_device_types.services import ServiceControlMixin
from napalm_device_types.updates import UpdateMixin
from napalm_device_types.residential_gateway import ResidentialGatewayDriver
from napalm_device_types.storage import StorageDriver
from napalm_device_types.switch import SwitchDriver
@@ -52,14 +77,25 @@ __all__ = [
"DhcpServerMixin",
"FingerprintRule",
"FirewallDriver",
"FirewallRuleMixin",
"HealthMetricsMixin",
"HypervisorDriver",
"InterfaceFilterMixin",
"MacAclMixin",
"NatVpnMixin",
"OSDriver",
"PackageManagementMixin",
"PingSweepMixin",
"PortSpec",
"ResidentialGatewayDriver",
"ServiceControlMixin",
"StorageDriver",
"SwitchDriver",
"UpdateMixin",
"driver_supports_ping",
"normalize_cidr",
"normalize_mac",
"primary_role_of",
"role_keys_of",
"roles_of",
]
napalm_device_types/access_point.py
+192 -459
View File
@@ -10,30 +10,25 @@ Usage::
...
"""
from typing import Any, Dict, List
from typing import Any, Dict, List, TYPE_CHECKING
from napalm_device_types.base import DeviceTypeDriver
from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
from napalm_device_types.mac_acl import MacAclMixin
from napalm_device_types.updates import UpdateMixin
from napalm_device_types.services import ServiceControlMixin
from napalm_device_types.packages import PackageManagementMixin
from napalm_device_types.health_metrics import HealthMetricsMixin
from napalm_device_types.interface_filter import InterfaceFilterMixin
from napalm_device_types.models import (
ChannelScanEntryDict,
Dot1XConfigDict,
FastTransitionConfigDict,
HealthMetricsDict,
MACACLDict,
MeshConfigDict,
MeshPeerDict,
PackageDict,
RadioStatusDict,
ServiceDict,
SSIDBridgeDict,
SSIDDict,
UpdateDict,
WirelessClientDict,
WirelessConfigDict,
)
class AccessPointDriver(DeviceTypeDriver):
TYPE_LABEL: str = "Access Point"
class AccessPointDriver(MacAclMixin, UpdateMixin, ServiceControlMixin, PackageManagementMixin, HealthMetricsMixin, InterfaceFilterMixin, DeviceTypeDriver):
"""
Abstract intermediate driver for wireless access points.
@@ -41,33 +36,16 @@ class AccessPointDriver(DeviceTypeDriver):
access-point-specific operations that concrete drivers must implement.
"""
# Interfaces that carry no operational meaning on an access point and
# should be excluded from get_interfaces() / get_facts() interface_list.
_EXCLUDED_INTERFACES: frozenset = frozenset({"lo"})
#: Stable key netOrk exposes as ``device_class``. The order in which a
#: driver lists its role bases is the ranking; see
#: :func:`napalm_device_types.roles.primary_role_of`.
ROLE: str = "access_point"
TYPE_LABEL: str = "Access Point"
# Interface name *prefixes* to exclude (e.g. Linux phy* are raw radio
# devices and have no IP/Ethernet significance at the AP level).
_EXCLUDED_INTERFACE_PREFIXES: tuple = ("phy",)
_SNMP_SKIP_IF = IF_SKIP_DEFAULT
_SNMP_TX_ERR_IS_DROP: bool = False
@classmethod
async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
return await collect_ucd_metrics(
snmp_get, snmp_walk,
tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
if_skip=cls._SNMP_SKIP_IF,
)
def _filter_interfaces(self, interfaces: Dict[str, Any]) -> Dict[str, Any]:
"""Remove loopback and radio-device (phy*) interfaces from an interface dict."""
return {
name: data
for name, data in interfaces.items()
if name not in self._EXCLUDED_INTERFACES
and not name.startswith(self._EXCLUDED_INTERFACE_PREFIXES)
}
# AP-specific methods (get_wireless_clients, get_ssids, get_radio_status,
# get_interfaces, get_vlans, …) are intentionally NOT defined here.
@@ -77,470 +55,225 @@ class AccessPointDriver(DeviceTypeDriver):
# all standard NAPALM methods, so AccessPointDriver must come LAST to avoid
# shadowing the mixin implementations.
def get_wireless_config(self) -> WirelessConfigDict:
"""
Returns global wireless configuration parameters that apply across
all radios and SSIDs.
if TYPE_CHECKING:
* 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
def get_wireless_config(self) -> WirelessConfigDict:
"""
Returns global wireless configuration parameters that apply across
all radios and SSIDs.
Example::
* 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
{
"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
Example::
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,
"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_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.
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:
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
* 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::
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
{
"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,
}
}
"""
...
def get_mac_acl(self) -> Dict[str, MACACLDict]:
"""
Returns the MAC-address-based access control lists configured per SSID.
def get_mesh_config(self) -> Dict[str, MeshConfigDict]:
"""
Returns the 802.11s mesh configuration per mesh interface.
Keys are SSID names. Each value contains:
Keys are mesh interface names (e.g. ``"mesh0"``). Each value contains:
* ssid (string) - SSID name (repeated for convenience)
* policy (string) - ACL mode:
* 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"``
* ``"allow"`` – whitelist: only listed MACs may associate
* ``"deny"`` – blacklist: listed MACs are blocked
* ``"disabled"`` – no MAC filtering active
Example::
* entries (list) - ACL entries, each with:
{
"mesh0": {
"enabled": True,
"radio": "radio1",
"mesh_id": "office-mesh",
"path_metric": "airtime",
"gate_announcements": True,
"is_gate": True,
"encryption": "SAE",
}
}
"""
...
* mac (string) - MAC address (normalised, colon-separated)
* action (string) - ``"allow"`` or ``"deny"``
* description (string) - optional human-readable label
def get_mesh_peers(self) -> List[MeshPeerDict]:
"""
Returns a list of currently active 802.11s mesh peers.
Example::
Each entry contains:
{
"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
* 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)
def push_mac_acl(self, ssid_name: str, mode: str, macs: List[str]) -> None:
"""
Rewrites the MAC-address access control list for a single SSID.
Example::
Full-rebuild semantics: replaces whatever ACL state currently exists
for *ssid_name* with *mode* + *macs* — not a diff/patch.
[
{
"mac": "AA:BB:CC:DD:EE:01",
"radio": "radio1",
"signal": -58,
"tx_rate": 300.0,
"rx_rate": 270.0,
"uptime": 7200,
"hop_count": 1,
}
]
"""
...
:param ssid_name: SSID name to apply the ACL to.
:param mode: ``"off"`` | ``"whitelist"`` | ``"blacklist"``.
:param macs: MAC addresses for the active list. Ignored when ``mode == "off"``.
:raises NotImplementedError: If the driver does not support MAC ACL push.
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.
Example::
Keys are SSID names. Each value contains:
driver.push_mac_acl("CorpWiFi", "whitelist", ["AA:BB:CC:DD:EE:01", "AA:BB:CC:DD:EE:02"])
driver.push_mac_acl("GuestNet", "off", [])
"""
raise NotImplementedError
* 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
def get_dot1x_config(self) -> Dict[str, Dot1XConfigDict]:
"""
Returns the 802.1X / WPA-Enterprise (RADIUS) configuration per SSID.
Example::
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,
{
"CorpWiFi": {
"ssid": "CorpWiFi",
"bridge": "br-corp",
"vlan_id": 10,
"tagged": True,
"client_isolation": False,
},
"acct_server": {
"host": "radius.corp.example",
"port": 1813,
"timeout": 5,
"retries": 3,
"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,
},
"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
def get_dot1x_config(self) -> Dict[str, Dot1XConfigDict]:
"""
Returns the 802.1X / WPA-Enterprise (RADIUS) configuration per SSID.
Example::
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::
[
{
"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
"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,
}
}
"""
...
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 get_services(self) -> List[ServiceDict]:
"""
Returns the list of system services known to the device's init system.
Each entry contains:
* name (string) - service name as registered with the init system
* running (bool) - ``True`` if the service process is currently running
* enabled (bool) - ``True`` if the service starts automatically at boot
* pid (int) - process ID of the main service process; 0 if not running
Example::
[
{"name": "lldpd", "running": True, "enabled": True, "pid": 2341},
{"name": "sshd", "running": True, "enabled": True, "pid": 1198},
{"name": "cron", "running": False, "enabled": False, "pid": 0},
]
"""
raise NotImplementedError
def manage_service(self, name: str, action: str) -> Dict[str, Any]:
"""
Execute a lifecycle action on a named service.
:param name: Service name as returned by :meth:`get_services`.
:param action: One of ``start``, ``stop``, ``restart``, ``enable``, ``disable``.
:returns: ``{"success": bool, "output": str}``
:raises ValueError: If ``name`` or ``action`` is invalid.
:raises NotImplementedError: If the driver does not support service management.
"""
raise NotImplementedError
def get_available_updates(self) -> List[UpdateDict]:
"""
Returns the list of installed packages that have a newer version available.
Uses the local package manager cache — does not run ``opkg update`` / ``apk update``.
:returns: List of :class:`~napalm_device_types.models.UpdateDict`.
:raises NotImplementedError: If the driver does not support update listing.
Example::
[
{"name": "busybox", "current_version": "1.36.1-1", "new_version": "1.37.0-1"},
{"name": "dropbear", "current_version": "2022.83-2", "new_version": "2024.86-1"},
]
"""
raise NotImplementedError
def apply_updates(self, packages: List[str]) -> Dict[str, Any]:
"""
Upgrade one or more packages to their newest available version.
:param packages: List of package names to upgrade.
:returns: ``{"success": bool, "output": str}``
:raises NotImplementedError: If the driver does not support package upgrades.
"""
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/config_lifecycle.py
+7 -5
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
import difflib
from typing import ClassVar
from typing import ClassVar, TYPE_CHECKING
from napalm.base.exceptions import (
CommandErrorException,
@@ -39,11 +39,13 @@ class ConfigLifecycleMixin:
_comment_chars: ClassVar[tuple[str, ...]] = ("!", "#")
def _get_running_config(self) -> str:
raise NotImplementedError
if TYPE_CHECKING:
def commit_config(self, message: str = "", revert_in: int | None = None) -> None:
raise NotImplementedError
def _get_running_config(self) -> str:
...
def commit_config(self, message: str = "", revert_in: int | None = None) -> None:
...
def load_merge_candidate(
self, filename: str | None = None, config: str | None = None
napalm_device_types/dhcp.py
+78 -76
View File
@@ -24,7 +24,7 @@ takes DHCP down for an entire VLAN.
from __future__ import annotations
import re
from typing import Any, Dict, Iterator, List, Optional
from typing import Any, Dict, Iterator, List, Optional, TYPE_CHECKING
from napalm_device_types.models import (
DhcpReservationDiffDict,
@@ -121,100 +121,102 @@ class DhcpServerMixin:
# Device-specific -- must be implemented by the concrete driver.
# ------------------------------------------------------------------
def get_dhcp_reservations(self) -> List[DhcpReservationDict]:
"""
Returns all static DHCP reservations currently configured on the
device, across all subnets.
if TYPE_CHECKING:
This is the *configured* state, not the observed leases -- see
``get_dhcp_leases()`` for the latter.
def get_dhcp_reservations(self) -> List[DhcpReservationDict]:
"""
Returns all static DHCP reservations currently configured on the
device, across all subnets.
:raises NotImplementedError: If the driver does not support reading
DHCP reservations.
"""
raise NotImplementedError
This is the *configured* state, not the observed leases -- see
``get_dhcp_leases()`` for the latter.
def apply_dhcp_reservation(
self, reservation: DhcpReservationDict, *, uuid: Optional[str] = None
) -> Dict[str, Any]:
"""
Creates or updates a single static DHCP reservation on the device.
:raises NotImplementedError: If the driver does not support reading
DHCP reservations.
"""
...
:param reservation: The desired reservation state, vendor-neutral.
:param uuid: If given, update the existing reservation with this ID
in-place. If ``None``, create a new one.
:raises NotImplementedError: If the driver does not support writing
DHCP reservations.
:raises ValueError: If `reservation` names a subnet the device does
not serve.
:raises RuntimeError: If the device rejects the write.
def apply_dhcp_reservation(
self, reservation: DhcpReservationDict, *, uuid: Optional[str] = None
) -> Dict[str, Any]:
"""
Creates or updates a single static DHCP reservation on the device.
:returns: A dict with at least ``{"success": bool}``.
"""
raise NotImplementedError
:param reservation: The desired reservation state, vendor-neutral.
:param uuid: If given, update the existing reservation with this ID
in-place. If ``None``, create a new one.
:raises NotImplementedError: If the driver does not support writing
DHCP reservations.
:raises ValueError: If `reservation` names a subnet the device does
not serve.
:raises RuntimeError: If the device rejects the write.
def get_dhcp_subnets(self) -> List[DhcpSubnetDict]:
"""
Returns every DHCPv4 subnet the device serves, with its pools and
per-subnet options.
:returns: A dict with at least ``{"success": bool}``.
"""
...
:raises NotImplementedError: If the driver does not support reading
DHCP subnets.
"""
raise NotImplementedError
def get_dhcp_subnets(self) -> List[DhcpSubnetDict]:
"""
Returns every DHCPv4 subnet the device serves, with its pools and
per-subnet options.
def apply_dhcp_subnet(
self, subnet: DhcpSubnetDict, *, uuid: Optional[str] = None
) -> Dict[str, Any]:
"""
Creates or updates a single DHCPv4 subnet on the device.
:raises NotImplementedError: If the driver does not support reading
DHCP subnets.
"""
...
Implementations must treat ``subnet["option_data"]`` as a partial
update: an option the caller did not name is left as the device has
it. Managing `domain_search` alone is the common case, and it must
not silently drop the `routers` the server autocollected.
def apply_dhcp_subnet(
self, subnet: DhcpSubnetDict, *, uuid: Optional[str] = None
) -> Dict[str, Any]:
"""
Creates or updates a single DHCPv4 subnet on the device.
:param subnet: The desired subnet state, vendor-neutral.
:param uuid: If given, update the existing subnet with this ID
in-place. If ``None``, create a new one.
:raises NotImplementedError: If the driver does not support writing
DHCP subnets.
:raises RuntimeError: If the device rejects the write.
Implementations must treat ``subnet["option_data"]`` as a partial
update: an option the caller did not name is left as the device has
it. Managing `domain_search` alone is the common case, and it must
not silently drop the `routers` the server autocollected.
:returns: A dict with at least ``{"success": bool}``.
"""
raise NotImplementedError
:param subnet: The desired subnet state, vendor-neutral.
:param uuid: If given, update the existing subnet with this ID
in-place. If ``None``, create a new one.
:raises NotImplementedError: If the driver does not support writing
DHCP subnets.
:raises RuntimeError: If the device rejects the write.
def commit_dhcp_subnets(self) -> Dict[str, Any]:
"""
Applies pending subnet changes (e.g. Kea's ``service/reconfigure``).
:returns: A dict with at least ``{"success": bool}``.
"""
...
Separate from ``commit_dhcp_reservations`` even where a driver
implements both with the same call: the two desired-state sets are
applied independently, and a caller that changed only subnets should
not have to know which reload the vendor happens to share.
def commit_dhcp_subnets(self) -> Dict[str, Any]:
"""
Applies pending subnet changes (e.g. Kea's ``service/reconfigure``).
:raises NotImplementedError: If the driver does not support this.
Separate from ``commit_dhcp_reservations`` even where a driver
implements both with the same call: the two desired-state sets are
applied independently, and a caller that changed only subnets should
not have to know which reload the vendor happens to share.
:returns: A dict with at least ``{"success": bool}``.
"""
raise NotImplementedError
:raises NotImplementedError: If the driver does not support this.
def commit_dhcp_reservations(self) -> Dict[str, Any]:
"""
Applies pending reservation changes (e.g. Kea's ``service/reconfigure``,
or a dnsmasq reload).
:returns: A dict with at least ``{"success": bool}``.
"""
...
Call once after one or more `apply_dhcp_reservation()` calls -- not
after every single reservation, and not at all when nothing changed:
on most implementations this reloads the DHCP daemon.
def commit_dhcp_reservations(self) -> Dict[str, Any]:
"""
Applies pending reservation changes (e.g. Kea's ``service/reconfigure``,
or a dnsmasq reload).
:raises NotImplementedError: If the driver does not support this
(e.g. reservations take effect immediately on write).
Call once after one or more `apply_dhcp_reservation()` calls -- not
after every single reservation, and not at all when nothing changed:
on most implementations this reloads the DHCP daemon.
:returns: A dict with at least ``{"success": bool}``.
"""
raise NotImplementedError
:raises NotImplementedError: If the driver does not support this
(e.g. reservations take effect immediately on write).
:returns: A dict with at least ``{"success": bool}``.
"""
...
# ------------------------------------------------------------------
# Generic, vendor-neutral algorithms.
napalm_device_types/firewall.py
+109 -433
View File
@@ -10,39 +10,21 @@ Usage::
...
"""
from typing import Any, Dict, Iterator, List, Optional
from typing import Any, ClassVar, Dict, Iterator, List, Optional, TYPE_CHECKING
from napalm_device_types.base import DeviceTypeDriver
from napalm_device_types.nat_vpn import NatVpnMixin
from napalm_device_types.packages import PackageManagementMixin
from napalm_device_types.health_metrics import HealthMetricsMixin
from napalm_device_types.dhcp import DhcpServerMixin
from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
from napalm_device_types.firewall_rules import FirewallRuleMixin
from napalm_device_types.models import (
FirewallRuleDict,
FirewallRuleDiffDict,
FirewallRuleUpdateDict,
HealthMetricsDict,
NATTranslationDict,
PackageDict,
SecurityZoneDict,
SessionDict,
VPNTunnelDict,
)
_FIREWALL_RULE_COMPARE_FIELDS = (
"action",
"interface",
"direction",
"protocol",
"source_net",
"source_port",
"destination_net",
"destination_port",
"log",
"quick",
"enabled",
)
class FirewallDriver(DhcpServerMixin, DeviceTypeDriver):
TYPE_LABEL: str = "Firewall"
class FirewallDriver(NatVpnMixin, PackageManagementMixin, HealthMetricsMixin, FirewallRuleMixin, DhcpServerMixin, DeviceTypeDriver):
"""
Abstract intermediate driver for firewall/security devices.
@@ -51,434 +33,128 @@ class FirewallDriver(DhcpServerMixin, DeviceTypeDriver):
that concrete drivers must implement.
"""
_SNMP_SKIP_IF = IF_SKIP_DEFAULT
#: Stable key netOrk exposes as ``device_class``. The order in which a
#: driver lists its role bases is the ranking; see
#: :func:`napalm_device_types.roles.primary_role_of`.
ROLE: str = "firewall"
TYPE_LABEL: str = "Firewall"
# OPNsense reports drops in the out-error counter.
_SNMP_TX_ERR_IS_DROP: bool = True
_SNMP_TX_ERR_IS_DROP: ClassVar[bool] = True
@classmethod
async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
return await collect_ucd_metrics(
snmp_get, snmp_walk,
tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
if_skip=cls._SNMP_SKIP_IF,
)
def get_nat_translations(self) -> List[NATTranslationDict]:
"""
Returns a list of active NAT translation entries.
if TYPE_CHECKING:
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
def get_security_zones(self) -> Dict[str, SecurityZoneDict]:
"""
Returns the security zone configuration.
Example::
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::
[
{
"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,
"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_security_zones(self) -> Dict[str, SecurityZoneDict]:
"""
Returns the security zone configuration.
def get_sessions(self) -> List[SessionDict]:
"""
Returns a list of active connection sessions (stateful flows).
Keys are zone names. Each value contains:
Each entry 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
* 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::
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 send_wake_on_lan(self, mac_address: str, interface: str = "") -> Dict[str, Any]:
"""
Sends a Wake-on-LAN "magic packet" to wake a host on the network.
:param mac_address: Target host's MAC address (colon-separated,
case-insensitive, e.g. ``"AA:BB:CC:DD:EE:FF"``).
:param interface: Driver-specific interface identifier to broadcast the
magic packet from. Required by drivers that scope WOL per interface
(e.g. OPNsense); an empty string means "use the driver's default/
only broadcast domain." Consult the concrete driver's docstring for
the exact expected format.
:raises NotImplementedError: If the driver does not support Wake-on-LAN.
:raises ValueError: If ``mac_address`` is malformed, or ``interface`` is
required by this driver but was not provided.
:returns: A dict with:
* success (bool) - ``True`` if the magic packet was sent without error
* output (string) - human-readable status message
Example::
driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
# → {"success": True, "output": "Magic packet sent to AA:BB:CC:DD:EE:FF via lan"}
.. note::
A driver whose ``interface`` is *not* the name :meth:`get_interfaces`
is keyed by must expose the name it does expect as an ``identifier``
key on each ``get_interfaces()`` entry. Without it a caller has no
way to offer a valid choice: OPNsense, for instance, keys interfaces
by the physical device ("em0") but wakes by the assigned name
("lan"), and rejects the former. ``identifier`` is a non-standard
NAPALM key, so it reaches consumers through the usual passthrough
for extra interface data.
"""
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
# ------------------------------------------------------------------
# Firewall rule diff/apply. get_firewall_rules/apply_firewall_rule/
# commit_firewall_rules are abstract (device communication); everything
# else here is a concrete, vendor-neutral algorithm -- see README.md
# "Design principle: generic vs. device-specific logic".
# ------------------------------------------------------------------
def get_firewall_rules(self) -> List[FirewallRuleDict]:
"""
Returns all firewall filter rules currently configured on the device.
`description` must be a stable, human-assigned identifier -- it is
the key used to match rules across calls (most firewall vendors
don't expose an ID a caller can pre-assign).
:raises NotImplementedError: If the driver does not support reading
firewall rules.
"""
raise NotImplementedError
def apply_firewall_rule(
self, rule: FirewallRuleDict, *, uuid: Optional[str] = None
) -> Dict[str, Any]:
"""
Creates or updates a single firewall filter rule on the device.
:param rule: The desired rule state, in vendor-neutral form.
:param uuid: If given, update the existing rule with this ID
in-place. If ``None``, create a new rule.
:raises NotImplementedError: If the driver does not support writing
firewall rules.
:raises ValueError: If `rule` references an alias/interface the
device doesn't know about.
:raises RuntimeError: If the device rejects the write.
:returns: A dict with at least ``{"success": bool}``.
"""
raise NotImplementedError
def commit_firewall_rules(self) -> Dict[str, Any]:
"""
Applies pending firewall filter rule changes (e.g. reloads pf/pfctl,
or whatever the device's equivalent of "Apply Changes" is).
Call once after one or more `apply_firewall_rule()` calls -- not
after every single rule.
:raises NotImplementedError: If the driver does not support this
(e.g. rules take effect immediately on write).
:returns: A dict with at least ``{"success": bool}``.
"""
raise NotImplementedError
def diff_firewall_rules(self, desired: List[FirewallRuleDict]) -> FirewallRuleDiffDict:
"""
Compares `desired` against the device's current rules and returns
what would need to change to reach that state.
Matches rules by `description`. A desired rule with no live
counterpart becomes an "add"; a live rule whose description matches
but whose other fields differ becomes an "update". Live rules with
no matching desired entry are **not** reported for deletion -- this
is intentionally conservative: a firewall may carry manually-created
or otherwise unmanaged rules that a caller's `desired` set was never
meant to describe, and this method has no way to distinguish those
from ones simply no longer wanted. Callers wanting delete/cleanup
semantics must implement that themselves, deliberately.
:param desired: The complete desired rule set.
:returns: ``{"add": [...], "update": [{"uuid", "rule",
"changed_fields"}, ...]}``.
"""
live_by_description: Dict[str, FirewallRuleDict] = {
rule["description"]: rule for rule in self.get_firewall_rules()
}
add: List[FirewallRuleDict] = []
update: List[FirewallRuleUpdateDict] = []
for desired_rule in desired:
live_rule = live_by_description.get(desired_rule["description"])
if live_rule is None:
add.append(desired_rule)
continue
changed_fields = [
field
for field in _FIREWALL_RULE_COMPARE_FIELDS
if live_rule.get(field) != desired_rule.get(field)
]
if changed_fields:
update.append(
[
{
"uuid": live_rule["uuid"],
"rule": desired_rule,
"changed_fields": changed_fields,
"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,
}
)
]
"""
...
return {"add": add, "update": update}
def apply_firewall_ruleset(self, desired: List[FirewallRuleDict]) -> Iterator[str]:
"""
Computes the diff against `desired` and applies it, yielding one
human-readable progress line per change, then commits.
def send_wake_on_lan(self, mac_address: str, interface: str = "") -> Dict[str, Any]:
"""
Sends a Wake-on-LAN "magic packet" to wake a host on the network.
:param mac_address: Target host's MAC address (colon-separated,
case-insensitive, e.g. ``"AA:BB:CC:DD:EE:FF"``).
:param interface: Driver-specific interface identifier to broadcast the
magic packet from. Required by drivers that scope WOL per interface
(e.g. OPNsense); an empty string means "use the driver's default/
only broadcast domain." Consult the concrete driver's docstring for
the exact expected format.
:raises NotImplementedError: If the driver does not support Wake-on-LAN.
:raises ValueError: If ``mac_address`` is malformed, or ``interface`` is
required by this driver but was not provided.
:returns: A dict with:
* success (bool) - ``True`` if the magic packet was sent without error
* output (string) - human-readable status message
Example::
driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
# → {"success": True, "output": "Magic packet sent to AA:BB:CC:DD:EE:FF via lan"}
.. note::
A driver whose ``interface`` is *not* the name :meth:`get_interfaces`
is keyed by must expose the name it does expect as an ``identifier``
key on each ``get_interfaces()`` entry. Without it a caller has no
way to offer a valid choice: OPNsense, for instance, keys interfaces
by the physical device ("em0") but wakes by the assigned name
("lan"), and rejects the former. ``identifier`` is a non-standard
NAPALM key, so it reaches consumers through the usual passthrough
for extra interface data.
"""
...
# ------------------------------------------------------------------
# Firewall rule diff/apply. get_firewall_rules/apply_firewall_rule/
# commit_firewall_rules are abstract (device communication); everything
# else here is a concrete, vendor-neutral algorithm -- see README.md
# "Design principle: generic vs. device-specific logic".
# ------------------------------------------------------------------
Intended for streaming to a caller (e.g. an SSE endpoint) that wants
live progress while writing to a real device.
:param desired: The complete desired rule set.
:yields: Progress lines, one per applied add/update, plus a final
commit line.
"""
diff = self.diff_firewall_rules(desired)
for rule in diff["add"]:
self.apply_firewall_rule(rule)
yield f"[add] {rule['description']}"
for entry in diff["update"]:
self.apply_firewall_rule(entry["rule"], uuid=entry["uuid"])
fields = ", ".join(entry["changed_fields"])
yield f"[update] {entry['rule']['description']} ({fields})"
self.commit_firewall_rules()
yield f"[commit] applied {len(diff['add'])} add(s), {len(diff['update'])} update(s)"
napalm_device_types/firewall_rules.py
+161
View File
@@ -0,0 +1,161 @@
# -*- coding: utf-8 -*-
"""Firewall rule reconciliation: generic here, device communication in the driver.
The split follows the rule in this package's README: matching, diffing and the
apply-then-commit sequence would be identical for any vendor's firewall, so they
live here as concrete methods. Only the three hooks below touch the device, and
a concrete driver supplies those.
Declared under ``if TYPE_CHECKING``, the hooks do not exist at runtime until a
driver implements them -- so ``hasattr`` stays an honest answer to "can this
driver manage firewall rules", and mixing this class in can never shadow a
working implementation inherited from elsewhere.
"""
from __future__ import annotations
from typing import Any, Dict, Iterator, List, Optional, TYPE_CHECKING
from napalm_device_types.models import (
FirewallRuleDict,
FirewallRuleDiffDict,
FirewallRuleUpdateDict,
)
_FIREWALL_RULE_COMPARE_FIELDS = (
"action",
"interface",
"direction",
"protocol",
"source_net",
"source_port",
"destination_net",
"destination_port",
"log",
"quick",
"enabled",
)
class FirewallRuleMixin:
"""Adds generic firewall-rule diff and apply on top of three device hooks."""
if TYPE_CHECKING:
def get_firewall_rules(self) -> List[FirewallRuleDict]:
"""
Returns all firewall filter rules currently configured on the device.
`description` must be a stable, human-assigned identifier -- it is
the key used to match rules across calls (most firewall vendors
don't expose an ID a caller can pre-assign).
:raises NotImplementedError: If the driver does not support reading
firewall rules.
"""
...
def apply_firewall_rule(
self, rule: FirewallRuleDict, *, uuid: Optional[str] = None
) -> Dict[str, Any]:
"""
Creates or updates a single firewall filter rule on the device.
:param rule: The desired rule state, in vendor-neutral form.
:param uuid: If given, update the existing rule with this ID
in-place. If ``None``, create a new rule.
:raises NotImplementedError: If the driver does not support writing
firewall rules.
:raises ValueError: If `rule` references an alias/interface the
device doesn't know about.
:raises RuntimeError: If the device rejects the write.
:returns: A dict with at least ``{"success": bool}``.
"""
...
def commit_firewall_rules(self) -> Dict[str, Any]:
"""
Applies pending firewall filter rule changes (e.g. reloads pf/pfctl,
or whatever the device's equivalent of "Apply Changes" is).
Call once after one or more `apply_firewall_rule()` calls -- not
after every single rule.
:raises NotImplementedError: If the driver does not support this
(e.g. rules take effect immediately on write).
:returns: A dict with at least ``{"success": bool}``.
"""
...
def diff_firewall_rules(self, desired: List[FirewallRuleDict]) -> FirewallRuleDiffDict:
"""
Compares `desired` against the device's current rules and returns
what would need to change to reach that state.
Matches rules by `description`. A desired rule with no live
counterpart becomes an "add"; a live rule whose description matches
but whose other fields differ becomes an "update". Live rules with
no matching desired entry are **not** reported for deletion -- this
is intentionally conservative: a firewall may carry manually-created
or otherwise unmanaged rules that a caller's `desired` set was never
meant to describe, and this method has no way to distinguish those
from ones simply no longer wanted. Callers wanting delete/cleanup
semantics must implement that themselves, deliberately.
:param desired: The complete desired rule set.
:returns: ``{"add": [...], "update": [{"uuid", "rule",
"changed_fields"}, ...]}``.
"""
live_by_description: Dict[str, FirewallRuleDict] = {
rule["description"]: rule for rule in self.get_firewall_rules()
}
add: List[FirewallRuleDict] = []
update: List[FirewallRuleUpdateDict] = []
for desired_rule in desired:
live_rule = live_by_description.get(desired_rule["description"])
if live_rule is None:
add.append(desired_rule)
continue
changed_fields = [
field
for field in _FIREWALL_RULE_COMPARE_FIELDS
if live_rule.get(field) != desired_rule.get(field)
]
if changed_fields:
update.append(
{
"uuid": live_rule["uuid"],
"rule": desired_rule,
"changed_fields": changed_fields,
}
)
return {"add": add, "update": update}
def apply_firewall_ruleset(self, desired: List[FirewallRuleDict]) -> Iterator[str]:
"""
Computes the diff against `desired` and applies it, yielding one
human-readable progress line per change, then commits.
Intended for streaming to a caller (e.g. an SSE endpoint) that wants
live progress while writing to a real device.
:param desired: The complete desired rule set.
:yields: Progress lines, one per applied add/update, plus a final
commit line.
"""
diff = self.diff_firewall_rules(desired)
for rule in diff["add"]:
self.apply_firewall_rule(rule)
yield f"[add] {rule['description']}"
for entry in diff["update"]:
self.apply_firewall_rule(entry["rule"], uuid=entry["uuid"])
fields = ", ".join(entry["changed_fields"])
yield f"[update] {entry['rule']['description']} ({fields})"
self.commit_firewall_rules()
yield f"[commit] applied {len(diff['add'])} add(s), {len(diff['update'])} update(s)"
napalm_device_types/health_metrics.py
+53
View File
@@ -0,0 +1,53 @@
# -*- coding: utf-8 -*-
"""SNMP health collection, shared by every device type that speaks UCD-MIB.
The collection itself is generic: CPU, memory, uptime, load and per-interface
counters come from the same standard OIDs on any UCD-MIB/IF-MIB capable device.
Only two details vary by device type, and both are class attributes rather than
code -- which is why this was five byte-identical copies of the same method
before it moved here.
A device type whose vendor publishes the same data under proprietary OIDs does
**not** mix this in; ``SwitchDriver`` is the example, and its concrete drivers
implement ``get_health_metrics`` themselves.
"""
from __future__ import annotations
import re
from typing import Any, Awaitable, Callable, ClassVar, Dict, cast
from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
from napalm_device_types.models import HealthMetricsDict
class HealthMetricsMixin:
"""Adds the standard UCD-MIB/IF-MIB ``get_health_metrics`` collector."""
#: Interfaces whose counters are noise rather than signal. Loopback,
#: tunnels and container bridges inflate error rates without meaning.
_SNMP_SKIP_IF: ClassVar[re.Pattern[str]] = IF_SKIP_DEFAULT
#: Whether the device reports discards in the TX-error counter. Firewalls
#: do -- a dropped packet is the point, not a fault -- so counting those as
#: errors would make a healthy firewall look broken.
_SNMP_TX_ERR_IS_DROP: ClassVar[bool] = False
@classmethod
async def get_health_metrics(
cls,
snmp_get: Callable[..., Awaitable[Any]],
snmp_walk: Callable[..., Awaitable[Dict[str, Any]]],
) -> HealthMetricsDict:
"""Collect SNMP health metrics (CPU, memory, uptime, load, interfaces)."""
# collect_ucd_metrics is untyped shared plumbing; the shape it returns is
# the contract HealthMetricsDict describes.
return cast(
HealthMetricsDict,
await collect_ucd_metrics(
snmp_get,
snmp_walk,
tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
if_skip=cls._SNMP_SKIP_IF,
),
)
napalm_device_types/hypervisor.py
+544 -657
View File
File diff suppressed because it is too large Load Diff
napalm_device_types/interface_filter.py
+31
View File
@@ -0,0 +1,31 @@
# -*- coding: utf-8 -*-
"""Dropping interfaces that are noise rather than signal.
An access point exposes radio PHYs and a loopback alongside its real
interfaces. Reporting them inflates every interface count and every error rate,
so drivers filter them out -- identically, whatever the vendor. The two class
attributes are the customisation point.
"""
from __future__ import annotations
from typing import Any, ClassVar, Dict, FrozenSet, Tuple
class InterfaceFilterMixin:
"""Adds :meth:`_filter_interfaces` for drivers that report raw interface dicts."""
#: Interface names dropped outright.
_EXCLUDED_INTERFACES: ClassVar[FrozenSet[str]] = frozenset({"lo"})
#: Name prefixes dropped -- ``phy*`` are radio devices, not links.
_EXCLUDED_INTERFACE_PREFIXES: ClassVar[Tuple[str, ...]] = ("phy",)
def _filter_interfaces(self, interfaces: Dict[str, Any]) -> Dict[str, Any]:
"""Remove loopback and radio-device (phy*) interfaces from an interface dict."""
return {
name: data
for name, data in interfaces.items()
if name not in self._EXCLUDED_INTERFACES
and not name.startswith(self._EXCLUDED_INTERFACE_PREFIXES)
}
napalm_device_types/mac_acl.py
+79
View File
@@ -0,0 +1,79 @@
# -*- coding: utf-8 -*-
"""MAC-based access control, per SSID on an AP and per port on a switch.
The reader is identical either way; only the writer is AP-specific so far.
Declared under ``if TYPE_CHECKING``: these are contracts, not placeholders.
Nothing exists at runtime until a concrete driver implements it, so mixing
this class in can never shadow a working implementation from a sibling base.
"""
from __future__ import annotations
from typing import Dict, List, TYPE_CHECKING
from napalm_device_types.models import MACACLDict
class MacAclMixin:
if TYPE_CHECKING:
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"},
],
},
}
"""
...
def push_mac_acl(self, ssid_name: str, mode: str, macs: List[str]) -> None:
"""
Rewrites the MAC-address access control list for a single SSID.
Full-rebuild semantics: replaces whatever ACL state currently exists
for *ssid_name* with *mode* + *macs* — not a diff/patch.
:param ssid_name: SSID name to apply the ACL to.
:param mode: ``"off"`` | ``"whitelist"`` | ``"blacklist"``.
:param macs: MAC addresses for the active list. Ignored when ``mode == "off"``.
:raises NotImplementedError: If the driver does not support MAC ACL push.
Example::
driver.push_mac_acl("CorpWiFi", "whitelist", ["AA:BB:CC:DD:EE:01", "AA:BB:CC:DD:EE:02"])
driver.push_mac_acl("GuestNet", "off", [])
"""
...
napalm_device_types/nat_vpn.py
+78
View File
@@ -0,0 +1,78 @@
# -*- coding: utf-8 -*-
"""Address translation and VPN tunnels.
A home gateway does a subset of what a firewall does, and these two readers
are where the sets overlap exactly.
Declared under ``if TYPE_CHECKING``: these are contracts, not placeholders.
Nothing exists at runtime until a concrete driver implements it, so mixing
this class in can never shadow a working implementation from a sibling base.
"""
from __future__ import annotations
from typing import Dict, List, TYPE_CHECKING
from napalm_device_types.models import NATTranslationDict, VPNTunnelDict
class NatVpnMixin:
if TYPE_CHECKING:
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,
}
]
"""
...
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,
}
}
"""
...
napalm_device_types/os.py
+254 -369
View File
@@ -10,27 +10,23 @@ Usage::
...
"""
import re
from typing import List, Optional
from typing import List, Optional, TYPE_CHECKING
from napalm_device_types.base import DeviceTypeDriver
from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
from napalm_device_types.updates import UpdateMixin
from napalm_device_types.services import ServiceControlMixin
from napalm_device_types.packages import PackageManagementMixin
from napalm_device_types.health_metrics import HealthMetricsMixin
from napalm_device_types.models import (
ApplyUpdatesResultDict,
CronJobDict,
DeviceActionResultDict,
DockerInfoDict,
HealthMetricsDict,
PackageDict,
ProcessDict,
ServiceDict,
SNMPConfigDict,
UpdateDict,
UserDict,
)
class OSDriver(DeviceTypeDriver):
TYPE_LABEL: str = "OS"
class OSDriver(UpdateMixin, ServiceControlMixin, PackageManagementMixin, HealthMetricsMixin, DeviceTypeDriver):
"""
Abstract intermediate driver for general-purpose operating systems
(e.g. Linux, BSD, macOS).
@@ -39,391 +35,280 @@ class OSDriver(DeviceTypeDriver):
operations that concrete drivers must implement.
"""
#: Stable key netOrk exposes as ``device_class``. The order in which a
#: driver lists its role bases is the ranking; see
#: :func:`napalm_device_types.roles.primary_role_of`.
ROLE: str = "linux"
TYPE_LABEL: str = "OS"
# Interfaces matching this pattern are excluded from health-metric collection.
_SNMP_SKIP_IF: re.Pattern = IF_SKIP_DEFAULT
# Set True for drivers where the out-error counter actually reports drops (e.g. OPNsense).
_SNMP_TX_ERR_IS_DROP: bool = False
@classmethod
async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
"""Collect SNMP health metrics (CPU, memory, uptime, load, interfaces).
:param snmp_get: async callable ``(oid: str) -> Optional[str]``
:param snmp_walk: async callable ``(oid: str) -> Dict[str, str]``
"""
return await collect_ucd_metrics(
snmp_get, snmp_walk,
tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
if_skip=cls._SNMP_SKIP_IF,
)
# ------------------------------------------------------------------
# Package management
# ------------------------------------------------------------------
def get_packages(self) -> List[PackageDict]:
"""
Returns a list of all installed software packages.
if TYPE_CHECKING:
Each entry contains:
* name (string) - package name
* version (string) - installed version string
* installed (bool) - always ``True`` for this method
* description (string) - short package description
* size (int) - installed size in bytes; ``0`` if unavailable
* source (string) - package source / repository name; empty string if unavailable
Example::
[
{
"name": "openssh-server",
"version": "1:9.2p1-2+deb12u2",
"installed": True,
"description": "secure shell (SSH) server, for secure access from remote machines",
"size": 524288,
"source": "Debian",
},
]
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Service management
# ------------------------------------------------------------------
def get_pending_updates(self) -> List[UpdateDict]:
"""
Returns a list of packages that have a newer version available.
Each entry contains:
# ------------------------------------------------------------------
# Users
# ------------------------------------------------------------------
* name (string) - package name
* current_version (string) - currently installed version
* new_version (string) - version available in the repository
def get_users(self) -> List[UserDict]:
"""
Returns a list of local OS user accounts.
Example::
Each entry contains:
[
{
"name": "openssh-server",
"current_version": "1:9.2p1-2+deb12u1",
"new_version": "1:9.2p1-2+deb12u2",
},
]
"""
raise NotImplementedError
* username (string) - login name
* uid (int) - numeric user ID
* gid (int) - primary group ID
* home (string) - home directory path
* shell (string) - login shell path
* groups (list of strings) - all supplementary group names
def apply_updates(self, packages: List[str]) -> ApplyUpdatesResultDict:
"""
Upgrades the given packages to the newest available version.
Example::
Only packages that are already installed may be upgraded; this method
does **not** install new packages. Pass an empty list to upgrade
**all** packages that have pending updates.
:param packages: List of package names to upgrade. Each name must
match ``^[a-zA-Z0-9_\\-\\+\\.]+$``; a :exc:`ValueError` is raised
for any name that does not conform.
:returns: A dict with:
* success (bool) – ``True`` if the package manager exited without error
* output (string) – combined stdout / stderr from the package manager
* error (string, optional) – short error message when *success* is ``False``
:raises ValueError: If any package name fails the safety check.
Example::
result = driver.apply_updates(["openssh-server", "curl"])
# → {"success": True, "output": "Reading package lists...\\n..."}
# Upgrade everything:
result = driver.apply_updates([])
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Service management
# ------------------------------------------------------------------
def get_services(self) -> List[ServiceDict]:
"""
Returns a list of system services and their current state.
Each entry contains:
* name (string) - service unit name (without ``.service`` suffix)
* running (bool) - ``True`` if the service is currently active
* enabled (bool) - ``True`` if the service starts automatically on boot
* pid (int) - main process ID; ``0`` if not running
Example::
[
{
"name": "ssh",
"running": True,
"enabled": True,
"pid": 1234,
},
{
"name": "cron",
"running": True,
"enabled": True,
"pid": 5678,
},
]
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Users
# ------------------------------------------------------------------
def get_users(self) -> List[UserDict]:
"""
Returns a list of local OS user accounts.
Each entry contains:
* username (string) - login name
* uid (int) - numeric user ID
* gid (int) - primary group ID
* home (string) - home directory path
* shell (string) - login shell path
* groups (list of strings) - all supplementary group names
Example::
[
{
"username": "admin",
"uid": 1000,
"gid": 1000,
"home": "/home/admin",
"shell": "/bin/bash",
"groups": ["sudo", "docker", "adm"],
},
]
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Processes
# ------------------------------------------------------------------
def get_processes(self) -> List[ProcessDict]:
"""
Returns a snapshot of currently running processes.
Each entry contains:
* pid (int) - process ID
* ppid (int) - parent process ID
* user (string) - effective user name
* cpu (float) - CPU utilisation percentage
* memory (float) - RSS as a percentage of total RAM
* vsz (int) - virtual memory size in KiB
* rss (int) - resident set size in KiB
* tty (string) - controlling terminal; empty string if none
* state (string) - process state: ``"R"`` running, ``"S"`` sleeping,
``"D"`` uninterruptible, ``"Z"`` zombie, ``"T"`` stopped, etc.
* started (string) - start time as printed by ``ps`` (e.g. ``"12:34"`` or ``"May28"``)
* command (string) - full command line
Example::
[
{
"pid": 1,
"ppid": 0,
"user": "root",
"cpu": 0.0,
"memory": 0.1,
"vsz": 168576,
"rss": 13312,
"tty": "",
"state": "S",
"started": "May28",
"command": "/sbin/init",
},
]
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Cron jobs
# ------------------------------------------------------------------
def get_cron_jobs(self) -> List[CronJobDict]:
"""
Returns scheduled cron tasks from all user crontabs and ``/etc/cron.d``.
Each entry contains:
* user (string) - owner of the crontab entry
* schedule (string) - five-field cron expression (e.g. ``"0 * * * *"``)
* command (string) - shell command to execute
* description (string, optional) - inline comment text if present
Example::
[
{
"user": "root",
"schedule": "0 4 * * *",
"command": "/usr/local/bin/backup.sh",
"description": "nightly backup",
},
]
"""
raise NotImplementedError
# ------------------------------------------------------------------
# SNMP
# ------------------------------------------------------------------
def get_snmp_config(self) -> Optional[SNMPConfigDict]:
"""
Returns the SNMP agent configuration currently active on the device,
or ``None`` if no SNMP daemon is running or detectable.
The returned dictionary contains:
* running (bool) - whether the SNMP daemon is currently active
* community (string) - the read community string (e.g. ``"public"``)
* port (int) - the UDP port the agent listens on (default ``161``)
* version (string) - highest supported SNMP version: ``"1"``, ``"2c"``, or ``"3"``
Example::
# snmpd running with community "public":
{
"running": True,
"community": "public",
"port": 161,
"version": "2c",
}
# snmpd not installed / not running:
None
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Docker
# ------------------------------------------------------------------
def get_docker_info(self) -> DockerInfoDict:
"""
Returns information about the local Docker environment.
If Docker is not installed or the current user lacks access to the
Docker socket, returns ``{"available": False}``. When the user has
no socket permission, ``permission_denied`` is additionally set to
``True``.
When Docker is available the dict contains:
* available (bool) - always ``True``
* version (string) - Docker Engine version string
* containers (list) - all containers (running and stopped), each with:
* id (string) - short container ID
* name (string) - container name(s)
* image (string) - image reference
* image_version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
* command (string) - entrypoint / command string
* created (string) - creation timestamp string
* status (string) - human-readable status (e.g. ``"Up 3 hours"``)
* ports (string) - port mapping string
* state (string) - ``"running"``, ``"exited"``, ``"paused"``, etc.
* images (list) - local images, each with:
* id (string) - short image ID
* repository (string) - image repository
* tag (string) - image tag
* size (string) - human-readable size string (e.g. ``"187MB"``)
* created (string) - creation timestamp string
* version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
* volumes (list) - Docker volumes, each with:
* name (string) - volume name
* driver (string) - volume driver
* mountpoint (string) - host filesystem path
* scope (string) - ``"local"`` or ``"global"``
* networks (list) - Docker networks, each with:
* id (string) - short network ID
* name (string) - network name
* driver (string) - network driver (e.g. ``"bridge"``, ``"host"``, ``"overlay"``)
* scope (string) - network scope
* ipv6 (string) - ``"true"`` if IPv6 is enabled
* internal (string) - ``"true"`` if the network is internal
* outdated_images (list of strings) - image names where the local digest
differs from the latest remote digest; empty list if all images are
current or update checks could not be performed.
Example::
# Docker not installed:
{"available": False}
# Docker installed, no socket permission:
{"available": False, "permission_denied": True}
# Docker available:
{
"available": True,
"version": "Docker version 27.3.1, build ce12230",
"containers": [
[
{
"id": "a1b2c3d4e5f6",
"name": "my-app",
"image": "nginx:latest",
"image_version": "1.27.0",
"command": "nginx -g 'daemon off;'",
"created": "2026-05-28 10:00:00 +0000 UTC",
"status": "Up 3 days",
"ports": "0.0.0.0:80->80/tcp",
"state": "running",
"username": "admin",
"uid": 1000,
"gid": 1000,
"home": "/home/admin",
"shell": "/bin/bash",
"groups": ["sudo", "docker", "adm"],
},
],
"images": [...],
"volumes": [],
"networks": [...],
"outdated_images": ["nginx:latest"],
}
"""
raise NotImplementedError
]
"""
...
# ------------------------------------------------------------------
# Generic device actions
# ------------------------------------------------------------------
# ------------------------------------------------------------------
# Processes
# ------------------------------------------------------------------
def run_device_action(self, action: str) -> DeviceActionResultDict:
"""
Executes a named administrative action on the device.
def get_processes(self) -> List[ProcessDict]:
"""
Returns a snapshot of currently running processes.
This method is an extensibility point for driver-specific one-off
operations that do not fit any other NAPALM API method. Each driver
documents the action names it supports.
Each entry contains:
:param action: Action identifier string (e.g. ``"fix_docker_permissions"``).
:raises NotImplementedError: If the driver does not implement this method.
:raises ValueError: If ``action`` is not a recognised action name for
this driver.
* pid (int) - process ID
* ppid (int) - parent process ID
* user (string) - effective user name
* cpu (float) - CPU utilisation percentage
* memory (float) - RSS as a percentage of total RAM
* vsz (int) - virtual memory size in KiB
* rss (int) - resident set size in KiB
* tty (string) - controlling terminal; empty string if none
* state (string) - process state: ``"R"`` running, ``"S"`` sleeping,
``"D"`` uninterruptible, ``"Z"`` zombie, ``"T"`` stopped, etc.
* started (string) - start time as printed by ``ps`` (e.g. ``"12:34"`` or ``"May28"``)
* command (string) - full command line
:returns: A dict with:
Example::
* success (bool) – ``True`` if the action completed without error
* output (string) – human-readable output or status message
[
{
"pid": 1,
"ppid": 0,
"user": "root",
"cpu": 0.0,
"memory": 0.1,
"vsz": 168576,
"rss": 13312,
"tty": "",
"state": "S",
"started": "May28",
"command": "/sbin/init",
},
]
"""
...
Example::
# ------------------------------------------------------------------
# Cron jobs
# ------------------------------------------------------------------
result = driver.run_device_action("fix_docker_permissions")
# → {"success": True, "output": "Added 'pi' to the docker group. Reconnect..."}
"""
raise NotImplementedError
def get_cron_jobs(self) -> List[CronJobDict]:
"""
Returns scheduled cron tasks from all user crontabs and ``/etc/cron.d``.
Each entry contains:
* user (string) - owner of the crontab entry
* schedule (string) - five-field cron expression (e.g. ``"0 * * * *"``)
* command (string) - shell command to execute
* description (string, optional) - inline comment text if present
Example::
[
{
"user": "root",
"schedule": "0 4 * * *",
"command": "/usr/local/bin/backup.sh",
"description": "nightly backup",
},
]
"""
...
# ------------------------------------------------------------------
# SNMP
# ------------------------------------------------------------------
def get_snmp_config(self) -> Optional[SNMPConfigDict]:
"""
Returns the SNMP agent configuration currently active on the device,
or ``None`` if no SNMP daemon is running or detectable.
The returned dictionary contains:
* running (bool) - whether the SNMP daemon is currently active
* community (string) - the read community string (e.g. ``"public"``)
* port (int) - the UDP port the agent listens on (default ``161``)
* version (string) - highest supported SNMP version: ``"1"``, ``"2c"``, or ``"3"``
Example::
# snmpd running with community "public":
{
"running": True,
"community": "public",
"port": 161,
"version": "2c",
}
# snmpd not installed / not running:
None
"""
...
# ------------------------------------------------------------------
# Docker
# ------------------------------------------------------------------
def get_docker_info(self) -> DockerInfoDict:
"""
Returns information about the local Docker environment.
If Docker is not installed or the current user lacks access to the
Docker socket, returns ``{"available": False}``. When the user has
no socket permission, ``permission_denied`` is additionally set to
``True``.
When Docker is available the dict contains:
* available (bool) - always ``True``
* version (string) - Docker Engine version string
* containers (list) - all containers (running and stopped), each with:
* id (string) - short container ID
* name (string) - container name(s)
* image (string) - image reference
* image_version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
* command (string) - entrypoint / command string
* created (string) - creation timestamp string
* status (string) - human-readable status (e.g. ``"Up 3 hours"``)
* ports (string) - port mapping string
* state (string) - ``"running"``, ``"exited"``, ``"paused"``, etc.
* images (list) - local images, each with:
* id (string) - short image ID
* repository (string) - image repository
* tag (string) - image tag
* size (string) - human-readable size string (e.g. ``"187MB"``)
* created (string) - creation timestamp string
* version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
* volumes (list) - Docker volumes, each with:
* name (string) - volume name
* driver (string) - volume driver
* mountpoint (string) - host filesystem path
* scope (string) - ``"local"`` or ``"global"``
* networks (list) - Docker networks, each with:
* id (string) - short network ID
* name (string) - network name
* driver (string) - network driver (e.g. ``"bridge"``, ``"host"``, ``"overlay"``)
* scope (string) - network scope
* ipv6 (string) - ``"true"`` if IPv6 is enabled
* internal (string) - ``"true"`` if the network is internal
* outdated_images (list of strings) - image names where the local digest
differs from the latest remote digest; empty list if all images are
current or update checks could not be performed.
Example::
# Docker not installed:
{"available": False}
# Docker installed, no socket permission:
{"available": False, "permission_denied": True}
# Docker available:
{
"available": True,
"version": "Docker version 27.3.1, build ce12230",
"containers": [
{
"id": "a1b2c3d4e5f6",
"name": "my-app",
"image": "nginx:latest",
"image_version": "1.27.0",
"command": "nginx -g 'daemon off;'",
"created": "2026-05-28 10:00:00 +0000 UTC",
"status": "Up 3 days",
"ports": "0.0.0.0:80->80/tcp",
"state": "running",
},
],
"images": [...],
"volumes": [],
"networks": [...],
"outdated_images": ["nginx:latest"],
}
"""
...
# ------------------------------------------------------------------
# Generic device actions
# ------------------------------------------------------------------
def run_device_action(self, action: str) -> DeviceActionResultDict:
"""
Executes a named administrative action on the device.
This method is an extensibility point for driver-specific one-off
operations that do not fit any other NAPALM API method. Each driver
documents the action names it supports.
:param action: Action identifier string (e.g. ``"fix_docker_permissions"``).
:raises NotImplementedError: If the driver does not implement this method.
:raises ValueError: If ``action`` is not a recognised action name for
this driver.
:returns: A dict with:
* success (bool) – ``True`` if the action completed without error
* output (string) – human-readable output or status message
Example::
result = driver.run_device_action("fix_docker_permissions")
# → {"success": True, "output": "Added 'pi' to the docker group. Reconnect..."}
"""
...
napalm_device_types/packages.py
+158
View File
@@ -0,0 +1,158 @@
# -*- coding: utf-8 -*-
"""Listing and managing installed software.
Identical on every device type that has a package manager -- this was five
byte-identical copies before it moved here. A driver whose device offers no
package management simply implements none of it.
Declared under ``if TYPE_CHECKING``: these are contracts, not placeholders.
Nothing exists at runtime until a concrete driver implements it, so mixing
this class in can never shadow a working implementation from a sibling base.
"""
from __future__ import annotations
from typing import Any, Dict, List, TYPE_CHECKING
from napalm_device_types.models import PackageDict
class PackageManagementMixin:
if TYPE_CHECKING:
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",
},
]
"""
...
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")
"""
...
def uninstall_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.uninstall_package("luci-app-statistics")
"""
...
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,
},
}
"""
...
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},
},
)
"""
...
napalm_device_types/residential_gateway.py
+109 -133
View File
@@ -18,25 +18,22 @@ Usage::
...
"""
from typing import Dict, List
from typing import Dict, List, TYPE_CHECKING
from napalm_device_types.base import DeviceTypeDriver
from napalm_device_types.nat_vpn import NatVpnMixin
from napalm_device_types.health_metrics import HealthMetricsMixin
from napalm_device_types.dhcp import DhcpServerMixin
from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
from napalm_device_types.models import (
HealthMetricsDict,
HostDict,
NATTranslationDict,
PortForwardDict,
RadioStatusDict,
SSIDDict,
VPNTunnelDict,
WANStatusDict,
WirelessClientDict,
)
class ResidentialGatewayDriver(DhcpServerMixin, DeviceTypeDriver):
TYPE_LABEL: str = "Gateway"
class ResidentialGatewayDriver(NatVpnMixin, HealthMetricsMixin, DhcpServerMixin, DeviceTypeDriver):
"""
Abstract intermediate driver for residential gateways (router + firewall + AP).
@@ -45,154 +42,133 @@ class ResidentialGatewayDriver(DhcpServerMixin, DeviceTypeDriver):
drivers must implement.
"""
_SNMP_SKIP_IF = IF_SKIP_DEFAULT
_SNMP_TX_ERR_IS_DROP: bool = False
#: Stable key netOrk exposes as ``device_class``. The order in which a
#: driver lists its role bases is the ranking; see
#: :func:`napalm_device_types.roles.primary_role_of`.
ROLE: str = "residential_gateway"
TYPE_LABEL: str = "Gateway"
@classmethod
async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
return await collect_ucd_metrics(
snmp_get, snmp_walk,
tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
if_skip=cls._SNMP_SKIP_IF,
)
def get_wan_status(self) -> WANStatusDict:
"""
Returns the status of the device's internet (WAN) uplink.
Contains:
if TYPE_CHECKING:
* connection_type (string) - e.g. ``"DSL"``, ``"Cable"``, ``"PPPoE"``, ``"DHCP"``
* is_connected (bool) - whether the WAN connection is currently established
* external_ip (string) - the public IPv4 address assigned to the WAN interface
* uptime (int) - seconds since the WAN connection was last (re-)established
* bytes_sent (int) - total bytes transmitted on the WAN interface
* bytes_received (int) - total bytes received on the WAN interface
* max_bitrate_up (int) - upstream sync rate in kbit/s
* max_bitrate_down (int) - downstream sync rate in kbit/s
* external_ipv6 (string, optional) - the public IPv6 address, if any
* link_status (string, optional) - physical line state, e.g. ``"Up"`` / ``"Down"``
def get_wan_status(self) -> WANStatusDict:
"""
Returns the status of the device's internet (WAN) uplink.
Example::
Contains:
{
"connection_type": "DSL",
"is_connected": True,
"external_ip": "203.0.113.7",
"uptime": 345600,
"bytes_sent": 1234567890,
"bytes_received": 9876543210,
"max_bitrate_up": 40000,
"max_bitrate_down": 250000,
"link_status": "Up",
}
"""
raise NotImplementedError
* connection_type (string) - e.g. ``"DSL"``, ``"Cable"``, ``"PPPoE"``, ``"DHCP"``
* is_connected (bool) - whether the WAN connection is currently established
* external_ip (string) - the public IPv4 address assigned to the WAN interface
* uptime (int) - seconds since the WAN connection was last (re-)established
* bytes_sent (int) - total bytes transmitted on the WAN interface
* bytes_received (int) - total bytes received on the WAN interface
* max_bitrate_up (int) - upstream sync rate in kbit/s
* max_bitrate_down (int) - downstream sync rate in kbit/s
* external_ipv6 (string, optional) - the public IPv6 address, if any
* link_status (string, optional) - physical line state, e.g. ``"Up"`` / ``"Down"``
def get_port_forwards(self) -> List[PortForwardDict]:
"""
Returns the configured port forwarding (port mapping) rules.
Example::
Each entry contains:
* name (string) - the rule's description/name
* protocol (string) - ``"TCP"`` or ``"UDP"``
* external_port (int) - the WAN-side port
* internal_ip (string) - the LAN host the traffic is forwarded to
* internal_port (int) - the LAN-side port
* enabled (bool) - whether the rule is currently active
* remote_host (string, optional) - restricts the forward to a specific
remote source address; empty/absent means "any"
Example::
[
{
"name": "Webserver HTTPS",
"protocol": "TCP",
"external_port": 443,
"internal_ip": "192.168.1.10",
"internal_port": 443,
"enabled": True,
"connection_type": "DSL",
"is_connected": True,
"external_ip": "203.0.113.7",
"uptime": 345600,
"bytes_sent": 1234567890,
"bytes_received": 9876543210,
"max_bitrate_up": 40000,
"max_bitrate_down": 250000,
"link_status": "Up",
}
]
"""
raise NotImplementedError
"""
...
def get_hosts(self) -> List[HostDict]:
"""
Returns the list of hosts known to the gateway (LAN clients).
def get_port_forwards(self) -> List[PortForwardDict]:
"""
Returns the configured port forwarding (port mapping) rules.
Each entry contains:
Each entry contains:
* mac (string) - the host's MAC address
* ip (string) - the host's current IP address
* hostname (string) - the host's reported hostname (empty if unknown)
* interface_type (string) - how the host is connected, e.g. ``"LAN"``, ``"WLAN"``
* is_active (bool) - whether the host is currently online
* lease_time_remaining (int, optional) - remaining DHCP lease time in seconds
* name (string) - the rule's description/name
* protocol (string) - ``"TCP"`` or ``"UDP"``
* external_port (int) - the WAN-side port
* internal_ip (string) - the LAN host the traffic is forwarded to
* internal_port (int) - the LAN-side port
* enabled (bool) - whether the rule is currently active
* remote_host (string, optional) - restricts the forward to a specific
remote source address; empty/absent means "any"
Example::
Example::
[
{
"mac": "AA:BB:CC:DD:EE:FF",
"ip": "192.168.1.42",
"hostname": "laptop",
"interface_type": "WLAN",
"is_active": True,
"lease_time_remaining": 3600,
}
]
"""
raise NotImplementedError
[
{
"name": "Webserver HTTPS",
"protocol": "TCP",
"external_port": 443,
"internal_ip": "192.168.1.10",
"internal_port": 443,
"enabled": True,
}
]
"""
...
def get_nat_translations(self) -> List[NATTranslationDict]:
"""
Returns a list of active NAT translation entries.
def get_hosts(self) -> List[HostDict]:
"""
Returns the list of hosts known to the gateway (LAN clients).
See :meth:`napalm_device_types.firewall.FirewallDriver.get_nat_translations`
for the entry format. Residential gateways typically derive this from
the active port-forwarding/NAT-PT table rather than a live connection
tracker; drivers that cannot provide this should return an empty list.
"""
raise NotImplementedError
Each entry contains:
def get_vpn_tunnels(self) -> Dict[str, VPNTunnelDict]:
"""
Returns the status of VPN tunnels (e.g. WireGuard road-warrior
access, IPsec site-to-site).
* mac (string) - the host's MAC address
* ip (string) - the host's current IP address
* hostname (string) - the host's reported hostname (empty if unknown)
* interface_type (string) - how the host is connected, e.g. ``"LAN"``, ``"WLAN"``
* is_active (bool) - whether the host is currently online
* lease_time_remaining (int, optional) - remaining DHCP lease time in seconds
See :meth:`napalm_device_types.firewall.FirewallDriver.get_vpn_tunnels`
for the entry format. Drivers that cannot provide this should return
an empty dict.
"""
raise NotImplementedError
Example::
def get_wireless_clients(self) -> List[WirelessClientDict]:
"""
Returns the list of wireless clients currently associated with the
device's built-in access point(s).
[
{
"mac": "AA:BB:CC:DD:EE:FF",
"ip": "192.168.1.42",
"hostname": "laptop",
"interface_type": "WLAN",
"is_active": True,
"lease_time_remaining": 3600,
}
]
"""
...
See :meth:`napalm_device_types.access_point.AccessPointDriver.get_wireless_clients`
for the entry format.
"""
raise NotImplementedError
def get_ssids(self) -> Dict[str, SSIDDict]:
"""
Returns the configured wireless networks (SSIDs).
See :meth:`napalm_device_types.access_point.AccessPointDriver.get_ssids`
for the entry format.
"""
raise NotImplementedError
def get_wireless_clients(self) -> List[WirelessClientDict]:
"""
Returns the list of wireless clients currently associated with the
device's built-in access point(s).
def get_radio_status(self) -> Dict[str, RadioStatusDict]:
"""
Returns the status of the device's wireless radios.
See :meth:`napalm_device_types.access_point.AccessPointDriver.get_wireless_clients`
for the entry format.
"""
...
See :meth:`napalm_device_types.access_point.AccessPointDriver.get_radio_status`
for the entry format.
"""
raise NotImplementedError
def get_ssids(self) -> Dict[str, SSIDDict]:
"""
Returns the configured wireless networks (SSIDs).
See :meth:`napalm_device_types.access_point.AccessPointDriver.get_ssids`
for the entry format.
"""
...
def get_radio_status(self) -> Dict[str, RadioStatusDict]:
"""
Returns the status of the device's wireless radios.
See :meth:`napalm_device_types.access_point.AccessPointDriver.get_radio_status`
for the entry format.
"""
...
napalm_device_types/roles.py
+59
View File
@@ -0,0 +1,59 @@
# -*- coding: utf-8 -*-
"""Which roles a driver class fills, and which of them is primary.
A *role* is what a device is: a firewall, a switch, a NAS. Real devices are
often several at once -- a QNAP is storage, hypervisor and Linux host -- so a
driver declares every role it fills by inheriting the matching base, and the
**order of those bases is the ranking**::
class QnapQtsDriver(StorageDriver, HypervisorDriver, LinuxDriver):
... # primary_role_of(...) == "storage"
That replaces the older arrangement, where an ``issubclass`` chain in netOrk
picked a single winner by a fixed precedence and a ``DEVICE_CLASS`` attribute
existed to overrule it. The author already states the ranking in the class
definition; nothing needs to restate it.
Whether a driver can perform a *specific operation* is a separate question with
a separate answer: ``hasattr``. Role bases declare their methods under
``if TYPE_CHECKING`` and implement nothing, so a method exists at runtime only
when a concrete driver provided it.
"""
from __future__ import annotations
from typing import Any, List, Optional, Type
def _is_role_base(cls: type) -> bool:
"""True for a class that introduces a role, not one that merely inherits one.
Keyed on ``ROLE`` appearing in the class's own ``__dict__``: a concrete
driver inherits the attribute but does not define it, and must not be
mistaken for a role base of its own.
"""
role = vars(cls).get("ROLE")
return isinstance(role, str) and bool(role)
def roles_of(driver_cls: type) -> List[Type[Any]]:
"""Every role base in ``driver_cls``'s MRO, most significant first.
MRO order is inheritance order, so the list reflects exactly what the driver
author wrote in the class definition.
"""
return [cls for cls in driver_cls.__mro__ if _is_role_base(cls)]
def role_keys_of(driver_cls: type) -> List[str]:
"""The stable string keys of :func:`roles_of`, e.g. ``["storage", "linux"]``."""
return [vars(cls)["ROLE"] for cls in roles_of(driver_cls)]
def primary_role_of(driver_cls: type) -> Optional[str]:
"""The driver's headline role, or ``None`` when it declares no role at all.
This is the value netOrk exposes as ``device_class``.
"""
keys = role_keys_of(driver_cls)
return keys[0] if keys else None
napalm_device_types/services.py
+55
View File
@@ -0,0 +1,55 @@
# -*- coding: utf-8 -*-
"""Init-system services: listing them and starting/stopping them.
Deliberately *not* the same thing as a NAS's exported shares, which live on
``StorageServiceCapability`` as ``get_storage_services``. Those two used to
share the name ``get_services`` with incompatible return types, which is why
a QNAP -- storage and OS at once -- could not satisfy both.
Declared under ``if TYPE_CHECKING``: these are contracts, not placeholders.
Nothing exists at runtime until a concrete driver implements it, so mixing
this class in can never shadow a working implementation from a sibling base.
"""
from __future__ import annotations
from typing import Any, Dict, List, TYPE_CHECKING
from napalm_device_types.models import ServiceDict
class ServiceControlMixin:
if TYPE_CHECKING:
def get_services(self) -> List[ServiceDict]:
"""
Returns the list of system services known to the device's init system.
Each entry contains:
* name (string) - service name as registered with the init system
* running (bool) - ``True`` if the service process is currently running
* enabled (bool) - ``True`` if the service starts automatically at boot
* pid (int) - process ID of the main service process; 0 if not running
Example::
[
{"name": "lldpd", "running": True, "enabled": True, "pid": 2341},
{"name": "sshd", "running": True, "enabled": True, "pid": 1198},
{"name": "cron", "running": False, "enabled": False, "pid": 0},
]
"""
...
def manage_service(self, name: str, action: str) -> Dict[str, Any]:
"""
Execute a lifecycle action on a named service.
:param name: Service name as returned by :meth:`get_services`.
:param action: One of ``start``, ``stop``, ``restart``, ``enable``, ``disable``.
:returns: ``{"success": bool, "output": str}``
:raises ValueError: If ``name`` or ``action`` is invalid.
:raises NotImplementedError: If the driver does not support service management.
"""
...
napalm_device_types/storage.py
+431 -531
View File
File diff suppressed because it is too large Load Diff
napalm_device_types/switch.py
+332 -358
View File
@@ -10,13 +10,13 @@ Usage::
...
"""
from typing import Dict, List
from typing import Any, Awaitable, Callable, Dict, List, TYPE_CHECKING
from napalm_device_types.base import DeviceTypeDriver
from napalm_device_types.mac_acl import MacAclMixin
from napalm_device_types.models import (
Dot1XPortDict,
HealthMetricsDict,
InterfaceConfigDict,
MACACLDict,
PoESummaryDict,
PortChannelDict,
SpanningTreeDict,
@@ -24,8 +24,7 @@ from napalm_device_types.models import (
)
class SwitchDriver(DeviceTypeDriver):
TYPE_LABEL: str = "Switch"
class SwitchDriver(MacAclMixin, DeviceTypeDriver):
"""
Abstract intermediate driver for Ethernet switches.
@@ -34,403 +33,378 @@ class SwitchDriver(DeviceTypeDriver):
operations that concrete drivers must implement.
"""
@classmethod
async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
"""Collect SNMP health metrics for this switch type.
#: Stable key netOrk exposes as ``device_class``. The order in which a
#: driver lists its role bases is the ranking; see
#: :func:`napalm_device_types.roles.primary_role_of`.
ROLE: str = "switch"
TYPE_LABEL: str = "Switch"
Switch vendors use proprietary OIDs — each concrete driver must
override this classmethod.
"""
raise NotImplementedError
if TYPE_CHECKING:
def get_spanning_tree(self) -> Dict[str, SpanningTreeDict]:
"""
Returns spanning tree status for each STP instance.
@classmethod
async def get_health_metrics(
cls,
snmp_get: Callable[..., Awaitable[Any]],
snmp_walk: Callable[..., Awaitable[Dict[str, Any]]],
) -> HealthMetricsDict:
"""Collect SNMP health metrics for this switch type.
Keys are STP instance identifiers (e.g. VLAN IDs for PVST,
``"MST0"`` for MSTP, or ``"0"`` for a single instance).
Each value contains:
Switch vendors use proprietary OIDs — each concrete driver must
override this classmethod.
"""
...
* 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:
def get_spanning_tree(self) -> Dict[str, SpanningTreeDict]:
"""
Returns spanning tree status for each STP instance.
* role (string) - ``"root"``, ``"designated"``, ``"alternate"``, ``"backup"``
* state (string) - ``"forwarding"``, ``"blocking"``, ``"learning"``, ``"listening"``
* cost (int) - port path cost
* port_priority (int) - port priority
Keys are STP instance identifiers (e.g. VLAN IDs for PVST,
``"MST0"`` for MSTP, or ``"0"`` for a single instance).
Each value contains:
Example::
* 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:
{
"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,
* 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,
},
},
"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.
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:
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
* 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::
Example::
{
"Port-Channel1": {
"members": ["GigabitEthernet0/1", "GigabitEthernet0/2"],
"protocol": "LACP",
"min_links": 1,
"is_up": True,
{
"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:
def get_dot1x_ports(self) -> Dict[str, Dot1XPortDict]:
"""
Returns the 802.1X / NAC configuration per switch port.
* name (string) - interface name (repeated for convenience)
* policy (string) - ACL mode:
Keys are interface names. Each value contains:
* ``"allow"`` – whitelist: only listed MACs may use this port
* ``"deny"`` – blacklist: listed MACs are blocked
* ``"disabled"`` – no MAC filtering active
* enabled (bool) - whether 802.1X is active on this port
* port_control (string) - authentication mode:
* entries (list) - ACL entries, each with:
* ``"auto"`` – port authenticates normally
* ``"force-authorized"`` – port always passes traffic (bypass)
* ``"force-unauthorized"`` – port always blocks traffic
* mac (string) - MAC address (normalised, colon-separated)
* action (string) - ``"allow"`` or ``"deny"``
* description (string) - optional human-readable label
* host_mode (string) - how many identities are authenticated per port:
Example::
* ``"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
{
"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
* 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)
def get_dot1x_config(self) -> Dict[str, Dot1XPortDict]:
"""
Returns the 802.1X / NAC configuration per switch port.
Note: RADIUS shared secrets are intentionally omitted.
Keys are interface names. Each value contains:
Example::
* 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,
"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,
},
"GigabitEthernet0/2": {
}
"""
...
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,
},
},
}
"""
...
def set_vlan(self, vlan_id: int, config: VlanConfigDict) -> None:
"""
Creates or updates a VLAN on the switch.
If the VLAN does not yet exist it is created first. Only the keys
present in *config* are applied; omitted keys leave the existing VLAN
configuration untouched.
:param vlan_id: VLAN ID (1–4094).
:param config: A (partial) :class:`~napalm_device_types.models.VlanConfigDict`
containing the fields to set. Supported keys:
* ``name`` (str) – human-readable VLAN name
* ``active`` (bool) – ``True`` = active, ``False`` = suspended
* ``interfaces`` (list of str) – access-port names to assign to
this VLAN (replaces the current membership list)
:raises NotImplementedError: If the driver does not implement this method.
:raises ValueError: If *vlan_id* is out of range or a field value is invalid.
Example – create VLAN 10 with a name::
driver.set_vlan(10, {"name": "Workstations", "active": True})
Example – assign ports to an existing VLAN::
driver.set_vlan(
10,
{"interfaces": ["GigabitEthernet0/1", "GigabitEthernet0/2"]},
)
"""
...
def delete_vlan(self, vlan_id: int) -> None:
"""
Removes a VLAN from the switch.
All ports that were assigned to this VLAN as their access VLAN are
moved to the default VLAN (1) by the driver before deletion. Trunk
ports that carry this VLAN will have it removed from their allowed
VLAN list.
:param vlan_id: VLAN ID (1–4094) to delete.
:raises NotImplementedError: If the driver does not implement this method.
:raises ValueError: If *vlan_id* is out of range or the VLAN does not
exist on the device.
Example::
driver.delete_vlan(10)
"""
...
def set_interface(self, interface: str, config: InterfaceConfigDict) -> None:
"""
Applies configuration to a single switch interface.
Only the keys present in *config* are changed; omitted keys leave the
current device configuration untouched.
:param interface: Interface name (e.g. ``"GigabitEthernet0/1"``).
:param config: A (partial) :class:`~napalm_device_types.models.InterfaceConfigDict`
containing the fields to update. Supported keys:
* ``description`` (str) – human-readable port label
* ``enabled`` (bool) – administrative state
* ``speed`` (int) – link speed in Mbps; ``0`` = auto-negotiate
* ``duplex`` (str) – ``"full"``, ``"half"``, or ``"auto"``
* ``mtu`` (int) – maximum transmission unit in bytes
* ``mode`` (str) – ``"access"``, ``"trunk"``, or ``"routed"``
* ``access_vlan`` (int) – untagged VLAN; relevant when mode is ``"access"``
* ``voice_vlan`` (int) – voice VLAN ID (``0`` = disabled)
* ``trunk_vlans`` (list of int) – tagged VLANs; empty = allow all
* ``native_vlan`` (int) – native VLAN on trunk ports
:raises NotImplementedError: If the driver does not implement this method.
:raises ValueError: If *interface* does not exist or an invalid value is
supplied for a configuration field.
Example – convert port to access VLAN 10 and add a description::
driver.set_interface(
"GigabitEthernet0/1",
{
"description": "Workstation port",
"enabled": True,
"status": "searching",
"poe_class": "unknown",
"power_draw": 0.0,
"power_budget": 30.0,
"voltage": 0.0,
"current": 0.0,
"mode": "access",
"access_vlan": 10,
},
"GigabitEthernet0/3": {
"enabled": False,
"status": "disabled",
"poe_class": "unknown",
"power_draw": 0.0,
"power_budget": 0.0,
"voltage": 0.0,
"current": 0.0,
)
Example – configure a trunk port::
driver.set_interface(
"GigabitEthernet0/2",
{
"mode": "trunk",
"trunk_vlans": [10, 20, 30],
"native_vlan": 1,
},
},
}
"""
raise NotImplementedError
)
"""
...
def set_vlan(self, vlan_id: int, config: VlanConfigDict) -> None:
"""
Creates or updates a VLAN on the switch.
def set_lag_members(self, lag_name: str, members: List[str]) -> None:
"""
Sets the full member-port list of a LAG/trunk interface.
If the VLAN does not yet exist it is created first. Only the keys
present in *config* are applied; omitted keys leave the existing VLAN
configuration untouched.
Diffs *members* against the LAG's current members (as reported by
``get_interfaces()``'s ``lag_members`` field) and adds/removes ports
on the device to match. The LAG itself must already exist on the
device; this method only manages its membership.
:param vlan_id: VLAN ID (1–4094).
:param config: A (partial) :class:`~napalm_device_types.models.VlanConfigDict`
containing the fields to set. Supported keys:
:param lag_name: Name of the LAG/trunk interface as returned by
``get_interfaces()`` (e.g. ``"Trk1"``, ``"ch1"``, ``"Lag1"``).
:param members: Full desired list of member port names.
:raises NotImplementedError: If the driver does not implement this method.
:raises ValueError: If *lag_name* does not refer to a valid LAG.
* ``name`` (str) – human-readable VLAN name
* ``active`` (bool) – ``True`` = active, ``False`` = suspended
* ``interfaces`` (list of str) – access-port names to assign to
this VLAN (replaces the current membership list)
Example::
:raises NotImplementedError: If the driver does not implement this method.
:raises ValueError: If *vlan_id* is out of range or a field value is invalid.
driver.set_lag_members("Lag1", ["GigabitEthernet0/1", "GigabitEthernet0/2"])
"""
...
Example – create VLAN 10 with a name::
def set_poe_enabled(self, interface: str, enabled: bool) -> None:
"""
Administratively enables or disables PoE on a single port.
driver.set_vlan(10, {"name": "Workstations", "active": True})
: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 – assign ports to an existing VLAN::
Example::
driver.set_vlan(
10,
{"interfaces": ["GigabitEthernet0/1", "GigabitEthernet0/2"]},
)
"""
raise NotImplementedError
driver.set_poe_enabled("GigabitEthernet0/1", False) # cut power
driver.set_poe_enabled("GigabitEthernet0/1", True) # restore
"""
...
def delete_vlan(self, vlan_id: int) -> None:
"""
Removes a VLAN from the switch.
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.
All ports that were assigned to this VLAN as their access VLAN are
moved to the default VLAN (1) by the driver before deletion. Trunk
ports that carry this VLAN will have it removed from their allowed
VLAN list.
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 vlan_id: VLAN ID (1–4094) to delete.
:raises NotImplementedError: If the driver does not implement this method.
:raises ValueError: If *vlan_id* is out of range or the VLAN does not
exist on the device.
: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::
Example::
driver.delete_vlan(10)
"""
raise NotImplementedError
def set_interface(self, interface: str, config: InterfaceConfigDict) -> None:
"""
Applies configuration to a single switch interface.
Only the keys present in *config* are changed; omitted keys leave the
current device configuration untouched.
:param interface: Interface name (e.g. ``"GigabitEthernet0/1"``).
:param config: A (partial) :class:`~napalm_device_types.models.InterfaceConfigDict`
containing the fields to update. Supported keys:
* ``description`` (str) – human-readable port label
* ``enabled`` (bool) – administrative state
* ``speed`` (int) – link speed in Mbps; ``0`` = auto-negotiate
* ``duplex`` (str) – ``"full"``, ``"half"``, or ``"auto"``
* ``mtu`` (int) – maximum transmission unit in bytes
* ``mode`` (str) – ``"access"``, ``"trunk"``, or ``"routed"``
* ``access_vlan`` (int) – untagged VLAN; relevant when mode is ``"access"``
* ``voice_vlan`` (int) – voice VLAN ID (``0`` = disabled)
* ``trunk_vlans`` (list of int) – tagged VLANs; empty = allow all
* ``native_vlan`` (int) – native VLAN on trunk ports
:raises NotImplementedError: If the driver does not implement this method.
:raises ValueError: If *interface* does not exist or an invalid value is
supplied for a configuration field.
Example – convert port to access VLAN 10 and add a description::
driver.set_interface(
"GigabitEthernet0/1",
{
"description": "Workstation port",
"enabled": True,
"mode": "access",
"access_vlan": 10,
},
)
Example – configure a trunk port::
driver.set_interface(
"GigabitEthernet0/2",
{
"mode": "trunk",
"trunk_vlans": [10, 20, 30],
"native_vlan": 1,
},
)
"""
raise NotImplementedError
def set_lag_members(self, lag_name: str, members: List[str]) -> None:
"""
Sets the full member-port list of a LAG/trunk interface.
Diffs *members* against the LAG's current members (as reported by
``get_interfaces()``'s ``lag_members`` field) and adds/removes ports
on the device to match. The LAG itself must already exist on the
device; this method only manages its membership.
:param lag_name: Name of the LAG/trunk interface as returned by
``get_interfaces()`` (e.g. ``"Trk1"``, ``"ch1"``, ``"Lag1"``).
:param members: Full desired list of member port names.
:raises NotImplementedError: If the driver does not implement this method.
:raises ValueError: If *lag_name* does not refer to a valid LAG.
Example::
driver.set_lag_members("Lag1", ["GigabitEthernet0/1", "GigabitEthernet0/2"])
"""
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
driver.power_cycle_port("GigabitEthernet0/1") # 5 s off
driver.power_cycle_port("GigabitEthernet0/1", delay=15) # 15 s off
"""
...
napalm_device_types/updates.py
+73
View File
@@ -0,0 +1,73 @@
# -*- coding: utf-8 -*-
"""Pending package updates and applying them.
The two device types that offer this used to declare it under two different
names -- ``get_available_updates`` and ``get_pending_updates`` -- with
``napalm-linux`` carrying an alias between them so netOrk could call either.
One name now, the one netOrk and four drivers already used.
Declared under ``if TYPE_CHECKING``: these are contracts, not placeholders.
Nothing exists at runtime until a concrete driver implements it, so mixing
this class in can never shadow a working implementation from a sibling base.
"""
from __future__ import annotations
from typing import List, TYPE_CHECKING
from napalm_device_types.models import ApplyUpdatesResultDict, UpdateDict
class UpdateMixin:
if TYPE_CHECKING:
def get_available_updates(self) -> List[UpdateDict]:
"""
Returns a list of packages that have a newer version available.
Each entry contains:
* name (string) - package name
* current_version (string) - currently installed version
* new_version (string) - version available in the repository
Example::
[
{
"name": "openssh-server",
"current_version": "1:9.2p1-2+deb12u1",
"new_version": "1:9.2p1-2+deb12u2",
},
]
"""
...
def apply_updates(self, packages: List[str]) -> ApplyUpdatesResultDict:
"""
Upgrades the given packages to the newest available version.
Only packages that are already installed may be upgraded; this method
does **not** install new packages. Pass an empty list to upgrade
**all** packages that have pending updates.
:param packages: List of package names to upgrade. Each name must
match ``^[a-zA-Z0-9_\\-\\+\\.]+$``; a :exc:`ValueError` is raised
for any name that does not conform.
:returns: A dict with:
* success (bool) – ``True`` if the package manager exited without error
* output (string) – combined stdout / stderr from the package manager
* error (string, optional) – short error message when *success* is ``False``
:raises ValueError: If any package name fails the safety check.
Example::
result = driver.apply_updates(["openssh-server", "curl"])
# → {"success": True, "output": "Reading package lists...\\n..."}
# Upgrade everything:
result = driver.apply_updates([])
"""
...
pyproject.toml
+3 -4
View File
@@ -4,10 +4,10 @@ build-backend = "setuptools.build_meta"
[project]
name = "napalm-device-types"
version = "0.5.0"
version = "1.0.0"
description = "Abstract device-type base classes for NAPALM drivers"
readme = "README.md"
requires-python = ">=3.9"
requires-python = ">=3.10"
license = { text = "Apache-2.0" }
authors = [
{ name = "Christian Manivong", email = "christian@manivong.de" },
@@ -31,7 +31,6 @@ classifiers = [
"License :: OSI Approved :: Apache Software License",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
@@ -61,6 +60,6 @@ where = ["."]
include = ["napalm_device_types*"]
[tool.mypy]
python_version = "3.9"
python_version = "3.10"
strict = true
ignore_missing_imports = true
tests/test_dhcp_diff_apply.py
+23 -10
View File
@@ -73,18 +73,31 @@ class TestNormalizeMac:
class TestAbstractContract:
def test_get_dhcp_reservations_raises_not_implemented_by_default(self):
with pytest.raises(NotImplementedError):
"""The device hooks are declared for type checkers, not implemented.
A driver that never implemented them does not carry them at runtime, so
`hasattr` is a truthful capability probe and the declaration cannot shadow a
working implementation inherited from a sibling base.
"""
@pytest.mark.parametrize(
"method",
[
"get_dhcp_reservations",
"apply_dhcp_reservation",
"commit_dhcp_reservations",
"get_dhcp_subnets",
"apply_dhcp_subnet",
"commit_dhcp_subnets",
],
)
def test_hook_absent_until_a_driver_implements_it(self, method):
assert not hasattr(DhcpServerMixin, method)
def test_calling_a_missing_hook_fails_loudly(self):
with pytest.raises(AttributeError):
DhcpServerMixin().get_dhcp_reservations()
def test_apply_dhcp_reservation_raises_not_implemented_by_default(self):
with pytest.raises(NotImplementedError):
DhcpServerMixin().apply_dhcp_reservation(_reservation())
def test_commit_dhcp_reservations_raises_not_implemented_by_default(self):
with pytest.raises(NotImplementedError):
DhcpServerMixin().commit_dhcp_reservations()
class TestMixedIntoDriverTypes:
"""Both firewalls and residential gateways run DHCP servers, so the mixin
tests/test_dhcp_subnet_diff_apply.py
+9 -9
View File
@@ -216,13 +216,13 @@ class _BareDriver(DhcpServerMixin):
@pytest.mark.parametrize(
"call",
[
lambda d: d.get_dhcp_subnets(),
lambda d: d.apply_dhcp_subnet({}), # type: ignore[typeddict-item]
lambda d: d.commit_dhcp_subnets(),
],
"method", ["get_dhcp_subnets", "apply_dhcp_subnet", "commit_dhcp_subnets"]
)
def test_device_specific_methods_raise_not_implemented(call: Any) -> None:
with pytest.raises(NotImplementedError):
call(_BareDriver())
def test_device_specific_methods_are_absent_until_implemented(method: str) -> None:
"""Declared under TYPE_CHECKING, so they do not exist until a driver adds them."""
assert not hasattr(_BareDriver, method)
def test_calling_a_missing_device_method_fails_loudly() -> None:
with pytest.raises(AttributeError):
_BareDriver().get_dhcp_subnets()
tests/test_firewall_diff_apply.py
+19 -18
View File
@@ -59,30 +59,31 @@ class _FakeFirewall(FirewallDriver):
class TestAbstractContract:
def test_get_firewall_rules_raises_not_implemented_by_default(self):
"""The three device hooks are declared, never implemented, on the base.
They exist for type checkers only, so a driver that did not implement them
does not carry them at runtime -- which is what keeps `hasattr` truthful and
stops the declaration from overriding a sibling base's working version.
"""
@pytest.mark.parametrize(
"method", ["get_firewall_rules", "apply_firewall_rule", "commit_firewall_rules"]
)
def test_hook_absent_until_a_driver_implements_it(self, method):
class _Bare(FirewallDriver):
def __init__(self) -> None:
pass
with pytest.raises(NotImplementedError):
assert not hasattr(_Bare, method)
def test_calling_a_missing_hook_fails_loudly(self):
class _Bare(FirewallDriver):
def __init__(self) -> None:
pass
with pytest.raises(AttributeError):
_Bare().get_firewall_rules()
def test_apply_firewall_rule_raises_not_implemented_by_default(self):
class _Bare(FirewallDriver):
def __init__(self) -> None:
pass
with pytest.raises(NotImplementedError):
_Bare().apply_firewall_rule(_rule())
def test_commit_firewall_rules_raises_not_implemented_by_default(self):
class _Bare(FirewallDriver):
def __init__(self) -> None:
pass
with pytest.raises(NotImplementedError):
_Bare().commit_firewall_rules()
class TestDiffFirewallRules:
def test_desired_rule_missing_live_is_an_add(self):
tests/test_role_contracts.py
+135
View File
@@ -0,0 +1,135 @@
# -*- coding: utf-8 -*-
"""The contract that makes multi-role drivers safe.
A role base declares what a device of that kind can be asked for. It must not
*implement* anything -- not even a placeholder. A ``NotImplementedError`` stub
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. That
failure has hit this codebase three times (OpenWrt's seven forwarding methods,
QNAP's ``get_services``, OpenMediaVault avoiding ``StorageDriver`` altogether).
These tests pin the property that prevents a fourth.
"""
from __future__ import annotations
import inspect
import pytest
from napalm_device_types import (
AccessPointDriver,
DeviceTypeDriver,
FirewallDriver,
HypervisorDriver,
OSDriver,
ResidentialGatewayDriver,
StorageDriver,
SwitchDriver,
)
from napalm_device_types.roles import primary_role_of, roles_of
ROLE_BASES = [
AccessPointDriver,
FirewallDriver,
HypervisorDriver,
OSDriver,
ResidentialGatewayDriver,
StorageDriver,
SwitchDriver,
]
@pytest.mark.parametrize("base", ROLE_BASES, ids=lambda b: b.__name__)
class TestRoleBasesAreContractsOnly:
def test_defines_no_methods_at_runtime(self, base):
"""A role base is a declaration. Anything callable it owns can shadow a
sibling base, so it must own nothing callable at all."""
own = [
name
for name, val in vars(base).items()
if not name.startswith("__")
and (inspect.isfunction(val) or isinstance(val, (classmethod, staticmethod)))
]
assert own == [], f"{base.__name__} implements {own}; move it to a function class"
def test_declares_a_role_key(self, base):
assert isinstance(vars(base).get("ROLE"), str) and vars(base)["ROLE"]
def test_has_a_real_docstring(self, base):
"""TYPE_LABEL used to be assigned above the triple-quoted string, which
made it a bare expression rather than a docstring -- __doc__ was None on
all seven bases, killing help() and IDE hovers."""
assert base.__doc__ and base.__doc__.strip()
class TestDeclaredMethodsDoNotExistAtRuntime:
"""`hasattr` is netOrk's capability probe. It only tells the truth when a
declared-but-unimplemented method is genuinely absent."""
@pytest.mark.parametrize(
("base", "method"),
[
(StorageDriver, "get_disks"),
(StorageDriver, "get_volumes"),
(HypervisorDriver, "get_vms"),
(FirewallDriver, "send_wake_on_lan"),
(SwitchDriver, "set_vlan"),
(AccessPointDriver, "get_wireless_config"),
(OSDriver, "get_processes"),
(ResidentialGatewayDriver, "get_wan_status"),
],
)
def test_absent_until_a_driver_implements_it(self, base, method):
assert not hasattr(base, method)
class TestNoShadowingAcrossRoles:
"""The regression test for the bug class."""
def test_role_base_listed_first_does_not_shadow_sibling(self):
"""StorageDriver precedes the working mixin -- the order that broke QNAP."""
class WorkingPackages:
def get_packages(self):
return [{"name": "vim", "version": "9.0"}]
def get_services(self):
return [{"name": "sshd", "state": "running"}]
class Combined(StorageDriver, WorkingPackages):
pass
assert Combined.get_packages is WorkingPackages.get_packages
assert Combined.get_services is WorkingPackages.get_services
def test_two_role_bases_can_be_combined(self):
"""A QNAP is NAS, hypervisor and Linux host. It must be able to say so."""
class Nas(StorageDriver, HypervisorDriver, OSDriver):
pass
assert [r.__name__ for r in roles_of(Nas)] == [
"StorageDriver",
"HypervisorDriver",
"OSDriver",
]
class TestRoleIntrospection:
def test_primary_role_follows_base_order(self):
class Nas(StorageDriver, OSDriver):
pass
class Host(OSDriver, StorageDriver):
pass
assert primary_role_of(Nas) == "storage"
assert primary_role_of(Host) == "linux"
def test_driver_without_a_role_has_none(self):
class Bare(DeviceTypeDriver):
pass
assert roles_of(Bare) == []
assert primary_role_of(Bare) is None
tests/test_send_wake_on_lan.py
+17 -8
View File
@@ -1,4 +1,12 @@
"""Tests for FirewallDriver.send_wake_on_lan default contract."""
"""Tests for the FirewallDriver.send_wake_on_lan contract.
``send_wake_on_lan`` is declared on ``FirewallDriver`` but implemented by very
few firewalls. It used to exist as a ``NotImplementedError`` stub; it is now a
``TYPE_CHECKING`` declaration, so a driver that never implemented it simply does
not have the attribute. That is what lets ``hasattr`` answer honestly, and what
stops the declaration from shadowing a working implementation inherited from a
sibling base.
"""
import pytest
from napalm_device_types import FirewallDriver
@@ -13,20 +21,21 @@ class _BareFirewall(FirewallDriver):
pass
def test_send_wake_on_lan_raises_not_implemented_by_default():
with pytest.raises(NotImplementedError):
def test_absent_on_a_driver_that_never_implemented_it():
assert not hasattr(_BareFirewall, "send_wake_on_lan")
def test_calling_it_anyway_fails_loudly():
"""A caller that skips the hasattr check must not get silence."""
with pytest.raises(AttributeError):
_BareFirewall().send_wake_on_lan("AA:BB:CC:DD:EE:FF")
def test_send_wake_on_lan_accepts_optional_interface():
with pytest.raises(NotImplementedError):
_BareFirewall().send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
def test_subclass_can_implement_send_wake_on_lan():
class MyFirewall(_BareFirewall):
def send_wake_on_lan(self, mac_address: str, interface: str = ""):
return {"success": True, "output": f"woke {mac_address} via {interface}"}
assert hasattr(MyFirewall, "send_wake_on_lan")
result = MyFirewall().send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
assert result == {"success": True, "output": "woke AA:BB:CC:DD:EE:FF via lan"}