Add support for WireGuard

Almost full support for WireGuard admin@server:/> show interface wg0 name : wg0 type : wireguard index : 10 mtu : 1420 operational status : up ipv4 addresses : 10.0.0.1/24 (static) ipv6 addresses : fd00::1/64 (static) peers : 2 Peer 1: public key : ROaZyvJc5DzA2XUAAeTj2YlwDsy2w0lr3t+rWj2imAk= status : UP endpoint : 192.168.10.2:51821 latest handshake : 2025-12-09T22:51:38+00:00 transfer tx : 1412 bytes transfer rx : 1324 bytes Peer 2: public key : Om9CPLYdK3l93GauKrq5WXo/gbcD+1CeqFpobRLLkB4= status : UP endpoint : 2001:db8:3c4d:20::2:51822 latest handshake : 2025-12-09T22:51:38+00:00 transfer tx : 1812 bytes transfer rx : 428 bytes in-octets : 1752 out-octets : 3224 admin@server:/>
2026-07-31 04:53:01 +02:00 · 2026-01-09 11:11:32 +01:00
parent 8e9fd27e24
commit 7e5da21cf6
50 changed files with 6193 additions and 16 deletions
@@ -1114,6 +1114,7 @@ class Iface:
        self.gre = self.data.get('infix-interfaces:gre')
        self.vxlan = self.data.get('infix-interfaces:vxlan')
        self.wifi = self.data.get('infix-interfaces:wifi')
+        self.wireguard = self.data.get('infix-interfaces:wireguard')

        if self.data.get('infix-interfaces:vlan'):
            self.lower_if = self.data.get('infix-interfaces:vlan', None).get('lower-layer-if',None)
@@ -1148,6 +1149,9 @@ class Iface:
    def is_gretap(self):
        return self.data['type'] == "infix-if-type:gretap"

+    def is_wireguard(self):
+        return self.data['type'] == "infix-if-type:wireguard"
+
    def oper(self, detail=False):
        """Remap in brief overview to fit column widths."""
        if not detail and self.oper_status == "lower-layer-down":
@@ -1219,6 +1223,23 @@ class Iface:
        row = self._pr_proto_common("vxlan", True, pipe);
        print(row)

+    def pr_proto_wireguard(self, pipe=''):
+        row = self._pr_proto_common("wireguard", False, pipe)
+
+        if self.wireguard:
+            peer_status = self.wireguard.get('peer-status', {})
+            peers = peer_status.get('peer', [])
+            total_peers = len(peers)
+            up_peers = sum(1 for p in peers if p.get('connection-status') == 'up')
+
+            if total_peers > 0:
+                row += f"{total_peers} peer"
+                if total_peers != 1:
+                    row += "s"
+                row += f" ({up_peers} up)"
+
+        print(row)
+
    def pr_proto_loopack(self, pipe=''):
        row = self._pr_proto_common("loopback", False, pipe);
        print(row)
@@ -1471,6 +1492,12 @@ class Iface:
        self.pr_proto_ipv4()
        self.pr_proto_ipv6()

+    def pr_wireguard(self):
+        self.pr_name(pipe="")
+        self.pr_proto_wireguard()
+        self.pr_proto_ipv4()
+        self.pr_proto_ipv6()
+
    def pr_wifi(self):
        self.pr_name(pipe="")
        self.pr_proto_wifi()
@@ -1598,9 +1625,6 @@ class Iface:
            print(f"{'in-octets':<{20}}: {self.in_octets}")
            print(f"{'out-octets':<{20}}: {self.out_octets}")

-        frame = get_json_data([], self.data,'ieee802-ethernet-interface:ethernet',
-                              'statistics', 'frame')
-
        if self.wifi:
            # Detect mode: AP has "stations", Station has "rssi" or "scan-results"
            ap = self.wifi.get('access-point')
@@ -1635,6 +1659,46 @@ class Iface:
            print(f"{'remote address':<{20}}: {self.vxlan['remote']}")
            print(f"{'VxLAN id':<{20}}: {self.vxlan['vni']}")

