Wala shell

Advanced PHP File Manager

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

Name	Size	Last Modified
📁 ..	-	2026-03-27 02:58
📄 __init__.py ## # # This module contains all the public symbols from the library. # ## # # Version # try: from pyroute2.config.version import __version__ except ImportError: __version__ = 'unknown' import sys from pyroute2.cli.console import Console from pyroute2.cli.server import Server ## # # Logging setup # # See the history: # * https://github.com/svinota/pyroute2/issues/246 # * https://github.com/svinota/pyroute2/issues/255 # * https://github.com/svinota/pyroute2/issues/270 # * https://github.com/svinota/pyroute2/issues/573 # * https://github.com/svinota/pyroute2/issues/601 # from pyroute2.config import log from pyroute2.conntrack import Conntrack, ConntrackEntry from pyroute2.devlink import DL from pyroute2.ethtool.ethtool import Ethtool from pyroute2.ipdb import IPDB, CommitException, CreateException from pyroute2.iproute import ( AsyncIPRoute, ChaoticIPRoute, IPBatch, IPRoute, NetNS, RawIPRoute, ) from pyroute2.ipset import IPSet from pyroute2.ipvs import IPVS, IPVSDest, IPVSService from pyroute2.iwutil import IW from pyroute2.ndb.main import NDB from pyroute2.netlink.connector.cn_proc import ProcEventSocket from pyroute2.netlink.devlink import DevlinkSocket from pyroute2.netlink.diag import DiagSocket, ss2 from pyroute2.netlink.event.acpi_event import AcpiEventSocket from pyroute2.netlink.event.dquot import DQuotSocket from pyroute2.netlink.exceptions import ( ChaoticException, NetlinkDecodeError, NetlinkDumpInterrupted, NetlinkError, ) from pyroute2.netlink.generic import GenericNetlinkSocket from pyroute2.netlink.generic.l2tp import L2tp from pyroute2.netlink.generic.mptcp import MPTCP from pyroute2.netlink.generic.wireguard import WireGuard from pyroute2.netlink.ipq import IPQSocket from pyroute2.netlink.nfnetlink.nfctsocket import NFCTSocket from pyroute2.netlink.nfnetlink.nftsocket import AsyncNFTSocket, NFTSocket from pyroute2.netlink.nl80211 import NL80211 from pyroute2.netlink.rtnl.iprsocket import AsyncIPRSocket, IPRSocket from pyroute2.netlink.taskstats import TaskStats from pyroute2.netlink.uevent import UeventSocket from pyroute2.nslink.nspopen import NSPopen from pyroute2.plan9.client import Plan9ClientSocket from pyroute2.plan9.server import Plan9ServerSocket from pyroute2.wiset import WiSet ## # # Windows platform specific: socket module monkey patching # # To use the library on Windows, run:: # pip install win-inet-pton # if sys.platform.startswith('win'): # noqa: E402 import win_inet_pton # noqa: F401 modules = [ AcpiEventSocket, AsyncIPRoute, AsyncIPRSocket, AsyncNFTSocket, ChaoticException, ChaoticIPRoute, CommitException, Conntrack, ConntrackEntry, Console, CreateException, DevlinkSocket, DiagSocket, DL, DQuotSocket, Ethtool, IPBatch, IPDB, IPQSocket, IPRoute, IPRSocket, IPSet, IPVS, IPVSDest, IPVSService, IW, GenericNetlinkSocket, L2tp, MPTCP, NDB, NetlinkError, NetlinkDecodeError, NetlinkDumpInterrupted, NetNS, NFCTSocket, NFTSocket, NL80211, NSPopen, Plan9ClientSocket, Plan9ServerSocket, ProcEventSocket, RawIPRoute, Server, ss2, TaskStats, UeventSocket, WireGuard, WiSet, log, ] __all__ = [] __all__.extend(modules)	3.27 KB	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 arp.py from pyroute2.common import map_namespace # ARP protocol HARDWARE identifiers. ARPHRD_NETROM = 0 # from KA9Q: NET/ROM pseudo ARPHRD_ETHER = 1 # Ethernet 10Mbps ARPHRD_EETHER = 2 # Experimental Ethernet ARPHRD_AX25 = 3 # AX.25 Level 2 ARPHRD_PRONET = 4 # PROnet token ring ARPHRD_CHAOS = 5 # Chaosnet ARPHRD_IEEE802 = 6 # IEEE 802.2 Ethernet/TR/TB ARPHRD_ARCNET = 7 # ARCnet ARPHRD_APPLETLK = 8 # APPLEtalk ARPHRD_DLCI = 15 # Frame Relay DLCI ARPHRD_ATM = 19 # ATM ARPHRD_METRICOM = 23 # Metricom STRIP (new IANA id) ARPHRD_IEEE1394 = 24 # IEEE 1394 IPv4 - RFC 2734 ARPHRD_EUI64 = 27 # EUI-64 ARPHRD_INFINIBAND = 32 # InfiniBand # Dummy types for non ARP hardware ARPHRD_SLIP = 256 ARPHRD_CSLIP = 257 ARPHRD_SLIP6 = 258 ARPHRD_CSLIP6 = 259 ARPHRD_RSRVD = 260 # Notional KISS type ARPHRD_ADAPT = 264 ARPHRD_ROSE = 270 ARPHRD_X25 = 271 # CCITT X.25 ARPHRD_HWX25 = 272 # Boards with X.25 in firmware ARPHRD_PPP = 512 ARPHRD_CISCO = 513 # Cisco HDLC ARPHRD_HDLC = ARPHRD_CISCO ARPHRD_LAPB = 516 # LAPB ARPHRD_DDCMP = 517 # Digital's DDCMP protocol ARPHRD_RAWHDLC = 518 # Raw HDLC ARPHRD_TUNNEL = 768 # IPIP tunnel ARPHRD_TUNNEL6 = 769 # IP6IP6 tunnel ARPHRD_FRAD = 770 # Frame Relay Access Device ARPHRD_SKIP = 771 # SKIP vif ARPHRD_LOOPBACK = 772 # Loopback device ARPHRD_LOCALTLK = 773 # Localtalk device ARPHRD_FDDI = 774 # Fiber Distributed Data Interface ARPHRD_BIF = 775 # AP1000 BIF ARPHRD_SIT = 776 # sit0 device - IPv6-in-IPv4 ARPHRD_IPDDP = 777 # IP over DDP tunneller ARPHRD_IPGRE = 778 # GRE over IP ARPHRD_PIMREG = 779 # PIMSM register interface ARPHRD_HIPPI = 780 # High Performance Parallel Interface ARPHRD_ASH = 781 # Nexus 64Mbps Ash ARPHRD_ECONET = 782 # Acorn Econet ARPHRD_IRDA = 783 # Linux-IrDA # ARP works differently on different FC media .. so ARPHRD_FCPP = 784 # Point to point fibrechannel ARPHRD_FCAL = 785 # Fibrechannel arbitrated loop ARPHRD_FCPL = 786 # Fibrechannel public loop ARPHRD_FCFABRIC = 787 # Fibrechannel fabric # 787->799 reserved for fibrechannel media types ARPHRD_IEEE802_TR = 800 # Magic type ident for TR ARPHRD_IEEE80211 = 801 # IEEE 802.11 ARPHRD_IEEE80211_PRISM = 802 # IEEE 802.11 + Prism2 header ARPHRD_IEEE80211_RADIOTAP = 803 # IEEE 802.11 + radiotap header ARPHRD_MPLS_TUNNEL = 899 # MPLS Tunnel Interface ARPHRD_VOID = 0xFFFF # Void type, nothing is known ARPHRD_NONE = 0xFFFE # zero header length (ARPHRD_NAMES, ARPHRD_VALUES) = map_namespace("ARPHRD_", globals())	2.42 KB	2025-05-25 12:26
📁 bsd	-	2025-05-25 12:26
📁 cli	-	2025-05-25 12:26
📄 common.py # -- coding: utf-8 -- ''' Common utilities ''' import errno import io import os import socket import struct import threading import time import types from functools import partial from typing import Any, Callable, Literal, Optional, TypeVar, Union basestring = (str, bytes) file = io.BytesIO AF_MPLS = 28 _uuid32 = 0 # (singleton) the last uuid32 value saved to avoid collisions _uuid32_lock = threading.Lock() USE_DEFAULT_TIMEOUT = -1 size_suffixes = { 'b': 1, 'k': 1024, 'kb': 1024, 'm': 1024 * 1024, 'mb': 1024 * 1024, 'g': 1024 * 1024 * 1024, 'gb': 1024 * 1024 * 1024, 'kbit': 1024 / 8, 'mbit': 1024 * 1024 / 8, 'gbit': 1024 * 1024 * 1024 / 8, } time_suffixes = { 's': 1, 'sec': 1, 'secs': 1, 'ms': 1000, 'msec': 1000, 'msecs': 1000, 'us': 1000000, 'usec': 1000000, 'usecs': 1000000, } rate_suffixes = { 'bit': 1, 'Kibit': 1024, 'kbit': 1000, 'mibit': 1024 * 1024, 'mbit': 1000000, 'gibit': 1024 * 1024 * 1024, 'gbit': 1000000000, 'tibit': 1024 * 1024 * 1024 * 1024, 'tbit': 1000000000000, 'Bps': 8, 'KiBps': 8 * 1024, 'KBps': 8000, 'MiBps': 8 * 1024 * 1024, 'MBps': 8000000, 'GiBps': 8 * 1024 * 1024 * 1024, 'GBps': 8000000000, 'TiBps': 8 * 1024 * 1024 * 1024 * 1024, 'TBps': 8000000000000, } ## # General purpose # class Namespace: def __init__(self, parent, override=None): raise NotImplementedError() def _no_change(s: str) -> str: return s def _default_normalize(s: str, prefix: str) -> str: return s[len(prefix) :].lower() def map_namespace( prefix: str, ns: dict[str, Any], normalize: Union[None, Literal[True], Callable[[str], str]] = None, ) -> tuple[dict[str, int], dict[int, str]]: ''' Take the namespace prefix, list all constants and build two dictionaries -- straight and reverse mappings. E.g.: ## neighbor attributes NDA_UNSPEC = 0 NDA_DST = 1 NDA_LLADDR = 2 NDA_CACHEINFO = 3 NDA_PROBES = 4 (NDA_NAMES, NDA_VALUES) = map_namespace('NDA', globals()) Will lead to:: NDA_NAMES = {'NDA_UNSPEC': 0, ... 'NDA_PROBES': 4} NDA_VALUES = {0: 'NDA_UNSPEC', ... 4: 'NDA_PROBES'} The `normalize` parameter can be: - None — no name transformation will be done - True — cut the prefix and `lower()` the rest - lambda x: … — apply the function to every name ''' transform: Callable[[str], str] if normalize is None: transform = _no_change elif normalize is True: transform = partial(_default_normalize, prefix=prefix) elif isinstance(normalize, types.FunctionType): transform = normalize else: raise ValueError("Invalid value for `normalize` parameter") by_name = {transform(i): ns[i] for i in ns.keys() if i.startswith(prefix)} by_value = {ns[i]: transform(i) for i in ns.keys() if i.startswith(prefix)} return (by_name, by_value) def getbroadcast( addr: str, mask: int, family: socket.AddressFamily = socket.AF_INET ) -> str: # 1. convert addr to int i = socket.inet_pton(family, addr) if family == socket.AF_INET: i_unpacked = struct.unpack('>I', i)[0] a = 0xFFFFFFFF length = 32 elif family == socket.AF_INET6: i_unpacked = struct.unpack('>QQ', i) i_unpacked = i_unpacked[0] << 64 \| i_unpacked[1] a = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF length = 128 else: raise NotImplementedError('family not supported') # 2. calculate mask m = (a << length - mask) & a # 3. calculate default broadcast n = (i_unpacked & m) \| a >> mask # 4. convert it back to the normal address form if family == socket.AF_INET: n = struct.pack('>I', n) else: n = struct.pack('>QQ', n >> 64, n & (a >> 64)) return socket.inet_ntop(family, n) def dqn2int(mask: str, family: socket.AddressFamily = socket.AF_INET) -> int: ''' IPv4 dotted quad notation to int mask conversion ''' ret = 0 binary = socket.inet_pton(family, mask) for offset in range(len(binary) // 4): ret += bin( struct.unpack('I', binary[offset * 4 : offset * 4 + 4])[0] ).count('1') return ret def get_address_family(address: str) -> socket.AddressFamily: if address.find(':') > -1: return socket.AF_INET6 else: return socket.AF_INET def hexdump(payload: bytes, length: int = 0) -> str: ''' Represent byte string as hex -- for debug purposes ''' return ':'.join('{0:02x}'.format(c) for c in payload[:length] or payload) def load_dump( f: Union[str, io.StringIO], meta: Optional[dict[str, str]] = None ) -> Union[bytes, str]: ''' Load a packet dump from an open file-like object or a string. Supported dump formats: * strace hex dump (\\x00\\x00...) * pyroute2 hex dump (00:00:...) Simple markup is also supported. Any data from # or ; till the end of the string is a comment and ignored. Any data after . till EOF is ignored as well. With #! starts an optional code block. All the data in the code block will be read and returned via metadata dictionary. ''' data = '' code = None meta_data = None meta_label = None io_obj: Union[io.StringIO, str] if isinstance(f, str): io_obj = io.StringIO() io_obj.write(f) io_obj.seek(0) else: io_obj = f for a in io_obj.readlines(): if code is not None: code += a continue if meta_data is not None: meta_data += a continue offset = 0 length = len(a) while offset < length: if a[offset] in (' ', '\t', '\n'): offset += 1 elif a[offset] == '#': if a[offset : offset + 2] == '#!': # read the code block until EOF code = '' elif a[offset : offset + 2] == '#:': # read data block until EOF meta_label = a.split(':')[1].strip() meta_data = '' break elif a[offset] == '.': return data elif a[offset] == '\\': # strace hex format data += chr(int(a[offset + 2 : offset + 4], 16)) offset += 4 else: # pyroute2 hex format data += chr(int(a[offset : offset + 2], 16)) offset += 3 if isinstance(meta, dict): if code is not None: meta['code'] = code if meta_data is not None and meta_label is not None: meta[meta_label] = meta_data return bytes(data, 'iso8859-1') class AddrPool(object): ''' Address pool ''' cell = 0xFFFFFFFFFFFFFFFF def __init__( self, minaddr: int = 0xF, maxaddr: int = 0xFFFFFF, reverse: bool = False, release: bool = False, ): self.cell_size = 0 # in bits mx = self.cell self.reverse = reverse self.release = release self.ban: list[dict[str, int]] = [] while mx: mx >>= 8 self.cell_size += 1 self.cell_size = 8 # calculate, how many ints we need to bitmap all addresses self.cells = int((maxaddr - minaddr) / self.cell_size + 1) # initial array self.addr_map = [self.cell] self.minaddr = minaddr self.maxaddr = maxaddr self.lock = threading.RLock() def alloc(self) -> int: with self.lock: # gc self.ban: for item in tuple(self.ban): if item['counter'] == 0: self.free(item['addr']) self.ban.remove(item) else: item['counter'] -= 1 # iterate through addr_map base = 0 for cell in self.addr_map: if cell: # not allocated addr bit = 0 while True: if (1 << bit) & self.addr_map[base]: self.addr_map[base] ^= 1 << bit break bit += 1 ret = base self.cell_size + bit if self.reverse: ret = self.maxaddr - ret else: ret = ret + self.minaddr if self.minaddr <= ret <= self.maxaddr: if self.release: self.free(ret, ban=self.release) return ret else: self.free(ret) raise KeyError('no free address available') base += 1 # no free address available if len(self.addr_map) < self.cells: # create new cell to allocate address from self.addr_map.append(self.cell) return self.alloc() else: raise KeyError('no free address available') def locate(self, addr: int) -> tuple[int, int, bool]: if self.reverse: addr = self.maxaddr - addr else: addr -= self.minaddr base = addr // self.cell_size bit = addr % self.cell_size try: is_allocated = not self.addr_map[base] & (1 << bit) except IndexError: is_allocated = False return (base, bit, is_allocated) def free(self, addr: int, ban: int = 0): with self.lock: if ban != 0: self.ban.append({'addr': addr, 'counter': ban}) else: base, bit, _ = self.locate(addr) if len(self.addr_map) <= base: raise KeyError('address is not allocated') if self.addr_map[base] & (1 << bit): raise KeyError('address is not allocated') self.addr_map[base] ^= 1 << bit def fnv1(data: bytes) -> int: ''' FNV1 -- 32bit hash, python3 version Returns: 32bit int hash See: http://www.isthe.com/chongo/tech/comp/fnv/index.html ''' hval = 0x811C9DC5 for i in range(len(data)): hval = 0x01000193 hval ^= data[i] return hval & 0xFFFFFFFF def uuid32() -> int: ''' Return 32bit UUID, based on the current time and pid. The uuid is guaranteed to be unique within one process. ''' global _uuid32 global _uuid32_lock with _uuid32_lock: candidate = _uuid32 while candidate == _uuid32: candidate = fnv1( struct.pack('QQ', int(time.time() 1000000), os.getpid()) ) _uuid32 = candidate return candidate def uifname() -> str: ''' Return a unique interface name based on a prime function ''' return 'pr%x' % uuid32() F = TypeVar("F", bound=Callable[..., Any]) def map_exception( match: Callable[[Exception], bool], subst: Callable[[Exception], Exception] ) -> Callable[[F], F]: ''' Decorator to map exception types ''' def wrapper(f: F) -> F: def decorated(argv: Any, kwarg: Any) -> Any: try: f(argv, *kwarg) except Exception as e: if match(e): raise subst(e) raise return decorated # type: ignore return wrapper def map_enoent(f: F) -> F: ''' Shortcut to map OSError(2) -> OSError(95) ''' return map_exception( lambda x: (isinstance(x, OSError) and x.errno == errno.ENOENT), lambda x: OSError(errno.EOPNOTSUPP, 'Operation not supported'), )(f) T = TypeVar('T', bound=type) def metaclass(mc: T) -> Callable[[T], T]: def wrapped(cls: T) -> T: nvars = {} skip = ['__dict__', '__weakref__'] slots = cls.__dict__.get('__slots__') if not isinstance(slots, (list, tuple)): slots = [slots] for k in slots: skip.append(k) for k, v in cls.__dict__.items(): if k not in skip: nvars[k] = v return mc(cls.__name__, cls.__bases__, nvars) return wrapped def msg_done(msg) -> bytes: newmsg = struct.pack('IHH', 40, 2, 0) newmsg += msg.data[8:16] newmsg += struct.pack('I', 0) # nlmsgerr struct alignment newmsg += b'\0' 20 return newmsg	12.41 KB	2025-05-25 12:26
📄 compat.py '''Compatibility with older but supported Python versions''' try: from enum import StrEnum # noqa: F401 except ImportError: from enum import Enum class StrEnum(str, Enum): '''Same as enum, but members are also strings.''' try: from socket import ETHERTYPE_IP except ImportError: # ETHERTYPE_* are new in python 3.12 ETHERTYPE_IP = 0x800 __all__ = ('StrEnum', 'ETHERTYPE_IP')	414 B	2025-05-25 12:26
📁 config	-	2025-05-25 12:26
📄 conntrack.py import socket from pyroute2.netlink.nfnetlink.nfctsocket import ( IP_CT_TCP_FLAG_TO_NAME, IPSBIT_TO_NAME, TCP_CONNTRACK_TO_NAME, NFCTAttrTuple, NFCTSocket, ) class NFCTATcpProtoInfo(object): __slots__ = ( 'state', 'wscale_orig', 'wscale_reply', 'flags_orig', 'flags_reply', ) def __init__( self, state, wscale_orig=None, wscale_reply=None, flags_orig=None, flags_reply=None, ): self.state = state self.wscale_orig = wscale_orig self.wscale_reply = wscale_reply self.flags_orig = flags_orig self.flags_reply = flags_reply def state_name(self): return TCP_CONNTRACK_TO_NAME.get(self.state, "UNKNOWN") def flags_name(self, flags): if flags is None: return '' s = '' for bit, name in IP_CT_TCP_FLAG_TO_NAME.items(): if flags & bit: s += '{},'.format(name) return s[:-1] @classmethod def from_netlink(cls, ndmsg): cta_tcp = ndmsg.get_attr('CTA_PROTOINFO_TCP') state = cta_tcp.get_attr('CTA_PROTOINFO_TCP_STATE') # second argument is the mask returned by kernel but useless for us flags_orig, _ = cta_tcp.get_attr('CTA_PROTOINFO_TCP_FLAGS_ORIGINAL') flags_reply, _ = cta_tcp.get_attr('CTA_PROTOINFO_TCP_FLAGS_REPLY') return cls(state=state, flags_orig=flags_orig, flags_reply=flags_reply) def __repr__(self): return 'TcpInfo(state={}, orig_flags={}, reply_flags={})'.format( self.state_name(), self.flags_name(self.flags_orig), self.flags_name(self.flags_reply), ) class ConntrackEntry(object): __slots__ = ( 'tuple_orig', 'tuple_reply', 'status', 'timeout', 'protoinfo', 'mark', 'id', 'use', ) def __init__( self, family, tuple_orig, tuple_reply, cta_status, cta_timeout, cta_protoinfo, cta_mark, cta_id, cta_use, ): self.tuple_orig = NFCTAttrTuple.from_netlink(family, tuple_orig) self.tuple_reply = NFCTAttrTuple.from_netlink(family, tuple_reply) self.status = cta_status self.timeout = cta_timeout if self.tuple_orig.proto == socket.IPPROTO_TCP: self.protoinfo = NFCTATcpProtoInfo.from_netlink(cta_protoinfo) else: self.protoinfo = None self.mark = cta_mark self.id = cta_id self.use = cta_use def status_name(self): s = '' for bit, name in IPSBIT_TO_NAME.items(): if self.status & bit: s += '{},'.format(name) return s[:-1] def __repr__(self): s = 'Entry(orig={}, reply={}, status={}'.format( self.tuple_orig, self.tuple_reply, self.status_name() ) if self.protoinfo is not None: s += ', protoinfo={}'.format(self.protoinfo) s += ')' return s class Conntrack(NFCTSocket): """ High level conntrack functions """ def __init__(self, nlm_generator=True, kwargs): super(Conntrack, self).__init__(nlm_generator=nlm_generator, kwargs) def stat(self): """Return current statistics per CPU Same result than conntrack -S command but a list of dictionaries """ stats = [] for msg in super(Conntrack, self).stat(): stats.append({'cpu': msg['res_id']}) stats[-1].update( (k[10:].lower(), v) for k, v in msg['attrs'] if k.startswith('CTA_STATS_') ) return stats def count(self): """Return current number of conntrack entries Same result than /proc/sys/net/netfilter/nf_conntrack_count file or conntrack -C command """ for ndmsg in super(Conntrack, self).count(): return ndmsg.get_attr('CTA_STATS_GLOBAL_ENTRIES') def conntrack_max_size(self): """ Return the max size of connection tracking table /proc/sys/net/netfilter/nf_conntrack_max """ for ndmsg in super(Conntrack, self).conntrack_max_size(): return ndmsg.get_attr('CTA_STATS_GLOBAL_MAX_ENTRIES') def delete(self, entry): if isinstance(entry, ConntrackEntry): tuple_orig = entry.tuple_orig elif isinstance(entry, NFCTAttrTuple): tuple_orig = entry else: raise NotImplementedError() for ndmsg in self.entry('del', tuple_orig=tuple_orig): return ndmsg def entry(self, cmd, kwargs): for res in super(Conntrack, self).entry(cmd, kwargs): return res def dump_entries( self, mark=None, mark_mask=0xFFFFFFFF, tuple_orig=None, tuple_reply=None, ): """ Dump all entries from conntrack table with filters Filters can be only part of a conntrack tuple :param NFCTAttrTuple tuple_orig: filter on original tuple :param NFCTAttrTuple tuple_reply: filter on reply tuple Examples:: # Filter only on tcp connections for entry in ct.dump_entries(tuple_orig=NFCTAttrTuple( proto=socket.IPPROTO_TCP)): print("This entry is tcp: {}".format(entry)) # Filter only on icmp message to 8.8.8.8 for entry in ct.dump_entries(tuple_orig=NFCTAttrTuple( proto=socket.IPPROTO_ICMP, daddr='8.8.8.8')): print("This entry is icmp to 8.8.8.8: {}".format(entry)) """ for ndmsg in self.dump( mark=mark, mark_mask=mark_mask, tuple_orig=tuple_orig, tuple_reply=tuple_reply, ): if tuple_orig is not None and not tuple_orig.nla_eq( ndmsg['nfgen_family'], ndmsg.get_attr('CTA_TUPLE_ORIG') ): continue if tuple_reply is not None and not tuple_reply.nla_eq( ndmsg['nfgen_family'], ndmsg.get_attr('CTA_TUPLE_REPLY') ): continue yield ConntrackEntry( ndmsg['nfgen_family'], ndmsg.get_attr('CTA_TUPLE_ORIG'), ndmsg.get_attr('CTA_TUPLE_REPLY'), ndmsg.get_attr('CTA_STATUS'), ndmsg.get_attr('CTA_TIMEOUT'), ndmsg.get_attr('CTA_PROTOINFO'), ndmsg.get_attr('CTA_MARK'), ndmsg.get_attr('CTA_ID'), ndmsg.get_attr('CTA_USE'), )	6.68 KB	2025-05-25 12:26
📁 decoder	-	2025-05-25 12:26
📄 devlink.py import logging from pyroute2.netlink import NLM_F_DUMP, NLM_F_REQUEST from pyroute2.netlink.devlink import DEVLINK_NAMES, DevlinkSocket, devlinkcmd log = logging.getLogger(__name__) class DL(DevlinkSocket): def __init__(self, argv, kwarg): # get specific groups kwarg if 'groups' in kwarg: groups = kwarg['groups'] del kwarg['groups'] else: groups = None # get specific async kwarg if 'async' in kwarg: # FIXME # raise deprecation error after 0.5.3 # log.warning( 'use "async_cache" instead of "async", ' '"async" is a keyword from Python 3.7' ) kwarg['async_cache'] = kwarg.pop('async') if 'async_cache' in kwarg: async_cache = kwarg.pop('async_cache') else: async_cache = False # align groups with async_cache if groups is None: groups = ~0 if async_cache else 0 # continue with init super(DL, self).__init__(argv, **kwarg) # do automatic bind # FIXME: unfortunately we can not omit it here try: self.bind(groups, async_cache=async_cache) except: # thanks to jtluka at redhat.com and the LNST # team for the fixed fd leak super(DL, self).close() raise def list(self): return self.get_dump() def get_dump(self): msg = devlinkcmd() msg['cmd'] = DEVLINK_NAMES['DEVLINK_CMD_GET'] return tuple( self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP ) ) def port_list(self): return self.get_port_dump() def get_port_dump(self): msg = devlinkcmd() msg['cmd'] = DEVLINK_NAMES['DEVLINK_CMD_PORT_GET'] return tuple( self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP ) )	2.02 KB	2025-05-25 12:26
📁 dhcp	-	2025-05-25 12:26
📁 ethtool	-	2025-05-25 12:26
📁 ext	-	2025-05-25 12:26
📁 fixtures	-	2025-05-25 12:26
📁 inotify	-	2025-05-25 12:26
📁 ipdb	-	2025-05-25 12:26
📁 iproute	-	2025-05-25 12:26
📄 ipset.py ''' ipset support. This module is tested with hash:ip, hash:net, list:set and several other ipset structures (like hash:net,iface). There is no guarantee that this module is working with all available ipset modules. It supports almost all kernel commands (create, destroy, flush, rename, swap, test...) ''' import errno import socket from pyroute2.common import basestring from pyroute2.netlink import ( NETLINK_NETFILTER, NLM_F_ACK, NLM_F_DUMP, NLM_F_EXCL, NLM_F_REQUEST, NLMSG_ERROR, ) from pyroute2.netlink.exceptions import IPSetError, NetlinkError from pyroute2.netlink.nfnetlink import NFNL_SUBSYS_IPSET from pyroute2.netlink.nfnetlink.ipset import ( IPSET_CMD_ADD, IPSET_CMD_CREATE, IPSET_CMD_DEL, IPSET_CMD_DESTROY, IPSET_CMD_FLUSH, IPSET_CMD_GET_BYINDEX, IPSET_CMD_GET_BYNAME, IPSET_CMD_HEADER, IPSET_CMD_LIST, IPSET_CMD_PROTOCOL, IPSET_CMD_RENAME, IPSET_CMD_SWAP, IPSET_CMD_TEST, IPSET_CMD_TYPE, IPSET_ERR_BUSY, IPSET_ERR_COMMENT, IPSET_ERR_COUNTER, IPSET_ERR_EXIST, IPSET_ERR_EXIST_SETNAME2, IPSET_ERR_FIND_TYPE, IPSET_ERR_INVALID_CIDR, IPSET_ERR_INVALID_FAMILY, IPSET_ERR_INVALID_MARKMASK, IPSET_ERR_INVALID_NETMASK, IPSET_ERR_IPADDR_IPV4, IPSET_ERR_IPADDR_IPV6, IPSET_ERR_MAX_SETS, IPSET_ERR_PROTOCOL, IPSET_ERR_REFERENCED, IPSET_ERR_SKBINFO, IPSET_ERR_TIMEOUT, IPSET_ERR_TYPE_MISMATCH, IPSET_FLAG_IFACE_WILDCARD, IPSET_FLAG_PHYSDEV, IPSET_FLAG_WITH_COMMENT, IPSET_FLAG_WITH_COUNTERS, IPSET_FLAG_WITH_FORCEADD, IPSET_FLAG_WITH_SKBINFO, ipset_msg, ) from pyroute2.netlink.nlsocket import NetlinkSocket def _nlmsg_error(msg): return msg['header']['type'] == NLMSG_ERROR class PortRange(object): """A simple container for port range with optional protocol Note that optional protocol parameter is not supported by all kernel ipset modules using ports. On the other hand, it's sometimes mandatory to set it (like for hash:net,port ipsets) Example:: udp_proto = socket.getprotobyname("udp") port_range = PortRange(1000, 2000, protocol=udp_proto) ipset.create("foo", stype="hash:net,port") ipset.add("foo", ("192.0.2.0/24", port_range), etype="net,port") ipset.test("foo", ("192.0.2.0/24", port_range), etype="net,port") """ def __init__(self, begin, end, protocol=None): self.begin = begin self.end = end self.protocol = protocol class PortEntry(object): """A simple container for port entry with optional protocol""" def __init__(self, port, protocol=None): self.port = port self.protocol = protocol class IPSet(NetlinkSocket): ''' NFNetlink socket (family=NETLINK_NETFILTER). Implements API to the ipset functionality. ''' policy = { IPSET_CMD_PROTOCOL: ipset_msg, IPSET_CMD_LIST: ipset_msg, IPSET_CMD_TYPE: ipset_msg, IPSET_CMD_HEADER: ipset_msg, IPSET_CMD_GET_BYNAME: ipset_msg, IPSET_CMD_GET_BYINDEX: ipset_msg, } attr_map = { 'iface': 'IPSET_ATTR_IFACE', 'mark': 'IPSET_ATTR_MARK', 'set': 'IPSET_ATTR_NAME', 'mac': 'IPSET_ATTR_ETHER', 'port': 'IPSET_ATTR_PORT', ('ip_from', 1): 'IPSET_ATTR_IP_FROM', ('ip_from', 2): 'IPSET_ATTR_IP2', ('cidr', 1): 'IPSET_ATTR_CIDR', ('cidr', 2): 'IPSET_ATTR_CIDR2', ('ip_to', 1): 'IPSET_ATTR_IP_TO', ('ip_to', 2): 'IPSET_ATTR_IP2_TO', } def __init__(self, version=None, attr_revision=None, nfgen_family=2): super(IPSet, self).__init__(family=NETLINK_NETFILTER) policy = dict( [ (x \| (NFNL_SUBSYS_IPSET << 8), y) for (x, y) in self.policy.items() ] ) self.register_policy(policy) self._nfgen_family = nfgen_family if version is None: msg = self.get_proto_version() version = msg[0].get_attr('IPSET_ATTR_PROTOCOL') self._proto_version = version self._attr_revision = attr_revision def request( self, msg, msg_type, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP, terminate=None, ): msg['nfgen_family'] = self._nfgen_family try: return tuple( self.nlm_request( msg, msg_type \| (NFNL_SUBSYS_IPSET << 8), msg_flags, terminate=terminate, ) ) except NetlinkError as err: raise _IPSetError(err.code, cmd=msg_type) def headers(self, name, kwargs): ''' Get headers of the named ipset. It can be used to test if one ipset exists, since it returns a no such file or directory. ''' return self._list_or_headers(IPSET_CMD_HEADER, name=name, kwargs) def get_proto_version(self, version=6): ''' Get supported protocol version by kernel. version parameter allow to set mandatory (but unused?) IPSET_ATTR_PROTOCOL netlink attribute in the request. ''' msg = ipset_msg() msg['attrs'] = [['IPSET_ATTR_PROTOCOL', version]] return self.request(msg, IPSET_CMD_PROTOCOL) def list(self, argv, kwargs): ''' List installed ipsets. If `name` is provided, list the named ipset or return an empty list. Be warned: netlink does not return an error if given name does not exit, you will receive an empty list. ''' if argv: kwargs['name'] = argv[0] return self._list_or_headers(IPSET_CMD_LIST, kwargs) def _list_or_headers(self, cmd, name=None, flags=None): msg = ipset_msg() msg['attrs'] = [['IPSET_ATTR_PROTOCOL', self._proto_version]] if name is not None: msg['attrs'].append(['IPSET_ATTR_SETNAME', name]) if flags is not None: msg['attrs'].append(['IPSET_ATTR_FLAGS', flags]) return self.request(msg, cmd) def destroy(self, name=None): ''' Destroy one (when name is set) or all ipset (when name is None) ''' msg = ipset_msg() msg['attrs'] = [['IPSET_ATTR_PROTOCOL', self._proto_version]] if name is not None: msg['attrs'].append(['IPSET_ATTR_SETNAME', name]) return self.request( msg, IPSET_CMD_DESTROY, msg_flags=NLM_F_REQUEST \| NLM_F_ACK \| NLM_F_EXCL, terminate=_nlmsg_error, ) def create( self, name, stype='hash:ip', family=socket.AF_INET, exclusive=True, counters=False, comment=False, maxelem=None, forceadd=False, hashsize=None, timeout=None, bitmap_ports_range=None, size=None, skbinfo=False, ): ''' Create an ipset `name` of type `stype`, by default `hash:ip`. Common ipset options are supported: exclusive -- if set, raise an error if the ipset exists * counters -- enable data/packets counters * comment -- enable comments capability * maxelem -- max size of the ipset * forceadd -- you should refer to the ipset manpage * hashsize -- size of the hashtable (if any) * timeout -- enable and set a default value for entries (if not None) * bitmap_ports_range -- set the specified inclusive portrange for the bitmap ipset structure (0, 65536) * size -- Size of the list:set, the default is 8 * skbinfo -- enable skbinfo capability ''' excl_flag = NLM_F_EXCL if exclusive else 0 msg = ipset_msg() cadt_flags = 0 if counters: cadt_flags \|= IPSET_FLAG_WITH_COUNTERS if comment: cadt_flags \|= IPSET_FLAG_WITH_COMMENT if forceadd: cadt_flags \|= IPSET_FLAG_WITH_FORCEADD if skbinfo: cadt_flags \|= IPSET_FLAG_WITH_SKBINFO if stype == 'bitmap:port' and bitmap_ports_range is None: raise ValueError('Missing value bitmap_ports_range') data = {'attrs': []} if cadt_flags: data['attrs'] += [['IPSET_ATTR_CADT_FLAGS', cadt_flags]] if maxelem is not None: data['attrs'] += [['IPSET_ATTR_MAXELEM', maxelem]] if hashsize is not None: data['attrs'] += [["IPSET_ATTR_HASHSIZE", hashsize]] elif size is not None and stype == 'list:set': data['attrs'] += [['IPSET_ATTR_SIZE', size]] if timeout is not None: data['attrs'] += [["IPSET_ATTR_TIMEOUT", timeout]] if bitmap_ports_range is not None and stype == 'bitmap:port': # Set the bitmap range A bitmap type of set # can store up to 65536 entries if isinstance(bitmap_ports_range, PortRange): data['attrs'] += [ ['IPSET_ATTR_PORT_FROM', bitmap_ports_range.begin] ] data['attrs'] += [ ['IPSET_ATTR_PORT_TO', bitmap_ports_range.end] ] else: data['attrs'] += [ ['IPSET_ATTR_PORT_FROM', bitmap_ports_range[0]] ] data['attrs'] += [ ['IPSET_ATTR_PORT_TO', bitmap_ports_range[1]] ] if self._attr_revision is None: # Get the last revision supported by kernel revision = self.get_supported_revisions(stype)[1] else: revision = self._attr_revision msg['attrs'] = [ ['IPSET_ATTR_PROTOCOL', self._proto_version], ['IPSET_ATTR_SETNAME', name], ['IPSET_ATTR_TYPENAME', stype], ['IPSET_ATTR_FAMILY', family], ['IPSET_ATTR_REVISION', revision], ["IPSET_ATTR_DATA", data], ] return self.request( msg, IPSET_CMD_CREATE, msg_flags=NLM_F_REQUEST \| NLM_F_ACK \| excl_flag, terminate=_nlmsg_error, ) @staticmethod def _family_to_version(family): if family is not None: if family == socket.AF_INET: return 'IPSET_ATTR_IPADDR_IPV4' elif family == socket.AF_INET6: return 'IPSET_ATTR_IPADDR_IPV6' elif family == socket.AF_UNSPEC: return None raise TypeError('unknown family') def _entry_to_data_attrs(self, entry, etype, ip_version): attrs = [] ip_count = 0 if etype == 'set': attrs += [['IPSET_ATTR_NAME', entry]] return attrs # We support string (for one element, and for users calling this # function like a command line), and tuple/list if isinstance(entry, basestring): entry = entry.split(',') if isinstance(entry, (int, PortRange, PortEntry)): entry = [entry] for e, t in zip(entry, etype.split(',')): if t in ('ip', 'net'): ip_count += 1 if t == 'net': if '/' in e: e, cidr = e.split('/') attrs += [ [self.attr_map[('cidr', ip_count)], int(cidr)] ] elif '-' in e: e, to = e.split('-') attrs += [ [ self.attr_map[('ip_to', ip_count)], {'attrs': [[ip_version, to]]}, ] ] attrs += [ [ self.attr_map[('ip_from', ip_count)], {'attrs': [[ip_version, e]]}, ] ] elif t == "port": if isinstance(e, PortRange): attrs += [['IPSET_ATTR_PORT_FROM', e.begin]] attrs += [['IPSET_ATTR_PORT_TO', e.end]] if e.protocol is not None: attrs += [['IPSET_ATTR_PROTO', e.protocol]] elif isinstance(e, PortEntry): attrs += [['IPSET_ATTR_PORT', e.port]] if e.protocol is not None: attrs += [['IPSET_ATTR_PROTO', e.protocol]] else: attrs += [[self.attr_map[t], e]] else: attrs += [[self.attr_map[t], e]] return attrs def _add_delete_test( self, name, entry, family, cmd, exclusive, comment=None, timeout=None, etype="ip", packets=None, bytes=None, skbmark=None, skbprio=None, skbqueue=None, wildcard=False, physdev=False, ): excl_flag = NLM_F_EXCL if exclusive else 0 adt_flags = 0 if wildcard: adt_flags \|= IPSET_FLAG_IFACE_WILDCARD if physdev: adt_flags \|= IPSET_FLAG_PHYSDEV ip_version = self._family_to_version(family) data_attrs = self._entry_to_data_attrs(entry, etype, ip_version) if comment is not None: data_attrs += [ ["IPSET_ATTR_COMMENT", comment], ["IPSET_ATTR_CADT_LINENO", 0], ] if timeout is not None: data_attrs += [["IPSET_ATTR_TIMEOUT", timeout]] if bytes is not None: data_attrs += [["IPSET_ATTR_BYTES", bytes]] if packets is not None: data_attrs += [["IPSET_ATTR_PACKETS", packets]] if skbmark is not None: data_attrs += [["IPSET_ATTR_SKBMARK", skbmark]] if skbprio is not None: data_attrs += [["IPSET_ATTR_SKBPRIO", skbprio]] if skbqueue is not None: data_attrs += [["IPSET_ATTR_SKBQUEUE", skbqueue]] if adt_flags: data_attrs += [["IPSET_ATTR_CADT_FLAGS", adt_flags]] msg = ipset_msg() msg['attrs'] = [ ['IPSET_ATTR_PROTOCOL', self._proto_version], ['IPSET_ATTR_SETNAME', name], ['IPSET_ATTR_DATA', {'attrs': data_attrs}], ] return self.request( msg, cmd, msg_flags=NLM_F_REQUEST \| NLM_F_ACK \| excl_flag, terminate=_nlmsg_error, ) def add( self, name, entry, family=socket.AF_INET, exclusive=True, comment=None, timeout=None, etype="ip", skbmark=None, skbprio=None, skbqueue=None, wildcard=False, kwargs ): ''' Add a member to the ipset. etype is the entry type that you add to the ipset. It's related to the ipset type. For example, use "ip" for one hash:ip or bitmap:ip ipset. When your ipset store a tuple, like "hash:net,iface", you must use a comma a separator (etype="net,iface") entry is a string for "ip" and "net" objects. For ipset with several dimensions, you must use a tuple (or a list) of objects. "port" type is specific, since you can use integer of specialized containers like :class:`PortEntry` and :class:`PortRange` Examples:: ipset = IPSet() ipset.create("foo", stype="hash:ip") ipset.add("foo", "198.51.100.1", etype="ip") ipset = IPSet() ipset.create("bar", stype="bitmap:port", bitmap_ports_range=(1000, 2000)) ipset.add("bar", 1001, etype="port") ipset.add("bar", PortRange(1500, 2000), etype="port") ipset = IPSet() import socket protocol = socket.getprotobyname("tcp") ipset.create("foobar", stype="hash:net,port") port_entry = PortEntry(80, protocol=protocol) ipset.add("foobar", ("198.51.100.0/24", port_entry), etype="net,port") wildcard option enable kernel wildcard matching on interface name for net,iface entries. ''' return self._add_delete_test( name, entry, family, IPSET_CMD_ADD, exclusive, comment=comment, timeout=timeout, etype=etype, skbmark=skbmark, skbprio=skbprio, skbqueue=skbqueue, wildcard=wildcard, kwargs ) def delete( self, name, entry, family=socket.AF_INET, exclusive=True, etype="ip" ): ''' Delete a member from the ipset. See :func:`add` method for more information on etype. ''' return self._add_delete_test( name, entry, family, IPSET_CMD_DEL, exclusive, etype=etype ) def test(self, name, entry, family=socket.AF_INET, etype="ip"): ''' Test if entry is part of an ipset See :func:`add` method for more information on etype. ''' try: self._add_delete_test( name, entry, family, IPSET_CMD_TEST, False, etype=etype ) return True except IPSetError as e: if e.code == IPSET_ERR_EXIST: return False raise e def swap(self, set_a, set_b): ''' Swap two ipsets. They must have compatible content type. ''' msg = ipset_msg() msg['attrs'] = [ ['IPSET_ATTR_PROTOCOL', self._proto_version], ['IPSET_ATTR_SETNAME', set_a], ['IPSET_ATTR_TYPENAME', set_b], ] return self.request( msg, IPSET_CMD_SWAP, msg_flags=NLM_F_REQUEST \| NLM_F_ACK, terminate=_nlmsg_error, ) def flush(self, name=None): ''' Flush all ipsets. When name is set, flush only this ipset. ''' msg = ipset_msg() msg['attrs'] = [['IPSET_ATTR_PROTOCOL', self._proto_version]] if name is not None: msg['attrs'].append(['IPSET_ATTR_SETNAME', name]) return self.request( msg, IPSET_CMD_FLUSH, msg_flags=NLM_F_REQUEST \| NLM_F_ACK, terminate=_nlmsg_error, ) def rename(self, name_src, name_dst): ''' Rename the ipset. ''' msg = ipset_msg() msg['attrs'] = [ ['IPSET_ATTR_PROTOCOL', self._proto_version], ['IPSET_ATTR_SETNAME', name_src], ['IPSET_ATTR_TYPENAME', name_dst], ] return self.request( msg, IPSET_CMD_RENAME, msg_flags=NLM_F_REQUEST \| NLM_F_ACK, terminate=_nlmsg_error, ) def _get_set_by(self, cmd, value): # Check that IPSet version is supported if self._proto_version < 7: raise NotImplementedError() msg = ipset_msg() if cmd == IPSET_CMD_GET_BYNAME: msg['attrs'] = [ ['IPSET_ATTR_PROTOCOL', self._proto_version], ['IPSET_ATTR_SETNAME', value], ] if cmd == IPSET_CMD_GET_BYINDEX: msg['attrs'] = [ ['IPSET_ATTR_PROTOCOL', self._proto_version], ['IPSET_ATTR_INDEX', value], ] return self.request(msg, cmd) def get_set_byname(self, name): ''' Get a set by its name ''' return self._get_set_by(IPSET_CMD_GET_BYNAME, name) def get_set_byindex(self, index): ''' Get a set by its index ''' return self._get_set_by(IPSET_CMD_GET_BYINDEX, index) def get_supported_revisions(self, stype, family=socket.AF_INET): ''' Return minimum and maximum of revisions supported by the kernel. Each ipset module (like hash:net, hash:ip, etc) has several revisions. Newer revisions often have more features or more performances. Thanks to this call, you can ask the kernel the list of supported revisions. You can manually set/force revisions used in IPSet constructor. Example:: ipset = IPSet() ipset.get_supported_revisions("hash:net") ipset.get_supported_revisions("hash:net,port,net") ''' msg = ipset_msg() msg['attrs'] = [ ['IPSET_ATTR_PROTOCOL', self._proto_version], ['IPSET_ATTR_TYPENAME', stype], ['IPSET_ATTR_FAMILY', family], ] response = self.request( msg, IPSET_CMD_TYPE, msg_flags=NLM_F_REQUEST \| NLM_F_ACK, terminate=_nlmsg_error, ) min_revision = response[0].get_attr("IPSET_ATTR_PROTOCOL_MIN") max_revision = response[0].get_attr("IPSET_ATTR_REVISION") return min_revision, max_revision class _IPSetError(IPSetError): ''' Proxy class to not import all specifics ipset code in exceptions.py Out of the ipset module, a caller should use parent class instead ''' def __init__(self, code, msg=None, cmd=None): if code in self.base_map: msg = self.base_map[code] elif cmd in self.cmd_map: error_map = self.cmd_map[cmd] if code in error_map: msg = error_map[code] super(_IPSetError, self).__init__(code, msg) base_map = { IPSET_ERR_PROTOCOL: "Kernel error received:" " ipset protocol error", IPSET_ERR_INVALID_CIDR: "The value of the CIDR parameter of" " the IP address is invalid", IPSET_ERR_TIMEOUT: "Timeout cannot be used: set was created" " without timeout support", IPSET_ERR_IPADDR_IPV4: "An IPv4 address is expected, but" " not received", IPSET_ERR_IPADDR_IPV6: "An IPv6 address is expected, but" " not received", IPSET_ERR_COUNTER: "Packet/byte counters cannot be used:" " set was created without counter support", IPSET_ERR_COMMENT: "Comment string is too long!", IPSET_ERR_SKBINFO: "Skbinfo mapping cannot be used: " " set was created without skbinfo support", } c_map = { errno.EEXIST: "Set cannot be created: set with the same" " name already exists", IPSET_ERR_FIND_TYPE: "Kernel error received: " "set type not supported", IPSET_ERR_MAX_SETS: "Kernel error received: maximal number of" " sets reached, cannot create more.", IPSET_ERR_INVALID_NETMASK: "The value of the netmask parameter" " is invalid", IPSET_ERR_INVALID_MARKMASK: "The value of the markmask parameter" " is invalid", IPSET_ERR_INVALID_FAMILY: "Protocol family not supported by the" " set type", } destroy_map = { IPSET_ERR_BUSY: "Set cannot be destroyed: it is in use" " by a kernel component" } r_map = { IPSET_ERR_EXIST_SETNAME2: "Set cannot be renamed: a set with the" " new name already exists", IPSET_ERR_REFERENCED: "Set cannot be renamed: it is in use by" " another system", } s_map = { IPSET_ERR_EXIST_SETNAME2: "Sets cannot be swapped: the second set" " does not exist", IPSET_ERR_TYPE_MISMATCH: "The sets cannot be swapped: their type" " does not match", } a_map = { IPSET_ERR_EXIST: "Element cannot be added to the set: it's" " already added" } del_map = { IPSET_ERR_EXIST: "Element cannot be deleted from the set:" " it's not added" } cmd_map = { IPSET_CMD_CREATE: c_map, IPSET_CMD_DESTROY: destroy_map, IPSET_CMD_RENAME: r_map, IPSET_CMD_SWAP: s_map, IPSET_CMD_ADD: a_map, IPSET_CMD_DEL: del_map, }	23.57 KB	2025-05-25 12:26
📄 ipvs.py ''' IPVS -- IP Virtual Server ------------------------- IPVS configuration is done via generic netlink protocol. At the low level one can use it with a GenericNetlinkSocket, binding it to "IPVS" generic netlink family. But for the convenience the library provides utility classes: * IPVS -- a socket class to access the API * IPVSService -- a class to define IPVS service records * IPVSDest -- a class to define real server records Dump all the records:: from pyroute2 import IPVS, IPVSDest, IPVSService # run the socket ipvs = IPVS() # iterate all the IPVS services for s in ipvs.service("dump"): # create a utility object from a netlink message service = IPVSService.from_message(s) print("Service: ", service) # iterate all the real servers for this service for d in ipvs.dest("dump", service=service): # create and print a utility object dest = IPVSDest.from_message(d) print(" Real server: ", dest) Create a service and a real server record:: from socket import IPPROTO_TCP from pyroute2 import IPVS, IPVSDest, IPVSService ipvs = IPVS() service = IPVSService(addr="192.168.122.1", port=80, protocol=IPPROTO_TCP) real_server = IPVSDest(addr="10.0.2.20", port=80) ipvs.service("add", service=service) ipvs.dest("add", service=service, dest=real_server) Delete a service:: from pyroute2 import IPVS, IPVSService ipvs = IPVS() ipvs.service("del", service=IPVSService( addr="192.168.122.1", port=80, protocol=IPPROTO_TCP ) ) ''' from socket import AF_INET from pyroute2.common import get_address_family from pyroute2.netlink.generic import ipvs from pyroute2.netlink.nlsocket import NetlinkRequest from pyroute2.requests.common import NLAKeyTransform from pyroute2.requests.main import RequestProcessor class ServiceFieldFilter(NLAKeyTransform): _nla_prefix = 'IPVS_SVC_ATTR_' def set_addr(self, context, value): ret = {"addr": value} if "af" in context.keys(): family = context["af"] else: family = ret["af"] = get_address_family(value) if family == AF_INET and "netmask" not in context.keys(): ret["netmask"] = "255.255.255.255" return ret class DestFieldFilter(NLAKeyTransform): _nla_prefix = 'IPVS_DEST_ATTR_' def set_addr(self, context, value): ret = {"addr": value} if "addr_family" not in context.keys(): ret["addr_family"] = get_address_family(value) return ret class NLAFilter(RequestProcessor): msg = None keys = tuple() field_filters = None nla = None default_values = {} def __init__(self, **kwarg): dict.update(self, self.default_values) # save field filters flt = self.field_filters # init resets the filters, not fixed yet super().__init__(prime=kwarg) # restore filters for f in flt: self.add_filter(f) @classmethod def from_message(cls, msg): obj = cls() for key, value in msg.get(cls.nla)["attrs"]: obj[key] = value obj.pop("stats", None) obj.pop("stats64", None) return obj def dump_nla(self, items=None): if items is None: items = self.items() self.update(self) self.finalize() return { "attrs": list( map(lambda x: (self.msg.name2nla(x[0]), x[1]), items) ) } def dump_key(self): return self.dump_nla( items=filter(lambda x: x[0] in self.key_fields, self.items()) ) class IPVSService(NLAFilter): field_filters = [ServiceFieldFilter()] msg = ipvs.ipvsmsg.service key_fields = ("af", "protocol", "addr", "port") nla = "IPVS_CMD_ATTR_SERVICE" default_values = { "timeout": 0, "sched_name": "wlc", "flags": {"flags": 0, "mask": 0xFFFF}, } class IPVSDest(NLAFilter): field_filters = [DestFieldFilter()] msg = ipvs.ipvsmsg.dest nla = "IPVS_CMD_ATTR_DEST" default_values = { "fwd_method": 3, "weight": 1, "tun_type": 0, "tun_port": 0, "tun_flags": 0, "u_thresh": 0, "l_thresh": 0, } class IPVS(ipvs.IPVSSocket): def service(self, command, service=None): command_map = { "add": (ipvs.IPVS_CMD_NEW_SERVICE, "create"), "set": (ipvs.IPVS_CMD_SET_SERVICE, "change"), "update": (ipvs.IPVS_CMD_DEL_SERVICE, "change"), "del": (ipvs.IPVS_CMD_DEL_SERVICE, "req"), "get": (ipvs.IPVS_CMD_GET_SERVICE, "get"), "dump": (ipvs.IPVS_CMD_GET_SERVICE, "dump"), } cmd, flags = NetlinkRequest.calculate_request_type( command, command_map ) msg = ipvs.ipvsmsg() msg["cmd"] = cmd msg["version"] = ipvs.GENL_VERSION if service is not None: msg["attrs"] = [("IPVS_CMD_ATTR_SERVICE", service.dump_nla())] return self.nlm_request(msg, msg_type=self.prid, msg_flags=flags) def dest(self, command, service, dest=None): command_map = { "add": (ipvs.IPVS_CMD_NEW_DEST, "create"), "set": (ipvs.IPVS_CMD_SET_DEST, "change"), "update": (ipvs.IPVS_CMD_DEL_DEST, "change"), "del": (ipvs.IPVS_CMD_DEL_DEST, "req"), "get": (ipvs.IPVS_CMD_GET_DEST, "get"), "dump": (ipvs.IPVS_CMD_GET_DEST, "dump"), } cmd, flags = self.make_request_type(command, command_map) msg = ipvs.ipvsmsg() msg["cmd"] = cmd msg["version"] = 0x1 msg["attrs"] = [("IPVS_CMD_ATTR_SERVICE", service.dump_key())] if dest is not None: msg["attrs"].append(("IPVS_CMD_ATTR_DEST", dest.dump_nla())) return self.nlm_request(msg, msg_type=self.prid, msg_flags=flags)	5.87 KB	2025-05-25 12:26
📄 iwutil.py # -- coding: utf-8 -- ''' IW module ========= Experimental wireless module — nl80211 support. Disclaimer ---------- Unlike IPRoute, which is mostly usable, though is far from complete yet, the IW module is in the very initial state. Neither the module itself, nor the message class cover the nl80211 functionality reasonably enough. So if you're going to use it, brace yourself — debug is coming. Messages -------- nl80211 messages are defined here:: pyroute2/netlink/nl80211/__init__.py Pls notice NLAs of type `hex`. On the early development stage `hex` allows to inspect incoming data as a hex dump and, occasionally, even make requests with such NLAs. But it's not a production way. The type `hex` in the NLA definitions means that this particular NLA is not handled yet properly. If you want to use some NLA which is defined as `hex` yet, pls find out a specific type, patch the message class and submit your pull request on github. If you're not familiar with NLA types, take a look at RTNL definitions:: pyroute2/netlink/rtnl/ndmsg.py and so on. Communication with the kernel ----------------------------- There are several methods of the communication with the kernel. * `sendto()` — lowest possible, send a raw binary data * `put()` — send a netlink message * `nlm_request()` — send a message, return the response * `get()` — get a netlink message * `recv()` — get a raw binary data from the kernel There are no errors on `put()` usually. Any `permission denied`, any `invalid value` errors are returned from the kernel with netlink also. So if you do `put()`, but don't do `get()`, be prepared to miss errors. The preferred method for the communication is `nlm_request()`. It tracks the message ID, returns the corresponding response. In the case of errors `nlm_request()` raises an exception. To get the response on any operation with nl80211, use flag `NLM_F_ACK`. Reverse it ---------- If you're too lazy to read the kernel sources, but still need something not implemented here, you can use reverse engineering on a reference implementation. E.g.:: # strace -e trace=network -f -x -s 4096 \\ iw phy phy0 interface add test type monitor Will dump all the netlink traffic between the program `iw` and the kernel. Three first packets are the generic netlink protocol discovery, you can ignore them. All that follows, is the nl80211 traffic:: sendmsg(3, {msg_name(12)={sa_family=AF_NETLINK, ... }, msg_iov(1)=[{"\\x30\\x00\\x00\\x00\\x1b\\x00\\x05 ...", 48}], msg_controllen=0, msg_flags=0}, 0) = 48 recvmsg(3, {msg_name(12)={sa_family=AF_NETLINK, ... }, msg_iov(1)=[{"\\x58\\x00\\x00\\x00\\x1b\\x00\\x00 ...", 16384}], msg_controllen=0, msg_flags=0}, 0) = 88 ... With `-s 4096` you will get the full dump. Then copy the strings from `msg_iov` to a file, let's say `data`, and run the decoder:: $ pwd /home/user/Projects/pyroute2 $ export PYTHONPATH=`pwd` $ python scripts/decoder.py pyroute2.netlink.nl80211.nl80211cmd data You will get the session decoded:: {'attrs': [['NL80211_ATTR_WIPHY', 0], ['NL80211_ATTR_IFNAME', 'test'], ['NL80211_ATTR_IFTYPE', 6]], 'cmd': 7, 'header': {'flags': 5, 'length': 48, 'pid': 3292542647, 'sequence_number': 1430426434, 'type': 27}, 'reserved': 0, 'version': 0} {'attrs': [['NL80211_ATTR_IFINDEX', 23811], ['NL80211_ATTR_IFNAME', 'test'], ['NL80211_ATTR_WIPHY', 0], ['NL80211_ATTR_IFTYPE', 6], ['NL80211_ATTR_WDEV', 4], ['NL80211_ATTR_MAC', 'a4:4e:31:43:1c:7c'], ['NL80211_ATTR_GENERATION', '02:00:00:00']], 'cmd': 7, 'header': {'flags': 0, 'length': 88, 'pid': 3292542647, 'sequence_number': 1430426434, 'type': 27}, 'reserved': 0, 'version': 1} Now you know, how to do a request and what you will get as a response. Sample collected data is in the `scripts` directory. Submit changes -------------- Please do not hesitate to submit the changes on github. Without your patches this module will not evolve. ''' import logging from pyroute2.netlink import NLM_F_ACK, NLM_F_DUMP, NLM_F_REQUEST from pyroute2.netlink.nl80211 import ( BSS_STATUS_NAMES, CHAN_WIDTH, IFTYPE_NAMES, NL80211, NL80211_NAMES, SCAN_FLAGS_NAMES, nl80211cmd, ) log = logging.getLogger(__name__) class IW(NL80211): def __init__(self, argv, kwarg): # get specific groups kwarg if 'groups' in kwarg: groups = kwarg['groups'] del kwarg['groups'] else: groups = None # get specific async kwarg if 'async' in kwarg: # FIXME # raise deprecation error after 0.5.3 # log.warning( 'use "async_cache" instead of "async", ' '"async" is a keyword from Python 3.7' ) kwarg['async_cache'] = kwarg.pop('async') if 'async_cache' in kwarg: async_cache = kwarg.pop('async_cache') else: async_cache = False # align groups with async_cache if groups is None: groups = ~0 if async_cache else 0 # continue with init super(IW, self).__init__(argv, *kwarg) # do automatic bind # FIXME: unfortunately we can not omit it here self.bind(groups, async_cache=async_cache) def del_interface(self, dev): ''' Delete a virtual interface - dev — device index ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_DEL_INTERFACE'] msg['attrs'] = [['NL80211_ATTR_IFINDEX', dev]] self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_ACK ) def add_interface(self, ifname, iftype, dev=None, phy=0): ''' Create a virtual interface - ifname — name of the interface to create - iftype — interface type to create - dev — device index - phy — phy index One should specify `dev` (device index) or `phy` (phy index). If no one specified, phy == 0. `iftype` can be integer or string: 1. adhoc 2. station 3. ap 4. ap_vlan 5. wds 6. monitor 7. mesh_point 8. p2p_client 9. p2p_go 10. p2p_device 11. ocb ''' # lookup the interface type iftype = IFTYPE_NAMES.get(iftype, iftype) if not isinstance(iftype, int): raise TypeError('iftype must be int') msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_NEW_INTERFACE'] msg['attrs'] = [ ['NL80211_ATTR_IFNAME', ifname], ['NL80211_ATTR_IFTYPE', iftype], ] if dev is not None: msg['attrs'].append(['NL80211_ATTR_IFINDEX', dev]) elif phy is not None: msg['attrs'].append(['NL80211_ATTR_WIPHY', phy]) else: raise TypeError('no device specified') self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_ACK ) def list_dev(self): ''' Get list of all wifi network interfaces ''' return self.get_interfaces_dump() def list_wiphy(self): ''' Get list of all phy devices ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_GET_WIPHY'] return self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP ) def _get_phy_name(self, attr): return 'phy%i' % attr.get_attr('NL80211_ATTR_WIPHY') def _get_frequency(self, attr): return attr.get_attr('NL80211_ATTR_WIPHY_FREQ') or 0 def get_interfaces_dict(self): ''' Get interfaces dictionary ''' ret = {} for wif in self.get_interfaces_dump(): chan_width = wif.get_attr('NL80211_ATTR_CHANNEL_WIDTH') freq = self._get_frequency(wif) if chan_width is not None else 0 wifname = wif.get_attr('NL80211_ATTR_IFNAME') ret[wifname] = [ wif.get_attr('NL80211_ATTR_IFINDEX'), self._get_phy_name(wif), wif.get_attr('NL80211_ATTR_MAC'), freq, chan_width, ] return ret def get_interfaces_dump(self): ''' Get interfaces dump ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_GET_INTERFACE'] return self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP ) def get_interface_by_phy(self, attr): ''' Get interface by phy ( use x.get_attr('NL80211_ATTR_WIPHY') ) ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_GET_INTERFACE'] msg['attrs'] = [['NL80211_ATTR_WIPHY', attr]] return self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP ) def get_interface_by_ifindex(self, ifindex): ''' Get interface by ifindex ( use x.get_attr('NL80211_ATTR_IFINDEX') ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_GET_INTERFACE'] msg['attrs'] = [['NL80211_ATTR_IFINDEX', ifindex]] return self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST ) def get_stations(self, ifindex): ''' Get stations by ifindex ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_GET_STATION'] msg['attrs'] = [['NL80211_ATTR_IFINDEX', ifindex]] return self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP ) def join_ibss( self, ifindex, ssid, freq, bssid=None, channel_fixed=False, width=None, center=None, center2=None, ): ''' Connect to network by ssid - ifindex - IFINDEX of the interface to perform the connection - ssid - Service set identification - freq - Frequency in MHz - bssid - The MAC address of target interface - channel_fixed: Boolean flag - width - Channel width - center - Central frequency of the 40/80/160 MHz channel - center2 - Center frequency of second segment if 80P80 If the flag of channel_fixed is True, one should specify both the width and center of the channel `width` can be integer of string: 0. 20_noht 1. 20 2. 40 3. 80 4. 80p80 5. 160 6. 5 7. 10 ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_JOIN_IBSS'] msg['attrs'] = [ ['NL80211_ATTR_IFINDEX', ifindex], ['NL80211_ATTR_SSID', ssid], ['NL80211_ATTR_WIPHY_FREQ', freq], ] if channel_fixed: msg['attrs'].append(['NL80211_ATTR_FREQ_FIXED', None]) width = CHAN_WIDTH.get(width, width) if not isinstance(width, int): raise TypeError('width must be int') if width in [2, 3, 5] and center: msg['attrs'].append(['NL80211_ATTR_CHANNEL_WIDTH', width]) msg['attrs'].append(['NL80211_ATTR_CENTER_FREQ1', center]) elif width == 4 and center and center2: msg['attrs'].append(['NL80211_ATTR_CHANNEL_WIDTH', width]) msg['attrs'].append(['NL80211_ATTR_CENTER_FREQ1', center]) msg['attrs'].append(['NL80211_ATTR_CENTER_FREQ2', center2]) elif width in [0, 1, 6, 7]: msg['attrs'].append(['NL80211_ATTR_CHANNEL_WIDTH', width]) else: raise TypeError('No channel specified') if bssid is not None: msg['attrs'].append(['NL80211_ATTR_MAC', bssid]) self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_ACK ) def leave_ibss(self, ifindex): ''' Leave the IBSS -- the IBSS is determined by the network interface ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_LEAVE_IBSS'] msg['attrs'] = [['NL80211_ATTR_IFINDEX', ifindex]] self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_ACK ) def authenticate(self, ifindex, bssid, ssid, freq, auth_type=0): ''' Send an Authentication management frame. ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_AUTHENTICATE'] msg['attrs'] = [ ['NL80211_ATTR_IFINDEX', ifindex], ['NL80211_ATTR_MAC', bssid], ['NL80211_ATTR_SSID', ssid], ['NL80211_ATTR_WIPHY_FREQ', freq], ['NL80211_ATTR_AUTH_TYPE', auth_type], ] self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_ACK ) def deauthenticate(self, ifindex, bssid, reason_code=0x01): ''' Send a Deauthentication management frame. ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_DEAUTHENTICATE'] msg['attrs'] = [ ['NL80211_ATTR_IFINDEX', ifindex], ['NL80211_ATTR_MAC', bssid], ['NL80211_ATTR_REASON_CODE', reason_code], ] self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_ACK ) def associate(self, ifindex, bssid, ssid, freq, info_elements=None): ''' Send an Association request frame. ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_ASSOCIATE'] msg['attrs'] = [ ['NL80211_ATTR_IFINDEX', ifindex], ['NL80211_ATTR_MAC', bssid], ['NL80211_ATTR_SSID', ssid], ['NL80211_ATTR_WIPHY_FREQ', freq], ] if info_elements is not None: msg['attrs'].append(['NL80211_ATTR_IE', info_elements]) self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_ACK ) def disassociate(self, ifindex, bssid, reason_code=0x03): ''' Send a Disassociation management frame. ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_DISASSOCIATE'] msg['attrs'] = [ ['NL80211_ATTR_IFINDEX', ifindex], ['NL80211_ATTR_MAC', bssid], ['NL80211_ATTR_REASON_CODE', reason_code], ] self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_ACK ) def connect(self, ifindex, ssid, bssid=None): ''' Connect to the ap with ssid and bssid ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_CONNECT'] msg['attrs'] = [ ['NL80211_ATTR_IFINDEX', ifindex], ['NL80211_ATTR_SSID', ssid], ] if bssid is not None: msg['attrs'].append(['NL80211_ATTR_MAC', bssid]) self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_ACK ) def disconnect(self, ifindex): ''' Disconnect the device ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_DISCONNECT'] msg['attrs'] = [['NL80211_ATTR_IFINDEX', ifindex]] self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_ACK ) def survey(self, ifindex): ''' Return the survey info. ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_GET_SURVEY'] msg['attrs'] = [['NL80211_ATTR_IFINDEX', ifindex]] return self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP ) def scan(self, ifindex, ssids=None, flush_cache=False): ''' Trigger scan and get results. Triggering scan usually requires root, and can take a couple of seconds. ''' # Prepare a second netlink socket to get the scan results. # The issue is that the kernel can send the results notification # before we get answer for the NL80211_CMD_TRIGGER_SCAN nsock = NL80211() nsock.bind() nsock.add_membership('scan') # send scan request msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_TRIGGER_SCAN'] msg['attrs'] = [['NL80211_ATTR_IFINDEX', ifindex]] # If a list of SSIDs is provided, active scanning should be performed if ssids is not None: if isinstance(ssids, list): msg['attrs'].append(['NL80211_ATTR_SCAN_SSIDS', ssids]) scan_flags = 0 if flush_cache: # Flush the cache before scanning scan_flags \|= SCAN_FLAGS_NAMES['NL80211_SCAN_FLAG_FLUSH'] msg['attrs'].append(['NL80211_ATTR_SCAN_FLAGS', scan_flags]) self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_ACK ) # monitor the results notification on the secondary socket scanResultNotFound = True while scanResultNotFound: listMsg = nsock.get() for msg in listMsg: if msg["event"] == "NL80211_CMD_NEW_SCAN_RESULTS": scanResultNotFound = False break # close the secondary socket nsock.close() # request the results msg2 = nl80211cmd() msg2['cmd'] = NL80211_NAMES['NL80211_CMD_GET_SCAN'] msg2['attrs'] = [['NL80211_ATTR_IFINDEX', ifindex]] return self.nlm_request( msg2, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP ) def get_associated_bss(self, ifindex): ''' Returns the same info like scan() does, but only about the currently associated BSS. Unlike scan(), it returns immediately and doesn't require root. ''' # When getting scan results without triggering scan first, # you'll always get the information about currently associated BSS # # However, it may return other BSS, if last scan wasn't very # long time go msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_GET_SCAN'] msg['attrs'] = [['NL80211_ATTR_IFINDEX', ifindex]] res = self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP ) for x in res: attr_bss = x.get_attr('NL80211_ATTR_BSS') if attr_bss is not None: status = attr_bss.get_attr('NL80211_BSS_STATUS') if status in ( BSS_STATUS_NAMES['associated'], BSS_STATUS_NAMES['ibss_joined'], ): return x return None def get_regulatory_domain(self, attr=None): ''' Get regulatory domain information. If attr specified, get regulatory domain information for this device ( use x.get_attr('NL80211_ATTR_WIPHY') ). ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_GET_REG'] flags = NLM_F_REQUEST if attr is None: flags \|= NLM_F_DUMP else: msg['attrs'] = [['NL80211_ATTR_WIPHY', attr]] return self.nlm_request(msg, msg_type=self.prid, msg_flags=flags) def set_regulatory_domain(self, alpha2): ''' Set regulatory domain. ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_REQ_SET_REG'] msg['attrs'] = [['NL80211_ATTR_REG_ALPHA2', alpha2]] self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_ACK ) def set_tx_power(self, dev, mode, mbm=None): ''' Set TX power of interface. - dev — device index - mode — TX power setting (0 - auto, 1 - limit, 2 - fixed) - mbm — TX power in mBm (dBm 100) ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_SET_WIPHY'] msg['attrs'] = [ ['NL80211_ATTR_IFINDEX', dev], ['NL80211_ATTR_WIPHY_TX_POWER_SETTING', mode], ] if mbm is not None: msg['attrs'].append(['NL80211_ATTR_WIPHY_TX_POWER_LEVEL', mbm]) self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_ACK ) def set_wiphy_netns_by_pid(self, wiphy, pid): ''' Set wiphy network namespace to process network namespace. ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_SET_WIPHY_NETNS'] msg['attrs'] = [ ['NL80211_ATTR_WIPHY', wiphy], ['NL80211_ATTR_PID', pid], ] self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_ACK ) def set_wiphy_netns_by_fd(self, wiphy, netns_fd): ''' Set wiphy network namespace to namespace referenced by fd. ''' msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_SET_WIPHY_NETNS'] msg['attrs'] = [ ['NL80211_ATTR_WIPHY', wiphy], ['NL80211_ATTR_NETNS_FD', netns_fd], ] self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_ACK ) def set_interface_type(self, ifindex, iftype): ''' Set interface type - ifindex — device index - iftype — interface type `iftype` can be integer or string: 1. adhoc 2. station 3. ap 4. ap_vlan 5. wds 6. monitor 7. mesh_point 8. p2p_client 9. p2p_go 10. p2p_device 11. ocb ''' iftype = IFTYPE_NAMES.get(iftype, iftype) if not isinstance(iftype, int): raise TypeError('iftype must be int') msg = nl80211cmd() msg['cmd'] = NL80211_NAMES['NL80211_CMD_SET_INTERFACE'] msg['attrs'] = [ ['NL80211_ATTR_IFINDEX', ifindex], ['NL80211_ATTR_IFTYPE', iftype], ] self.nlm_request( msg, msg_type=self.prid, msg_flags=NLM_F_REQUEST \| NLM_F_ACK ) def get_interface_type(self, ifindex) -> str: ''' return interface type name ''' dump = self.get_interface_by_ifindex(ifindex) type = None for d in dump: type = d.get_attr('NL80211_ATTR_IFTYPE') if type is not None: for key, value in IFTYPE_NAMES.items(): if value == type: res = key else: res = 'Not Found Type' return res	22.85 KB	2025-05-25 12:26
📄 minimal.py ## # # This module contains all the public symbols from the library. # ## # # Version # try: from pyroute2.config.version import __version__ except ImportError: __version__ = 'unknown' from pyroute2.conntrack import Conntrack, ConntrackEntry from pyroute2.iproute import ChaoticIPRoute, IPBatch, IPRoute, RawIPRoute from pyroute2.iwutil import IW from pyroute2.netlink.devlink import DevlinkSocket from pyroute2.netlink.diag import DiagSocket from pyroute2.netlink.event.acpi_event import AcpiEventSocket from pyroute2.netlink.event.dquot import DQuotSocket from pyroute2.netlink.exceptions import ( ChaoticException, NetlinkDecodeError, NetlinkDumpInterrupted, NetlinkError, ) from pyroute2.netlink.generic import GenericNetlinkSocket from pyroute2.netlink.generic.l2tp import L2tp from pyroute2.netlink.generic.mptcp import MPTCP from pyroute2.netlink.generic.wireguard import WireGuard from pyroute2.netlink.ipq import IPQSocket from pyroute2.netlink.nfnetlink.nfctsocket import NFCTSocket from pyroute2.netlink.nfnetlink.nftsocket import NFTSocket from pyroute2.netlink.nl80211 import NL80211 from pyroute2.netlink.rtnl.iprsocket import IPRSocket from pyroute2.netlink.taskstats import TaskStats from pyroute2.netlink.uevent import UeventSocket modules = [ AcpiEventSocket, ChaoticException, ChaoticIPRoute, Conntrack, ConntrackEntry, DevlinkSocket, DiagSocket, DQuotSocket, IPBatch, IPQSocket, IPRoute, IPRSocket, IW, GenericNetlinkSocket, L2tp, MPTCP, NetlinkError, NetlinkDecodeError, NetlinkDumpInterrupted, NFCTSocket, NFTSocket, NL80211, RawIPRoute, TaskStats, UeventSocket, WireGuard, ] __all__ = [] __all__.extend(modules)	1.73 KB	2025-05-25 12:26
📁 ndb	-	2025-05-25 12:26
📁 netlink	-	2025-05-25 12:26
📁 netns	-	2025-05-25 12:26
📁 nftables	-	2025-05-25 12:26
📁 nslink	-	2025-05-25 12:26
📁 plan9	-	2025-05-25 12:26
📄 process.py import builtins import gc import json import logging import multiprocessing import os import select import signal import socket import struct from collections import namedtuple from typing import Any, Callable, Optional, Union from pyroute2 import config from pyroute2.common import USE_DEFAULT_TIMEOUT from pyroute2.netlink import exceptions as pyroute2_exceptions log = logging.getLogger(__name__) ChildProcessReturnValue = namedtuple( 'ChildProcessReturnValue', ('payload', 'fds') ) ChildFuncReturnType = Union[None, ChildProcessReturnValue, bytearray, bytes] def wrapper( ctrl: socket.socket, func: Callable[..., ChildFuncReturnType], argv: list[Any], ) -> None: '''Child function wrapper. This code will be executed in the child process after running fork(). The internal data structures might be damaged, so the function `func` must be as simple as possible, and use only local variables of simple types. The garbage collector should be disabled as well to minimize possible deadlocks. If process doesn't response in time, it will get killed. ''' gc.disable() payload: bytes = b'' ret_data: bytes = b'' fds: list[int] = [] sockets: list[socket.socket] = [] try: ret = func(argv) if isinstance(ret, bytes): ret_data = ret if isinstance(ret, ChildProcessReturnValue): ret_data, sockets = ret if isinstance(ret_data, bytearray): ret_data = bytes(ret_data) if not isinstance(ret_data, bytes): raise TypeError('return values not supported') payload = struct.pack('B', 2) + ret_data if sockets: fds = [x.fileno() for x in sockets] except Exception as e: payload = struct.pack('B', 1) + json.dumps( {'exception': e.__class__.__name__, 'options': e.args} ).encode('utf-8') fds = [] finally: socket.send_fds(ctrl, [payload], fds, len(fds)) class ChildProcess: ctrl_r: socket.socket ctrl_w: socket.socket def __init__( self, target: Callable[..., ChildFuncReturnType], args: list[Any] ): self._mode: str = config.child_process_mode self._target: Callable[..., ChildFuncReturnType] = target self._args: list[Any] = args self._proc: Optional[multiprocessing.Process] = None self._running: bool = False self._exitcode: Optional[int] = None self._pid: Optional[int] = None def close(self): self.stop() def __enter__(self): self.run() return self def __exit__(self, _): self.close() @property def mode(self): return self._mode @property def pid(self): return self._pid def communicate( self, timeout: int = USE_DEFAULT_TIMEOUT ) -> tuple[bytes, list[int]]: '''Communicate with the child process. Raises: * struct.error -- error unpacking response from the child process * OSError -- OS level error communicating with the child process * TimeoutError -- the child process is alive, but doesn't response * RuntimeError -- the child process is dead or not started * TypeError -- error loading propagated exception ''' if not self._running: raise RuntimeError('child process not started yet') if timeout == USE_DEFAULT_TIMEOUT: timeout = config.default_communicate_timeout rl, _, _ = select.select([self.ctrl_r], [], [], timeout) if not len(rl): # no data received within timeout # 1. the child process is killed # 2. the child process is stuck # # So first check the process, if it is killed, raise # RuntimeError, otherwise raise TimeoutError. # c_pid, w_status = os.waitpid(self.pid, os.WNOHANG) if c_pid != 0: # process is dead self._exitcode = os.waitstatus_to_exitcode(w_status) raise RuntimeError( f'child process is dead, status: {self.exitcode}' ) # process is alive, but stuck, kill it self.stop(kill=True, reason='no response from the child') raise TimeoutError(f'no response from the child pid {self.pid}') ret_data = b'' # raises OSError (raw_data, fds, _, _) = socket.recv_fds(self.ctrl_r, 1024, 1) # get the return type # raises struct.error (ret_type,) = struct.unpack('B', raw_data[:1]) raw_data = raw_data[1:] if ret_type == 1: # exception payload = json.loads(raw_data.decode('utf-8')) if not set(payload.keys()) == set(('exception', 'options')): raise TypeError('error loading child exception') if payload.get('exception') is not None: error_class = getattr(builtins, payload['exception'], None) if error_class is None: error_class = getattr( pyroute2_exceptions, payload['exception'], None ) if error_class is None: error_class = Exception if not issubclass(error_class, Exception): raise TypeError('error loading child error') raise error_class(*payload['options']) elif ret_type == 2: # raw_data ret_data = raw_data return ret_data, fds def get_data(self, timeout: int = USE_DEFAULT_TIMEOUT) -> bytes: return self.communicate(timeout)[0] def get_fds(self, timeout: int = USE_DEFAULT_TIMEOUT) -> list[int]: return self.communicate(timeout)[1] @property def proc(self) -> multiprocessing.Process: if self._proc is None: raise RuntimeError('not started') return self._proc @proc.setter def proc(self, value: Optional[multiprocessing.Process]) -> None: self._proc = value def _unsupported(self) -> Exception: return TypeError('unsupported mode') def run(self) -> None: if self._running: return self._running = True self.ctrl_r, self.ctrl_w = socket.socketpair( socket.AF_UNIX, socket.SOCK_DGRAM ) if self.mode == 'fork': self._pid = os.fork() if self._pid == 0: wrapper(self.ctrl_w, self._target, self._args) os._exit(0) elif self.mode == 'mp': self.proc = multiprocessing.Process( target=wrapper, args=[self.ctrl_w, self._target, self._args] ) self.proc.start() self._pid = self.proc.pid else: raise self._unsupported() def stop(self, kill: bool = False, reason: Optional[str] = None) -> None: if not self._running: return self._running = False self.ctrl_r.close() self.ctrl_w.close() if config.force_gc: gc.collect() if self.mode == 'fork': try: if kill: os.kill(self.pid, signal.SIGKILL) else: os.kill(self.pid, signal.SIGTERM) _, status = os.waitpid(self.pid, 0) self._exitcode = os.waitstatus_to_exitcode(status) except ProcessLookupError: # the child process has already exited pass elif self.mode == 'mp': if kill: self.proc.kill() else: self.proc.terminate() self.proc.join() else: raise self._unsupported() if reason is not None: log.warning(reason) @property def exitcode(self) -> Optional[int]: if self.mode == 'fork': return self._exitcode elif self.mode == 'mp': return self.proc.exitcode else: raise self._unsupported()	7.92 KB	2025-05-25 12:26
📁 protocols	-	2025-05-25 12:26
📁 requests	-	2025-05-25 12:26
📄 statsd.py import ctypes import os import socket from typing import Literal, Optional, Union from pyroute2 import netns from pyroute2.netlink.coredata import CoreConfig, CoreSocketSpec metric_type = Literal['c', 'g', 'ms'] class StatsDClientSocket(socket.socket): '''StatsD client.''' def __init__( self, address: Optional[tuple[str, int]] = None, use_socket: Optional[socket.socket] = None, flags: int = os.O_CREAT, libc: Optional[ctypes.CDLL] = None, ): self.spec = CoreSocketSpec( CoreConfig( netns=None, address=address, use_socket=use_socket is not None ) ) self.status = self.spec.status self.buffer: str = '' if use_socket is not None: fd = use_socket.fileno() else: prime = netns.create_socket( self.spec['netns'], socket.AF_INET, socket.SOCK_DGRAM ) fd = os.dup(prime.fileno()) prime.close() super().__init__(fileno=fd) def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): self.close() def commit(self): self.sendto(self.buffer.encode(), self.spec['address']) self.buffer = '' def put( self, name: str, value: Union[int, str], kind: metric_type ) -> None: self.buffer += f'{name}:{value}\|{kind}\n' def incr(self, name: str, value: int = 1) -> None: self.put(name, value, 'c') self.commit() def gauge(self, name: str, value: int) -> None: self.put(name, value, 'g') self.commit() def timing(self, name: str, value: int) -> None: self.put(name, value, 'ms') self.commit()	1.72 KB	2025-05-25 12:26
📄 wiset.py ''' High level ipset support. When :doc:`ipset` is providing a direct netlink socket with low level functions, a :class:`WiSet` object is built to map ipset objects from kernel. It helps to add/remove entries, list content, etc. For example, adding an entry with :class:`pyroute2.ipset.IPSet` object implies to set a various number of parameters: .. doctest:: :skipif: True >>> ipset = IPSet() >>> ipset.add("foo", "1.2.3.4/24", etype="net") >>> ipset.close() When they are discovered by a :class:`WiSet`: .. doctest:: :skipif: True >>> wiset = load_ipset("foo") >>> wiset.add("1.2.3.4/24") Listing entries is also easier using :class:`WiSet`, since it parses for you netlink messages: .. doctest:: :skipif: True >>> wiset.content {'1.2.3.0/24': IPStats(packets=None, bytes=None, comment=None, timeout=None, skbmark=None, physdev=False)} ''' import errno import uuid from collections import namedtuple from inspect import getcallargs from socket import AF_INET from pyroute2.common import basestring from pyroute2.ipset import IPSet from pyroute2.netlink.exceptions import IPSetError from pyroute2.netlink.nfnetlink.ipset import ( IPSET_FLAG_IFACE_WILDCARD, IPSET_FLAG_PHYSDEV, IPSET_FLAG_WITH_COMMENT, IPSET_FLAG_WITH_COUNTERS, IPSET_FLAG_WITH_SKBINFO, ) from pyroute2.netlink.nfnetlink.nfctsocket import IP_PROTOCOLS # Debug variable to detect netlink socket leaks COUNT = {"count": 0} def need_ipset_socket(fun): """Decorator to create netlink socket if needed. In many of our helpers, we need to open a netlink socket. This can be expensive for someone using many times the functions: instead to have only one socket and use several requests, we will open it again and again. This helper allow our functions to be flexible: the caller can pass an optional socket, or do nothing. In this last case, this decorator will open a socket for the caller (and close it after call) It also help to mix helpers. One helper can call another one: the socket will be opened only once. We just have to pass the ipset variable. Note that all functions using this helper must use ipset as variable name for the socket. """ def wrap(args, kwargs): callargs = getcallargs(fun, args, kwargs) if callargs["sock"] is None: # This variable is used only to debug leak in tests COUNT['count'] += 1 with IPSet() as sock: callargs["sock"] = sock # We must pop kwargs here, else the function will receive # a dict of dict if "kwargs" in callargs: callargs.update(callargs.pop("kwargs")) return fun(callargs) # pylint:disable=star-args return fun(args, kwargs) return wrap class IPStats( namedtuple( "IPStats", [ "packets", "bytes", "comment", "timeout", "skbmark", "physdev", "wildcard", ], ) ): __slots__ = () def __new__( cls, packets, bytes, comment, timeout, skbmark, physdev=False, wildcard=False, ): return super(IPStats, cls).__new__( cls, packets, bytes, comment, timeout, skbmark, physdev=physdev, wildcard=wildcard, ) # pylint: disable=too-many-instance-attributes class WiSet(object): """Main high level ipset manipulation class. Every high level ipset operation should be possible with this class, you probably don't need other helpers of this module, except tools to load data from kernel (:func:`load_all_ipsets` and :func:`load_ipset`) For example, you can create and an entry in a ipset just with: .. doctest:: :skipif: True >>> with WiSet(name="mysuperipset") as myset: >>> myset.create() # add the ipset in the kernel >>> myset.add("198.51.100.1") # add one IP to the set Netlink sockets are opened by __enter__ and __exit__ function, so you don't have to manage it manually if you use the "with" keyword. If you want to manage it manually (for example for long operation in a daemon), you can do the following: .. doctest:: :skipif: True >>> myset = WiSet(name="mysuperipset") >>> myset.open_netlink() >>> # do stuff >>> myset.close_netlink() You can also don't initiate at all any netlink socket, this code will work: .. doctest:: :skipif: True >>> myset = WiSet(name="mysuperipset") >>> myset.create() >>> myset.destroy() But do it very carefully. In that case, a netlink socket will be opened in background for any operation. No socket will be leaked, but that can consume resources. You can also instantiate WiSet objects with :func:`load_all_ipsets` and :func:`load_ipset`: .. doctest:: :skipif: True >>> all_sets_dict = load_all_ipsets() >>> one_set = load_ipset(name="myset") Have a look on content variable if you need list of entries in the Set. """ # pylint: disable=too-many-arguments def __init__( self, name=None, attr_type='hash:ip', family=AF_INET, sock=None, timeout=None, counters=False, comment=False, hashsize=None, revision=None, skbinfo=False, ): self.name = name self.hashsize = hashsize self._attr_type = None self.entry_type = None self.attr_type = attr_type self.family = family self._content = None self.sock = sock self.timeout = timeout self.counters = counters self.comment = comment self.revision = revision self.index = None self.skbinfo = skbinfo def open_netlink(self): """ Open manually a netlink socket. You can use "with WiSet()" statement instead. """ if self.sock is None: self.sock = IPSet() def close_netlink(self): """Clone any opened netlink socket""" if self.sock is not None: self.sock.close() self.sock = None @property def attr_type(self): return self._attr_type @attr_type.setter def attr_type(self, value): self._attr_type = value self.entry_type = value.split(":", 1)[1] def __enter__(self): self.open_netlink() return self def __exit__(self, exc_type, exc_value, traceback): self.close_netlink() @classmethod def from_netlink(cls, ndmsg, content=False): """Create a ipset objects based on a parsed netlink message :param ndmsg: the netlink message to parse :param content: should we fill (and parse) entries info (can be slow on very large set) :type content: bool """ self = cls() self.attr_type = ndmsg.get_attr("IPSET_ATTR_TYPENAME") self.name = ndmsg.get_attr("IPSET_ATTR_SETNAME") self.hashsize = ndmsg.get_attr("IPSET_ATTR_HASHSIZE") self.family = ndmsg.get_attr("IPSET_ATTR_FAMILY") self.revision = ndmsg.get_attr("IPSET_ATTR_REVISION") self.index = ndmsg.get_attr("IPSET_ATTR_INDEX") data = ndmsg.get_attr("IPSET_ATTR_DATA") self.timeout = data.get_attr("IPSET_ATTR_TIMEOUT") flags = data.get_attr("IPSET_ATTR_CADT_FLAGS") if flags is not None: self.counters = bool(flags & IPSET_FLAG_WITH_COUNTERS) self.comment = bool(flags & IPSET_FLAG_WITH_COMMENT) self.skbinfo = bool(flags & IPSET_FLAG_WITH_SKBINFO) if content: self.update_dict_content(ndmsg) return self def update_dict_content(self, ndmsg): """Update a dictionary statistics with values sent in netlink message :param ndmsg: the netlink message :type ndmsg: netlink message """ family = "IPSET_ATTR_IPADDR_IPV4" ip_attr = "IPSET_ATTR_IP_FROM" if self._content is None: self._content = {} timeout = None entries = ndmsg.get_attr("IPSET_ATTR_ADT").get_attrs("IPSET_ATTR_DATA") for entry in entries: key = "" for parse_type in self.entry_type.split(","): if parse_type == "ip": ip = entry.get_attr(ip_attr).get_attr(family) key += ip elif parse_type == "net": ip = entry.get_attr(ip_attr).get_attr(family) key += ip cidr = entry.get_attr("IPSET_ATTR_CIDR") if cidr is not None: key += "/{0}".format(cidr) elif parse_type == "iface": key += entry.get_attr("IPSET_ATTR_IFACE") elif parse_type == "set": key += entry.get_attr("IPSET_ATTR_NAME") elif parse_type == "mark": key += str(hex(entry.get_attr("IPSET_ATTR_MARK"))) elif parse_type == "port": proto = entry.get_attr('IPSET_ATTR_PROTO') if proto is not None: proto = IP_PROTOCOLS.get(proto, str(proto)).lower() key += '{proto}:'.format(proto=proto) key += str(entry.get_attr("IPSET_ATTR_PORT_FROM")) elif parse_type == "mac": key += entry.get_attr("IPSET_ATTR_ETHER") key += "," key = key.strip(",") if self.timeout is not None: timeout = entry.get_attr("IPSET_ATTR_TIMEOUT") skbmark = entry.get_attr("IPSET_ATTR_SKBMARK") if skbmark is not None: # Convert integer to hex for mark/mask # Only display mask if != 0xffffffff if skbmark[1] != (232 - 1): skbmark = "/".join([str(hex(mark)) for mark in skbmark]) else: skbmark = str(hex(skbmark[0])) entry_flag_parsed = {"physdev": False} flags = entry.get_attr("IPSET_ATTR_CADT_FLAGS") if flags is not None: entry_flag_parsed["physdev"] = bool(flags & IPSET_FLAG_PHYSDEV) entry_flag_parsed["wildcard"] = bool( flags & IPSET_FLAG_IFACE_WILDCARD ) value = IPStats( packets=entry.get_attr("IPSET_ATTR_PACKETS"), bytes=entry.get_attr("IPSET_ATTR_BYTES"), comment=entry.get_attr("IPSET_ATTR_COMMENT"), skbmark=skbmark, timeout=timeout, entry_flag_parsed ) self._content[key] = value def create(self, kwargs): """Insert this Set in the kernel Many options are set with python object attributes (like comments, counters, etc). For non-supported type, kwargs are provided. See :doc:`ipset` documentation for more information. """ create_ipset( self.name, stype=self.attr_type, family=self.family, sock=self.sock, timeout=self.timeout, comment=self.comment, counters=self.counters, hashsize=self.hashsize, skbinfo=self.skbinfo, kwargs ) def destroy(self): """Destroy this ipset in the kernel list. It does not delete this python object (any content or other stored values are keep in memory). This function will fail if the ipset is still referenced (by example in iptables rules), you have been warned. """ destroy_ipset(self.name, sock=self.sock) def add(self, entry, kwargs): """Add an entry in this ipset. If counters are enabled on the set, reset by default the value when we add the element. Without this reset, kernel sometimes store old values and can add very strange behavior on counters. """ if isinstance(entry, dict): kwargs.update(entry) entry = kwargs.pop("entry") if self.counters: kwargs["packets"] = kwargs.pop("packets", 0) kwargs["bytes"] = kwargs.pop("bytes", 0) skbmark = kwargs.get("skbmark") if isinstance(skbmark, basestring): skbmark = skbmark.split('/') mark = int(skbmark[0], 16) try: mask = int(skbmark[1], 16) except IndexError: mask = int("0xffffffff", 16) kwargs["skbmark"] = (mark, mask) add_ipset_entry( self.name, entry, etype=self.entry_type, sock=self.sock, kwargs ) def delete(self, entry, kwargs): """Delete/remove an entry in this ipset""" delete_ipset_entry( self.name, entry, etype=self.entry_type, sock=self.sock, kwargs ) def test(self, entry, kwargs): """Test if an entry is in this ipset""" return test_ipset_entry( self.name, entry, etype=self.entry_type, sock=self.sock, kwargs ) def test_list(self, entries, kwargs): """Test if a list of a set of entries is in this ipset Return a set of entries found in the IPSet """ return test_ipset_entries( self.name, entries, etype=self.entry_type, sock=self.sock, kwargs ) def update_content(self): """Update the content dictionary with values from kernel""" self._content = {} update_wiset_content(self, sock=self.sock) def flush(self): """Flush entries of the ipset""" flush_ipset(self.name, sock=self.sock) @property def content(self): """Dictionary of entries in the set. Keys are IP addresses (as string), values are IPStats tuples. """ if self._content is None: self.update_content() return self._content def insert_list(self, entries): """Just a small helper to reduce the number of loops in main code.""" for entry in entries: self.add(entry) def replace_entries(self, new_list): """Replace the content of an ipset with a new list of entries. This operation is like a flush() and adding all entries one by one. But this call is atomic: it creates a temporary ipset and swap the content. :param new_list: list of entries to add :type new_list: list or :py:class:`set` of basestring or of keyword arguments dict """ temp_name = str(uuid.uuid4())[0:8] # Get a copy of ourself temp = load_ipset(self.name, sock=self.sock) temp.name = temp_name temp.sock = self.sock temp.create() temp.insert_list(new_list) swap_ipsets(self.name, temp_name, sock=self.sock) temp.destroy() @need_ipset_socket def create_ipset( name, stype=None, family=AF_INET, exclusive=False, sock=None, kwargs ): """Create an ipset.""" sock.create( name, stype=stype, family=family, exclusive=exclusive, kwargs ) @need_ipset_socket def load_all_ipsets(content=False, sock=None, inherit_sock=False, prefix=None): """List all ipset as WiSet objects. Get full ipset data from kernel and parse it in WiSet objects. Result is a dictionary with ipset names as keys, and WiSet objects as values. :param content: parse the list of entries and fill it in WiSet content dictionary :type content: bool :param inherit_sock: use the netlink sock passed in ipset arg to fill WiSets sock :type inherit_sock: bool :param prefix: filter out all ipset with a name not beginning by this prefix :type prefix: str or None """ res = {} for myset in sock.list(): # on large sets, we can receive data in several messages name = myset.get_attr("IPSET_ATTR_SETNAME") if prefix is not None and not name.startswith(prefix): continue if name not in res: wiset = WiSet.from_netlink(myset, content=content) if inherit_sock: wiset.sock = sock res[wiset.name] = wiset elif content: res[wiset.name].update_dict_content(myset) return res @need_ipset_socket def load_ipset(name, content=False, sock=None, inherit_sock=False): """Get one ipset as WiSet object Helper to get current WiSet object. More efficient that :func:`load_all_ipsets` since the kernel does the filtering itself. Return None if the ipset does not exist :param name: name of the ipset :type name: str :param content: parse or not content and statistics on entries :type content: bool :param inherit_sock: use the netlink sock passed in ipset arg to fill WiSet sock :type inherit_sock: bool """ res = None try: messages = sock.list(name=name) except IPSetError as e: if e.code == errno.ENOENT: return res raise for msg in messages: if res is None: res = WiSet.from_netlink(msg, content=content) if inherit_sock: res.sock = sock elif content: res.update_dict_content(msg) return res @need_ipset_socket def update_wiset_content(wiset, sock=None): """Update content/statistics of a wiset. You should never call yourself this function. It is only a helper to use the :func:`need_ipset_socket` decorator out of WiSet object. """ for msg in sock.list(name=wiset.name): wiset.update_dict_content(msg) @need_ipset_socket def destroy_ipset(name, sock=None): """Remove an ipset in the kernel.""" sock.destroy(name) @need_ipset_socket def add_ipset_entry(name, entry, sock=None, kwargs): """Add an entry""" sock.add(name, entry, kwargs) @need_ipset_socket def delete_ipset_entry(name, entry, sock=None, kwargs): """Remove one entry""" sock.delete(name, entry, kwargs) @need_ipset_socket def test_ipset_exist(name, sock=None): """Test if the given ipset exist""" try: sock.headers(name) return True except IPSetError as e: if e.code == errno.ENOENT: return False raise @need_ipset_socket def test_ipset_entry(name, entry, sock=None, kwargs): """Test if an entry is in one ipset""" return sock.test(name, entry, kwargs) @need_ipset_socket def test_ipset_entries(name, entries, sock=None, kwargs): """Test a list (or a set) of entries.""" res = set() for entry in entries: if sock.test(name, entry, kwargs): res.add(entry) return res @need_ipset_socket def flush_ipset(name, sock=None): """Flush all ipset content""" sock.flush(name) @need_ipset_socket def swap_ipsets(name_a, name_b, sock=None): """Swap the content of ipset a and b. ipsets must have compatible content. """ sock.swap(name_a, name_b) def get_ipset_socket(kwargs): """Get a socket that one can pass to several WiSet objects""" return IPSet(*kwargs)	19.16 KB	2025-05-25 12:26

Edit File:

Terminal