Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:26
📄 __init__.py # -- coding: utf-8 -- ''' Generic netlink =============== Describe ''' import errno import logging from pyroute2.netlink import ( CTRL_CMD_GETFAMILY, CTRL_CMD_GETPOLICY, GENL_ID_CTRL, NETLINK_ADD_MEMBERSHIP, NETLINK_DROP_MEMBERSHIP, NLM_F_ACK, NLM_F_DUMP, NLM_F_REQUEST, SOL_NETLINK, ctrlmsg, ) from pyroute2.netlink.nlsocket import NetlinkSocket class GenericNetlinkSocket(NetlinkSocket): ''' Low-level socket interface. Provides all the usual socket does, can be used in poll/select, doesn't create any implicit threads. ''' mcast_groups = {} module_err_message = None module_err_level = 'error' _prid = None @property def prid(self): if self._prid is None: raise RuntimeError( 'generic netlink protocol id is not obtained' ' yet, run bind() before placing any requests' ) else: return self._prid def bind(self, proto, msg_class, groups=0, pid=None, kwarg): ''' Bind the socket and performs generic netlink proto lookup. The `proto` parameter is a string, like "TASKSTATS", `msg_class` is a class to parse messages with. ''' NetlinkSocket.bind(self, groups, pid, kwarg) self.marshal.msg_map[GENL_ID_CTRL] = ctrlmsg msg = self.discovery(proto) self._prid = msg.get_attr('CTRL_ATTR_FAMILY_ID') self.mcast_groups = dict( [ ( x.get_attr('CTRL_ATTR_MCAST_GRP_NAME'), x.get_attr('CTRL_ATTR_MCAST_GRP_ID'), ) for x in msg.get_attr('CTRL_ATTR_MCAST_GROUPS', []) ] ) self.marshal.msg_map[self.prid] = msg_class def add_membership(self, group): self.setsockopt( SOL_NETLINK, NETLINK_ADD_MEMBERSHIP, self.mcast_groups[group] ) def drop_membership(self, group): self.setsockopt( SOL_NETLINK, NETLINK_DROP_MEMBERSHIP, self.mcast_groups[group] ) def discovery(self, proto): ''' Resolve generic netlink protocol -- takes a string as the only parameter, return protocol description ''' msg = ctrlmsg() msg['cmd'] = CTRL_CMD_GETFAMILY msg['version'] = 1 msg['attrs'].append(['CTRL_ATTR_FAMILY_NAME', proto]) msg['header']['type'] = GENL_ID_CTRL msg['header']['flags'] = NLM_F_REQUEST msg['header']['pid'] = self.pid msg.encode() self.sendto(msg.data, (0, 0)) msg = self.get()[0] err = msg['header'].get('error', None) if err is not None: if hasattr(err, 'code') and err.code == errno.ENOENT: err.extra_code = errno.ENOTSUP logger = getattr(logging, self.module_err_level) logger('Generic netlink protocol %s not found' % proto) logger('Please check if the protocol module is loaded') if self.module_err_message is not None: logger(self.module_err_message) raise err return msg def policy(self, proto): ''' Extract policy information for a generic netlink protocol -- takes a string as the only parameter, return protocol policy ''' self.marshal.msg_map[GENL_ID_CTRL] = ctrlmsg msg = ctrlmsg() msg['cmd'] = CTRL_CMD_GETPOLICY msg['attrs'].append(['CTRL_ATTR_FAMILY_NAME', proto]) return self.nlm_request( msg, msg_type=GENL_ID_CTRL, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP \| NLM_F_ACK, ) def get(self, argv, kwarg): return tuple(super().get(argv, *kwarg)) def nlm_request(self, argv, *kwarg): return tuple(super().nlm_request(argv, **kwarg))	3.82 KB	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 ethtool.py from pyroute2.netlink import ( NLA_F_NESTED, NLM_F_ACK, NLM_F_REQUEST, genlmsg, nla, ) from pyroute2.netlink.exceptions import NetlinkError from pyroute2.netlink.generic import GenericNetlinkSocket ETHTOOL_GENL_NAME = "ethtool" ETHTOOL_GENL_VERSION = 1 ETHTOOL_MSG_USER_NONE = 0 ETHTOOL_MSG_STRSET_GET = 1 ETHTOOL_MSG_LINKINFO_GET = 2 ETHTOOL_MSG_LINKINFO_SET = 3 ETHTOOL_MSG_LINKMODES_GET = 4 ETHTOOL_MSG_LINKMODES_SET = 5 ETHTOOL_MSG_LINKSTATE_GET = 6 ETHTOOL_MSG_DEBUG_GET = 7 ETHTOOL_MSG_DEBUG_SET = 8 ETHTOOL_MSG_WOL_GET = 9 ETHTOOL_MSG_WOL_SET = 10 ETHTOOL_MSG_FEATURES_GET = 11 ETHTOOL_MSG_FEATURES_SET = 12 ETHTOOL_MSG_PRIVFLAGS_GET = 13 ETHTOOL_MSG_PRIVFLAGS_SET = 14 ETHTOOL_MSG_RINGS_GET = 15 ETHTOOL_MSG_RINGS_SET = 16 class ethtoolheader(nla): nla_flags = NLA_F_NESTED nla_map = ( ('ETHTOOL_A_HEADER_UNSPEC', 'none'), ('ETHTOOL_A_HEADER_DEV_INDEX', 'uint32'), ('ETHTOOL_A_HEADER_DEV_NAME', 'asciiz'), ('ETHTOOL_A_HEADER_FLAGS', 'uint32'), ) class ethtoolbitset(nla): nla_flags = NLA_F_NESTED nla_map = ( ("ETHTOOL_A_BITSET_UNSPEC", 'none'), ("ETHTOOL_A_BITSET_NOMASK", "flag"), ("ETHTOOL_A_BITSET_SIZE", 'uint32'), ("ETHTOOL_A_BITSET_BITS", 'bitset_bits'), ("ETHTOOL_A_BITSET_VALUE", 'hex'), ("ETHTOOL_A_BITSET_MASK", 'hex'), ) class bitset_bits(nla): nla_flags = NLA_F_NESTED nla_map = ( ('ETHTOOL_A_BITSET_BIT_UNSPEC', 'none'), ('ETHTOOL_A_BITSET_BITS_BIT', 'bitset_bits_bit'), ) class bitset_bits_bit(nla): nla_flags = NLA_F_NESTED nla_map = ( ('ETHTOOL_A_BITSET_BIT_UNSPEC', 'none'), ('ETHTOOL_A_BITSET_BIT_INDEX', 'uint32'), ('ETHTOOL_A_BITSET_BIT_NAME', 'asciiz'), ('ETHTOOL_A_BITSET_BIT_VALUE', 'flag'), ) class ethtool_strset_msg(genlmsg): nla_map = ( ('ETHTOOL_A_STRSET_UNSPEC', 'none'), ('ETHTOOL_A_STRSET_HEADER', 'ethtoolheader'), ('ETHTOOL_A_STRSET_STRINGSETS', 'strings_sets'), ('ETHTOOL_A_STRSET_COUNTS_ONLY', 'flag'), ) ethtoolheader = ethtoolheader class strings_sets(nla): nla_flags = NLA_F_NESTED nla_map = ( ('ETHTOOL_A_STRINGSETS_UNSPEC', 'none'), ('ETHTOOL_A_STRINGSETS_STRINGSET', 'string_set'), ) class string_set(nla): nla_flags = NLA_F_NESTED nla_map = ( ('ETHTOOL_A_STRINGSET_UNSPEC', 'none'), ('ETHTOOL_A_STRINGSET_ID', 'uint32'), ('ETHTOOL_A_STRINGSET_COUNT', 'uint32'), ('ETHTOOL_A_STRINGSET_STRINGS', 'stringset_strings'), ) class stringset_strings(nla): nla_flags = NLA_F_NESTED nla_map = ( ('ETHTOOL_A_STRINGS_UNSPEC', 'none'), ('ETHTOOL_A_STRINGS_STRING', 'strings_string'), ) class strings_string(nla): nla_flags = NLA_F_NESTED nla_map = ( ('ETHTOOL_A_STRING_UNSPEC', 'none'), ('ETHTOOL_A_STRING_INDEX', 'uint32'), ('ETHTOOL_A_STRING_VALUE', 'asciiz'), ) class ethtool_linkinfo_msg(genlmsg): nla_map = ( ('ETHTOOL_A_LINKINFO_UNSPEC', 'none'), ('ETHTOOL_A_LINKINFO_HEADER', 'ethtoolheader'), ('ETHTOOL_A_LINKINFO_PORT', 'uint8'), ('ETHTOOL_A_LINKINFO_PHYADDR', 'uint8'), ('ETHTOOL_A_LINKINFO_TP_MDIX', 'uint8'), ('ETHTOOL_A_LINKINFO_TP_MDIX_CTR', 'uint8'), ('ETHTOOL_A_LINKINFO_TRANSCEIVER', 'uint8'), ) ethtoolheader = ethtoolheader class ethtool_linkmode_msg(genlmsg): nla_map = ( ('ETHTOOL_A_LINKMODES_UNSPEC', 'none'), ('ETHTOOL_A_LINKMODES_HEADER', 'ethtoolheader'), ('ETHTOOL_A_LINKMODES_AUTONEG', 'uint8'), ('ETHTOOL_A_LINKMODES_OURS', 'ethtoolbitset'), ('ETHTOOL_A_LINKMODES_PEER', 'ethtoolbitset'), ('ETHTOOL_A_LINKMODES_SPEED', 'uint32'), ('ETHTOOL_A_LINKMODES_DUPLEX', 'uint8'), ) ethtoolheader = ethtoolheader ethtoolbitset = ethtoolbitset class ethtool_linkstate_msg(genlmsg): nla_map = ( ('ETHTOOL_A_LINKSTATE_UNSPEC', 'none'), ('ETHTOOL_A_LINKSTATE_HEADER', 'ethtoolheader'), ('ETHTOOL_A_LINKSTATE_LINK', 'uint8'), ) ethtoolheader = ethtoolheader class ethtool_wol_msg(genlmsg): nla_map = ( ('ETHTOOL_A_WOL_UNSPE', 'none'), ('ETHTOOL_A_WOL_HEADER', 'ethtoolheader'), ('ETHTOOL_A_WOL_MODES', 'ethtoolbitset'), ('ETHTOOL_A_WOL_SOPASS', 'hex'), ) ethtoolheader = ethtoolheader ethtoolbitset = ethtoolbitset class ethtool_rings_msg(genlmsg): nla_map = ( ('ETHTOOL_A_RINGS_UNSPEC', 'none'), ('ETHTOOL_A_RINGS_HEADER', 'ethtoolheader'), ('ETHTOOL_A_RINGS_RX_MAX', 'uint32'), ('ETHTOOL_A_RINGS_RX_MINI_MAX', 'uint32'), ('ETHTOOL_A_RINGS_RX_JUMBO_MAX', 'uint32'), ('ETHTOOL_A_RINGS_TX_MAX', 'uint32'), ('ETHTOOL_A_RINGS_RX', 'uint32'), ('ETHTOOL_A_RINGS_RX_MINI', 'uint32'), ('ETHTOOL_A_RINGS_RX_JUMBO', 'uint32'), ('ETHTOOL_A_RINGS_TX', 'uint32'), ('ETHTOOL_A_RINGS_RX_BUF_LEN', 'uint32'), ('ETHTOOL_A_RINGS_TCP_DATA_SPLIT', 'uint8'), ('ETHTOOL_A_RINGS_CQE_SIZE', 'uint32'), ('ETHTOOL_A_RINGS_TX_PUSH', 'uint8'), ('ETHTOOL_A_RINGS_RX_PUSH', 'uint8'), ('ETHTOOL_A_RINGS_TX_PUSH_BUF_LEN', 'uint32'), ('ETHTOOL_A_RINGS_TX_PUSH_BUF_LEN_MAX', 'uint32'), ) ethtoolheader = ethtoolheader class NlEthtool(GenericNetlinkSocket): def _do_request(self, msg, msg_flags=NLM_F_REQUEST): return self.nlm_request(msg, msg_type=self.prid, msg_flags=msg_flags) def is_nlethtool_in_kernel(self): try: self.bind(ETHTOOL_GENL_NAME, ethtool_linkinfo_msg) except NetlinkError: return False return True def _get_dev_header(self, ifname=None, ifindex=None): if ifindex is not None: return {'attrs': [['ETHTOOL_A_HEADER_DEV_INDEX', ifindex]]} elif ifname is not None: return {'attrs': [['ETHTOOL_A_HEADER_DEV_NAME', ifname]]} else: raise ValueError("Need ifname or ifindex") def get_linkinfo(self, ifname=None, ifindex=None): msg = ethtool_linkinfo_msg() msg["cmd"] = ETHTOOL_MSG_LINKINFO_GET msg['version'] = ETHTOOL_GENL_VERSION msg["attrs"].append( ( 'ETHTOOL_A_LINKINFO_HEADER', self._get_dev_header(ifname, ifindex), ) ) self.bind(ETHTOOL_GENL_NAME, ethtool_linkinfo_msg) return self._do_request(msg) def get_linkmode(self, ifname=None, ifindex=None): msg = ethtool_linkmode_msg() msg["cmd"] = ETHTOOL_MSG_LINKMODES_GET msg['version'] = ETHTOOL_GENL_VERSION msg["attrs"].append( ( 'ETHTOOL_A_LINKMODES_HEADER', self._get_dev_header(ifname, ifindex), ) ) self.bind(ETHTOOL_GENL_NAME, ethtool_linkmode_msg) return self._do_request(msg) def get_stringset(self, ifname=None, ifindex=None): msg = ethtool_strset_msg() msg["cmd"] = ETHTOOL_MSG_STRSET_GET msg['version'] = ETHTOOL_GENL_VERSION msg["attrs"].append( ('ETHTOOL_A_STRSET_HEADER', self._get_dev_header(ifname, ifindex)) ) self.bind(ETHTOOL_GENL_NAME, ethtool_strset_msg) return self._do_request(msg) def get_linkstate(self, ifname=None, ifindex=None): msg = ethtool_linkstate_msg() msg["cmd"] = ETHTOOL_MSG_LINKSTATE_GET msg['version'] = ETHTOOL_GENL_VERSION msg["attrs"].append( ( 'ETHTOOL_A_LINKSTATE_HEADER', self._get_dev_header(ifname, ifindex), ) ) self.bind(ETHTOOL_GENL_NAME, ethtool_linkstate_msg) return self._do_request(msg) def get_wol(self, ifname=None, ifindex=None): msg = ethtool_wol_msg() msg["cmd"] = ETHTOOL_MSG_WOL_GET msg['version'] = ETHTOOL_GENL_VERSION msg["attrs"].append( ('ETHTOOL_A_WOL_HEADER', self._get_dev_header(ifname, ifindex)) ) self.bind(ETHTOOL_GENL_NAME, ethtool_wol_msg) return self._do_request(msg) def get_rings(self, ifname=None, ifindex=None): msg = ethtool_rings_msg() msg["cmd"] = ETHTOOL_MSG_RINGS_GET msg["version"] = ETHTOOL_GENL_VERSION msg["attrs"].append( ('ETHTOOL_A_RINGS_HEADER', self._get_dev_header(ifname, ifindex)) ) self.bind(ETHTOOL_GENL_NAME, ethtool_rings_msg) return self._do_request(msg) def set_rings(self, rings, ifname=None, ifindex=None): rings["cmd"] = ETHTOOL_MSG_RINGS_SET rings["version"] = ETHTOOL_GENL_VERSION rings["attrs"].append( ('ETHTOOL_A_RINGS_HEADER', self._get_dev_header(ifname, ifindex)) ) self.bind(ETHTOOL_GENL_NAME, ethtool_rings_msg) return self._do_request(rings, msg_flags=NLM_F_REQUEST \| NLM_F_ACK)	9.17 KB	2025-05-25 12:26
📄 ipvs.py from pyroute2.netlink import genlmsg, nla from pyroute2.netlink.generic import GenericNetlinkSocket GENL_NAME = "IPVS" GENL_VERSION = 0x1 IPVS_CMD_UNSPEC = 0 IPVS_CMD_NEW_SERVICE = 1 IPVS_CMD_SET_SERVICE = 2 IPVS_CMD_DEL_SERVICE = 3 IPVS_CMD_GET_SERVICE = 4 IPVS_CMD_NEW_DEST = 5 IPVS_CMD_SET_DEST = 6 IPVS_CMD_DEL_DEST = 7 IPVS_CMD_GET_DEST = 8 IPVS_CMD_NEW_DAEMON = 9 IPVS_CMD_DEL_DAEMON = 10 IPVS_CMD_GET_DAEMON = 11 IPVS_CMD_SET_CONFIG = 12 IPVS_CMD_GET_CONFIG = 13 IPVS_CMD_SET_INFO = 14 IPVS_CMD_GET_INFO = 15 IPVS_CMD_ZERO = 16 IPVS_CMD_FLUSH = 17 class ipvsstats: class stats(nla): nla_map = ( ("IPVS_STATS_ATTR_UNSPEC", "none"), ("IPVS_STATS_ATTR_CONNS", "uint32"), ("IPVS_STATS_ATTR_INPKTS", "uint32"), ("IPVS_STATS_ATTR_OUTPKTS", "uint32"), ("IPVS_STATS_ATTR_INBYTES", "uint64"), ("IPVS_STATS_ATTR_OUTBYTES", "uint64"), ("IPVS_STATS_ATTR_CPS", "uint32"), ("IPVS_STATS_ATTR_INPPS", "uint32"), ("IPVS_STATS_ATTR_OUTPPS", "uint32"), ("IPVS_STATS_ATTR_INBPS", "uint32"), ("IPVS_STATS_ATTR_OUTBPS", "uint32"), ) class stats64(nla): nla_map = ( ("IPVS_STATS_ATTR_UNSPEC", "none"), ("IPVS_STATS_ATTR_CONNS", "uint64"), ("IPVS_STATS_ATTR_INPKTS", "uint64"), ("IPVS_STATS_ATTR_OUTPKTS", "uint64"), ("IPVS_STATS_ATTR_INBYTES", "uint64"), ("IPVS_STATS_ATTR_OUTBYTES", "uint64"), ("IPVS_STATS_ATTR_CPS", "uint64"), ("IPVS_STATS_ATTR_INPPS", "uint64"), ("IPVS_STATS_ATTR_OUTPPS", "uint64"), ("IPVS_STATS_ATTR_INBPS", "uint64"), ("IPVS_STATS_ATTR_OUTBPS", "uint64"), ) class ipvsmsg(genlmsg): prefix = "IPVS_CMD_ATTR_" nla_map = ( ("IPVS_CMD_ATTR_UNSPEC", "none"), ("IPVS_CMD_ATTR_SERVICE", "service"), ("IPVS_CMD_ATTR_DEST", "dest"), ("IPVS_CMD_ATTR_DAEMON", "hex"), ("IPVS_CMD_ATTR_TIMEOUT_TCP", "hex"), ("IPVS_CMD_ATTR_TIMEOUT_TCP_FIN", "hex"), ("IPVS_CMD_ATTR_TIMEOUT_UDP", "hex"), ) class service(nla, ipvsstats): prefix = "IPVS_SVC_ATTR_" nla_map = ( ("IPVS_SVC_ATTR_UNSPEC", "none"), ("IPVS_SVC_ATTR_AF", "uint16"), ("IPVS_SVC_ATTR_PROTOCOL", "uint16"), ("IPVS_SVC_ATTR_ADDR", "target(nla,IPVS_SVC_ATTR_AF)"), ("IPVS_SVC_ATTR_PORT", "be16"), ("IPVS_SVC_ATTR_FWMARK", "uint32"), ("IPVS_SVC_ATTR_SCHED_NAME", "asciiz"), ("IPVS_SVC_ATTR_FLAGS", "flags"), ("IPVS_SVC_ATTR_TIMEOUT", "uint32"), ("IPVS_SVC_ATTR_NETMASK", "ip4addr"), ("IPVS_SVC_ATTR_STATS", "stats"), ("IPVS_SVC_ATTR_PE_NAME", "asciiz"), ("IPVS_SVC_ATTR_STATS64", "stats64"), ) class flags(nla): fields = (("flags", "I"), ("mask", "I")) class dest(nla, ipvsstats): prefix = "IPVS_DEST_ATTR_" nla_map = ( ("IPVS_DEST_ATTR_UNSPEC", "none"), ("IPVS_DEST_ATTR_ADDR", "target(nla,IPVS_DEST_ATTR_ADDR_FAMILY)"), ("IPVS_DEST_ATTR_PORT", "be16"), ("IPVS_DEST_ATTR_FWD_METHOD", "uint32"), ("IPVS_DEST_ATTR_WEIGHT", "uint32"), ("IPVS_DEST_ATTR_U_THRESH", "uint32"), ("IPVS_DEST_ATTR_L_THRESH", "uint32"), ("IPVS_DEST_ATTR_ACTIVE_CONNS", "uint32"), ("IPVS_DEST_ATTR_INACT_CONNS", "uint32"), ("IPVS_DEST_ATTR_PERSIST_CONNS", "uint32"), ("IPVS_DEST_ATTR_STATS", "stats"), ("IPVS_DEST_ATTR_ADDR_FAMILY", "uint16"), ("IPVS_DEST_ATTR_STATS64", "stats64"), ("IPVS_DEST_ATTR_TUN_TYPE", "uint8"), ("IPVS_DEST_ATTR_TUN_PORT", "uint16"), ("IPVS_DEST_ATTR_TUN_FLAGS", "uint16"), ) class IPVSSocket(GenericNetlinkSocket): def __init__(self, argv, kwargs): super().__init__(argv, **kwargs) self.bind(GENL_NAME, ipvsmsg)	4 KB	2025-05-25 12:26
📄 l2tp.py from pyroute2.netlink import ( NLA_F_NESTED, NLM_F_ACK, NLM_F_DUMP, NLM_F_REQUEST, genlmsg, nla, ) from pyroute2.netlink.generic import GenericNetlinkSocket # Defines from uapi/linux/l2tp.h L2TP_GENL_NAME = "l2tp" L2TP_GENL_VERSION = 1 L2TP_CMD_NOOP = 0 L2TP_CMD_TUNNEL_CREATE = 1 L2TP_CMD_TUNNEL_DELETE = 2 L2TP_CMD_TUNNEL_MODIFY = 3 L2TP_CMD_TUNNEL_GET = 4 L2TP_CMD_SESSION_CREATE = 5 L2TP_CMD_SESSION_DELETE = 6 L2TP_CMD_SESSION_MODIFY = 7 L2TP_CMD_SESSION_GET = 8 # ATTR types defined for L2TP L2TP_ATTR_NONE = 0 L2TP_ATTR_PW_TYPE = 1 L2TP_ATTR_ENCAP_TYPE = 2 L2TP_ATTR_OFFSET = 3 L2TP_ATTR_DATA_SEQ = 4 L2TP_ATTR_L2SPEC_TYPE = 5 L2TP_ATTR_L2SPEC_LEN = 6 L2TP_ATTR_PROTO_VERSION = 7 L2TP_ATTR_IFNAME = 8 L2TP_ATTR_CONN_ID = 9 L2TP_ATTR_PEER_CONN_ID = 10 L2TP_ATTR_SESSION_ID = 11 L2TP_ATTR_PEER_SESSION_ID = 12 L2TP_ATTR_UDP_CSUM = 13 L2TP_ATTR_VLAN_ID = 14 L2TP_ATTR_COOKIE = 15 L2TP_ATTR_PEER_COOKIE = 16 L2TP_ATTR_DEBUG = 17 L2TP_ATTR_RECV_SEQ = 18 L2TP_ATTR_SEND_SEQ = 19 L2TP_ATTR_LNS_MODE = 20 L2TP_ATTR_USING_IPSEC = 21 L2TP_ATTR_RECV_TIMEOUT = 22 L2TP_ATTR_FD = 23 L2TP_ATTR_IP_SADDR = 24 L2TP_ATTR_IP_DADDR = 25 L2TP_ATTR_UDP_SPORT = 26 L2TP_ATTR_UDP_DPORT = 27 L2TP_ATTR_MTU = 28 L2TP_ATTR_MRU = 29 L2TP_ATTR_STATS = 30 L2TP_ATTR_IP6_SADDR = 31 L2TP_ATTR_IP6_DADDR = 32 L2TP_ATTR_UDP_ZERO_CSUM6_TX = 33 L2TP_ATTR_UDP_ZERO_CSUM6_RX = 34 L2TP_ATTR_PAD = 35 # Nested L2TP_ATTR_STATS L2TP_ATTR_STATS_NONE = 0 L2TP_ATTR_TX_PACKETS = 1 L2TP_ATTR_TX_BYTES = 2 L2TP_ATTR_TX_ERRORS = 3 L2TP_ATTR_RX_PACKETS = 4 L2TP_ATTR_RX_BYTES = 5 L2TP_ATTR_RX_SEQ_DISCARDS = 6 L2TP_ATTR_RX_OOS_PACKETS = 7 L2TP_ATTR_RX_ERRORS = 8 L2TP_ATTR_STATS_PAD = 9 L2TP_PWTYPE_NONE = 0x0000 L2TP_PWTYPE_ETH_VLAN = 0x0004 L2TP_PWTYPE_ETH = 0x0005 L2TP_PWTYPE_PPP = 0x0007 L2TP_PWTYPE_PPP_AC = 0x0008 L2TP_PWTYPE_IP = 0x000B L2TP_L2SPECTYPE_NONE = 0 L2TP_L2SPECTYPE_DEFAULT = 1 L2TP_ENCAPTYPE_UDP = 0 L2TP_ENCAPTYPE_IP = 1 class l2tpmsg(genlmsg): prefix = "L2TP_ATTR_" nla_map = ( ("L2TP_ATTR_NONE", "none"), ("L2TP_ATTR_PW_TYPE", "uint16"), ("L2TP_ATTR_ENCAP_TYPE", "uint16"), ("L2TP_ATTR_OFFSET", "uint16"), ("L2TP_ATTR_DATA_SEQ", "uint8"), ("L2TP_ATTR_L2SPEC_TYPE", "uint8"), ("L2TP_ATTR_L2SPEC_LEN", "uint8"), ("L2TP_ATTR_PROTO_VERSION", "uint8"), ("L2TP_ATTR_IFNAME", "asciiz"), ("L2TP_ATTR_CONN_ID", "uint32"), ("L2TP_ATTR_PEER_CONN_ID", "uint32"), ("L2TP_ATTR_SESSION_ID", "uint32"), ("L2TP_ATTR_PEER_SESSION_ID", "uint32"), ("L2TP_ATTR_UDP_CSUM", "uint8"), ("L2TP_ATTR_VLAN_ID", "uint16"), ("L2TP_ATTR_COOKIE", "hex"), ("L2TP_ATTR_PEER_COOKIE", "hex"), ("L2TP_ATTR_DEBUG", "uint32"), ("L2TP_ATTR_RECV_SEQ", "uint8"), ("L2TP_ATTR_SEND_SEQ", "uint8"), ("L2TP_ATTR_LNS_MODE", "uint8"), ("L2TP_ATTR_USING_IPSEC", "uint8"), ("L2TP_ATTR_RECV_TIMEOUT", "uint64"), ("L2TP_ATTR_FD", "uint32"), ("L2TP_ATTR_IP_SADDR", "ip4addr"), ("L2TP_ATTR_IP_DADDR", "ip4addr"), ("L2TP_ATTR_UDP_SPORT", "uint16"), ("L2TP_ATTR_UDP_DPORT", "uint16"), ("L2TP_ATTR_MTU", "uint16"), ("L2TP_ATTR_MRU", "uint16"), ("L2TP_ATTR_STATS", "l2tp_stats"), ("L2TP_ATTR_IP6_SADDR", "ip6addr"), ("L2TP_ATTR_IP6_DADDR", "ip6addr"), ("L2TP_ATTR_UDP_ZERO_CSUM6_TX", "flag"), ("L2TP_ATTR_UDP_ZERO_CSUM6_RX", "flag"), ("L2TP_ATTR_PAD", "none"), ) class l2tp_stats(nla): nla_flags = NLA_F_NESTED nla_map = ( ("L2TP_ATTR_STATS_NONE", "none"), ("L2TP_ATTR_TX_PACKETS", "uint64"), ("L2TP_ATTR_TX_BYTES", "uint64"), ("L2TP_ATTR_TX_ERRORS", "uint64"), ("L2TP_ATTR_RX_PACKETS", "uint64"), ("L2TP_ATTR_RX_BYTES", "uint64"), ("L2TP_ATTR_RX_SEQ_DISCARDS", "uint64"), ("L2TP_ATTR_RX_OOS_PACKETS", "uint64"), ("L2TP_ATTR_RX_ERRORS", "uint64"), ("L2TP_ATTR_STATS_PAD", "none"), ) class L2tp(GenericNetlinkSocket): def __init__(self, args, kwargs): GenericNetlinkSocket.__init__(self, args, **kwargs) self.bind(L2TP_GENL_NAME, l2tpmsg) def _do_request(self, msg, msg_flags=NLM_F_REQUEST \| NLM_F_ACK): return self.nlm_request(msg, msg_type=self.prid, msg_flags=msg_flags) def _send_tunnel( self, cmd, tunnel_id, peer_tunnel_id=None, protocol=3, remote=None, local=None, fd=None, encap="udp", udp_sport=None, udp_dport=None, udp_csum=None, udp6_csum_rx=None, udp6_csum_tx=None, debug=None, ): """ Send L2TP tunnel create or modify commands :param cmd: Netlink command to use :param tunnel_id: local tunnel id :param peer_tunnel_id: remote tunnel id :param protocol: L2TP version :param remote: IP address of the remote peer :param local: IP address of the local interface :param fd: file descriptor of socket to use :param encap: encapsulation type of the tunnel (udp, ip) :param udp_sport: UDP source port to be used for the tunnel :param udp_dport: UDP destination port to be used for the tunnel :param udp_csum: control if IPv4 UDP checksums should be calculated and checked :param udp6_csum_rx: control if IPv6 UDP rx checksums should be calculated :param udp6_csum_tx: control if IPv6 UDP tx checksums should be calculated :param debug: enable or disable debugging using kernel printk for the tunnel :return: Netlink response """ msg = l2tpmsg() msg["cmd"] = cmd msg["version"] = L2TP_GENL_VERSION msg["attrs"].append(["L2TP_ATTR_CONN_ID", tunnel_id]) if cmd == L2TP_CMD_TUNNEL_CREATE: msg["attrs"].append(["L2TP_ATTR_PEER_CONN_ID", peer_tunnel_id]) msg["attrs"].append(["L2TP_ATTR_PROTO_VERSION", protocol]) if encap == "ip": msg["attrs"].append(["L2TP_ATTR_ENCAP_TYPE", L2TP_ENCAPTYPE_IP]) elif encap == "udp": msg["attrs"].append(["L2TP_ATTR_ENCAP_TYPE", L2TP_ENCAPTYPE_UDP]) if fd: msg["attrs"].append(["L2TP_ATTR_FD", fd]) else: local_ip_version = 4 if local: if local.find(":") > -1: local_ip_version = 6 if local_ip_version == 6: msg["attrs"].append(["L2TP_ATTR_IP6_SADDR", local]) else: msg["attrs"].append(["L2TP_ATTR_IP_SADDR", local]) remote_ip_version = 4 if remote: if remote.find(":") > -1: remote_ip_version = 6 if remote_ip_version == 6: msg["attrs"].append(["L2TP_ATTR_IP6_DADDR", remote]) else: msg["attrs"].append(["L2TP_ATTR_IP_DADDR", remote]) if local and remote: if remote_ip_version != local_ip_version: raise ValueError( "Local and remote peer address version mismatch" ) if encap == "udp" and cmd == L2TP_CMD_TUNNEL_CREATE: if udp_sport: msg["attrs"].append(["L2TP_ATTR_UDP_SPORT", udp_sport]) if udp_dport: msg["attrs"].append(["L2TP_ATTR_UDP_DPORT", udp_dport]) if udp_csum: msg["attrs"].append(["L2TP_ATTR_UDP_CSUM", True]) if udp6_csum_rx: msg["attrs"].append(["L2TP_ATTR_UDP_ZERO_CSUM6_TX", True]) if udp6_csum_tx: msg["attrs"].append(["L2TP_ATTR_UDP_ZERO_CSUM6_RX", True]) if debug is not None: msg["attrs"].append(["L2TP_ATTR_DEBUG", debug]) return self._do_request(msg) def create_tunnel( self, tunnel_id, peer_tunnel_id, protocol=3, remote=None, local=None, fd=None, encap="udp", udp_sport=None, udp_dport=None, udp_csum=None, udp6_csum_rx=None, udp6_csum_tx=None, debug=False, ): """ Create a new L2TP tunnel :param tunnel_id: local tunnel id :param peer_tunnel_id: remote tunnel id :param protocol: L2TP version :param remote: IP address of the remote peer :param local: IP address of the local interface :param fd: file descriptor of socket to use :param encap: encapsulation type of the tunnel (udp, ip) :param udp_sport: UDP source port to be used for the tunnel :param udp_dport: UDP destination port to be used for the tunnel :param udp_csum: control if IPv4 UDP checksums should be calculated and checked :param udp6_csum_rx: control if IPv6 UDP rx checksums should be calculated :param udp6_csum_tx: control if IPv6 UDP tx checksums should be calculated :param debug: enable or disable debugging using kernel printk for the tunnel :return: Netlink response """ if not remote: raise ValueError("remote endpoint missing") if not local: raise ValueError("local endpoint missing") if encap == "udp": if not udp_sport: raise ValueError( "udp_sport is required when UDP encapsulation is " "selected" ) if not udp_dport: raise ValueError( "udp_dport is required when UDP encapsulation is " "selected" ) return self._send_tunnel( cmd=L2TP_CMD_TUNNEL_CREATE, tunnel_id=tunnel_id, peer_tunnel_id=peer_tunnel_id, protocol=protocol, remote=remote, local=local, fd=fd, encap=encap, udp_sport=udp_sport, udp_dport=udp_dport, udp_csum=udp_csum, udp6_csum_rx=udp6_csum_rx, udp6_csum_tx=udp6_csum_tx, debug=debug, ) def modify_tunnel(self, tunnel_id, debug): """ Modify an existing L2TP tunnel :param tunnel_id: local tunnel id :param debug: enable or disable debugging using kernel printk for the tunnel :return: netlink response """ return self._send_tunnel( L2TP_CMD_TUNNEL_MODIFY, tunnel_id=tunnel_id, debug=debug ) def delete_tunnel(self, tunnel_id): """ Delete a tunnel :param tunnel_id: tunnel id of the tunnel to be deleted :return: netlink response """ msg = l2tpmsg() msg["cmd"] = L2TP_CMD_TUNNEL_DELETE msg["version"] = L2TP_GENL_VERSION msg["attrs"].append(["L2TP_ATTR_CONN_ID", tunnel_id]) return self._do_request(msg) def _send_session( self, cmd, tunnel_id, session_id, peer_session_id=None, ifname=None, l2spec_type=None, cookie=None, peer_cookie=None, debug=None, seq=None, lns_mode=None, recv_timeout=None, pwtype=L2TP_PWTYPE_ETH, ): """ Send session create or modify commands :param cmd: Netlink command to use :param tunnel_id: local tunnel id :param session_id: local session id :param peer_session_id: remote session id :param ifname: interface name :param l2spec_type: layer2 specific header type of the session :param cookie: local cookie value to be assigned to the session :param peer_cookie: remote cookie value to be assigned to the session :param debug: enable or disable debugging using kernel printk for the session :param seq: controls sequence numbering to prevent or detect out of order packets :param lns_mode: LNS mode :param recv_timeout: Reorder timeout :param pwtype: pseudowire type :return: netlink response """ msg = l2tpmsg() msg["cmd"] = cmd msg["version"] = L2TP_GENL_VERSION msg["attrs"].append(["L2TP_ATTR_CONN_ID", tunnel_id]) msg["attrs"].append(["L2TP_ATTR_SESSION_ID", session_id]) if cmd == L2TP_CMD_SESSION_CREATE: if peer_session_id: msg["attrs"].append( ["L2TP_ATTR_PEER_SESSION_ID", peer_session_id] ) else: raise ValueError( "peer_session_id required when creating a session" ) if ifname: msg["attrs"].append(["L2TP_ATTR_IFNAME", ifname]) if cmd == L2TP_CMD_SESSION_CREATE: if l2spec_type == "none": l2spec_type_value = L2TP_L2SPECTYPE_NONE else: l2spec_type_value = L2TP_L2SPECTYPE_DEFAULT msg["attrs"].append(["L2TP_ATTR_L2SPEC_TYPE", l2spec_type_value]) if cookie: if len(cookie) - 2 not in (8, 16): raise ValueError("cookie must be either 8 or 16 hex digits") msg["attrs"].append(["L2TP_ATTR_COOKIE", cookie]) if peer_cookie: if len(peer_cookie) - 2 not in (8, 16): raise ValueError( "peer_cookie must be either 8 or 16 hex digits" ) msg["attrs"].append(["L2TP_ATTR_PEER_COOKIE", peer_cookie]) if debug is not None: msg["attrs"].append(["L2TP_ATTR_DEBUG", debug]) if seq == "both": msg["attrs"].append(["L2TP_ATTR_RECV_SEQ", True]) msg["attrs"].append(["L2TP_ATTR_SEND_SEQ", True]) elif seq == "recv": msg["attrs"].append(["L2TP_ATTR_RECV_SEQ", True]) elif seq == "send": msg["attrs"].append(["L2TP_ATTR_SEND_SEQ", True]) if lns_mode: msg["attrs"].append(["L2TP_ATTR_LNS_MODE", lns_mode]) if recv_timeout is not None: msg["attrs"].append(["L2TP_ATTR_RECV_TIMEOUT", recv_timeout]) if cmd == L2TP_CMD_SESSION_CREATE: msg["attrs"].append(["L2TP_ATTR_PW_TYPE", pwtype]) return self._do_request(msg) def create_session( self, tunnel_id, session_id, peer_session_id=None, ifname=None, l2spec_type=None, cookie=None, peer_cookie=None, debug=None, seq=None, lns_mode=None, recv_timeout=None, pwtype=L2TP_PWTYPE_ETH, ): """ Add a new session to a tunnel :param tunnel_id: local tunnel id :param session_id: local session id :param peer_session_id: remote session id :param ifname: interface name :param l2spec_type: layer2 specific header type of the session :param cookie: local cookie value to be assigned to the session :param peer_cookie: remote cookie value to be assigned to the session :param debug: enable or disable debugging using kernel printk for the session :param seq: controls sequence numbering to prevent or detect out of order packets :param lns_mode: LNS mode :param recv_timeout: Reorder timeout :param pwtype: pseudowire type :return: netlink response """ self._send_session( cmd=L2TP_CMD_SESSION_CREATE, tunnel_id=tunnel_id, session_id=session_id, peer_session_id=peer_session_id, ifname=ifname, l2spec_type=l2spec_type, cookie=cookie, peer_cookie=peer_cookie, debug=debug, seq=seq, lns_mode=lns_mode, recv_timeout=recv_timeout, pwtype=pwtype, ) def modify_session( self, tunnel_id, session_id, debug=None, seq=None, lns_mode=None, recv_timeout=None, ): """ Modify an existing session :param tunnel_id: local tunnel id :param session_id: local session id :param debug: enable or disable debugging for the session :param seq: controls sequence numbering to prevent or detect out of order packets :param lns_mode: LNS mode :param recv_timeout: Reorder timeout :return: netlink response """ self._send_session( cmd=L2TP_CMD_SESSION_MODIFY, tunnel_id=tunnel_id, session_id=session_id, debug=debug, seq=seq, lns_mode=lns_mode, recv_timeout=recv_timeout, ) def delete_session(self, tunnel_id, session_id): """ Delete a session :param tunnel_id: tunnel id in which the session to be deleted is located :param session_id: session id of the session to be deleted :return:z """ msg = l2tpmsg() msg["cmd"] = L2TP_CMD_SESSION_DELETE msg["version"] = L2TP_GENL_VERSION msg["attrs"].append(["L2TP_ATTR_CONN_ID", tunnel_id]) msg["attrs"].append(["L2TP_ATTR_SESSION_ID", session_id]) return self._do_request(msg) def get_tunnel(self, tunnel_id): """ Get one tunnel :param tunnel_id: tunnel id of the tunnel to show :return: netlink response """ msg = l2tpmsg() msg["cmd"] = L2TP_CMD_TUNNEL_GET msg["version"] = L2TP_GENL_VERSION msg["attrs"].append(["L2TP_ATTR_CONN_ID", tunnel_id]) return self._do_request(msg, msg_flags=NLM_F_REQUEST)[0] def dump_tunnels(self, tunnel_id): """ Dump all tunnels :return: netlink response """ msg = l2tpmsg() msg["cmd"] = L2TP_CMD_TUNNEL_GET msg["version"] = L2TP_GENL_VERSION return self._do_request(msg, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP) def get_session(self, tunnel_id, session_id): """ Get one session :param tunnel_id: tunnel id of the session :param session_id: session id of the session :return: netlink response """ msg = l2tpmsg() msg["cmd"] = L2TP_CMD_SESSION_GET msg["version"] = L2TP_GENL_VERSION msg["attrs"].append(["L2TP_ATTR_CONN_ID", tunnel_id]) msg["attrs"].append(["L2TP_ATTR_SESSION_ID", session_id]) return self._do_request(msg, msg_flags=NLM_F_REQUEST)[0] def dump_sessions(self): """ Dump all sessions :return: netlink response """ msg = l2tpmsg() msg["cmd"] = L2TP_CMD_SESSION_GET msg["version"] = L2TP_GENL_VERSION return self._do_request(msg, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP)	18.92 KB	2025-05-25 12:26
📄 mptcp.py from socket import AF_INET, AF_INET6 from pyroute2.netlink import NLM_F_ACK, NLM_F_DUMP, NLM_F_REQUEST, genlmsg, nla from pyroute2.netlink.generic import GenericNetlinkSocket MPTCP_GENL_NAME = 'mptcp_pm' MPTCP_PM_CMD_UNSPEC = 0 MPTCP_PM_CMD_ADD_ADDR = 1 MPTCP_PM_CMD_DEL_ADDR = 2 MPTCP_PM_CMD_GET_ADDR = 3 MPTCP_PM_CMD_FLUSH_ADDRS = 4 MPTCP_PM_CMD_SET_LIMITS = 5 MPTCP_PM_CMD_GET_LIMITS = 6 MPTCP_PM_CMD_SET_FLAGS = 7 class mptcp_msg(genlmsg): prefix = 'MPTCP_PM_ATTR_' nla_map = ( ('MPTCP_PM_ATTR_UNSPEC', 'none'), ('MPTCP_PM_ATTR_ADDR', 'pm_addr'), ('MPTCP_PM_ATTR_RCV_ADD_ADDRS', 'uint32'), ('MPTCP_PM_ATTR_SUBFLOWS', 'uint32'), ) class pm_addr(nla): prefix = 'MPTCP_PM_ADDR_ATTR_' nla_map = ( ('MPTCP_PM_ADDR_ATTR_UNSPEC', 'none'), ('MPTCP_PM_ADDR_ATTR_FAMILY', 'uint16'), ('MPTCP_PM_ADDR_ATTR_ID', 'uint8'), ('MPTCP_PM_ADDR_ATTR_ADDR4', 'ipaddr'), ('MPTCP_PM_ADDR_ATTR_ADDR6', 'ipaddr'), ('MPTCP_PM_ADDR_ATTR_PORT', 'uint16'), ('MPTCP_PM_ADDR_ATTR_FLAGS', 'uint32'), ('MPTCP_PM_ADDR_ATTR_IF_IDX', 'hex'), ) class MPTCP(GenericNetlinkSocket): def __init__(self, ext_ack=True): super(MPTCP, self).__init__(ext_ack=ext_ack) try: self.bind(MPTCP_GENL_NAME, mptcp_msg) except Exception as e: self.close() raise e def endpoint(self, cmd, kwarg): ''' Usage:: mptcp.endpoint('show') mptcp.endpoint('add', addr='172.17.20.2') mptcp.endpoint('del', id=4) mptcp.endpoint('flush') Argument `addr` is equal to `addr4` and implies `family=AF_INET`, while `addr6` implies `family=AF_INET6` ''' flags_dump = NLM_F_REQUEST \| NLM_F_DUMP flags_base = NLM_F_REQUEST \| NLM_F_ACK commands = { 'show': (MPTCP_PM_CMD_GET_ADDR, flags_dump), 'add': (MPTCP_PM_CMD_ADD_ADDR, flags_base), 'del': (MPTCP_PM_CMD_DEL_ADDR, flags_base), 'flush': (MPTCP_PM_CMD_FLUSH_ADDRS, flags_base), } (command, flags) = commands.get(cmd, cmd) msg = mptcp_msg() msg['cmd'] = command msg['version'] = 1 if cmd in ('add', 'del'): addr_info = {'attrs': []} if 'addr' in kwarg: kwarg['addr4'] = kwarg.pop('addr') if 'addr4' in kwarg: kwarg['family'] = AF_INET elif 'addr6' in kwarg: kwarg['family'] = AF_INET6 for key, value in kwarg.items(): addr_info['attrs'].append( (mptcp_msg.pm_addr.name2nla(key), value) ) msg['attrs'] = [('MPTCP_PM_ATTR_ADDR', addr_info, 0x8000)] return self.nlm_request(msg, msg_type=self.prid, msg_flags=flags) def limits(self, cmd, kwarg): ''' Usage:: mptcp.limits('show') mptcp.limits('set', subflows=10) ''' flags_dump = NLM_F_REQUEST flags_base = NLM_F_REQUEST \| NLM_F_ACK commands = { 'show': (MPTCP_PM_CMD_GET_LIMITS, flags_dump), 'set': (MPTCP_PM_CMD_SET_LIMITS, flags_base), } (command, flags) = commands.get(cmd, cmd) msg = mptcp_msg() msg['cmd'] = command msg['version'] = 1 if cmd == 'set': if not set(kwarg) < set( ('subflows', 'rcv_add_addrs', 'add_addr_accepted') ): raise TypeError('invalid parameter') for key, value in kwarg.items(): if key == 'add_addr_accepted': key = 'rcv_add_addrs' msg['attrs'].append((mptcp_msg.name2nla(key), value)) return self.nlm_request(msg, msg_type=self.prid, msg_flags=flags)	3.84 KB	2025-05-25 12:26
📄 wireguard.py ''' Usage:: # Imports from pyroute2 import NDB, WireGuard IFNAME = 'wg1' # Create a WireGuard interface with NDB() as ndb: with ndb.interfaces.create(kind='wireguard', ifname=IFNAME) as link: link.add_ip('10.0.0.1/24') link.set(state='up') # Create WireGuard object wg = WireGuard() # Add a WireGuard configuration + first peer peer = {'public_key': 'TGFHcm9zc2VCaWNoZV9DJ2VzdExhUGx1c0JlbGxlPDM=', 'endpoint_addr': '8.8.8.8', 'endpoint_port': 8888, 'persistent_keepalive': 15, 'allowed_ips': ['10.0.0.0/24', '8.8.8.8/32']} wg.set(IFNAME, private_key='RCdhcHJlc0JpY2hlLEplU2VyYWlzTGFQbHVzQm9ubmU=', fwmark=0x1337, listen_port=2525, peer=peer) # Add second peer with preshared key peer = {'public_key': 'RCdBcHJlc0JpY2hlLFZpdmVMZXNQcm9iaW90aXF1ZXM=', 'preshared_key': 'Pz8/V2FudFRvVHJ5TXlBZXJvR3Jvc3NlQmljaGU/Pz8=', 'endpoint_addr': '8.8.8.8', 'endpoint_port': 9999, 'persistent_keepalive': 25, 'allowed_ips': ['::/0']} wg.set(IFNAME, peer=peer) # Delete second peer peer = {'public_key': 'RCdBcHJlc0JpY2hlLFZpdmVMZXNQcm9iaW90aXF1ZXM=', 'remove': True} wg.set(IFNAME, peer=peer) # Get information of the interface wg.info(IFNAME) # Get specific value from the interface wg.info(IFNAME)[0].get('WGDEVICE_A_PRIVATE_KEY') NOTES: * The `get()` method always returns iterable * Using `set()` method only requires an interface name * The `peer` structure is described as follow:: struct peer_s { public_key: # Base64 public key - required remove: # Boolean - optional preshared_key: # Base64 preshared key - optional endpoint_addr: # IPv4 or IPv6 endpoint - optional endpoint_port : # endpoint Port - required only if endpoint_addr persistent_keepalive: # time in seconds to send keep alive - optional allowed_ips: # list of CIDRs allowed - optional } ''' import errno import logging import struct from base64 import b64decode, b64encode from binascii import a2b_hex from socket import AF_INET, AF_INET6, inet_ntop, inet_pton from time import ctime from pyroute2.netlink import ( NLA_F_NESTED, NLM_F_ACK, NLM_F_DUMP, NLM_F_REQUEST, genlmsg, nla, ) from pyroute2.netlink.generic import GenericNetlinkSocket # Defines from uapi/wireguard.h WG_GENL_NAME = "wireguard" WG_GENL_VERSION = 1 WG_KEY_LEN = 32 # WireGuard Device commands WG_CMD_GET_DEVICE = 0 WG_CMD_SET_DEVICE = 1 # Wireguard Device attributes WGDEVICE_A_UNSPEC = 0 WGDEVICE_A_IFINDEX = 1 WGDEVICE_A_IFNAME = 2 WGDEVICE_A_PRIVATE_KEY = 3 WGDEVICE_A_PUBLIC_KEY = 4 WGDEVICE_A_FLAGS = 5 WGDEVICE_A_LISTEN_PORT = 6 WGDEVICE_A_FWMARK = 7 WGDEVICE_A_PEERS = 8 # WireGuard Device flags WGDEVICE_F_REPLACE_PEERS = 1 # WireGuard Allowed IP attributes WGALLOWEDIP_A_UNSPEC = 0 WGALLOWEDIP_A_FAMILY = 1 WGALLOWEDIP_A_IPADDR = 2 WGALLOWEDIP_A_CIDR_MASK = 3 # WireGuard Peer flags WGPEER_F_REMOVE_ME = 0 WGPEER_F_REPLACE_ALLOWEDIPS = 1 WGPEER_F_UPDATE_ONLY = 2 # Specific defines WG_MAX_PEERS = 1000 WG_MAX_ALLOWEDIPS = 1000 class wgmsg(genlmsg): prefix = 'WGDEVICE_A_' nla_map = ( ('WGDEVICE_A_UNSPEC', 'none'), ('WGDEVICE_A_IFINDEX', 'uint32'), ('WGDEVICE_A_IFNAME', 'asciiz'), ('WGDEVICE_A_PRIVATE_KEY', 'parse_wg_key'), ('WGDEVICE_A_PUBLIC_KEY', 'parse_wg_key'), ('WGDEVICE_A_FLAGS', 'uint32'), ('WGDEVICE_A_LISTEN_PORT', 'uint16'), ('WGDEVICE_A_FWMARK', 'uint32'), ('WGDEVICE_A_PEERS', 'wgdevice_peer'), ) class wgdevice_peer(nla): prefix = 'WGPEER_A_' nla_flags = NLA_F_NESTED nla_map = ( ('WGPEER_A_UNSPEC', 'none'), ('WGPEER_A_PUBLIC_KEY', 'parse_peer_key'), ('WGPEER_A_PRESHARED_KEY', 'parse_peer_key'), ('WGPEER_A_FLAGS', 'uint32'), ('WGPEER_A_ENDPOINT', 'parse_endpoint'), ('WGPEER_A_PERSISTENT_KEEPALIVE_INTERVAL', 'uint16'), ('WGPEER_A_LAST_HANDSHAKE_TIME', 'parse_handshake_time'), ('WGPEER_A_RX_BYTES', 'uint64'), ('WGPEER_A_TX_BYTES', 'uint64'), ('WGPEER_A_ALLOWEDIPS', 'wgpeer_allowedip'), ('WGPEER_A_PROTOCOL_VERSION', 'uint32'), ) class parse_peer_key(nla): fields = (('key', '32s'),) def decode(self): nla.decode(self) self['value'] = b64encode(self['key']) def encode(self): self['key'] = b64decode(self['value']) nla.encode(self) @staticmethod def parse_endpoint(nla, argv, kwarg): family = AF_INET if 'data' in kwarg: # decoding, fetch the famliy from the NLA data data = kwarg['data'] offset = kwarg['offset'] family = struct.unpack('H', data[offset + 4 : offset + 6])[0] elif kwarg['value']['addr'].find(':') > -1: # encoding, setup family from the addr format family = AF_INET6 if family == AF_INET: return nla.endpoint_ipv4 else: return nla.endpoint_ipv6 class endpoint_ipv4(nla): fields = ( ('family', 'H'), ('port', '>H'), ('addr', '4s'), ('__pad', '8x'), ) def decode(self): nla.decode(self) self['addr'] = inet_ntop(AF_INET, self['addr']) def encode(self): self['family'] = AF_INET self['addr'] = inet_pton(AF_INET, self['addr']) nla.encode(self) class endpoint_ipv6(nla): fields = ( ('family', 'H'), ('port', '>H'), ('flowinfo', '>I'), ('addr', '16s'), ('scope_id', '>I'), ) def decode(self): nla.decode(self) self['addr'] = inet_ntop(AF_INET6, self['addr']) def encode(self): self['family'] = AF_INET6 self['addr'] = inet_pton(AF_INET6, self['addr']) nla.encode(self) class parse_handshake_time(nla): fields = (('tv_sec', 'Q'), ('tv_nsec', 'Q')) def decode(self): nla.decode(self) self['latest handshake'] = ctime(self['tv_sec']) class wgpeer_allowedip(nla): prefix = 'WGALLOWEDIP_A_' nla_flags = NLA_F_NESTED nla_map = ( ('WGALLOWEDIP_A_UNSPEC', 'none'), ('WGALLOWEDIP_A_FAMILY', 'uint16'), ('WGALLOWEDIP_A_IPADDR', 'hex'), ('WGALLOWEDIP_A_CIDR_MASK', 'uint8'), ) def decode(self): nla.decode(self) family = self.get_attr('WGALLOWEDIP_A_FAMILY') if family is None: # Prevent when decode() is called without attrs because all # datas transfered to 'value' entry. # {'attrs': [], 'value': [{'attrs' ... return ipaddr = self.get_attr('WGALLOWEDIP_A_IPADDR') cidr = self.get_attr('WGALLOWEDIP_A_CIDR_MASK') self['addr'] = '{ipaddr}/{cidr}'.format( ipaddr=inet_ntop(family, a2b_hex(ipaddr.replace(':', ''))), cidr=cidr, ) class parse_wg_key(nla): fields = (('key', '32s'),) def decode(self): nla.decode(self) self['value'] = b64encode(self['key']) def encode(self): self['key'] = b64decode(self['value']) nla.encode(self) class WireGuard(GenericNetlinkSocket): def __init__(self, args, *kwargs): GenericNetlinkSocket.__init__(self, args, **kwargs) self.bind(WG_GENL_NAME, wgmsg) def info(self, interface): msg = wgmsg() msg['cmd'] = WG_CMD_GET_DEVICE msg['attrs'].append(['WGDEVICE_A_IFNAME', interface]) return self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP ) def set( self, interface, listen_port=None, fwmark=None, private_key=None, peer=None, ): msg = wgmsg() msg['attrs'].append(['WGDEVICE_A_IFNAME', interface]) if private_key is not None: self._wg_test_key(private_key) msg['attrs'].append(['WGDEVICE_A_PRIVATE_KEY', private_key]) if listen_port is not None: msg['attrs'].append(['WGDEVICE_A_LISTEN_PORT', listen_port]) if fwmark is not None: msg['attrs'].append(['WGDEVICE_A_FWMARK', fwmark]) if peer is not None: self._wg_set_peer(msg, peer) # Message attributes msg['cmd'] = WG_CMD_SET_DEVICE msg['version'] = WG_GENL_VERSION msg['header']['type'] = self.prid msg['header']['flags'] = NLM_F_REQUEST \| NLM_F_ACK msg['header']['pid'] = self.pid msg.encode() self.sendto(msg.data, (0, 0)) msg = self.get()[0] err = msg['header'].get('error', None) if err is not None: if hasattr(err, 'code') and err.code == errno.ENOENT: logging.error( 'Generic netlink protocol %s not found' % self.prid ) logging.error('Please check if the protocol module is loaded') raise err return msg def _wg_test_key(self, key): try: if len(b64decode(key)) != WG_KEY_LEN: raise ValueError('Invalid WireGuard key length') except TypeError: raise ValueError('Failed to decode Base64 key') def _wg_set_peer(self, msg, peer): attrs = [] wg_peer = [{'attrs': attrs}] if 'public_key' not in peer: raise ValueError('Peer Public key required') # Check public key validity public_key = peer['public_key'] self._wg_test_key(public_key) attrs.append(['WGPEER_A_PUBLIC_KEY', public_key]) # If peer removal is set to True if 'remove' in peer and peer['remove']: attrs.append(['WGPEER_A_FLAGS', WGDEVICE_F_REPLACE_PEERS]) msg['attrs'].append(['WGDEVICE_A_PEERS', wg_peer]) return # Set Endpoint if 'endpoint_addr' in peer and 'endpoint_port' in peer: attrs.append( [ 'WGPEER_A_ENDPOINT', { 'addr': peer['endpoint_addr'], 'port': peer['endpoint_port'], }, ] ) # Set Preshared key if 'preshared_key' in peer: pkey = peer['preshared_key'] self._wg_test_key(pkey) attrs.append(['WGPEER_A_PRESHARED_KEY', pkey]) # Set Persistent Keepalive time if 'persistent_keepalive' in peer: keepalive = peer['persistent_keepalive'] attrs.append(['WGPEER_A_PERSISTENT_KEEPALIVE_INTERVAL', keepalive]) # Set Peer flags attrs.append(['WGPEER_A_FLAGS', WGPEER_F_UPDATE_ONLY]) # Set allowed IPs if 'allowed_ips' in peer: allowed_ips = self._wg_build_allowedips(peer['allowed_ips']) attrs.append(['WGPEER_A_ALLOWEDIPS', allowed_ips]) msg['attrs'].append(['WGDEVICE_A_PEERS', wg_peer]) def _wg_build_allowedips(self, allowed_ips): ret = [] for index, ip in enumerate(allowed_ips): allowed_ip = [] ret.append({'attrs': allowed_ip}) if ip.find("/") == -1: raise ValueError('No CIDR set in allowed ip #{}'.format(index)) addr, mask = ip.split('/') family = AF_INET if addr.find(":") == -1 else AF_INET6 allowed_ip.append(['WGALLOWEDIP_A_FAMILY', family]) allowed_ip.append( ['WGALLOWEDIP_A_IPADDR', inet_pton(family, addr)] ) allowed_ip.append(['WGALLOWEDIP_A_CIDR_MASK', int(mask)]) return ret	12.22 KB	2025-05-25 12:26

Terminal