+        if self.wireguard:
+            peer_status = self.wireguard.get('peer-status', {})
+            peers = peer_status.get('peer', [])
+            if peers:
+                print(f"{'peers':<{20}}: {len(peers)}")
+                for idx, peer in enumerate(peers, 1):
+                    print(f"\n  Peer {idx}:")
+
+                    # Public key (always 44 chars: 43 + '=')
+                    if pubkey := peer.get('public-key'):
+                        print(f"    {'public key':<{18}}: {pubkey}")
+
+                    # Connection status with color
+                    status = peer.get('connection-status', 'unknown')
+                    if status == 'up':
+                        status_str = Decore.green(status.upper())
+                    else:
+                        status_str = Decore.red(status.upper())
+                    print(f"    {'status':<{18}}: {status_str}")
+
+                    # Endpoint information
+                    if endpoint := peer.get('endpoint-address'):
+                        port = peer.get('endpoint-port', '')
+                        endpoint_str = f"{endpoint}:{port}" if port else endpoint
+                        print(f"    {'endpoint':<{18}}: {endpoint_str}")
+
+                    # Latest handshake
+                    if handshake := peer.get('latest-handshake'):
+                        print(f"    {'latest handshake':<{18}}: {handshake}")
+
+                    # Transfer statistics
+                    if transfer := peer.get('transfer'):
+                        tx = transfer.get('tx-bytes', '0')
+                        rx = transfer.get('rx-bytes', '0')
+                        print(f"    {'transfer tx':<{18}}: {tx} bytes")
+                        print(f"    {'transfer rx':<{18}}: {rx} bytes")
+
+
+        frame = get_json_data([], self.data,'ieee802-ethernet-interface:ethernet',
+                              'statistics', 'frame')
        if frame:
            print("")
            for key, val in frame.items():
@@ -1799,6 +1863,10 @@ def pr_interface_list(json):
            iface.pr_vxlan()
            continue

+        if iface.is_wireguard():
+            iface.pr_wireguard()
+            continue
+
        if iface.is_wifi():
            iface.pr_wifi()
            continue
@@ -9,6 +9,7 @@ from . import tun
 from . import veth
 from . import vlan
 from . import wifi
+from . import wireguard


 def statistics(iplink):
@@ -27,6 +28,7 @@ def statistics(iplink):

 def iplink2yang_type(iplink):
    ifname=iplink["ifname"]
+
    match iplink["link_type"]:
        case "loopback":
            return "infix-if-type:loopback"
@@ -36,6 +38,8 @@ def iplink2yang_type(iplink):
            data = HOST.run(tuple(f"ls /sys/class/net/{ifname}/wireless/".split()), default="no")
            if data != "no":
                return "infix-if-type:wifi"
+        case "none":
+            pass # WireGuard interfaces is for some reason link_type none
        case _:
            return "infix-if-type:other"

@@ -54,6 +58,8 @@ def iplink2yang_type(iplink):
            return "infix-if-type:veth"
        case "vlan":
            return "infix-if-type:vlan"
+        case "wireguard":
+            return "infix-if-type:wireguard"

    return "infix-if-type:ethernet"

@@ -141,6 +147,9 @@ def interface(iplink, ipaddr):
        case "infix-if-type:wifi":
            if w := wifi.wifi(iplink["ifname"]):
                interface["infix-interfaces:wifi"] = w
+        case "infix-if-type:wireguard":
+            if wg := wireguard.wireguard(iplink):
+                interface["infix-interfaces:wireguard"] = wg

    match iplink2yang_lower(iplink):
        case "infix-interfaces:bridge-port":
