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 -- import sys from pyroute2 import config from pyroute2.iproute.linux import RTNL_API, IPBatch # compatibility fix -- LNST: from pyroute2.netlink.rtnl import ( RTM_DELADDR, RTM_DELLINK, RTM_GETADDR, RTM_GETLINK, RTM_NEWADDR, RTM_NEWLINK, ) AsyncIPRoute = None if sys.platform.startswith('win'): from pyroute2.iproute.windows import ( ChaoticIPRoute, IPRoute, NetNS, RawIPRoute, ) elif sys.platform.startswith('darwin'): from pyroute2.iproute.darwin import ( ChaoticIPRoute, IPRoute, NetNS, RawIPRoute, ) elif config.uname[0][-3:] == 'BSD': from pyroute2.iproute.bsd import ChaoticIPRoute, IPRoute, NetNS, RawIPRoute else: from pyroute2.iproute.linux import ( AsyncIPRoute, ChaoticIPRoute, IPRoute, NetNS, RawIPRoute, ) classes = [ AsyncIPRoute, RTNL_API, IPBatch, IPRoute, RawIPRoute, ChaoticIPRoute, NetNS, ] constants = [ RTM_GETLINK, RTM_NEWLINK, RTM_DELLINK, RTM_GETADDR, RTM_NEWADDR, RTM_DELADDR, ]	1.11 KB	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 bsd.py ''' The library provides very basic RTNL API for BSD systems via protocol emulation. Only getters are supported yet, no setters. BSD employs PF_ROUTE sockets to send notifications about network object changes, but the protocol doesn not allow changing links/addresses/etc like Netlink. To change network setting one have to rely on system calls or external tools. Thus IPRoute on BSD systems is not as effective as on Linux, where all the changes are done via Netlink. The monitoring started with `bind()` is implemented as an implicit thread, started by the `bind()` call. This is done to have only one notification FD, used both for normal calls and notifications. This allows to use IPRoute objects in poll/select calls. On Linux systems RTNL API is provided by the netlink protocol, so no implicit threads are started by default to monitor the system updates. `IPRoute.bind(...)` may start the async cache thread, but only when asked explicitly:: # # Normal monitoring. Always starts monitoring thread on # FreeBSD / OpenBSD, no threads on Linux. # with IPRoute() as ipr: ipr.bind() ... # # Monitoring with async cache. Always starts cache thread # on Linux, ignored on FreeBSD / OpenBSD. # with IPRoute() as ipr: ipr.bind(async_cache=True) ... On all the supported platforms, be it Linux or BSD, the `IPRoute.recv(...)` method returns valid netlink RTNL raw binary payload and `IPRoute.get(...)` returns parsed RTNL messages. ''' import errno import os import queue import select import struct import threading from pyroute2 import config from pyroute2.bsd.pf_route import IFF_VALUES from pyroute2.bsd.rtmsocket import RTMSocket from pyroute2.bsd.util import ARP, Ifconfig, Route from pyroute2.netlink import NLM_F_DUMP, NLM_F_MULTI, NLM_F_REQUEST, NLMSG_DONE from pyroute2.netlink.rtnl import ( RTM_GETADDR, RTM_GETLINK, RTM_GETNEIGH, RTM_GETROUTE, RTM_NEWADDR, RTM_NEWLINK, RTM_NEWNEIGH, RTM_NEWROUTE, ) from pyroute2.netlink.rtnl.ifaddrmsg import ifaddrmsg from pyroute2.netlink.rtnl.ifinfmsg import IFF_NAMES, ifinfmsg from pyroute2.netlink.rtnl.marshal import MarshalRtnl from pyroute2.netlink.rtnl.ndmsg import ndmsg from pyroute2.netlink.rtnl.rtmsg import rtmsg class IPRoute(object): def __init__(self, argv, kwarg): if 'ssh' in kwarg: self._ssh = ['ssh', kwarg.pop('ssh')] else: self._ssh = [] async_qsize = kwarg.get('async_qsize') self._ifc = Ifconfig(cmd=self._ssh + ['ifconfig', '-a']) self._arp = ARP(cmd=self._ssh + ['arp', '-an']) self._route = Route(cmd=self._ssh + ['netstat', '-rn']) self.marshal = MarshalRtnl() self.target = kwarg.get('target') or 'localhost' self._mon_th = None self._rtm = None self._brd_socket = None self._pfdr, self._pfdw = os.pipe() # notify external poll/select self._ctlr, self._ctlw = os.pipe() # notify monitoring thread self._outq = queue.Queue(maxsize=async_qsize or config.async_qsize) self._system_lock = threading.Lock() self.closed = threading.Event() def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): self.close() def clone(self): return self def close(self, code=errno.ECONNRESET): with self._system_lock: if self.closed.is_set(): return if self._mon_th is not None: os.write(self._ctlw, b'\0') self._mon_th.join() self._rtm.close() if code > 0: self._outq.put(struct.pack('IHHQIQQ', 28, 2, 0, 0, code, 0, 0)) os.write(self._pfdw, b'\0') for ep in (self._pfdr, self._pfdw, self._ctlr, self._ctlw): try: os.close(ep) except OSError: pass self.closed.set() def bind(self, argv, *kwarg): with self._system_lock: if self._mon_th is not None: return if self._ssh: return self._mon_th = threading.Thread( target=self._monitor_thread, name='PF_ROUTE monitoring' ) self._mon_th.setDaemon(True) self._mon_th.start() def _monitor_thread(self): # Monitoring thread to convert arriving PF_ROUTE data into # the netlink format, enqueue it and notify poll/select. self._rtm = RTMSocket(output='netlink') inputs = [self._rtm.fileno(), self._ctlr] outputs = [] while True: try: events, _, _ = select.select(inputs, outputs, inputs) except: continue for fd in events: if fd == self._ctlr: # Main thread <-> monitor thread protocol is # pretty simple: discard the data and terminate # the monitor thread. os.read(self._ctlr, 1) return else: # Read the data from the socket and queue it msg = self._rtm.get() if msg is not None: msg.encode() self._outq.put(msg.data) # Notify external poll/select os.write(self._pfdw, b'\0') def fileno(self): # Every time when some new data arrives, one should write # into self._pfdw one byte to kick possible poll/select. # # Resp. recv() discards one byte from self._pfdr each call. return self._pfdr def get(self): data = self.recv() return self.marshal.parse(data) def recv(self, bufsize=None): os.read(self._pfdr, 1) return self._outq.get() def getsockopt(self, argv, *kwarg): return 1024 1024 def sendto_gate(self, msg, addr): # # handle incoming netlink requests # # sendto_gate() receives single RTNL messages as objects # cmd = msg['header']['type'] flags = msg['header']['flags'] seq = msg['header']['sequence_number'] # work only on dump requests for now if flags != NLM_F_REQUEST \| NLM_F_DUMP: return # if cmd == RTM_GETLINK: rtype = RTM_NEWLINK ret = self.get_links() elif cmd == RTM_GETADDR: rtype = RTM_NEWADDR ret = self.get_addr() elif cmd == RTM_GETROUTE: rtype = RTM_NEWROUTE ret = self.get_routes() elif cmd == RTM_GETNEIGH: rtype = RTM_NEWNEIGH ret = self.get_neighbours() # # set response type and finalize the message for r in ret: r['header']['type'] = rtype r['header']['flags'] = NLM_F_MULTI r['header']['sequence_number'] = seq # r = type(msg)() r['header']['type'] = NLMSG_DONE r['header']['sequence_number'] = seq ret.append(r) data = b'' for r in ret: r.encode() data += r.data self._outq.put(data) os.write(self._pfdw, b'\0') # 8<--------------------------------------------------------------- # def dump(self, groups=None): ''' Iterate all the objects -- links, routes, addresses etc. ''' for method in ( self.get_links, self.get_addr, self.get_neighbours, self.get_routes, ): for msg in method(): yield msg # 8<--------------------------------------------------------------- def get_links(self, argv, kwarg): ret = [] data = self._ifc.run() parsed = self._ifc.parse(data) for name, spec in parsed['links'].items(): msg = ifinfmsg().load(spec) msg['header']['type'] = RTM_NEWLINK msg['header']['target'] = self.target del msg['value'] flags = msg['flags'] new_flags = 0 for value, name in IFF_VALUES.items(): if value & flags and name in IFF_NAMES: new_flags \|= IFF_NAMES[name] msg['flags'] = new_flags ret.append(msg) return ret def get_addr(self, argv, *kwarg): ret = [] data = self._ifc.run() parsed = self._ifc.parse(data) for name, specs in parsed['addrs'].items(): for spec in specs: msg = ifaddrmsg().load(spec) msg['header']['type'] = RTM_NEWADDR msg['header']['target'] = self.target del msg['value'] ret.append(msg) return ret def get_neighbours(self, argv, *kwarg): ifc = self._ifc.parse(self._ifc.run()) arp = self._arp.parse(self._arp.run()) ret = [] for spec in arp: if spec['ifname'] not in ifc['links']: continue spec['ifindex'] = ifc['links'][spec['ifname']]['index'] msg = ndmsg().load(spec) msg['header']['type'] = RTM_NEWNEIGH msg['header']['target'] = self.target del msg['value'] ret.append(msg) return ret def get_routes(self, argv, *kwarg): ifc = self._ifc.parse(self._ifc.run()) rta = self._route.parse(self._route.run()) ret = [] for spec in rta: if spec['ifname'] not in ifc['links']: continue idx = ifc['links'][spec['ifname']]['index'] spec['attrs'].append(['RTA_OIF', idx]) msg = rtmsg().load(spec) msg['header']['type'] = RTM_NEWROUTE msg['header']['target'] = self.target del msg['value'] ret.append(msg) return ret class RawIPRoute(IPRoute): pass class ChaoticIPRoute: def __init__(self, argv, *kwarg): raise NotImplementedError() class NetNS: def __init__(self, argv, **kwarg): raise NotImplementedError()	10.05 KB	2025-05-25 12:26
📄 darwin.py import os import socket import subprocess from socket import AF_INET, AF_INET6 from pyroute2.common import dqn2int from pyroute2.netlink import NLM_F_DUMP, NLM_F_MULTI, NLM_F_REQUEST, NLMSG_DONE from pyroute2.netlink.rtnl import ( RTM_GETADDR, RTM_GETLINK, RTM_GETNEIGH, RTM_GETROUTE, RTM_NEWADDR, RTM_NEWLINK, RTM_NEWNEIGH, RTM_NEWROUTE, ) from pyroute2.netlink.rtnl.ifaddrmsg import ifaddrmsg from pyroute2.netlink.rtnl.ifinfmsg import ifinfmsg from pyroute2.netlink.rtnl.marshal import MarshalRtnl class IPRoute(object): """ macOS-specific IPRoute class (proof-of-concept). Attempts to fetch interfaces, addresses, neighbors, and routes by parsing macOS command output (ifconfig, netstat, arp, ndp, etc.). """ def __init__(self, argv, kwarg): self.marshal = MarshalRtnl() self.target = kwarg.get('target') or 'localhost' self._outq = kwarg.get('_outq', None) self._pfdw = kwarg.get('_pfdw', None) def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): self.close() def clone(self): return self def close(self, code=None): pass def bind(self, argv, *kwarg): pass def getsockopt(self, argv, *kwarg): return 1024 1024 def sendto_gate(self, msg, addr): """ Handle incoming netlink requests (simulation for macOS). """ cmd = msg['header']['type'] flags = msg['header']['flags'] seq = msg['header']['sequence_number'] # Only handle dump requests if flags != NLM_F_REQUEST \| NLM_F_DUMP: return if cmd == RTM_GETLINK: rtype = RTM_NEWLINK ret = self.get_links() elif cmd == RTM_GETADDR: rtype = RTM_NEWADDR ret = self.get_addr() elif cmd == RTM_GETROUTE: rtype = RTM_NEWROUTE ret = self.get_routes() elif cmd == RTM_GETNEIGH: rtype = RTM_NEWNEIGH ret = self.get_neighbours() else: ret = [] # Set response type and finalize for r in ret: r['header']['type'] = rtype r['header']['flags'] = NLM_F_MULTI r['header']['sequence_number'] = seq done_msg = type(msg)() done_msg['header']['type'] = NLMSG_DONE done_msg['header']['sequence_number'] = seq ret.append(done_msg) # If using an output queue (as done in Windows code), # write data there if self._outq and self._pfdw: data = b'' for r in ret: r.encode() data += r.data self._outq.put(data) os.write(self._pfdw, b'\0') def _parse_ifconfig(self): """ Parse `ifconfig` output to gather interface info, MAC, and addresses. Returns a dictionary: { 'interfaces': [ifinfmsg, ifinfmsg, ...], 'addresses': [ifaddrmsg, ifaddrmsg, ...] } """ ret = {'interfaces': [], 'addresses': []} try: output = subprocess.check_output(["ifconfig"]).decode( "utf-8", errors="replace" ) except (subprocess.CalledProcessError, FileNotFoundError): return ret blocks = output.strip().split("\n\n") for block in blocks: lines = block.strip().splitlines() if not lines: continue # The first line usually has "en0: flags=..." first_line = lines[0] iface_name = first_line.split(":")[0].strip() # Attempt to fetch interface index and MAC address try: iface_index = socket.if_nametoindex(iface_name) except OSError: iface_index = 0 # fallback if something goes wrong mac_addr = "00:00:00:00:00:00" for line in lines: # Lines with "ether " on macOS contain MAC, e.g.: # "ether 00:1c:42:aa:bb:cc" if line.strip().startswith("ether "): parts = line.strip().split() if len(parts) >= 2: mac_addr = parts[1] break # Build ifinfmsg spec_if = { 'index': iface_index, 'attrs': ( ['IFLA_ADDRESS', mac_addr], ['IFLA_IFNAME', iface_name], ), } msg_if = ifinfmsg().load(spec_if) msg_if['header']['target'] = self.target msg_if['header']['type'] = RTM_NEWLINK # remove 'value' if present msg_if.pop('value', None) ret['interfaces'].append(msg_if) # Look for IPv4 or IPv6 addresses # "inet 192.168.1.10 netmask 0xffffff00 broadcast 192.168.1.255" # "inet6 fe80::... prefixlen 64 ..." for line in lines[1:]: parts = line.strip().split() if line.strip().startswith("inet "): # IPv4 try: ip_idx = parts.index("inet") + 1 ip_addr = parts[ip_idx] mask_str = "255.255.255.0" # fallback if "netmask" in parts: mask_idx = parts.index("netmask") + 1 mask_hex = parts[mask_idx] if mask_hex.startswith("0x"): nm_value = int(mask_hex, 16) dotted_mask = [] for _ in range(4): dotted_mask.insert(0, str(nm_value & 0xFF)) nm_value >>= 8 mask_str = ".".join(dotted_mask) prefix_len = dqn2int(mask_str) spec_addr = { 'index': iface_index, 'family': AF_INET, 'prefixlen': prefix_len, 'attrs': ( ['IFA_ADDRESS', ip_addr], ['IFA_LOCAL', ip_addr], ['IFA_LABEL', iface_name], ), } msg_addr = ifaddrmsg().load(spec_addr) msg_addr['header']['target'] = self.target msg_addr['header']['type'] = RTM_NEWADDR msg_addr.pop('value', None) ret['addresses'].append(msg_addr) except (ValueError, IndexError): continue elif line.strip().startswith("inet6 "): # IPv6 try: ip_idx = parts.index("inet6") + 1 ip_addr = parts[ip_idx] prefix_len = 64 # default if not found # On macOS you'll often see # "inet6 fe80::xxxx prefixlen 64 ..." if "prefixlen" in parts: pre_idx = parts.index("prefixlen") + 1 prefix_len = int(parts[pre_idx]) spec_addr = { 'index': iface_index, 'family': AF_INET6, 'prefixlen': prefix_len, 'attrs': ( ['IFA_ADDRESS', ip_addr], ['IFA_LOCAL', ip_addr], ['IFA_LABEL', iface_name], ), } msg_addr = ifaddrmsg().load(spec_addr) msg_addr['header']['target'] = self.target msg_addr['header']['type'] = RTM_NEWADDR msg_addr.pop('value', None) ret['addresses'].append(msg_addr) except (ValueError, IndexError): continue return ret def _parse_routes(self): """ Parse `netstat -rn` to retrieve routing table entries. Return list of ifinfmsg-like route objects, though strictly they'd be `rtmsg` in a real Netlink environment. For demonstration, we just return RTM_NEWROUTE messages with minimal info: destination, gateway, interface index, etc. """ routes = [] try: output = subprocess.check_output(["netstat", "-rn"]).decode( "utf-8", errors="replace" ) except (subprocess.CalledProcessError, FileNotFoundError): return routes # On macOS, `netstat -rn` has headers like: # Destination Gateway Flags Netif Expire # default 192.168.1.1 UGSc en0 # 127 127.0.0.1 UCS lo0 # ... lines = output.strip().splitlines() # skip header lines # find the header first start_index = 0 for i, line in enumerate(lines): if "Destination" in line and "Gateway" in line: start_index = i + 1 break # parse from start_index onward for line in lines[start_index:]: parts = line.split() if len(parts) < 4: continue destination, gateway, flags, netif = parts[:4] # Attempt to get interface index try: idx = socket.if_nametoindex(netif) except OSError: idx = 0 # We'll store the route as ifinfmsg for # demonstration, but real netlink # code would use `rtmsg`. We'll keep it # consistent with the rest of this # POC approach. spec_route = { 'index': idx, # not strictly correct, but placing # them in attrs for demonstration 'attrs': ( ['ROUTE_DST', destination], ['ROUTE_GATEWAY', gateway], ['ROUTE_FLAGS', flags], ['ROUTE_IFNAME', netif], ), } # We'll build it as ifinfmsg with a 'type' = RTM_NEWROUTE msg_route = ifinfmsg().load(spec_route) msg_route['header']['target'] = self.target msg_route['header']['type'] = RTM_NEWROUTE msg_route.pop('value', None) routes.append(msg_route) return routes def _parse_arp_neighbors(self): """ Parse `arp -an` for IPv4 neighbors and (optionally) `ndp -an` for IPv6. Returns a list of ifinfmsg (or ifaddrmsg) objects simulating neighbor entries. """ neighbors = [] # IPv4 neighbors (ARP) try: output = subprocess.check_output(["arp", "-an"]).decode( "utf-8", errors="replace" ) # Lines look like: # "? (192.168.1.10) at 00:1c:42:xx:yy:zz on en0 ifscope [ethernet]" for line in output.strip().splitlines(): line = line.strip() if not line: continue # Quick parse: # 1) IP in parentheses # 2) MAC after "at" # 3) interface name after "on" parts = line.split() # example parts: # ["?", "(192.168.1.10)", # "at", "00:1c:42:xx:yy:zz", # "on", "en0", ...] if len(parts) < 7: continue ip_str = parts[1].strip("()") mac_str = parts[3] iface_str = parts[5] try: idx = socket.if_nametoindex(iface_str) except OSError: idx = 0 spec_neigh = { 'index': idx, 'attrs': ( ['NEIGH_IP', ip_str], ['NEIGH_LLADDR', mac_str], ['NEIGH_IFNAME', iface_str], ), } msg_neigh = ifinfmsg().load(spec_neigh) msg_neigh['header']['target'] = self.target msg_neigh['header']['type'] = RTM_NEWNEIGH msg_neigh.pop('value', None) neighbors.append(msg_neigh) except (subprocess.CalledProcessError, FileNotFoundError): pass # Optional: IPv6 neighbors (NDP) try: output = subprocess.check_output(["ndp", "-an"]).decode( "utf-8", errors="replace" ) # Lines look like: # "fe80::1%lo0 lladdr 00:00:00:... router STALE" for line in output.strip().splitlines(): line = line.strip() if not line: continue parts = line.split() if len(parts) < 4: continue # typical parts: # ["fe80::1%lo0", # "lladdr", "00:00:00:..", # "router", "STALE"] # or ["2001:db8::1234%en0", # "lladdr", # "00:11:22:33:44:55", # "REACHABLE"] addr_part = parts[0] # separate out the interface from the IP if "%" in addr_part: ip_str, iface_str = addr_part.split("%", 1) else: ip_str = addr_part iface_str = "??" mac_str = None if "lladdr" in parts: ll_idx = parts.index("lladdr") + 1 if ll_idx < len(parts): mac_str = parts[ll_idx] if not mac_str: continue try: idx = socket.if_nametoindex(iface_str) except OSError: idx = 0 spec_neigh = { 'index': idx, 'attrs': ( ['NEIGH_IP', ip_str], ['NEIGH_LLADDR', mac_str], ['NEIGH_IFNAME', iface_str], ), } msg_neigh = ifinfmsg().load(spec_neigh) msg_neigh['header']['target'] = self.target msg_neigh['header']['type'] = RTM_NEWNEIGH msg_neigh.pop('value', None) neighbors.append(msg_neigh) except (subprocess.CalledProcessError, FileNotFoundError): pass return neighbors def dump(self, groups=None): # Return all info in sequence for method in ( self.get_links, self.get_addr, self.get_neighbours, self.get_routes, ): for msg in method(): yield msg def get_links(self, argv, kwarg): ''' { 'attrs': (['IFLA_ADDRESS', '9a:9d:81:90:d2:8f'], ['IFLA_IFNAME', 'lo0']), 'header': {'target': 'localhost', 'type': 16}, 'index': 1 } ''' return self._parse_ifconfig()['interfaces'] def get_addr(self, argv, *kwarg): ''' { 'attrs': (['IFA_ADDRESS', '127.0.0.1'], ['IFA_LOCAL', '127.0.0.1'], ['IFA_LABEL', 'lo0']), 'header': {'target': 'localhost', 'type': 20}, 'index': 1, 'family': <AddressFamily.AF_INET: 2>, 'prefixlen': 8 } ''' return self._parse_ifconfig()['addresses'] def get_neighbours(self, argv, *kwarg): return self._parse_arp_neighbors() def get_routes(self, argv, *kwarg): return self._parse_routes() class RawIPRoute(IPRoute): pass class ChaoticIPRoute: """ Placeholder, mirroring its Windows counterpart. """ def __init__(self, argv, *kwarg): raise NotImplementedError() class NetNS: """ Another placeholder class to match the Windows code structure. """ def __init__(self, argv, **kwarg): raise NotImplementedError()	16.23 KB	2025-05-25 12:26
📄 ipmock.py import copy import errno import os import socket import struct import threading from itertools import count from pyroute2.config import AF_NETLINK from pyroute2.netlink import ( NLM_F_DUMP, NLM_F_MULTI, NLMSG_DONE, nlmsg, nlmsgerr, ) from pyroute2.netlink.core import Stats from pyroute2.netlink.rtnl import ( RTM_DELADDR, RTM_DELROUTE, RTM_GETADDR, RTM_GETLINK, RTM_GETROUTE, RTM_NEWADDR, RTM_NEWLINK, RTM_NEWROUTE, ) from pyroute2.netlink.rtnl.ifaddrmsg import ifaddrmsg from pyroute2.netlink.rtnl.ifinfmsg import ifinfmsg from pyroute2.netlink.rtnl.marshal import MarshalRtnl from pyroute2.netlink.rtnl.rtmsg import rtmsg interface_counter = count(3) MAGIC_CLOSE = 0x42 class MockLink: def __init__( self, index=0, ifname='', address='00:00:00:00:00:00', broadcast='ff:ff:ff:ff:ff:ff', perm_address=None, flags=1, rx_bytes=0, tx_bytes=0, rx_packets=0, tx_packets=0, mtu=0, qdisc='noqueue', kind=None, link=None, vlan_id=None, master=0, br_max_age=0, br_forward_delay=0, alt_ifname_list=None, ): self.index = index if index > 0 else next(interface_counter) self.ifname = ifname self.flags = flags self.address = address self.broadcast = broadcast self.perm_address = perm_address self.rx_bytes = rx_bytes self.tx_bytes = tx_bytes self.rx_packets = rx_packets self.tx_packets = tx_packets self.mtu = mtu self.qdisc = qdisc self.kind = kind self.link = link self.vlan_id = vlan_id self.master = master self.br_max_age = br_max_age self.br_forward_delay = br_forward_delay self.alt_ifname_list = alt_ifname_list or [] def update_from_msg(self, msg): [ setattr(self, x, msg.get(x)) for x in ['address', 'broadcast', 'mtu', 'ifname', 'master'] if msg.get(x) is not None ] # self.kind = msg.get(('linkinfo', 'kind')) if msg.get('change') != 0: self.flags = msg.get('flags') # vlan if self.kind == 'vlan': self.vlan_id = msg.get(('linkinfo', 'data', 'vlan_id')) self.link = msg.get('link') elif self.kind == 'bridge': self.br_max_age = msg.get(('linkinfo', 'data', 'br_max_age')) self.br_forward_delay = msg.get( ('linkinfo', 'data', 'br_forward_delay') ) @classmethod def load_from_msg(cls, msg): ret = cls() ret.update_from_msg(msg) return ret def export(self): ret = { 'attrs': [ ['IFLA_IFNAME', self.ifname], ['IFLA_TXQLEN', 1000], ['IFLA_OPERSTATE', 'UNKNOWN'], ['IFLA_LINKMODE', 0], ['IFLA_MTU', self.mtu], ['IFLA_GROUP', 0], ['IFLA_PROMISCUITY', 0], ['IFLA_NUM_TX_QUEUES', 1], ['IFLA_GSO_MAX_SEGS', 65535], ['IFLA_GSO_MAX_SIZE', 65536], ['IFLA_GRO_MAX_SIZE', 65536], ['IFLA_NUM_RX_QUEUES', 1], ['IFLA_CARRIER', 1], ['IFLA_QDISC', self.qdisc], ['IFLA_CARRIER_CHANGES', 0], ['IFLA_CARRIER_UP_COUNT', 0], ['IFLA_CARRIER_DOWN_COUNT', 0], ['IFLA_PROTO_DOWN', 0], [ 'IFLA_MAP', { 'base_addr': 0, 'dma': 0, 'irq': 0, 'mem_end': 0, 'mem_start': 0, 'port': 0, }, ], ['IFLA_ADDRESS', self.address], ['IFLA_BROADCAST', self.broadcast], [ 'IFLA_STATS64', { 'collisions': 0, 'multicast': 0, 'rx_bytes': self.rx_bytes, 'rx_compressed': 0, 'rx_crc_errors': 0, 'rx_dropped': 0, 'rx_errors': 0, 'rx_fifo_errors': 0, 'rx_frame_errors': 0, 'rx_length_errors': 0, 'rx_missed_errors': 0, 'rx_over_errors': 0, 'rx_packets': self.rx_packets, 'tx_aborted_errors': 0, 'tx_bytes': self.tx_bytes, 'tx_carrier_errors': 0, 'tx_compressed': 0, 'tx_dropped': 0, 'tx_errors': 0, 'tx_fifo_errors': 0, 'tx_heartbeat_errors': 0, 'tx_packets': self.tx_packets, 'tx_window_errors': 0, }, ], [ 'IFLA_STATS', { 'collisions': 0, 'multicast': 0, 'rx_bytes': self.rx_bytes, 'rx_compressed': 0, 'rx_crc_errors': 0, 'rx_dropped': 0, 'rx_errors': 0, 'rx_fifo_errors': 0, 'rx_frame_errors': 0, 'rx_length_errors': 0, 'rx_missed_errors': 0, 'rx_over_errors': 0, 'rx_packets': self.rx_packets, 'tx_aborted_errors': 0, 'tx_bytes': self.tx_bytes, 'tx_carrier_errors': 0, 'tx_compressed': 0, 'tx_dropped': 0, 'tx_errors': 0, 'tx_fifo_errors': 0, 'tx_heartbeat_errors': 0, 'tx_packets': self.tx_packets, 'tx_window_errors': 0, }, ], ['IFLA_XDP', {'attrs': [['IFLA_XDP_ATTACHED', None]]}], ( 'IFLA_PERM_ADDRESS', self.perm_address if self.perm_address else self.address, ), [ 'IFLA_AF_SPEC', { 'attrs': [ [ 'AF_INET', { 'accept_local': 0, 'accept_redirects': 1, 'accept_source_route': 0, 'arp_accept': 0, 'arp_announce': 0, 'arp_ignore': 0, 'arp_notify': 0, 'arpfilter': 0, 'bootp_relay': 0, 'dummy': 65672, 'force_igmp_version': 0, 'forwarding': 1, 'log_martians': 0, 'mc_forwarding': 0, 'medium_id': 0, 'nopolicy': 1, 'noxfrm': 1, 'promote_secondaries': 1, 'proxy_arp': 0, 'proxy_arp_pvlan': 0, 'route_localnet': 0, 'rp_filter': 2, 'secure_redirects': 1, 'send_redirects': 1, 'shared_media': 1, 'src_vmark': 0, 'tag': 0, }, ] ] }, ], ], 'change': 0, 'event': 'RTM_NEWLINK', 'family': 0, 'flags': self.flags, 'header': { 'error': None, 'flags': 2, 'length': 1364, 'pid': 303471, 'sequence_number': 260, 'stats': Stats(qsize=0, delta=0, delay=0), 'target': 'localhost', 'type': 16, }, 'ifi_type': 772, 'index': self.index, 'state': 'up' if self.flags & 1 else 'down', } linkinfo = None infodata = {'attrs': []} if self.kind is not None: linkinfo = {'attrs': [('IFLA_INFO_KIND', self.kind)]} if self.kind not in (None, 'dummy'): linkinfo['attrs'].append(('IFLA_INFO_DATA', infodata)) if self.kind == 'vlan': infodata['attrs'].append(('IFLA_VLAN_ID', self.vlan_id)) ret['attrs'].append(('IFLA_LINK', self.link)) if self.kind == 'bridge': infodata['attrs'].extend( ( ('IFLA_BR_MAX_AGE', self.br_max_age), ('IFLA_BR_FORWARD_DELAY', self.br_forward_delay), ) ) if linkinfo is not None: ret['attrs'].append(('IFLA_LINKINFO', linkinfo)) if self.master != 0: ret['attrs'].append(('IFLA_MASTER', self.master)) return ret class MockAddress: def __init__( self, index=0, address=None, prefixlen=None, broadcast=None, label=None, family=2, local=None, kwarg, ): self.address = address self.local = local self.broadcast = broadcast self.prefixlen = prefixlen self.index = index self.label = label self.family = family def update_from_msg(self, msg): [ setattr(self, x, msg.get(x)) for x in ['index', 'address', 'broadcast', 'prefixlen'] if msg.get(x) is not None ] @classmethod def load_from_msg(cls, msg): ret = cls() ret.update_from_msg(msg) return ret def export(self): ret = { 'family': self.family, 'prefixlen': self.prefixlen, 'flags': 0, 'scope': 0, 'index': self.index, 'attrs': [ ('IFA_ADDRESS', self.address), ('IFA_LOCAL', self.local if self.local else self.address), ('IFA_FLAGS', 512), ( 'IFA_CACHEINFO', { 'ifa_preferred': 3476, 'ifa_valid': 3476, 'cstamp': 138655779, 'tstamp': 141288674, }, ), ], 'header': { 'length': 88, 'type': 20, 'flags': 2, 'sequence_number': 256, 'pid': 320994, 'error': None, 'target': 'localhost', 'stats': Stats(qsize=0, delta=0, delay=0), }, 'event': 'RTM_NEWADDR', } if self.label is not None: ret['attrs'].append(('IFA_LABEL', self.label)) if self.broadcast is not None: ret['attrs'].append(('IFA_BROADCAST', self.broadcast)) return ret class MockRoute: def __init__( self, dst=None, oif=0, gateway=None, prefsrc=None, family=2, dst_len=24, table=254, scope=253, proto=2, route_type=1, kwarg, ): self.dst = dst self.gateway = gateway self.prefsrc = prefsrc self.oif = oif self.family = family self.dst_len = dst_len self.table = table self.scope = scope self.proto = proto self.route_type = route_type self.priority = kwarg.get('priority', 0) self.tos = kwarg.get('tos', 0) self._type = kwarg.get('type', 2) def update_from_msg(self, msg): [ setattr(self, x, msg.get(x)) for x in [ 'dst', 'dst_len', 'gateway', 'oif', 'family', 'table', 'priority', ] if msg.get(x) is not None ] @classmethod def load_from_msg(cls, msg): ret = cls() ret.update_from_msg(msg) return ret def export(self): ret = { 'family': self.family, 'dst_len': self.dst_len, 'src_len': 0, 'tos': self.tos, 'table': self.table if self.table <= 255 else 252, 'proto': self.proto, 'scope': self.scope, 'type': self._type, 'flags': 0, 'attrs': [('RTA_TABLE', self.table), ('RTA_OIF', self.oif)], 'header': { 'length': 60, 'type': 24, 'flags': 2, 'sequence_number': 255, 'pid': 325359, 'error': None, 'target': 'localhost', 'stats': Stats(qsize=0, delta=0, delay=0), }, 'event': 'RTM_NEWROUTE', } if self.dst is not None: ret['attrs'].append(('RTA_DST', self.dst)) if self.prefsrc is not None: ret['attrs'].append(('RTA_PREFSRC', self.prefsrc)) if self.gateway is not None: ret['attrs'].append(('RTA_GATEWAY', self.gateway)) if self.priority > 0: ret['attrs'].append(('RTA_PRIORITY', self.priority)) return ret presets = { 'default': { 'links': [ MockLink( index=1, ifname='lo', address='00:00:00:00:00:00', broadcast='00:00:00:00:00:00', rx_bytes=43309665, tx_bytes=43309665, rx_packets=173776, tx_packets=173776, mtu=65536, qdisc='noqueue', ), MockLink( index=2, ifname='eth0', address='52:54:00:72:58:b2', broadcast='ff:ff:ff:ff:ff:ff', rx_bytes=175340, tx_bytes=175340, rx_packets=10251, tx_packets=10251, mtu=1500, qdisc='fq_codel', ), ], 'addr': [ MockAddress( index=1, label='lo', address='127.0.0.1', broadcast='127.255.255.255', prefixlen=8, ), MockAddress( index=2, label='eth0', address='192.168.122.28', broadcast='192.168.122.255', prefixlen=24, ), ], 'routes': [ MockRoute( dst=None, gateway='192.168.122.1', oif=2, dst_len=0, table=254, scope=0, ), MockRoute(dst='192.168.122.0', oif=2, dst_len=24, table=254), MockRoute( dst='127.0.0.0', oif=1, dst_len=8, table=255, route_type=2 ), MockRoute( dst='127.0.0.1', oif=1, dst_len=32, table=255, route_type=2 ), MockRoute( dst='127.255.255.255', oif=1, dst_len=32, table=255, route_type=3, ), MockRoute( dst='192.168.122.28', oif=2, dst_len=32, table=255, route_type=2, ), MockRoute( dst='192.168.122.255', oif=2, dst_len=32, table=255, route_type=3, ), ], }, 'netns': { 'links': [ MockLink( index=1, ifname='lo', address='00:00:00:00:00:00', broadcast='00:00:00:00:00:00', rx_bytes=43309665, tx_bytes=43309665, rx_packets=173776, tx_packets=173776, mtu=65536, qdisc='noqueue', ) ], 'addr': [ MockAddress( index=1, label='lo', address='127.0.0.1', broadcast='127.255.255.255', prefixlen=8, ) ], 'routes': [ MockRoute( dst='127.0.0.0', oif=1, dst_len=8, table=255, route_type=2 ), MockRoute( dst='127.0.0.1', oif=1, dst_len=32, table=255, route_type=2 ), MockRoute( dst='127.255.255.255', oif=1, dst_len=32, table=255, route_type=3, ), ], }, } class IPEngine: '''Mock network objects database with the socket API. WIP: work in progress. A drop-in replacement to use instead of a low level RTNL socket. Implements all the required socket properties and provides a network objects database with RTNL protocol. Example:: >>> ipe = IPEngine() >>> ipr = IPRoute(use_socket=ipe) >>> [ x.get('ifname') for x in ipr.link('dump') ] ['lo', 'eth0'] ''' def __init__( self, sfamily=AF_NETLINK, stype=socket.SOCK_DGRAM, sproto=0, netns='default', flags=os.O_CREAT, ): self.marshal = MarshalRtnl() self.netns = netns self.flags = flags self.magic = 0 self._stype = stype self._sfamily = sfamily self._sproto = sproto self._local = threading.local() self._lock = threading.Lock() self._broadcast = set() self.processors = { RTM_GETADDR: self.RTM_GETADDR, RTM_GETLINK: self.RTM_GETLINK, RTM_NEWADDR: self.RTM_NEWADDR, RTM_DELADDR: self.RTM_DELADDR, RTM_NEWLINK: self.RTM_NEWLINK, RTM_DELROUTE: self.RTM_DELROUTE, RTM_NEWROUTE: self.RTM_NEWROUTE, RTM_GETROUTE: self.RTM_GETROUTE, } self.initdb() @property def loopback_r(self): if not hasattr(self._local, 'loopback_r'): self._local.loopback_r, self._local.loopback_w = socket.socketpair( socket.AF_UNIX, socket.SOCK_DGRAM ) return self._local.loopback_r @property def loopback_w(self): if not hasattr(self._local, 'loopback_w'): self._local.loopback_r, self._local.loopback_w = socket.socketpair( socket.AF_UNIX, socket.SOCK_DGRAM ) return self._local.loopback_w def initdb(self): if self.netns not in presets: if not self.flags & os.O_CREAT: raise FileNotFoundError() presets[self.netns] = copy.deepcopy(presets['netns']) self.database = copy.deepcopy(presets[self.netns]) def close(self): if self.magic == MAGIC_CLOSE: self.loopback_r.close() self.loopback_w.close() def bind(self, address=None): self._broadcast.add(self.loopback_w) msg = rtmsg() msg.load(self.database['routes'][0].export()) msg.encode() self.loopback_w.send(msg.data) def fileno(self): return self.loopback_r.fileno() def recv(self, bufsize, flags=0): return self.loopback_r.recv(bufsize, flags) def recvfrom(self, bufsize, flags=0): return self.loopback_r.recvfrom(bufsize, flags) def recvmsg(self, bufsize, ancbufsize=0, flags=0): return self.loopback_r.recvmsg(bufsize, ancbufsize, flags) def recv_into(self, buffer, nbytes=0, flags=0): return self.loopback_r.recv_into(buffer, nbytes, flags) def recvfrom_into(self, buffer, nbytes=0, flags=0): return self.loopback_r.recvfrom_into(buffer, nbytes, flags) def recvmsg_into(self, buffers, ancbufsize=0, flags=0): return self.loopback_r.recvmsg_into(buffers, ancbufsize, flags) def send(self, data, flags=0): return self.nl_handle(data) def sendall(self, data, flags=0): return self.nl_handle(data) def sendto(self, data, flags, address=0): return self.nl_handle(data) def sendmsg(self, buffers, ancdata=None, flags=None, address=None): raise NotImplementedError() def setblocking(self, flag): return self.loopback_r.setblocking(flag) def getblocking(self): return self.loopback_r.getblocking() def getsockname(self): return self.loopback_r.getsockname() def getpeername(self): return self.loopback_r.getpeername() @property def type(self): return self._stype @property def family(self): return self._sfamily @property def proto(self): return self._sproto def nl_handle(self, data): with self._lock: for msg in self.marshal.parse(data): key = msg['header']['type'] tag = msg['header']['sequence_number'] if key in self.processors: self.processors[key](msg) else: self.nl_done(tag) return len(data) def nl_dump(self, registry, msg_class, tag): for item in registry: msg = msg_class() msg.load(item.export()) msg['header']['flags'] = NLM_F_MULTI msg['header']['sequence_number'] = tag msg.encode() yield msg def nl_broadcast(self, msg): for sock in self._broadcast: msg['header']['sequence_number'] = 0 msg.reset() msg.encode() sock.send(msg.data) def nl_done(self, tag): msg = nlmsg() msg['header']['type'] = NLMSG_DONE msg['header']['sequence_number'] = tag msg.encode() self.loopback_w.send(msg.data) def nl_error(self, tag, code): msg = nlmsgerr() msg['header']['sequence_number'] = tag msg['error'] = code msg.encode() self.loopback_w.send(msg.data) def RTM_GETROUTE(self, req): tag = req['header']['sequence_number'] for msg in self.nl_dump(self.database['routes'], rtmsg, tag): self.loopback_w.send(msg.data) self.nl_done(tag) def RTM_GETADDR(self, msg_in): tag = msg_in['header']['sequence_number'] for msg_out in self.nl_dump(self.database['addr'], ifaddrmsg, tag): self.loopback_w.send(msg_out.data) self.nl_done(tag) def RTM_GETLINK(self, msg_in): tag = msg_in['header']['sequence_number'] database = self.database['links'] if msg_in.get('index') > 0: database = [ x for x in self.database['links'] if x.index == msg_in.get('index') ] elif msg_in.get('ifname'): database = [ x for x in self.database['links'] if x.ifname == msg_in.get('ifname') ] for msg_out in self.nl_dump(database, ifinfmsg, tag): self.loopback_w.send(msg_out.data) if msg_in.get(('header', 'flags')) & NLM_F_DUMP: return self.nl_done(tag) self.nl_error(tag, 0) def RTM_NEWROUTE(self, req): tag = req['header']['sequence_number'] if not req.get('oif'): (gateway,) = struct.unpack( '>I', socket.inet_aton(req.get('gateway')) ) for route in self.database['routes']: if route.dst is None: continue (dst,) = struct.unpack('>I', socket.inet_aton(route.dst)) if (gateway & (0xFFFFFFFF << (32 - route.dst_len))) == dst: req['attrs'].append(('RTA_OIF', route.oif)) break else: return self.nl_error(tag, errno.ENOENT) idx = { (x.dst, x.dst_len, x.oif, x.priority, x.gateway, x.table): x for x in self.database['routes'] } req_index = ( req.get('dst'), req.get('dst_len'), req.get('oif'), req.get('priority'), req.get('gateway'), req.get('IFLA_TABLE') or req.get('table'), ) if req_index in idx: return self.nl_error(tag, errno.EEXIST) route = MockRoute.load_from_msg(req) self.database['routes'].append(route) self.nl_error(tag, 0) msg = rtmsg() msg.load(route.export()) self.nl_broadcast(msg) def RTM_DELROUTE(self, req): idx = { (x.dst, x.dst_len, x.oif, x.priority, x.table): x for x in self.database['routes'] } req_index = ( req.get('dst'), req.get('dst_len'), req.get('oif'), req.get('priority'), req.get('IFLA_TABLE') or req.get('table'), ) tag = req['header']['sequence_number'] if req_index not in idx: return self.nl_error(tag, errno.ENOENT) self.database['routes'].remove(idx[req_index]) self.nl_error(tag, 0) self.nl_broadcast(req) def RTM_DELADDR(self, req): idx = { (x.index, x.address, x.prefixlen): x for x in self.database['addr'] } req_index = ( req.get("index"), req.get("address"), req.get("prefixlen"), ) tag = req['header']['sequence_number'] if req_index not in idx: return self.nl_error(tag, errno.ENOENT) self.database['addr'].remove(idx[req_index]) self.nl_error(tag, 0) self.nl_broadcast(req) def RTM_NEWADDR(self, req): idx = { (x.index, x.address, x.prefixlen) for x in self.database['addr'] } req_index = ( req.get("index"), req.get("address"), req.get("prefixlen"), ) tag = req['header']['sequence_number'] if req_index in idx: return self.nl_error(tag, errno.EEXIST) addr = MockAddress.load_from_msg(req) self.database['addr'].append(addr) self.nl_error(tag, 0) msg = ifaddrmsg() msg.load(addr.export()) self.nl_broadcast(msg) def RTM_NEWLINK(self, req): idx = {x.index: x for x in self.database['links']} nmx = {x.ifname: x for x in self.database['links']} tag = req['header']['sequence_number'] if req.get('index') in idx: link = idx[req.get('index')] if link.ifname == req.get('ifname'): return self.nl_error(tag, errno.EEXIST) link.update_from_msg(req) elif req.get('ifname') in nmx: return self.nl_error(tag, errno.EEXIST) else: link = MockLink.load_from_msg(req) self.database['links'].append(link) self.nl_error(tag, 0) msg = ifinfmsg() msg.load(link.export()) self.nl_broadcast(msg)	27.48 KB	2025-05-25 12:26
📄 linux.py # -- coding: utf-8 -- import asyncio import io import logging import os import struct import time import warnings from functools import partial from socket import AF_INET, AF_INET6, AF_UNSPEC from pyroute2 import netns from pyroute2.common import AF_MPLS, basestring from pyroute2.config import AF_BRIDGE from pyroute2.netlink import NLM_F_ACK, NLM_F_DUMP, NLM_F_REQUEST, NLMSG_ERROR from pyroute2.netlink.exceptions import ( NetlinkDumpInterrupted, NetlinkError, SkipInode, ) from pyroute2.netlink.nlsocket import NetlinkRequest, NetlinkSocket from pyroute2.netlink.rtnl import ( RTM_DELADDR, RTM_DELLINK, RTM_DELLINKPROP, RTM_DELNEIGH, RTM_DELQDISC, RTM_DELROUTE, RTM_DELRULE, RTM_DELTCLASS, RTM_DELTFILTER, RTM_GETADDR, RTM_GETLINK, RTM_GETNEIGH, RTM_GETNEIGHTBL, RTM_GETNSID, RTM_GETQDISC, RTM_GETROUTE, RTM_GETRULE, RTM_GETSTATS, RTM_GETTCLASS, RTM_GETTFILTER, RTM_NEWADDR, RTM_NEWLINK, RTM_NEWLINKPROP, RTM_NEWNEIGH, RTM_NEWNETNS, RTM_NEWNSID, RTM_NEWPROBE, RTM_NEWQDISC, RTM_NEWROUTE, RTM_NEWRULE, RTM_NEWTCLASS, RTM_NEWTFILTER, RTM_SETLINK, RTMGRP_DEFAULTS, RTMGRP_IPV4_IFADDR, RTMGRP_IPV4_ROUTE, RTMGRP_IPV4_RULE, RTMGRP_IPV6_IFADDR, RTMGRP_IPV6_ROUTE, RTMGRP_IPV6_RULE, RTMGRP_LINK, RTMGRP_MPLS_ROUTE, RTMGRP_NEIGH, ndmsg, ) 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.iprsocket import AsyncIPRSocket, ChaoticIPRSocket from pyroute2.netlink.rtnl.ndtmsg import ndtmsg from pyroute2.netlink.rtnl.nsidmsg import nsidmsg from pyroute2.netlink.rtnl.nsinfmsg import nsinfmsg from pyroute2.netlink.rtnl.probe_msg import probe_msg from pyroute2.netlink.rtnl.rtmsg import rtmsg from pyroute2.netlink.rtnl.tcmsg import plugins as tc_plugins from pyroute2.netlink.rtnl.tcmsg import tcmsg from pyroute2.requests.address import AddressFieldFilter, AddressIPRouteFilter from pyroute2.requests.bridge import ( BridgeFieldFilter, BridgeIPRouteFilter, BridgePortFieldFilter, ) from pyroute2.requests.link import LinkFieldFilter, LinkIPRouteFilter from pyroute2.requests.main import RequestProcessor from pyroute2.requests.neighbour import ( NeighbourFieldFilter, NeighbourIPRouteFilter, ) from pyroute2.requests.probe import ProbeFieldFilter from pyroute2.requests.route import RouteFieldFilter, RouteIPRouteFilter from pyroute2.requests.rule import RuleFieldFilter, RuleIPRouteFilter from pyroute2.requests.tc import TcIPRouteFilter, TcRequestFilter from .parsers import default_routes, export_routes DEFAULT_TABLE = 254 IPROUTE2_DUMP_MAGIC = 0x45311224 log = logging.getLogger(__name__) def get_default_request_filters(mode, command): filters = { 'link': [LinkFieldFilter(), LinkIPRouteFilter(command)], 'addr': [AddressFieldFilter(), AddressIPRouteFilter(command)], 'neigh': [NeighbourFieldFilter(), NeighbourIPRouteFilter(command)], 'route': [ RouteFieldFilter(add_defaults=(command not in ('dump', 'show'))), RouteIPRouteFilter(command), ], 'rule': [RuleFieldFilter(), RuleIPRouteFilter(command)], 'tc': [TcRequestFilter(), TcIPRouteFilter(command)], 'brport': [BridgePortFieldFilter(command)], 'vlan_filter': [BridgeFieldFilter(), BridgeIPRouteFilter(command)], 'probe': [ProbeFieldFilter()], 'stats': [], } return filters[mode] def get_dump_filter(mode, command, query): if 'dump_filter' in query: return query.pop('dump_filter'), query if command not in ('dump', 'show'): return RequestProcessor(), query new_query = {} if 'family' in query: new_query['family'] = query.pop('family') if 'ext_mask' in query: new_query['ext_mask'] = query.pop('ext_mask') if 'match' in query: query = query['match'] if callable(query): return query, {} dump_filter = RequestProcessor(context=query, prime=query) for rf in query.pop( 'dump_filter', get_default_request_filters(mode, command) ): dump_filter.add_filter(rf) dump_filter.finalize() return dump_filter, new_query def get_arguments_processor(mode, command, query, parameters=None): if 'request_filter' in query: return query['request_filter'] parameters = parameters or {} # if not query: # return RequestProcessor() processor = RequestProcessor(context=query, prime=query) for pname, pvalue in parameters.items(): processor.set_parameter(pname, pvalue) for rf in get_default_request_filters(mode, command): processor.add_filter(rf) processor.finalize() return processor def transform_handle(handle): if isinstance(handle, basestring): (major, minor) = [int(x if x else '0', 16) for x in handle.split(':')] handle = (major << 8 * 2) \| minor return handle class RTNL_API: '''A mixin RTNL API class. `RTNL_API` should not be instantiated by itself, it is intended to be used as a mixin class. Following classes use `RTNL_API`: * `AsyncIPRoute` -- Asynchronous RTNL API * `IPRoute` -- Synchronous RTNL API * `NetNS` -- Legace netns-enabled RTNL API This class was started as iproute2 ip/tc equivalent, but as a Python API. It does not provide any complicated logic, but instead runs simple RTNL queries: It is an old-school API, that provides access to rtnetlink as is. It helps you to retrieve and change almost all the data, available through rtnetlink: .. testcode:: cls01 from pyroute2 import IPRoute ipr = IPRoute() # create an interface ipr.link("add", ifname="brx", kind="bridge") # lookup the index dev = ipr.link_lookup(ifname="brx")[0] # bring it up ipr.link("set", index=dev, state="up") # change the interface MAC address and rename it just for fun ipr.link( "set", index=dev, address="00:11:22:33:44:55", ifname="br-ctrl", ) # add primary IP address ipr.addr( "add", index=dev, address="10.0.0.1", prefixlen=24, broadcast="10.0.0.255", ) # add secondary IP address ipr.addr( "add", index=dev, address="10.0.0.2", prefixlen=24, broadcast="10.0.0.255", ) .. testcode:: cls01 :hide: br = ipr.link("get", index=dev)[0] assert br.get("flags") & 1 try: assert br.get("flags") & 2 except AssertionError: pass assert br.get("ifname") == "br-ctrl" assert br.get("address") == "00:11:22:33:44:55" addr1, addr2 = ipr.addr("dump", index=dev) assert addr1.get("address") == "10.0.0.1" assert addr2.get("address") == "10.0.0.2" assert addr1.get("broadcast") == "10.0.0.255" assert addr2.get("broadcast") == "10.0.0.255" assert addr1.get("prefixlen") == 24 assert addr2.get("prefixlen") == 24 ''' def filter_messages(self, dump_filter, msgs): ''' Filter messages using `dump_filter`. The filter might be a callable, then it will be called for every message in the list. Or it might be a dict, where keys are used to get values from messages, and dict values are used to match the message. The method might be called directly. It is also used by calls like `ipr.link('dump', ....)`, where keyword arguments work as `dump_filter` for `ipr.filter_messages()`. A callable `dump_filter` must return True or False: .. testcode:: fm01 # get all links with names starting with eth: # for link in ipr.filter_messages( lambda x: x.get("ifname").startswith("eth"), ipr.link("dump"), ): print(link.get("ifname")) .. testoutput:: fm01 eth0 A dict `dump_filter` can have callables as values: .. testcode:: fm02 # get all links with names starting with eth, and # MAC address in a database: # database = [ "52:54:00:72:58:b2", ] for link in ipr.filter_messages( { "ifname": lambda x: x.startswith("eth"), "address": lambda x: x in database, }, ipr.link("dump"), ): print(link.get("ifname")) .. testoutput:: fm02 eth0 ... or constants to compare with: .. testcode:: fm03 # get all links in state up: # for link in ipr.filter_messages( {"state": "up"}, ipr.link("dump"), ): print(link.get("ifname")) .. testoutput:: fm03 lo eth0 ''' # filtered results, the generator version for msg in msgs: if NetlinkRequest.match_one_message(dump_filter, msg): yield msg # 8<--------------------------------------------------------------- # async def dump(self, groups=None): ''' Dump network objects. * get_links() * get_addr() * get_neighbours() * get_vlans() * dump FDB * IPv4 and IPv6 rules ''' groups_map = { RTMGRP_LINK: [ self.get_links, self.get_vlans, partial(self.fdb, 'dump'), ], RTMGRP_IPV4_IFADDR: [partial(self.get_addr, family=AF_INET)], RTMGRP_IPV6_IFADDR: [partial(self.get_addr, family=AF_INET6)], RTMGRP_NEIGH: [self.get_neighbours], RTMGRP_IPV4_ROUTE: [partial(self.get_routes, family=AF_INET)], RTMGRP_IPV6_ROUTE: [partial(self.get_routes, family=AF_INET6)], RTMGRP_MPLS_ROUTE: [partial(self.get_routes, family=AF_MPLS)], RTMGRP_IPV4_RULE: [partial(self.get_rules, family=AF_INET)], RTMGRP_IPV6_RULE: [partial(self.get_rules, family=AF_INET6)], } async def ret(): for group, methods in groups_map.items(): if group & (groups if groups is not None else self.groups): for method in methods: async for msg in await method(): yield msg return ret() async def poll(self, method, command, timeout=10, interval=0.2, spec): '''Wait for a method to succeed. Run `method` with a positional argument `command` and keyword arguments `spec` every `interval` seconds, but not more than `timeout`, until it returns a result which doesn't evaluate to `False`. Example: .. testcode:: p0 # create a bridge interface and wait for it: # spec = { 'ifname': 'br0', 'kind': 'bridge', 'state': 'up', } ipr.link('add', spec) ret = ipr.poll(ipr.link, 'dump', spec) assert ret[0].get('ifname') == 'br0' assert ret[0].get('flags') & 1 assert ret[0].get('state') == 'up' assert ret[0].get(('linkinfo', 'kind')) == 'bridge' .. testcode:: p1 :hide: try: ipr.poll(ipr.link, 'dump', ifname='br1') except TimeoutError: pass ''' ctime = time.time() ret = tuple() while ctime + timeout > time.time(): try: ret = await method(command, spec) if not isinstance(ret, list): ret = [x async for x in ret] if ret: return ret await asyncio.sleep(interval) except NetlinkDumpInterrupted: pass raise asyncio.TimeoutError() # 8<--------------------------------------------------------------- # # Diagnostics # async def probe(self, command, kwarg): '''Run a network probe. The API will trigger a network probe from the environment it works in. For NetNS it will be the network namespace, for remote IPRoute instances it will be the host it runs on. Running probes via API allows to test network connectivity between the environments in a simple uniform way. Supported arguments: * kind -- probe type, for now only ping is supported * dst -- target to run the probe against * num -- number of probes to run * timeout -- timeout for the whole request Examples:: ipr.probe("add", kind="ping", dst="10.0.0.1") By default ping probe will send one ICMP request towards the target. To change this, use num argument:: ipr.probe( "add", kind="ping", dst="10.0.0.1", num=4, timeout=10 ) Timeout for the ping probe by default is 1 second, which may not be enough to run multiple requests. In the next release more probe types are planned, like TCP port probe. ''' msg = probe_msg() arguments = get_arguments_processor('probe', command, kwarg) request = NetlinkRequest( self, msg, msg_type=RTM_NEWPROBE, msg_flags=1, request_filter=arguments, ) await request.send() return [x async for x in request.response()] # 8<--------------------------------------------------------------- # # Binary streams methods # async def route_dump(self, fd, family=AF_UNSPEC, fmt='iproute2'): '''Save routes as a binary dump into a file object. fd -- an open file object, must support `write()` family -- AF_UNSPEC, AF_INET, etc. -- filter routes by family fmt -- dump format, "iproute2" (default) or "raw" The binary dump is just a set of unparsed netlink messages. The `iproute2` prepends the dump with a magic uint32, so `IPRoute` does the same for compatibility. If you want a raw dump without any additional magic data, use `fmt="raw"`. This routine neither close the file object, nor uses `seek()` to rewind, it's up to the user. ''' if fmt == 'iproute2': fd.write(struct.pack('I', IPROUTE2_DUMP_MAGIC)) elif fmt != 'raw': raise TypeError('dump format not supported') msg = rtmsg() msg['family'] = family request = NetlinkRequest( self, msg, msg_type=RTM_GETROUTE, msg_flags=NLM_F_DUMP \| NLM_F_REQUEST, parser=export_routes(fd), ) await request.send() return [x async for x in request.response()] async def route_dumps(self, family=AF_UNSPEC, fmt='iproute2'): '''Save routes and returns as a `bytes` object. The same as `.route_dump()`, but returns `bytes`. ''' fd = io.BytesIO() await self.route_dump(fd, family, fmt) return fd.getvalue() async def route_load(self, fd, fmt='iproute2'): '''Load routes from a binary dump. fd -- an open file object, must support `read()` fmt -- dump format, "iproute2" (default) or "raw" The current version parses the dump and loads routes one by one. This behavior will be changed in the future to optimize the performance, but the result will be the same. If `fmt == "iproute2"`, then the loader checks the magic iproute2 prefix in the dump. Otherwise it parses the data from byte 0. ''' if fmt == 'iproute2': if ( not struct.unpack('I', fd.read(struct.calcsize('I')))[0] == IPROUTE2_DUMP_MAGIC ): raise TypeError('wrong dump magic') elif fmt != 'raw': raise TypeError('dump format not supported') ret = [] for msg in self.marshal.parse(fd.read()): request = NetlinkRequest( self, msg, command='replace', command_map={'replace': (RTM_NEWROUTE, 'replace')}, ) await request.send() ret.extend( [ x['header']['error'] is None async for x in request.response() ] ) if not all(ret): raise NetlinkError('error loading route dump') return [] async def route_loads(self, data, fmt='iproute2'): '''Load routes from a `bytes` object. Like `.route_load()`, but accepts `bytes` instead of an file file. ''' fd = io.BytesIO() fd.write(data) fd.seek(0) return await self.route_load(fd, fmt) # 8<--------------------------------------------------------------- # # Listing methods # async def get_qdiscs(self, index=None): ''' Get all queue disciplines for all interfaces or for the selected one. A compatibility method, == .tc("dump") ''' return await self.tc('dump', index=index) async def get_filters(self, index=0, handle=0, parent=0): ''' Get filters for specified interface, handle and parent. ''' msg = tcmsg() msg['family'] = AF_UNSPEC msg['index'] = index msg['handle'] = transform_handle(handle) msg['parent'] = transform_handle(parent) request = NetlinkRequest(self, msg, msg_type=RTM_GETTFILTER) await request.send() return request.response() async def get_classes(self, index=0): ''' Get classes for specified interface. ''' msg = tcmsg() msg['family'] = AF_UNSPEC msg['index'] = index request = NetlinkRequest(self, msg, msg_type=RTM_GETTCLASS) await request.send() return request.response() async def get_vlans(self, *kwarg): ''' Dump available vlan info on bridge ports ''' # IFLA_EXT_MASK, extended info mask # # include/uapi/linux/rtnetlink.h # 1 << 0 => RTEXT_FILTER_VF # 1 << 1 => RTEXT_FILTER_BRVLAN # 1 << 2 => RTEXT_FILTER_BRVLAN_COMPRESSED # 1 << 3 => RTEXT_FILTER_SKIP_STATS # # maybe place it as mapping into ifinfomsg.py? # return await self.link( 'dump', family=AF_BRIDGE, ext_mask=2, match=kwarg ) async def get_links(self, argv, *kwarg): ''' Get network interfaces. By default returns all interfaces. Arguments vector can contain interface indices or a special keyword 'all':: ip.get_links() ip.get_links('all') ip.get_links(1, 2, 3) interfaces = [1, 2, 3] ip.get_links(interfaces) ''' links = argv or [0] if links[0] == 'all': # compat syntax links = [0] if links[0] == 0: return await self.link('dump', kwarg) async def dump(): for index in links: for link in await self.link('get', index=index, kwarg): yield link return dump() async def get_neighbours(self, family=AF_UNSPEC, match=None, kwarg): ''' Dump ARP cache records. The `family` keyword sets the family for the request: e.g. `AF_INET` or `AF_INET6` for arp cache, `AF_BRIDGE` for fdb. If other keyword arguments not empty, they are used as filter. Also, one can explicitly set filter as a function with the `match` parameter. Examples:: # get neighbours on the 3rd link: ip.get_neighbours(ifindex=3) # get a particular record by dst: ip.get_neighbours(dst='172.16.0.1') # get fdb records: ip.get_neighbours(AF_BRIDGE) # and filter them by a function: ip.get_neighbours(AF_BRIDGE, match=lambda x: x['state'] == 2) ''' return await self.neigh('dump', family=family, match=match or kwarg) async def get_ntables(self, family=AF_UNSPEC): ''' Get neighbour tables ''' msg = ndtmsg() msg['family'] = family request = NetlinkRequest(self, msg, msg_type=RTM_GETNEIGHTBL) await request.send() return request.response() async def get_addr(self, family=AF_UNSPEC, match=None, kwarg): ''' Dump addresses. If family is not specified, both AF_INET and AF_INET6 addresses will be dumped:: # get all addresses ip.get_addr() It is possible to apply filters on the results:: # get addresses for the 2nd interface ip.get_addr(index=2) # get addresses with IFA_LABEL == 'eth0' ip.get_addr(label='eth0') # get all the subnet addresses on the interface, identified # by broadcast address (should be explicitly specified upon # creation) ip.get_addr(index=2, broadcast='192.168.1.255') A custom predicate can be used as a filter:: ip.get_addr(match=lambda x: x['index'] == 1) ''' return await self.addr('dump', family=family, match=match or kwarg) async def get_rules(self, family=AF_UNSPEC, match=None, kwarg): ''' Get all rules. By default return all rules. To explicitly request the IPv4 rules use `family=AF_INET`. Example:: ip.get_rules() # get all the rules for all families ip.get_rules(family=AF_INET6) # get only IPv6 rules ''' return await self.rule('dump', family=family, match=match or kwarg) async def get_routes(self, family=255, match=None, kwarg): ''' Get all routes. You can specify the table. There are up to 4294967295 routing classes (tables), and the kernel returns all the routes on each request. So the routine filters routes from full output. Note the number of tables is increased from 255 in Linux 2.6+. Example:: ip.get_routes() # get all the routes for all families ip.get_routes(family=AF_INET6) # get only IPv6 routes ip.get_routes(table=254) # get routes from 254 table The default family=255 is a hack. Despite the specs, the kernel returns only IPv4 routes for AF_UNSPEC family. But it returns all the routes for all the families if one uses an invalid value here. Hack but true. And let's hope the kernel team will not fix this bug. ''' # get a particular route? async def dump(dst): for route in await self.route('get', dst=dst): yield route if isinstance(kwarg.get('dst'), str): return dump(kwarg['dst']) else: return await self.route( 'dump', family=family, match=match or kwarg ) # 8<--------------------------------------------------------------- # 8<--------------------------------------------------------------- @staticmethod def open_file(path): '''Open a file (read only) and return its (fd, inode).''' fd = os.open(path, os.O_RDONLY) inode = os.fstat(fd).st_ino return (fd, inode) @staticmethod def close_file(fd): '''Close a file that was previously opened with open_file().''' os.close(fd) @staticmethod def get_pid(): '''Return the PID of the current process.''' return os.getpid() def register_link_kind(self, path=None, pkg=None, module=None): return ifinfmsg.ifinfo.register_link_kind(path, pkg, module) def unregister_link_kind(self, kind): return ifinfmsg.ifinfo.unregister_link_kind(kind) def list_link_kind(self): return ifinfmsg.ifinfo.list_link_kind() # # List NetNS info # async def _dump_one_ns(self, path, registry): item = nsinfmsg() item['netnsid'] = 0xFFFFFFFF # default netnsid "unknown" nsfd = 0 info = nsidmsg() msg = nsidmsg() try: (nsfd, inode) = self.open_file(path) item['inode'] = inode # # if the inode is registered, skip it # if item['inode'] in registry: raise SkipInode() registry.add(item['inode']) # # request NETNSA_NSID # # may not work on older kernels ( <4.20 ?) # msg['attrs'] = [('NETNSA_FD', nsfd)] request = NetlinkRequest( self, msg, msg_type=RTM_GETNSID, msg_flags=NLM_F_REQUEST ) await request.send() try: async for info in request.response(): # response to nlm_request() is a list or a generator, # that's why loop item['netnsid'] = info.get_attr('NETNSA_NSID') break except Exception: pass item['attrs'] = [('NSINFO_PATH', path)] except OSError as e: raise SkipInode(e.errno) finally: if nsfd > 0: self.close_file(nsfd) item['header']['type'] = RTM_NEWNETNS item['header']['target'] = self.target item['event'] = 'RTM_NEWNETNS' return item async def _dump_dir(self, path, registry): for name in os.listdir(path): # strictly speaking, there is no need to use os.sep, # since the code is not portable outside of Linux nspath = '%s%s%s' % (path, os.sep, name) try: yield await self._dump_one_ns(nspath, registry) except SkipInode: pass async def _dump_proc(self, registry): for name in os.listdir('/proc'): try: int(name) except ValueError: continue try: yield await self._dump_one_ns( '/proc/%s/ns/net' % name, registry ) except SkipInode: pass async def get_netnsid( self, nsid=None, pid=None, fd=None, target_nsid=None ): '''Return a dict containing the result of a RTM_GETNSID query. This loosely corresponds to the "ip netns list-id" command. ''' msg = nsidmsg() if nsid is not None: msg['attrs'].append(('NETNSA_NSID', nsid)) if pid is not None: msg['attrs'].append(('NETNSA_PID', pid)) if fd is not None: msg['attrs'].append(('NETNSA_FD', fd)) if target_nsid is not None: msg['attrs'].append(('NETNSA_TARGET_NSID', target_nsid)) request = NetlinkRequest( self, msg, msg_type=RTM_GETNSID, msg_flags=NLM_F_REQUEST ) await request.send() async for r in request.response(): return { 'nsid': r.get_attr('NETNSA_NSID'), 'current_nsid': r.get_attr('NETNSA_CURRENT_NSID'), } return None async def get_netns_info(self, list_proc=False): ''' A prototype method to list available netns and associated interfaces. A bit weird to have it here and not under `pyroute2.netns`, but it uses RTNL to get all the info. ''' # # register all the ns inodes, not to repeat items in the output # registry = set() # # fetch veth peers # peers = {} async for peer in await self.link('dump'): netnsid = peer.get_attr('IFLA_LINK_NETNSID') if netnsid is not None: if netnsid not in peers: peers[netnsid] = [] peers[netnsid].append(peer.get_attr('IFLA_IFNAME')) # # chain iterators: # # * one iterator for every item in self.path # * one iterator for /proc/<pid>/ns/net # views = [] for path in self.status['netns_path']: views.append(self._dump_dir(path, registry)) if list_proc: views.append(self._dump_proc(registry)) # # iterate all the items # async def ret(): for view in views: try: async for item in view: # # remove uninitialized 'value' field # del item['value'] # # fetch peers for that ns # for peer in peers.get(item['netnsid'], []): item['attrs'].append(('NSINFO_PEER', peer)) yield item except OSError: pass return ret() def set_netnsid(self, nsid=None, pid=None, fd=None): '''Assigns an id to a peer netns using RTM_NEWNSID query. The kernel chooses an unique id if nsid is omitted. This corresponds to the "ip netns set" command. ''' msg = nsidmsg() if nsid is None or nsid < 0: # kernel auto select msg['attrs'].append(('NETNSA_NSID', 4294967295)) else: msg['attrs'].append(('NETNSA_NSID', nsid)) if pid is not None: msg['attrs'].append(('NETNSA_PID', pid)) if fd is not None: msg['attrs'].append(('NETNSA_FD', fd)) return self.nlm_request(msg, RTM_NEWNSID, NLM_F_REQUEST \| NLM_F_ACK) # 8<--------------------------------------------------------------- # 8<--------------------------------------------------------------- # # Shortcuts # async def get_default_routes(self, family=AF_UNSPEC, table=DEFAULT_TABLE): ''' Get default routes ''' msg = rtmsg() msg['family'] = family dump_filter, _ = get_dump_filter( 'route', 'dump', {'table': table} if table is not None else {} ) request = NetlinkRequest( self, msg, msg_type=RTM_GETROUTE, msg_flags=NLM_F_DUMP \| NLM_F_REQUEST, parser=default_routes, ) await request.send() return request.response() async def link_lookup(self, match=None, kwarg): ''' Lookup interface index (indeces) by first level NLA value. Example:: ip.link_lookup(address="52:54:00:9d:4e:3d") ip.link_lookup(ifname="lo") ip.link_lookup(operstate="UP") Please note, that link_lookup() returns list, not one value. ''' if kwarg and set(kwarg) < {'index', 'ifname', 'altname'}: # shortcut for index and ifname try: for link in await self.link('get', kwarg): return [link['index']] except (NetlinkError, KeyError): return [] else: # otherwise fallback to the userspace filter return [ link['index'] async for link in await self.get_links(match=match or kwarg) ] # 8<--------------------------------------------------------------- # 8<--------------------------------------------------------------- # # Shortcuts to flush RTNL objects # async def flush_routes(self, argv, kwarg): ''' Flush routes -- purge route records from a table. Arguments are the same as for `get_routes()` routine. Actually, this routine implements a pipe from `get_routes()` to `nlm_request()`. ''' ret = [] async for route in await self.get_routes(argv, *kwarg): request = NetlinkRequest( self, route, msg_type=RTM_DELROUTE, msg_flags=NLM_F_REQUEST \| NLM_F_ACK, ) await request.send() ret.extend([y async for y in request.response()]) return ret async def flush_addr(self, argv, *kwarg): ''' Flush IP addresses. Examples:: # flush all addresses on the interface with index 2: ipr.flush_addr(index=2) # flush all addresses with IFA_LABEL='eth0': ipr.flush_addr(label='eth0') ''' ret = [] work = [] async for addr in await self.get_addr(argv, kwarg): work.append( { 'index': addr.get('index'), 'address': addr.get('address'), 'prefixlen': addr.get('prefixlen'), } ) for addr in work: try: ret.extend(await self.addr('del', addr)) except NetlinkError: if not ret: raise return ret async def flush_rules(self, argv, kwarg): ''' Flush rules. Please keep in mind, that by default the function operates on all* rules of all families. To work only on IPv4 rules, one should explicitly specify `family=AF_INET`. Examples:: # flush all IPv4 rule with priorities above 5 and below 32000 ipr.flush_rules(family=AF_INET, priority=lambda x: 5 < x < 32000) # flush all IPv6 rules that point to table 250: ipr.flush_rules(family=socket.AF_INET6, table=250) ''' ret = [] for rule in tuple([x async for x in await self.rule('dump', kwarg)]): request = NetlinkRequest( self, rule, msg_type=RTM_DELRULE, msg_flags=NLM_F_REQUEST \| NLM_F_ACK, ) await request.send() ret.extend([y async for y in request.response()]) return ret # 8<--------------------------------------------------------------- # 8<--------------------------------------------------------------- # # Extensions to low-level functions # async def brport(self, command, kwarg): ''' Set bridge port parameters. Example:: idx = ip.link_lookup(ifname='eth0') ip.brport("set", index=idx, unicast_flood=0, cost=200) ip.brport("show", index=idx) Possible keywords are NLA names for the `protinfo_bridge` class, without the prefix and in lower letters. ''' # forward the set command to link() if command == 'set': linkkwarg = dict() linkkwarg['index'] = kwarg.pop('index', 0) linkkwarg['kind'] = 'bridge_slave' for key in kwarg: linkkwarg[key] = kwarg[key] return await self.link(command, linkkwarg) command_map = { 'dump': (RTM_GETLINK, 'dump'), 'show': (RTM_GETLINK, 'dump'), } if command not in command_map: raise TypeError('command not supported') dump_filter, kwarg = get_dump_filter('brport', command, kwarg) arguments = get_arguments_processor('brport', command, kwarg) request = NetlinkRequest( self, ifinfmsg(), command, command_map, dump_filter, arguments ) await request.send() return request.response() async def vlan_filter(self, command, kwarg): ''' Vlan filters is another approach to support vlans in Linux. Before vlan filters were introduced, there was only one way to bridge vlans: one had to create vlan interfaces and then add them as ports: .. aafig:: +------+ +----------+ net --> \| eth0 \| <--> \| eth0.500 \| <---+ +------+ +----------+ \| v +------+ +-----+ net --> \| eth1 \| \| br0 \| +------+ +-----+ ^ +------+ +----------+ \| net --> \| eth2 \| <--> \| eth2.500 \| <---+ +------+ +----------+ It means that one has to create as many bridges, as there were vlans. Vlan filters allow to bridge together underlying interfaces and create vlans already on the bridge: .. aafig:: # v500 label shows which interfaces have vlan filter +------+ v500 net --> \| eth0 \| <-------+ +------+ \| v +------+ +-----+ +---------+ net --> \| eth1 \| <--> \| br0 \|<-->\| br0v500 \| +------+ +-----+ +---------+ ^ +------+ v500 \| net --> \| eth2 \| <-------+ +------+ In this example vlan 500 will be allowed only on ports `eth0` and `eth2`, though all three eth nics are bridged. Some example code:: # create bridge ip.link("add", ifname="br0", kind="bridge") # attach a port ip.link("set", index=ip.link_lookup(ifname="eth0")[0], master=ip.link_lookup(ifname="br0")[0]) # set vlan filter ip.vlan_filter("add", index=ip.link_lookup(ifname="eth0")[0], vlan_info={"vid": 500}) # create vlan interface on the bridge ip.link("add", ifname="br0v500", kind="vlan", link=ip.link_lookup(ifname="br0")[0], vlan_id=500) # set all UP ip.link("set", index=ip.link_lookup(ifname="br0")[0], state="up") ip.link("set", index=ip.link_lookup(ifname="br0v500")[0], state="up") ip.link("set", index=ip.link_lookup(ifname="eth0")[0], state="up") # set IP address ip.addr("add", index=ip.link_lookup(ifname="br0v500")[0], address="172.16.5.2", mask=24) Now all the traffic to the network 172.16.5.2/24 will go to vlan 500 only via ports that have such vlan filter. Required arguments for `vlan_filter()`: `index` and `vlan_info`. Vlan info dict:: ip.vlan_filter('add', index=<ifindex>, vlan_info = {'vid': <single or range>, 'pvid': <bool>, 'flags': int or list} More details: * kernel:Documentation/networking/switchdev.txt * pyroute2.netlink.rtnl.ifinfmsg:... vlan_info Setting PVID or specifying a range will specify the approprate flags. One can specify `flags` as int or as a list of flag names: * `master` == 0x1 * `pvid` == 0x2 * `untagged` == 0x4 * `range_begin` == 0x8 * `range_end` == 0x10 * `brentry` == 0x20 E.g.:: {'vid': 20, 'pvid': true } # is equal to {'vid': 20, 'flags': ['pvid', 'untagged']} # is equal to {'vid': 20, 'flags': 6} # range {'vid': '100-199'} Required arguments for `vlan_filter()`: `index` and `vlan_tunnel_info`. Vlan tunnel info dict:: ip.vlan_filter('add', index=<ifindex>, vlan_tunnel_info = {'vid': <single or range>, 'id': <single or range>} vlan_tunnel_info appears to only use the 'range_begin' and 'range_end' flags from vlan_info. Specifying a range will automatically send the needed flags. Example:: {'vid': 20, 'id: 20} {'vid': '200-299', 'id': '200-299'} The above directives can be combined as in the example:: ip.vlan_filter('add', index=7, vlan_info={'vid': 600}, vlan_tunnel_info={'vid': 600, 'id': 600}) Commands: add Add vlan filter to a bridge port. Example:: ip.vlan_filter("add", index=2, vlan_info={"vid": 200}) del Remove vlan filter from a bridge port. Example:: ip.vlan_filter("del", index=2, vlan_info={"vid": 200}) ''' command_map = { 'add': (RTM_SETLINK, 'req'), 'del': (RTM_DELLINK, 'req'), } kwarg['family'] = AF_BRIDGE kwarg['command_map'] = command_map kwarg['dump_filter'] = None kwarg['request_filter'] = get_arguments_processor( 'vlan_filter', command, kwarg ) return await self.link(command, kwarg) async def fdb(self, command, kwarg): ''' Bridge forwarding database management. More details: * kernel:Documentation/networking/switchdev.txt * pyroute2.netlink.rtnl.ndmsg add Add a new FDB record. Works in the same way as ARP cache management, but some additional NLAs can be used:: # simple FDB record # ip.fdb('add', ifindex=ip.link_lookup(ifname='br0')[0], lladdr='00:11:22:33:44:55', dst='10.0.0.1') # specify vlan # NB: vlan should exist on the device, use # `vlan_filter()` # ip.fdb('add', ifindex=ip.link_lookup(ifname='br0')[0], lladdr='00:11:22:33:44:55', dst='10.0.0.1', vlan=200) # specify vxlan id and port # NB: works only for vxlan devices, use # `link("add", kind="vxlan", ...)` # # if port is not specified, the default one is used # by the kernel. # # if vni (vxlan id) is equal to the device vni, # the kernel doesn't report it back # ip.fdb('add', ifindex=ip.link_lookup(ifname='vx500')[0] lladdr='00:11:22:33:44:55', dst='10.0.0.1', port=5678, vni=600) # or specify src_vni for a vlan-aware vxlan device ip.fdb('add', ifindex=ip.link_lookup(ifname='vx500')[0] lladdr='00:11:22:33:44:55', dst='10.0.0.1', port=5678, src_vni=600) append Append a new FDB record. The same syntax as for add. del Remove an existing FDB record. The same syntax as for add. dump Dump all the FDB records. If any `kwarg` is provided, results will be filtered:: # dump all the records ip.fdb('dump') # show only specific lladdr, dst, vlan etc. ip.fdb('dump', lladdr='00:11:22:33:44:55') ip.fdb('dump', dst='10.0.0.1') ip.fdb('dump', vlan=200) ''' kwarg['family'] = AF_BRIDGE # nud -> state if 'nud' in kwarg: kwarg['state'] = kwarg.pop('nud') if (command in ('add', 'del', 'append')) and not ( kwarg.get('state', 0) & ndmsg.states['noarp'] ): # state must contain noarp in add / del / append kwarg['state'] = kwarg.pop('state', 0) \| ndmsg.states['noarp'] # other assumptions if not kwarg.get('state', 0) & ( ndmsg.states['permanent'] \| ndmsg.states['reachable'] ): # permanent (default) or reachable kwarg['state'] \|= ndmsg.states['permanent'] if not kwarg.get('flags', 0) & ( ndmsg.flags['self'] \| ndmsg.flags['master'] ): # self (default) or master kwarg['flags'] = kwarg.get('flags', 0) \| ndmsg.flags['self'] # return await self.neigh(command, kwarg) # 8<--------------------------------------------------------------- # # General low-level configuration methods # async def neigh(self, command, kwarg): ''' Neighbours operations, same as `ip neigh` or `bridge fdb` add Add a neighbour record, e.g.:: from pyroute2 import IPRoute from pyroute2.netlink.rtnl import ndmsg # add a permanent record on veth0 idx = ip.link_lookup(ifname='veth0')[0] ip.neigh('add', dst='172.16.45.1', lladdr='00:11:22:33:44:55', ifindex=idx, state=ndmsg.states['permanent']) set Set an existing record or create a new one, if it doesn't exist. The same as above, but the command is "set":: ip.neigh('set', dst='172.16.45.1', lladdr='00:11:22:33:44:55', ifindex=idx, state=ndmsg.states['permanent']) change Change an existing record. If the record doesn't exist, fail. del Delete an existing record. dump Dump all the records in the NDB:: ip.neigh('dump') get Get specific record (dst and ifindex are mandatory). Available only on recent kernel:: ip.neigh('get', dst='172.16.45.1', ifindex=idx) ''' command_map = { 'add': (RTM_NEWNEIGH, 'create'), 'set': (RTM_NEWNEIGH, 'replace'), 'replace': (RTM_NEWNEIGH, 'replace'), 'change': (RTM_NEWNEIGH, 'change'), 'del': (RTM_DELNEIGH, 'req'), 'remove': (RTM_DELNEIGH, 'req'), 'delete': (RTM_DELNEIGH, 'req'), 'dump': (RTM_GETNEIGH, 'dump'), 'get': (RTM_GETNEIGH, 'get'), 'append': (RTM_NEWNEIGH, 'append'), } if isinstance(kwarg.get('match'), str): kwarg['match'] = {'ifname': kwarg['match']} dump_filter, kwarg = get_dump_filter('neigh', command, kwarg) arguments = get_arguments_processor('neigh', command, kwarg) request = NetlinkRequest( self, ndmsg.ndmsg(), command, command_map, dump_filter, arguments ) await request.send() if command == 'dump': return request.response() return [x async for x in request.response()] async def link(self, command, kwarg): ''' Link operations. Keywords to set up ifinfmsg fields: * index -- interface index * family -- AF_BRIDGE for bridge operations, otherwise 0 * flags -- device flags * change -- change mask All other keywords will be translated to NLA names, e.g. `mtu -> IFLA_MTU`, `af_spec -> IFLA_AF_SPEC` etc. You can provide a complete NLA structure or let filters do it for you. E.g., these pairs show equal statements:: # set device MTU ip.link("set", index=x, mtu=1000) ip.link("set", index=x, IFLA_MTU=1000) # add vlan device ip.link("add", ifname="test", kind="dummy") ip.link("add", ifname="test", IFLA_LINKINFO={'attrs': [['IFLA_INFO_KIND', 'dummy']]}) Filters are implemented in the `pyroute2.iproute.req` module. You can contribute your own if you miss shortcuts. Commands: add To create an interface, one should specify the interface kind:: ip.link("add", ifname="test", kind="dummy") The kind can be any of those supported by kernel. It can be `dummy`, `bridge`, `bond` etc. On modern kernels one can specify even interface index:: ip.link("add", ifname="br-test", kind="bridge", index=2345) Specific type notes: ► geneve Create GENEVE tunnel:: ip.link("add", ifname="genx", kind="geneve", geneve_id=42, geneve_remote="172.16.0.101") Support for GENEVE over IPv6 is also included; use `geneve_remote6` to configure a remote IPv6 address. ► gre Create GRE tunnel:: ip.link("add", ifname="grex", kind="gre", gre_local="172.16.0.1", gre_remote="172.16.0.101", gre_ttl=16) The keyed GRE requires explicit iflags/oflags specification:: ip.link("add", ifname="grex", kind="gre", gre_local="172.16.0.1", gre_remote="172.16.0.101", gre_ttl=16, gre_ikey=10, gre_okey=10, gre_iflags=32, gre_oflags=32) Support for GRE over IPv6 is also included; use `kind=ip6gre` and `ip6gre_` as the prefix for its values. ► ipip Create ipip tunnel:: ip.link("add", ifname="tun1", kind="ipip", ipip_local="172.16.0.1", ipip_remote="172.16.0.101", ipip_ttl=16) Support for sit and ip6tnl is also included; use `kind=sit` and `sit_` as prefix for sit tunnels, and `kind=ip6tnl` and `ip6tnl_` prefix for ip6tnl tunnels. ► macvlan Macvlan interfaces act like VLANs within OS. The macvlan driver provides an ability to add several MAC addresses on one interface, where every MAC address is reflected with a virtual interface in the system. In some setups macvlan interfaces can replace bridge interfaces, providing more simple and at the same time high-performance solution:: ip.link("add", ifname="mvlan0", kind="macvlan", link=ip.link_lookup(ifname="em1")[0], macvlan_mode="private").commit() Several macvlan modes are available: "private", "vepa", "bridge", "passthru". Ususally the default is "vepa". ► macvtap Almost the same as macvlan, but creates also a character tap device:: ip.link("add", ifname="mvtap0", kind="macvtap", link=ip.link_lookup(ifname="em1")[0], macvtap_mode="vepa").commit() Will create a device file `"/dev/tap%s" % index` ► tuntap Possible `tuntap` keywords: * `mode` — "tun" or "tap" * `uid` — integer * `gid` — integer * `ifr` — dict of tuntap flags (see ifinfmsg:... tuntap_data) Create a tap interface:: ip.link("add", ifname="tap0", kind="tuntap", mode="tap") Tun/tap interfaces are created using `ioctl()`, but the library provides a transparent way to manage them using netlink API. ► veth To properly create `veth` interface, one should specify `peer` also, since `veth` interfaces are created in pairs:: # simple call ip.link("add", ifname="v1p0", kind="veth", peer="v1p1") # set up specific veth peer attributes ip.link("add", ifname="v1p0", kind="veth", peer={"ifname": "v1p1", "net_ns_fd": "test_netns"}) ► vlan VLAN interfaces require additional parameters, `vlan_id` and `link`, where `link` is a master interface to create VLAN on:: ip.link("add", ifname="v100", kind="vlan", link=ip.link_lookup(ifname="eth0")[0], vlan_id=100) There is a possibility to create also 802.1ad interfaces:: # create external vlan 802.1ad, s-tag ip.link("add", ifname="v100s", kind="vlan", link=ip.link_lookup(ifname="eth0")[0], vlan_id=100, vlan_protocol=0x88a8) # create internal vlan 802.1q, c-tag ip.link("add", ifname="v200c", kind="vlan", link=ip.link_lookup(ifname="v100s")[0], vlan_id=200, vlan_protocol=0x8100) ► vrf VRF interfaces (see linux/Documentation/networking/vrf.txt):: ip.link("add", ifname="vrf-foo", kind="vrf", vrf_table=42) ► vxlan VXLAN interfaces are like VLAN ones, but require a bit more parameters:: ip.link("add", ifname="vx101", kind="vxlan", vxlan_link=ip.link_lookup(ifname="eth0")[0], vxlan_id=101, vxlan_group='239.1.1.1', vxlan_ttl=16) All possible vxlan parameters are listed in the module `pyroute2.netlink.rtnl.ifinfmsg:... vxlan_data`. ► ipoib IPoIB driver provides an ability to create several ip interfaces on one interface. IPoIB interfaces requires the following parameter: `link` : The master interface to create IPoIB on. The following parameters can also be provided: * `pkey`- Inifiniband partition key the ip interface is associated with * `mode`- Underlying infiniband transport mode. One of: ['datagram' ,'connected'] * `umcast`- If set(1), multicast group membership for this interface is handled by user space. Example:: ip.link("add", ifname="ipoib1", kind="ipoib", link=ip.link_lookup(ifname="ib0")[0], pkey=10) set Set interface attributes:: # get interface index x = ip.link_lookup(ifname="eth0")[0] # put link down ip.link("set", index=x, state="down") # rename and set MAC addr ip.link("set", index=x, address="00:11:22:33:44:55", name="bala") # set MTU and TX queue length ip.link("set", index=x, mtu=1000, txqlen=2000) # bring link up ip.link("set", index=x, state="up") Seting bridge or tunnel attributes require `kind` to be specified in order to properly encode `IFLA_LINKINFO`:: ip.link("set", index=x, kind="bridge", br_forward_delay=2000) ip.link("set", index=x, kind="gre", gre_local="10.0.0.1", gre_remote="10.1.0.103") Keyword "state" is reserved. State can be "up" or "down", it is a shortcut:: state="up": flags=1, change=1 state="down": flags=0, change=1 SR-IOV virtual function setup:: # get PF index x = ip.link_lookup(ifname="eth0")[0] # setup macaddr ip.link("set", index=x, # PF index vf={"vf": 0, # VF index "mac": "00:11:22:33:44:55"}) # address # setup vlan ip.link("set", index=x, # PF index vf={"vf": 0, # VF index "vlan": 100}) # the simplest case # setup QinQ ip.link("set", index=x, # PF index vf={"vf": 0, # VF index "vlan": [{"vlan": 100, # vlan id "proto": 0x88a8}, # 802.1ad {"vlan": 200, # vlan id "proto": 0x8100}]}) # 802.1q update Almost the same as `set`, except it uses different flags and message type. Mostly does the same, but in some cases differs. If you're not sure what to use, use `set`. del Destroy the interface:: ip.link("del", index=ip.link_lookup(ifname="dummy0")[0]) dump Dump info for all interfaces get Get specific interface info:: ip.link("get", index=ip.link_lookup(ifname="br0")[0]) Get extended attributes like SR-IOV setup:: ip.link("get", index=3, ext_mask=1) ''' command_map = { 'set': (RTM_NEWLINK, 'req'), 'update': (RTM_SETLINK, 'create'), 'add': (RTM_NEWLINK, 'create'), 'del': (RTM_DELLINK, 'req'), 'property_add': (RTM_NEWLINKPROP, 'append'), 'property_del': (RTM_DELLINKPROP, 'req'), 'remove': (RTM_DELLINK, 'req'), 'delete': (RTM_DELLINK, 'req'), 'dump': (RTM_GETLINK, 'dump'), 'get': (RTM_GETLINK, 'get'), } if isinstance(kwarg.get('match'), str): kwarg['match'] = {'ifname': kwarg['match']} if 'command_map' in kwarg: command_map = kwarg.pop('command_map') dump_filter, kwarg = get_dump_filter('link', command, kwarg) arguments = get_arguments_processor('link', command, kwarg) request = NetlinkRequest( self, ifinfmsg(), command, command_map, dump_filter, arguments ) await request.send() if command == 'dump': return request.response() return [x async for x in request.response()] async def addr(self, command, *kwarg): ''' Address operations command -- add, delete, replace, dump * index -- device index * address -- IPv4 or IPv6 address * mask -- address mask * family -- socket.AF_INET for IPv4 or socket.AF_INET6 for IPv6 * scope -- the address scope, see /etc/iproute2/rt_scopes * kwarg -- dictionary, any ifaddrmsg field or NLA Later the method signature will be changed to:: def addr(self, command, match=None, kwarg): # the method body So only keyword arguments (except of the command) will be accepted. The reason for this change is an unification of API. Example:: idx = 62 ip.addr('add', index=idx, address='10.0.0.1', mask=24) ip.addr('add', index=idx, address='10.0.0.2', mask=24) With more NLAs:: # explicitly set broadcast address ip.addr('add', index=idx, address='10.0.0.3', broadcast='10.0.0.255', prefixlen=24) # make the secondary address visible to ifconfig: add label ip.addr('add', index=idx, address='10.0.0.4', broadcast='10.0.0.255', prefixlen=24, label='eth0:1') Configure p2p address on an interface:: ip.addr('add', index=idx, address='10.1.1.2', mask=24, local='10.1.1.1') ''' if command in ('get', 'set'): return if 'mask' in kwarg: warnings.warn( 'usage of mask is deprecated, use prefixlen instead', DeprecationWarning, ) command_map = { 'add': (RTM_NEWADDR, 'create'), 'del': (RTM_DELADDR, 'req'), 'remove': (RTM_DELADDR, 'req'), 'delete': (RTM_DELADDR, 'req'), 'replace': (RTM_NEWADDR, 'replace'), 'dump': (RTM_GETADDR, 'dump'), } # quirks in the filter: flags are supplied as NLA, not as a field dump_filter, kwarg = get_dump_filter('addr', command, kwarg) arguments = get_arguments_processor('addr', command, kwarg) request = NetlinkRequest( self, ifaddrmsg(), command, command_map, dump_filter, arguments, terminate=lambda x: x['header']['type'] == NLMSG_ERROR, ) await request.send() if command == 'dump': return request.response() return [x async for x in request.response()] async def tc(self, command, kind=None, index=None, handle=None, kwarg): ''' "Swiss knife" for traffic control. With the method you can dump, add, delete or modify qdiscs, classes and filters. * command -- add or delete qdisc, class, filter. * kind -- a string identifier -- "sfq", "htb", "u32" and so on. * handle -- integer or string Command can be one of ("add", "del", "add-class", "del-class", "add-filter", "del-filter") (see `commands` dict in the code). Handle notice: traditional iproute2 notation, like "1:0", actually represents two parts in one four-bytes integer:: 1:0 -> 0x10000 1:1 -> 0x10001 ff:0 -> 0xff0000 ffff:1 -> 0xffff0001 Target notice: if your target is a class/qdisc that applies an algorithm that can only apply to upstream traffic profile, but your keys variable explicitly references a match that is only relevant for upstream traffic, the kernel will reject the filter. Unless you're dealing with devices like IMQs For pyroute2 tc() you can use both forms: integer like 0xffff0000 or string like 'ffff:0000'. By default, handle is 0, so you can add simple classless queues w/o need to specify handle. Ingress queue causes handle to be 0xffff0000. So, to set up sfq queue on interface 1, the function call will be like that:: ip = IPRoute() ip.tc("add", "sfq", 1) Instead of string commands ("add", "del"...), you can use also module constants, `RTM_NEWQDISC`, `RTM_DELQDISC` and so on:: ip = IPRoute() flags = NLM_F_REQUEST \| NLM_F_ACK \| NLM_F_CREATE \| NLM_F_EXCL ip.tc((RTM_NEWQDISC, flags), "sfq", 1) It should be noted that "change", "change-class" and "change-filter" work like "replace", "replace-class" and "replace-filter", except they will fail if the node doesn't exist (while it would have been created by "replace"). This is not the same behaviour as with "tc" where "change" can be used to modify the value of some options while leaving the others unchanged. However, as not all entities support this operation, we believe the "change" commands as implemented here are more useful. Also available "modules" (returns tc plugins dict) and "help" commands:: help(ip.tc("modules")["htb"]) print(ip.tc("help", "htb")) ''' if command == 'set': return if command == 'modules': return tc_plugins if command == 'help': p = tc_plugins.get(kind) if p is not None and hasattr(p, '__doc__'): return p.__doc__ else: return 'No help available' command_map = { 'dump': (RTM_GETQDISC, 'dump'), 'get': (RTM_GETQDISC, 'req'), 'add': (RTM_NEWQDISC, 'create'), 'del': (RTM_DELQDISC, 'req'), 'remove': (RTM_DELQDISC, 'req'), 'delete': (RTM_DELQDISC, 'req'), 'change': (RTM_NEWQDISC, 'change'), 'replace': (RTM_NEWQDISC, 'replace'), 'dump-class': (RTM_GETTCLASS, 'dump'), 'get-class': (RTM_GETTCLASS, 'dump'), 'add-class': (RTM_NEWTCLASS, 'create'), 'del-class': (RTM_DELTCLASS, 'req'), 'change-class': (RTM_NEWTCLASS, 'change'), 'replace-class': (RTM_NEWTCLASS, 'replace'), 'dump-filter': (RTM_GETTFILTER, 'dump'), 'get-filter': (RTM_GETTFILTER, 'dump'), 'add-filter': (RTM_NEWTFILTER, 'create'), 'del-filter': (RTM_DELTFILTER, 'req'), 'change-filter': (RTM_NEWTFILTER, 'change'), 'replace-filter': (RTM_NEWTFILTER, 'replace'), } if command[:3] in ('add', 'cha'): if kind is None: raise ValueError('must specify kind for add/change commands') if kind is not None: kwarg['kind'] = kind # 8<----------------------------------------------- # compatibility section, to be cleaned up? if index is not None: kwarg['index'] = index if handle is not None: kwarg['handle'] = handle # 8<----------------------------------------------- dump_filter, kwarg = get_dump_filter('tc', command, kwarg) arguments = get_arguments_processor('tc', command, kwarg) request = NetlinkRequest( self, tcmsg(), command, command_map, dump_filter, arguments ) await request.send() if command.startswith('dump'): return request.response() return [x async for x in request.response()] async def route(self, command, *kwarg): ''' Route operations. Keywords to set up rtmsg fields: dst_len, src_len -- destination and source mask(see `dst` below) * tos -- type of service * table -- routing table * proto -- `redirect`, `boot`, `static` (see `rt_proto`) * scope -- routing realm * type -- `unicast`, `local`, etc. (see `rt_type`) `pyroute2/netlink/rtnl/rtmsg.py` rtmsg.nla_map: * table -- routing table to use (default: 254) * gateway -- via address * prefsrc -- preferred source IP address * dst -- the same as `prefix` * iif -- incoming traffic interface * oif -- outgoing traffic interface etc. One can specify mask not as `dst_len`, but as a part of `dst`, e.g.: `dst="10.0.0.0/24"`. Commands: add Example:: ipr.route("add", dst="10.0.0.0/24", gateway="192.168.0.1") ... More `route()` examples. Blackhole route:: ipr.route( "add", dst="10.0.0.0/24", type="blackhole", ) Create a route with metrics:: ipr.route( "add", dst="172.16.0.0/24", gateway="10.0.0.10", metrics={ "mtu": 1400, "hoplimit": 16, }, ) Multipath route:: ipr.route( "add", dst="10.0.0.0/24", multipath=[ {"gateway": "192.168.0.1", "hops": 2}, {"gateway": "192.168.0.2", "hops": 1}, {"gateway": "192.168.0.3"}, ], ) MPLS lwtunnel on eth0:: ipr.route( "add", dst="10.0.0.0/24", oif=ip.link_lookup(ifname="eth0"), encap={ "type": "mpls", "labels": "200/300", }, ) IPv6 next hop for IPv4 dst:: ipr.route( "add", prefsrc="10.127.30.4", dst="172.16.0.0/24", via={"family": AF_INET6, "addr": "fe80::1337"}, oif=ipr.link_lookup(ifname="eth0"), table=100, ) Create MPLS route: push label:: # $ sudo modprobe mpls_router # $ sudo sysctl net.mpls.platform_labels=1024 ipr.route( "add", family=AF_MPLS, oif=ipr.link_lookup(ifname="eth0"), dst=0x200, newdst=[0x200, 0x300], ) MPLS multipath:: ipr.route( "add", dst="10.0.0.0/24", table=20, multipath=[ { "gateway": "192.168.0.1", "encap": {"type": "mpls", "labels": 200}, }, { "ifindex": ipr.link_lookup(ifname="eth0"), "encap": {"type": "mpls", "labels": 300}, }, ], ) MPLS target can be int, string, dict or list:: "labels": 300 # simple label "labels": "300" # the same "labels": (200, 300) # stacked "labels": "200/300" # the same # explicit label definition "labels": { "bos": 1, "label": 300, "tc": 0, "ttl": 16, } Create SEG6 tunnel encap mode (kernel >= 4.10):: ipr.route( "add", dst="2001:0:0:10::2/128", oif=idx, encap={ "type": "seg6", "mode": "encap", "segs": "2000::5,2000::6", }, ) Create SEG6 tunnel inline mode (kernel >= 4.10):: ipr.route( "add", dst="2001:0:0:10::2/128", oif=idx, encap={ "type": "seg6", "mode": "inline", "segs": ["2000::5", "2000::6"], }, ) Create SEG6 tunnel inline mode with hmac (kernel >= 4.10):: ipr.route( "add", dst="2001:0:0:22::2/128", oif=idx, encap={ "type": "seg6", "mode": "inline", "segs": "2000::5,2000::6,2000::7,2000::8", "hmac": 0xf, }, ) Create SEG6 tunnel with ip4ip6 encapsulation (kernel >= 4.14):: ipr.route( "add", dst="172.16.0.0/24", oif=idx, encap={ "type": "seg6", "mode": "encap", "segs": "2000::5,2000::6", }, ) Create SEG6LOCAL tunnel End.DX4 action (kernel >= 4.14):: ipr.route( "add", dst="2001:0:0:10::2/128", oif=idx, encap={ "type": "seg6local", "action": "End.DX4", "nh4": "172.16.0.10", }, ) Create SEG6LOCAL tunnel End.DT6 action (kernel >= 4.14):: ipr.route( "add", dst="2001:0:0:10::2/128", oif=idx, encap={ "type": "seg6local", "action": "End.DT6", "table": "10", }, ) Create SEG6LOCAL tunnel End.DT4 action (kernel >= 5.11):: # $ sudo modprobe vrf # $ sudo sysctl -w net.vrf.strict_mode=1 ipr.link( "add", ifname="vrf-foo", kind="vrf", vrf_table=10, ) ipr.route( "add", dst="2001:0:0:10::2/128", oif=idx, encap={ "type": "seg6local", "action": "End.DT4", "vrf_table": 10, }, ) Create SEG6LOCAL tunnel End.DT46 action (kernel >= 5.14):: # $ sudo modprobe vrf # $ sudo sysctl -w net.vrf.strict_mode=1 ip.link('add', ifname='vrf-foo', kind='vrf', vrf_table=10) ip.route('add', dst='2001:0:0:10::2/128', oif=idx, encap={'type': 'seg6local', 'action': 'End.DT46', 'vrf_table': 10}) Create SEG6LOCAL tunnel End.B6 action (kernel >= 4.14):: ipr.route( "add", dst="2001:0:0:10::2/128", oif=idx, encap={ "type": "seg6local", "action": "End.B6", "srh": {"segs": "2000::5,2000::6"}, }, ) Create SEG6LOCAL tunnel End.B6 action with hmac (kernel >= 4.14):: ipr.route( "add", dst="2001:0:0:10::2/128", oif=idx, encap={ "type": "seg6local", "action": "End.B6", "srh": { "segs": "2000::5,2000::6", "hmac": 0xf, }, }, ) change, replace, append Commands `change`, `replace` and `append` have the same meanings as in ip-route(8): `change` modifies only existing route, while `replace` creates a new one, if there is no such route yet. `append` allows to create an IPv6 multipath route. del Remove the route. The same syntax as for add. get Get route by spec. dump Dump all routes. ''' # transform kwarg if command in ('add', 'set', 'replace', 'change', 'append'): kwarg['proto'] = kwarg.get('proto', 'static') or 'static' kwarg['type'] = kwarg.get('type', 'unicast') or 'unicast' command_map = { 'add': (RTM_NEWROUTE, 'create'), 'set': (RTM_NEWROUTE, 'replace'), 'replace': (RTM_NEWROUTE, 'replace'), 'change': (RTM_NEWROUTE, 'change'), 'append': (RTM_NEWROUTE, 'append'), 'del': (RTM_DELROUTE, 'req'), 'remove': (RTM_DELROUTE, 'req'), 'delete': (RTM_DELROUTE, 'req'), 'get': (RTM_GETROUTE, 'get'), 'show': (RTM_GETROUTE, 'dump'), 'dump': (RTM_GETROUTE, 'dump'), } msg = rtmsg() # table is mandatory without strict_check; by default == 254 # if table is not defined in kwarg, save it there # also for nla_attr. Do not set it in strict_check, use # NLA instead # # begin # FIXME: move this block to the request processor if not self.status['strict_check']: table = kwarg.get('table', 254) msg['table'] = table if table <= 255 else 252 # end dump_filter, kwarg = get_dump_filter('route', command, kwarg) arguments = get_arguments_processor( 'route', command, kwarg, {'strict_check': self.status['strict_check']}, ) request = NetlinkRequest( self, msg, command, command_map, dump_filter, arguments ) await request.send() if command in ('dump', 'show'): return request.response() return [x async for x in request.response()] async def rule(self, command, kwarg): ''' Rule operations - command — add, delete - table — 0 < table id < 253 - priority — 0 < rule's priority < 32766 - action — type of rule, default 'FR_ACT_NOP' (see fibmsg.py) - rtscope — routing scope, default RT_SCOPE_UNIVERSE `(RT_SCOPE_UNIVERSE\|RT_SCOPE_SITE\|\ RT_SCOPE_LINK\|RT_SCOPE_HOST\|RT_SCOPE_NOWHERE)` - family — rule's family (socket.AF_INET (default) or socket.AF_INET6) - src — IP source for Source Based (Policy Based) routing's rule - dst — IP for Destination Based (Policy Based) routing's rule - src_len — Mask for Source Based (Policy Based) routing's rule - dst_len — Mask for Destination Based (Policy Based) routing's rule - iifname — Input interface for Interface Based (Policy Based) routing's rule - oifname — Output interface for Interface Based (Policy Based) routing's rule - uid_range — Range of user identifiers, a string like "1000:1234" - dport_range — Range of destination ports, a string like "80-120" - sport_range — Range of source ports, as a string like "80-120" All packets route via table 10:: # 32000: from all lookup 10 # ... ip.rule('add', table=10, priority=32000) Default action:: # 32001: from all lookup 11 unreachable # ... iproute.rule('add', table=11, priority=32001, action='FR_ACT_UNREACHABLE') Use source address to choose a routing table:: # 32004: from 10.64.75.141 lookup 14 # ... iproute.rule('add', table=14, priority=32004, src='10.64.75.141') Use dst address to choose a routing table:: # 32005: from 10.64.75.141/24 lookup 15 # ... iproute.rule('add', table=15, priority=32005, dst='10.64.75.141', dst_len=24) Match fwmark:: # 32006: from 10.64.75.141 fwmark 0xa lookup 15 # ... iproute.rule('add', table=15, priority=32006, dst='10.64.75.141', fwmark=10) ''' if command == 'set': return [] command_map = { 'add': (RTM_NEWRULE, 'create'), 'del': (RTM_DELRULE, 'req'), 'remove': (RTM_DELRULE, 'req'), 'delete': (RTM_DELRULE, 'req'), 'dump': (RTM_GETRULE, 'root'), } if isinstance(kwarg.get('match'), str): kwarg['match'] = {'ifname': kwarg['match']} msg = fibmsg() dump_filter, kwarg = get_dump_filter('rule', command, kwarg) arguments = get_arguments_processor('rule', command, kwarg) request = NetlinkRequest( self, msg, command, command_map, dump_filter, arguments ) await request.send() if command == 'dump': return request.response() return [x async for x in request.response()] async def stats(self, command, kwarg): ''' Stats prototype. ''' command_map = { 'dump': (RTM_GETSTATS, 'dump'), 'get': (RTM_GETSTATS, 'get'), } msg = ifstatsmsg() msg['filter_mask'] = kwarg.get('filter_mask', 31) msg['ifindex'] = kwarg.get('ifindex', 0) dump_filter, kwarg = get_dump_filter('stats', command, kwarg) request = NetlinkRequest(self, msg, command, command_map, dump_filter) await request.send() if command == 'dump': return request.response() return [x async for x in request.response()] # 8<--------------------------------------------------------------- class AsyncIPRoute(AsyncIPRSocket, RTNL_API): ''' Regular ordinary async utility class, provides RTNL API using AsyncIPRSocket as the transport level. .. warning:: The project core is currently undergoing refactoring, so some methods may still use the old synchronous API. This will be addressed in future updates. The main RTNL API class is built on an asyncio core. All methods that send netlink requests are asynchronous and return awaitables. Dump requests return asynchronous generators, while other requests return iterables, such as tuples or lists. This design choice addresses the fact that RTNL dumps, such as routes or neighbors, can return an extremely large number of objects. Buffering the entire response in memory could lead to performance issues. .. testcode:: import asyncio from pyroute2 import AsyncIPRoute async def main(): async with AsyncIPRoute() as ipr: # create a link: immediate evaluation await ipr.link("add", ifname="test0", kind="dummy") # dump links: lazy evaluation async for link in await ipr.link("dump"): print(link.get("ifname")) asyncio.run(main()) .. testoutput:: lo eth0 test0 ''' async def __aenter__(self): return self async def __aexit__(self, exc_type, exc, tb): self.close() class IPRoute(NetlinkSocket): ''' A synchronous version of AsyncIPRoute. All the same API, but sync. Provides a legacy API for the old code that is not using asyncio. This API is designed to be compatible with the old synchronous `IPRoute` from version 0.8.x and earlier: .. testcode:: from pyroute2 import IPRoute with IPRoute() as ipr: for msg in ipr.addr("dump"): addr = msg.get("address") mask = msg.get("prefixlen") print(f"{addr}/{mask}") .. testoutput:: 127.0.0.1/8 192.168.122.28/24 .. testcode:: from pyroute2 import IPRoute with IPRoute() as ipr: # this request returns one match, one interface index eth0 = ipr.link_lookup(ifname="eth0") assert len(eth0) == 1 # 1 if exists else 0 # this requests uses a lambda to filter interfaces # and returns all interfaces that are up nics_up = set(ipr.link_lookup(lambda x: x.get("flags") & 1)) assert len(nics_up) == 2 assert nics_up == {1, 2} ''' 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=RTMGRP_DEFAULTS, nlm_echo=False, netns=None, flags=os.O_CREAT, libc=None, use_socket=None, use_event_loop=None, telemetry=None, ): self.asyncore = AsyncIPRoute( 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, ) self.asyncore.status['event_loop'] = 'new' self.asyncore.local.keep_event_loop = True self.asyncore.event_loop.run_until_complete( self.asyncore.setup_endpoint() ) if self.asyncore.socket.fileno() == -1: raise OSError(9, 'Bad file descriptor') @classmethod def from_asyncore(cls, iproute): ret = cls() ret.asyncore = iproute return ret def poll(self, method, command, timeout=10, interval=0.2, spec): ctime = time.time() ret = tuple() while ctime + timeout > time.time(): try: ret = [x for x in method(command, spec)] if ret: return ret time.sleep(interval) except NetlinkDumpInterrupted: pass raise TimeoutError() def __getattr__(self, name): async_generic_methods = [ 'addr', 'link', 'neigh', 'route', 'rule', 'tc', 'fdb', 'brport', 'probe', 'stats', 'link_lookup', 'vlan_filter', 'flush_addr', 'flush_rules', 'flush_routes', 'get_netnsid', 'route_dump', 'route_dumps', 'route_load', 'route_loads', ] async_dump_methods = [ 'dump', 'get_qdiscs', 'get_filters', 'get_classes', 'get_links', 'get_addr', 'get_neighbours', 'get_routes', 'get_rules', 'get_netns_info', 'get_vlans', 'get_default_routes', 'get_ntables', ] symbol = getattr(self.asyncore, name) def synchronize_generic(argv, kwarg): async def collect_dump(): return [i async for i in await symbol(argv, *kwarg)] async def collect_op(): return await symbol(argv, *kwarg) if ( len(argv) > 0 and isinstance(argv[0], str) and (argv[0].startswith('dump') or argv[0].startswith('show')) ): task = collect_dump else: task = collect_op return task() def synchronize_dump(argv, *kwarg): async def collect_dump(): return [i async for i in await symbol(argv, **kwarg)] return collect_dump() # create an event loop cmd = f'iproute-{name}' if name in async_generic_methods: return partial(self._run_with_cleanup, synchronize_generic, cmd) elif name in async_dump_methods: return partial(self._run_with_cleanup, synchronize_dump, cmd) return symbol class IPBatch(IPRoute): ''' Netlink requests compiler. Does not send any requests, but instead stores them in the internal binary buffer. The contents of the buffer can be used to send batch requests, to test custom netlink parsers and so on. Uses `RTNL_API` and provides all the same API as normal `IPRoute` objects:: # create the batch compiler ipb = IPBatch() # compile requests into the internal buffer ipb.link("add", index=550, ifname="test", kind="dummy") ipb.link("set", index=550, state="up") ipb.addr("add", index=550, address="10.0.0.2", mask=24) # save the buffer data = ipb.batch # reset the buffer ipb.reset() ... # send the buffer IPRoute().sendto(data, (0, 0)) ''' def __init__(self): super().__init__() self.reset() def reset(self): self.asyncore.batch = bytearray() class RawIPRoute(IPRoute): def __init__(self): super().__init__() self.asyncore.request_proxy = None class NetNS(IPRoute): ''' The `NetNS` class, prior to version 0.9.1, was used to run the RTNL API in a network namespace. Starting with pyroute2 version 0.9.1, the network namespace functionality has been integrated into the library core. To run an `IPRoute` or `AsyncIPRoute` instance in a network namespace, simply use the `netns` argument: .. testcode:: from pyroute2 import IPRoute with IPRoute(netns="test") as ipr: assert ipr.status["netns"] == "test" After initialization, the netns name is available as `.status["netns"]`. The old synchronous `NetNS` class is still available for compatibility but now serves as a wrapper around `IPRoute`. .. testcode:: from pyroute2 import NetNS with NetNS("test") as ns: assert ns.status["netns"] == "test" ''' def __init__( self, netns=None, flags=os.O_CREAT, target='localhost', libc=None, groups=RTMGRP_DEFAULTS, ): super().__init__( target=target, netns=netns, flags=flags, libc=libc, groups=groups ) def remove(self): self.close() netns.remove(self.status['netns']) class ChaoticIPRoute(RTNL_API, ChaoticIPRSocket): ''' IPRoute interface for chaotic tests - raising exceptions randomly. ''' pass	88.93 KB	2025-05-25 12:26
📄 parsers.py import struct from functools import partial from pyroute2.netlink import NLMSG_DONE, nlmsg from pyroute2.netlink.exceptions import NetlinkError from pyroute2.netlink.rtnl import RTM_NEWROUTE from pyroute2.netlink.rtnl.rtmsg import rtmsg def get_header(data, offset): # get message header header = dict( zip( ('length', 'type', 'flags', 'sequence_number'), struct.unpack_from('IHHI', data, offset), ) ) header['error'] = None return header def msg_done(header): msg = nlmsg() msg['header'] = header msg.length = msg['header']['length'] return msg def _export_routes(fd, data, offset, length): '''Export RTM_NEWROUTE messages binary data. Otherwise return NLMSG_DONE. ''' header = get_header(data, offset) if header['type'] == NLMSG_DONE: return msg_done(header) elif header['type'] == RTM_NEWROUTE: fd.write(data[offset : offset + length]) return raise NetlinkError() def export_routes(fd): return partial(_export_routes, fd) def default_routes(data, offset, length): ''' Only for RTM_NEWROUTE. This parser returns: * rtmsg() -- only for default routes (no RTA_DST) * nlmsg() -- NLMSG_DONE * None for any other messages ''' header = get_header(data, offset) if header['type'] == NLMSG_DONE: return msg_done(header) # skip to NLA: offset + nlmsg header + rtmsg data cursor = offset + 28 # iterate NLA, if meet RTA_DST -- return None (not a default route) while cursor < offset + length: nla_length, nla_type = struct.unpack_from('HH', data, cursor) nla_length = (nla_length + 3) & ~3 # align, page size = 4 cursor += nla_length if nla_type == 1: return # no RTA_DST, a default route -- spend time to decode using the # standard routine msg = rtmsg(data, offset=offset) msg.decode() msg['header']['error'] = None # required return msg	1.96 KB	2025-05-25 12:26
📄 windows.py ''' Windows systems are not supported, but the library provides some proof-of-concept how to build an RTNL-compatible core on top of WinAPI calls. Only two methods are provided so far. If you're interested in extending the functionality, you're welcome to propose PRs. .. warning:: Using pyroute2 on Windows requires installing `win_inet_pton` module, you can use `pip install win_inet_pton`. ''' import ctypes import os from socket import AF_INET from pyroute2.common import dqn2int from pyroute2.netlink import NLM_F_DUMP, NLM_F_MULTI, NLM_F_REQUEST, NLMSG_DONE from pyroute2.netlink.rtnl import ( RTM_GETADDR, RTM_GETLINK, RTM_GETNEIGH, RTM_GETROUTE, RTM_NEWADDR, RTM_NEWLINK, RTM_NEWNEIGH, RTM_NEWROUTE, ) from pyroute2.netlink.rtnl.ifaddrmsg import ifaddrmsg from pyroute2.netlink.rtnl.ifinfmsg import ifinfmsg from pyroute2.netlink.rtnl.marshal import MarshalRtnl MAX_ADAPTER_NAME_LENGTH = 256 MAX_ADAPTER_DESCRIPTION_LENGTH = 128 MAX_ADAPTER_ADDRESS_LENGTH = 8 class IP_ADDRESS_STRING(ctypes.Structure): pass PIP_ADDRESS_STRING = ctypes.POINTER(IP_ADDRESS_STRING) IP_ADDRESS_STRING._fields_ = [ ('Next', PIP_ADDRESS_STRING), ('IpAddress', ctypes.c_byte * 16), ('IpMask', ctypes.c_byte * 16), ('Context', ctypes.c_ulong), ] class IP_ADAPTER_INFO(ctypes.Structure): pass PIP_ADAPTER_INFO = ctypes.POINTER(IP_ADAPTER_INFO) IP_ADAPTER_INFO._fields_ = [ ('Next', PIP_ADAPTER_INFO), ('ComboIndex', ctypes.c_ulong), ('AdapterName', ctypes.c_byte * (256 + 4)), ('Description', ctypes.c_byte * (128 + 4)), ('AddressLength', ctypes.c_uint), ('Address', ctypes.c_ubyte * 8), ('Index', ctypes.c_ulong), ('Type', ctypes.c_uint), ('DhcpEnabled', ctypes.c_uint), ('CurrentIpAddress', PIP_ADDRESS_STRING), ('IpAddressList', IP_ADDRESS_STRING), ('GatewayList', IP_ADDRESS_STRING), ('DhcpServer', IP_ADDRESS_STRING), ('HaveWins', ctypes.c_byte), ('PrimaryWinsServer', IP_ADDRESS_STRING), ('SecondaryWinsServer', IP_ADDRESS_STRING), ('LeaseObtained', ctypes.c_ulong), ('LeaseExpires', ctypes.c_ulong), ] class IPRoute(object): def __init__(self, argv, kwarg): self.marshal = MarshalRtnl() self.target = kwarg.get('target') or 'localhost' def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): self.close() def clone(self): return self def close(self, code=None): pass def bind(self, argv, *kwarg): pass def getsockopt(self, argv, *kwarg): return 1024 1024 def sendto_gate(self, msg, addr): # # handle incoming netlink requests # # sendto_gate() receives single RTNL messages as objects # cmd = msg['header']['type'] flags = msg['header']['flags'] seq = msg['header']['sequence_number'] # work only on dump requests for now if flags != NLM_F_REQUEST \| NLM_F_DUMP: return # if cmd == RTM_GETLINK: rtype = RTM_NEWLINK ret = self.get_links() elif cmd == RTM_GETADDR: rtype = RTM_NEWADDR ret = self.get_addr() elif cmd == RTM_GETROUTE: rtype = RTM_NEWROUTE ret = self.get_routes() elif cmd == RTM_GETNEIGH: rtype = RTM_NEWNEIGH ret = self.get_neighbours() # # set response type and finalize the message for r in ret: r['header']['type'] = rtype r['header']['flags'] = NLM_F_MULTI r['header']['sequence_number'] = seq # r = type(msg)() r['header']['type'] = NLMSG_DONE r['header']['sequence_number'] = seq ret.append(r) data = b'' for r in ret: r.encode() data += r.data self._outq.put(data) os.write(self._pfdw, b'\0') def _GetAdaptersInfo(self): ret = {'interfaces': [], 'addresses': []} # prepare buffer buf = ctypes.create_string_buffer(15000) buf_len = ctypes.c_ulong(15000) ( ctypes.windll.iphlpapi.GetAdaptersInfo( ctypes.byref(buf), ctypes.byref(buf_len) ) ) adapter = IP_ADAPTER_INFO.from_address(ctypes.addressof(buf)) while True: mac = ':'.join(['%02x' % x for x in adapter.Address][:6]) ifname = ctypes.string_at( ctypes.addressof(adapter.AdapterName) ).decode('utf-8') spec = { 'index': adapter.Index, 'attrs': (['IFLA_ADDRESS', mac], ['IFLA_IFNAME', ifname]), } msg = ifinfmsg().load(spec) msg['header']['target'] = self.target msg['header']['type'] = RTM_NEWLINK del msg['value'] ret['interfaces'].append(msg) ipaddr = adapter.IpAddressList while True: addr = ctypes.string_at( ctypes.addressof(ipaddr.IpAddress) ).decode('utf-8') mask = ctypes.string_at( ctypes.addressof(ipaddr.IpMask) ).decode('utf-8') spec = { 'index': adapter.Index, 'family': AF_INET, 'prefixlen': dqn2int(mask), 'attrs': ( ['IFA_ADDRESS', addr], ['IFA_LOCAL', addr], ['IFA_LABEL', ifname], ), } msg = ifaddrmsg().load(spec) msg['header']['target'] = self.target msg['header']['type'] = RTM_NEWADDR del msg['value'] ret['addresses'].append(msg) if ipaddr.Next: ipaddr = ipaddr.Next.contents else: break if adapter.Next: adapter = adapter.Next.contents else: break return ret def dump(self, groups=None): for method in ( self.get_links, self.get_addr, self.get_neighbours, self.get_routes, ): for msg in method(): yield msg def get_links(self, argv, kwarg): ''' Get network interfaces list:: >>> pprint(ipr.get_links()) [{'attrs': (['IFLA_ADDRESS', '52:54:00:7a:8a:49'], ['IFLA_IFNAME', '{F444467B-3549-455D-81F2-AB617C7421AB}']), 'change': 0, 'family': 0, 'flags': 0, 'header': {}, 'ifi_type': 0, 'index': 7}] ''' return self._GetAdaptersInfo()['interfaces'] def get_addr(self, argv, *kwarg): ''' Get IP addresses:: >>> pprint(ipr.get_addr()) [{'attrs': (['IFA_ADDRESS', '192.168.122.81'], ['IFA_LOCAL', '192.168.122.81'], ['IFA_LABEL', '{F444467B-3549-455D-81F2-AB617C7421AB}']), 'family': <AddressFamily.AF_INET: 2>, 'flags': 0, 'header': {}, 'index': 7, 'prefixlen': 24, 'scope': 0}] ''' return self._GetAdaptersInfo()['addresses'] def get_neighbours(self, argv, *kwarg): ret = [] return ret def get_routes(self, argv, *kwarg): ret = [] return ret class RawIPRoute(IPRoute): pass class ChaoticIPRoute: def __init__(self, argv, *kwarg): raise NotImplementedError() class NetNS: def __init__(self, argv, **kwarg): raise NotImplementedError()	7.74 KB	2025-05-25 12:26

Terminal