Wala shell

Advanced PHP File Manager

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

Name	Size	Last Modified
📁 ..	-	2025-05-25 12:26
📄 __init__.py ''' RTNetlink: network setup ======================== RTNL is a netlink protocol, used to get and set information about different network objects -- addresses, routes, interfaces etc. RTNL protocol-specific data in messages depends on the object type. E.g., complete packet with the interface address information:: nlmsg header: + uint32 length + uint16 type + uint16 flags + uint32 sequence number + uint32 pid ifaddrmsg structure: + unsigned char ifa_family + unsigned char ifa_prefixlen + unsigned char ifa_flags + unsigned char ifa_scope + uint32 ifa_index [ optional NLA tree ] NLA for this kind of packets can be of type IFA_ADDRESS, IFA_LOCAL etc. -- please refer to the corresponding source. Other objects types require different structures, sometimes really complex. All these structures are described in sources. --------------------------- Module contents: ''' from pyroute2.common import map_namespace # Useful RTNL related constants from the kernel sources: IFNAMSIZ = 16 ALTIFNAMSIZ = 128 # RTnetlink multicast group flags (for use with bind()) RTMGRP_NONE = 0x0 RTMGRP_LINK = 0x1 RTMGRP_NOTIFY = 0x2 RTMGRP_NEIGH = 0x4 RTMGRP_TC = 0x8 RTMGRP_IPV4_IFADDR = 0x10 RTMGRP_IPV4_MROUTE = 0x20 RTMGRP_IPV4_ROUTE = 0x40 RTMGRP_IPV4_RULE = 0x80 RTMGRP_IPV6_IFADDR = 0x100 RTMGRP_IPV6_MROUTE = 0x200 RTMGRP_IPV6_ROUTE = 0x400 RTMGRP_IPV6_IFINFO = 0x800 RTMGRP_DECnet_IFADDR = 0x1000 RTMGRP_NOP2 = 0x2000 RTMGRP_DECnet_ROUTE = 0x4000 RTMGRP_DECnet_RULE = 0x8000 RTMGRP_NOP4 = 0x10000 RTMGRP_IPV6_PREFIX = 0x20000 RTMGRP_IPV6_RULE = 0x40000 RTMGRP_MPLS_ROUTE = 0x4000000 # multicast group ids (for use with {add,drop}_membership) RTNLGRP_NONE = 0 RTNLGRP_LINK = 1 RTNLGRP_NOTIFY = 2 RTNLGRP_NEIGH = 3 RTNLGRP_TC = 4 RTNLGRP_IPV4_IFADDR = 5 RTNLGRP_IPV4_MROUTE = 6 RTNLGRP_IPV4_ROUTE = 7 RTNLGRP_IPV4_RULE = 8 RTNLGRP_IPV6_IFADDR = 9 RTNLGRP_IPV6_MROUTE = 10 RTNLGRP_IPV6_ROUTE = 11 RTNLGRP_IPV6_IFINFO = 12 RTNLGRP_DECnet_IFADDR = 13 RTNLGRP_NOP2 = 14 RTNLGRP_DECnet_ROUTE = 15 RTNLGRP_DECnet_RULE = 16 RTNLGRP_NOP4 = 17 RTNLGRP_IPV6_PREFIX = 18 RTNLGRP_IPV6_RULE = 19 RTNLGRP_ND_USEROPT = 20 RTNLGRP_PHONET_IFADDR = 21 RTNLGRP_PHONET_ROUTE = 22 RTNLGRP_DCB = 23 RTNLGRP_IPV4_NETCONF = 24 RTNLGRP_IPV6_NETCONF = 25 RTNLGRP_MDB = 26 RTNLGRP_MPLS_ROUTE = 27 RTNLGRP_NSID = 28 RTNLGRP_MPLS_NETCONF = 29 RTNLGRP_IPV4_MROUTE_R = 30 RTNLGRP_IPV6_MROUTE_R = 31 # Types of messages # RTM_BASE = 16 RTM_NEWLINK = 16 RTM_DELLINK = 17 RTM_GETLINK = 18 RTM_SETLINK = 19 RTM_NEWADDR = 20 RTM_DELADDR = 21 RTM_GETADDR = 22 RTM_NEWROUTE = 24 RTM_DELROUTE = 25 RTM_GETROUTE = 26 RTM_NEWNEIGH = 28 RTM_DELNEIGH = 29 RTM_GETNEIGH = 30 RTM_NEWRULE = 32 RTM_DELRULE = 33 RTM_GETRULE = 34 RTM_NEWQDISC = 36 RTM_DELQDISC = 37 RTM_GETQDISC = 38 RTM_NEWTCLASS = 40 RTM_DELTCLASS = 41 RTM_GETTCLASS = 42 RTM_NEWTFILTER = 44 RTM_DELTFILTER = 45 RTM_GETTFILTER = 46 RTM_NEWACTION = 48 RTM_DELACTION = 49 RTM_GETACTION = 50 RTM_NEWPREFIX = 52 RTM_GETMULTICAST = 58 RTM_GETANYCAST = 62 RTM_NEWNEIGHTBL = 64 RTM_GETNEIGHTBL = 66 RTM_SETNEIGHTBL = 67 RTM_NEWNDUSEROPT = 68 RTM_NEWADDRLABEL = 72 RTM_DELADDRLABEL = 73 RTM_GETADDRLABEL = 74 RTM_GETDCB = 78 RTM_SETDCB = 79 RTM_NEWNETCONF = 80 RTM_DELNETCONF = 81 RTM_GETNETCONF = 82 RTM_NEWMDB = 84 RTM_DELMDB = 85 RTM_GETMDB = 86 RTM_NEWNSID = 88 RTM_DELNSID = 89 RTM_GETNSID = 90 RTM_NEWSTATS = 92 RTM_GETSTATS = 94 RTM_NEWCACHEREPORT = 96 RTM_NEWLINKPROP = 108 RTM_DELLINKPROP = 109 RTM_GETLINKPROP = 110 # fake internal message types RTM_NEWNETNS = 500 RTM_DELNETNS = 501 RTM_GETNETNS = 502 RTM_NEWPROBE = 504 RTM_DELPROBE = 505 RTM_GETPROBE = 506 (RTM_NAMES, RTM_VALUES) = map_namespace('RTM_', globals()) TC_H_INGRESS = 0xFFFFFFF1 TC_H_CLSACT = TC_H_INGRESS TC_H_ROOT = 0xFFFFFFFF RTMGRP_DEFAULTS = ( RTMGRP_IPV4_IFADDR \| RTMGRP_IPV6_IFADDR \| RTMGRP_IPV4_ROUTE \| RTMGRP_IPV6_ROUTE \| RTMGRP_IPV4_RULE \| RTMGRP_IPV6_RULE \| RTMGRP_NEIGH \| RTMGRP_LINK \| RTMGRP_TC \| RTMGRP_MPLS_ROUTE ) encap_type = {'unspec': 0, 'mpls': 1, 0: 'unspec', 1: 'mpls'} rtypes = { 'RTN_UNSPEC': 0, 'RTN_UNICAST': 1, # Gateway or direct route 'RTN_LOCAL': 2, # Accept locally 'RTN_BROADCAST': 3, # Accept locally as broadcast # send as broadcast 'RTN_ANYCAST': 4, # Accept locally as broadcast, # but send as unicast 'RTN_MULTICAST': 5, # Multicast route 'RTN_BLACKHOLE': 6, # Drop 'RTN_UNREACHABLE': 7, # Destination is unreachable 'RTN_PROHIBIT': 8, # Administratively prohibited 'RTN_THROW': 9, # Not in this table 'RTN_NAT': 10, # Translate this address 'RTN_XRESOLVE': 11, } # Use external resolver # normalized rt_type = dict( [(x[0][4:].lower(), x[1]) for x in rtypes.items()] + [(x[1], x[0][4:].lower()) for x in rtypes.items()] ) rtprotos = { 'RTPROT_UNSPEC': 0, 'RTPROT_REDIRECT': 1, # Route installed by ICMP redirects; # not used by current IPv4 'RTPROT_KERNEL': 2, # Route installed by kernel 'RTPROT_BOOT': 3, # Route installed during boot 'RTPROT_STATIC': 4, # Route installed by administrator # Values of protocol >= RTPROT_STATIC are not # interpreted by kernel; # keep in sync with iproute2 ! 'RTPROT_GATED': 8, # gated 'RTPROT_RA': 9, # RDISC/ND router advertisements 'RTPROT_MRT': 10, # Merit MRT 'RTPROT_ZEBRA': 11, # Zebra 'RTPROT_BIRD': 12, # BIRD 'RTPROT_DNROUTED': 13, # DECnet routing daemon 'RTPROT_XORP': 14, # XORP 'RTPROT_NTK': 15, # Netsukuku 'RTPROT_DHCP': 16, } # DHCP client # normalized rt_proto = dict( [(x[0][7:].lower(), x[1]) for x in rtprotos.items()] + [(x[1], x[0][7:].lower()) for x in rtprotos.items()] ) rtscopes = { 'RT_SCOPE_UNIVERSE': 0, 'RT_SCOPE_SITE': 200, 'RT_SCOPE_LINK': 253, 'RT_SCOPE_HOST': 254, 'RT_SCOPE_NOWHERE': 255, } # normalized rt_scope = dict( [(x[0][9:].lower(), x[1]) for x in rtscopes.items()] + [(x[1], x[0][9:].lower()) for x in rtscopes.items()] )	5.99 KB	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 errmsg.py from pyroute2.netlink import nlmsg class errmsg(nlmsg): ''' Custom message type Error ersatz-message ''' fields = (('code', 'i'),)	155 B	2025-05-25 12:26
📄 fibmsg.py from pyroute2.common import map_namespace from pyroute2.netlink import nla, nlmsg FR_ACT_UNSPEC = 0 FR_ACT_TO_TBL = 1 FR_ACT_GOTO = 2 FR_ACT_NOP = 3 FR_ACT_BLACKHOLE = 6 FR_ACT_UNREACHABLE = 7 FR_ACT_PROHIBIT = 8 (FR_ACT_NAMES, FR_ACT_VALUES) = map_namespace('FR_ACT', globals()) class fibmsg(nlmsg): ''' IP rule message C structure:: struct fib_rule_hdr { __u8 family; __u8 dst_len; __u8 src_len; __u8 tos; __u8 table; __u8 res1; /* reserved / __u8 res2; / reserved */ __u8 action; __u32 flags; }; ''' prefix = 'FRA_' fields = ( ('family', 'B'), ('dst_len', 'B'), ('src_len', 'B'), ('tos', 'B'), ('table', 'B'), ('res1', 'B'), ('res2', 'B'), ('action', 'B'), ('flags', 'I'), ) # fibmsg NLA numbers are not sequential, so # give them here explicitly nla_map = ( (0, 'FRA_UNSPEC', 'none'), (1, 'FRA_DST', 'ipaddr'), (2, 'FRA_SRC', 'ipaddr'), (3, 'FRA_IIFNAME', 'asciiz'), (4, 'FRA_GOTO', 'uint32'), (6, 'FRA_PRIORITY', 'uint32'), (10, 'FRA_FWMARK', 'uint32'), (11, 'FRA_FLOW', 'uint32'), (12, 'FRA_TUN_ID', 'be64'), (13, 'FRA_SUPPRESS_IFGROUP', 'uint32'), (14, 'FRA_SUPPRESS_PREFIXLEN', 'uint32'), (15, 'FRA_TABLE', 'uint32'), (16, 'FRA_FWMASK', 'uint32'), (17, 'FRA_OIFNAME', 'asciiz'), (18, 'FRA_PAD', 'hex'), (19, 'FRA_L3MDEV', 'uint8'), (20, 'FRA_UID_RANGE', 'uid_range'), (21, 'FRA_PROTOCOL', 'uint8'), (22, 'FRA_IP_PROTO', 'uint8'), (23, 'FRA_SPORT_RANGE', 'port_range'), (24, 'FRA_DPORT_RANGE', 'port_range'), ) class fra_range(nla): __slots__ = () sql_type = 'TEXT' def encode(self): self['start'], self['end'] = [ int(x) for x in self.value.split(':') ] nla.encode(self) def decode(self): nla.decode(self) self.value = '%s:%s' % (self['start'], self['end']) class uid_range(fra_range): fields = (('start', 'I'), ('end', 'I')) class port_range(fra_range): fields = (('start', 'H'), ('end', 'H'))	2.37 KB	2025-05-25 12:26
📄 ifaddrmsg.py import socket from pyroute2.common import map_namespace from pyroute2.netlink import nla, nlmsg # address attributes # # Important comment: # For IPv4, IFA_ADDRESS is a prefix address, not a local interface # address. It makes no difference for normal interfaces, but # for point-to-point ones IFA_ADDRESS means DESTINATION address, # and the local address is supplied in IFA_LOCAL attribute. # IFA_F_SECONDARY = 0x01 # IFA_F_TEMPORARY IFA_F_SECONDARY IFA_F_NODAD = 0x02 IFA_F_OPTIMISTIC = 0x04 IFA_F_DADFAILED = 0x08 IFA_F_HOMEADDRESS = 0x10 IFA_F_DEPRECATED = 0x20 IFA_F_TENTATIVE = 0x40 IFA_F_PERMANENT = 0x80 IFA_F_MANAGETEMPADDR = 0x100 IFA_F_NOPREFIXROUTE = 0x200 IFA_F_MCAUTOJOIN = 0x400 IFA_F_STABLE_PRIVACY = 0x800 (IFA_F_NAMES, IFA_F_VALUES) = map_namespace('IFA_F', globals()) # 8<---------------------------------------------- IFA_F_TEMPORARY = IFA_F_SECONDARY IFA_F_NAMES['IFA_F_TEMPORARY'] = IFA_F_TEMPORARY IFA_F_VALUES6 = IFA_F_VALUES IFA_F_VALUES6[IFA_F_TEMPORARY] = 'IFA_F_TEMPORARY' # 8<---------------------------------------------- class ifaddrmsg(nlmsg): ''' IP address information C structure:: struct ifaddrmsg { unsigned char ifa_family; /* Address type / unsigned char ifa_prefixlen; / Prefixlength of address / unsigned char ifa_flags; / Address flags / unsigned char ifa_scope; / Address scope / int ifa_index; / Interface index */ }; ''' prefix = 'IFA_' sql_constraints = {'IFA_LOCAL': "NOT NULL DEFAULT ''"} fields = ( ('family', 'B'), ('prefixlen', 'B'), ('flags', 'B'), ('scope', 'B'), ('index', 'I'), ) nla_map = ( ('IFA_UNSPEC', 'hex'), ('IFA_ADDRESS', 'ipaddr'), ('IFA_LOCAL', 'ipaddr'), ('IFA_LABEL', 'asciiz'), ('IFA_BROADCAST', 'ipaddr'), ('IFA_ANYCAST', 'ipaddr'), ('IFA_CACHEINFO', 'cacheinfo'), ('IFA_MULTICAST', 'ipaddr'), ('IFA_FLAGS', 'uint32'), ) class cacheinfo(nla): fields = ( ('ifa_preferred', 'I'), ('ifa_valid', 'I'), ('cstamp', 'I'), ('tstamp', 'I'), ) @staticmethod def flags2names(flags, family=socket.AF_INET): if family == socket.AF_INET6: ifa_f_values = IFA_F_VALUES6 else: ifa_f_values = IFA_F_VALUES ret = [] for f in ifa_f_values: if f & flags: ret.append(ifa_f_values[f]) return ret @staticmethod def names2flags(flags): ret = 0 for f in flags: if f[0] == '!': f = f[1:] else: ret \|= IFA_F_NAMES[f] return ret	2.73 KB	2025-05-25 12:26
📁 ifinfmsg	-	2025-05-25 12:26
📄 ifstatsmsg.py from pyroute2.netlink import nla, nlmsg from pyroute2.netlink.rtnl.ifinfmsg import ifinfmsg class ifstatsmsg(nlmsg): fields = ( ('family', 'B'), ('pad1', 'B'), ('pad2', 'H'), ('ifindex', 'I'), ('filter_mask', 'I'), ) nla_map = ( ('IFLA_STATS_UNSPEC', 'none'), ('IFLA_STATS_LINK_64', 'ifstats64'), ('IFLA_STATS_LINK_XSTATS', 'ifxstats'), ('IFLA_STATS_LINK_XSTATS_SLAVE', 'ifxstats'), ('IFLA_STATS_LINK_OFFLOAD_XSTATS', 'hex'), ('IFLA_STATS_AF_SPEC', 'hex'), ) class ifstats64(ifinfmsg.ifstats64): pass class ifxstats(nla): nla_map = ( ('LINK_XSTATS_TYPE_UNSPEC', 'none'), ('LINK_XSTATS_TYPE_BRIDGE', 'bridge'), ('LINK_XSTATS_TYPE_BOND', 'hex'), ) class bridge(nla): nla_map = ( ('BRIDGE_XSTATS_UNSPEC', 'none'), ('BRIDGE_XSTATS_VLAN', 'vlan'), ('BRIDGE_XSTATS_MCAST', 'mcast'), ('BRIDGE_XSTATS_PAD', 'hex'), ('BRIDGE_XSTATS_STP', 'stp'), ) class vlan(nla): fields = ( ('rx_bytes', 'Q'), ('rx_packets', 'Q'), ('tx_bytes', 'Q'), ('tx_packets', 'Q'), ('vid', 'H'), ('flags', 'H'), ('pad2', 'I'), ) class mcast(nla): fields = ( ('igmp_v1queries', 'QQ'), ('igmp_v2queries', 'QQ'), ('igmp_v3queries', 'QQ'), ('igmp_leaves', 'QQ'), ('igmp_v1reports', 'QQ'), ('igmp_v2reports', 'QQ'), ('igmp_v3reports', 'QQ'), ('igmp_parse_errors', 'Q'), ('mld_v1queries', 'QQ'), ('mld_v2queries', 'QQ'), ('mld_leaves', 'QQ'), ('mld_v1reports', 'QQ'), ('mld_v2reports', 'QQ'), ('mld_parse_errors', 'Q'), ('mcast_bytes', 'QQ'), ('mcast_packets', 'QQ'), ) class stp(nla): fields = ( ('transition_blk', 'Q'), ('transition_fwd', 'Q'), ('rx_bpdu', 'Q'), ('tx_bpdu', 'Q'), ('rx_tcn', 'Q'), ('tx_tcn', 'Q'), )	2.52 KB	2025-05-25 12:26
📄 iprsocket.py import errno import os import struct from collections.abc import Callable from typing import Optional from unittest import mock from pyroute2 import config from pyroute2.common import msg_done from pyroute2.iproute.ipmock import IPEngine from pyroute2.netlink import NETLINK_ROUTE, NLM_F_REQUEST, nlmsg, rtnl from pyroute2.netlink.exceptions import NetlinkError from pyroute2.netlink.nlsocket import ( AsyncNetlinkSocket, ChaoticNetlinkSocket, NetlinkRequest, NetlinkSocket, ) from pyroute2.netlink.rtnl.ifinfmsg.tuntap import manage_tun, manage_tuntap from pyroute2.netlink.rtnl.marshal import MarshalRtnl from pyroute2.netlink.rtnl.probe_msg import proxy_newprobe from pyroute2.netns import setns from pyroute2.process import ChildProcess, ChildProcessReturnValue def _run_in_netns( netns: str, target: Callable[[nlmsg], bytes], msg: nlmsg ) -> ChildProcessReturnValue: setns(netns) return ChildProcessReturnValue(target(msg), []) class IPRouteProxy: route: dict[ int, tuple[tuple[Callable[[nlmsg], bool], Callable[[nlmsg], bytes]], ...], ] = { rtnl.RTM_NEWLINK: ( (lambda x: x.get(('linkinfo', 'kind')) == 'tuntap', manage_tuntap), (lambda x: x.get(('linkinfo', 'kind')) == 'tun', manage_tun), ), rtnl.RTM_NEWPROBE: ((lambda x: True, proxy_newprobe),), } def __init__(self, netns: Optional[str] = None): self.netns = netns def handle(self, msg: nlmsg) -> bytes: ret: bytes = b'' key = msg['header']['type'] if key not in self.route: return ret for predicate, target in self.route[key]: if predicate(msg): try: if self.netns is not None: with ChildProcess( target=_run_in_netns, args=[self.netns, target, msg], ) as proc: ret = proc.get_data(timeout=4) else: ret = target(msg) return ret or msg_done(msg) except Exception as e: # errmsg if isinstance(e, (OSError, IOError)): code = e.errno or errno.ENODATA elif isinstance(e, NetlinkError): code = e.code else: code = errno.ECOMM newmsg = struct.pack('HH', 2, 0) newmsg += msg.data[8:16] newmsg += struct.pack('I', code) newmsg += msg.data newmsg = struct.pack('I', len(newmsg) + 4) + newmsg return newmsg return b'' class AsyncIPRSocket(AsyncNetlinkSocket): '''A low-level class to provide RTNL socket. This is a low-level class designed to provide an RTNL asyncio-controlled socket. It does not include high-level methods like those found in AsyncIPRoute. Instead, it provides only common netlink methods such as `get()` and `put()`. For more details, refer to the `AsyncNetlinkSocket` documentation. .. testcode:: :hide: from pyroute2 import AsyncIPRSocket iprsock = AsyncIPRSocket() assert callable(iprsock.get) assert callable(iprsock.put) assert callable(iprsock.nlm_request) assert callable(iprsock.bind) Since the underlying socket is controlled by asyncio, it is not possible to use it in poll/select loops. If you want such API, consider using synchronous `IPRSocket`. .. warning:: Your code must process incoming messages quickly enough to prevent the RCVBUF from overflowing. If the RCVBUF overflows, all subsequent socket operations will raise an OSError: .. code:: >>> [ x async for x in ipr.get() ] Traceback (most recent call last): File ".../python3.13/futures/_base.py", line 456, in result return self.__get_result() ~~~~~~~~~~~~~~~~~^^ ... OSError: [Errno 105] No buffer space available If this exception occurs, the only solution is to close the socket and create a new one. This class does not handle protocol-level error propagation; it only provides socket-level error handling. It is the user's responsibility to catch and manage protocol-level errors: .. testsetup:: as0 from pyroute2.netlink import nlmsgerr, NLMSG_ERROR msg = nlmsgerr() msg['header']['type'] = NLMSG_ERROR msg['error'] = 42 msg.reset() msg.encode() msg.decode() .. testcode:: as0 if msg.get(('header', 'type')) == NLMSG_ERROR: # prints error code and the request that # triggered the error print( msg.get('error'), msg.get('msg'), ) .. testoutput:: as0 :hide: 42 None ''' def __init__( self, port=None, pid=None, fileno=None, sndbuf=1048576, rcvbuf=1048576, rcvsize=16384, all_ns=False, async_qsize=None, nlm_generator=None, target='localhost', ext_ack=False, strict_check=False, groups=rtnl.RTMGRP_DEFAULTS, nlm_echo=False, netns=None, netns_path=None, flags=os.O_CREAT, libc=None, use_socket=None, use_event_loop=None, telemetry=None, ): if config.mock_netlink: use_socket = IPEngine() if netns is not None: use_socket.netns = netns use_socket.flags = flags use_socket.initdb() self.marshal = MarshalRtnl() super().__init__( family=NETLINK_ROUTE, port=port, pid=pid, fileno=fileno, sndbuf=sndbuf, rcvbuf=rcvbuf, rcvsize=rcvsize, all_ns=all_ns, async_qsize=async_qsize, nlm_generator=nlm_generator, target=target, ext_ack=ext_ack, strict_check=strict_check, groups=groups, nlm_echo=nlm_echo, use_socket=use_socket, netns=netns, flags=flags, libc=libc, use_event_loop=use_event_loop, telemetry=telemetry, ) if not config.mock_netlink: self.request_proxy = IPRouteProxy(netns) self.status['netns_path'] = netns_path or config.netns_path async def bind(self, groups=None, kwarg): return await super().bind( groups if groups is not None else self.status['groups'], kwarg ) class NotLocal: event_loop = None msg_queue = mock.Mock() class IPRSocket(NetlinkSocket): '''Synchronous select-compatible netlink socket. `IPRSocket` is the synchronous counterpart to `AsyncIPRSocket`. A key feature of `IPRSocket` is that the underlying netlink socket operates out of asyncio control, allowing it to be used in poll/select loops. .. warning:: Your code must process incoming messages quickly enough to prevent the RCVBUF from overflowing. If the RCVBUF overflows, all subsequent socket operations will raise an OSError: .. code:: >>> iprsock.get() Traceback (most recent call last): File "<python-input-12>", line 1, in <module> iprsock.get() ~~~~~~~~^^ File ".../pyroute2/netlink/rtnl/iprsocket.py", line 276, in get data = self.socket.recv(16384) OSError: [Errno 105] No buffer space available >>> If this exception occurs, the only solution is to close the socket and create a new one. Some usage examples: .. testcode:: import select from pyroute2 import IPRSocket from pyroute2.netlink import NLM_F_DUMP, NLM_F_REQUEST from pyroute2.netlink.rtnl import RTM_GETLINK from pyroute2.netlink.rtnl.ifinfmsg import ifinfmsg with IPRSocket() as iprsock: iprsock.put( ifinfmsg(), msg_type=RTM_GETLINK, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP ) ret = [] while True: rl, wl, xl = select.select([iprsock], [], [], 0) if not len(rl): break ret.extend(iprsock.get()) for link in ret: if link.get('event') == 'RTM_NEWLINK': print( link.get('ifname'), link.get('state'), link.get('address'), ) .. testoutput:: lo up 00:00:00:00:00:00 eth0 up 52:54:00:72:58:b2 Threadless RT netlink monitoring with blocking I/O calls: >>> from pyroute2 import IPRSocket >>> from pprint import pprint >>> s = IPRSocket() >>> s.bind() >>> pprint(s.get()) [{'attrs': [('RTA_TABLE', 254), ('RTA_OIF', 2), ('RTA_GATEWAY', '192.168.122.1')], 'dst_len': 0, 'event': 'RTM_NEWROUTE', 'family': 2, 'flags': 0, 'header': {'error': None, 'flags': 2, 'length': 52, 'pid': 325359, 'sequence_number': 255, 'type': 24}, 'proto': 2, 'scope': 0, 'src_len': 0, 'table': 254, 'tos': 0, 'type': 2}] >>> Like `AsyncIPRSocket`, it does not perform response reassembly, protocol-level error propagation, or packet buffering. ''' def __init__( self, port=None, pid=None, fileno=None, sndbuf=1048576, rcvbuf=1048576, rcvsize=16384, all_ns=False, async_qsize=None, nlm_generator=None, target='localhost', ext_ack=False, strict_check=False, groups=rtnl.RTMGRP_DEFAULTS, nlm_echo=False, netns=None, netns_path=None, flags=os.O_CREAT, libc=None, use_socket=None, use_event_loop=None, ): self.asyncore = AsyncIPRSocket( port=port, pid=pid, fileno=fileno, sndbuf=sndbuf, rcvbuf=rcvbuf, rcvsize=rcvsize, all_ns=all_ns, target=target, ext_ack=ext_ack, strict_check=strict_check, groups=groups, nlm_echo=nlm_echo, netns=netns, netns_path=netns_path, flags=flags, libc=libc, use_socket=use_socket, use_event_loop=use_event_loop, ) self.asyncore.local = NotLocal() @property def addr_pool(self): return self.asyncore.addr_pool @property def socket(self): return self.asyncore.socket @property def fileno(self): return self.asyncore.local.socket.fileno def bind(self, groups=None, pid=None, kwarg): self.asyncore._sync_bind(groups, pid, kwarg) def put( self, msg, msg_type, msg_flags=NLM_F_REQUEST, addr=(0, 0), msg_seq=0, msg_pid=None, ): if msg is None: msg_class = self.marshal.msg_map[msg_type] msg = msg_class() request = NetlinkRequest( self, msg, msg_type=msg_type, msg_flags=msg_flags, msg_seq=msg_seq, msg_pid=msg_pid, ) request.msg.encode() return request.sock.send(request.msg.data) def get(self, msg_seq=0, terminate=None, callback=None, noraise=False): data = self.socket.recv(16384) return [x for x in self.marshal.parse(data)] class ChaoticIPRSocket(AsyncIPRSocket, ChaoticNetlinkSocket): pass	11.92 KB	2025-05-25 12:26
📄 iw_event.py from pyroute2.netlink import nla class iw_event(nla): nla_map = ( (0xB00, 'SIOCSIWCOMMIT', 'hex'), (0xB01, 'SIOCGIWNAME', 'hex'), # Basic operations (0xB02, 'SIOCSIWNWID', 'hex'), (0xB03, 'SIOCGIWNWID', 'hex'), (0xB04, 'SIOCSIWFREQ', 'hex'), (0xB05, 'SIOCGIWFREQ', 'hex'), (0xB06, 'SIOCSIWMODE', 'hex'), (0xB07, 'SIOCGIWMODE', 'hex'), (0xB08, 'SIOCSIWSENS', 'hex'), (0xB09, 'SIOCGIWSENS', 'hex'), # Informative stuff (0xB0A, 'SIOCSIWRANGE', 'hex'), (0xB0B, 'SIOCGIWRANGE', 'hex'), (0xB0C, 'SIOCSIWPRIV', 'hex'), (0xB0D, 'SIOCGIWPRIV', 'hex'), (0xB0E, 'SIOCSIWSTATS', 'hex'), (0xB0F, 'SIOCGIWSTATS', 'hex'), # Spy support (statistics per MAC address - # used for Mobile IP support) (0xB10, 'SIOCSIWSPY', 'hex'), (0xB11, 'SIOCGIWSPY', 'hex'), (0xB12, 'SIOCSIWTHRSPY', 'hex'), (0xB13, 'SIOCGIWTHRSPY', 'hex'), # Access Point manipulation (0xB14, 'SIOCSIWAP', 'hex'), (0xB15, 'SIOCGIWAP', 'hex'), (0xB17, 'SIOCGIWAPLIST', 'hex'), (0xB18, 'SIOCSIWSCAN', 'hex'), (0xB19, 'SIOCGIWSCAN', 'hex'), # 802.11 specific support (0xB1A, 'SIOCSIWESSID', 'hex'), (0xB1B, 'SIOCGIWESSID', 'hex'), (0xB1C, 'SIOCSIWNICKN', 'hex'), (0xB1D, 'SIOCGIWNICKN', 'hex'), # Other parameters useful in 802.11 and # some other devices (0xB20, 'SIOCSIWRATE', 'hex'), (0xB21, 'SIOCGIWRATE', 'hex'), (0xB22, 'SIOCSIWRTS', 'hex'), (0xB23, 'SIOCGIWRTS', 'hex'), (0xB24, 'SIOCSIWFRAG', 'hex'), (0xB25, 'SIOCGIWFRAG', 'hex'), (0xB26, 'SIOCSIWTXPOW', 'hex'), (0xB27, 'SIOCGIWTXPOW', 'hex'), (0xB28, 'SIOCSIWRETRY', 'hex'), (0xB29, 'SIOCGIWRETRY', 'hex'), # Encoding stuff (scrambling, hardware security, WEP...) (0xB2A, 'SIOCSIWENCODE', 'hex'), (0xB2B, 'SIOCGIWENCODE', 'hex'), # Power saving stuff (power management, unicast # and multicast) (0xB2C, 'SIOCSIWPOWER', 'hex'), (0xB2D, 'SIOCGIWPOWER', 'hex'), # WPA : Generic IEEE 802.11 information element # (e.g., for WPA/RSN/WMM). (0xB30, 'SIOCSIWGENIE', 'hex'), (0xB31, 'SIOCGIWGENIE', 'hex'), # WPA : IEEE 802.11 MLME requests (0xB16, 'SIOCSIWMLME', 'hex'), # WPA : Authentication mode parameters (0xB32, 'SIOCSIWAUTH', 'hex'), (0xB33, 'SIOCGIWAUTH', 'hex'), # WPA : Extended version of encoding configuration (0xB34, 'SIOCSIWENCODEEXT', 'hex'), (0xB35, 'SIOCGIWENCODEEXT', 'hex'), # WPA2 : PMKSA cache management (0xB36, 'SIOCSIWPMKSA', 'hex'), # Events s.str. (0xC00, 'IWEVTXDROP', 'hex'), (0xC01, 'IWEVQUAL', 'hex'), (0xC02, 'IWEVCUSTOM', 'hex'), (0xC03, 'IWEVREGISTERED', 'hex'), (0xC04, 'IWEVEXPIRED', 'hex'), (0xC05, 'IWEVGENIE', 'hex'), (0xC06, 'IWEVMICHAELMICFAILURE', 'hex'), (0xC07, 'IWEVASSOCREQIE', 'hex'), (0xC08, 'IWEVASSOCRESPIE', 'hex'), (0xC09, 'IWEVPMKIDCAND', 'hex'), )	3.17 KB	2025-05-25 12:26
📄 marshal.py from pyroute2.netlink import rtnl from pyroute2.netlink.nlsocket import Marshal from pyroute2.netlink.rtnl.fibmsg import fibmsg from pyroute2.netlink.rtnl.ifaddrmsg import ifaddrmsg from pyroute2.netlink.rtnl.ifinfmsg import ifinfmsg from pyroute2.netlink.rtnl.ifstatsmsg import ifstatsmsg from pyroute2.netlink.rtnl.ndmsg import ndmsg from pyroute2.netlink.rtnl.ndtmsg import ndtmsg from pyroute2.netlink.rtnl.nsidmsg import nsidmsg from pyroute2.netlink.rtnl.probe_msg import probe_msg from pyroute2.netlink.rtnl.rtmsg import rtmsg from pyroute2.netlink.rtnl.tcmsg import tcmsg class MarshalRtnl(Marshal): msg_map = { rtnl.RTM_NEWLINK: ifinfmsg, rtnl.RTM_DELLINK: ifinfmsg, rtnl.RTM_GETLINK: ifinfmsg, rtnl.RTM_SETLINK: ifinfmsg, rtnl.RTM_NEWADDR: ifaddrmsg, rtnl.RTM_DELADDR: ifaddrmsg, rtnl.RTM_GETADDR: ifaddrmsg, rtnl.RTM_NEWROUTE: rtmsg, rtnl.RTM_DELROUTE: rtmsg, rtnl.RTM_GETROUTE: rtmsg, rtnl.RTM_NEWRULE: fibmsg, rtnl.RTM_DELRULE: fibmsg, rtnl.RTM_GETRULE: fibmsg, rtnl.RTM_NEWNEIGH: ndmsg, rtnl.RTM_DELNEIGH: ndmsg, rtnl.RTM_GETNEIGH: ndmsg, rtnl.RTM_NEWQDISC: tcmsg, rtnl.RTM_DELQDISC: tcmsg, rtnl.RTM_GETQDISC: tcmsg, rtnl.RTM_NEWTCLASS: tcmsg, rtnl.RTM_DELTCLASS: tcmsg, rtnl.RTM_GETTCLASS: tcmsg, rtnl.RTM_NEWTFILTER: tcmsg, rtnl.RTM_DELTFILTER: tcmsg, rtnl.RTM_GETTFILTER: tcmsg, rtnl.RTM_NEWNEIGHTBL: ndtmsg, rtnl.RTM_GETNEIGHTBL: ndtmsg, rtnl.RTM_SETNEIGHTBL: ndtmsg, rtnl.RTM_NEWNSID: nsidmsg, rtnl.RTM_DELNSID: nsidmsg, rtnl.RTM_GETNSID: nsidmsg, rtnl.RTM_NEWSTATS: ifstatsmsg, rtnl.RTM_GETSTATS: ifstatsmsg, rtnl.RTM_NEWLINKPROP: ifinfmsg, rtnl.RTM_DELLINKPROP: ifinfmsg, rtnl.RTM_NEWPROBE: probe_msg, } def fix_message(self, msg): # FIXME: pls do something with it try: msg['event'] = rtnl.RTM_VALUES[msg['header']['type']] except: pass	2.06 KB	2025-05-25 12:26
📄 ndmsg.py from pyroute2.common import map_namespace from pyroute2.netlink import nla, nlmsg # neighbor cache entry flags NTF_USE = 0x01 NTF_SELF = 0x02 NTF_MASTER = 0x04 NTF_PROXY = 0x08 NTF_EXT_LEARNED = 0x10 NTF_ROUTER = 0x80 # neighbor cache entry states NUD_INCOMPLETE = 0x01 NUD_REACHABLE = 0x02 NUD_STALE = 0x04 NUD_DELAY = 0x08 NUD_PROBE = 0x10 NUD_FAILED = 0x20 # dummy states NUD_NOARP = 0x40 NUD_PERMANENT = 0x80 NUD_NONE = 0x00 (NTF_NAMES, NTF_VALUES) = map_namespace('NTF_', globals()) (NUD_NAMES, NUD_VALUES) = map_namespace('NUD_', globals()) flags = dict([(x[0][4:].lower(), x[1]) for x in NTF_NAMES.items()]) states = dict([(x[0][4:].lower(), x[1]) for x in NUD_NAMES.items()]) def states_a2n(s): # parse state string ss = s.split(',') ret = 0 for state in ss: state = state.upper() if not state.startswith('NUD_'): state = 'NUD_' + state ret \|= NUD_NAMES[state] return ret class ndmsg(nlmsg): ''' ARP cache update message C structure:: struct ndmsg { unsigned char ndm_family; int ndm_ifindex; /* Interface index / __u16 ndm_state; / State / __u8 ndm_flags; / Flags */ __u8 ndm_type; }; Cache info structure:: struct nda_cacheinfo { __u32 ndm_confirmed; __u32 ndm_used; __u32 ndm_updated; __u32 ndm_refcnt; }; ''' __slots__ = () prefix = 'NDA_' fields = ( ('family', 'B'), ('__pad', '3x'), ('ifindex', 'i'), ('state', 'H'), ('flags', 'B'), ('ndm_type', 'B'), ) # Please note, that nla_map creates implicit # enumeration. In this case it will be: # # NDA_UNSPEC = 0 # NDA_DST = 1 # NDA_LLADDR = 2 # NDA_CACHEINFO = 3 # NDA_PROBES = 4 # ... # nla_map = ( ('NDA_UNSPEC', 'none'), ('NDA_DST', 'ipaddr'), ('NDA_LLADDR', 'lladdr'), ('NDA_CACHEINFO', 'cacheinfo'), ('NDA_PROBES', 'uint32'), ('NDA_VLAN', 'uint16'), ('NDA_PORT', 'be16'), ('NDA_VNI', 'uint32'), ('NDA_IFINDEX', 'uint32'), ('NDA_MASTER', 'uint32'), ('NDA_LINK_NETNSID', 'uint32'), ('NDA_SRC_VNI', 'uint32'), ) class cacheinfo(nla): __slots__ = () fields = ( ('ndm_confirmed', 'I'), ('ndm_used', 'I'), ('ndm_updated', 'I'), ('ndm_refcnt', 'I'), )	2.55 KB	2025-05-25 12:26
📄 ndtmsg.py from pyroute2.netlink import nla, nlmsg class ndtmsg(nlmsg): ''' Neighbour table message ''' __slots__ = () fields = (('family', 'B'), ('__pad', '3x')) nla_map = ( ('NDTA_UNSPEC', 'none'), ('NDTA_NAME', 'asciiz'), ('NDTA_THRESH1', 'uint32'), ('NDTA_THRESH2', 'uint32'), ('NDTA_THRESH3', 'uint32'), ('NDTA_CONFIG', 'config'), ('NDTA_PARMS', 'parms'), ('NDTA_STATS', 'stats'), ('NDTA_GC_INTERVAL', 'uint64'), ) class config(nla): __slots__ = () fields = ( ('key_len', 'H'), ('entry_size', 'H'), ('entries', 'I'), ('last_flush', 'I'), # delta to now in msecs ('last_rand', 'I'), # delta to now in msecs ('hash_rnd', 'I'), ('hash_mask', 'I'), ('hash_chain_gc', 'I'), ('proxy_qlen', 'I'), ) class stats(nla): __slots__ = () fields = ( ('allocs', 'Q'), ('destroys', 'Q'), ('hash_grows', 'Q'), ('res_failed', 'Q'), ('lookups', 'Q'), ('hits', 'Q'), ('rcv_probes_mcast', 'Q'), ('rcv_probes_ucast', 'Q'), ('periodic_gc_runs', 'Q'), ('forced_gc_runs', 'Q'), ) class parms(nla): __slots__ = () nla_map = ( ('NDTPA_UNSPEC', 'none'), ('NDTPA_IFINDEX', 'uint32'), ('NDTPA_REFCNT', 'uint32'), ('NDTPA_REACHABLE_TIME', 'uint64'), ('NDTPA_BASE_REACHABLE_TIME', 'uint64'), ('NDTPA_RETRANS_TIME', 'uint64'), ('NDTPA_GC_STALETIME', 'uint64'), ('NDTPA_DELAY_PROBE_TIME', 'uint64'), ('NDTPA_QUEUE_LEN', 'uint32'), ('NDTPA_APP_PROBES', 'uint32'), ('NDTPA_UCAST_PROBES', 'uint32'), ('NDTPA_MCAST_PROBES', 'uint32'), ('NDTPA_ANYCAST_DELAY', 'uint64'), ('NDTPA_PROXY_DELAY', 'uint64'), ('NDTPA_PROXY_QLEN', 'uint32'), ('NDTPA_LOCKTIME', 'uint64'), ('NDTPA_QUEUE_LENBYTES', 'uint32'), )	2.14 KB	2025-05-25 12:26
📄 nsidmsg.py from pyroute2.netlink.rtnl.rtgenmsg import rtgenmsg class nsidmsg(rtgenmsg): nla_map = ( ('NETNSA_NONE', 'none'), ('NETNSA_NSID', 'uint32'), ('NETNSA_PID', 'uint32'), ('NETNSA_FD', 'uint32'), ('NETNSA_TARGET_NSID', 'uint32'), ('NETNSA_CURRENT_NSID', 'uint32'), )	321 B	2025-05-25 12:26
📄 nsinfmsg.py from pyroute2.netlink import nlmsg, nlmsg_atoms class nsinfmsg(nlmsg): ''' Fake message type to represent network namespace information. This is a prototype, the NLA layout is subject to change without notification. ''' __slots__ = () prefix = 'NSINFO_' fields = (('inode', 'I'), ('netnsid', 'I')) nla_map = ( ('NSINFO_UNSPEC', 'none'), ('NSINFO_PATH', 'string'), ('NSINFO_PEER', 'peer'), ) class peer(nlmsg_atoms.string): sql_type = None	522 B	2025-05-25 12:26
📄 p2pmsg.py from pyroute2.netlink import nlmsg class p2pmsg(nlmsg): ''' Fake message type to represent peer to peer connections, be it GRE or PPP ''' __slots__ = () prefix = 'P2P_' fields = (('index', 'I'), ('family', 'I')) nla_map = ( ('P2P_UNSPEC', 'none'), ('P2P_LOCAL', 'target'), ('P2P_REMOTE', 'target'), )	366 B	2025-05-25 12:26
📄 probe_msg.py import errno import json import shutil import socket import ssl import subprocess from pyroute2.netlink import nlmsg from pyroute2.netlink.exceptions import NetlinkError class probe_msg(nlmsg): ''' Fake message type to represent network probe info. This is a prototype, the NLA layout is subject to change without notification. ''' __slots__ = () prefix = 'PROBE_' fields = (('family', 'B'), ('proto', 'B'), ('port', 'H'), ('dst_len', 'I')) nla_map = ( ('PROBE_UNSPEC', 'none'), ('PROBE_KIND', 'asciiz'), ('PROBE_STDOUT', 'asciiz'), ('PROBE_STDERR', 'asciiz'), ('PROBE_SRC', 'asciiz'), ('PROBE_DST', 'asciiz'), ('PROBE_NUM', 'uint8'), ('PROBE_TIMEOUT', 'uint8'), ('PROBE_HOSTNAME', 'asciiz'), ('PROBE_SSL_VERIFY', 'uint8'), ('PROBE_SSL_VERSION', 'asciiz'), ('PROBE_SSL_CERT_JSON', 'asciiz'), ('PROBE_SSL_CERT_DER', 'cdata'), ) def probe_ping(msg): num = msg.get('num') timeout = msg.get('timeout') dst = msg.get('dst') kind = msg.get('kind') args = [shutil.which(kind), '-c', f'{num}', '-W', f'{timeout}', f'{dst}'] if args[0] is None: raise NetlinkError(errno.ENOENT, 'probe not found') process = subprocess.Popen( args, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) try: out, err = process.communicate(timeout=timeout) if out: msg['attrs'].append(['PROBE_STDOUT', out]) if err: msg['attrs'].append(['PROBE_STDERR', err]) except subprocess.TimeoutExpired: process.terminate() raise NetlinkError(errno.ETIMEDOUT, 'timeout expired') finally: process.stdout.close() process.stderr.close() return_code = process.wait() if return_code != 0: raise NetlinkError(errno.EHOSTUNREACH, 'probe failed') def probe_tcp(msg, close=True): timeout = msg.get('timeout') dst = msg.get('dst') port = msg.get('port') connection = None try: connection = socket.create_connection((dst, port), timeout=timeout) except ConnectionRefusedError: raise NetlinkError(errno.ECONNREFUSED, 'connection refused') except TimeoutError: raise NetlinkError(errno.ETIMEDOUT, 'timeout expired') except Exception: raise NetlinkError(errno.ECOMM, 'probe failed') finally: if close and connection is not None: connection.close() return connection def probe_ssl(msg): hostname = msg.get('hostname') or msg.get('dst') context = ssl.create_default_context() context.verify_mode = msg.get('ssl_verify', ssl.CERT_REQUIRED) with probe_tcp(msg, close=False) as connection: try: with context.wrap_socket( connection, server_hostname=hostname ) as ssl_wrap: version = ssl_wrap.version() peer_cert_json = ssl_wrap.getpeercert(binary_form=False) peer_cert_der = ssl_wrap.getpeercert(binary_form=True) if peer_cert_json is not None: msg['attrs'].append( ['PROBE_SSL_CERT_JSON', json.dumps(peer_cert_json)] ) if peer_cert_der is not None: msg['attrs'].append(['PROBE_SSL_CERT_DER', peer_cert_der]) if version is not None: msg['attrs'].append(['PROBE_SSL_VERSION', version]) except ssl.SSLError as e: code = errno.EPROTO if e.reason == 'UNSUPPORTED_PROTOCOL': code = errno.EPROTONOSUPPORT elif e.reason == 'CERTIFICATE_VERIFY_FAILED': code = errno.EACCES raise NetlinkError(code, e.strerror) except Exception: raise NetlinkError(errno.ECOMM, 'probe failed') def proxy_newprobe(msg): kind = msg.get('kind') if kind.startswith('ping'): probe_ping(msg) elif kind == 'tcp': probe_tcp(msg) elif kind == 'ssl': probe_ssl(msg) else: raise NetlinkError(errno.ENOTSUP, 'probe type not supported') msg.reset() msg.encode() return msg.data	4.13 KB	2025-05-25 12:26
📄 rtgenmsg.py from pyroute2.netlink import nlmsg class rtgenmsg(nlmsg): fields = (('rtgen_family', 'B'), ('__pad', '3x'))	114 B	2025-05-25 12:26
📄 rtmsg.py import struct from socket import AF_INET, AF_INET6, AF_UNSPEC, inet_ntop, inet_pton from pyroute2.common import AF_MPLS, hexdump, map_namespace from pyroute2.netlink import nla, nla_string, nlmsg from pyroute2.netlink.rtnl import rt_proto, rt_scope, rt_type RTNH_F_DEAD = 1 RTNH_F_PERVASIVE = 2 RTNH_F_ONLINK = 4 RTNH_F_OFFLOAD = 8 RTNH_F_LINKDOWN = 16 (RTNH_F_NAMES, RTNH_F_VALUES) = map_namespace('RTNH_F_', globals()) RT_SCOPE_UNIVERSE = 0 RT_SCOPE_SITE = 200 RT_SCOPE_LINK = 253 RT_SCOPE_HOST = 254 RT_SCOPE_NOWHERE = 255 (RT_SCOPE_NAMES, RT_SCOPE_VALUES) = map_namespace('RT_SCOPE_', globals()) LWTUNNEL_ENCAP_NONE = 0 LWTUNNEL_ENCAP_MPLS = 1 LWTUNNEL_ENCAP_IP = 2 LWTUNNEL_ENCAP_ILA = 3 LWTUNNEL_ENCAP_IP6 = 4 LWTUNNEL_ENCAP_SEG6 = 5 LWTUNNEL_ENCAP_BPF = 6 LWTUNNEL_ENCAP_SEG6_LOCAL = 7 IP6_RT_PRIO_USER = 1024 class nlflags(object): def encode(self): if isinstance(self['flags'], str): self['flags'] = self['flags'].split(',') if isinstance(self['flags'], (set, tuple, list)): self['flags'] = self.names2flags(self['flags']) if isinstance(self.get('scope'), str): self['scope'] = self.name2scope(self['scope']) return super(nlflags, self).encode() @staticmethod def scope2name(scope): return RT_SCOPE_VALUES[scope].lower()[9:] @staticmethod def name2scope(scope): return RT_SCOPE_NAMES['RT_SCOPE_' + scope.upper()] @staticmethod def flags2names(flags): ret = [] for flag in RTNH_F_VALUES: if (flag & flags) == flag: ret.append(RTNH_F_VALUES[flag].lower()[7:]) return ret @staticmethod def names2flags(flags): ret = 0 if isinstance(flags, str): flags = [flags] for flag in flags: ret \|= RTNH_F_NAMES['RTNH_F_' + flag.upper()] return ret class rtmsg_base(nlflags): ''' Route message C structure:: struct rtmsg { unsigned char rtm_family; /* Address family of route / unsigned char rtm_dst_len; / Length of destination / unsigned char rtm_src_len; / Length of source / unsigned char rtm_tos; / TOS filter / unsigned char rtm_table; / Routing table ID / unsigned char rtm_protocol; / Routing protocol; see below / unsigned char rtm_scope; / See below / unsigned char rtm_type; / See below / unsigned int rtm_flags; }; ''' __slots__ = () prefix = 'RTA_' sql_constraints = { 'RTA_TABLE': 'NOT NULL DEFAULT 0', 'RTA_DST': "NOT NULL DEFAULT ''", 'RTA_OIF': 'NOT NULL DEFAULT 0', 'RTA_PRIORITY': 'NOT NULL DEFAULT 0', 'RTA_VIA': "NOT NULL DEFAULT ''", 'RTA_NEWDST': "NOT NULL DEFAULT ''", } defaults = { 'family': AF_INET, 'scope': rt_scope['universe'], 'type': rt_type['unspec'], 'proto': rt_proto['unspec'], } field_policies = {('field', 'table'): lambda x: 252 if x > 255 else x} fields = ( ('family', 'B'), ('dst_len', 'B'), ('src_len', 'B'), ('tos', 'B'), ('table', 'B'), ('proto', 'B'), ('scope', 'B'), ('type', 'B'), ('flags', 'I'), ) nla_map = ( ('RTA_UNSPEC', 'none'), ('RTA_DST', 'target'), ('RTA_SRC', 'target'), ('RTA_IIF', 'uint32'), ('RTA_OIF', 'uint32'), ('RTA_GATEWAY', 'target'), ('RTA_PRIORITY', 'uint32'), ('RTA_PREFSRC', 'target'), ('RTA_METRICS', 'metrics'), ('RTA_MULTIPATH', 'get_nh'), ('RTA_PROTOINFO', 'uint32'), ('RTA_FLOW', 'uint32'), ('RTA_CACHEINFO', 'cacheinfo'), ('RTA_SESSION', 'hex'), ('RTA_MP_ALGO', 'hex'), ('RTA_TABLE', 'uint32'), ('RTA_MARK', 'uint32'), ('RTA_MFC_STATS', 'rta_mfc_stats'), ('RTA_VIA', 'rtvia'), ('RTA_NEWDST', 'target'), ('RTA_PREF', 'uint8'), ('RTA_ENCAP_TYPE', 'uint16'), ('RTA_ENCAP', 'encap_info'), ('RTA_EXPIRES', 'hex'), ) @staticmethod def encap_info(self, argv, kwarg): encap_type = None # Check, if RTA_ENCAP_TYPE is decoded already # for name, value in self['attrs']: if name == 'RTA_ENCAP_TYPE': encap_type = value break else: # No RTA_ENCAP_TYPE met, so iterate all the chain. # Ugly, but to do otherwise would be too complicated. # data = kwarg['data'] offset = kwarg['offset'] while offset < len(data): # Shift offset to the next NLA # NLA header: # # uint16 length # uint16 type # try: offset += struct.unpack('H', data[offset : offset + 2])[0] # 21 == RTA_ENCAP_TYPE # FIXME: should not be hardcoded if ( struct.unpack('H', data[offset + 2 : offset + 4])[0] == 21 ): encap_type = struct.unpack( 'H', data[offset + 4 : offset + 6] )[0] break except: # in the case of any decoding error return self.hex break # return specific classes # return self.encaps.get(encap_type, self.hex) class mpls_encap_info(nla): prefix = 'MPLS_IPTUNNEL_' __slots__ = () nla_map = ( ('MPLS_IPTUNNEL_UNSPEC', 'none'), ('MPLS_IPTUNNEL_DST', 'mpls_target'), ('MPLS_IPTUNNEL_TTL', 'uint8'), ) class seg6_encap_info(nla): __slots__ = () nla_map = ( ('SEG6_IPTUNNEL_UNSPEC', 'none'), ('SEG6_IPTUNNEL_SRH', 'ipv6_sr_hdr'), ) class ipv6_sr_hdr(nla): __slots__ = () fields = ( ('encapmode', 'I'), ('nexthdr', 'B'), ('hdrlen', 'B'), ('type', 'B'), ('segments_left', 'B'), ('first_segment', 'B'), ('flags', 'B'), ('reserved', 'H'), ('segs', 's'), # Potentially several type-length-value ('tlvs', 's'), ) # Corresponding values for seg6 encap modes SEG6_IPTUN_MODE_INLINE = 0 SEG6_IPTUN_MODE_ENCAP = 1 # Mapping string to nla value encapmodes = { "inline": SEG6_IPTUN_MODE_INLINE, "encap": SEG6_IPTUN_MODE_ENCAP, } # Reverse mapping: mapping nla value to string r_encapmodes = {v: k for k, v in encapmodes.items()} # Nla value for seg6 type SEG6_TYPE = 4 # Flag value for hmac SR6_FLAG1_HMAC = 1 << 3 # Tlv value for hmac SR6_TLV_HMAC = 5 # Utility function to get the family from the msg def get_family(self): pointer = self while pointer.parent is not None: pointer = pointer.parent return pointer.get('family', AF_UNSPEC) def encode(self): # Retrieve the family family = self.get_family() # Seg6 can be applied only to IPv6 and IPv4 if family == AF_INET6 or family == AF_INET: # Get mode mode = self['mode'] # Get segs segs = self['segs'] # Get hmac hmac = self.get('hmac', None) # With "inline" mode there is not # encap into an outer IPv6 header if mode == "inline": # Add :: to segs segs.insert(0, "::") # Add mode to value self['encapmode'] = self.encapmodes.get( mode, self.SEG6_IPTUN_MODE_ENCAP ) # Calculate srlen srhlen = 8 + 16 len(segs) # If we are using hmac we have a tlv as trailer data if hmac: # Since we can use sha1 or sha256 srhlen += 40 # Calculate and set hdrlen self['hdrlen'] = (srhlen >> 3) - 1 # Add seg6 type self['type'] = self.SEG6_TYPE # Add segments left self['segments_left'] = len(segs) - 1 # Add fitst segment self['first_segment'] = len(segs) - 1 # If hmac is used we have to set the flags if hmac: # Add SR6_FLAG1_HMAC self['flags'] \|= self.SR6_FLAG1_HMAC # Init segs self['segs'] = b'' # Iterate over segments for seg in segs: # Convert to network byte order and add to value self['segs'] += inet_pton(AF_INET6, seg) # Initialize tlvs self['tlvs'] = b'' # If hmac is used we have to properly init tlvs if hmac: # Put type self['tlvs'] += struct.pack('B', self.SR6_TLV_HMAC) # Put length -> 40-2 self['tlvs'] += struct.pack('B', 38) # Put reserved self['tlvs'] += struct.pack('H', 0) # Put hmac key self['tlvs'] += struct.pack('>I', hmac) # Put hmac self['tlvs'] += struct.pack('QQQQ', 0, 0, 0, 0) else: raise TypeError( 'Family %s not supported for seg6 tunnel' % family ) # Finally encode as nla nla.encode(self) # Utility function to verify if hmac is present def has_hmac(self): # Useful during the decoding return self['flags'] & self.SR6_FLAG1_HMAC def decode(self): # Decode the data nla.decode(self) # Extract the encap mode self['mode'] = self.r_encapmodes.get( self['encapmode'], "encap" ) # Calculate offset of the segs offset = self.offset + 16 # Point the addresses addresses = self.data[offset:] # Extract the number of segs n_segs = self['segments_left'] + 1 # Init segs segs = [] # Move 128 bit in each step for i in range(n_segs): # Save the segment segs.append( inet_ntop(AF_INET6, addresses[i * 16 : i * 16 + 16]) ) # Save segs self['segs'] = segs # Init tlvs self['tlvs'] = '' # If hmac is used if self.has_hmac(): # Point to the start of hmac hmac = addresses[n_segs * 16 : n_segs * 16 + 40] # Save tlvs section self['tlvs'] = hexdump(hmac) # Show also the hmac key self['hmac'] = hexdump(hmac[4:8]) class bpf_encap_info(nla): __slots__ = () nla_map = ( ('LWT_BPF_UNSPEC', 'none'), ('LWT_BPF_IN', 'bpf_obj'), ('LWT_BPF_OUT', 'bpf_obj'), ('LWT_BPF_XMIT', 'bpf_obj'), ('LWT_BPF_XMIT_HEADROOM', 'uint32'), ) class bpf_obj(nla): __slots__ = () nla_map = ( ('LWT_BPF_PROG_UNSPEC', 'none'), ('LWT_BPF_PROG_FD', 'uint32'), ('LWT_BPF_PROG_NAME', 'asciiz'), ) class seg6local_encap_info(nla): __slots__ = () nla_map = ( ('SEG6_LOCAL_UNSPEC', 'none'), ('SEG6_LOCAL_ACTION', 'action'), ('SEG6_LOCAL_SRH', 'ipv6_sr_hdr'), ('SEG6_LOCAL_TABLE', 'table'), ('SEG6_LOCAL_NH4', 'nh4'), ('SEG6_LOCAL_NH6', 'nh6'), ('SEG6_LOCAL_IIF', 'iif'), ('SEG6_LOCAL_OIF', 'oif'), ('SEG6_LOCAL_BPF', 'bpf_obj'), ('SEG6_LOCAL_VRFTABLE', 'vrf_table'), ) class bpf_obj(nla): __slots__ = () nla_map = ( ('LWT_BPF_PROG_UNSPEC', 'none'), ('LWT_BPF_PROG_FD', 'uint32'), ('LWT_BPF_PROG_NAME', 'asciiz'), ) class ipv6_sr_hdr(nla): __slots__ = () fields = ( ('nexthdr', 'B'), ('hdrlen', 'B'), ('type', 'B'), ('segments_left', 'B'), ('first_segment', 'B'), ('flags', 'B'), ('reserved', 'H'), ('segs', 's'), # Potentially several type-length-value ('tlvs', 's'), ) # Corresponding values for seg6 encap modes SEG6_IPTUN_MODE_INLINE = 0 SEG6_IPTUN_MODE_ENCAP = 1 # Mapping string to nla value encapmodes = { "inline": SEG6_IPTUN_MODE_INLINE, "encap": SEG6_IPTUN_MODE_ENCAP, } # Reverse mapping: mapping nla value to string r_encapmodes = {v: k for k, v in encapmodes.items()} # Nla value for seg6 type SEG6_TYPE = 4 # Flag value for hmac SR6_FLAG1_HMAC = 1 << 3 # Tlv value for hmac SR6_TLV_HMAC = 5 # Utility function to get the family from the msg def get_family(self): pointer = self while pointer.parent is not None: pointer = pointer.parent return pointer.get('family', AF_UNSPEC) def encode(self): # Retrieve the family family = self.get_family() # Seg6 can be applied only to IPv6 if family == AF_INET6: # Get mode mode = self['mode'] # Get segs segs = self['segs'] # Get hmac hmac = self.get('hmac', None) # With "inline" mode there is not # encap into an outer IPv6 header if mode == "inline": # Add :: to segs segs.insert(0, "::") # Add mode to value self['encapmode'] = self.encapmodes.get( mode, self.SEG6_IPTUN_MODE_ENCAP ) # Calculate srlen srhlen = 8 + 16 * len(segs) # If we are using hmac we have a tlv as trailer data if hmac: # Since we can use sha1 or sha256 srhlen += 40 # Calculate and set hdrlen self['hdrlen'] = (srhlen >> 3) - 1 # Add seg6 type self['type'] = self.SEG6_TYPE # Add segments left self['segments_left'] = len(segs) - 1 # Add fitst segment self['first_segment'] = len(segs) - 1 # If hmac is used we have to set the flags if hmac: # Add SR6_FLAG1_HMAC self['flags'] \|= self.SR6_FLAG1_HMAC # Init segs self['segs'] = b'' # Iterate over segments for seg in segs: # Convert to network byte order and add to value self['segs'] += inet_pton(family, seg) # Initialize tlvs self['tlvs'] = b'' # If hmac is used we have to properly init tlvs if hmac: # Put type self['tlvs'] += struct.pack('B', self.SR6_TLV_HMAC) # Put length -> 40-2 self['tlvs'] += struct.pack('B', 38) # Put reserved self['tlvs'] += struct.pack('H', 0) # Put hmac key self['tlvs'] += struct.pack('>I', hmac) # Put hmac self['tlvs'] += struct.pack('QQQQ', 0, 0, 0, 0) else: raise TypeError( 'Family %s not supported for seg6 tunnel' % family ) # Finally encode as nla nla.encode(self) # Utility function to verify if hmac is present def has_hmac(self): # Useful during the decoding return self['flags'] & self.SR6_FLAG1_HMAC def decode(self): # Decode the data nla.decode(self) # Extract the encap mode self['mode'] = self.r_encapmodes.get( self['encapmode'], "encap" ) # Calculate offset of the segs offset = self.offset + 16 # Point the addresses addresses = self.data[offset:] # Extract the number of segs n_segs = self['segments_left'] + 1 # Init segs segs = [] # Move 128 bit in each step for i in range(n_segs): # Save the segment segs.append( inet_ntop(AF_INET6, addresses[i * 16 : i * 16 + 16]) ) # Save segs self['segs'] = segs # Init tlvs self['tlvs'] = '' # If hmac is used if self.has_hmac(): # Point to the start of hmac hmac = addresses[n_segs * 16 : n_segs * 16 + 40] # Save tlvs section self['tlvs'] = hexdump(hmac) # Show also the hmac key self['hmac'] = hexdump(hmac[4:8]) class table(nla): __slots__ = () # Table ID fields = (('value', 'I'),) class action(nla): __slots__ = () # Action fields = (('value', 'I'),) SEG6_LOCAL_ACTION_UNSPEC = 0 SEG6_LOCAL_ACTION_END = 1 SEG6_LOCAL_ACTION_END_X = 2 SEG6_LOCAL_ACTION_END_T = 3 SEG6_LOCAL_ACTION_END_DX2 = 4 SEG6_LOCAL_ACTION_END_DX6 = 5 SEG6_LOCAL_ACTION_END_DX4 = 6 SEG6_LOCAL_ACTION_END_DT6 = 7 SEG6_LOCAL_ACTION_END_DT4 = 8 SEG6_LOCAL_ACTION_END_B6 = 9 SEG6_LOCAL_ACTION_END_B6_ENCAP = 10 SEG6_LOCAL_ACTION_END_BM = 11 SEG6_LOCAL_ACTION_END_S = 12 SEG6_LOCAL_ACTION_END_AS = 13 SEG6_LOCAL_ACTION_END_AM = 14 SEG6_LOCAL_ACTION_END_BPF = 15 SEG6_LOCAL_ACTION_END_DT46 = 16 actions = { 'End': SEG6_LOCAL_ACTION_END, 'End.X': SEG6_LOCAL_ACTION_END_X, 'End.T': SEG6_LOCAL_ACTION_END_T, 'End.DX2': SEG6_LOCAL_ACTION_END_DX2, 'End.DX6': SEG6_LOCAL_ACTION_END_DX6, 'End.DX4': SEG6_LOCAL_ACTION_END_DX4, 'End.DT6': SEG6_LOCAL_ACTION_END_DT6, 'End.DT4': SEG6_LOCAL_ACTION_END_DT4, 'End.B6': SEG6_LOCAL_ACTION_END_B6, 'End.B6.Encaps': SEG6_LOCAL_ACTION_END_B6_ENCAP, 'End.BM': SEG6_LOCAL_ACTION_END_BM, 'End.S': SEG6_LOCAL_ACTION_END_S, 'End.AS': SEG6_LOCAL_ACTION_END_AS, 'End.AM': SEG6_LOCAL_ACTION_END_AM, 'End.BPF': SEG6_LOCAL_ACTION_END_BPF, 'End.DT46': SEG6_LOCAL_ACTION_END_DT46, } def encode(self): # Get action type and convert string to value action = self['value'] self['value'] = self.actions.get( action, self.SEG6_LOCAL_ACTION_UNSPEC ) # Convert action type to u32 self['value'] = self['value'] & 0xFFFFFFFF # Finally encode as nla nla.encode(self) class iif(nla): __slots__ = () # Index of the incoming interface fields = (('value', 'I'),) class oif(nla): __slots__ = () # Index of the outcoming interface fields = (('value', 'I'),) class nh4(nla_string): __slots__ = () # Nexthop of the IPv4 family def encode(self): # Convert to network byte order self['value'] = inet_pton(AF_INET, self['value']) # Finally encode as nla nla_string.encode(self) def decode(self): # Decode the data nla_string.decode(self) # Convert the packed IP address to its string representation self['value'] = inet_ntop(AF_INET, self['value']) class nh6(nla_string): __slots__ = () # Nexthop of the IPv6 family def encode(self): # Convert to network byte order self['value'] = inet_pton(AF_INET6, self['value']) # Finally encode as nla nla_string.encode(self) def decode(self): # Decode the data nla_string.decode(self) # Convert the packed IP address to its string representation self['value'] = inet_ntop(AF_INET6, self['value']) class vrf_table(nla): __slots__ = () # VRF Table ID fields = (('value', 'I'),) # # TODO: add here other lwtunnel types # encaps = { LWTUNNEL_ENCAP_MPLS: mpls_encap_info, LWTUNNEL_ENCAP_SEG6: seg6_encap_info, LWTUNNEL_ENCAP_BPF: bpf_encap_info, LWTUNNEL_ENCAP_SEG6_LOCAL: seg6local_encap_info, } class rta_mfc_stats(nla): __slots__ = () fields = ( ('mfcs_packets', 'uint64'), ('mfcs_bytes', 'uint64'), ('mfcs_wrong_if', 'uint64'), ) class metrics(nla): __slots__ = () prefix = 'RTAX_' nla_map = ( ('RTAX_UNSPEC', 'none'), ('RTAX_LOCK', 'uint32'), ('RTAX_MTU', 'uint32'), ('RTAX_WINDOW', 'uint32'), ('RTAX_RTT', 'uint32'), ('RTAX_RTTVAR', 'uint32'), ('RTAX_SSTHRESH', 'uint32'), ('RTAX_CWND', 'uint32'), ('RTAX_ADVMSS', 'uint32'), ('RTAX_REORDERING', 'uint32'), ('RTAX_HOPLIMIT', 'uint32'), ('RTAX_INITCWND', 'uint32'), ('RTAX_FEATURES', 'uint32'), ('RTAX_RTO_MIN', 'uint32'), ('RTAX_INITRWND', 'uint32'), ('RTAX_QUICKACK', 'uint32'), ) @staticmethod def get_nh(self, argv, *kwarg): return nh class rtvia(nla_string): __slots__ = () sql_type = 'TEXT' def encode(self): family = self.get('family', AF_UNSPEC) if family in (AF_INET, AF_INET6): addr = inet_pton(family, self['addr']) else: raise TypeError('Family %s not supported for RTA_VIA' % family) self['value'] = struct.pack('H', family) + addr nla_string.encode(self) def decode(self): nla_string.decode(self) family = struct.unpack('H', self['value'][:2])[0] addr = self['value'][2:] if addr: if (family == AF_INET and len(addr) == 4) or ( family == AF_INET6 and len(addr) == 16 ): addr = inet_ntop(family, addr) else: addr = hexdump(addr) self.value = {'family': family, 'addr': addr} class cacheinfo(nla): __slots__ = () fields = ( ('rta_clntref', 'I'), ('rta_lastuse', 'I'), ('rta_expires', 'i'), ('rta_error', 'I'), ('rta_used', 'I'), ('rta_id', 'I'), ('rta_ts', 'I'), ('rta_tsage', 'I'), ) class rtmsg(rtmsg_base, nlmsg): __slots__ = () def encode(self): if self.get('family') == AF_MPLS: # force fields self['dst_len'] = 20 self['table'] = 254 self['type'] = 1 # assert NLA types for n in self.get('attrs', []): if n[0] not in ( 'RTA_OIF', 'RTA_DST', 'RTA_VIA', 'RTA_NEWDST', 'RTA_MULTIPATH', ): raise TypeError('Incorrect NLA type %s for AF_MPLS' % n[0]) super(rtmsg_base, self).encode() class nh(rtmsg_base, nla): __slots__ = () is_nla = False sql_constraints = {} cell_header = (('length', 'H'),) fields = (('flags', 'B'), ('hops', 'B'), ('oif', 'i'))	25.76 KB	2025-05-25 12:26
📁 tcmsg	-	2025-05-25 12:26

Edit File:

Terminal