@@ -0,0 +1,127 @@
+import subprocess
+import json
+from datetime import datetime, timezone
+
+from ..host import HOST
+
+
+def _parse_wg_show(ifname):
+    """Parse `wg show <ifname> dump` output into structured data"""
+    try:
+        result = HOST.run(("wg", "show", ifname, "dump"), default="")
+        if not result:
+            return None
+
+        lines = result.strip().split('\n')
+        if len(lines) < 2:  # Need at least interface line + one peer
+            return None
+
+        peers = []
+        # Skip first line (interface info), process peer lines
+        for line in lines[1:]:
+            parts = line.split('\t')
+            if len(parts) < 8:
+                continue
+
+            public_key, preshared_key, endpoint, allowed_ips, \
+                latest_handshake, rx_bytes, tx_bytes, persistent_keepalive = parts
+
+            peer = {
+                "public_key": public_key,
+                "endpoint": endpoint if endpoint != "(none)" else None,
+                "allowed_ips": allowed_ips.split(',') if allowed_ips else [],
+                "latest_handshake": int(latest_handshake) if latest_handshake != "0" else None,
+                "rx_bytes": int(rx_bytes),
+                "tx_bytes": int(tx_bytes),
+            }
+            peers.append(peer)
+
+        return peers
+    except Exception:
+        return None
+
+
+def _format_timestamp(epoch_seconds):
+    """Convert Unix timestamp to YANG date-and-time format"""
+    if not epoch_seconds:
+        return None
+    dt = datetime.fromtimestamp(epoch_seconds, tz=timezone.utc)
+    # YANG date-and-time requires timezone with colon: +00:00 not +0000
+    timestamp = dt.strftime("%Y-%m-%dT%H:%M:%S%z")
+    # Insert colon in timezone offset: +0000 -> +00:00
+    return timestamp[:-2] + ':' + timestamp[-2:]
+
+
+def _parse_endpoint(endpoint_str):
+    """Parse endpoint string like '192.168.1.1:51820' or '[2001:db8::1]:51820'"""
+    if not endpoint_str or endpoint_str == "(none)":
+        return None, None
+
+    # Handle IPv6 with brackets
+    if endpoint_str.startswith('['):
+        addr_end = endpoint_str.find(']')
+        if addr_end == -1:
+            return None, None
+        addr = endpoint_str[1:addr_end]
+        port_part = endpoint_str[addr_end+1:]
+        port = int(port_part.lstrip(':')) if ':' in port_part else None
+        return addr, port
+
+    # Handle IPv4
+    parts = endpoint_str.rsplit(':', 1)
+    if len(parts) == 2:
+        return parts[0], int(parts[1])
+    return parts[0], None
+
+
+def _connection_status(latest_handshake_epoch):
+    """Determine connection status based on handshake time"""
+    if not latest_handshake_epoch:
+        return "down"
+
+    # Consider connection up if handshake within last 3 minutes
+    age = datetime.now(timezone.utc).timestamp() - latest_handshake_epoch
+    return "up" if age < 180 else "down"
+
+
+def wireguard(iplink):
+    """Get WireGuard operational state data"""
+    ifname = iplink.get("ifname")
+    if not ifname:
+        return None
+
+    peers_data = _parse_wg_show(ifname)
+    if not peers_data:
+        return None
+
+    peers = []
+    for peer_data in peers_data:
+        peer = {
+            "public-key": peer_data["public_key"]
+        }
+
+        # Connection status (always include)
+        if peer_data["latest_handshake"]:
+            peer["latest-handshake"] = _format_timestamp(peer_data["latest_handshake"])
+            peer["connection-status"] = _connection_status(peer_data["latest_handshake"])
+        else:
+            peer["connection-status"] = "down"
+
+        # Parse endpoint
+        if peer_data["endpoint"]:
+            addr, port = _parse_endpoint(peer_data["endpoint"])
+            if addr:
+                peer["endpoint-address"] = addr
+            if port:
+                peer["endpoint-port"] = port
+
+        # Transfer statistics
+        if peer_data["tx_bytes"] or peer_data["rx_bytes"]:
+            peer["transfer"] = {
+                "tx-bytes": str(peer_data["tx_bytes"]),
+                "rx-bytes": str(peer_data["rx_bytes"]),
+            }
+
+        peers.append(peer)
+
+    return {"peer-status": {"peer": peers}} if peers else None