refactor: WireGuard tunnels are read in their own module (#32)
driver / lint (pull_request) Successful in 13s
driver / standard (pull_request) Successful in 14s
driver / test (3.10) (pull_request) Failing after 6s
driver / types (pull_request) Successful in 1m2s
driver / test (3.11) (pull_request) Successful in 1m5s
driver / test (3.12) (pull_request) Successful in 1m5s
driver / lowest (pull_request) Successful in 33s
driver / test (3.13) (pull_request) Successful in 59s
driver / build (pull_request) Successful in 27s
driver / security (pull_request) Successful in 1m1s
CI / driver (pull_request) Failing after 2m25s
driver / publish (pull_request) Skipped

get_vpn_tunnels() grew past the quality baseline with the key fix. The
WireGuard part is pure parsing of /api/wireguard/service/show, like the
listen ports in listening.py, so it moves to wireguard.py and the driver
calls it. opnsense.py shrinks from 3125 to 3082 lines.
This commit is contained in:
Christian Manivong committed 2026-10-27 10:03:30 +01:00
1 parent 9f45952605
commit eaafa540e5
3 files changed
+78 -70

No files matched your search

+73
View File
@@ -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