Test program to print out WireGuard interface config using Python pyroute2 netlink library

pyroute_netlink_wireguard.py

#!/usr/bin/env python3
"""
This is free and unencumbered software released into the public domain.

Anyone is free to copy, modify, publish, use, compile, sell, or
distribute this software, either in source code form or as a compiled
binary, for any purpose, commercial or non-commercial, and by any
means.

In jurisdictions that recognize copyright laws, the author or authors
of this software dedicate any and all copyright interest in the
software to the public domain. We make this dedication for the benefit
of the public at large and to the detriment of our heirs and
successors. We intend this dedication to be an overt act of
relinquishment in perpetuity of all present and future rights to this
software under copyright law.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
OTHER DEALINGS IN THE SOFTWARE.

For more information, please refer to <http://unlicense.org>
"""

from pprint import pprint

import logging
logging.basicConfig(level=logging.DEBUG)

import struct
from socket import AF_INET, AF_INET6
from socket import inet_pton, inet_ntop

import datetime
import base64

from pyroute2.netlink import NLM_F_REQUEST, NLM_F_DUMP, NLM_F_ACK
from pyroute2.netlink import nla, nla_base
from pyroute2.netlink import genlmsg
from pyroute2.netlink.generic import GenericNetlinkSocket

WG_GENL_NAME = "wireguard"
WG_GENL_VERSION = 1

WG_KEY_LEN = 32

WG_CMD_GET_DEVICE = 0
WG_CMD_SET_DEVICE = 1


class wireguardcmd(genlmsg):
    prefix = 'WGDEVICE_A_'
    nla_map = (
        ('WGDEVICE_A_UNSPEC', 'none'),
        ('WGDEVICE_A_IFINDEX', 'uint32'),
        ('WGDEVICE_A_IFNAME', 'asciiz'),
        ('WGDEVICE_A_PRIVATE_KEY', 'cdata'),
        ('WGDEVICE_A_PUBLIC_KEY', 'cdata'),
        ('WGDEVICE_A_FLAGS', 'uint32'),
        ('WGDEVICE_A_LISTEN_PORT', 'uint16'),
        ('WGDEVICE_A_FWMARK', 'uint32'),
        ('WGDEVICE_A_PEERS', '*wgpeer')
    )

    class wgpeer(nla):
        prefix = 'WGPEER_A_'
        nla_map = (
            ('WGPEER_A_UNSPEC', 'none'),
            ('WGPEER_A_PUBLIC_KEY', 'cdata'),
            ('WGPEER_A_PRESHARED_KEY', 'cdata'),
            ('WGPEER_A_FLAGS', 'uint32'),
            ('WGPEER_A_ENDPOINT', 'sockaddr'),
            ('WGPEER_A_PERSISTENT_KEEPALIVE_INTERVAL', 'uint16'),
            ('WGPEER_A_LAST_HANDSHAKE_TIME', 'timespec'),
            ('WGPEER_A_RX_BYTES', 'uint64'),
            ('WGPEER_A_TX_BYTES', 'uint64'),
            ('WGPEER_A_ALLOWEDIPS', '*wgallowedip')
        )

        class sockaddr(nla_base):
            """

            # IPv4
            struct sockaddr_in {
               uint16_t      sin_family; /* address family: AF_INET */
               uint16_t      sin_port;   /* port in network byte order */
               uint32_t      sin_addr;   /* internet address */
            };

            # IPv6
            struct sockaddr_in6 {
               uint16_t        sin6_family;   /* AF_INET6 */
               uint16_t        sin6_port;     /* port number */
               uint32_t        sin6_flowinfo; /* IPv6 flow information */
               uint8_t         sin6_addr[16]; /* IPv6 address */
               uint32_t        sin6_scope_id; /* Scope ID (new in 2.4) */
            };
            """

            fields = [('value', 's')]

            def encode(self):
                nla_base.encode(self)
                #print(self.value)

            def decode(self):
                nla_base.decode(self)
                family = struct.unpack('H', self['value'][:2])[0]
                if family == AF_INET:
                    port, host = struct.unpack('!H4s', self['value'][2:8])
                    self.value = (inet_ntop(family, host), port)
                elif family == AF_INET6:
                    port, flowinfo, host, scopeid = struct.unpack('!HI16sI', self['value'][2:28])
                    self.value = (inet_ntop(family, host), port, flowinfo, scopeid)

        class timespec(nla):
            fields = [('value', 's')]

            def decode(self):
                nla_base.decode(self)
                sec, nsec = struct.unpack('ll', self['value'])
                #FIXME: don't drop nanoseconds
                self.value = datetime.datetime.fromtimestamp(sec)

        class wgallowedip(nla):
            prefix = 'WGALLOWEDIP_A_'
            nla_map = (
                ('WGALLOWEDIP_A_UNSPEC', 'none'),
                ('WGALLOWEDIP_A_FAMILY', 'uint16'),
                ('WGALLOWEDIP_A_IPADDR', 'ipaddr'),
                ('WGALLOWEDIP_A_CIDR_MASK', 'uint8')
            )


