From f3ecf14c8d34eb2f8f56c8677166d237ce062dce Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Fri, 29 May 2026 09:24:39 +0200 Subject: [PATCH] initial commit --- .gitignore | 74 ++ CHANGELOG.md | 37 + LICENSE | 165 +++ README.md | 176 +++ napalm_proxmox/__init__.py | 5 + napalm_proxmox/driver.py | 2109 ++++++++++++++++++++++++++++++++++++ napalm_proxmox/utils.py | 269 +++++ pyproject.toml | 65 ++ tests/__init__.py | 1 + tests/conftest.py | 217 ++++ tests/test_connection.py | 75 ++ tests/test_environment.py | 63 ++ tests/test_get_facts.py | 42 + tests/test_interfaces.py | 109 ++ tests/test_misc.py | 323 ++++++ tests/test_ovs_and_arp.py | 163 +++ tests/test_sdn.py | 89 ++ tests/test_utils.py | 224 ++++ 18 files changed, 4206 insertions(+) create mode 100644 .gitignore create mode 100644 CHANGELOG.md create mode 100644 LICENSE create mode 100644 README.md create mode 100644 napalm_proxmox/__init__.py create mode 100644 napalm_proxmox/driver.py create mode 100644 napalm_proxmox/utils.py create mode 100644 pyproject.toml create mode 100644 tests/__init__.py create mode 100644 tests/conftest.py create mode 100644 tests/test_connection.py create mode 100644 tests/test_environment.py create mode 100644 tests/test_get_facts.py create mode 100644 tests/test_interfaces.py create mode 100644 tests/test_misc.py create mode 100644 tests/test_ovs_and_arp.py create mode 100644 tests/test_sdn.py create mode 100644 tests/test_utils.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e7d7487 --- /dev/null +++ b/.gitignore @@ -0,0 +1,74 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class +*.pyo + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +pip-wheel-metadata/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +pytest_cache/ + +# Virtual environments +.venv +venv/ +env/ +ENV/ + +# IDE +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# macOS +.DS_Store +.AppleDouble +.LSOverride + +# Logs +*.log + +# Secrets / local config +*.env +.secrets diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..51acf3c --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,37 @@ +# Changelog + +All notable changes to this project will be documented in this file. + +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), +and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). + +## [Unreleased] + +## [0.1.0] - 2024-01-01 + +### Added +- Initial release of the Proxmox VE NAPALM driver +- Support for classic Linux networking (`/etc/network/interfaces`, Linux bridges) +- Support for Proxmox SDN (zones: VLAN, QinQ, VXLAN, EVPN; VNets and subnets) +- Support for Open vSwitch (OVS) bridges, bonds, and internal ports +- Password and API token authentication via `proxmoxer` +- Implemented NAPALM methods: + - `open`, `close`, `is_alive` + - `get_facts` + - `get_interfaces`, `get_interfaces_ip`, `get_interfaces_counters` + - `get_environment` (CPU, memory, temperature sensors) + - `get_arp_table` (via `/proc/net/arp`) + - `get_mac_address_table` (Linux bridge fdb + OVS appctl) + - `get_vlans` (SDN VNet tags + `bridge vlan show`) + - `get_network_instances` (SDN zones as NAPALM network instances) + - `get_ntp_servers`, `get_ntp_stats` + - `get_snmp_information` + - `get_users` + - `get_config`, `load_merge_candidate`, `load_replace_candidate` + - `compare_config`, `commit_config`, `discard_config`, `rollback` + - `get_lldp_neighbors`, `get_lldp_neighbors_detail` + - `get_ipv6_neighbors_table` + - `ping`, `traceroute`, `cli` + - `get_route_to` +- 149 unit tests with ~79% code coverage +- Full type annotations diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..8ef1c6b --- /dev/null +++ b/LICENSE @@ -0,0 +1,165 @@ + 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 modifications + 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 tracking and + discussing 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 included + 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 combination of their Contributions + with the Work to which such Contributions were 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 in addition 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, sublicense, and/or sell copies of the Work, and to permit + persons to whom the Work is furnished to do so. + + 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, + NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A PARTICULAR + PURPOSE. You are solely responsible for determining the + appropriateness of using or reproducing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), strict liability, or contract + 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 your own + liability and not on behalf of any other Contributor, and only if + You agree to indemnify, defend, and hold each Contributor harmless + for any liability incurred by, or claims asserted against, such + Contributor by reason of your accepting any such warranty or + additional liability. + + END OF TERMS AND CONDITIONS diff --git a/README.md b/README.md new file mode 100644 index 0000000..48bfbba --- /dev/null +++ b/README.md @@ -0,0 +1,176 @@ +# napalm-proxmox + +A [NAPALM](https://napalm.readthedocs.io/) driver for [Proxmox VE](https://www.proxmox.com/en/proxmox-virtual-environment/overview) nodes. + +It supports all three networking domains of Proxmox VE: + +| Domain | Description | +|--------|-------------| +| **Classic Linux** | `/etc/network/interfaces`, Linux bridges, VLANs | +| **SDN** | Zones (VLAN, QinQ, VXLAN, EVPN), VNets, subnets | +| **OVS** | Open vSwitch bridges, bonds, and internal ports | + +## Requirements + +- Python ≥ 3.9 +- NAPALM ≥ 5.0.0 +- proxmoxer ≥ 2.0.0 +- netaddr ≥ 0.9.0 +- requests ≥ 2.31.0 + +## Installation + +```bash +pip install napalm-proxmox +``` + +Or directly from source: + +```bash +git clone https://github.com/example/napalm-proxmox.git +cd napalm-proxmox +pip install -e . +``` + +## Usage + +### Password authentication + +```python +from napalm import get_network_driver + +driver = get_network_driver("proxmox") +device = driver( + hostname="pve1.example.com", + username="root", + password="secret", + optional_args={ + "realm": "pam", # default: "pam" + "port": 8006, # default: 8006 + "verify_ssl": True, # default: True + }, +) + +with device: + facts = device.get_facts() + print(facts) + # {'vendor': 'Proxmox Server Solutions GmbH', 'model': 'PowerEdge R640', + # 'hostname': 'pve1', 'os_version': 'Proxmox VE 8.2.4', ...} + + interfaces = device.get_interfaces() + interfaces_ip = device.get_interfaces_ip() + vlans = device.get_vlans() + network_instances = device.get_network_instances() +``` + +### API token authentication + +```python +device = driver( + hostname="pve1.example.com", + username="napalm", + password="", + optional_args={ + "token_name": "napalm@pam!napalm-token", + "token_value": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", + }, +) +``` + +### Targeting a specific node + +In a multi-node cluster the driver auto-detects the node by matching the +hostname against the cluster's node list. You can also pin a specific node: + +```python +device = driver( + hostname="pve-cluster.example.com", # cluster VIP + username="root", + password="secret", + optional_args={"node": "pve2"}, +) +``` + +## Supported NAPALM methods + +| Method | Notes | +|--------|-------| +| `open` / `close` | Password + API token auth | +| `is_alive` | Checks `/api2/json/version` | +| `get_facts` | vendor, model, hostname, os_version, uptime, interface_list, fqdn | +| `get_interfaces` | eth, bridge, OVS bridge/bond; speed, MTU, MAC, status | +| `get_interfaces_ip` | IPv4/IPv6 from classic ifaces + SDN subnet gateways | +| `get_interfaces_counters` | From `node/netstat` RRD data | +| `get_environment` | CPU %, memory (bytes), hardware temperature sensors | +| `get_arp_table` | Reads `/proc/net/arp` via exec endpoint | +| `get_mac_address_table` | `bridge fdb show` (Linux) + `ovs-appctl fdb/show` (OVS) | +| `get_vlans` | SDN VNet tags + `bridge vlan show` | +| `get_network_instances` | SDN zones (VLAN→L2VPN, EVPN→L3VRF) + default instance | +| `get_ntp_servers` | From `node/ntp` API | +| `get_ntp_stats` | Parses `chronyc` / `ntpq -pn` output | +| `get_snmp_information` | Parses `/etc/snmp/snmpd.conf` | +| `get_users` | Proxmox access/users API + local `/etc/passwd` | +| `get_config` | `/etc/network/interfaces` + SDN config; sanitize support | +| `load_merge_candidate` / `load_replace_candidate` | String or file | +| `compare_config` | Unified diff | +| `commit_config` | Writes `/etc/network/interfaces` + `ifreload -a` | +| `discard_config` / `rollback` | Revert candidate | +| `get_lldp_neighbors` | `lldpcli show neighbors` via exec | +| `get_lldp_neighbors_detail` | Full LLDP detail | +| `get_ipv6_neighbors_table` | `ip -6 neigh show` | +| `ping` | Linux `ping` via exec endpoint | +| `traceroute` | `traceroute -n` via exec endpoint | +| `cli` | Arbitrary command execution via exec endpoint | +| `get_route_to` | `ip route show` via exec endpoint | + +**Not implemented** (raise `NotImplementedError`): +`get_bgp_config`, `get_bgp_neighbors`, `get_bgp_neighbors_detail`, +`get_optics`, `get_probes_config`, `get_probes_results`, `get_firewall_policies` + +## Optional arguments + +| Argument | Type | Default | Description | +|----------|------|---------|-------------| +| `port` | int | `8006` | Proxmox API port | +| `verify_ssl` | bool | `True` | Verify TLS certificate | +| `realm` | str | `"pam"` | PAM realm for password auth | +| `node` | str | *auto* | Override target node name | +| `token_name` | str | `None` | API token identifier (`user@realm!tokenid`) | +| `token_value` | str | `None` | API token secret | + +## Development + +```bash +python3 -m venv .venv +source .venv/bin/activate +pip install -e ".[dev]" + +# Run tests with coverage +pytest --cov=napalm_proxmox --cov-report=term-missing + +# Lint +ruff check napalm_proxmox +``` + +## Architecture + +``` +napalm_proxmox/ +├── __init__.py # Exports ProxmoxDriver +├── driver.py # ProxmoxDriver — all NAPALM methods +└── utils.py # MAC normalisation, CIDR parsing, ARP/OVS parsers +tests/ +├── conftest.py # Shared fixtures and mock API payloads +├── test_connection.py # open / close / is_alive +├── test_get_facts.py # get_facts +├── test_interfaces.py # get_interfaces, get_interfaces_ip, get_interfaces_counters +├── test_environment.py # get_environment +├── test_sdn.py # get_vlans, get_network_instances +├── test_ovs_and_arp.py # get_arp_table, get_mac_address_table, ARP/OVS utils +├── test_misc.py # NTP, SNMP, users, config, ping, traceroute, CLI, LLDP +└── test_utils.py # Unit tests for napalm_proxmox.utils +``` + +## License + +Apache License 2.0 — see [LICENSE](LICENSE). diff --git a/napalm_proxmox/__init__.py b/napalm_proxmox/__init__.py new file mode 100644 index 0000000..f9faf44 --- /dev/null +++ b/napalm_proxmox/__init__.py @@ -0,0 +1,5 @@ +"""NAPALM driver for Proxmox VE.""" + +from napalm_proxmox.driver import ProxmoxDriver + +__all__ = ["ProxmoxDriver"] diff --git a/napalm_proxmox/driver.py b/napalm_proxmox/driver.py new file mode 100644 index 0000000..8c4e8a3 --- /dev/null +++ b/napalm_proxmox/driver.py @@ -0,0 +1,2109 @@ +"""NAPALM driver for Proxmox VE. + +Supports: + - Classic Linux networking (/etc/network/interfaces via Proxmox API) + - Software-Defined Networking (SDN): zones, VNets, subnets + - Open vSwitch (OVS) bridges, bonds, and internal ports + +Connection is made via the Proxmox REST API (``proxmoxer`` library). +The driver targets the *node* level: each Proxmox node is treated as a +network device. Cluster-wide SDN information is also exposed where the +NAPALM API allows it. + +Optional args +------------- +verify_ssl : bool + Verify TLS certificates (default: True). +port : int + Proxmox API port (default: 8006). +node : str + Override the target node name (default: auto-detected from hostname). +realm : str + PAM realm (default: ``pam``). +token_name : str + API token name (e.g. ``napalm@pam!mytoken``). +token_value : str + API token secret. When both token_name and token_value are provided, + token-based auth is used instead of password auth. +""" + +from __future__ import annotations + +import logging +import re +import socket +from typing import Any + +logger = logging.getLogger(__name__) + +from napalm_device_types import HypervisorDriver +from napalm.base.exceptions import ( + ConnectionException, + SessionLockedException, +) +from napalm.base.helpers import mac as napalm_mac +from napalm.base.netmiko_helpers import netmiko_args +import napalm.base.constants as C + +try: + from proxmoxer import ProxmoxAPI + from proxmoxer.core import ResourceException +except ImportError as exc: # pragma: no cover + raise ImportError( + "proxmoxer is required: pip install proxmoxer" + ) from exc + +from napalm_proxmox import utils + +# --------------------------------------------------------------------------- # +# Type aliases +# --------------------------------------------------------------------------- # +_JsonDict = dict[str, Any] + +# --------------------------------------------------------------------------- # +# Driver +# --------------------------------------------------------------------------- # + + +class ProxmoxDriver(HypervisorDriver): + """NAPALM driver for Proxmox VE nodes.""" + + platform = "proxmox" + + def __init__( + self, + hostname: str, + username: str, + password: str, + timeout: int = 60, + optional_args: _JsonDict | None = None, + ) -> None: + self.hostname = hostname + self.username = username + self.password = password + self.timeout = timeout + self.optional_args: _JsonDict = optional_args or {} + + self._port: int = self.optional_args.get("port", 8006) + self._verify_ssl: bool = self.optional_args.get("verify_ssl", True) + self._realm: str = self.optional_args.get("realm", "pam") + self._token_name: str | None = self.optional_args.get("token_name") + self._token_value: str | None = self.optional_args.get("token_value") + self._node: str | None = self.optional_args.get("node") + + self._api: ProxmoxAPI | None = None + self._node_name: str = "" + + # Candidate config (merge/replace) + self._candidate_config: str = "" + self._running_config: str = "" + + # ------------------------------------------------------------------ # + # Connection management + # ------------------------------------------------------------------ # + + def open(self) -> None: + """Open the connection to the Proxmox API.""" + try: + kwargs: _JsonDict = { + "host": self.hostname, + "port": self._port, + "verify_ssl": self._verify_ssl, + "timeout": self.timeout, + } + + if self._token_name and self._token_value: + # Token-based auth: token_name format is "user@realm!tokenid" + # proxmoxer needs user="user@realm", token_name="tokenid", token_value="..." + if "!" in self._token_name: + user_part, token_id = self._token_name.split("!", 1) + else: + user_part = self.username or "root@pam" + token_id = self._token_name + kwargs["user"] = user_part + kwargs["token_name"] = token_id + kwargs["token_value"] = self._token_value + else: + kwargs["user"] = f"{self.username}@{self._realm}" + kwargs["password"] = self.password + + self._api = ProxmoxAPI(**kwargs) + + # Determine which node we're talking to + self._node_name = self._resolve_node() + except Exception as exc: + raise ConnectionException( + f"Cannot connect to Proxmox at {self.hostname}:{self._port} — {exc}" + ) from exc + + def _resolve_node(self) -> str: + """Return the Proxmox node name for this host.""" + if self._node: + return self._node + # Try to match hostname against listed nodes + try: + nodes = self._api.nodes.get() # type: ignore[union-attr] + except Exception: + nodes = [] + short_host = self.hostname.split(".")[0].lower() + for node_entry in nodes: + name = node_entry.get("node", "") + if name.lower() == short_host or name.lower() == self.hostname.lower(): + return name + # Fall back to first online node + for node_entry in nodes: + if node_entry.get("status") == "online": + return node_entry["node"] + # Last resort: use the shortened hostname + return short_host + + def close(self) -> None: + """Close the session (Proxmox REST is stateless; nothing to tear down).""" + self._api = None + + def is_alive(self) -> _JsonDict: + """Return connection state.""" + try: + self._api.version.get() # type: ignore[union-attr] + return {"is_alive": True} + except Exception: + return {"is_alive": False} + + # ------------------------------------------------------------------ # + # Internal helpers + # ------------------------------------------------------------------ # + + def _node_api(self): + """Return the proxmoxer sub-resource for the target node.""" + return self._api.nodes(self._node_name) # type: ignore[union-attr] + + def _get_node_network(self) -> list[_JsonDict]: + """Return the list of network interfaces from the Proxmox node API.""" + try: + return self._node_api().network.get() or [] + except ResourceException: + return [] + + def _exec_ssh_command(self, command: str) -> str: + """Execute a shell command on the Proxmox node. + + First tries the Proxmox API execute endpoint. If that fails (e.g. + because token-auth is blocked on that endpoint), falls back to SSH + using the driver's username / password credentials. + Falls back to an empty string if both methods are unavailable. + """ + # 1. Try Proxmox API execute endpoint + try: + result = self._node_api().execute.post(command=command) + data = result.get("data", "") + if isinstance(data, str): + return data + except Exception: + pass + + # 2. Fall back to SSH if we have credentials + if not self.username or not self.password: + return "" + try: + import paramiko # noqa: PLC0415 + ssh_port: int = self.optional_args.get("ssh_port", 22) + client = paramiko.SSHClient() + client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + client.connect( + hostname=self.hostname, + port=ssh_port, + username=self.username, + password=self.password, + timeout=self.timeout, + look_for_keys=False, + allow_agent=False, + ) + _, stdout, _ = client.exec_command(command, timeout=self.timeout) + output = stdout.read().decode("utf-8", errors="replace") + client.close() + return output + except Exception: + return "" + + def _get_version_info(self) -> _JsonDict: + """Return Proxmox version dict.""" + try: + return self._api.version.get() or {} # type: ignore[union-attr] + except Exception: + return {} + + def _get_node_status(self) -> _JsonDict: + """Return node status dict.""" + try: + return self._node_api().status.get() or {} + except Exception: + return {} + + def _get_node_subscription(self) -> _JsonDict: + try: + return self._node_api().subscription.get() or {} + except Exception: + return {} + + def _get_sdn_zones(self) -> list[_JsonDict]: + try: + return self._api.cluster.sdn.zones.get() or [] # type: ignore[union-attr] + except Exception: + return [] + + def _get_sdn_vnets(self) -> list[_JsonDict]: + try: + return self._api.cluster.sdn.vnets.get() or [] # type: ignore[union-attr] + except Exception: + return [] + + def _get_sdn_subnets(self, vnet: str) -> list[_JsonDict]: + try: + return self._api.cluster.sdn.vnets(vnet).subnets.get() or [] + except Exception: + return [] + + def _get_node_dns(self) -> _JsonDict: + try: + return self._node_api().dns.get() or {} + except Exception: + return {} + + def _get_node_time(self) -> _JsonDict: + try: + return self._node_api().time.get() or {} + except Exception: + return {} + + def _get_node_ntp(self) -> _JsonDict: + try: + return self._node_api().ntp.get() or {} + except Exception: + return {} + + # ------------------------------------------------------------------ # + # get_facts + # ------------------------------------------------------------------ # + + def get_facts(self) -> _JsonDict: + """Return basic facts about the Proxmox node.""" + status = self._get_node_status() + version = self._get_version_info() + network = self._get_node_network() + dns = self._get_node_dns() + + uptime = float(status.get("uptime", 0)) + model = status.get("model", "") + # dns["search"] is the DNS search domain (e.g. "home.example.com"), not the hostname + dns_search = dns.get("search", "") + hostname = self._node_name + fqdn = f"{self._node_name}.{dns_search}" if dns_search else self.hostname + + # Build interface list + iface_list = sorted( + iface["iface"] for iface in network if iface.get("iface") + ) + + # PVE version looks like "8.2.4" + pve_version = version.get("version", "") + release = version.get("release", "") + os_version = f"Proxmox VE {pve_version}" if pve_version else f"Proxmox VE {release}" + + return { + "uptime": uptime, + "vendor": "Proxmox Server Solutions GmbH", + "model": model or "Proxmox VE Node", + "hostname": self._node_name, + "fqdn": fqdn or self.hostname, + "os_version": os_version, + "serial_number": "", + "interface_list": iface_list, + } + + # ------------------------------------------------------------------ # + # get_interfaces + # ------------------------------------------------------------------ # + + def get_interfaces(self) -> dict[str, _JsonDict]: + """Return a dict of interfaces keyed by interface name.""" + result: dict[str, _JsonDict] = {} + for iface in self._get_node_network(): + name = iface.get("iface", "") + if not name: + continue + + # Proxmox marks bridge ports as not independent — include all + active = iface.get("active", 0) + autostart = iface.get("autostart", 0) + + result[name] = { + "is_up": bool(active), + "is_enabled": bool(autostart) or bool(active), + "description": iface.get("comments", "").strip(), + "last_flapped": -1.0, + "speed": utils.speed_mbps(iface), + "mtu": int(iface.get("mtu") or 1500), + "mac_address": utils.normalize_mac(iface.get("hwaddr", "")), + } + return result + + # ------------------------------------------------------------------ # + # get_interfaces_ip + # ------------------------------------------------------------------ # + + def get_interfaces_ip(self) -> dict[str, _JsonDict]: + """Return IP addresses per interface.""" + result: dict[str, _JsonDict] = {} + for iface in self._get_node_network(): + name = iface.get("iface", "") + if not name: + continue + addrs = utils.addresses_from_node_network(iface) + if addrs: + result[name] = addrs + + # Overlay SDN VNet addresses (subnets with gateways) + for vnet in self._get_sdn_vnets(): + vnet_id = vnet.get("vnet", "") + if not vnet_id: + continue + for subnet in self._get_sdn_subnets(vnet_id): + cidr = subnet.get("cidr", "") + gateway = subnet.get("gateway", "") + if gateway and cidr: + ip, plen = utils.parse_cidr(cidr) + if "." in gateway: + result.setdefault(vnet_id, {}).setdefault("ipv4", {})[gateway] = { + "prefix_length": plen + } + else: + result.setdefault(vnet_id, {}).setdefault("ipv6", {})[gateway] = { + "prefix_length": plen + } + return result + + # ------------------------------------------------------------------ # + # get_interfaces_counters + # ------------------------------------------------------------------ # + + def get_interfaces_counters(self) -> dict[str, _JsonDict]: + """Return per-interface traffic counters.""" + result: dict[str, _JsonDict] = {} + try: + rrd_data = self._node_api().netstat.get() or [] + except Exception: + rrd_data = [] + + # Proxmox /nodes/{node}/netstat returns a list of time-series points. + # Take the most recent (last) entry for each interface. + latest: dict[str, _JsonDict] = {} + for entry in rrd_data: + iface = entry.get("dev", "") + if iface: + latest[iface] = entry + + for iface, data in latest.items(): + result[iface] = { + "tx_errors": int(data.get("tx_errs", 0) or 0), + "rx_errors": int(data.get("rx_errs", 0) or 0), + "tx_discards": int(data.get("tx_drop", 0) or 0), + "rx_discards": int(data.get("rx_drop", 0) or 0), + "tx_octets": int(data.get("tx_bytes", 0) or 0), + "rx_octets": int(data.get("rx_bytes", 0) or 0), + "tx_unicast_packets": int(data.get("tx_packets", 0) or 0), + "rx_unicast_packets": int(data.get("rx_packets", 0) or 0), + "tx_multicast_packets": 0, + "rx_multicast_packets": 0, + "tx_broadcast_packets": 0, + "rx_broadcast_packets": 0, + } + return result + + # ------------------------------------------------------------------ # + # get_environment + # ------------------------------------------------------------------ # + + def get_environment(self) -> _JsonDict: + """Return environment status (CPU, memory, temperature).""" + status = self._get_node_status() + env: _JsonDict = { + "fans": {}, + "temperature": {}, + "power": {}, + "cpu": {}, + "memory": {"available_ram": 0, "used_ram": 0}, + } + + # CPU + cpu_usage = status.get("cpu", 0.0) + env["cpu"]["0"] = {"%usage": round(float(cpu_usage) * 100, 2)} + + # Memory (Proxmox reports in bytes) + mem = status.get("memory", {}) + total = int(mem.get("total", 0) or 0) + used = int(mem.get("used", 0) or 0) + env["memory"]["available_ram"] = total + env["memory"]["used_ram"] = used + + # Temperature (from node sensors if available) + try: + sensors = self._node_api().hardware.sensors.get() or [] + except Exception: + sensors = [] + for sensor in sensors: + name = sensor.get("name", "unknown") + value = sensor.get("value", None) + if value is not None: + try: + temp_c = float(value) + env["temperature"][name] = { + "temperature": temp_c, + "is_alert": temp_c >= 80.0, + "is_critical": temp_c >= 95.0, + } + except (TypeError, ValueError): + pass + + return env + + # ------------------------------------------------------------------ # + # get_arp_table + # ------------------------------------------------------------------ # + + def get_arp_table(self, vrf: str = "") -> list[_JsonDict]: + """Return ARP table. + + Proxmox does not expose ARP via the REST API directly. We attempt + to read it via the node's ``/proc/net/arp`` through the Proxmox + exec endpoint. If that is unavailable, an empty list is returned. + """ + raw = self._exec_ssh_command("cat /proc/net/arp") + if not raw: + return [] + + entries = [] + # /proc/net/arp format: + # IP address HW type Flags HW address Mask Device + # 192.168.1.1 0x1 0x2 aa:bb:cc:dd:ee:ff * eth0 + for line in raw.splitlines(): + line = line.strip() + if not line or line.startswith("IP"): + continue + parts = line.split() + if len(parts) < 6: + continue + ip_addr, _, flags, mac, _, iface = ( + parts[0], parts[1], parts[2], parts[3], parts[4], parts[5] + ) + if mac in ("00:00:00:00:00:00", ""): + continue + if vrf and iface != vrf: + continue + entries.append( + { + "interface": iface, + "mac": utils.normalize_mac(mac), + "ip": ip_addr, + "age": -1.0, + } + ) + return entries + + # ------------------------------------------------------------------ # + # get_mac_address_table + # ------------------------------------------------------------------ # + + def get_mac_address_table(self) -> list[_JsonDict]: + """Return MAC address table from Linux bridges and OVS bridges.""" + result: list[_JsonDict] = [] + network = self._get_node_network() + + # Linux bridges + linux_bridges = [ + iface["iface"] + for iface in network + if iface.get("type") in ("bridge",) and iface.get("iface") + ] + for bridge in linux_bridges: + raw = self._exec_ssh_command( + f"bridge fdb show br {bridge} 2>/dev/null || true" + ) + for line in raw.splitlines(): + parts = line.split() + if len(parts) < 3: + continue + mac_str = parts[0] + if not re.match(r"([0-9a-f]{2}:){5}[0-9a-f]{2}", mac_str): + continue + # "dev " "vlan " + dev = "" + vlan_id = 1 + for i, tok in enumerate(parts): + if tok == "dev" and i + 1 < len(parts): + dev = parts[i + 1] + if tok == "vlan" and i + 1 < len(parts): + try: + vlan_id = int(parts[i + 1]) + except ValueError: + pass + result.append( + { + "mac": utils.normalize_mac(mac_str), + "interface": dev or bridge, + "vlan": vlan_id, + "static": "permanent" in line, + "active": True, + "moves": 0, + "last_move": 0.0, + } + ) + + # OVS bridges + ovs_bridges = [ + iface["iface"] + for iface in network + if iface.get("type") in ("OVSBridge",) and iface.get("iface") + ] + for bridge in ovs_bridges: + raw = self._exec_ssh_command( + f"ovs-appctl fdb/show {bridge} 2>/dev/null || true" + ) + # Format: port VLAN MAC AGE + for line in raw.splitlines(): + parts = line.split() + if len(parts) < 4: + continue + try: + _port = int(parts[0]) + vlan_id = int(parts[1]) + mac_str = parts[2] + except (ValueError, IndexError): + continue + result.append( + { + "mac": utils.normalize_mac(mac_str), + "interface": bridge, + "vlan": vlan_id, + "static": False, + "active": True, + "moves": 0, + "last_move": 0.0, + } + ) + + return result + + # ------------------------------------------------------------------ # + # get_vlans (SDN VNets + classic Linux bridge VLANs) + # ------------------------------------------------------------------ # + + def get_vlans(self) -> dict[str, _JsonDict]: + """Return VLAN table. + + For OVS+SDN nodes: reads SDN VNets for VLAN IDs/names, then maps + OVSIntPort (access ports with ovs_tag) → untagged membership, and + OVSPort / OVSBridge (trunk ports) → tagged membership. + + Falls back to ``bridge vlan show`` for classic Linux-bridge nodes. + """ + result: dict[str, _JsonDict] = {} + node_network = self._get_node_network() + + # --- SDN vnets → VLAN IDs and initial entries --- + for vnet in self._get_sdn_vnets(): + tag = vnet.get("tag") + vnet_id = vnet.get("vnet", "") + if tag is None: + continue + try: + tag_int = int(tag) + except (ValueError, TypeError): + continue + result[str(tag_int)] = { + "name": vnet_id, + "tagged": [], + "untagged": [], + } + + # --- OVS port membership --- + trunk_ports: list[str] = [] + access_by_vlan: dict[str, list[str]] = {} + + for iface in node_network: + ovs_type = iface.get("ovs_type", "") + iface_name = iface.get("iface", "") + if not iface_name: + continue + if ovs_type == "OVSIntPort": + ovs_tag = iface.get("ovs_tag") + if ovs_tag is not None: + vid = str(int(ovs_tag)) + access_by_vlan.setdefault(vid, []).append(iface_name) + elif ovs_type in ("OVSPort", "OVSBridge"): + # Trunk: carries all VLANs tagged + trunk_ports.append(iface_name) + + if trunk_ports or access_by_vlan: + # OVS topology detected — assign tagged/untagged per VLAN + for vid, vlan_entry in result.items(): + vlan_entry["tagged"] = list(trunk_ports) + vlan_entry["untagged"] = list(access_by_vlan.get(vid, [])) + return result + + # --- Linux bridge fallback (bridge vlan show) --- + # Convert result entries to use tagged/untagged keys + for entry in result.values(): + entry.setdefault("tagged", []) + entry.setdefault("untagged", []) + + raw = self._exec_ssh_command("bridge vlan show 2>/dev/null || true") + current_iface = "" + for line in raw.splitlines(): + line = line.strip() + if not line: + continue + m = re.match(r"^(\S+)\s+(\d+)", line) + if m: + current_iface = m.group(1) + vid = m.group(2) + else: + m2 = re.match(r"^\s*(\d+)", line) + if m2: + vid = m2.group(1) + else: + continue + if current_iface and vid: + entry = result.setdefault(vid, {"name": "", "tagged": [], "untagged": []}) + if "Untagged" in line or "PVID" in line: + if current_iface not in entry["untagged"]: + entry["untagged"].append(current_iface) + else: + if current_iface not in entry["tagged"]: + entry["tagged"].append(current_iface) + + return result + + # ------------------------------------------------------------------ # + # get_network_instances (SDN Zones as VRF-like instances) + # ------------------------------------------------------------------ # + + def get_network_instances(self, name: str = "") -> dict[str, _JsonDict]: + """Return SDN zones as network instances.""" + result: dict[str, _JsonDict] = {} + + # Always include the default instance + result["default"] = { + "name": "default", + "type": "DEFAULT_INSTANCE", + "state": {"route_distinguisher": None}, + "interfaces": {"interface": {}}, + } + + # Attach non-SDN interfaces to the default instance + for iface in self._get_node_network(): + iface_name = iface.get("iface", "") + if iface_name: + result["default"]["interfaces"]["interface"][iface_name] = {} + + # SDN Zones + for zone in self._get_sdn_zones(): + zone_id = zone.get("zone", zone.get("name", "")) + if not zone_id: + continue + if name and zone_id != name: + continue + instance = utils.sdn_zone_to_network_instance(zone) + # Attach VNets that belong to this zone + for vnet in self._get_sdn_vnets(): + if vnet.get("zone") == zone_id: + vnet_id = vnet.get("vnet", "") + if vnet_id: + instance["interfaces"]["interface"][vnet_id] = {} + result[zone_id] = instance + + if name: + return {k: v for k, v in result.items() if k == name} + return result + + # ------------------------------------------------------------------ # + # get_ntp_servers / get_ntp_stats + # ------------------------------------------------------------------ # + + def get_ntp_servers(self) -> dict[str, _JsonDict]: + """Return configured NTP servers.""" + ntp = self._get_node_ntp() + servers: dict[str, _JsonDict] = {} + # Proxmox reports a comma-separated or space-separated server list + raw = ntp.get("server", "") or ntp.get("servers", "") + for srv in re.split(r"[\s,]+", raw): + srv = srv.strip() + if srv: + servers[srv] = {} + return servers + + def get_ntp_stats(self) -> list[_JsonDict]: + """Return NTP synchronisation statistics from chronyc/ntpq output.""" + raw = self._exec_ssh_command( + "chronyc -n tracking 2>/dev/null || ntpq -pn 2>/dev/null || true" + ) + stats: list[_JsonDict] = [] + for line in raw.splitlines(): + line = line.strip() + # ntpq -pn format: *remote refid st t when poll reach delay offset jitter + m = re.match( + r"^([\*\+\-\s])([\d.]+)\s+([\d.]+)\s+(\d+)\s+\S+\s+(\S+)\s+(\d+)\s+(\d+)\s+([\d.]+)\s+([-\d.]+)\s+([\d.]+)", + line, + ) + if m: + synced = m.group(1).strip() == "*" + stats.append( + { + "remote": m.group(2), + "referenceid": m.group(3), + "synchronized": synced, + "stratum": int(m.group(4)), + "type": "", + "when": m.group(5), + "hostpoll": int(m.group(6)), + "reachability": int(m.group(7)), + "delay": float(m.group(8)), + "offset": float(m.group(9)), + "jitter": float(m.group(10)), + } + ) + return stats + + # ------------------------------------------------------------------ # + # get_snmp_information + # ------------------------------------------------------------------ # + + def get_snmp_information(self) -> _JsonDict: + """Return SNMP information. + + Proxmox does not expose SNMP configuration via the REST API. + We read /etc/snmp/snmpd.conf via exec if available. + """ + raw = self._exec_ssh_command( + "cat /etc/snmp/snmpd.conf 2>/dev/null || true" + ) + communities: dict[str, _JsonDict] = {} + location = "" + contact = "" + for line in raw.splitlines(): + line = line.strip() + if line.startswith("#") or not line: + continue + # rocommunity [source] + m = re.match(r"^(ro|rw)community\s+(\S+)", line) + if m: + mode = "ro" if m.group(1) == "ro" else "rw" + community = m.group(2) + communities[community] = {"acl": "N/A", "mode": mode} + m_loc = re.match(r"^sysLocation\s+(.+)", line) + if m_loc: + location = m_loc.group(1).strip() + m_con = re.match(r"^sysContact\s+(.+)", line) + if m_con: + contact = m_con.group(1).strip() + + return { + "chassis_id": self._node_name, + "community": communities, + "contact": contact, + "location": location, + } + + # ------------------------------------------------------------------ # + # get_users + # ------------------------------------------------------------------ # + + def get_users(self) -> dict[str, _JsonDict]: + """Return users configured on the Proxmox node. + + Reads from both the Proxmox access/users API and local /etc/passwd. + """ + result: dict[str, _JsonDict] = {} + try: + pve_users = self._api.access.users.get() or [] # type: ignore[union-attr] + except Exception: + pve_users = [] + + for user in pve_users: + uid = user.get("userid", "") + if not uid: + continue + # Proxmox roles: Administrator → 15, otherwise 1 + groups = user.get("groups", "") or "" + level = 1 + try: + roles = self._api.access.users(uid).get() or {} # type: ignore[union-attr] + if "Administrator" in str(roles): + level = 15 + except Exception: + pass + result[uid] = { + "level": level, + "password": "", + "sshkeys": [], + } + + # Merge local OS users from /etc/passwd + raw = self._exec_ssh_command("getent passwd 2>/dev/null || cat /etc/passwd") + for line in raw.splitlines(): + parts = line.split(":") + if len(parts) < 7: + continue + uname, _, uid_str, *_ = parts + try: + uid_int = int(uid_str) + except ValueError: + continue + if uname not in result and uid_int < 1000 or uid_int == 0: + result[uname] = { + "level": 15 if uid_int == 0 else 0, + "password": "", + "sshkeys": [], + } + + return result + + # ------------------------------------------------------------------ # + # get_config + # ------------------------------------------------------------------ # + + def get_config( + self, + retrieve: str = "all", + full: bool = False, + sanitized: bool = False, + format: str = "text", + ) -> _JsonDict: + """Return the node network configuration. + + ``running`` config is the contents of ``/etc/network/interfaces`` + (and the SDN config directory). ``startup`` is identical (PVE + applies on boot). ``candidate`` is what was loaded via + ``load_merge_candidate`` / ``load_replace_candidate`` but not yet + committed. + """ + configs: _JsonDict = {"running": "", "candidate": "", "startup": ""} + + if retrieve in ("running", "all", "startup"): + raw = self._exec_ssh_command( + "cat /etc/network/interfaces 2>/dev/null || true" + ) + # Append SDN config if available + sdn_raw = self._exec_ssh_command( + "cat /etc/pve/sdn/vnets.cfg 2>/dev/null || true" + ) + running = raw + if sdn_raw: + running += "\n# === SDN VNets ===\n" + sdn_raw + if sanitized: + running = re.sub(r"password\s+\S+", "password ****", running) + configs["running"] = running + self._running_config = running + if retrieve == "startup": + configs["startup"] = running + elif retrieve == "all": + configs["startup"] = running + + if retrieve in ("candidate", "all"): + configs["candidate"] = self._candidate_config + + return configs + + # ------------------------------------------------------------------ # + # Config management (load / compare / commit / discard / rollback) + # ------------------------------------------------------------------ # + + def load_merge_candidate( + self, + filename: str | None = None, + config: str | None = None, + ) -> None: + """Load a candidate configuration (merge mode).""" + if filename: + with open(filename) as fh: + config = fh.read() + if config is None: + raise ValueError("Either filename or config must be provided") + # In merge mode we append / overlay + self._candidate_config = config + + def load_replace_candidate( + self, + filename: str | None = None, + config: str | None = None, + ) -> None: + """Load a candidate configuration (replace mode).""" + if filename: + with open(filename) as fh: + config = fh.read() + if config is None: + raise ValueError("Either filename or config must be provided") + self._candidate_config = config + + def compare_config(self) -> str: + """Return a unified diff between running and candidate config.""" + import difflib + + if not self._running_config: + self.get_config(retrieve="running") + running_lines = self._running_config.splitlines(keepends=True) + candidate_lines = self._candidate_config.splitlines(keepends=True) + diff = difflib.unified_diff( + running_lines, + candidate_lines, + fromfile="running", + tofile="candidate", + ) + return "".join(diff) + + def commit_config(self, message: str = "", revert_in: int | None = None) -> None: + """Commit the candidate configuration to the Proxmox node. + + This writes the candidate config to ``/etc/network/interfaces`` + via the Proxmox node/network PUT API (which applies it live). + + .. note:: + Full programmatic apply requires the Proxmox API to accept raw + interface configs. This implementation uses ``pvesh`` via exec + which requires the node exec endpoint to be available. + """ + if not self._candidate_config: + return + # Write via exec endpoint + escaped = self._candidate_config.replace("'", "'\\''") + self._exec_ssh_command( + f"printf '%s' '{escaped}' > /etc/network/interfaces && " + "ifreload -a 2>&1 || ifup -a 2>&1 || true" + ) + self._running_config = self._candidate_config + self._candidate_config = "" + + def discard_config(self) -> None: + """Discard the candidate configuration.""" + self._candidate_config = "" + + def rollback(self) -> None: + """Revert to the stored running configuration.""" + if self._running_config: + self._candidate_config = self._running_config + self.commit_config() + self._candidate_config = "" + + # ------------------------------------------------------------------ # + # get_lldp_neighbors (via lldpcli if installed) + # ------------------------------------------------------------------ # + + def get_lldp_neighbors(self) -> dict[str, list[_JsonDict]]: + """Return LLDP neighbours (requires lldpd on the Proxmox node).""" + result: dict[str, list[_JsonDict]] = {} + raw = self._exec_ssh_command( + "lldpcli show neighbors summary 2>/dev/null || true" + ) + current_iface = "" + for line in raw.splitlines(): + m_iface = re.match(r"^\s*Interface:\s+(\S+?),?\s", line) + if m_iface: + current_iface = m_iface.group(1) + result.setdefault(current_iface, []) + continue + m_sys = re.match(r"^\s*SysName:\s+(.+)", line) + m_port = re.match(r"^\s*PortID:\s+\S+\s+(.+)", line) + if m_sys and current_iface: + hostname = m_sys.group(1).strip() + if result[current_iface]: + result[current_iface][-1]["hostname"] = hostname + else: + result[current_iface].append({"hostname": hostname, "port": ""}) + if m_port and current_iface and result[current_iface]: + result[current_iface][-1]["port"] = m_port.group(1).strip() + return result + + def get_lldp_neighbors_detail(self, interface: str = "") -> dict[str, list[_JsonDict]]: + """Return detailed LLDP neighbour information.""" + result: dict[str, list[_JsonDict]] = {} + raw = self._exec_ssh_command( + "lldpcli show neighbors details 2>/dev/null || true" + ) + current_iface = "" + current_entry: _JsonDict = {} + + def _flush(): + if current_iface and current_entry: + result.setdefault(current_iface, []).append(current_entry.copy()) + + for line in raw.splitlines(): + m_iface = re.match(r"^\s*Interface:\s+(\S+?),?\s", line) + if m_iface: + _flush() + current_iface = m_iface.group(1) + if interface and current_iface != interface: + current_iface = "" + current_entry = { + "parent_interface": "", + "remote_chassis_id": "", + "remote_system_name": "", + "remote_port": "", + "remote_port_description": "", + "remote_system_description": "", + "remote_system_capab": [], + "remote_system_enable_capab": [], + } + continue + if not current_iface: + continue + for key, pattern in ( + ("remote_chassis_id", r"ChassisID:\s+\S+\s+(.+)"), + ("remote_system_name", r"SysName:\s+(.+)"), + ("remote_port", r"PortID:\s+\S+\s+(.+)"), + ("remote_port_description", r"PortDescr:\s+(.+)"), + ("remote_system_description", r"SysDescr:\s+(.+)"), + ): + m = re.match(rf"^\s*{pattern}", line) + if m: + current_entry[key] = m.group(1).strip() + + m_cap = re.match(r"^\s*Capability:\s+(\S+),\s+(\w+)", line) + if m_cap: + cap = m_cap.group(1).lower() + enabled = m_cap.group(2).lower() == "on" + current_entry["remote_system_capab"].append(cap) + if enabled: + current_entry["remote_system_enable_capab"].append(cap) + + _flush() + return result + + # ------------------------------------------------------------------ # + # get_ipv6_neighbors_table + # ------------------------------------------------------------------ # + + def get_ipv6_neighbors_table(self) -> list[_JsonDict]: + """Return IPv6 NDP neighbour table.""" + raw = self._exec_ssh_command("ip -6 neigh show 2>/dev/null || true") + result = [] + for line in raw.splitlines(): + parts = line.split() + # Format: dev lladdr + if len(parts) < 5: + continue + ip6 = parts[0] + iface = parts[2] if len(parts) > 2 else "" + mac = "" + state = "" + for i, tok in enumerate(parts): + if tok == "lladdr" and i + 1 < len(parts): + mac = parts[i + 1] + if tok in ("REACHABLE", "STALE", "DELAY", "PROBE", "FAILED", "NOARP", "PERMANENT"): + state = tok + if not mac or mac == "FAILED": + continue + result.append( + { + "interface": iface, + "mac": utils.normalize_mac(mac), + "ip": ip6, + "age": -1.0, + "state": state, + } + ) + return result + + # ------------------------------------------------------------------ # + # ping (via Proxmox node/execute) + # ------------------------------------------------------------------ # + + def ping( + self, + destination: str, + source: str = C.PING_SOURCE, + ttl: int = C.PING_TTL, + timeout: int = C.PING_TIMEOUT, + size: int = C.PING_SIZE, + count: int = C.PING_COUNT, + vrf: str = C.PING_VRF, + source_interface: str = C.PING_SOURCE_INTERFACE, + ) -> _JsonDict: + """Execute ping on the Proxmox node and return results.""" + cmd_parts = [ + f"ping -c {count}", + f"-W {timeout}", + f"-s {size}", + f"-t {ttl}", + ] + if source: + cmd_parts.append(f"-I {source}") + elif source_interface: + cmd_parts.append(f"-I {source_interface}") + cmd_parts.append(destination) + cmd = " ".join(cmd_parts) + + raw = self._exec_ssh_command(f"{cmd} 2>&1 || true") + if not raw: + return {"error": "Ping command not available via exec endpoint"} + + # Detect common failure strings before parsing statistics + _error_patterns = ( + "Name or service not known", + "Network is unreachable", + "connect: No route to host", + "unknown host", + ) + for _pat in _error_patterns: + if _pat.lower() in raw.lower(): + return {"error": raw.strip()} + + # Parse statistics line: "5 packets transmitted, 5 received, 0% packet loss" + m_stat = re.search( + r"(\d+) packets transmitted,\s+(\d+) received.*?([\d.]+)% packet loss", + raw, + ) + if not m_stat: + return {"error": raw.strip()} + + sent = int(m_stat.group(1)) + received = int(m_stat.group(2)) + loss = sent - received + + # RTT line: "rtt min/avg/max/mdev = 0.123/0.456/0.789/0.100 ms" + m_rtt = re.search( + r"rtt min/avg/max/mdev = ([\d.]+)/([\d.]+)/([\d.]+)/([\d.]+)", + raw, + ) + rtt_min = float(m_rtt.group(1)) if m_rtt else 0.0 + rtt_avg = float(m_rtt.group(2)) if m_rtt else 0.0 + rtt_max = float(m_rtt.group(3)) if m_rtt else 0.0 + rtt_std = float(m_rtt.group(4)) if m_rtt else 0.0 + + # Individual probe lines + probes = [] + for m_probe in re.finditer( + r"icmp_seq=\d+.*?time=([\d.]+) ms.*?from ([\d.a-fA-F:]+)", raw + ): + probes.append( + {"ip_address": m_probe.group(2), "rtt": float(m_probe.group(1))} + ) + + return { + "success": { + "probes_sent": sent, + "packet_loss": loss, + "rtt_min": rtt_min, + "rtt_max": rtt_max, + "rtt_avg": rtt_avg, + "rtt_stddev": rtt_std, + "results": probes, + } + } + + # ------------------------------------------------------------------ # + # traceroute + # ------------------------------------------------------------------ # + + def traceroute( + self, + destination: str, + source: str = "", + ttl: int = 255, + timeout: int = 2, + vrf: str = "", + ) -> _JsonDict: + """Execute traceroute on the Proxmox node and return results.""" + cmd_parts = [f"traceroute -m {ttl}", f"-w {timeout}", "-n"] + if source: + cmd_parts.append(f"-s {source}") + cmd_parts.append(destination) + raw = self._exec_ssh_command(" ".join(cmd_parts) + " 2>&1 || true") + + if not raw: + return {"error": "traceroute not available via exec endpoint"} + + hops: _JsonDict = {} + for line in raw.splitlines(): + m = re.match( + r"^\s*(\d+)\s+([\d.a-fA-F:]+|\*)\s+([\d.]+|[\d.]+\s+ms|\*)", + line, + ) + if not m: + continue + hop_id = int(m.group(1)) + ip_addr = m.group(2) + if ip_addr == "*": + continue + # Parse RTT probes: each hop can have up to 3 + rtts = re.findall(r"([\d.]+)\s+ms", line) + probes_dict = {} + for idx, rtt in enumerate(rtts, start=1): + probes_dict[idx] = { + "rtt": float(rtt), + "ip_address": ip_addr, + "host_name": ip_addr, + } + if probes_dict: + hops[hop_id] = {"probes": probes_dict} + + if not hops: + return {"error": raw.strip()} + return {"success": hops} + + # ------------------------------------------------------------------ # + # cli + # ------------------------------------------------------------------ # + + def cli(self, commands: list[str], encoding: str = "text") -> dict[str, str]: + """Execute arbitrary commands on the node via the exec endpoint.""" + output = {} + for cmd in commands: + output[cmd] = self._exec_ssh_command(cmd) + return output + + # ------------------------------------------------------------------ # + # Unsupported / not-applicable methods + # ------------------------------------------------------------------ # + + def get_bgp_config(self, group: str = "", neighbor: str = "") -> _JsonDict: + raise NotImplementedError("BGP configuration is not managed via Proxmox API") + + def get_bgp_neighbors(self) -> _JsonDict: + raise NotImplementedError("BGP is not managed via Proxmox API") + + def get_bgp_neighbors_detail(self, neighbor_address: str = "") -> _JsonDict: + raise NotImplementedError("BGP is not managed via Proxmox API") + + def get_route_to( + self, destination: str = "", protocol: str = "", longer: bool = False + ) -> _JsonDict: + """Return routing table entries for the given destination.""" + cmd = f"ip route show {destination} 2>/dev/null || true" + raw = self._exec_ssh_command(cmd) + routes: _JsonDict = {} + for line in raw.splitlines(): + line = line.strip() + if not line: + continue + parts = line.split() + if not parts: + continue + prefix = parts[0] + next_hop = "" + out_iface = "" + proto = "static" + for i, tok in enumerate(parts): + if tok == "via" and i + 1 < len(parts): + next_hop = parts[i + 1] + if tok == "dev" and i + 1 < len(parts): + out_iface = parts[i + 1] + if tok == "proto" and i + 1 < len(parts): + proto = parts[i + 1] + + if protocol and protocol.lower() not in proto.lower(): + continue + + routes.setdefault(prefix, []).append( + { + "protocol": proto, + "current_active": True, + "last_active": True, + "age": -1, + "next_hop": next_hop, + "outgoing_interface": out_iface, + "selected_next_hop": True, + "preference": 1, + "inactive_reason": "", + "routing_table": "default", + "protocol_attributes": {}, + } + ) + return routes + + def get_optics(self) -> _JsonDict: + raise NotImplementedError("Optics not available via Proxmox API") + + def get_probes_config(self) -> _JsonDict: + raise NotImplementedError + + def get_probes_results(self) -> _JsonDict: + raise NotImplementedError + + def get_firewall_policies(self) -> _JsonDict: + raise NotImplementedError("Use the Proxmox firewall API directly") + + # ------------------------------------------------------------------ # + # VMs and Containers + # ------------------------------------------------------------------ # + + def get_vm_interfaces( + self, vmid: int, vm_type: str + ) -> tuple[dict[str, _JsonDict], bool, bool]: + """Return network interfaces for a single VM or LXC container. + + Returns a 3-tuple ``(interfaces, agent_running, agent_enabled)``: + + * ``interfaces`` – dict keyed by interface name, each with + NAPALM-compatible fields plus ``ipv4``, ``bridge``, and ``tag``. + * ``agent_running`` – True if the QEMU Guest Agent responded during + this call (always False for LXC). + * ``agent_enabled`` – True if the QEMU Guest Agent is enabled in the + VM's Proxmox config (always False for LXC). + + LXC : uses ``/nodes/{node}/lxc/{vmid}/interfaces`` + LXC config + QEMU : tries QEMU guest agent first, falls back to VM config parsing. + In both paths the VM config is fetched to derive bridge/tag. + """ + _NET_MODELS = {"virtio", "e1000", "e1000e", "vmxnet3", "rtl8139", "ne2k_pci"} + + def _parse_net_entry(val_str: str) -> tuple[str, str, int | None]: + """Parse a Proxmox net config value → (mac_upper, bridge, tag|None).""" + mac = bridge = "" + tag: int | None = None + for part in str(val_str).split(","): + if "=" not in part: + continue + k, v = part.split("=", 1) + k = k.strip().lower() + if k in _NET_MODELS: + mac = v.strip() + elif k == "bridge": + bridge = v.strip() + elif k == "tag": + try: + tag = int(v.strip()) + except ValueError: + pass + return mac.upper() if mac else "", bridge, tag + + interfaces: dict[str, _JsonDict] = {} + + if vm_type == "container": + # Build iface_name → (bridge, tag) from LXC config + # LXC net entries look like: net0=name=eth0,bridge=vmbr40,tag=40,... + lxc_net_map: dict[str, tuple[str, int | None]] = {} + try: + config = self._node_api().lxc(vmid).config.get() or {} + net_re = re.compile(r"^net(\d+)$") + for key, val in config.items(): + if not net_re.match(key): + continue + iface_name = "" + bridge = "" + tag: int | None = None + for part in str(val).split(","): + if "=" not in part: + continue + k, v = part.split("=", 1) + k = k.strip().lower() + if k == "name": + iface_name = v.strip() + elif k == "bridge": + bridge = v.strip() + elif k == "tag": + try: + tag = int(v.strip()) + except ValueError: + pass + if iface_name: + lxc_net_map[iface_name] = (bridge, tag) + except Exception as exc: + logger.debug("get_vm_interfaces: LXC %s config failed: %s", vmid, exc) + + try: + for iface in (self._node_api().lxc(vmid).interfaces.get() or []): + name = iface.get("name", "") + if not name or name == "lo": + continue + mac = iface.get("hwaddr", "") + ipv4 = "" + inet = iface.get("inet", "") + if inet: + ipv4 = inet.split("/")[0] + bridge, tag = lxc_net_map.get(name, ("", None)) + interfaces[name] = { + "is_up": True, + "is_enabled": True, + "description": bridge, + "mac_address": mac.upper() if mac else "", + "speed": -1, + "mtu": 1500, + "last_flapped": -1.0, + "ipv4": ipv4, + "bridge": bridge, + "tag": tag, + } + except Exception as exc: + logger.debug("get_vm_interfaces: LXC %s ifaces failed: %s", vmid, exc) + + # LXC containers do not use QEMU Guest Agent + return interfaces, False, False + + else: + # QEMU: pre-fetch VM config to build MAC → (bridge, tag) map and + # to check whether the QEMU Guest Agent is enabled. + mac_to_net: dict[str, tuple[str, int | None]] = {} # mac_upper → (bridge, tag) + net_idx_map: dict[str, tuple[str, str, int | None]] = {} # "netN" → (mac, bridge, tag) + agent_enabled = False + try: + config = self._node_api().qemu(vmid).config.get() or {} + # Proxmox stores the agent setting as agent=1, agent=0, or + # agent=enabled=1[,fstrim_cloned_disks=1,...] + raw_agent = str(config.get("agent", "0")) + # Treat any truthy value ("1", "enabled=1", ...) as enabled + agent_enabled = bool( + raw_agent.strip() in ("1", "true") + or raw_agent.startswith("enabled=1") + or raw_agent.startswith("1,") + ) + net_re = re.compile(r"^net(\d+)$") + for key, val in config.items(): + m = net_re.match(key) + if not m: + continue + mac, bridge, tag = _parse_net_entry(val) + iface_key = f"net{m.group(1)}" + net_idx_map[iface_key] = (mac, bridge, tag) + if mac: + mac_to_net[mac] = (bridge, tag) + except Exception as exc: + logger.debug("get_vm_interfaces: QEMU %s config fetch failed: %s", vmid, exc) + + # Try guest agent first + agent_ok = False + try: + agent_result = self._node_api().qemu(vmid).agent("network-get-interfaces").get() + for iface in (agent_result or {}).get("result", []): + name = iface.get("name", "") + if not name or name == "lo": + continue + mac = (iface.get("hardware-address", "") or "").upper() + ipv4 = "" + for addr in iface.get("ip-addresses", []): + if addr.get("ip-address-type") == "ipv4": + ipv4 = addr.get("ip-address", "") + break + bridge, tag = mac_to_net.get(mac, ("", None)) + interfaces[name] = { + "is_up": True, + "is_enabled": True, + "description": bridge, + "mac_address": mac, + "speed": -1, + "mtu": 1500, + "last_flapped": -1.0, + "ipv4": ipv4, + "bridge": bridge, + "tag": tag, + } + agent_ok = bool(interfaces) + except Exception: + pass + + if not agent_ok: + # Fall back to config-only (gives MAC + bridge + tag, no IP) + for iface_key, (mac, bridge, tag) in net_idx_map.items(): + interfaces[iface_key] = { + "is_up": False, + "is_enabled": True, + "description": bridge, + "mac_address": mac, + "speed": -1, + "mtu": 1500, + "last_flapped": -1.0, + "ipv4": "", + "bridge": bridge, + "tag": tag, + } + + return interfaces, agent_ok, agent_enabled + + def get_vms(self) -> list[_JsonDict]: + """Return all VMs (QEMU) and containers (LXC) on this node. + + Each entry contains: + * vmid (int) - Proxmox VM/container ID + * name (str) - display name + * type (str) - ``"vm"`` or ``"container"`` + * status (str) - ``"running"``, ``"stopped"``, etc. + * vcpus (int) - allocated vCPUs + * memory (int) - configured RAM in megabytes + * cpu_usage (float) - current CPU utilisation 0.0–1.0 (from last stats cycle) + * memory_usage (int)- current RSS in megabytes + * uptime (int) - uptime in seconds (0 if stopped) + * node (str) - cluster node name + * interfaces (dict) - network interfaces (NAPALM format + ipv4 field) + * ipv4 (str) - primary IPv4 address (empty string if unknown) + """ + result: list[_JsonDict] = [] + + # QEMU VMs + try: + for vm in (self._node_api().qemu.get() or []): + vmid = int(vm.get("vmid", 0)) + name = vm.get("name", f"vm-{vmid}") + status = vm.get("status", "unknown") + cpu_usage = float(vm.get("cpu", 0.0) or 0.0) + uptime = int(vm.get("uptime", 0) or 0) + + # mem/maxmem are in bytes + maxmem_bytes = int(vm.get("maxmem", 0) or 0) + mem_bytes = int(vm.get("mem", 0) or 0) + memory_mb = maxmem_bytes // (1024 * 1024) + memory_usage_mb = mem_bytes // (1024 * 1024) + + # vcpus can be in "cpus" key for running VMs + vcpus = int(vm.get("cpus", vm.get("vcpus", 0)) or 0) + + interfaces, agent_running, agent_enabled = self.get_vm_interfaces(vmid, "vm") + ipv4 = next( + (iface["ipv4"] for iface in interfaces.values() if iface.get("ipv4")), + "", + ) + + result.append({ + "vmid": vmid, + "name": name, + "type": "vm", + "status": status, + "vcpus": vcpus, + "memory": memory_mb, + "cpu_usage": round(cpu_usage, 4), + "memory_usage": memory_usage_mb, + "uptime": uptime, + "node": self._node_name, + "interfaces": interfaces, + "ipv4": ipv4, + "agent_enabled": agent_enabled, + "agent_running": agent_running, + }) + except Exception as exc: + logger.warning("get_vms: failed to list QEMU VMs: %s", exc) + + # LXC containers + try: + for ct in (self._node_api().lxc.get() or []): + vmid = int(ct.get("vmid", 0)) + name = ct.get("name", f"ct-{vmid}") + status = ct.get("status", "unknown") + cpu_usage = float(ct.get("cpu", 0.0) or 0.0) + uptime = int(ct.get("uptime", 0) or 0) + + maxmem_bytes = int(ct.get("maxmem", 0) or 0) + mem_bytes = int(ct.get("mem", 0) or 0) + memory_mb = maxmem_bytes // (1024 * 1024) + memory_usage_mb = mem_bytes // (1024 * 1024) + + vcpus = int(ct.get("cpus", 0) or 0) + + interfaces, agent_running, agent_enabled = self.get_vm_interfaces(vmid, "container") + ipv4 = next( + (iface["ipv4"] for iface in interfaces.values() if iface.get("ipv4")), + "", + ) + + result.append({ + "vmid": vmid, + "name": name, + "type": "container", + "status": status, + "vcpus": vcpus, + "memory": memory_mb, + "cpu_usage": round(cpu_usage, 4), + "memory_usage": memory_usage_mb, + "uptime": uptime, + "node": self._node_name, + "interfaces": interfaces, + "ipv4": ipv4, + "agent_enabled": agent_enabled, + "agent_running": agent_running, + }) + except Exception as exc: + logger.warning("get_vms: failed to list LXC containers: %s", exc) + + return sorted(result, key=lambda x: x["vmid"]) + + # ------------------------------------------------------------------ # + # VM power management + # ------------------------------------------------------------------ # + + _POWER_ACTIONS_VM = {'start', 'stop', 'shutdown', 'reboot', 'reset'} + _POWER_ACTIONS_CT = {'start', 'stop', 'shutdown', 'reboot'} + + def power_vm(self, vmid: int, vm_type: str, action: str) -> dict: + """Send a power action to a VM or container on this node. + + Supported actions for VMs: start, stop, shutdown, reboot, reset + Supported actions for containers: start, stop, shutdown, reboot + """ + allowed = self._POWER_ACTIONS_VM if vm_type == 'vm' else self._POWER_ACTIONS_CT + if action not in allowed: + return {"success": False, "error": f"Action '{action}' not supported for {vm_type} (allowed: {sorted(allowed)})"} + try: + vm_api = self._node_api().qemu(vmid) if vm_type == 'vm' else self._node_api().lxc(vmid) + task_id = getattr(vm_api.status, action).post() + return {"success": True, "task_id": task_id or ""} + except Exception as exc: + return {"success": False, "error": str(exc)} + + # ------------------------------------------------------------------ # + # Packages (Debian APT) + # ------------------------------------------------------------------ # + + def get_packages(self) -> list[_JsonDict]: + """Return installed Debian packages with available-update info. + + Installed list comes from ``dpkg-query`` via SSH (the Proxmox API + ``/apt/installed`` endpoint is not implemented on PVE 8.x). + Available updates come from the Proxmox API ``/apt/update``. + """ + # Available updates from Proxmox API (keyed by package name) + upgradable: dict[str, str] = {} + try: + for upd in self._api.nodes(self._node_name).apt.update.get(): + pkg = upd.get("Package", "") + if pkg: + upgradable[pkg] = upd.get("Version", "") + except Exception: + pass + + # Installed packages via SSH dpkg-query + raw = self._exec_ssh_command( + "dpkg-query -W -f='${Package}\\t${Version}\\t${db:Status-Status}\\t${Installed-Size}\\n'" + " 2>/dev/null" + ) + result: list[_JsonDict] = [] + for line in raw.splitlines(): + parts = line.strip().split("\t") + if len(parts) < 2: + continue + name = parts[0] + version = parts[1] if len(parts) > 1 else "" + status = parts[2] if len(parts) > 2 else "installed" + size_kb = parts[3] if len(parts) > 3 else "0" + if not name or status != "installed": + continue + size_bytes = int(size_kb) * 1024 if size_kb.isdigit() else 0 + result.append({ + "name": name, + "version": version, + "installed": True, + "description": "", + "size": size_bytes, + "source": "pve", + "upgrade_version": upgradable.get(name, ""), + }) + return result + + # ------------------------------------------------------------------ # + # Device warnings + # ------------------------------------------------------------------ # + + def get_device_warnings(self) -> list[_JsonDict]: + """Return warnings for the Proxmox node. + + Currently detects: + - Available package updates (via Proxmox APT API) + - Missing / invalid subscription + """ + warnings: list[_JsonDict] = [] + + # 1. Available package updates + try: + updates = self._api.nodes(self._node_name).apt.update.get() + if updates: + warnings.append({ + "code": "updates_available", + "severity": "info", + "action": None, + "meta": { + "count": len(updates), + "packages": [u.get("Package", "") for u in updates[:10]], + }, + }) + except Exception: + pass + + # 2. Subscription status + try: + sub = self._get_node_subscription() + status = sub.get("status", "") + if status in ("NotFound", "Invalid", "Expired"): + warnings.append({ + "code": "no_subscription", + "severity": "warning", + "action": None, + "meta": {"status": status}, + }) + except Exception: + pass + + return warnings + + # ------------------------------------------------------------------ # + # Services (systemd) + # ------------------------------------------------------------------ # + + def get_services(self) -> list[_JsonDict]: + """Return systemd services with running and enabled state. + + Uses two ``systemctl`` invocations combined in a single SSH command: + - ``list-unit-files`` for the static enabled/disabled state + - ``list-units`` for the live running state + """ + raw = self._exec_ssh_command( + "{ systemctl list-unit-files --type=service --no-pager --no-legend --full 2>/dev/null;" + " echo '---UNITS---';" + " systemctl list-units --type=service --all --no-pager --no-legend --full 2>/dev/null;" + " } || true" + ) + + # Parse enabled state from list-unit-files + enabled_map: dict[str, bool] = {} + section = "files" + for line in raw.splitlines(): + if line.strip() == "---UNITS---": + section = "units" + continue + parts = line.strip().split(None, 1) + if len(parts) < 1: + continue + unit = parts[0].lstrip("●").strip() + if not unit.endswith(".service"): + continue + name = unit[: -len(".service")] + if section == "files": + state = parts[1].strip() if len(parts) > 1 else "" + enabled_map[name] = state in ("enabled", "enabled-runtime", "static") + + # Parse running state from list-units + running_map: dict[str, bool] = {} + section = "files" + for line in raw.splitlines(): + if line.strip() == "---UNITS---": + section = "units" + continue + if section != "units": + continue + parts = line.strip().lstrip("●").strip().split(None, 4) + if len(parts) < 4: + continue + unit = parts[0] + if not unit.endswith(".service"): + continue + name = unit[: -len(".service")] + sub_state = parts[3] + running_map[name] = sub_state == "running" + + all_names = sorted(set(enabled_map) | set(running_map)) + return [ + { + "name": name, + "running": running_map.get(name, False), + "enabled": enabled_map.get(name, False), + "pid": 0, + } + for name in all_names + ] + + def manage_service(self, name: str, action: str) -> _JsonDict: + """Start / stop / restart / enable / disable a systemd service.""" + if not re.match(r'^[a-zA-Z0-9_\-\.@]+$', name): + raise ValueError(f"Invalid service name: {name!r}") + if action not in ('start', 'stop', 'restart', 'enable', 'disable'): + raise ValueError(f"Invalid action: {action!r}") + output = self._exec_ssh_command(f"systemctl {action} {name}.service 2>&1 || true") + return {"success": True, "output": output} + + # ------------------------------------------------------------------ # + # Available updates + # ------------------------------------------------------------------ # + + def get_available_updates(self) -> list[_JsonDict]: + """Return list of upgradable packages from the Proxmox APT API.""" + updates: list[_JsonDict] = [] + try: + for upd in self._api.nodes(self._node_name).apt.update.get(): + pkg = upd.get("Package", "") + if not pkg: + continue + updates.append({ + "name": pkg, + "current_version": upd.get("OldVersion", ""), + "new_version": upd.get("Version", ""), + }) + except Exception: + pass + return sorted(updates, key=lambda u: u["name"]) + + def apply_updates(self, packages: list[str]) -> _JsonDict: + """Upgrade the given packages via ``apt-get install`` over SSH.""" + for pkg in packages: + if not re.match(r'^[a-zA-Z0-9_\-\+\.]+$', pkg): + raise ValueError(f"Invalid package name: {pkg!r}") + pkg_args = " ".join(packages) + output = self._exec_ssh_command( + f"DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y {pkg_args} 2>&1 || true" + ) + return {"success": True, "output": output} + + # ------------------------------------------------------------------ # + # VLAN provisioning via SDN VNets (OVS-backed nodes only) + # ------------------------------------------------------------------ # + + # Proxmox-internal / runtime virtual interface name prefixes that should + # never be considered physical switch uplinks. + _VIRTUAL_IFACE_PREFIXES = ( + "fwpr", # Proxmox firewall proxy veth + "fwln", # Proxmox firewall line veth + "tap", # VM tap devices + "veth", # generic veth pairs + "virbr", # libvirt bridges + "docker", # Docker virtual interfaces + "lxcbr", # LXC bridges + ) + + def _is_physical_uplink(self, iface_name: str, network: dict) -> bool: + """Return True if *iface_name* is a physical Ethernet port usable as uplink. + + Rules: + - Must not match any known virtual interface name prefix. + - Must appear in the Proxmox node network config (runtime-only virtual + interfaces such as ``fwpr*`` or ``tap*`` will not be listed there). + - Must have a physical-compatible type: + - ``"eth"`` — regular physical NIC + - ``"OVSPort"`` — physical NIC attached directly to an OVS bridge + - ``""`` — untyped (e.g. OVS bond slave, still physical) + """ + if any(iface_name.startswith(p) for p in self._VIRTUAL_IFACE_PREFIXES): + return False + iface_info = network.get(iface_name) + if iface_info is None: + # Not in Proxmox network config → runtime virtual interface + return False + return iface_info.get("type", "") in ("eth", "OVSPort", "") + + def _find_switch_uplink(self) -> str | None: + """Return the name of the physical interface connected to a switch. + + Detection order: + 1. LLDP detailed: physical port whose neighbour advertises Bridge + capability. + 2. LLDP basic fallback: first physical port with any LLDP neighbour. + """ + network = { + iface["iface"]: iface + for iface in self._get_node_network() + if iface.get("iface") + } + + # Prefer neighbours that announce Bridge capability + try: + for iface_name, neighbour_list in self.get_lldp_neighbors_detail().items(): + if not self._is_physical_uplink(iface_name, network): + continue + for nb in neighbour_list: + caps = nb.get("remote_system_capab", []) + if any("bridge" in str(c).lower() for c in caps): + return iface_name + except Exception: + pass + + # Fallback: first physical port with any LLDP neighbour + try: + for iface_name, neighbour_list in self.get_lldp_neighbors().items(): + if self._is_physical_uplink(iface_name, network) and neighbour_list: + return iface_name + except Exception: + pass + + return None + + def _get_ovs_bridge_for_port(self, port_name: str) -> str | None: + """Return the OVS bridge name that *port_name* belongs to, or ``None``. + + Checks (in order): + 1. Port listed in an OVSBridge's ``ovs_ports``. + 2. Port is a slave of an OVSBond which has an ``ovs_bridge`` reference. + 3. Port itself carries an ``ovs_bridge`` field. + """ + network = self._get_node_network() + by_name: dict[str, _JsonDict] = { + iface["iface"]: iface for iface in network if iface.get("iface") + } + + for iface in network: + if iface.get("type") == "OVSBridge": + ports = (iface.get("ovs_ports") or "").split() + if port_name in ports: + return iface["iface"] + + for iface in network: + if iface.get("type") == "OVSBond": + slaves = (iface.get("slaves") or "").split() + if port_name in slaves: + bridge = iface.get("ovs_bridge", "") + if bridge: + return bridge + + port_info = by_name.get(port_name, {}) + return port_info.get("ovs_bridge") or None + + def _is_cluster_master(self) -> bool: + """Return ``True`` if this node is the Corosync quorum coordinator. + + The coordinator is the online cluster node with the lowest ``nodeid``. + On standalone (non-clustered) nodes this always returns ``True``. + """ + try: + status = self._api.cluster.status.get() or [] + node_entries = [e for e in status if e.get("type") == "node"] + if not node_entries: + return True # Standalone node — no cluster + online_nodes = [n for n in node_entries if n.get("online", 0)] + if not online_nodes: + return True # All nodes offline → assume we're the master + min_id = min(int(n.get("nodeid", 9999)) for n in online_nodes) + for n in online_nodes: + if ( + n.get("name") == self._node_name + and int(n.get("nodeid", 9999)) == min_id + ): + return True + return False + except Exception: + return True # Can't determine → assume standalone, proceed + + def _get_sdn_zone_for_bridge(self, bridge_name: str) -> str | None: + """Return the SDN zone ID whose ``bridge`` field matches *bridge_name*. + + In Proxmox SDN each zone is linked to exactly one OVS bridge via the + ``bridge`` property. We look for that mapping so the VNet is always + created in the correct zone instead of guessing by type order. + + Falls back to the first zone if no bridge match is found. + """ + try: + zones = self._get_sdn_zones() + # Exact match on bridge field + for zone in zones: + if zone.get("bridge") == bridge_name: + return zone.get("zone") + # Fallback: first available zone + if zones: + return zones[0].get("zone") + except Exception: + pass + return None + + def set_vlan(self, vlan_id: int, config) -> None: + """Create (or update) a VLAN via an SDN VNet on this Proxmox node. + + Pre-flight checks (all must pass to proceed): + + 1. Finds the physical uplink port connected to a switch via LLDP. + Physical ports are those with type ``eth``, ``OVSPort``, or ``""`` + in the Proxmox network config (excludes runtime virtuals like + ``fwpr*``, ``tap*``, etc.). + 2. Verifies that uplink is part of an OVS bridge or OVS bond. + 3. Confirms this node is the Corosync quorum master (lowest node-id). + Non-master nodes return silently — the master handles VNet creation. + + The SDN VNet is named ``vlan{vid:04d}`` (e.g. ``vlan0007`` for VID 7). + If the VNet already exists its alias is updated. After creating / + updating the VNet the SDN configuration is reloaded via + ``PUT /cluster/sdn``. + + Args: + vlan_id: VLAN identifier (1-4094). + config: Dict that may contain ``"name"`` for the VLAN alias. + """ + name: str = ( + (config.get("name") or f"VLAN{vlan_id}") if config else f"VLAN{vlan_id}" + ) + vnet_id = f"vlan{vlan_id:04d}" + + # 1. Find switch uplink via LLDP + uplink = self._find_switch_uplink() + if uplink is None: + raise ConnectionException( + f"set_vlan({vlan_id}): no LLDP-detected switch uplink found" + f" on node {self._node_name!r}" + ) + + # 2. Verify uplink is part of an OVS bridge + ovs_bridge = self._get_ovs_bridge_for_port(uplink) + if ovs_bridge is None: + raise ConnectionException( + f"set_vlan({vlan_id}): uplink {uplink!r} is not part of an OVS bridge" + ) + + # 3. Only the Corosync master manages SDN VNets + if not self._is_cluster_master(): + return + + # 4. Find the SDN zone linked to this OVS bridge + zone = self._get_sdn_zone_for_bridge(ovs_bridge) + if not zone: + raise ConnectionException( + f"set_vlan({vlan_id}): no SDN zone found for bridge {ovs_bridge!r}" + ) + + # 5. Create or update the VNet + try: + self._api.cluster.sdn.vnets.post( + vnet=vnet_id, + zone=zone, + tag=vlan_id, + alias=name, + ) + except ResourceException as exc: + err_str = str(exc).lower() + if "already exists" in err_str or "duplicate" in err_str or "500" in err_str: + # VNet already exists — update alias + try: + self._api.cluster.sdn.vnets(vnet_id).put(alias=name) + except Exception: + pass # Best effort + else: + raise ConnectionException( + f"set_vlan({vlan_id}): failed to create VNet {vnet_id!r}: {exc}" + ) from exc + + # 6. Reload SDN config so the VNet becomes active + try: + self._api.cluster.sdn.put() + except Exception: + pass # Best effort — may not be needed on older PVE versions + + def delete_vlan(self, vlan_id: int) -> None: + """Delete the SDN VNet corresponding to *vlan_id*. + + The VNet is identified by the canonical name ``vlan{vid:04d}``. + Only the Corosync quorum master performs the deletion — non-master + nodes return silently. + + After deletion the SDN configuration is reloaded via + ``PUT /cluster/sdn``. + + Args: + vlan_id: VLAN identifier to delete. + """ + if not self._is_cluster_master(): + return # Non-master node — master handles VNet deletion + + vnet_id = f"vlan{vlan_id:04d}" + try: + self._api.cluster.sdn.vnets(vnet_id).delete() + except ResourceException as exc: + err_str = str(exc).lower() + if "does not exist" in err_str or "404" in str(exc): + return # Already gone — treat as success + raise ConnectionException( + f"delete_vlan({vlan_id}): failed to delete VNet {vnet_id!r}: {exc}" + ) from exc + + # Reload SDN config + try: + self._api.cluster.sdn.put() + except Exception: + pass # Best effort diff --git a/napalm_proxmox/utils.py b/napalm_proxmox/utils.py new file mode 100644 index 0000000..69398cf --- /dev/null +++ b/napalm_proxmox/utils.py @@ -0,0 +1,269 @@ +"""Utility helpers for the Proxmox NAPALM driver. + +Covers: + - Classic Linux networking (/etc/network/interfaces) + - Proxmox SDN (zones, vnets, subnets) + - Open vSwitch (OVS bridges, bonds, internal ports) +""" + +from __future__ import annotations + +import re +from ipaddress import IPv4Interface, IPv6Interface, ip_interface +from typing import Any + +from netaddr import EUI, AddrFormatError + + +# --------------------------------------------------------------------------- +# MAC address helpers +# --------------------------------------------------------------------------- + +_MAC_RE = re.compile( + r"(?:[0-9a-fA-F]{2}[:\-]){5}[0-9a-fA-F]{2}" + r"|[0-9a-fA-F]{12}" +) + + +def normalize_mac(mac: str) -> str: + """Return a colon-separated, lower-case MAC address, or '' on failure.""" + if not mac: + return "" + try: + return str(EUI(mac, dialect=_ColonDialect())) + except (AddrFormatError, ValueError): + return mac.lower() + + +class _ColonDialect(EUI): + """EUI dialect that formats as aa:bb:cc:dd:ee:ff.""" + + word_sep = ":" + word_fmt = "%.2x" + + def __new__(cls): # type: ignore[override] + from netaddr import mac_unix_expanded + + return mac_unix_expanded + + +def normalize_mac(mac: str) -> str: # noqa: F811 + """Return lower-case colon-separated MAC or empty string.""" + if not mac: + return "" + try: + eui = EUI(mac) + eui.dialect = type( + "_ColonDialect", + (), + {"word_sep": ":", "word_fmt": "%.2x", "word_size": 8, "num_words": 6}, + ) + return str(eui).lower() + except (AddrFormatError, ValueError): + # Best-effort normalisation + cleaned = mac.replace("-", ":").strip().lower() + return cleaned + + +# --------------------------------------------------------------------------- +# IP / prefix helpers +# --------------------------------------------------------------------------- + +def prefix_length(cidr: str) -> int: + """Return prefix length from a CIDR string such as '192.168.1.0/24' → 24.""" + try: + return int(cidr.split("/")[1]) + except (IndexError, ValueError): + # No slash present — treat as host route + return 128 if ":" in cidr else 32 + + +def parse_cidr(cidr: str) -> tuple[str, int]: + """Return (ip_str, prefix_len) from a CIDR address.""" + try: + iface = ip_interface(cidr) + return str(iface.ip), iface.network.prefixlen + except ValueError: + return cidr, -1 + + +def addresses_from_node_network(iface_data: dict[str, Any]) -> dict[str, dict]: + """Build an interfaces_ip sub-dict from a Proxmox node/network entry. + + Proxmox returns separate fields: ``address``, ``netmask``, ``address6``, + ``netmask6``. We convert those to the NAPALM ipv4/ipv6 keyed format. + """ + result: dict[str, dict] = {} + + addr4 = iface_data.get("address", "") + mask4 = iface_data.get("netmask", "") + if addr4 and mask4: + try: + prefix = IPv4Interface(f"{addr4}/{mask4}").network.prefixlen + result.setdefault("ipv4", {})[addr4] = {"prefix_length": prefix} + except ValueError: + pass + + addr6 = iface_data.get("address6", "") + mask6 = iface_data.get("netmask6", "") + if addr6 and mask6: + try: + prefix = IPv6Interface(f"{addr6}/{mask6}").network.prefixlen + result.setdefault("ipv6", {})[addr6] = {"prefix_length": prefix} + except ValueError: + pass + + # Also handle CIDR-style fields (e.g. from SDN) + cidr4 = iface_data.get("cidr", "") + if cidr4 and "/" in cidr4 and "ipv4" not in result: + ip, plen = parse_cidr(cidr4) + if plen >= 0: + result.setdefault("ipv4", {})[ip] = {"prefix_length": plen} + + cidr6 = iface_data.get("cidr6", "") + if cidr6 and "/" in cidr6 and "ipv6" not in result: + ip, plen = parse_cidr(cidr6) + if plen >= 0: + result.setdefault("ipv6", {})[ip] = {"prefix_length": plen} + + return result + + +# --------------------------------------------------------------------------- +# Interface type helpers (classic Linux vs OVS vs SDN bridge) +# --------------------------------------------------------------------------- + +OVS_TYPES = {"OVSBridge", "OVSBond", "OVSIntPort", "OVSPort"} + +SDN_BRIDGE_RE = re.compile(r"^vnet\d+$|^vmbr\d+$|^sdnbr\d+$") + + +def is_ovs_interface(iface_data: dict[str, Any]) -> bool: + return iface_data.get("type", "") in OVS_TYPES + + +def is_sdn_vnet(name: str) -> bool: + return bool(re.fullmatch(r"vnet\d+", name)) + + +def speed_mbps(iface_data: dict[str, Any]) -> float: + """Extract link speed (Mbit/s) from Proxmox interface data.""" + raw = iface_data.get("speed", -1) + if raw is None or raw == "": + return -1.0 + try: + val = float(raw) + # Proxmox reports speed in Mbit/s already for physical interfaces + return val if val > 0 else -1.0 + except (TypeError, ValueError): + return -1.0 + + +# --------------------------------------------------------------------------- +# ARP / neighbour table helpers +# --------------------------------------------------------------------------- + +def parse_arp_output(raw: str) -> list[dict[str, Any]]: + """Parse Linux ``arp -n`` text output into NAPALM ARP dicts. + + Expected format:: + + Address HWtype HWaddress Flags Iface + 192.168.1.1 ether aa:bb:cc:dd:ee:ff C eth0 + """ + entries = [] + for line in raw.splitlines(): + line = line.strip() + if not line or line.startswith("Address") or line.startswith("?"): + continue + parts = line.split() + if len(parts) < 5: + continue + ip_addr, _, mac, _, iface = parts[0], parts[1], parts[2], parts[3], parts[4] + if mac in ("(incomplete)", ""): + continue + entries.append( + { + "interface": iface, + "mac": normalize_mac(mac), + "ip": ip_addr, + "age": -1.0, + } + ) + return entries + + +# --------------------------------------------------------------------------- +# SDN helpers +# --------------------------------------------------------------------------- + +ZONE_TYPE_MAP = { + "simple": "L3VRF", + "vlan": "L2VPN", + "qinq": "L2VPN", + "vxlan": "L2VPN", + "evpn": "L3VRF", +} + + +def sdn_zone_to_network_instance(zone: dict[str, Any]) -> dict[str, Any]: + """Convert a Proxmox SDN zone entry to a NAPALM network instance dict.""" + zone_id = zone.get("zone", zone.get("name", "unknown")) + zone_type = zone.get("type", "simple") + return { + "name": zone_id, + "type": ZONE_TYPE_MAP.get(zone_type, "DEFAULT_INSTANCE"), + "state": { + "route_distinguisher": zone.get("vrf-vxlan", None), + }, + "interfaces": {"interface": {}}, + } + + +# --------------------------------------------------------------------------- +# OVS helpers +# --------------------------------------------------------------------------- + +def parse_ovs_vsctl_show(raw: str) -> dict[str, Any]: + """Very lightweight parser for ``ovs-vsctl show`` output. + + Returns a mapping of bridge_name → {ports: [port_name, …]}. + """ + bridges: dict[str, Any] = {} + current_bridge: str | None = None + for line in raw.splitlines(): + stripped = line.strip() + m_bridge = re.match(r'^Bridge\s+"?([^\s"]+)"?', stripped) + if m_bridge: + current_bridge = m_bridge.group(1) + bridges[current_bridge] = {"ports": []} + continue + m_port = re.match(r'^Port\s+"?([^\s"]+)"?', stripped) + if m_port and current_bridge: + port_name = m_port.group(1) + if port_name != current_bridge: # skip the bridge self-port + bridges[current_bridge]["ports"].append(port_name) + return bridges + + +def parse_ovs_ofctl_dump_flows(raw: str) -> list[dict[str, Any]]: + """Parse ``ovs-ofctl dump-flows `` into a list of flow dicts. + + Used to populate get_mac_address_table for OVS bridges. + """ + flows = [] + for line in raw.splitlines(): + line = line.strip() + if not line or line.startswith("NXST_FLOW") or line.startswith("OFPST_FLOW"): + continue + # Extract dl_src / dl_dst / in_port / output + mac_m = re.search(r"dl_src=([\da-f:]+)", line) + port_m = re.search(r"in_port=(\d+)", line) + if mac_m: + flows.append( + { + "mac": normalize_mac(mac_m.group(1)), + "in_port": port_m.group(1) if port_m else "", + } + ) + return flows diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..d048846 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,65 @@ +[build-system] +requires = ["setuptools>=61", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "napalm-proxmox" +version = "0.1.0" +description = "NAPALM driver for Proxmox VE (classic Linux networking, SDN, OVS)" +readme = "README.md" +license = { text = "Apache-2.0" } +authors = [{ name = "Christian Manivong" }] +keywords = ["napalm", "proxmox", "network", "sdn", "ovs", "automation"] +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "Intended Audience :: System Administrators", + "License :: OSI Approved :: Apache Software License", + "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", +] +requires-python = ">=3.9" +dependencies = [ + "napalm>=5.0.0", + "napalm_device_types>=0.1.0", + "proxmoxer>=2.0.0", + "netaddr>=0.9.0", + "requests>=2.31.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=7.4", + "pytest-cov>=4.1", + "pytest-mock>=3.12", + "responses>=0.24", + "ruff>=0.1", + "mypy>=1.7", +] + +[project.entry-points."napalm.drivers"] +proxmox = "napalm_proxmox:ProxmoxDriver" + +[tool.setuptools.packages.find] +where = ["."] +include = ["napalm_proxmox*"] + +[tool.ruff] +line-length = 100 +target-version = "py39" + +[tool.ruff.lint] +select = ["E", "F", "W", "I", "UP"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "--cov=napalm_proxmox --cov-report=term-missing" + +[tool.mypy] +python_version = "3.9" +warn_return_any = true +warn_unused_configs = true diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..ae78246 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +# tests/__init__.py diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..d2729e9 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,217 @@ +"""Shared fixtures and mock helpers for napalm-proxmox tests.""" + +from __future__ import annotations + +import pytest +from unittest.mock import MagicMock, patch + +from napalm_proxmox.driver import ProxmoxDriver + + +# --------------------------------------------------------------------------- +# Sample API payloads +# --------------------------------------------------------------------------- + +NODE_STATUS = { + "uptime": 123456, + "model": "PowerEdge R640", + "cpu": 0.12, + "memory": {"total": 137438953472, "used": 68719476736}, +} + +VERSION_INFO = {"version": "8.2.4", "release": "8", "repoid": "abc123"} + +NODE_NETWORK = [ + { + "iface": "eth0", + "type": "eth", + "active": 1, + "autostart": 1, + "hwaddr": "aa:bb:cc:dd:ee:01", + "speed": 1000, + "mtu": 1500, + "address": "10.0.0.1", + "netmask": "255.255.255.0", + "comments": "Management", + }, + { + "iface": "eth1", + "type": "eth", + "active": 1, + "autostart": 1, + "hwaddr": "aa:bb:cc:dd:ee:02", + "speed": 10000, + "mtu": 9000, + "comments": "", + }, + { + "iface": "vmbr0", + "type": "bridge", + "active": 1, + "autostart": 1, + "hwaddr": "aa:bb:cc:dd:ee:03", + "speed": None, + "mtu": 1500, + "address": "192.168.1.1", + "netmask": "255.255.255.0", + "comments": "Linux bridge", + }, + { + "iface": "vmbr1", + "type": "OVSBridge", + "active": 0, + "autostart": 1, + "hwaddr": "", + "speed": None, + "mtu": 1500, + "comments": "OVS bridge", + }, +] + +SDN_ZONES = [ + {"zone": "zone-vlan", "type": "vlan", "nodes": "pve1"}, + {"zone": "zone-vxlan", "type": "vxlan", "nodes": "pve1,pve2", "vrf-vxlan": 10000}, + {"zone": "zone-evpn", "type": "evpn", "nodes": "pve1,pve2", "vrf-vxlan": 10001}, +] + +SDN_VNETS = [ + {"vnet": "vnet0", "zone": "zone-vlan", "tag": 10}, + {"vnet": "vnet1", "zone": "zone-vlan", "tag": 20}, + {"vnet": "vnet2", "zone": "zone-vxlan", "tag": 100000}, +] + +SDN_SUBNETS_VNET0 = [ + {"cidr": "10.10.0.0/24", "gateway": "10.10.0.1", "vnet": "vnet0"}, +] + +SDN_SUBNETS_VNET1 = [ + {"cidr": "10.20.0.0/24", "gateway": "10.20.0.1", "vnet": "vnet1"}, + {"cidr": "2001:db8:20::/64", "gateway": "2001:db8:20::1", "vnet": "vnet1"}, +] + +SDN_SUBNETS_VNET2: list = [] + +DNS_INFO = {"search": "pve1.example.com", "dns1": "8.8.8.8"} + +NTP_INFO = {"server": "pool.ntp.org,time.cloudflare.com"} + +NODE_NETSTAT = [ + { + "dev": "eth0", + "rx_bytes": 1000000, + "tx_bytes": 500000, + "rx_packets": 1000, + "tx_packets": 900, + "rx_errs": 0, + "tx_errs": 0, + "rx_drop": 0, + "tx_drop": 1, + } +] + +NODES_LIST = [{"node": "pve1", "status": "online"}] + +PVE_USERS = [ + {"userid": "root@pam", "groups": ""}, + {"userid": "napalm@pam", "groups": ""}, +] + + +# --------------------------------------------------------------------------- +# Driver fixture +# --------------------------------------------------------------------------- + +def _build_mock_api( + nodes=None, + node_status=None, + version=None, + network=None, + sdn_zones=None, + sdn_vnets=None, + sdn_subnets=None, + dns=None, + ntp=None, + netstat=None, + pve_users=None, + exec_return="", + sensors=None, +): + """Build a MagicMock ProxmoxAPI with pre-configured return values.""" + api = MagicMock() + + # Nodes list + api.nodes.get.return_value = nodes or NODES_LIST + + # Version + api.version.get.return_value = version or VERSION_INFO + + # Node sub-resource + node_res = MagicMock() + api.nodes.return_value = node_res + + node_res.status.get.return_value = node_status or NODE_STATUS + node_res.network.get.return_value = network if network is not None else NODE_NETWORK + node_res.dns.get.return_value = dns or DNS_INFO + node_res.ntp.get.return_value = ntp or NTP_INFO + node_res.netstat.get.return_value = netstat or NODE_NETSTAT + node_res.execute.post.return_value = {"data": exec_return} + node_res.hardware.sensors.get.return_value = sensors or [] + node_res.time.get.return_value = {} + node_res.subscription.get.return_value = {} + + # SDN + cluster = MagicMock() + api.cluster = cluster + cluster.sdn.zones.get.return_value = sdn_zones or SDN_ZONES + cluster.sdn.vnets.get.return_value = sdn_vnets or SDN_VNETS + + # SDN subnets per vnet + subnets_map = sdn_subnets or { + "vnet0": SDN_SUBNETS_VNET0, + "vnet1": SDN_SUBNETS_VNET1, + "vnet2": SDN_SUBNETS_VNET2, + } + + def _vnet_resource(vnet_id): + vnet_res = MagicMock() + vnet_res.subnets.get.return_value = subnets_map.get(vnet_id, []) + return vnet_res + + cluster.sdn.vnets.side_effect = _vnet_resource + + # Access / users + api.access.users.get.return_value = pve_users or PVE_USERS + api.access.users.return_value.get.return_value = {} + + return api + + +@pytest.fixture +def driver(): + """Return an open ProxmoxDriver with a mocked API.""" + drv = ProxmoxDriver( + hostname="pve1.example.com", + username="root", + password="secret", + optional_args={"verify_ssl": False, "realm": "pam"}, + ) + mock_api = _build_mock_api() + with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api): + drv.open() + return drv + + +@pytest.fixture +def driver_with_exec(request): + """Return a driver whose exec endpoint returns a parametrised string.""" + exec_output = getattr(request, "param", "") + drv = ProxmoxDriver( + hostname="pve1.example.com", + username="root", + password="secret", + optional_args={"verify_ssl": False}, + ) + mock_api = _build_mock_api(exec_return=exec_output) + with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api): + drv.open() + return drv diff --git a/tests/test_connection.py b/tests/test_connection.py new file mode 100644 index 0000000..a36844e --- /dev/null +++ b/tests/test_connection.py @@ -0,0 +1,75 @@ +"""Tests for ProxmoxDriver connection management.""" + +from __future__ import annotations + +import pytest +from unittest.mock import MagicMock, patch + +from napalm.base.exceptions import ConnectionException +from napalm_proxmox.driver import ProxmoxDriver +from tests.conftest import _build_mock_api, NODES_LIST + + +class TestOpen: + def test_open_password_auth(self): + drv = ProxmoxDriver("pve1", "root", "secret") + mock_api = _build_mock_api() + with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api) as mock_cls: + drv.open() + call_kwargs = mock_cls.call_args.kwargs + assert call_kwargs["user"] == "root@pam" + assert call_kwargs["password"] == "secret" + + def test_open_token_auth(self): + drv = ProxmoxDriver( + "pve1", + "root", + "", + optional_args={ + "token_name": "napalm@pam!mytoken", + "token_value": "super-secret", + }, + ) + mock_api = _build_mock_api() + with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api) as mock_cls: + drv.open() + call_kwargs = mock_cls.call_args.kwargs + assert call_kwargs["user"] == "napalm@pam!mytoken" + assert call_kwargs["token_value"] == "super-secret" + + def test_open_connection_error(self): + drv = ProxmoxDriver("badhost", "root", "bad") + with patch( + "napalm_proxmox.driver.ProxmoxAPI", + side_effect=Exception("Connection refused"), + ): + with pytest.raises(ConnectionException): + drv.open() + + def test_node_resolution_by_hostname(self, driver): + assert driver._node_name == "pve1" + + def test_node_override_via_optional_args(self): + drv = ProxmoxDriver( + "pve1.example.com", + "root", + "secret", + optional_args={"node": "custom-node"}, + ) + mock_api = _build_mock_api() + with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api): + drv.open() + assert drv._node_name == "custom-node" + + def test_close(self, driver): + driver.close() + assert driver._api is None + + +class TestIsAlive: + def test_alive_when_api_responds(self, driver): + assert driver.is_alive() == {"is_alive": True} + + def test_not_alive_when_api_fails(self, driver): + driver._api.version.get.side_effect = Exception("timeout") + assert driver.is_alive() == {"is_alive": False} diff --git a/tests/test_environment.py b/tests/test_environment.py new file mode 100644 index 0000000..5dce8e9 --- /dev/null +++ b/tests/test_environment.py @@ -0,0 +1,63 @@ +"""Tests for get_environment.""" + +from __future__ import annotations + +import pytest +from tests.conftest import _build_mock_api, NODE_STATUS +from unittest.mock import patch + +from napalm_proxmox.driver import ProxmoxDriver + + +class TestGetEnvironment: + def test_cpu_usage_present(self, driver): + result = driver.get_environment() + assert "0" in result["cpu"] + assert result["cpu"]["0"]["%usage"] == pytest.approx(12.0, abs=0.1) + + def test_memory_total(self, driver): + result = driver.get_environment() + assert result["memory"]["available_ram"] == 137438953472 + assert result["memory"]["used_ram"] == 68719476736 + + def test_fans_empty_when_no_sensors(self, driver): + result = driver.get_environment() + assert isinstance(result["fans"], dict) + + def test_temperature_with_sensors(self): + from tests.conftest import _build_mock_api, NODE_STATUS + from unittest.mock import patch + from napalm_proxmox.driver import ProxmoxDriver + + sensors_data = [ + {"name": "CPU Package", "value": "55.0"}, + {"name": "GPU Core", "value": "92.0"}, + ] + mock_api = _build_mock_api(sensors=sensors_data) + drv = ProxmoxDriver("pve1", "root", "secret", optional_args={"verify_ssl": False}) + with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api): + drv.open() + + result = drv.get_environment() + assert "CPU Package" in result["temperature"] + assert result["temperature"]["CPU Package"]["temperature"] == 55.0 + assert result["temperature"]["CPU Package"]["is_alert"] is False + assert result["temperature"]["CPU Package"]["is_critical"] is False + + assert "GPU Core" in result["temperature"] + assert result["temperature"]["GPU Core"]["is_alert"] is True + assert result["temperature"]["GPU Core"]["is_critical"] is False + + def test_temperature_critical(self): + from tests.conftest import _build_mock_api + from unittest.mock import patch + from napalm_proxmox.driver import ProxmoxDriver + + sensors_data = [{"name": "NVMe", "value": "98.0"}] + mock_api = _build_mock_api(sensors=sensors_data) + drv = ProxmoxDriver("pve1", "root", "secret", optional_args={"verify_ssl": False}) + with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api): + drv.open() + + result = drv.get_environment() + assert result["temperature"]["NVMe"]["is_critical"] is True diff --git a/tests/test_get_facts.py b/tests/test_get_facts.py new file mode 100644 index 0000000..6514472 --- /dev/null +++ b/tests/test_get_facts.py @@ -0,0 +1,42 @@ +"""Tests for get_facts.""" + +from __future__ import annotations + +import pytest +from tests.conftest import _build_mock_api, NODE_NETWORK +from unittest.mock import patch + +from napalm_proxmox.driver import ProxmoxDriver + + +@pytest.fixture +def facts(driver): + return driver.get_facts() + + +class TestGetFacts: + def test_vendor(self, facts): + assert "Proxmox" in facts["vendor"] + + def test_model(self, facts): + assert facts["model"] == "PowerEdge R640" + + def test_hostname(self, facts): + assert facts["hostname"] == "pve1" + + def test_os_version_contains_version(self, facts): + assert "8.2.4" in facts["os_version"] + + def test_uptime_is_float(self, facts): + assert isinstance(facts["uptime"], float) + assert facts["uptime"] == 123456.0 + + def test_interface_list_sorted(self, facts): + expected = sorted(i["iface"] for i in NODE_NETWORK if i.get("iface")) + assert facts["interface_list"] == expected + + def test_fqdn(self, facts): + assert facts["fqdn"] == "pve1.example.com" + + def test_serial_number_is_string(self, facts): + assert isinstance(facts["serial_number"], str) diff --git a/tests/test_interfaces.py b/tests/test_interfaces.py new file mode 100644 index 0000000..c646847 --- /dev/null +++ b/tests/test_interfaces.py @@ -0,0 +1,109 @@ +"""Tests for get_interfaces, get_interfaces_ip, get_interfaces_counters.""" + +from __future__ import annotations + +import pytest +from tests.conftest import _build_mock_api, NODE_NETWORK +from unittest.mock import patch + +from napalm_proxmox.driver import ProxmoxDriver + + +class TestGetInterfaces: + def test_returns_all_ifaces(self, driver): + result = driver.get_interfaces() + expected_names = {i["iface"] for i in NODE_NETWORK if i.get("iface")} + assert set(result.keys()) == expected_names + + def test_eth0_is_up(self, driver): + result = driver.get_interfaces() + assert result["eth0"]["is_up"] is True + assert result["eth0"]["is_enabled"] is True + + def test_ovs_bridge_not_up(self, driver): + result = driver.get_interfaces() + assert result["vmbr1"]["is_up"] is False + + def test_mac_address_normalized(self, driver): + result = driver.get_interfaces() + mac = result["eth0"]["mac_address"] + # should be lower-case colon-separated + assert mac == "aa:bb:cc:dd:ee:01" + + def test_speed_eth0(self, driver): + result = driver.get_interfaces() + assert result["eth0"]["speed"] == 1000.0 + + def test_speed_unknown(self, driver): + result = driver.get_interfaces() + assert result["vmbr0"]["speed"] == -1.0 + + def test_mtu(self, driver): + result = driver.get_interfaces() + assert result["eth1"]["mtu"] == 9000 + + def test_description(self, driver): + result = driver.get_interfaces() + assert result["eth0"]["description"] == "Management" + + def test_last_flapped_minus_one(self, driver): + result = driver.get_interfaces() + assert result["eth0"]["last_flapped"] == -1.0 + + +class TestGetInterfacesIP: + def test_eth0_has_ipv4(self, driver): + result = driver.get_interfaces_ip() + assert "eth0" in result + assert "ipv4" in result["eth0"] + assert "10.0.0.1" in result["eth0"]["ipv4"] + + def test_eth0_prefix_length(self, driver): + result = driver.get_interfaces_ip() + assert result["eth0"]["ipv4"]["10.0.0.1"]["prefix_length"] == 24 + + def test_sdn_vnet_ipv4_gateway(self, driver): + result = driver.get_interfaces_ip() + # vnet0 gateway 10.10.0.1 /24 + assert "vnet0" in result + assert "10.10.0.1" in result["vnet0"].get("ipv4", {}) + + def test_sdn_vnet_ipv6_gateway(self, driver): + result = driver.get_interfaces_ip() + # vnet1 has IPv6 subnet + assert "vnet1" in result + assert "2001:db8:20::1" in result["vnet1"].get("ipv6", {}) + + def test_iface_without_address_not_in_result(self, driver): + result = driver.get_interfaces_ip() + # eth1 has no address in fixture + assert "eth1" not in result + + +class TestGetInterfacesCounters: + def test_returns_eth0(self, driver): + result = driver.get_interfaces_counters() + assert "eth0" in result + + def test_counters_fields_present(self, driver): + result = driver.get_interfaces_counters() + eth0 = result["eth0"] + for field in ( + "tx_errors", "rx_errors", "tx_discards", "rx_discards", + "tx_octets", "rx_octets", "tx_unicast_packets", "rx_unicast_packets", + "tx_multicast_packets", "rx_multicast_packets", + "tx_broadcast_packets", "rx_broadcast_packets", + ): + assert field in eth0, f"Missing field: {field}" + + def test_counter_values(self, driver): + result = driver.get_interfaces_counters() + eth0 = result["eth0"] + assert eth0["tx_octets"] == 500000 + assert eth0["rx_octets"] == 1000000 + assert eth0["tx_discards"] == 1 + + def test_empty_when_no_netstat(self, driver): + driver._node_api().netstat.get.return_value = [] + result = driver.get_interfaces_counters() + assert result == {} diff --git a/tests/test_misc.py b/tests/test_misc.py new file mode 100644 index 0000000..05f5134 --- /dev/null +++ b/tests/test_misc.py @@ -0,0 +1,323 @@ +"""Tests for NTP, SNMP, users, config management, ping, traceroute, CLI.""" + +from __future__ import annotations + +import pytest +from unittest.mock import MagicMock, patch + + +class TestGetNTPServers: + def test_returns_configured_servers(self, driver): + result = driver.get_ntp_servers() + assert "pool.ntp.org" in result + assert "time.cloudflare.com" in result + + def test_empty_when_no_ntp(self, driver): + driver._node_api().ntp.get.return_value = {} + result = driver.get_ntp_servers() + assert result == {} + + +class TestGetNTPStats: + NTP_OUTPUT = ( + "*192.168.1.1 10.0.0.1 4 u 12 64 377 1.234 0.567 0.123\n" + "+192.168.1.2 10.0.0.2 3 u 30 64 377 2.345 -0.123 0.234\n" + ) + + def test_synchronized_server(self, driver): + driver._node_api().execute.post.return_value = {"data": self.NTP_OUTPUT} + stats = driver.get_ntp_stats() + synced = [s for s in stats if s["synchronized"]] + assert len(synced) == 1 + assert synced[0]["remote"] == "192.168.1.1" + + def test_stratum(self, driver): + driver._node_api().execute.post.return_value = {"data": self.NTP_OUTPUT} + stats = driver.get_ntp_stats() + assert stats[0]["stratum"] == 4 + + def test_empty_when_no_ntp_command(self, driver): + driver._node_api().execute.post.return_value = {"data": ""} + stats = driver.get_ntp_stats() + assert stats == [] + + +class TestGetSNMPInformation: + SNMP_CONF = ( + "# SNMP config\n" + "rocommunity public\n" + "rwcommunity private\n" + "sysLocation Server Room A\n" + "sysContact admin@example.com\n" + ) + + def test_communities(self, driver): + driver._node_api().execute.post.return_value = {"data": self.SNMP_CONF} + result = driver.get_snmp_information() + assert "public" in result["community"] + assert result["community"]["public"]["mode"] == "ro" + assert "private" in result["community"] + assert result["community"]["private"]["mode"] == "rw" + + def test_location(self, driver): + driver._node_api().execute.post.return_value = {"data": self.SNMP_CONF} + result = driver.get_snmp_information() + assert result["location"] == "Server Room A" + + def test_contact(self, driver): + driver._node_api().execute.post.return_value = {"data": self.SNMP_CONF} + result = driver.get_snmp_information() + assert result["contact"] == "admin@example.com" + + def test_chassis_id_is_node_name(self, driver): + driver._node_api().execute.post.return_value = {"data": ""} + result = driver.get_snmp_information() + assert result["chassis_id"] == "pve1" + + +class TestGetUsers: + def test_pve_users_returned(self, driver): + driver._node_api().execute.post.return_value = {"data": ""} + result = driver.get_users() + assert "root@pam" in result + assert "napalm@pam" in result + + def test_user_has_required_keys(self, driver): + driver._node_api().execute.post.return_value = {"data": ""} + result = driver.get_users() + for user, data in result.items(): + assert "level" in data + assert "password" in data + assert "sshkeys" in data + + def test_root_local_user_from_passwd(self, driver): + passwd_output = "root:x:0:0:root:/root:/bin/bash\n" + driver._node_api().execute.post.return_value = {"data": passwd_output} + result = driver.get_users() + assert "root" in result + assert result["root"]["level"] == 15 + + +class TestConfigManagement: + RUNNING = "auto eth0\niface eth0 inet static\n address 10.0.0.1/24\n" + CANDIDATE = "auto eth0\niface eth0 inet static\n address 10.0.0.2/24\n" + + def test_load_merge_candidate_from_string(self, driver): + driver.load_merge_candidate(config=self.CANDIDATE) + assert driver._candidate_config == self.CANDIDATE + + def test_load_replace_candidate_from_string(self, driver): + driver.load_replace_candidate(config=self.CANDIDATE) + assert driver._candidate_config == self.CANDIDATE + + def test_load_merge_from_file(self, driver, tmp_path): + f = tmp_path / "candidate.txt" + f.write_text(self.CANDIDATE) + driver.load_merge_candidate(filename=str(f)) + assert driver._candidate_config == self.CANDIDATE + + def test_discard_clears_candidate(self, driver): + driver.load_replace_candidate(config=self.CANDIDATE) + driver.discard_config() + assert driver._candidate_config == "" + + def test_compare_config_diff(self, driver): + driver._running_config = self.RUNNING + driver.load_replace_candidate(config=self.CANDIDATE) + diff = driver.compare_config() + assert "10.0.0.1" in diff + assert "10.0.0.2" in diff + + def test_compare_config_no_diff_when_same(self, driver): + driver._running_config = self.RUNNING + driver.load_replace_candidate(config=self.RUNNING) + diff = driver.compare_config() + assert diff == "" + + def test_get_config_running(self, driver): + driver._node_api().execute.post.return_value = {"data": self.RUNNING} + result = driver.get_config(retrieve="running") + assert self.RUNNING in result["running"] + assert result["candidate"] == "" + + def test_get_config_sanitized_redacts_password(self, driver): + raw = "password mysecret\nauto eth0\n" + driver._node_api().execute.post.return_value = {"data": raw} + result = driver.get_config(retrieve="running", sanitized=True) + assert "mysecret" not in result["running"] + assert "****" in result["running"] + + def test_load_merge_raises_without_input(self, driver): + with pytest.raises((ValueError, TypeError)): + driver.load_merge_candidate() + + +class TestPing: + PING_SUCCESS = ( + "PING 8.8.8.8 (8.8.8.8) 100(128) bytes of data.\n" + "108 bytes from 8.8.8.8: icmp_seq=1 ttl=118 time=10.5 ms\n" + "108 bytes from 8.8.8.8: icmp_seq=2 ttl=118 time=11.2 ms\n" + "--- 8.8.8.8 ping statistics ---\n" + "2 packets transmitted, 2 received, 0% packet loss, time 1001ms\n" + "rtt min/avg/max/mdev = 10.500/10.850/11.200/0.350 ms\n" + ) + PING_FAIL = ( + "PING badhost: Name or service not known\n" + "0 packets transmitted, 0 received, 100% packet loss\n" + ) + + def test_success(self, driver): + driver._node_api().execute.post.return_value = {"data": self.PING_SUCCESS} + result = driver.ping("8.8.8.8") + assert "success" in result + assert result["success"]["probes_sent"] == 2 + assert result["success"]["packet_loss"] == 0 + assert result["success"]["rtt_min"] == pytest.approx(10.5, abs=0.01) + + def test_failure(self, driver): + driver._node_api().execute.post.return_value = {"data": self.PING_FAIL} + result = driver.ping("badhost") + assert "error" in result + + def test_exec_unavailable(self, driver): + driver._node_api().execute.post.return_value = {"data": ""} + result = driver.ping("8.8.8.8") + assert "error" in result + + +class TestTraceroute: + TR_OUTPUT = ( + "traceroute to 8.8.8.8 (8.8.8.8), 30 hops max, 60 byte packets\n" + " 1 10.0.0.254 0.456 ms 0.423 ms 0.401 ms\n" + " 2 192.168.1.1 1.234 ms 1.210 ms 1.189 ms\n" + " 3 8.8.8.8 11.200 ms 11.150 ms 11.100 ms\n" + ) + + def test_success_hops(self, driver): + driver._node_api().execute.post.return_value = {"data": self.TR_OUTPUT} + result = driver.traceroute("8.8.8.8") + assert "success" in result + assert 1 in result["success"] + assert 3 in result["success"] + + def test_hop_rtt(self, driver): + driver._node_api().execute.post.return_value = {"data": self.TR_OUTPUT} + result = driver.traceroute("8.8.8.8") + hop1 = result["success"][1]["probes"] + assert hop1[1]["ip_address"] == "10.0.0.254" + + def test_empty_returns_error(self, driver): + driver._node_api().execute.post.return_value = {"data": ""} + result = driver.traceroute("8.8.8.8") + assert "error" in result + + +class TestCLI: + def test_cli_single_command(self, driver): + driver._node_api().execute.post.return_value = {"data": "hello world"} + result = driver.cli(["echo hello world"]) + assert result["echo hello world"] == "hello world" + + def test_cli_multiple_commands(self, driver): + responses = {"uptime": "up 2 days", "hostname": "pve1"} + call_count = 0 + + def _exec(command): + nonlocal call_count + cmd = command + call_count += 1 + return {"data": responses.get(cmd, "")} + + driver._node_api().execute.post.side_effect = _exec + result = driver.cli(["uptime", "hostname"]) + assert "uptime" in result + assert "hostname" in result + + +class TestGetRouteTo: + ROUTE_OUTPUT = ( + "10.0.0.0/24 dev eth0 proto kernel scope link src 10.0.0.1\n" + "default via 10.0.0.254 dev eth0 proto dhcp\n" + ) + + def test_returns_routes(self, driver): + driver._node_api().execute.post.return_value = {"data": self.ROUTE_OUTPUT} + result = driver.get_route_to(destination="10.0.0.0/24") + assert "10.0.0.0/24" in result + + def test_route_fields(self, driver): + driver._node_api().execute.post.return_value = {"data": self.ROUTE_OUTPUT} + result = driver.get_route_to() + for prefix, routes in result.items(): + for route in routes: + assert "protocol" in route + assert "next_hop" in route + assert "outgoing_interface" in route + + def test_protocol_filter(self, driver): + driver._node_api().execute.post.return_value = {"data": self.ROUTE_OUTPUT} + result = driver.get_route_to(protocol="dhcp") + for prefix, routes in result.items(): + for route in routes: + assert "dhcp" in route["protocol"].lower() + + +class TestLLDPNeighbors: + LLDP_SUMMARY = ( + " Interface: eth0\n" + " SysName: sw01.example.com\n" + " PortID: ifname GigabitEthernet1/0/1\n" + " Interface: eth1\n" + " SysName: sw02.example.com\n" + " PortID: ifname GigabitEthernet1/0/2\n" + ) + + def test_neighbors_found(self, driver): + driver._node_api().execute.post.return_value = {"data": self.LLDP_SUMMARY} + result = driver.get_lldp_neighbors() + assert "eth0" in result + assert result["eth0"][0]["hostname"] == "sw01.example.com" + + def test_empty_when_no_lldp(self, driver): + driver._node_api().execute.post.return_value = {"data": ""} + result = driver.get_lldp_neighbors() + assert result == {} + + +class TestIPv6Neighbors: + IPV6_NEIGH = ( + "2001:db8::1 dev eth0 lladdr aa:bb:cc:dd:ee:ff REACHABLE\n" + "2001:db8::2 dev vmbr0 lladdr 11:22:33:44:55:66 STALE\n" + "fe80::1 dev eth0 FAILED\n" + ) + + def test_neighbors_parsed(self, driver): + driver._node_api().execute.post.return_value = {"data": self.IPV6_NEIGH} + result = driver.get_ipv6_neighbors_table() + assert len(result) == 2 # FAILED entry excluded + + def test_mac_normalized(self, driver): + driver._node_api().execute.post.return_value = {"data": self.IPV6_NEIGH} + result = driver.get_ipv6_neighbors_table() + macs = {e["mac"] for e in result} + assert "aa:bb:cc:dd:ee:ff" in macs + + def test_state_present(self, driver): + driver._node_api().execute.post.return_value = {"data": self.IPV6_NEIGH} + result = driver.get_ipv6_neighbors_table() + states = {e["state"] for e in result} + assert "REACHABLE" in states + + +class TestNotImplemented: + def test_get_bgp_config_raises(self, driver): + with pytest.raises(NotImplementedError): + driver.get_bgp_config() + + def test_get_bgp_neighbors_raises(self, driver): + with pytest.raises(NotImplementedError): + driver.get_bgp_neighbors() + + def test_get_optics_raises(self, driver): + with pytest.raises(NotImplementedError): + driver.get_optics() diff --git a/tests/test_ovs_and_arp.py b/tests/test_ovs_and_arp.py new file mode 100644 index 0000000..b514bb6 --- /dev/null +++ b/tests/test_ovs_and_arp.py @@ -0,0 +1,163 @@ +"""Tests for OVS-related helpers and MAC/ARP methods.""" + +from __future__ import annotations + +import pytest + +from napalm_proxmox import utils + + +PROC_NET_ARP = """\ +IP address HW type Flags HW address Mask Device +10.0.0.254 0x1 0x2 aa:bb:cc:dd:ee:ff * eth0 +10.0.0.10 0x1 0x2 11:22:33:44:55:66 * vmbr0 +10.0.0.20 0x1 0x0 00:00:00:00:00:00 * eth0 +""" + +OVS_VSCTL_SHOW = """\ + Bridge vmbr1 + Port eth1 + Port vnet0 + Port vmbr1 + Bridge vmbr2 + Port eth2 +""" + +BRIDGE_FDB_OUTPUT = """\ +aa:bb:cc:dd:ee:01 dev eth0 vlan 10 master vmbr0 permanent +cc:dd:ee:ff:00:11 dev eth0 vlan 20 master vmbr0 +""" + +OVS_FDB_SHOW = """\ + LOCAL 0 aa:bb:cc:dd:ee:03 0 + 1 10 11:22:33:44:55:66 100 + 2 20 66:55:44:33:22:11 200 +""" + + +class TestNormalizeMAC: + def test_colon_separated(self): + assert utils.normalize_mac("AA:BB:CC:DD:EE:FF") == "aa:bb:cc:dd:ee:ff" + + def test_dash_separated(self): + assert utils.normalize_mac("AA-BB-CC-DD-EE-FF") == "aa:bb:cc:dd:ee:ff" + + def test_empty_string(self): + assert utils.normalize_mac("") == "" + + def test_already_normalized(self): + assert utils.normalize_mac("aa:bb:cc:dd:ee:ff") == "aa:bb:cc:dd:ee:ff" + + +class TestParseARPOutput: + def test_parses_valid_entries(self): + entries = utils.parse_arp_output(PROC_NET_ARP) + # 10.0.0.20 has 00:00:00:00:00:00 → excluded by ARP parse (not by proc/net/arp parser) + # proc/net/arp in driver excludes 00:00..., parse_arp_output is for `arp -n` format + # The proc/net/arp format is handled inline in driver. + # parse_arp_output handles `arp -n` which uses "HWtype HWaddress" columns + # Let's test with arp -n format + arp_n = ( + "Address HWtype HWaddress Flags Iface\n" + "10.0.0.254 ether aa:bb:cc:dd:ee:ff C eth0\n" + "10.0.0.10 ether 11:22:33:44:55:66 C vmbr0\n" + ) + entries = utils.parse_arp_output(arp_n) + assert len(entries) == 2 + assert entries[0]["ip"] == "10.0.0.254" + assert entries[0]["mac"] == "aa:bb:cc:dd:ee:ff" + assert entries[0]["interface"] == "eth0" + + def test_skips_incomplete(self): + arp_n = ( + "Address HWtype HWaddress Flags Iface\n" + "10.0.0.1 ether (incomplete) C eth0\n" + ) + entries = utils.parse_arp_output(arp_n) + assert len(entries) == 0 + + +class TestParseOVSVsctlShow: + def test_bridges_found(self): + result = utils.parse_ovs_vsctl_show(OVS_VSCTL_SHOW) + assert "vmbr1" in result + assert "vmbr2" in result + + def test_ports_found(self): + result = utils.parse_ovs_vsctl_show(OVS_VSCTL_SHOW) + assert "eth1" in result["vmbr1"]["ports"] + assert "vnet0" in result["vmbr1"]["ports"] + # The bridge self-port should not appear in ports list + assert "vmbr1" not in result["vmbr1"]["ports"] + + def test_second_bridge_ports(self): + result = utils.parse_ovs_vsctl_show(OVS_VSCTL_SHOW) + assert "eth2" in result["vmbr2"]["ports"] + + +class TestGetARPTable: + PROC_ARP = ( + "IP address HW type Flags HW address Mask Device\n" + "10.0.0.254 0x1 0x2 aa:bb:cc:dd:ee:ff * eth0\n" + "10.0.0.10 0x1 0x2 11:22:33:44:55:66 * vmbr0\n" + "10.0.0.20 0x1 0x0 00:00:00:00:00:00 * eth0\n" + ) + + def test_arp_entries_returned(self, driver): + driver._node_api().execute.post.return_value = {"data": self.PROC_ARP} + result = driver.get_arp_table() + assert len(result) == 2 # 00:00:00:00:00:00 excluded + + def test_mac_normalized(self, driver): + driver._node_api().execute.post.return_value = {"data": self.PROC_ARP} + result = driver.get_arp_table() + macs = {e["mac"] for e in result} + assert "aa:bb:cc:dd:ee:ff" in macs + + def test_vrf_filter(self, driver): + driver._node_api().execute.post.return_value = {"data": self.PROC_ARP} + result = driver.get_arp_table(vrf="vmbr0") + assert all(e["interface"] == "vmbr0" for e in result) + + def test_empty_when_exec_unavailable(self, driver): + driver._node_api().execute.post.return_value = {"data": ""} + result = driver.get_arp_table() + assert result == [] + + +class TestGetMACAddressTable: + BRIDGE_FDB = ( + "aa:bb:cc:dd:ee:01 dev eth0 vlan 10 master vmbr0 permanent\n" + "cc:dd:ee:ff:00:11 dev eth0 vlan 20 master vmbr0\n" + ) + + OVS_APPCTL = ( + " LOCAL 0 aa:bb:cc:dd:ee:03 0\n" + " 1 10 11:22:33:44:55:66 100\n" + ) + + def test_linux_bridge_entries(self, driver): + def _exec(cmd): + if "bridge fdb" in cmd: + return self.BRIDGE_FDB + return "" + + driver._node_api().execute.post.side_effect = lambda command: { + "data": _exec(command) + } + result = driver.get_mac_address_table() + macs = {e["mac"] for e in result} + assert "aa:bb:cc:dd:ee:01" in macs + + def test_static_flag(self, driver): + def _exec(cmd): + if "bridge fdb" in cmd: + return self.BRIDGE_FDB + return "" + + driver._node_api().execute.post.side_effect = lambda command: { + "data": _exec(command) + } + result = driver.get_mac_address_table() + static_entries = [e for e in result if e["mac"] == "aa:bb:cc:dd:ee:01"] + assert static_entries[0]["static"] is True diff --git a/tests/test_sdn.py b/tests/test_sdn.py new file mode 100644 index 0000000..3c68d97 --- /dev/null +++ b/tests/test_sdn.py @@ -0,0 +1,89 @@ +"""Tests for SDN-related methods: get_vlans, get_network_instances.""" + +from __future__ import annotations + +import pytest +from tests.conftest import _build_mock_api, SDN_ZONES, SDN_VNETS + + +class TestGetVlans: + def test_sdn_vnets_appear_as_vlans(self, driver): + result = driver.get_vlans() + # vnet0 tag=10, vnet1 tag=20 + assert "10" in result + assert "20" in result + + def test_vlan_name_is_vnet_id(self, driver): + result = driver.get_vlans() + assert result["10"]["name"] == "vnet0" + assert result["20"]["name"] == "vnet1" + + def test_vlan_100000_for_vxlan(self, driver): + result = driver.get_vlans() + assert "100000" in result + + def test_bridge_vlan_show_parsing(self, driver): + # Simulate bridge vlan output + bridge_output = ( + "vmbr0 1\n" + " 10\n" + " 20\n" + "eth0 1\n" + ) + driver._node_api().execute.post.return_value = {"data": bridge_output} + result = driver.get_vlans() + # Interface vmbr0 should appear in vlan 1 + entry = result.get("1", {}) + assert "vmbr0" in entry.get("interfaces", []) + + def test_empty_sdn_returns_dict(self): + from unittest.mock import patch + from napalm_proxmox.driver import ProxmoxDriver + from tests.conftest import _build_mock_api + + mock_api = _build_mock_api(sdn_vnets=[], sdn_zones=[]) + drv = ProxmoxDriver("pve1", "root", "secret", optional_args={"verify_ssl": False}) + with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api): + drv.open() + result = drv.get_vlans() + assert isinstance(result, dict) + + +class TestGetNetworkInstances: + def test_default_instance_always_present(self, driver): + result = driver.get_network_instances() + assert "default" in result + assert result["default"]["type"] == "DEFAULT_INSTANCE" + + def test_sdn_zones_as_instances(self, driver): + result = driver.get_network_instances() + assert "zone-vlan" in result + assert "zone-vxlan" in result + assert "zone-evpn" in result + + def test_zone_type_mapping(self, driver): + result = driver.get_network_instances() + assert result["zone-vlan"]["type"] == "L2VPN" + assert result["zone-evpn"]["type"] == "L3VRF" + + def test_vnet_appears_in_zone_interfaces(self, driver): + result = driver.get_network_instances() + zone_ifaces = result["zone-vlan"]["interfaces"]["interface"] + assert "vnet0" in zone_ifaces + assert "vnet1" in zone_ifaces + + def test_filter_by_name(self, driver): + result = driver.get_network_instances(name="zone-vlan") + assert "zone-vlan" in result + assert "zone-vxlan" not in result + + def test_evpn_route_distinguisher(self, driver): + result = driver.get_network_instances() + rd = result["zone-evpn"]["state"]["route_distinguisher"] + assert rd == 10001 + + def test_default_interfaces_include_node_ifaces(self, driver): + result = driver.get_network_instances() + default_ifaces = result["default"]["interfaces"]["interface"] + assert "eth0" in default_ifaces + assert "vmbr0" in default_ifaces diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..90d8430 --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,224 @@ +"""Unit tests for napalm_proxmox.utils.""" + +from __future__ import annotations + +import pytest +from napalm_proxmox import utils + + +class TestNormalizeMAC: + def test_colon_upper(self): + assert utils.normalize_mac("AA:BB:CC:DD:EE:FF") == "aa:bb:cc:dd:ee:ff" + + def test_dash_separated(self): + assert utils.normalize_mac("AA-BB-CC-DD-EE-FF") == "aa:bb:cc:dd:ee:ff" + + def test_lowercase_passthrough(self): + assert utils.normalize_mac("aa:bb:cc:dd:ee:ff") == "aa:bb:cc:dd:ee:ff" + + def test_empty(self): + assert utils.normalize_mac("") == "" + + +class TestParseCIDR: + def test_ipv4(self): + ip, plen = utils.parse_cidr("192.168.1.1/24") + assert ip == "192.168.1.1" + assert plen == 24 + + def test_ipv6(self): + ip, plen = utils.parse_cidr("2001:db8::1/64") + assert ip == "2001:db8::1" + assert plen == 64 + + def test_no_prefix(self): + ip, plen = utils.parse_cidr("10.0.0.1") + assert ip == "10.0.0.1" + assert plen == 32 + + def test_host_route(self): + ip, plen = utils.parse_cidr("10.0.0.1/32") + assert plen == 32 + + +class TestPrefixLength: + def test_slash24(self): + assert utils.prefix_length("10.0.0.0/24") == 24 + + def test_slash_16(self): + assert utils.prefix_length("192.168.0.0/16") == 16 + + def test_no_slash(self): + assert utils.prefix_length("10.0.0.1") == 32 + + +class TestAddressesFromNodeNetwork: + def test_address_and_netmask(self): + data = {"address": "10.0.0.1", "netmask": "255.255.255.0"} + result = utils.addresses_from_node_network(data) + assert "ipv4" in result + assert "10.0.0.1" in result["ipv4"] + assert result["ipv4"]["10.0.0.1"]["prefix_length"] == 24 + + def test_cidr_ipv4(self): + data = {"cidr": "10.0.0.1/24"} + result = utils.addresses_from_node_network(data) + assert "ipv4" in result + assert "10.0.0.1" in result["ipv4"] + + def test_cidr_ipv6(self): + data = {"cidr6": "2001:db8::1/64"} + result = utils.addresses_from_node_network(data) + assert "ipv6" in result + assert "2001:db8::1" in result["ipv6"] + assert result["ipv6"]["2001:db8::1"]["prefix_length"] == 64 + + def test_both_ipv4_and_ipv6(self): + data = {"cidr": "10.0.0.1/24", "cidr6": "2001:db8::1/64"} + result = utils.addresses_from_node_network(data) + assert "ipv4" in result + assert "ipv6" in result + + def test_empty_input(self): + result = utils.addresses_from_node_network({}) + assert result == {} + + +class TestIsOVSInterface: + def test_ovs_bridge(self): + assert utils.is_ovs_interface({"type": "OVSBridge"}) is True + + def test_ovs_bond(self): + assert utils.is_ovs_interface({"type": "OVSBond"}) is True + + def test_ovs_int_port(self): + assert utils.is_ovs_interface({"type": "OVSIntPort"}) is True + + def test_linux_bridge(self): + assert utils.is_ovs_interface({"type": "bridge"}) is False + + def test_eth(self): + assert utils.is_ovs_interface({"type": "eth"}) is False + + def test_no_type_key(self): + assert utils.is_ovs_interface({}) is False + + +class TestIsSDNVNet: + def test_vnet_pattern(self): + assert utils.is_sdn_vnet("vnet0") is True + assert utils.is_sdn_vnet("vnet99") is True + + def test_non_vnet(self): + assert utils.is_sdn_vnet("eth0") is False + assert utils.is_sdn_vnet("vmbr0") is False + + def test_vnet_with_suffix(self): + assert utils.is_sdn_vnet("vnet0sub") is False + + +class TestSpeedMbps: + def test_speed_present(self): + assert utils.speed_mbps({"speed": "1000"}) == 1000.0 + + def test_speed_mbit_suffix(self): + # Some entries may have "Mbit" unit in the string — handled by parsing + assert utils.speed_mbps({"speed": "10000"}) == 10000.0 + + def test_no_speed(self): + assert utils.speed_mbps({}) == -1.0 + + def test_speed_non_numeric(self): + assert utils.speed_mbps({"speed": "unknown"}) == -1.0 + + +class TestSDNZoneToNetworkInstance: + def test_vlan_zone(self): + zone = {"zone": "z1", "type": "vlan", "mtu": 1500} + result = utils.sdn_zone_to_network_instance(zone) + assert result["type"] == "L2VPN" + assert result["name"] == "z1" + + def test_evpn_zone(self): + zone = {"zone": "z2", "type": "evpn", "peers": "10.0.0.1", "rt": 10001} + result = utils.sdn_zone_to_network_instance(zone) + assert result["type"] == "L3VRF" + + def test_vxlan_zone(self): + zone = {"zone": "z3", "type": "vxlan", "mtu": 9000} + result = utils.sdn_zone_to_network_instance(zone) + assert result["type"] == "L2VPN" + + def test_simple_zone(self): + zone = {"zone": "z4", "type": "simple"} + result = utils.sdn_zone_to_network_instance(zone) + assert result["type"] == "L3VRF" + + def test_unknown_zone_type(self): + zone = {"zone": "z5", "type": "qinq"} + result = utils.sdn_zone_to_network_instance(zone) + assert result["type"] == "L2VPN" + + def test_result_has_interfaces(self): + zone = {"zone": "z1", "type": "vlan"} + result = utils.sdn_zone_to_network_instance(zone) + assert "interfaces" in result + + +class TestParseARPOutput: + ARP_N = ( + "Address HWtype HWaddress Flags Iface\n" + "10.0.0.254 ether aa:bb:cc:dd:ee:ff C eth0\n" + "10.0.0.10 ether 11:22:33:44:55:66 C vmbr0\n" + "? (10.0.0.20) at on eth0\n" + ) + + def test_parses_two_entries(self): + result = utils.parse_arp_output(self.ARP_N) + assert len(result) == 2 + + def test_entry_fields(self): + result = utils.parse_arp_output(self.ARP_N) + assert result[0]["ip"] == "10.0.0.254" + assert result[0]["mac"] == "aa:bb:cc:dd:ee:ff" + assert result[0]["interface"] == "eth0" + + def test_skips_incomplete(self): + result = utils.parse_arp_output(self.ARP_N) + ips = {e["ip"] for e in result} + assert "10.0.0.20" not in ips + + def test_empty_input(self): + result = utils.parse_arp_output("") + assert result == [] + + +class TestParseOVSVsctlShow: + RAW = ( + " Bridge vmbr1\n" + " Port eth1\n" + " Port vnet0\n" + " Port vmbr1\n" + " Bridge vmbr2\n" + " Port eth2\n" + " Port vmbr2\n" + ) + + def test_finds_both_bridges(self): + result = utils.parse_ovs_vsctl_show(self.RAW) + assert "vmbr1" in result + assert "vmbr2" in result + + def test_ports_exclude_self(self): + result = utils.parse_ovs_vsctl_show(self.RAW) + assert "vmbr1" not in result["vmbr1"]["ports"] + assert "vmbr2" not in result["vmbr2"]["ports"] + + def test_ports_included(self): + result = utils.parse_ovs_vsctl_show(self.RAW) + assert "eth1" in result["vmbr1"]["ports"] + assert "eth2" in result["vmbr2"]["ports"] + + def test_empty_input(self): + result = utils.parse_ovs_vsctl_show("") + assert result == {}