CHANGELOG.md

@@ -6,6 +6,12 @@ All notable changes to this driver. The format follows
 
 ## [Unreleased]
 
+### Fixed
+- `get_vpn_tunnels()` keys a WireGuard peer by its interface and the start of its
+  public key (`wireguard-wg0-aBcD1234`, URL-safe) instead of its position, so a peer
+  added ahead of another no longer renumbers it (#32). Every WireGuard key changes
+  once with this release.
+
 ## [0.5.0] - 2026-10-23
 
 The first release on PyPI.

napalm_opnsense/opnsense.py

@@ -69,6 +69,7 @@ from napalm_opnsense.ping_mixin import OPNsensePingMixin
 from napalm_opnsense.port_forwards import alias_index, port_forwards, wan_interfaces
 from napalm_opnsense.resolver_mixin import OPNsenseResolverMixin
 from napalm_opnsense.virtual_ips import carp_addresses
+from napalm_opnsense.wireguard import wireguard_tunnels
 
 logger = logging.getLogger(__name__)
 
@@ -2167,51 +2168,7 @@ class OPNsenseDriver(
 
         # ── WireGuard ──────────────────────────────────────────────────────
         try:
-            # /api/wireguard/service/show returns structured JSON:
-            # {"total": N, "rows": [
-            #   {"if":"wg0", "type":"interface", ...},
-            #   {"if":"wg0", "type":"peer", "public-key":"...",
-            #    "endpoint":"1.2.3.4:51820", "transfer-rx":N, "transfer-tx":N,
-            #    "name":"HCQ", "peer-status":"online",
-            #    "latest-handshake-age":45, "ifname":"HCQ"}, ...
-            # ]}
-            data = self._get("/api/wireguard/service/show")
-            rows = data.get("rows", []) if isinstance(data, dict) else []
-
-            wg_idx = 0
-            for row in rows:
-                if row.get("type") != "peer":
-                    continue
-                wg_idx += 1
-
-                endpoint = (row.get("endpoint") or "").strip()
-                if endpoint and endpoint != "(none)":
-                    remote_ip = endpoint.rsplit(":", 1)[0].strip("[]")
-                else:
-                    remote_ip = ""
-
-                is_up = row.get("peer-status") == "online"
-                bytes_in = int(row.get("transfer-rx") or 0)
-                bytes_out = int(row.get("transfer-tx") or 0)
-                hs_age = row.get("latest-handshake-age")
-                uptime = int(hs_age) if hs_age else 0
-
-                description = (
-                    (row.get("name") or "").strip()
-                    or (row.get("ifname") or "").strip()
-                    or f"wg-peer-{wg_idx}"
-                )
-
-                tunnels[f"wireguard-{wg_idx}"] = {
-                    "type": "WireGuard",
-                    "local_endpoint": "",
-                    "remote_endpoint": remote_ip,
-                    "is_up": is_up,
-                    "uptime": uptime,
-                    "bytes_in": bytes_in,
-                    "bytes_out": bytes_out,
-                    "description": description,
-                }
+            tunnels.update(wireguard_tunnels(self._get("/api/wireguard/service/show")))
         except Exception as exc:
             logger.debug("WireGuard status fetch failed (plugin not installed?): %s", exc)
 

napalm_opnsense/wireguard.py

@@ -0,0 +1,73 @@
+"""WireGuard peers as VPN tunnels. Pure: no I/O.
+
+``/api/wireguard/service/show`` lists every WireGuard interface and peer::
+
+    {"total": N, "rows": [
+      {"if": "wg0", "type": "interface", ...},
+      {"if": "wg0", "type": "peer", "public-key": "...",
+       "endpoint": "1.2.3.4:51820", "transfer-rx": N, "transfer-tx": N,
+       "name": "office", "peer-status": "online",
+       "latest-handshake-age": 45, "ifname": "office"}, ...
+    ]}
+
+**The key is the peer, not its place.** It used to be a running counter, so a
+peer added ahead of another renumbered it, and whatever a consumer keeps per
+key (netOrk's site link, its tunnel check) moved to a different peer (#32).
+Now it is the interface and the start of the public key, as napalm-linux does
+it, with the base64 made URL-safe because the key is used as a path segment.
+Two peers sharing the prefix get the whole key; a row without a public key
+keeps its position.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+#: How much of a public key names a peer, the same length as napalm-linux uses.
+_KEY_PREFIX = 8
+
+
+def _key(row: dict[str, Any], position: int, taken: dict[str, Any]) -> str:
+    public_key = (row.get("public-key") or "").strip()
+    if not public_key:
+        return f"wireguard-{position}"
+    safe = public_key.rstrip("=").replace("+", "-").replace("/", "_")
+    key = f"wireguard-{row.get('if') or 'wg'}-{safe[:_KEY_PREFIX]}"
+    if key in taken:
+        key = f"wireguard-{row.get('if') or 'wg'}-{safe}"
+    return key
+
+
+def _remote_ip(row: dict[str, Any]) -> str:
+    endpoint = (row.get("endpoint") or "").strip()
+    if not endpoint or endpoint == "(none)":
+        return ""
+    return endpoint.rsplit(":", 1)[0].strip("[]")
+
+
+def _tunnel(row: dict[str, Any], position: int) -> dict[str, Any]:
+    handshake_age = row.get("latest-handshake-age")
+    return {
+        "type": "WireGuard",
+        "local_endpoint": "",
+        "remote_endpoint": _remote_ip(row),
+        "is_up": row.get("peer-status") == "online",
+        "uptime": int(handshake_age) if handshake_age else 0,
+        "bytes_in": int(row.get("transfer-rx") or 0),
+        "bytes_out": int(row.get("transfer-tx") or 0),
+        "description": (
+            (row.get("name") or "").strip()
+            or (row.get("ifname") or "").strip()
+            or f"wg-peer-{position}"
+        ),
+    }
+
+
+def wireguard_tunnels(show: Any) -> dict[str, dict[str, Any]]:
+    """One ``VPNTunnelDict`` per peer in ``/api/wireguard/service/show``, by tunnel key."""
+    rows = show.get("rows", []) if isinstance(show, dict) else []
+    tunnels: dict[str, dict[str, Any]] = {}
+    peers = [row for row in rows if isinstance(row, dict) and row.get("type") == "peer"]
+    for position, row in enumerate(peers, start=1):
+        tunnels[_key(row, position, tunnels)] = _tunnel(row, position)
+    return tunnels

tests/quality_baseline.json

@@ -24,7 +24,7 @@
   "napalm_opnsense/opnsense.py::OPNsenseDriver.get_lldp_neighbors_detail": 12,
   "napalm_opnsense/opnsense.py::OPNsenseDriver.get_networks": 25,
   "napalm_opnsense/opnsense.py::OPNsenseDriver.get_route_to": 23,
-  "napalm_opnsense/opnsense.py::OPNsenseDriver.get_vpn_tunnels": 60,
+  "napalm_opnsense/opnsense.py::OPNsenseDriver.get_vpn_tunnels": 47,
   "napalm_opnsense/opnsense.py::OPNsenseDriver.load_merge_candidate": 13,
   "napalm_opnsense/opnsense.py::OPNsenseDriver.sync_dns_zone": 20,
   "napalm_opnsense/port_forwards.py::_forwards_of": 12,
@@ -43,7 +43,7 @@
   "napalm_opnsense/opnsense.py::OPNsenseDriver.get_interfaces_ip": 68,
   "napalm_opnsense/opnsense.py::OPNsenseDriver.get_networks": 85,
   "napalm_opnsense/opnsense.py::OPNsenseDriver.get_route_to": 85,
-  "napalm_opnsense/opnsense.py::OPNsenseDriver.get_vpn_tunnels": 189,
+  "napalm_opnsense/opnsense.py::OPNsenseDriver.get_vpn_tunnels": 145,
   "napalm_opnsense/opnsense.py::OPNsenseDriver.load_merge_candidate": 56,
   "napalm_opnsense/opnsense.py::OPNsenseDriver.sync_dns_zone": 112
  },
@@ -57,7 +57,7 @@
   "napalm_opnsense.resolver_mixin"
  ],
  "python_file_lines": {
-  "napalm_opnsense/opnsense.py": 3125
+  "napalm_opnsense/opnsense.py": 3082
  },
  "untyped_public": [
   "napalm_opnsense/opnsense.py::OPNsenseDriver.get_snmp_config"

tests/unit/test_vpn_tunnels.py

@@ -0,0 +1,106 @@
+"""WireGuard peers in ``get_vpn_tunnels()`` keep their key when others come and go (#32).
+
+The key was a running counter over the peer rows, so a peer added ahead of an
+existing one renumbered every peer after it, and whatever a consumer stored per
+key (netOrk's site link, its tunnel check) silently moved to another peer. A
+peer is its interface and its public key, so that is what the key is made of.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+from napalm_opnsense.opnsense import OPNsenseDriver
+
+OFFICE = {
+    "if": "wg0",
+    "type": "peer",
+    "name": "office",
+    "public-key": "aBcD1234efGH5678ijKL9012mnOP3456qrST7890uvw=",
+    "endpoint": "198.51.100.7:51820",
+    "peer-status": "online",
+    "transfer-rx": 10,
+    "transfer-tx": 20,
+    "latest-handshake-age": 45,
+}
+LAPTOP = {
+    "if": "wg0",
+    "type": "peer",
+    "name": "laptop",
+    "public-key": "Zz9/+yXw8765vuTS4321rqPO0987nmLK6543jiHG210=",
+    "endpoint": "(none)",
+    "peer-status": "offline",
+}
+SERVER = {
+    "if": "wg1",
+    "type": "peer",
+    "name": "server",
+    "public-key": "QqQqWwWwEeEeRrRrTtTtYyYyUuUuIiIiOoOoPpPp123=",
+    "endpoint": "203.0.113.9:51821",
+    "peer-status": "online",
+}
+
+
+def _driver(*rows: dict) -> OPNsenseDriver:
+    with patch("napalm_opnsense.opnsense.requests.Session"):
+        drv = OPNsenseDriver(hostname="fw", username="k", password="s")
+    drv.session = MagicMock()
+    show = {"rows": [{"if": "wg0", "type": "interface"}, *rows]}
+
+    def get(path: str) -> dict:
+        if path == "/api/wireguard/service/show":
+            return show
+        raise RuntimeError("not installed")  # IPsec and OpenVPN are not part of this
+
+    drv._get = get  # type: ignore[method-assign]
+    return drv
+
+
+def _by_name(tunnels: dict) -> dict[str, str]:
+    return {t["description"]: key for key, t in tunnels.items()}
+
+
+class TestTheKey:
+    def test_is_the_interface_and_the_start_of_the_public_key(self):
+        assert list(_driver(OFFICE).get_vpn_tunnels()) == ["wireguard-wg0-aBcD1234"]
+
+    def test_survives_a_peer_added_ahead_of_it(self):
+        before = _by_name(_driver(OFFICE, SERVER).get_vpn_tunnels())
+        after = _by_name(_driver(OFFICE, LAPTOP, SERVER).get_vpn_tunnels())
+        assert after["server"] == before["server"]
+        assert after["office"] == before["office"]
+
+    def test_is_safe_in_a_url_path(self):
+        # The key is a path segment (PATCH /devices/{id}/vpn-tunnels/{key}), and
+        # a "/" from base64 would split it even when encoded.
+        key = _by_name(_driver(LAPTOP).get_vpn_tunnels())["laptop"]
+        assert key == "wireguard-wg0-Zz9_-yXw"
+
+    def test_two_peers_with_the_same_key_start_both_stay(self):
+        twin = {
+            **SERVER,
+            "if": "wg0",
+            "name": "twin",
+            "public-key": OFFICE["public-key"][:8] + "x=",
+        }
+        tunnels = _driver(OFFICE, twin).get_vpn_tunnels()
+        assert len(tunnels) == 2
+
+    def test_without_a_public_key_falls_back_to_the_position(self):
+        keyless = {k: v for k, v in OFFICE.items() if k != "public-key"}
+        assert list(_driver(keyless).get_vpn_tunnels()) == ["wireguard-1"]
+
+
+class TestTheEntry:
+    def test_carries_the_peer_s_reading(self):
+        tunnel = _driver(OFFICE).get_vpn_tunnels()["wireguard-wg0-aBcD1234"]
+        assert tunnel == {
+            "type": "WireGuard",
+            "local_endpoint": "",
+            "remote_endpoint": "198.51.100.7",
+            "is_up": True,
+            "uptime": 45,
+            "bytes_in": 10,
+            "bytes_out": 20,
+            "description": "office",
+        }