class WireGuard(GenericNetlinkSocket):

    def __init__(self, *args, **kwargs):
        GenericNetlinkSocket.__init__(self, *args, **kwargs)
        GenericNetlinkSocket.bind(self, WG_GENL_NAME, wireguardcmd, *args, **kwargs)

    def get_device_dump(self, ifname=None, ifindex=None):
        """
        Return current configuration of wireguard device
        by name or by interface index
        """

        msg = wireguardcmd()
        msg['cmd'] = WG_CMD_GET_DEVICE
        msg['version'] = WG_GENL_VERSION
        if ifname != None:
            msg['attrs'].append(['WGDEVICE_A_IFNAME', ifname])
        elif ifindex != None:
            msg['attrs'].append(['WGDEVICE_A_IFINDEX', ifindex])
        else:
            raise ValueError("ifname or ifindex are unset")
        return self.nlm_request(msg,
                                msg_type=self.prid,
                                msg_flags=NLM_F_REQUEST | NLM_F_ACK | NLM_F_DUMP)

    def get_device_dict(self, *args, **kwargs):
        msg = self.get_device_dump(*args, **kwargs)
        ret = {}
        for x in msg:
            dev = ret[x.get_attr('WGDEVICE_A_IFNAME')] = {}
            for key, val in x['attrs']:
                key = key.replace(x.prefix, '', 1).lower()
                if key == 'peers':
                    peers = []
                    for peer in val:
                        p = {}
                        peers.append(p)
                        for peer_key, peer_val in peer['attrs']:
                            peer_key = peer_key.replace(peer.prefix, '', 1).lower()
                            if peer_key == 'allowedips':
                                allowedips = []
                                for ips in peer_val:
                                    ip = ips.get_attr("WGALLOWEDIP_A_IPADDR")
                                    mask = ips.get_attr("WGALLOWEDIP_A_CIDR_MASK")
                                    allowedips.append("{}/{}".format(ip, mask))
                                peer_val = allowedips
                            p[peer_key] = peer_val
                    val = peers
                dev[key] = val
        return ret

    def set_device(self, ifname=None, ifindex=None, *, listen_port=None, fwmark=None, private_key=None):
        msg = wireguardcmd()
        msg['cmd'] = WG_CMD_SET_DEVICE
        msg['version'] = WG_GENL_VERSION

        if ifname != None:
            msg['attrs'].append(['WGDEVICE_A_IFNAME', ifname])
        elif ifindex != None:
            msg['attrs'].append(['WGDEVICE_A_IFINDEX', ifindex])
        else:
            raise ValueError("ifname or ifindex are unset")

        if listen_port != None:
            msg['attrs'].append(['WGDEVICE_A_LISTEN_PORT', listen_port])
        if fwmark != None:
            msg['attrs'].append(['WGDEVICE_A_FWMARK', fwmark])
        if private_key != None:
            if type(private_key) != bytes:
                raise ValueError("private_key has to be bytes")
            elif len(private_key) != WG_KEY_LEN:
                raise ValueError("private_key length has to be {}".format(WG_KEY_LEN))
            msg['attrs'].append(['WGDEVICE_A_PRIVATE_KEY', private_key])

        return self.put(msg,
                        msg_type=self.prid,
                        msg_flags=NLM_F_REQUEST)



wg = WireGuard()
pprint(wg.get_device_dict(ifname='wgtest'))
wg.set_device('wgtest', listen_port=555)

Hey, nice work, what licence is this under?

Public domain, do what ever you want with it.