From b03e4355c95c3a5080d611a025f15742cf1c06eb Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Mon, 11 May 2026 21:31:52 +0200 Subject: [PATCH] Initial release (v0.1.0) Add abstract device-type base classes for NAPALM drivers: - AccessPointDriver (wireless APs) - SwitchDriver (Ethernet switches) - FirewallDriver (firewalls / UTM) - HypervisorDriver (Proxmox VE, ESXi, KVM, Hyper-V) - StorageDriver (NAS/SAN appliances) All return types modelled as TypedDicts in napalm_device_types.models. --- .gitignore | 63 +++ LICENSE | 191 +++++++++ README.md | 160 ++++++++ napalm_device_types/__init__.py | 37 ++ napalm_device_types/access_point.py | 530 ++++++++++++++++++++++++ napalm_device_types/firewall.py | 285 +++++++++++++ napalm_device_types/hypervisor.py | 523 ++++++++++++++++++++++++ napalm_device_types/models.py | 430 ++++++++++++++++++++ napalm_device_types/storage.py | 599 ++++++++++++++++++++++++++++ napalm_device_types/switch.py | 298 ++++++++++++++ pyproject.toml | 66 +++ 11 files changed, 3182 insertions(+) create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 README.md create mode 100644 napalm_device_types/__init__.py create mode 100644 napalm_device_types/access_point.py create mode 100644 napalm_device_types/firewall.py create mode 100644 napalm_device_types/hypervisor.py create mode 100644 napalm_device_types/models.py create mode 100644 napalm_device_types/storage.py create mode 100644 napalm_device_types/switch.py create mode 100644 pyproject.toml diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0b4053e --- /dev/null +++ b/.gitignore @@ -0,0 +1,63 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# Virtual environments +.venv/ +venv/ +ENV/ +env/ +.env + +# Testing +.tox/ +.nox/ +.pytest_cache/ +htmlcov/ +.coverage +.coverage.* +coverage.xml +*.cover +*.py,cover +nosetests.xml +test-results/ + +# Type checking +.mypy_cache/ +.dmypy.json +dmypy.json +.pytype/ + +# IDE +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# macOS +.DS_Store + +# Logs +*.log diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..68d39ec --- /dev/null +++ b/LICENSE @@ -0,0 +1,191 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship made available under + the License, as indicated by a copyright notice that is included in + or attached to the work (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other transformations + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean, as submitted to the Licensor for inclusion + in the Work by the copyright owner or by an individual or Legal Entity + authorized to submit on behalf of the copyright owner. For the purposes + of this definition, "submitted" means any form of electronic, verbal, + or written communication sent to the Licensor or its representatives, + including but not limited to communication on electronic mailing lists, + source code control systems, and issue tracking systems that are managed + by, or on behalf of, the Licensor for the purpose of discussing and + improving the Work, but excluding communication that is conspicuously + marked or designated in writing by the copyright owner as "Not a + Contribution." + + "Contributor" shall mean Licensor and any Legal Entity on behalf of + whom a Contribution has been received by the Licensor and incorporated + within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by the combined Contribution(s) and the Work + to which such Contribution(s) was submitted. If You institute patent + litigation against any entity (including a cross-claim or counterclaim + in a lawsuit) alleging that the Work or any Contribution embodied + within the Work constitutes direct or contributory patent infringement, + then any patent licenses granted to You under this License for that + Work shall terminate as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or Derivative + Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, You must include a readable copy of the + attribution notices contained within such NOTICE file, in + at least one of the following places: within a NOTICE text file + distributed as part of the Derivative Works; within the Source + form or documentation, if provided along with the Derivative + Works; or, within a display generated by the Derivative Works, + if and wherever such third-party notices normally appear. The + contents of the NOTICE file are for informational purposes only + and do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own license statement for Your modifications and + may provide additional grant of rights to use, copy, modify, merge, + publish, distribute, sublicense, and/or sell copies of the Work, + and to permit persons to whom the Work is furnished to do so, + subject to the following conditions: + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any conditions of TITLE, + MERCHANTIBILITY, or FITNESS FOR A PARTICULAR PURPOSE. + See the License for the specific language governing permissions and + limitations under the License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or exemplary damages of any character arising as a + result of this License or out of the use or inability to use the + Work (even if such Contributor has been advised of the possibility + of such damages). + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may offer only + conditions consistent with this License, subject to the following + additional conditions: + + (a) You may offer additional warranty, indemnity, or other liability + terms and conditions; and + + (b) You may offer support for the Work in return for a fee. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format in question. It may also be + possible to make it available with a creative commons license. + + Copyright 2026 Christian Manivong + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/README.md b/README.md new file mode 100644 index 0000000..4ada66b --- /dev/null +++ b/README.md @@ -0,0 +1,160 @@ +# napalm-device-types + +Abstract intermediate device-type base classes for [NAPALM](https://napalm.readthedocs.io/) drivers. + +Instead of inheriting directly from `napalm.base.NetworkDriver`, a driver can inherit from one of the device-type classes here to gain a richer, type-specific contract: + +```python +# without napalm-device-types +class OpenWrtDriver(NetworkDriver): + ... + +# with napalm-device-types +from napalm_device_types import AccessPointDriver + +class OpenWrtDriver(AccessPointDriver): + ... +``` + +## Why? + +NAPALM's `NetworkDriver` defines a common interface for all network devices. In practice, devices fall into distinct categories with very different capabilities. A switch exposes spanning-tree and PoE data; a firewall exposes NAT tables and VPN tunnels; a NAS exposes disk pools and shares. Writing these methods directly in a concrete driver mixes concerns and makes drivers harder to discover and compare. + +`napalm-device-types` sits in between: it adds one well-typed layer of abstract methods per device category, so every driver for the same category exposes the same interface. + +## Installation + +```bash +pip install napalm-device-types +``` + +Requires Python ≥ 3.9 and NAPALM ≥ 4.0. + +## Available base classes + +| Class | Target devices | Example implementations | +|---|---|---| +| `AccessPointDriver` | Wireless access points | OpenWrt, Ubiquiti UniFi, Cisco Meraki AP | +| `SwitchDriver` | Ethernet switches | Cisco IOS, Arista EOS, Juniper EX | +| `FirewallDriver` | Firewalls & UTM appliances | pfSense, Fortinet FortiOS, Cisco ASA | +| `HypervisorDriver` | Hypervisors & virtualisation platforms | Proxmox VE, VMware ESXi, KVM/libvirt | +| `StorageDriver` | Storage appliances & NAS/SAN | TrueNAS, Synology DSM, QNAP QTS | + +## Usage + +### Access Point + +```python +from napalm_device_types import AccessPointDriver + +class OpenWrtDriver(AccessPointDriver): + + def get_wireless_clients(self): + # return List[WirelessClientDict] + ... + + def get_ssids(self): + # return Dict[str, SSIDDict] + ... +``` + +### Switch + +```python +from napalm_device_types import SwitchDriver + +class CiscoIOSDriver(SwitchDriver): + + def get_spanning_tree(self): + # return Dict[str, SpanningTreeDict] + ... + + def get_poe_status(self): + # return PoESummaryDict + ... +``` + +### Firewall + +```python +from napalm_device_types import FirewallDriver + +class PfSenseDriver(FirewallDriver): + + def get_nat_translations(self): + # return List[NATTranslationDict] + ... + + def get_vpn_tunnels(self): + # return Dict[str, VPNTunnelDict] + ... +``` + +### Hypervisor + +```python +from napalm_device_types import HypervisorDriver + +class ProxmoxDriver(HypervisorDriver): + + def get_vms(self): + # return List[VMDict] + ... + + def snapshot_create(self, name, snapshot, description="", include_memory=False): + ... +``` + +### Storage / NAS + +```python +from napalm_device_types import StorageDriver + +class TrueNASDriver(StorageDriver): + + def get_disks(self): + # return List[PhysicalDiskDict] + ... + + def get_shares(self): + # return Dict[str, NASShareDict] + ... +``` + +## Return types + +All return types are `TypedDict` classes defined in `napalm_device_types.models`. Import them directly for type annotations in your driver: + +```python +from napalm_device_types.models import ( + PhysicalDiskDict, + DiskPoolDict, + NASShareDict, + VolumeSnapshotDict, +) +``` + +## Development + +```bash +git clone https://github.com/chrismanivong/napalm-device-types.git +cd napalm-device-types +pip install -e ".[dev]" +``` + +Run type-checking: + +```bash +mypy napalm_device_types +``` + +## Contributing + +1. Fork the repository +2. Create a feature branch (`git checkout -b feature/router-driver`) +3. Implement your changes +4. Open a Pull Request + +## License + +Apache-2.0 – see [LICENSE](LICENSE) for details. diff --git a/napalm_device_types/__init__.py b/napalm_device_types/__init__.py new file mode 100644 index 0000000..8faa8aa --- /dev/null +++ b/napalm_device_types/__init__.py @@ -0,0 +1,37 @@ +""" +napalm-device-types +~~~~~~~~~~~~~~~~~~~ + +Abstract device-type base classes for NAPALM drivers. + +Instead of inheriting directly from ``napalm.base.NetworkDriver``, a driver +can inherit from one of the device-type classes defined here to gain +type-specific abstract methods and a clearer contract:: + + from napalm_device_types import AccessPointDriver + + class OpenWrtDriver(AccessPointDriver): + ... + +Available base classes: + +* :class:`~napalm_device_types.access_point.AccessPointDriver` +* :class:`~napalm_device_types.switch.SwitchDriver` +* :class:`~napalm_device_types.firewall.FirewallDriver` +* :class:`~napalm_device_types.hypervisor.HypervisorDriver` +* :class:`~napalm_device_types.storage.StorageDriver` +""" + +from napalm_device_types.access_point import AccessPointDriver +from napalm_device_types.firewall import FirewallDriver +from napalm_device_types.hypervisor import HypervisorDriver +from napalm_device_types.storage import StorageDriver +from napalm_device_types.switch import SwitchDriver + +__all__ = [ + "AccessPointDriver", + "FirewallDriver", + "HypervisorDriver", + "StorageDriver", + "SwitchDriver", +] diff --git a/napalm_device_types/access_point.py b/napalm_device_types/access_point.py new file mode 100644 index 0000000..32d1e38 --- /dev/null +++ b/napalm_device_types/access_point.py @@ -0,0 +1,530 @@ +""" +Abstract base class for wireless access point drivers. + +Usage:: + + from napalm_device_types import AccessPointDriver + + class OpenWrtDriver(AccessPointDriver): + def get_wireless_clients(self): + ... +""" + +from typing import Any, Dict, List +from napalm.base import NetworkDriver +from napalm_device_types.models import ( + Dot1XConfigDict, + FastTransitionConfigDict, + MACACLDict, + MeshConfigDict, + MeshPeerDict, + PackageDict, + RadioStatusDict, + SSIDBridgeDict, + SSIDDict, + WirelessClientDict, + WirelessConfigDict, +) + + +class AccessPointDriver(NetworkDriver): + """ + Abstract intermediate driver for wireless access points. + + Inherits all standard NAPALM NetworkDriver methods and adds + access-point-specific operations that concrete drivers must implement. + """ + + def get_wireless_clients(self) -> List[WirelessClientDict]: + """ + Returns a list of wireless clients currently associated with this + access point. + + Each entry contains: + + * mac (string) - client MAC address + * ssid (string) - SSID the client is connected to + * radio (string) - radio identifier (e.g. ``"radio0"``, ``"5GHz"``) + * signal (int) - received signal strength in dBm + * noise (int) - noise floor in dBm + * tx_rate (float) - TX bitrate in Mbit/s + * rx_rate (float) - RX bitrate in Mbit/s + * uptime (int) - association duration in seconds + + Example:: + + [ + { + "mac": "AA:BB:CC:DD:EE:FF", + "ssid": "MyNetwork", + "radio": "radio1", + "signal": -65, + "noise": -95, + "tx_rate": 300.0, + "rx_rate": 144.0, + "uptime": 3600, + } + ] + """ + raise NotImplementedError + + def get_ssids(self) -> Dict[str, SSIDDict]: + """ + Returns the configured SSIDs (VAPs) on this access point. + + Keys are SSID names. Each value contains: + + * enabled (bool) - whether the SSID is currently broadcasting + * radio (string) - radio the SSID is bound to + * bssid (string) - BSSID (MAC) of the VAP + * encryption (string) - e.g. ``"WPA2-PSK"``, ``"WPA3-SAE"``, ``"open"`` + * hidden (bool) - whether the SSID is hidden + * clients (int) - number of currently associated clients + + Example:: + + { + "MyNetwork": { + "enabled": True, + "radio": "radio1", + "bssid": "AA:BB:CC:DD:EE:F0", + "encryption": "WPA2-PSK", + "hidden": False, + "clients": 3, + }, + "GuestNet": { + "enabled": True, + "radio": "radio0", + "bssid": "AA:BB:CC:DD:EE:F1", + "encryption": "WPA2-PSK", + "hidden": False, + "clients": 1, + }, + } + """ + raise NotImplementedError + + def get_radio_status(self) -> Dict[str, RadioStatusDict]: + """ + Returns the status of each radio interface. + + Keys are radio identifiers (e.g. ``"radio0"``, ``"radio1"``). + Each value contains: + + * enabled (bool) - whether the radio is active + * band (string) - frequency band, e.g. ``"2.4GHz"``, ``"5GHz"``, ``"6GHz"`` + * channel (int) - operating channel number + * channel_width (int) - channel width in MHz (e.g. 20, 40, 80, 160) + * tx_power (int) - transmit power in dBm + * frequency (float) - center frequency in MHz + + Example:: + + { + "radio0": { + "enabled": True, + "band": "2.4GHz", + "channel": 6, + "channel_width": 20, + "tx_power": 20, + "frequency": 2437.0, + }, + "radio1": { + "enabled": True, + "band": "5GHz", + "channel": 36, + "channel_width": 80, + "tx_power": 23, + "frequency": 5180.0, + }, + } + """ + raise NotImplementedError + + def get_wireless_config(self) -> WirelessConfigDict: + """ + Returns global wireless configuration parameters that apply across + all radios and SSIDs. + + * country_code (string) - ISO 3166-1 alpha-2 country code (e.g. ``"DE"``) + * regulatory_domain (string) - regulatory domain string (e.g. ``"ETSI"``) + * beacon_interval (int) - beacon interval in TUs (default 100) + * dtim_period (int) - DTIM period (default 2) + * rts_threshold (int) - RTS/CTS threshold in bytes (2347 = disabled) + * fragmentation_threshold (int) - fragmentation threshold in bytes + * short_preamble (bool) - whether short preamble is enabled + * wmm_enabled (bool) - whether WMM/QoS is enabled + + Example:: + + { + "country_code": "DE", + "regulatory_domain": "ETSI", + "beacon_interval": 100, + "dtim_period": 2, + "rts_threshold": 2347, + "fragmentation_threshold": 2346, + "short_preamble": True, + "wmm_enabled": True, + } + """ + raise NotImplementedError + + def get_fast_transition_config(self) -> Dict[str, FastTransitionConfigDict]: + """ + Returns the 802.11r Fast BSS Transition (FT) configuration per SSID. + + Keys are SSID names. Each value contains: + + * enabled (bool) - whether FT is active on this SSID + * ssid (string) - SSID name (repeated for convenience) + * mobility_domain (string) - 4-hex-digit Mobility Domain ID (MDID) + * reassociation_deadline (int) - FT reassociation deadline in TUs + * r0_key_lifetime (int) - PMK-R0 key lifetime in minutes + * r1_key_holder (string) - R1 Key Holder identifier (MAC-like string) + * pmk_r1_push (bool) - whether PMK-R1 is proactively pushed to neighbours + * over_ds (bool) - whether FT over DS (instead of FT over air) is used + + Example:: + + { + "CorpWiFi": { + "enabled": True, + "ssid": "CorpWiFi", + "mobility_domain": "a1b2", + "reassociation_deadline": 1000, + "r0_key_lifetime": 10000, + "r1_key_holder": "00:11:22:33:44:55", + "pmk_r1_push": True, + "over_ds": False, + } + } + """ + raise NotImplementedError + + def get_mesh_config(self) -> Dict[str, MeshConfigDict]: + """ + Returns the 802.11s mesh configuration per mesh interface. + + Keys are mesh interface names (e.g. ``"mesh0"``). Each value contains: + + * enabled (bool) - whether the mesh interface is active + * radio (string) - underlying radio (e.g. ``"radio0"``) + * mesh_id (string) - 802.11s Mesh ID (analogous to SSID) + * path_metric (string) - path selection metric, e.g. ``"airtime"`` or ``"hopcount"`` + * gate_announcements (bool) - whether gate announcements (GANN) are sent + * is_gate (bool) - whether this node acts as a mesh gate to the DS + * encryption (string) - e.g. ``"SAE"``, ``"open"`` + + Example:: + + { + "mesh0": { + "enabled": True, + "radio": "radio1", + "mesh_id": "office-mesh", + "path_metric": "airtime", + "gate_announcements": True, + "is_gate": True, + "encryption": "SAE", + } + } + """ + raise NotImplementedError + + def get_mesh_peers(self) -> List[MeshPeerDict]: + """ + Returns a list of currently active 802.11s mesh peers. + + Each entry contains: + + * mac (string) - peer MAC address + * radio (string) - radio on which the peering was established + * signal (int) - received signal strength in dBm + * tx_rate (float) - TX bitrate to peer in Mbit/s + * rx_rate (float) - RX bitrate from peer in Mbit/s + * uptime (int) - peering duration in seconds + * hop_count (int) - number of hops to the mesh gate (0 = this node is the gate) + + Example:: + + [ + { + "mac": "AA:BB:CC:DD:EE:01", + "radio": "radio1", + "signal": -58, + "tx_rate": 300.0, + "rx_rate": 270.0, + "uptime": 7200, + "hop_count": 1, + } + ] + """ + raise NotImplementedError + + def get_ssid_bridge_config(self) -> Dict[str, SSIDBridgeDict]: + """ + Returns the Layer-2 bridging configuration for each SSID, i.e. which + bridge interface and VLAN each SSID is mapped to. + + Keys are SSID names. Each value contains: + + * ssid (string) - SSID name (repeated for convenience) + * bridge (string) - bridge interface the VAP is attached to (e.g. ``"br-lan"``, ``"br-guest"``) + * vlan_id (int) - 802.1Q VLAN ID (0 = untagged / no VLAN separation) + * tagged (bool) - whether traffic is 802.1Q-tagged on the uplink port + * client_isolation (bool) - whether clients on this SSID are isolated from each other + + Example:: + + { + "CorpWiFi": { + "ssid": "CorpWiFi", + "bridge": "br-corp", + "vlan_id": 10, + "tagged": True, + "client_isolation": False, + }, + "GuestNet": { + "ssid": "GuestNet", + "bridge": "br-guest", + "vlan_id": 20, + "tagged": True, + "client_isolation": True, + }, + "IoT": { + "ssid": "IoT", + "bridge": "br-iot", + "vlan_id": 30, + "tagged": True, + "client_isolation": True, + }, + } + """ + raise NotImplementedError + + def get_mac_acl(self) -> Dict[str, MACACLDict]: + """ + Returns the MAC-address-based access control lists configured per SSID. + + Keys are SSID names. Each value contains: + + * ssid (string) - SSID name (repeated for convenience) + * policy (string) - ACL mode: + + * ``"allow"`` – whitelist: only listed MACs may associate + * ``"deny"`` – blacklist: listed MACs are blocked + * ``"disabled"`` – no MAC filtering active + + * entries (list) - ACL entries, each with: + + * mac (string) - MAC address (normalised, colon-separated) + * action (string) - ``"allow"`` or ``"deny"`` + * description (string) - optional human-readable label + + Example:: + + { + "CorpWiFi": { + "name": "CorpWiFi", + "policy": "allow", + "entries": [ + {"mac": "AA:BB:CC:DD:EE:01", "action": "allow", "description": "CEO-Laptop"}, + {"mac": "AA:BB:CC:DD:EE:02", "action": "allow", "description": "CFO-Laptop"}, + ], + }, + "GuestNet": { + "name": "GuestNet", + "policy": "deny", + "entries": [ + {"mac": "DE:AD:BE:EF:00:01", "action": "deny", "description": "blocked device"}, + ], + }, + } + """ + raise NotImplementedError + + def get_dot1x_config(self) -> Dict[str, Dot1XConfigDict]: + """ + Returns the 802.1X / WPA-Enterprise (RADIUS) configuration per SSID. + + Keys are SSID names. Each value contains: + + * enabled (bool) - whether 802.1X authentication is active on this SSID + * ssid (string) - SSID name (repeated for convenience) + * auth_server (dict) - RADIUS authentication server: + + * host (string) - IP or FQDN of the RADIUS server + * port (int) - UDP port (default 1812) + * timeout (int) - request timeout in seconds + * retries (int) - number of retransmissions + + * acct_server (dict or None) - RADIUS accounting server (same keys as auth_server, + ``None`` if accounting is not configured) + * reauth_interval (int) - re-authentication interval in seconds (0 = disabled) + * pmksa_caching (bool) - whether PMKSA caching (opportunistic key caching) is enabled + + Note: The RADIUS shared secret is intentionally omitted from the return + value for security reasons. + + Example:: + + { + "CorpWiFi": { + "enabled": True, + "ssid": "CorpWiFi", + "auth_server": { + "host": "radius.corp.example", + "port": 1812, + "timeout": 5, + "retries": 3, + }, + "acct_server": { + "host": "radius.corp.example", + "port": 1813, + "timeout": 5, + "retries": 3, + }, + "reauth_interval": 3600, + "pmksa_caching": True, + } + } + """ + raise NotImplementedError + + def get_packages(self) -> List[PackageDict]: + """ + Returns all packages currently known to the device's package manager + (e.g. ``opkg`` on OpenWrt, ``apk`` on Alpine-based APs). + + Each entry contains: + + * name (string) - package name + * version (string) - installed or available version string + * installed (bool) - ``True`` if the package is currently installed + * description (string) - short package description + * size (int) - package size in bytes (0 if unknown) + * source (string) - repository / feed the package comes from + + Example:: + + [ + { + "name": "luci-app-statistics", + "version": "git-24.001.00000-1", + "installed": True, + "description": "LuCI Statistics application", + "size": 20480, + "source": "openwrt/packages", + }, + { + "name": "collectd-mod-wireless", + "version": "5.12.0-24", + "installed": False, + "description": "Wireless statistics plugin for collectd", + "size": 8192, + "source": "openwrt/packages", + }, + ] + """ + raise NotImplementedError + + def install_package(self, name: str, version: str = "") -> None: + """ + Installs a package on the device. + + The method blocks until the installation is complete. After it returns + successfully the package is available for use without a reboot + (where the underlying package manager supports this). + + :param name: Package name as known to the package manager. + :param version: Exact version to install. An empty string (default) + installs the latest available version. + :raises NotImplementedError: If the driver does not support package management. + :raises ValueError: If the package name is unknown or the version does + not exist in any configured feed. + :raises RuntimeError: If the installation fails on the device side + (e.g. dependency conflict, disk full). + + Example:: + + driver.install_package("luci-app-statistics") + driver.install_package("collectd", version="5.12.0-24") + """ + raise NotImplementedError + + def remove_package(self, name: str) -> None: + """ + Removes an installed package from the device. + + The method blocks until the removal is complete. + + :param name: Package name to remove. + :raises NotImplementedError: If the driver does not support package management. + :raises ValueError: If the package is not currently installed. + :raises RuntimeError: If the removal fails on the device side + (e.g. other packages depend on it). + + Example:: + + driver.remove_package("luci-app-statistics") + """ + raise NotImplementedError + + def get_package_config(self, name: str) -> Dict[str, Any]: + """ + Returns the current configuration of an installed package as a + dictionary. The structure is package-specific. + + :param name: Package name. + :raises NotImplementedError: If the driver does not support package management. + :raises ValueError: If the package is not installed. + + Example:: + + driver.get_package_config("luci-app-statistics") + # → + { + "collectd": { + "enabled": True, + "interval": 30, + }, + "rrdtool": { + "datadir": "/tmp/rrd", + "stepsize": 30, + "heartbeat": 60, + }, + } + """ + raise NotImplementedError + + def set_package_config(self, name: str, config: Dict[str, Any]) -> None: + """ + Writes a new configuration for an installed package. + + The ``config`` dict must match the structure returned by + :meth:`get_package_config`. Unknown keys are ignored or raise a + ``ValueError`` depending on the driver implementation. + + Changes take effect immediately where the package supports live + reload; otherwise a package restart or device reboot may be + required – behaviour is driver-specific. + + :param name: Package name. + :param config: New configuration as a nested dictionary. + :raises NotImplementedError: If the driver does not support package management. + :raises ValueError: If the package is not installed or the configuration + contains invalid values. + :raises RuntimeError: If the device rejects the configuration. + + Example:: + + driver.set_package_config( + "luci-app-statistics", + { + "collectd": {"enabled": True, "interval": 60}, + "rrdtool": {"datadir": "/tmp/rrd", "stepsize": 60, "heartbeat": 120}, + }, + ) + """ + raise NotImplementedError diff --git a/napalm_device_types/firewall.py b/napalm_device_types/firewall.py new file mode 100644 index 0000000..cdb4a5f --- /dev/null +++ b/napalm_device_types/firewall.py @@ -0,0 +1,285 @@ +""" +Abstract base class for firewall drivers. + +Usage:: + + from napalm_device_types import FirewallDriver + + class FortiGateDriver(FirewallDriver): + def get_security_zones(self): + ... +""" + +from typing import Any, Dict, List +from napalm.base import NetworkDriver +from napalm_device_types.models import ( + NATTranslationDict, + PackageDict, + SecurityZoneDict, + SessionDict, + VPNTunnelDict, +) + + +class FirewallDriver(NetworkDriver): + """ + Abstract intermediate driver for firewall/security devices. + + Inherits all standard NAPALM NetworkDriver methods (including + ``get_firewall_policies()``) and adds firewall-specific operations + that concrete drivers must implement. + """ + + def get_nat_translations(self) -> List[NATTranslationDict]: + """ + Returns a list of active NAT translation entries. + + Each entry contains: + + * protocol (string) - ``"tcp"``, ``"udp"``, ``"icmp"`` + * inside_local (string) - original source address (IP or IP:port) + * inside_global (string) - translated source address (IP or IP:port) + * outside_local (string) - destination as seen from inside + * outside_global (string) - actual destination address + * age (float) - translation entry age in seconds + + Example:: + + [ + { + "protocol": "tcp", + "inside_local": "192.168.1.10:54321", + "inside_global": "203.0.113.1:54321", + "outside_local": "1.1.1.1:443", + "outside_global": "1.1.1.1:443", + "age": 120.5, + } + ] + """ + raise NotImplementedError + + def get_security_zones(self) -> Dict[str, SecurityZoneDict]: + """ + Returns the security zone configuration. + + Keys are zone names. Each value contains: + + * interfaces (list of strings) - interfaces assigned to this zone + * policy (string) - name of the security policy applied to this zone + * description (string) - zone description + + Example:: + + { + "LAN": { + "interfaces": ["eth0", "eth1"], + "policy": "LAN-policy", + "description": "Internal LAN zone", + }, + "WAN": { + "interfaces": ["eth2"], + "policy": "WAN-policy", + "description": "Uplink to internet", + }, + } + """ + raise NotImplementedError + + def get_sessions(self) -> List[SessionDict]: + """ + Returns a list of active connection sessions (stateful flows). + + Each entry contains: + + * protocol (string) - ``"tcp"``, ``"udp"``, ``"icmp"`` + * src_ip (string) - source IP address + * src_port (int) - source port (0 for ICMP) + * dst_ip (string) - destination IP address + * dst_port (int) - destination port (0 for ICMP) + * state (string) - session state, e.g. ``"established"``, ``"syn_sent"`` + * age (float) - session age in seconds + + Example:: + + [ + { + "protocol": "tcp", + "src_ip": "192.168.1.10", + "src_port": 54321, + "dst_ip": "1.1.1.1", + "dst_port": 443, + "state": "established", + "age": 30.2, + } + ] + """ + raise NotImplementedError + + def get_vpn_tunnels(self) -> Dict[str, VPNTunnelDict]: + """ + Returns the status of VPN tunnels. + + Keys are tunnel names or identifiers. Each value contains: + + * type (string) - tunnel type: ``"IPsec"``, ``"SSL"``, ``"GRE"``, ``"WireGuard"`` + * local_endpoint (string) - local tunnel endpoint IP + * remote_endpoint (string) - remote tunnel endpoint IP + * is_up (bool) - whether the tunnel is operationally up + * uptime (int) - tunnel uptime in seconds (0 if down) + * bytes_in (int) - total bytes received through the tunnel + * bytes_out (int) - total bytes sent through the tunnel + + Example:: + + { + "vpn-to-branch": { + "type": "IPsec", + "local_endpoint": "203.0.113.1", + "remote_endpoint": "198.51.100.1", + "is_up": True, + "uptime": 86400, + "bytes_in": 104857600, + "bytes_out": 52428800, + } + } + """ + raise NotImplementedError + + def get_packages(self) -> List[PackageDict]: + """ + Returns all packages / plugins currently known to the firewall's + package manager (e.g. ``pkg`` on pfSense/OPNsense, ``FortiGate + License`` add-ons, ``apt`` on Debian-based firewalls). + + Each entry contains: + + * name (string) - package name + * version (string) - installed or available version string + * installed (bool) - ``True`` if the package is currently installed + * description (string) - short package description + * size (int) - package size in bytes (0 if unknown) + * source (string) - repository / channel the package comes from + + Example:: + + [ + { + "name": "pfBlockerNG", + "version": "3.2.0_4", + "installed": True, + "description": "IP and DNS blocking for pfSense", + "size": 2097152, + "source": "pfSense-pkg", + }, + { + "name": "suricata", + "version": "7.0.3_1", + "installed": False, + "description": "High-performance Network IDS/IPS", + "size": 51380224, + "source": "pfSense-pkg", + }, + ] + """ + raise NotImplementedError + + def install_package(self, name: str, version: str = "") -> None: + """ + Installs a package or plugin on the firewall. + + The method blocks until the installation is complete. Whether a + reboot is required afterwards depends on the device; check the vendor + documentation. + + :param name: Package name as known to the package manager. + :param version: Exact version to install. An empty string (default) + installs the latest available version. + :raises NotImplementedError: If the driver does not support package management. + :raises ValueError: If the package name is unknown or the requested + version is not available. + :raises RuntimeError: If the installation fails on the device side + (e.g. license missing, dependency conflict, disk full). + + Example:: + + driver.install_package("pfBlockerNG") + driver.install_package("suricata", version="7.0.3_1") + """ + raise NotImplementedError + + def remove_package(self, name: str) -> None: + """ + Removes an installed package or plugin from the firewall. + + The method blocks until the removal is complete. + + :param name: Package name to remove. + :raises NotImplementedError: If the driver does not support package management. + :raises ValueError: If the package is not currently installed. + :raises RuntimeError: If the removal fails on the device side + (e.g. the package is a system dependency). + + Example:: + + driver.remove_package("pfBlockerNG") + """ + raise NotImplementedError + + def get_package_config(self, name: str) -> Dict[str, Any]: + """ + Returns the current configuration of an installed package or plugin + as a dictionary. The structure is package-specific. + + :param name: Package name. + :raises NotImplementedError: If the driver does not support package management. + :raises ValueError: If the package is not installed. + + Example:: + + driver.get_package_config("pfBlockerNG") + # → + { + "enable": True, + "maxmind_key": "", + "blocklists": [ + {"name": "PRI1", "action": "Deny_Both", "enabled": True}, + {"name": "DNSBL_ADs", "action": "Unbound", "enabled": True}, + ], + "update_interval": "Once a day", + } + """ + raise NotImplementedError + + def set_package_config(self, name: str, config: Dict[str, Any]) -> None: + """ + Writes a new configuration for an installed package or plugin. + + The ``config`` dict must match the structure returned by + :meth:`get_package_config`. Unknown keys are ignored or raise a + ``ValueError`` depending on the driver implementation. + + Changes take effect immediately where the package supports live + reload; otherwise a package restart or device reboot may be + required – behaviour is driver-specific. + + :param name: Package name. + :param config: New configuration as a nested dictionary. + :raises NotImplementedError: If the driver does not support package management. + :raises ValueError: If the package is not installed or the configuration + contains invalid values. + :raises RuntimeError: If the device rejects the configuration. + + Example:: + + driver.set_package_config( + "pfBlockerNG", + { + "enable": True, + "blocklists": [ + {"name": "PRI1", "action": "Deny_Both", "enabled": True}, + ], + "update_interval": "Twice a day", + }, + ) + """ + raise NotImplementedError diff --git a/napalm_device_types/hypervisor.py b/napalm_device_types/hypervisor.py new file mode 100644 index 0000000..11ffd86 --- /dev/null +++ b/napalm_device_types/hypervisor.py @@ -0,0 +1,523 @@ +""" +Abstract base class for hypervisor drivers. + +Usage:: + + from napalm_device_types import HypervisorDriver + + class ProxmoxDriver(HypervisorDriver): + def get_vms(self): + ... +""" + +from typing import Any, Dict, List +from napalm.base import NetworkDriver +from napalm_device_types.models import ( + PackageDict, + SnapshotDict, + StorageVolumeDict, + VMConfigDict, + VMDict, + VirtualNetworkDict, +) + + +class HypervisorDriver(NetworkDriver): + """ + Abstract intermediate driver for hypervisors and virtualisation platforms + (e.g. Proxmox VE, VMware ESXi, KVM/libvirt, Hyper-V). + + Inherits all standard NAPALM NetworkDriver methods and adds + hypervisor-specific operations that concrete drivers must implement. + """ + + # ------------------------------------------------------------------ + # Virtual machines – read + # ------------------------------------------------------------------ + + def get_vms(self) -> List[VMDict]: + """ + Returns a list of all virtual machines known to this hypervisor, + including their runtime status. + + Each entry contains: + + * name (string) - VM display name + * vmid (int) - hypervisor-internal numeric ID + * status (string) - ``"running"``, ``"stopped"``, ``"paused"``, ``"suspended"`` + * vcpus (int) - number of virtual CPUs assigned + * memory (int) - configured RAM in megabytes + * cpu_usage (float) - current CPU utilisation 0.0–1.0 + * memory_usage (int) - current RAM usage in megabytes + * uptime (int) - uptime in seconds (0 if not running) + * node (string) - cluster node this VM lives on (empty string for standalone) + + Example:: + + [ + { + "name": "web01", + "vmid": 100, + "status": "running", + "vcpus": 4, + "memory": 8192, + "cpu_usage": 0.12, + "memory_usage": 3200, + "uptime": 864000, + "node": "pve1", + }, + { + "name": "db-backup", + "vmid": 101, + "status": "stopped", + "vcpus": 2, + "memory": 4096, + "cpu_usage": 0.0, + "memory_usage": 0, + "uptime": 0, + "node": "pve1", + }, + ] + """ + raise NotImplementedError + + def get_vm_config(self, name: str) -> VMConfigDict: + """ + Returns the full hardware configuration of a virtual machine. + + :param name: VM name or numeric VMID as a string. + :raises ValueError: If no VM with the given name/ID exists. + + The returned dictionary contains: + + * name (string) - VM display name + * vmid (int) - hypervisor-internal numeric ID + * vcpus (int) - number of virtual CPUs + * memory (int) - RAM in megabytes + * os_type (string) - guest OS type hint (e.g. ``"l26"``, ``"win11"``, ``"other"``) + * boot_order (list of strings) - boot device sequence (e.g. ``["scsi0", "net0"]``) + * disks (list) - attached virtual disks, each with: + + * device (string) - device ID (e.g. ``"scsi0"``) + * storage (string) - backing storage pool + * size (int) - disk size in gigabytes + * format (string) - image format: ``"qcow2"``, ``"raw"``, ``"vmdk"`` + * bootable (bool) - whether this disk is in the boot order + + * nics (list) - virtual network interfaces, each with: + + * device (string) - device ID (e.g. ``"net0"``) + * mac (string) - MAC address + * model (string) - NIC model (e.g. ``"virtio"``, ``"e1000"``) + * bridge (string) - host bridge the NIC is connected to + * vlan_id (int) - VLAN tag (0 = untagged) + + * description (string) - free-text notes / description + * tags (list of strings) - organisational tags + + Example:: + + { + "name": "web01", + "vmid": 100, + "vcpus": 4, + "memory": 8192, + "os_type": "l26", + "boot_order": ["scsi0"], + "disks": [ + { + "device": "scsi0", + "storage": "local-lvm", + "size": 32, + "format": "raw", + "bootable": True, + } + ], + "nics": [ + { + "device": "net0", + "mac": "BC:24:11:AA:BB:CC", + "model": "virtio", + "bridge": "vmbr0", + "vlan_id": 10, + } + ], + "description": "Production web server", + "tags": ["prod", "web"], + } + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Virtual machines – power actions + # ------------------------------------------------------------------ + + def start_vm(self, name: str) -> None: + """ + Powers on a stopped or suspended virtual machine. + + The method blocks until the hypervisor reports the VM as running. + + :param name: VM name or numeric VMID as a string. + :raises ValueError: If no VM with the given name/ID exists. + :raises RuntimeError: If the VM cannot be started (e.g. resource limit). + + Example:: + + driver.start_vm("web01") + """ + raise NotImplementedError + + def stop_vm(self, name: str, force: bool = False) -> None: + """ + Shuts down a virtual machine. + + With ``force=False`` (default) a graceful ACPI shutdown is requested + and the method blocks until the VM is stopped. With ``force=True`` + the VM is immediately powered off (equivalent to pulling the plug). + + :param name: VM name or numeric VMID as a string. + :param force: ``True`` for immediate power-off, ``False`` for graceful shutdown. + :raises ValueError: If no VM with the given name/ID exists. + :raises RuntimeError: If the VM is already stopped. + + Example:: + + driver.stop_vm("web01") # graceful + driver.stop_vm("web01", force=True) # hard off + """ + raise NotImplementedError + + def reboot_vm(self, name: str, force: bool = False) -> None: + """ + Reboots a virtual machine. + + With ``force=False`` (default) a graceful ACPI reboot is requested. + With ``force=True`` the VM is reset immediately without OS shutdown. + + :param name: VM name or numeric VMID as a string. + :param force: ``True`` for an immediate reset, ``False`` for graceful reboot. + :raises ValueError: If no VM with the given name/ID exists. + :raises RuntimeError: If the VM is not currently running. + + Example:: + + driver.reboot_vm("web01") + driver.reboot_vm("web01", force=True) + """ + raise NotImplementedError + + def suspend_vm(self, name: str) -> None: + """ + Suspends (pauses) a running virtual machine, preserving its in-memory + state. The VM can be resumed with :meth:`start_vm`. + + :param name: VM name or numeric VMID as a string. + :raises ValueError: If no VM with the given name/ID exists. + :raises RuntimeError: If the VM is not currently running. + + Example:: + + driver.suspend_vm("web01") + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Snapshots + # ------------------------------------------------------------------ + + def get_snapshots(self, name: str) -> List[SnapshotDict]: + """ + Returns all snapshots of a virtual machine. + + :param name: VM name or numeric VMID as a string. + :raises ValueError: If no VM with the given name/ID exists. + + Each entry contains: + + * name (string) - snapshot name + * vm (string) - VM name this snapshot belongs to + * created (float) - creation timestamp (Unix epoch) + * description (string) - optional snapshot description + * has_memory (bool) - whether the snapshot includes RAM state + * parent (string) - name of the parent snapshot (empty string for root) + + Example:: + + [ + { + "name": "before-upgrade", + "vm": "web01", + "created": 1746921600.0, + "description": "Clean state before kernel upgrade", + "has_memory": False, + "parent": "", + }, + { + "name": "post-upgrade", + "vm": "web01", + "created": 1746925200.0, + "description": "", + "has_memory": False, + "parent": "before-upgrade", + }, + ] + """ + raise NotImplementedError + + def snapshot_create(self, name: str, snapshot: str, + description: str = "", include_memory: bool = False) -> None: + """ + Creates a snapshot of a virtual machine. + + :param name: VM name or numeric VMID as a string. + :param snapshot: Name for the new snapshot. + :param description: Optional human-readable description. + :param include_memory: Whether to include the current RAM state + (only possible while the VM is running). + :raises ValueError: If no VM with the given name/ID exists, or a + snapshot with that name already exists. + :raises RuntimeError: If snapshot creation fails. + + Example:: + + driver.snapshot_create("web01", "before-upgrade", + description="Clean state before kernel upgrade") + """ + raise NotImplementedError + + def snapshot_delete(self, name: str, snapshot: str) -> None: + """ + Deletes a snapshot of a virtual machine. + + :param name: VM name or numeric VMID as a string. + :param snapshot: Name of the snapshot to delete. + :raises ValueError: If the VM or snapshot does not exist. + :raises RuntimeError: If other snapshots depend on this one (must delete children first). + + Example:: + + driver.snapshot_delete("web01", "before-upgrade") + """ + raise NotImplementedError + + def snapshot_rollback(self, name: str, snapshot: str) -> None: + """ + Reverts a virtual machine to a previously created snapshot. + + The VM is stopped (if running), reverted, and then left in the state + the snapshot recorded (running or stopped depending on ``has_memory``). + + :param name: VM name or numeric VMID as a string. + :param snapshot: Name of the snapshot to roll back to. + :raises ValueError: If the VM or snapshot does not exist. + :raises RuntimeError: If the rollback fails. + + Example:: + + driver.snapshot_rollback("web01", "before-upgrade") + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Storage + # ------------------------------------------------------------------ + + def get_storage(self) -> Dict[str, StorageVolumeDict]: + """ + Returns the storage pools / datastores configured on this hypervisor. + + Keys are storage pool names. Each value contains: + + * name (string) - pool name (repeated for convenience) + * type (string) - backend type: ``"dir"``, ``"lvm"``, ``"zfs"``, + ``"nfs"``, ``"ceph"``, ``"iscsi"`` etc. + * total (int) - total capacity in bytes + * used (int) - used space in bytes + * available (int) - free space in bytes + * enabled (bool) - whether the pool is administratively enabled + * shared (bool) - whether the pool is accessible from multiple cluster nodes + + Example:: + + { + "local-lvm": { + "name": "local-lvm", + "type": "lvm", + "total": 107374182400, + "used": 53687091200, + "available": 53687091200, + "enabled": True, + "shared": False, + }, + "ceph-pool": { + "name": "ceph-pool", + "type": "ceph", + "total": 1099511627776, + "used": 274877906944, + "available": 824633720832, + "enabled": True, + "shared": True, + }, + } + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Virtual networking + # ------------------------------------------------------------------ + + def get_virtual_networks(self) -> Dict[str, VirtualNetworkDict]: + """ + Returns the virtual networks / bridges defined on this hypervisor. + + Keys are network names. Each value contains: + + * name (string) - network name (repeated for convenience) + * type (string) - network type: ``"bridge"``, ``"ovs"``, ``"nat"``, ``"vxlan"`` + * bridge (string) - underlying host bridge interface + * vlan_id (int) - associated VLAN tag (0 = untagged / all VLANs) + * autostart (bool) - whether the network starts automatically at boot + * active (bool) - whether the network is currently active + + Example:: + + { + "vmbr0": { + "name": "vmbr0", + "type": "bridge", + "bridge": "vmbr0", + "vlan_id": 0, + "autostart": True, + "active": True, + }, + "vmbr10": { + "name": "vmbr10", + "type": "bridge", + "bridge": "vmbr10", + "vlan_id": 10, + "autostart": True, + "active": True, + }, + } + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Package management (hypervisor extensions / plugins) + # ------------------------------------------------------------------ + + def get_packages(self) -> List[PackageDict]: + """ + Returns all packages known to the hypervisor's package manager + (e.g. ``apt`` on Proxmox VE, vendor extension bundles on ESXi). + + Each entry contains: + + * name (string) - package name + * version (string) - installed or available version string + * installed (bool) - ``True`` if the package is currently installed + * description (string) - short package description + * size (int) - package size in bytes (0 if unknown) + * source (string) - repository the package comes from + + Example:: + + [ + { + "name": "proxmox-backup-client", + "version": "3.2.4-1", + "installed": True, + "description": "Proxmox Backup Client tools", + "size": 8388608, + "source": "pve-no-subscription", + }, + { + "name": "ifupdown2", + "version": "3.2.0-1+pmx4", + "installed": True, + "description": "Network interface management daemon", + "size": 524288, + "source": "pve-no-subscription", + }, + ] + """ + raise NotImplementedError + + def install_package(self, name: str, version: str = "") -> None: + """ + Installs a package on the hypervisor host. + + :param name: Package name as known to the package manager. + :param version: Exact version to install. Empty string installs latest. + :raises NotImplementedError: If the driver does not support package management. + :raises ValueError: If the package name is unknown or the version unavailable. + :raises RuntimeError: If the installation fails on the host side. + + Example:: + + driver.install_package("proxmox-backup-client") + """ + raise NotImplementedError + + def remove_package(self, name: str) -> None: + """ + Removes an installed package from the hypervisor host. + + :param name: Package name to remove. + :raises NotImplementedError: If the driver does not support package management. + :raises ValueError: If the package is not currently installed. + :raises RuntimeError: If removal fails (e.g. required dependency). + + Example:: + + driver.remove_package("proxmox-backup-client") + """ + raise NotImplementedError + + def get_package_config(self, name: str) -> Dict[str, Any]: + """ + Returns the current configuration of an installed hypervisor package + as a dictionary. The structure is package-specific. + + :param name: Package name. + :raises NotImplementedError: If the driver does not support package management. + :raises ValueError: If the package is not installed. + + Example:: + + driver.get_package_config("proxmox-backup-client") + # → + { + "server": "backup.corp.example", + "datastore": "vm-backups", + "fingerprint": "AB:CD:EF:...", + "schedule": "daily", + "retention": {"keep_last": 7, "keep_weekly": 4}, + } + """ + raise NotImplementedError + + def set_package_config(self, name: str, config: Dict[str, Any]) -> None: + """ + Writes a new configuration for an installed hypervisor package. + + :param name: Package name. + :param config: New configuration as a nested dictionary. + :raises NotImplementedError: If the driver does not support package management. + :raises ValueError: If the package is not installed or config is invalid. + :raises RuntimeError: If the device rejects the configuration. + + Example:: + + driver.set_package_config( + "proxmox-backup-client", + { + "server": "backup.corp.example", + "datastore": "vm-backups", + "schedule": "daily", + "retention": {"keep_last": 14, "keep_weekly": 4}, + }, + ) + """ + raise NotImplementedError diff --git a/napalm_device_types/models.py b/napalm_device_types/models.py new file mode 100644 index 0000000..8188854 --- /dev/null +++ b/napalm_device_types/models.py @@ -0,0 +1,430 @@ +""" +TypedDicts for NAPALM device-type-specific return values. + +These supplement the types defined in napalm.base.models and are used by the +abstract device-type driver classes in this package. +""" + +from typing import Dict, List, Optional +from typing_extensions import TypedDict + + +# --------------------------------------------------------------------------- +# Common (shared across device types) +# --------------------------------------------------------------------------- + + +class RadiusServerDict(TypedDict): + host: str + port: int + timeout: int + retries: int + + +class MACACLEntryDict(TypedDict): + mac: str + action: str + description: str + + +class MACACLDict(TypedDict): + """MAC access-control list for a single binding point. + + ``name`` is the SSID name on an access point, or the interface name on a + switch. + """ + + name: str + policy: str + entries: List[MACACLEntryDict] + + +class PackageDict(TypedDict): + """A software package / plugin installed on the device.""" + + name: str + version: str + installed: bool + description: str + size: int + source: str + + +# --------------------------------------------------------------------------- +# Access Point +# --------------------------------------------------------------------------- + + +class WirelessClientDict(TypedDict): + mac: str + ssid: str + radio: str + signal: int + noise: int + tx_rate: float + rx_rate: float + uptime: int + + +class SSIDDict(TypedDict): + enabled: bool + radio: str + bssid: str + encryption: str + hidden: bool + clients: int + + +class RadioStatusDict(TypedDict): + enabled: bool + band: str + channel: int + channel_width: int + tx_power: int + frequency: float + + +class WirelessConfigDict(TypedDict): + country_code: str + regulatory_domain: str + beacon_interval: int + dtim_period: int + rts_threshold: int + fragmentation_threshold: int + short_preamble: bool + wmm_enabled: bool + + +class FastTransitionConfigDict(TypedDict): + enabled: bool + ssid: str + mobility_domain: str + reassociation_deadline: int + r0_key_lifetime: int + r1_key_holder: str + pmk_r1_push: bool + over_ds: bool + + +class MeshPeerDict(TypedDict): + mac: str + radio: str + signal: int + tx_rate: float + rx_rate: float + uptime: int + hop_count: int + + +class MeshConfigDict(TypedDict): + enabled: bool + radio: str + mesh_id: str + path_metric: str + gate_announcements: bool + is_gate: bool + encryption: str + + +class SSIDBridgeDict(TypedDict): + ssid: str + bridge: str + vlan_id: int + tagged: bool + client_isolation: bool + + +class Dot1XConfigDict(TypedDict): + """802.1X / WPA-Enterprise config for an access-point SSID.""" + + enabled: bool + ssid: str + auth_server: RadiusServerDict + acct_server: Optional[RadiusServerDict] + reauth_interval: int + pmksa_caching: bool + + +# --------------------------------------------------------------------------- +# Switch +# --------------------------------------------------------------------------- + + +class STPInterfaceDict(TypedDict): + role: str + state: str + cost: int + port_priority: int + + +class SpanningTreeDict(TypedDict): + mode: str + root_bridge: bool + root_id: str + root_priority: int + bridge_id: str + bridge_priority: int + interfaces: Dict[str, STPInterfaceDict] + + +class PortChannelDict(TypedDict): + members: List[str] + protocol: str + min_links: int + is_up: bool + + +class Dot1XPortDict(TypedDict): + """802.1X / NAC configuration for a single switch port.""" + + enabled: bool + port_control: str + host_mode: str + auth_server: RadiusServerDict + acct_server: Optional[RadiusServerDict] + reauthentication: bool + reauth_interval: int + guest_vlan: int + auth_fail_vlan: int + + +class PoEPortDict(TypedDict): + enabled: bool + status: str + poe_class: str + power_draw: float + power_budget: float + voltage: float + current: float + + +class PoESummaryDict(TypedDict): + total_power_budget: float + total_power_draw: float + ports: Dict[str, PoEPortDict] + + +# --------------------------------------------------------------------------- +# Firewall +# --------------------------------------------------------------------------- + + +class NATTranslationDict(TypedDict): + protocol: str + inside_local: str + inside_global: str + outside_local: str + outside_global: str + age: float + + +class SecurityZoneDict(TypedDict): + interfaces: List[str] + policy: str + description: str + + +class SessionDict(TypedDict): + protocol: str + src_ip: str + src_port: int + dst_ip: str + dst_port: int + state: str + age: float + + +class VPNTunnelDict(TypedDict): + type: str + local_endpoint: str + remote_endpoint: str + is_up: bool + uptime: int + bytes_in: int + bytes_out: int + + +# --------------------------------------------------------------------------- +# Hypervisor +# --------------------------------------------------------------------------- + + +class VMDiskDict(TypedDict): + device: str + storage: str + size: int + format: str + bootable: bool + + +class VMNICDict(TypedDict): + device: str + mac: str + model: str + bridge: str + vlan_id: int + + +class VMDict(TypedDict): + name: str + vmid: int + status: str + vcpus: int + memory: int + cpu_usage: float + memory_usage: int + uptime: int + node: str + + +class VMConfigDict(TypedDict): + name: str + vmid: int + vcpus: int + memory: int + os_type: str + boot_order: List[str] + disks: List[VMDiskDict] + nics: List[VMNICDict] + description: str + tags: List[str] + + +class StorageVolumeDict(TypedDict): + name: str + type: str + total: int + used: int + available: int + enabled: bool + shared: bool + + +class VirtualNetworkDict(TypedDict): + name: str + type: str + bridge: str + vlan_id: int + autostart: bool + active: bool + + +class SnapshotDict(TypedDict): + name: str + vm: str + created: float + description: str + has_memory: bool + parent: str + + +# --------------------------------------------------------------------------- +# Storage / NAS +# --------------------------------------------------------------------------- + + +class PhysicalDiskDict(TypedDict): + """A physical drive installed in the storage device.""" + + slot: str + model: str + serial: str + vendor: str + type: str # "hdd", "ssd", "nvme" + size: int # bytes + rpm: int # 0 for SSD/NVMe + temperature: int # Celsius; -1 if unavailable + health: str # "healthy", "warning", "failed", "unknown" + pool: str # name of the containing pool; empty string if spare/unassigned + + +class DiskPoolDict(TypedDict): + """A RAID array, ZFS pool, or volume group.""" + + name: str + type: str # "zfs", "lvm", "md", "hardware-raid", "btrfs" + level: str # "stripe", "mirror", "raidz1", "raidz2", "raidz3", + # "raid0" … "raid60", "single", etc. + status: str # "online", "degraded", "faulted", "offline", "unknown" + total: int # bytes + used: int # bytes + available: int # bytes + disks: List[str] # slot identifiers of member disks + auto_expand: bool + dedup: bool + compression: str # "off", "lz4", "gzip", "zstd", etc. + + +class LogicalVolumeDict(TypedDict): + """A logical volume, ZFS dataset, or LUN exposed to clients.""" + + name: str + pool: str + type: str # "filesystem", "volume" (block device / LUN), "zvol" + total: int # bytes + used: int # bytes + available: int # bytes + mountpoint: str # empty string for block volumes + compression: str # "off", "lz4", etc. + dedup: bool + readonly: bool + snapshots: int # number of snapshots currently held + + +class NASShareDict(TypedDict): + """A network share or iSCSI target exported by the storage device.""" + + name: str + protocol: str # "nfs", "smb", "afp", "ftp", "sftp", "iscsi", "webdav" + path: str # local filesystem path or iSCSI target IQN + volume: str # logical volume or dataset this share is backed by + enabled: bool + readonly: bool + description: str + clients: List[str] # IP/subnet allow-list; empty list = all hosts allowed + + +class VolumeSnapshotDict(TypedDict): + """A point-in-time snapshot of a logical volume or dataset.""" + + name: str + volume: str # parent volume / dataset + created: float # Unix epoch + size: int # bytes of unique data referenced by this snapshot + description: str + clones: List[str] # volumes cloned from this snapshot + + +class StorageQuotaDict(TypedDict): + """A filesystem quota applied to a user, group, or dataset.""" + + target: str # username, group name, or dataset path + target_type: str # "user", "group", "dataset" + volume: str # volume / dataset the quota applies to + used: int # bytes currently used + quota: int # hard limit in bytes; 0 = no limit + ref_quota: int # referenced-data limit in bytes; 0 = no limit + + +class StorageServiceDict(TypedDict): + """Status of a file-sharing or access service running on the device.""" + + name: str # "nfs", "smb", "ftp", "ssh", "iscsi", "webdav", etc. + enabled: bool # administratively enabled + running: bool # currently active / listening + port: int # primary listening port; 0 if N/A + version: str # protocol version string (e.g. "4.1" for NFSv4.1) + + +class ReplicationJobDict(TypedDict): + """A scheduled replication or sync task.""" + + name: str + source: str # source path / dataset + target: str # destination path / dataset (may be remote: host:path) + direction: str # "push" or "pull" + schedule: str # cron expression or human label ("daily", "hourly") + enabled: bool + last_run: float # Unix epoch; 0.0 if never run + last_status: str # "success", "failed", "running", "pending" + bytes_sent: int # bytes transferred in the last run diff --git a/napalm_device_types/storage.py b/napalm_device_types/storage.py new file mode 100644 index 0000000..84f9eb8 --- /dev/null +++ b/napalm_device_types/storage.py @@ -0,0 +1,599 @@ +""" +Abstract base class for storage and NAS/SAN device drivers. + +Usage:: + + from napalm_device_types import StorageDriver + + class TrueNASDriver(StorageDriver): + def get_disks(self): + ... +""" + +from typing import Any, Dict, List +from napalm.base import NetworkDriver +from napalm_device_types.models import ( + DiskPoolDict, + LogicalVolumeDict, + NASShareDict, + PackageDict, + PhysicalDiskDict, + ReplicationJobDict, + StorageQuotaDict, + StorageServiceDict, + VolumeSnapshotDict, +) + + +class StorageDriver(NetworkDriver): + """ + Abstract intermediate driver for storage appliances and NAS/SAN devices + (e.g. TrueNAS SCALE/CORE, Synology DSM, QNAP QTS, NetApp ONTAP, + OpenMediaVault, Pure Storage, IBM Storwize). + + Inherits all standard NAPALM NetworkDriver methods and adds + storage-specific operations that concrete drivers must implement. + """ + + # ------------------------------------------------------------------ + # Physical hardware + # ------------------------------------------------------------------ + + def get_disks(self) -> List[PhysicalDiskDict]: + """ + Returns all physical drives detected by the storage device. + + Each entry contains: + + * slot (string) - bay or device identifier (e.g. ``"bay1"``, ``"sda"``, ``"nvme0n1"``) + * model (string) - drive model string + * serial (string) - drive serial number + * vendor (string) - drive manufacturer + * type (string) - ``"hdd"``, ``"ssd"``, or ``"nvme"`` + * size (int) - raw capacity in bytes + * rpm (int) - rotational speed; ``0`` for SSD/NVMe + * temperature (int) - current temperature in Celsius; ``-1`` if unavailable + * health (string) - ``"healthy"``, ``"warning"``, ``"failed"``, or ``"unknown"`` + * pool (string) - name of the pool this disk belongs to; empty string if unassigned/spare + + Example:: + + [ + { + "slot": "bay1", + "model": "HGST HUS726T6TALE6L4", + "serial": "K3GXXXXX", + "vendor": "HGST", + "type": "hdd", + "size": 6001175126016, + "rpm": 7200, + "temperature": 34, + "health": "healthy", + "pool": "tank", + }, + { + "slot": "bay5", + "model": "Samsung SSD 870 EVO 1TB", + "serial": "S5XXXXXXX", + "vendor": "Samsung", + "type": "ssd", + "size": 1000204886016, + "rpm": 0, + "temperature": 28, + "health": "healthy", + "pool": "fast-pool", + }, + ] + """ + raise NotImplementedError + + def get_disk_pools(self) -> Dict[str, DiskPoolDict]: + """ + Returns the disk pools (RAID arrays, ZFS pools, LVM volume groups, etc.) + configured on the device. + + Keys are pool names. Each value contains: + + * name (string) - pool name (repeated for convenience) + * type (string) - ``"zfs"``, ``"lvm"``, ``"md"``, ``"hardware-raid"``, ``"btrfs"`` + * level (string) - RAID/redundancy level: ``"mirror"``, ``"raidz1"``, ``"raidz2"``, + ``"raidz3"``, ``"stripe"``, ``"raid0"`` … ``"raid60"``, ``"single"`` + * status (string) - ``"online"``, ``"degraded"``, ``"faulted"``, ``"offline"``, ``"unknown"`` + * total (int) - usable capacity in bytes + * used (int) - used space in bytes + * available (int) - free space in bytes + * disks (list of strings) - slot identifiers of member drives + * auto_expand (bool) - whether the pool grows automatically when disks are replaced with larger ones + * dedup (bool) - whether deduplication is enabled + * compression (string) - pool-level compression algorithm: ``"off"``, ``"lz4"``, + ``"gzip"``, ``"zstd"``, etc. + + Example:: + + { + "tank": { + "name": "tank", + "type": "zfs", + "level": "raidz2", + "status": "online", + "total": 21990232555520, + "used": 8796093022208, + "available": 13194139533312, + "disks": ["bay1", "bay2", "bay3", "bay4", "bay5", "bay6"], + "auto_expand": True, + "dedup": False, + "compression": "lz4", + }, + } + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Logical volumes / datasets + # ------------------------------------------------------------------ + + def get_volumes(self) -> Dict[str, LogicalVolumeDict]: + """ + Returns all logical volumes, ZFS datasets, or LUNs on the device. + + Keys are volume paths (e.g. ``"tank/data"``, ``"tank/media"``). + Each value contains: + + * name (string) - volume name (leaf component or full path) + * pool (string) - parent pool + * type (string) - ``"filesystem"``, ``"volume"`` (block device / LUN), or ``"zvol"`` + * total (int) - quota or provisioned size in bytes; ``0`` means unlimited + * used (int) - space currently used in bytes + * available (int) - space available in bytes + * mountpoint (string) - local mount path; empty string for block volumes / LUNs + * compression (string) - active compression algorithm (``"off"``, ``"lz4"``, ``"zstd"``, etc.) + * dedup (bool) - whether deduplication is active on this volume + * readonly (bool) - whether the volume is mounted read-only + * snapshots (int) - number of snapshots currently held + + Example:: + + { + "tank/media": { + "name": "media", + "pool": "tank", + "type": "filesystem", + "total": 0, + "used": 4398046511104, + "available": 13194139533312, + "mountpoint": "/mnt/tank/media", + "compression": "lz4", + "dedup": False, + "readonly": False, + "snapshots": 7, + }, + "tank/backups": { + "name": "backups", + "pool": "tank", + "type": "filesystem", + "total": 5497558138880, + "used": 1099511627776, + "available": 4398046511104, + "mountpoint": "/mnt/tank/backups", + "compression": "zstd", + "dedup": False, + "readonly": False, + "snapshots": 14, + }, + } + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Shares + # ------------------------------------------------------------------ + + def get_shares(self) -> Dict[str, NASShareDict]: + """ + Returns all network shares and iSCSI targets exported by the device. + + Keys are share names. Each value contains: + + * name (string) - share name (repeated for convenience) + * protocol (string) - ``"nfs"``, ``"smb"``, ``"afp"``, ``"ftp"``, ``"sftp"``, + ``"iscsi"``, or ``"webdav"`` + * path (string) - local filesystem path; for iSCSI the target IQN + * volume (string) - logical volume or dataset this share is backed by + * enabled (bool) - whether the share is currently exported + * readonly (bool) - whether the share is exported read-only + * description (string) - optional human-readable description + * clients (list of strings) - IP address or subnet allow-list; + empty list means all hosts are permitted + + Example:: + + { + "media": { + "name": "media", + "protocol": "nfs", + "path": "/mnt/tank/media", + "volume": "tank/media", + "enabled": True, + "readonly": False, + "description": "Media library", + "clients": ["192.168.1.0/24"], + }, + "homes": { + "name": "homes", + "protocol": "smb", + "path": "/mnt/tank/homes", + "volume": "tank/homes", + "enabled": True, + "readonly": False, + "description": "User home directories", + "clients": [], + }, + } + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Snapshots + # ------------------------------------------------------------------ + + def get_volume_snapshots(self, volume: str = "") -> List[VolumeSnapshotDict]: + """ + Returns volume / dataset snapshots. + + :param volume: Restrict results to this volume path (e.g. ``"tank/media"``). + Pass an empty string (default) to list snapshots for all volumes. + + Each entry contains: + + * name (string) - snapshot name (e.g. ``"auto-2026-05-11"`` or full ``"tank/media@auto-2026-05-11"``) + * volume (string) - parent volume / dataset path + * created (float) - creation timestamp (Unix epoch) + * size (int) - bytes of unique data referenced only by this snapshot + * description (string) - optional description + * clones (list of strings) - volumes that were cloned from this snapshot + + Example:: + + [ + { + "name": "auto-2026-05-11", + "volume": "tank/media", + "created": 1746921600.0, + "size": 2097152, + "description": "Automatic daily snapshot", + "clones": [], + }, + { + "name": "before-migration", + "volume": "tank/backups", + "created": 1746835200.0, + "size": 1073741824, + "description": "Snapshot before storage migration", + "clones": ["tank/backups-clone"], + }, + ] + """ + raise NotImplementedError + + def snapshot_create(self, volume: str, name: str, description: str = "") -> None: + """ + Creates a snapshot of a logical volume or dataset. + + :param volume: Volume / dataset path (e.g. ``"tank/media"``). + :param name: Name for the new snapshot. + :param description: Optional human-readable description. + :raises ValueError: If the volume does not exist, or a snapshot with that + name already exists. + :raises RuntimeError: If snapshot creation fails on the device side. + + Example:: + + driver.snapshot_create("tank/media", "before-migration", + description="Snapshot before storage migration") + """ + raise NotImplementedError + + def snapshot_delete(self, volume: str, name: str) -> None: + """ + Deletes a snapshot of a logical volume or dataset. + + :param volume: Volume / dataset path. + :param name: Name of the snapshot to delete. + :raises ValueError: If the volume or snapshot does not exist. + :raises RuntimeError: If other clones depend on this snapshot + (delete or promote clones first). + + Example:: + + driver.snapshot_delete("tank/media", "auto-2026-05-01") + """ + raise NotImplementedError + + def snapshot_rollback(self, volume: str, name: str) -> None: + """ + Reverts a volume / dataset to a previously created snapshot. + + All data written after the snapshot was taken is permanently discarded. + Any snapshots created after the target snapshot are also deleted. + + :param volume: Volume / dataset path. + :param name: Name of the snapshot to roll back to. + :raises ValueError: If the volume or snapshot does not exist. + :raises RuntimeError: If the rollback fails (e.g. active clones block it). + + Example:: + + driver.snapshot_rollback("tank/media", "before-migration") + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Quotas + # ------------------------------------------------------------------ + + def get_quotas(self) -> List[StorageQuotaDict]: + """ + Returns all filesystem quotas configured on the device. + + Each entry contains: + + * target (string) - username, group name, or dataset path depending on ``target_type`` + * target_type (string) - ``"user"``, ``"group"``, or ``"dataset"`` + * volume (string) - volume / dataset the quota applies to + * used (int) - bytes currently consumed by this target + * quota (int) - hard storage limit in bytes; ``0`` means no limit + * ref_quota (int) - referenced-data limit in bytes (excludes snapshots); + ``0`` means no limit + + Example:: + + [ + { + "target": "alice", + "target_type": "user", + "volume": "tank/homes", + "used": 53687091200, + "quota": 107374182400, + "ref_quota": 0, + }, + { + "target": "tank/backups", + "target_type": "dataset", + "volume": "tank/backups", + "used": 1099511627776, + "quota": 5497558138880, + "ref_quota": 0, + }, + ] + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Services + # ------------------------------------------------------------------ + + def get_services(self) -> Dict[str, StorageServiceDict]: + """ + Returns the file-sharing and access services available on the device. + + Keys are service names (e.g. ``"nfs"``, ``"smb"``). Each value contains: + + * name (string) - service name (repeated for convenience) + * enabled (bool) - administratively enabled (will start on next boot) + * running (bool) - currently active and listening + * port (int) - primary listening port; ``0`` if not applicable + * version (string) - protocol version string (e.g. ``"4.1"`` for NFSv4.1, + ``"3.1.1"`` for SMB3); empty string if unknown + + Example:: + + { + "nfs": { + "name": "nfs", + "enabled": True, + "running": True, + "port": 2049, + "version": "4.2", + }, + "smb": { + "name": "smb", + "enabled": True, + "running": True, + "port": 445, + "version": "3.1.1", + }, + "ftp": { + "name": "ftp", + "enabled": False, + "running": False, + "port": 21, + "version": "", + }, + } + """ + raise NotImplementedError + + def set_service_enabled(self, service: str, enabled: bool) -> None: + """ + Administratively enables or disables a file-sharing service. + + Disabling stops the service immediately; enabling starts it immediately. + + :param service: Service name (e.g. ``"nfs"``, ``"smb"``, ``"ftp"``). + :param enabled: ``True`` to start and enable; ``False`` to stop and disable. + :raises ValueError: If the service name is not recognised. + :raises RuntimeError: If the operation fails on the device side. + + Example:: + + driver.set_service_enabled("ftp", False) # disable FTP + driver.set_service_enabled("nfs", True) # enable NFS + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Replication + # ------------------------------------------------------------------ + + def get_replication_jobs(self) -> List[ReplicationJobDict]: + """ + Returns all replication / sync jobs configured on the device. + + Each entry contains: + + * name (string) - job name + * source (string) - source path or dataset + * target (string) - destination path or dataset; may include a remote + host prefix (e.g. ``"backup-server:tank/replica"``) + * direction (string) - ``"push"`` (local → remote) or ``"pull"`` (remote → local) + * schedule (string) - cron expression or descriptive label (``"daily"``, + ``"hourly"``, etc.) + * enabled (bool) - whether the job is scheduled to run + * last_run (float) - Unix epoch of the last run; ``0.0`` if never run + * last_status (string) - ``"success"``, ``"failed"``, ``"running"``, or ``"pending"`` + * bytes_sent (int) - bytes transferred in the most recent run; ``0`` if never run + + Example:: + + [ + { + "name": "media-offsite", + "source": "tank/media", + "target": "backup-nas:tank/media-replica", + "direction": "push", + "schedule": "0 2 * * *", + "enabled": True, + "last_run": 1746921600.0, + "last_status": "success", + "bytes_sent": 1073741824, + }, + { + "name": "backups-local", + "source": "tank/backups", + "target": "tank/backups-mirror", + "direction": "push", + "schedule": "hourly", + "enabled": True, + "last_run": 1746918000.0, + "last_status": "success", + "bytes_sent": 104857600, + }, + ] + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Package management (plugins / extensions) + # ------------------------------------------------------------------ + + def get_packages(self) -> List[PackageDict]: + """ + Returns all packages / plugins installed on the storage appliance + (e.g. TrueNAS SCALE Apps, Synology packages, OpenMediaVault plugins). + + Each entry contains: + + * name (string) - package name + * version (string) - installed or available version string + * installed (bool) - ``True`` if the package is currently installed + * description (string) - short package description + * size (int) - package size in bytes; ``0`` if unknown + * source (string) - repository or catalogue the package comes from + + Example:: + + [ + { + "name": "plex-media-server", + "version": "1.40.0", + "installed": True, + "description": "Plex Media Server", + "size": 134217728, + "source": "TrueNAS Community", + }, + { + "name": "nextcloud", + "version": "28.0.3", + "installed": True, + "description": "Nextcloud – self-hosted file sync and share", + "size": 268435456, + "source": "TrueNAS Community", + }, + ] + """ + raise NotImplementedError + + def install_package(self, name: str, version: str = "") -> None: + """ + Installs a package or plugin on the storage appliance. + + :param name: Package name as known to the package catalogue. + :param version: Exact version to install. Empty string installs latest. + :raises NotImplementedError: If the driver does not support package management. + :raises ValueError: If the package name is unknown or the version unavailable. + :raises RuntimeError: If the installation fails on the device side. + + Example:: + + driver.install_package("nextcloud") + """ + raise NotImplementedError + + def remove_package(self, name: str) -> None: + """ + Removes an installed package from the storage appliance. + + :param name: Package name to remove. + :raises NotImplementedError: If the driver does not support package management. + :raises ValueError: If the package is not currently installed. + :raises RuntimeError: If removal fails (e.g. required dependency). + + Example:: + + driver.remove_package("plex-media-server") + """ + raise NotImplementedError + + def get_package_config(self, name: str) -> Dict[str, Any]: + """ + Returns the current configuration of an installed package as a dictionary. + The structure is package-specific. + + :param name: Package name. + :raises NotImplementedError: If the driver does not support package management. + :raises ValueError: If the package is not installed. + + Example:: + + driver.get_package_config("nextcloud") + # → + { + "admin_user": "admin", + "trusted_domains": ["nas.corp.example"], + "mail_smtphost": "smtp.corp.example", + "maintenance_window_start": 2, + } + """ + raise NotImplementedError + + def set_package_config(self, name: str, config: Dict[str, Any]) -> None: + """ + Writes a new configuration for an installed package. + + :param name: Package name. + :param config: New configuration as a nested dictionary. + :raises NotImplementedError: If the driver does not support package management. + :raises ValueError: If the package is not installed or config is invalid. + :raises RuntimeError: If the device rejects the configuration. + + Example:: + + driver.set_package_config( + "nextcloud", + { + "trusted_domains": ["nas.corp.example", "192.168.1.10"], + "maintenance_window_start": 3, + }, + ) + """ + raise NotImplementedError diff --git a/napalm_device_types/switch.py b/napalm_device_types/switch.py new file mode 100644 index 0000000..f2cd49e --- /dev/null +++ b/napalm_device_types/switch.py @@ -0,0 +1,298 @@ +""" +Abstract base class for switch drivers. + +Usage:: + + from napalm_device_types import SwitchDriver + + class CiscoSGDriver(SwitchDriver): + def get_spanning_tree(self): + ... +""" + +from typing import Dict +from napalm.base import NetworkDriver +from napalm_device_types.models import ( + Dot1XPortDict, + MACACLDict, + PoESummaryDict, + PortChannelDict, + SpanningTreeDict, +) + + +class SwitchDriver(NetworkDriver): + """ + Abstract intermediate driver for Ethernet switches. + + Inherits all standard NAPALM NetworkDriver methods (including + ``get_vlans()``, ``get_mac_address_table()``) and adds switch-specific + operations that concrete drivers must implement. + """ + + def get_spanning_tree(self) -> Dict[str, SpanningTreeDict]: + """ + Returns spanning tree status for each STP instance. + + Keys are STP instance identifiers (e.g. VLAN IDs for PVST, + ``"MST0"`` for MSTP, or ``"0"`` for a single instance). + Each value contains: + + * mode (string) - STP variant: ``"STP"``, ``"RSTP"``, ``"MSTP"``, ``"PVST"`` + * root_bridge (bool) - whether this device is the root bridge + * root_id (string) - root bridge MAC address + * root_priority (int) - root bridge priority + * bridge_id (string) - this bridge's MAC address + * bridge_priority (int) - this bridge's priority + * interfaces (dict) - per-interface STP state: + + * role (string) - ``"root"``, ``"designated"``, ``"alternate"``, ``"backup"`` + * state (string) - ``"forwarding"``, ``"blocking"``, ``"learning"``, ``"listening"`` + * cost (int) - port path cost + * port_priority (int) - port priority + + Example:: + + { + "1": { + "mode": "RSTP", + "root_bridge": False, + "root_id": "00:11:22:33:44:55", + "root_priority": 4096, + "bridge_id": "AA:BB:CC:DD:EE:FF", + "bridge_priority": 32768, + "interfaces": { + "GigabitEthernet0/1": { + "role": "root", + "state": "forwarding", + "cost": 4, + "port_priority": 128, + }, + "GigabitEthernet0/2": { + "role": "designated", + "state": "forwarding", + "cost": 4, + "port_priority": 128, + }, + }, + } + } + """ + raise NotImplementedError + + def get_port_channels(self) -> Dict[str, PortChannelDict]: + """ + Returns port-channel (LAG) configuration and status. + + Keys are port-channel interface names (e.g. ``"Port-Channel1"``). + Each value contains: + + * members (list of strings) - names of member interfaces + * protocol (string) - aggregation protocol: ``"LACP"``, ``"PAgP"``, ``"static"`` + * min_links (int) - minimum number of active members required + * is_up (bool) - whether the LAG is operationally up + + Example:: + + { + "Port-Channel1": { + "members": ["GigabitEthernet0/1", "GigabitEthernet0/2"], + "protocol": "LACP", + "min_links": 1, + "is_up": True, + } + } + """ + raise NotImplementedError + + def get_mac_acl(self) -> Dict[str, MACACLDict]: + """ + Returns the MAC-address-based access control lists configured per port. + + Keys are interface names. Each value contains: + + * name (string) - interface name (repeated for convenience) + * policy (string) - ACL mode: + + * ``"allow"`` – whitelist: only listed MACs may use this port + * ``"deny"`` – blacklist: listed MACs are blocked + * ``"disabled"`` – no MAC filtering active + + * entries (list) - ACL entries, each with: + + * mac (string) - MAC address (normalised, colon-separated) + * action (string) - ``"allow"`` or ``"deny"`` + * description (string) - optional human-readable label + + Example:: + + { + "GigabitEthernet0/1": { + "name": "GigabitEthernet0/1", + "policy": "allow", + "entries": [ + {"mac": "AA:BB:CC:DD:EE:01", "action": "allow", "description": "printer"}, + ], + }, + "GigabitEthernet0/2": { + "name": "GigabitEthernet0/2", + "policy": "disabled", + "entries": [], + }, + } + """ + raise NotImplementedError + + def get_dot1x_config(self) -> Dict[str, Dot1XPortDict]: + """ + Returns the 802.1X / NAC configuration per switch port. + + Keys are interface names. Each value contains: + + * enabled (bool) - whether 802.1X is active on this port + * port_control (string) - authentication mode: + + * ``"auto"`` – port authenticates normally + * ``"force-authorized"`` – port always passes traffic (bypass) + * ``"force-unauthorized"`` – port always blocks traffic + + * host_mode (string) - how many identities are authenticated per port: + + * ``"single-host"`` – one device, then port is locked + * ``"multi-host"`` – first auth unlocks port for all devices + * ``"multi-domain"`` – one data + one voice device (IP phone scenario) + * ``"multi-auth"`` – each device authenticates individually + + * auth_server (dict) - RADIUS authentication server (host, port, timeout, retries) + * acct_server (dict or None) - RADIUS accounting server, ``None`` if unused + * reauthentication (bool) - whether periodic re-authentication is enabled + * reauth_interval (int) - re-authentication interval in seconds (0 = disabled) + * guest_vlan (int) - VLAN ID for unauthenticated clients (0 = disabled) + * auth_fail_vlan (int) - VLAN ID for clients that fail authentication (0 = disabled) + + Note: RADIUS shared secrets are intentionally omitted. + + Example:: + + { + "GigabitEthernet0/1": { + "enabled": True, + "port_control": "auto", + "host_mode": "multi-domain", + "auth_server": { + "host": "radius.corp.example", + "port": 1812, + "timeout": 5, + "retries": 3, + }, + "acct_server": None, + "reauthentication": True, + "reauth_interval": 3600, + "guest_vlan": 99, + "auth_fail_vlan": 999, + }, + } + """ + raise NotImplementedError + + def get_poe_status(self) -> PoESummaryDict: + """ + Returns the PoE status of the switch as a whole and per port. + + The returned dictionary contains: + + * total_power_budget (float) - total PoE power available in watts + * total_power_draw (float) - total PoE power currently consumed in watts + * ports (dict) - per-interface PoE state, keyed by interface name: + + * enabled (bool) - whether PoE is configured on this port + * status (string) - operational state: + + * ``"delivering"`` – power is being delivered to a PD + * ``"searching"`` – port is looking for a powered device + * ``"fault"`` – an error condition was detected + * ``"disabled"`` – PoE is administratively off + * ``"denied"`` – PD detected but power budget exceeded + + * poe_class (string) - IEEE 802.3 class: ``"Class 0"`` … ``"Class 8"`` + (``"unknown"`` if not yet negotiated) + * power_draw (float) - current power consumption in watts + * power_budget (float) - per-port power limit in watts + * voltage (float) - measured port voltage in volts + * current (float) - measured port current in milliamps + + Example:: + + { + "total_power_budget": 740.0, + "total_power_draw": 43.2, + "ports": { + "GigabitEthernet0/1": { + "enabled": True, + "status": "delivering", + "poe_class": "Class 3", + "power_draw": 12.4, + "power_budget": 30.0, + "voltage": 53.5, + "current": 231.0, + }, + "GigabitEthernet0/2": { + "enabled": True, + "status": "searching", + "poe_class": "unknown", + "power_draw": 0.0, + "power_budget": 30.0, + "voltage": 0.0, + "current": 0.0, + }, + "GigabitEthernet0/3": { + "enabled": False, + "status": "disabled", + "poe_class": "unknown", + "power_draw": 0.0, + "power_budget": 0.0, + "voltage": 0.0, + "current": 0.0, + }, + }, + } + """ + raise NotImplementedError + + def set_poe_enabled(self, interface: str, enabled: bool) -> None: + """ + Administratively enables or disables PoE on a single port. + + :param interface: Interface name (e.g. ``"GigabitEthernet0/1"``). + :param enabled: ``True`` to enable PoE, ``False`` to disable it. + :raises NotImplementedError: If the driver does not support PoE control. + :raises ValueError: If the interface does not exist or does not support PoE. + + Example:: + + driver.set_poe_enabled("GigabitEthernet0/1", False) # cut power + driver.set_poe_enabled("GigabitEthernet0/1", True) # restore + """ + raise NotImplementedError + + def power_cycle_port(self, interface: str, delay: int = 5) -> None: + """ + Power-cycles the PoE port: cuts power, waits ``delay`` seconds, then + restores power. Useful for rebooting a hung IP camera, AP, or IP phone + without physical access. + + The method blocks until the full cycle (off → wait → on) is complete. + After it returns the port is back in the delivering/searching state. + + :param interface: Interface name (e.g. ``"GigabitEthernet0/1"``). + :param delay: Seconds to keep the port powered off (default: 5). + :raises NotImplementedError: If the driver does not support PoE control. + :raises ValueError: If the interface does not exist, does not support PoE, + or PoE is administratively disabled on the port. + + Example:: + + driver.power_cycle_port("GigabitEthernet0/1") # 5 s off + driver.power_cycle_port("GigabitEthernet0/1", delay=15) # 15 s off + """ + raise NotImplementedError diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..029750f --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,66 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.backends.legacy:build" + +[project] +name = "napalm-device-types" +version = "0.1.0" +description = "Abstract device-type base classes for NAPALM drivers" +readme = "README.md" +requires-python = ">=3.9" +license = { text = "Apache-2.0" } +authors = [ + { name = "Christian Manivong", email = "christian@manivong.de" }, +] +keywords = [ + "napalm", + "network", + "automation", + "driver", + "access-point", + "switch", + "firewall", + "hypervisor", + "storage", + "nas", +] +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "Intended Audience :: System Administrators", + "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", + "Topic :: System :: Networking", + "Topic :: System :: Systems Administration", + "Typing :: Typed", +] +dependencies = [ + "napalm>=4.0", +] + +[project.urls] +Homepage = "https://github.com/chrismanivong/napalm-device-types" +Repository = "https://github.com/chrismanivong/napalm-device-types" +Issues = "https://github.com/chrismanivong/napalm-device-types/issues" + +[project.optional-dependencies] +dev = [ + "pytest", + "mypy", + "build", + "twine", +] + +[tool.setuptools.packages.find] +where = ["."] +include = ["napalm_device_types*"] + +[tool.mypy] +python_version = "3.9" +strict = true +ignore_missing_imports = true