Wala shell

Advanced PHP File Manager

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

Name	Size	Last Modified
📁 ..	-	2025-05-25 12:26
📄 __init__.py ''' Netlink ------- basics ====== General netlink packet structure:: nlmsg packet: header data Generic netlink message header:: nlmsg header: uint32 length uint16 type uint16 flags uint32 sequence number uint32 pid The `length` field is the length of all the packet, including data and header. The `type` field is used to distinguish different message types, commands etc. Please note, that there is no explicit protocol field -- you choose a netlink protocol, when you create a socket. The `sequence number` is very important. Netlink is an asynchronous protocol -- it means, that the packet order doesn't matter and is not guaranteed. But responses to a request are always marked with the same sequence number, so you can treat it as a cookie. Please keep in mind, that a netlink request can initiate a cascade of events, and netlink messages from these events can carry sequence number == 0. E.g., it is so when you remove a primary IP addr from an interface, when `promote_secondaries` sysctl is set. Beside of incapsulated headers and other protocol-specific data, netlink messages can carry NLA (netlink attributes). NLA structure is as follows:: NLA header: uint16 length uint16 type NLA data: data-specific struct # optional: NLA NLA ... So, NLA structures can be nested, forming a tree. Complete structure of a netlink packet:: nlmsg header: uint32 length uint16 type uint16 flags uint32 sequence number uint32 pid [ optional protocol-specific data ] [ optional NLA tree ] More information about netlink protocol you can find in the man pages. pyroute2 and netlink ==================== packets ~~~~~~~ To simplify the development, pyroute2 provides an easy way to describe packet structure. As an example, you can take the ifaddrmsg description -- `pyroute2/netlink/rtnl/ifaddrmsg.py`. To describe a packet, you need to inherit from `nlmsg` class:: from pyroute2.netlink import nlmsg class foo_msg(nlmsg): fields = ( ... ) nla_map = ( ... ) NLA are described in the same way, but the parent class should be `nla`, instead of `nlmsg`. And yes, it is important to use the proper parent class -- it affects the header structure. fields attribute ~~~~~~~~~~~~~~~~ The `fields` attribute describes the structure of the protocol-specific data. It is a tuple of tuples, where each member contains a field name and its data format. Field data format should be specified as for Python `struct` module. E.g., ifaddrmsg structure:: struct ifaddrmsg { __u8 ifa_family; __u8 ifa_prefixlen; __u8 ifa_flags; __u8 ifa_scope; __u32 ifa_index; }; should be described as follows:: class ifaddrmsg(nlmsg): fields = (('family', 'B'), ('prefixlen', 'B'), ('flags', 'B'), ('scope', 'B'), ('index', 'I')) Format strings are passed directly to the `struct` module, so you can use all the notations like `>I`, `16s` etc. All fields are parsed from the stream separately, so if you want to explicitly fix alignemt, as if it were C struct, use the `pack` attribute:: class tstats(nla): pack = 'struct' fields = (('version', 'H'), ('ac_exitcode', 'I'), ('ac_flag', 'B'), ...) Explicit padding bytes also can be used, when struct packing doesn't work well:: class ipq_mode_msg(nlmsg): pack = 'struct' fields = (('value', 'B'), ('__pad', '7x'), ('range', 'I'), ('__pad', '12x')) nla_map attribute ~~~~~~~~~~~~~~~~~ The `nla_map` attribute is a tuple of NLA descriptions. Each description is also a tuple in two different forms: either two fields, name and format, or three fields: type, name and format. Please notice, that the format field is a string name of corresponding NLA class:: class ifaddrmsg(nlmsg): ... nla_map = (('IFA_UNSPEC', 'hex'), ('IFA_ADDRESS', 'ipaddr'), ('IFA_LOCAL', 'ipaddr'), ...) This code will create mapping, where IFA_ADDRESS NLA will be of type 1 and IFA_LOCAL -- of type 2, etc. Both NLA will be decoded as IP addresses (class `ipaddr`). IFA_UNSPEC will be of type 0, and if it will be in the NLA tree, it will be just dumped in hex. NLA class names are should be specified as strings, since they are resolved in runtime. There are several pre-defined NLA types, that you will get with `nla` class: - `none` -- ignore this NLA - `flag` -- boolean flag NLA (no payload; NLA exists = True) - `uint8`, `uint16`, `uint32`, `uint64` -- unsigned int - `be8`, `be16`, `be32`, `be64` -- big-endian unsigned int - `ipaddr` -- IP address, IPv4 or IPv6 - `ip4addr` -- only IPv4 address type - `ip6addr` -- only IPv6 address type - `target` -- a univeral target (IPv4, IPv6, MPLS) - `l2addr` -- MAC address - `lladdr` -- link layer address (MAC, IPv4, IPv6) - `hex` -- hex dump as a string -- useful for debugging - `cdata` -- a binary data - `string` -- UTF-8 string - `asciiz` -- zero-terminated ASCII string, no decoding - `array` -- array of simple types (uint8, uint16 etc.) Please refer to `pyroute2/netlink/__init__.py` for details. You can also make your own NLA descriptions:: class ifaddrmsg(nlmsg): ... nla_map = (... ('IFA_CACHEINFO', 'cacheinfo'), ...) class cacheinfo(nla): fields = (('ifa_preferred', 'I'), ('ifa_valid', 'I'), ('cstamp', 'I'), ('tstamp', 'I')) Custom NLA descriptions should be defined in the same class, where they are used. explicit NLA type ids ~~~~~~~~~~~~~~~~~~~~~ Also, it is possible to use not autogenerated type numbers, as for ifaddrmsg, but specify them explicitly:: class iw_event(nla): ... nla_map = ((0x8B00, 'SIOCSIWCOMMIT', 'hex'), (0x8B01, 'SIOCGIWNAME', 'hex'), (0x8B02, 'SIOCSIWNWID', 'hex'), (0x8B03, 'SIOCGIWNWID', 'hex'), ...) Here you can see custom NLA type numbers -- 0x8B00, 0x8B01 etc. It is not permitted to mix these two forms in one class: you should use ether autogenerated type numbers (two fields tuples), or explicit numbers (three fields typles). nla map adapters ~~~~~~~~~~~~~~~~ If the default declarative NLA map is not flexible enough, one can use a custom map adapter. In order to do so, one should define at least one function to return `pyroute2.netlink.NlaSpec()`, and one optional function to tell the parser if the attribute is supported. The simplest definition only to decode packets: .. code-block:: python from pyroute2.netlink import NlaMapAdapter, NlaSpec, nlmsg def my_flexible_nla_spec(key): return NlaSpec(nlmsg_atoms.hex, key, f'NLA_CLASS_{key}') class my_msg(nlmsg): nla_map = NlaMapAdapter(my_flexible_nla_spec) # example result [ { 'attrs': [ ('NLA_CLASS_1', '00:00:00:00'), ('NLA_CLASS_5', '00:00:00:00'), ], 'header': { ... }, }, ] In this example the same routine is used both for decoding and encoding workflows, but the workflows are not equal, thus the example will fail on encoding. Still the example may be useful if you don't plan to encode packets of this type. The decoding workflow will pass an integer as the `key` for NLA type, while the encoding workflow passes a string as the `key` for NLA name. To correctly handle both workflows, you can use either the `key` type discrimination, or the explicit declaration syntax: .. code-block:: python # discriminate workflows by the key type def my_flexible_nla_spec(key): if isinstance(key, int): # decoding workflow ... else: # encoding workflow ... class my_msg(nlmsg): nla_map = NlaMapAdapter(my_flexible_nla_spec) .. code-block:: python # declarate separate workflows def my_flexible_nla_spec_encode(key): # receives a string -- nla type name ... def my_flexible_nla_spec_decode(key): # receives an int -- nla type id ... class my_msg(nlmsg): nla_map = { 'decode': NlaMapAdapter(my_flexible_nla_spec_decode), 'encode': NlaMapAdapter(my_flexible_nla_spec_encode), } array types ~~~~~~~~~~~ There are different array-like NLA types in the kernel, and some of them are covered by pyroute2. An array of simple type elements:: # declaration nla_map = (('NLA_TYPE', 'array(uint8)'), ...) # data layout +======+======+---------------------------- \| len \| type \| uint8 \| uint8 \| uint 8 \| ... +======+======+---------------------------- # decoded {'attrs': [['NLA_TYPE', (2, 3, 4, 5, ...)], ...], ...} An array of NLAs:: # declaration nla_map = (('NLA_TYPE', 'type'), ...) # data layout +=======+=======+-----------------------+-----------------------+-- \| len \| type \| len \| type \| payload \| len \| type \| payload \| ... +=======+=======+-----------------------+-----------------------+-- # type* -- in that case the type is OR'ed with NLA_F_NESTED # decoded {'attrs': [['NLA_TYPE', [payload, payload, ...]], ...], ...} parsed netlink message ~~~~~~~~~~~~~~~~~~~~~~ Netlink messages are represented by pyroute2 as dictionaries as follows:: {'header': {'pid': ..., 'length: ..., 'flags': ..., 'error': None, # if you are lucky 'type': ..., 'sequence_number': ...}, # fields attributes 'field_name1': value, ... 'field_nameX': value, # nla tree 'attrs': [['NLA_NAME1', value], ... ['NLA_NAMEX', value], ['NLA_NAMEY', {'field_name1': value, ... 'field_nameX': value, 'attrs': [['NLA_NAME.... ]]}]]} As an example, a message from the wireless subsystem about new scan event:: {'index': 4, 'family': 0, '__align': 0, 'header': {'pid': 0, 'length': 64, 'flags': 0, 'error': None, 'type': 16, 'sequence_number': 0}, 'flags': 69699, 'ifi_type': 1, 'event': 'RTM_NEWLINK', 'change': 0, 'attrs': [['IFLA_IFNAME', 'wlp3s0'], ['IFLA_WIRELESS', {'attrs': [['SIOCGIWSCAN', '00:00:00:00:00:00:00:00:00:00:00:00']]}]]} One important detail is that NLA chain is represented as a list of elements `['NLA_TYPE', value]`, not as a dictionary. The reason is that though in the kernel usually NLA chain is a dictionary, the netlink protocol by itself doesn't require elements of each type to be unique. In a message there may be several NLA of the same type. encoding and decoding algo ~~~~~~~~~~~~~~~~~~~~~~~~~~ The message encoding works as follows: 1. Reserve space for the message header (if there is) 2. Iterate defined `fields`, encoding values with `struct.pack()` 3. Iterate NLA from the `attrs` field, looking up types in `nla_map` 4. Encode the header Since every NLA is also an `nlmsg` object, there is a recursion. The decoding process is a bit simpler: 1. Decode the header 2. Iterate `fields`, decoding values with `struct.unpack()` 3. Iterate NLA until the message ends If the `fields` attribute is an empty list, the step 2 will be skipped. The step 3 will be skipped in the case of the empty `nla_map`. If both attributes are empty lists, only the header will be encoded/decoded. create and send messages ~~~~~~~~~~~~~~~~~~~~~~~~ Using high-level interfaces like `IPRoute` or `IPDB`, you will never need to manually construct and send netlink messages. But in the case you really need it, it is simple as well. Having a description class, like `ifaddrmsg` from above, you need to: - instantiate it - fill the fields - encode the packet - send the encoded data The code:: from pyroute2.netlink import NLM_F_REQUEST from pyroute2.netlink import NLM_F_ACK from pyroute2.netlink import NLM_F_CREATE from pyroute2.netlink import NLM_F_EXCL from pyroute2.iproute import RTM_NEWADDR from pyroute2.netlink.rtnl.ifaddrmsg import ifaddrmsg ## # add an addr to an interface # # create the message msg = ifaddrmsg() # fill the protocol-specific fields msg['index'] = index # index of the interface msg['family'] = AF_INET # address family msg['prefixlen'] = 24 # the address mask msg['scope'] = scope # see /etc/iproute2/rt_scopes # attach NLA -- it MUST be a list / mutable msg['attrs'] = [['IFA_LOCAL', '192.168.0.1'], ['IFA_ADDRESS', '192.162.0.1']] # fill generic netlink fields msg['header']['sequence_number'] = nonce # an unique seq number msg['header']['pid'] = os.getpid() msg['header']['type'] = RTM_NEWADDR msg['header']['flags'] = NLM_F_REQUEST \|\\ NLM_F_ACK \|\\ NLM_F_CREATE \|\\ NLM_F_EXCL # encode the packet msg.encode() # send the buffer nlsock.sendto(msg.data, (0, 0)) Please notice, that NLA list MUST be mutable. ''' import io import logging import socket import struct import sys import threading import traceback import types import weakref from collections import OrderedDict from socket import AF_INET, AF_INET6, AF_UNSPEC, inet_ntop, inet_pton from pyroute2.common import AF_MPLS, basestring, hexdump from pyroute2.netlink.exceptions import ( NetlinkDecodeError, NetlinkError, NetlinkNLADecodeError, ) log = logging.getLogger(__name__) # make pep8 happy _ne = NetlinkError # reexport for compatibility _de = NetlinkDecodeError # class NotInitialized(Exception): pass ## # That's a hack for the code linter, which works under # Python3, see unicode reference in the code below unicode = str NLMSG_MIN_TYPE = 0x10 GENL_NAMSIZ = 16 # length of family name GENL_MIN_ID = NLMSG_MIN_TYPE GENL_MAX_ID = 1023 GENL_ADMIN_PERM = 0x01 GENL_CMD_CAP_DO = 0x02 GENL_CMD_CAP_DUMP = 0x04 GENL_CMD_CAP_HASPOL = 0x08 # # List of reserved static generic netlink identifiers: # GENL_ID_GENERATE = 0 GENL_ID_CTRL = NLMSG_MIN_TYPE # # Controller # CTRL_CMD_UNSPEC = 0x0 CTRL_CMD_NEWFAMILY = 0x1 CTRL_CMD_DELFAMILY = 0x2 CTRL_CMD_GETFAMILY = 0x3 CTRL_CMD_NEWOPS = 0x4 CTRL_CMD_DELOPS = 0x5 CTRL_CMD_GETOPS = 0x6 CTRL_CMD_NEWMCAST_GRP = 0x7 CTRL_CMD_DELMCAST_GRP = 0x8 CTRL_CMD_GETMCAST_GRP = 0x9 # unused CTRL_CMD_GETPOLICY = 0xA CTRL_ATTR_UNSPEC = 0x0 CTRL_ATTR_FAMILY_ID = 0x1 CTRL_ATTR_FAMILY_NAME = 0x2 CTRL_ATTR_VERSION = 0x3 CTRL_ATTR_HDRSIZE = 0x4 CTRL_ATTR_MAXATTR = 0x5 CTRL_ATTR_OPS = 0x6 CTRL_ATTR_MCAST_GROUPS = 0x7 CTRL_ATTR_POLICY = 0x8 CTRL_ATTR_OP_POLICY = 0x9 CTRL_ATTR_OP = 0xA CTRL_ATTR_OP_UNSPEC = 0x0 CTRL_ATTR_OP_ID = 0x1 CTRL_ATTR_OP_FLAGS = 0x2 CTRL_ATTR_MCAST_GRP_UNSPEC = 0x0 CTRL_ATTR_MCAST_GRP_NAME = 0x1 CTRL_ATTR_MCAST_GRP_ID = 0x2 NL_ATTR_TYPE_INVALID = 0 NL_ATTR_TYPE_FLAG = 1 NL_ATTR_TYPE_U8 = 2 NL_ATTR_TYPE_U16 = 3 NL_ATTR_TYPE_U32 = 4 NL_ATTR_TYPE_U64 = 5 NL_ATTR_TYPE_S8 = 6 NL_ATTR_TYPE_S16 = 7 NL_ATTR_TYPE_S32 = 8 NL_ATTR_TYPE_S64 = 9 NL_ATTR_TYPE_BINARY = 10 NL_ATTR_TYPE_STRING = 11 NL_ATTR_TYPE_NUL_STRING = 12 NL_ATTR_TYPE_NESTED = 13 NL_ATTR_TYPE_NESTED_ARRAY = 14 NL_ATTR_TYPE_BITFIELD32 = 15 NL_POLICY_TYPE_ATTR_UNSPEC = 0 NL_POLICY_TYPE_ATTR_TYPE = 1 NL_POLICY_TYPE_ATTR_MIN_VALUE_S = 2 NL_POLICY_TYPE_ATTR_MAX_VALUE_S = 3 NL_POLICY_TYPE_ATTR_MIN_VALUE_U = 4 NL_POLICY_TYPE_ATTR_MAX_VALUE_U = 5 NL_POLICY_TYPE_ATTR_MIN_LENGTH = 6 NL_POLICY_TYPE_ATTR_MAX_LENGTH = 7 NL_POLICY_TYPE_ATTR_POLICY_IDX = 8 NL_POLICY_TYPE_ATTR_POLICY_MAXTYPE = 9 NL_POLICY_TYPE_ATTR_BITFIELD32_MASK = 10 NL_POLICY_TYPE_ATTR_PAD = 11 NL_POLICY_TYPE_ATTR_MASK = 12 # Different Netlink families # NETLINK_ROUTE = 0 # Routing/device hook NETLINK_UNUSED = 1 # Unused number NETLINK_USERSOCK = 2 # Reserved for user mode socket protocols NETLINK_FIREWALL = 3 # Firewalling hook NETLINK_SOCK_DIAG = 4 # INET socket monitoring NETLINK_NFLOG = 5 # netfilter/iptables ULOG NETLINK_XFRM = 6 # ipsec NETLINK_SELINUX = 7 # SELinux event notifications NETLINK_ISCSI = 8 # Open-iSCSI NETLINK_AUDIT = 9 # auditing NETLINK_FIB_LOOKUP = 10 NETLINK_CONNECTOR = 11 NETLINK_NETFILTER = 12 # netfilter subsystem NETLINK_IP6_FW = 13 NETLINK_DNRTMSG = 14 # DECnet routing messages NETLINK_KOBJECT_UEVENT = 15 # Kernel messages to userspace NETLINK_GENERIC = 16 # leave room for NETLINK_DM (DM Events) NETLINK_SCSITRANSPORT = 18 # SCSI Transports # NLA flags NLA_F_NESTED = 1 << 15 NLA_F_NET_BYTEORDER = 1 << 14 # Netlink message flags values (nlmsghdr.flags) # NLM_F_REQUEST = 1 # It is request message. NLM_F_MULTI = 2 # Multipart message, terminated by NLMSG_DONE NLM_F_ACK = 4 # Reply with ack, with zero or error code NLM_F_ECHO = 8 # Echo this request NLM_F_DUMP_INTR = 0x10 # Dump was inconsistent due to sequence change NLM_F_DUMP_FILTERED = 0x20 # Dump was filtered as requested # Modifiers to GET request NLM_F_ROOT = 0x100 # specify tree root NLM_F_MATCH = 0x200 # return all matching NLM_F_ATOMIC = 0x400 # atomic GET NLM_F_DUMP = NLM_F_ROOT \| NLM_F_MATCH # Modifiers to NEW request NLM_F_REPLACE = 0x100 # Override existing NLM_F_EXCL = 0x200 # Do not touch, if it exists NLM_F_CREATE = 0x400 # Create, if it does not exist NLM_F_APPEND = 0x800 # Add to end of list NLM_F_CAPPED = 0x100 NLM_F_ACK_TLVS = 0x200 NLMSG_NOOP = 0x1 # Nothing NLMSG_ERROR = 0x2 # Error NLMSG_DONE = 0x3 # End of a dump NLMSG_OVERRUN = 0x4 # Data lost NLMSG_CONTROL = 0xE # Custom message type for messaging control NLMSG_TRANSPORT = 0xF # Custom message type for NL as a transport NLMSG_MIN_TYPE = 0x10 # < 0x10: reserved control messages NLMSG_MAX_LEN = 0xFFFF # Max message length mtypes = { 1: 'NLMSG_NOOP', 2: 'NLMSG_ERROR', 3: 'NLMSG_DONE', 4: 'NLMSG_OVERRUN', } IPRCMD_NOOP = 0 IPRCMD_STOP = 1 IPRCMD_ACK = 2 IPRCMD_ERR = 3 IPRCMD_REGISTER = 4 IPRCMD_RELOAD = 5 IPRCMD_ROUTE = 6 IPRCMD_CONNECT = 7 IPRCMD_DISCONNECT = 8 IPRCMD_SERVE = 9 IPRCMD_SHUTDOWN = 10 IPRCMD_SUBSCRIBE = 11 IPRCMD_UNSUBSCRIBE = 12 IPRCMD_PROVIDE = 13 IPRCMD_REMOVE = 14 IPRCMD_DISCOVER = 15 IPRCMD_UNREGISTER = 16 SOL_NETLINK = 270 NETLINK_ADD_MEMBERSHIP = 1 NETLINK_DROP_MEMBERSHIP = 2 NETLINK_PKTINFO = 3 NETLINK_BROADCAST_ERROR = 4 NETLINK_NO_ENOBUFS = 5 NETLINK_RX_RING = 6 NETLINK_TX_RING = 7 NETLINK_LISTEN_ALL_NSID = 8 NETLINK_EXT_ACK = 11 NETLINK_GET_STRICT_CHK = 12 clean_cbs = threading.local() # Cached results for some struct operations. # No cache invalidation required. cache_fmt = {} cache_hdr = {} cache_jit = {} class NlaSpec(dict): def __init__( self, nla_class, nla_type, nla_name, nla_flags=0, nla_array=False, init=None, ): self.update( { 'class': nla_class, 'type': nla_type, 'name': nla_name, 'nla_flags': nla_flags, 'nla_array': nla_array, 'init': init, } ) class NlaMapAdapter: def __init__(self, api_get, api_contains=lambda x: True): self.api_get = api_get self.api_contains = api_contains self.types = None def __contains__(self, key): return self.api_contains(key) def __getitem__(self, key): ret = self.api_get(key) if isinstance(ret['class'], str): ret['class'] = getattr(self.types, ret['class']) return ret class SQLSchema: def __init__(self, cls): ret = [] for field in cls.fields: if field[0][0] != '_': ret.append( ( (field[0],), ' '.join( ('BIGINT', cls.sql_constraints.get(field[0], '')) ), ) ) for nla_tuple in cls.nla_map: if isinstance(nla_tuple[0], basestring): nla_name = nla_tuple[0] nla_type = nla_tuple[1] else: nla_name = nla_tuple[1] nla_type = nla_tuple[2] nla_type = getattr(cls, nla_type, None) sql_type = getattr(nla_type, 'sql_type', None) if sql_type: sql_type = ' '.join( (sql_type, cls.sql_constraints.get(nla_name, '')) ) ret.append(((nla_name,), sql_type)) for fname, ftype in cls.sql_extra_fields: if isinstance(fname, basestring): fname = (fname,) ret.append((fname, ftype)) for dcls, prefix in cls.sql_extend: for fname, ftype in dcls.sql_schema(): ret.append(((prefix,) + fname, ftype)) self.spec = ret self.index = [] self.foreign_keys = [] def unique_index(self, index): self.index = index return self def constraint(self, name, spec): idx = 0 for field, tspec in self.spec: if field[0] == name: break idx += 1 else: raise KeyError() self.spec[idx] = (field, f'{tspec} {spec}') return self def foreign_key(self, parent, fields, parent_fields): self.foreign_keys.append( { 'fields': fields, 'parent_fields': parent_fields, 'parent': parent, } ) return self def push(self, spec): f_type = spec[-1] f_name = spec[:-1] self.spec.append((f_name, f_type)) return self def __iter__(self): return iter(self.spec) def as_dict(self): return OrderedDict(self.spec) class nlmsg_base(dict): ''' Netlink base class. You do not need to inherit it directly, unless you're inventing completely new protocol structure. Use nlmsg or nla classes. The class provides several methods, but often one need to customize only `decode()` and `encode()`. ''' fields = () header = () defaults = {} pack = None # pack pragma cell_header = None align = 4 nla_map = {} # NLA mapping sql_constraints = {} sql_extra_fields = () sql_extend = () lookup_fallbacks = {} nla_flags = 0 # NLA flags value_map = {} is_nla = False prefix = None own_parent = False header_type = None header_fmt = None decode_as = None # caches __compiled_nla = False __compiled_ft = False __t_nla_map = None __r_nla_map = None # schema __schema = None __slots__ = ( "_buf", "data", "chain", "offset", "length", "parent", "decoded", "_nla_init", "_nla_array", "_nla_flags", "value", "_r_value_map", "__weakref__", ) def msg_align(self, length): return (length + self.align - 1) & ~(self.align - 1) def __init__( self, data=None, offset=0, length=None, parent=None, init=None ): global cache_jit dict.__init__(self) self._buf = None self.data = data or bytearray() self.offset = offset self.length = length or 0 self.chain = [self] if parent is not None: # some structures use parents, some not, # so don't create cycles without need self.parent = parent if self.own_parent else weakref.proxy(parent) else: self.parent = None self.decoded = False self._nla_init = init self._nla_array = False self._nla_flags = self.nla_flags self['attrs'] = [] self.value = NotInitialized # work only on non-empty mappings if self.nla_map and not self.__class__.__compiled_nla: self.compile_nla_table() if self.header: self['header'] = {} @classmethod def sql_schema(cls): return SQLSchema(cls) @property def buf(self): logging.error( 'nlmsg.buf is deprecated:\n%s', ''.join(traceback.format_stack()) ) if self._buf is None: self._buf = io.BytesIO() self._buf.write(self.data[self.offset : self.length or None]) self._buf.seek(0) return self._buf def copy(self): ''' Return a decoded copy of the netlink message. Works correctly only if the message was encoded, or is received from the socket. ''' ret = type(self)(data=self.data, offset=self.offset) ret.decode() return ret def reset(self, buf=None): self.data = bytearray() self.offset = 0 self.decoded = False def register_clean_cb(self, cb): global clean_cbs if self.parent is not None: return self.parent.register_clean_cb(cb) else: # get the msg_seq -- if applicable seq = self.get('header', {}).get('sequence_number', None) if seq is not None and seq not in clean_cbs.__dict__: clean_cbs.__dict__[seq] = [] # attach the callback clean_cbs.__dict__[seq].append(cb) def unregister_clean_cb(self): global clean_cbs seq = self.get('header', {}).get('sequence_number', None) msf = self.get('header', {}).get('flags', 0) if ( (seq is not None) and (not msf & NLM_F_REQUEST) and seq in clean_cbs.__dict__ ): for cb in clean_cbs.__dict__[seq]: try: cb() except: log.error('Cleanup callback fail: %s' % (cb)) log.error(traceback.format_exc()) del clean_cbs.__dict__[seq] def _strip_one(self, name): for i in tuple(self['attrs']): if i[0] == name: self['attrs'].remove(i) return self def strip(self, attrs): ''' Remove an NLA from the attrs chain. The `attrs` parameter can be either string, or iterable. In the latter case, will be stripped NLAs, specified in the provided list. ''' if isinstance(attrs, basestring): self._strip_one(attrs) else: for name in attrs: self._strip_one(name) return self def __ops(self, rvalue, op0, op1): if rvalue is None: return None lvalue = self.getvalue() res = self.__class__() for key, _ in res.fields: res.pop(key, None) if 'header' in res: res.pop('header', None) if 'value' in res: res.pop('value', None) for key in lvalue: if key not in ('header', 'attrs', '__align'): if op0 == '__sub__': # operator -, complement if (key not in rvalue) or (lvalue[key] != rvalue[key]): res[key] = lvalue[key] elif op0 == '__and__': # operator &, intersection if (key in rvalue) and (lvalue[key] == rvalue[key]): res[key] = lvalue[key] if 'attrs' in lvalue: res['attrs'] = [] for attr in lvalue['attrs']: if isinstance(attr[1], nlmsg_base): diff = getattr(attr[1], op0)(rvalue.get_attr(attr[0])) if diff is not None: res['attrs'].append([attr[0], diff]) else: if op0 == '__sub__': # operator -, complement if rvalue.get_attr(attr[0]) != attr[1]: res['attrs'].append(attr) elif op0 == '__and__': # operator &, intersection if rvalue.get_attr(attr[0]) == attr[1]: res['attrs'].append(attr) if 'attrs' in res and not res['attrs']: del res['attrs'] if not res: return None return res def __bool__(self): return len(self.keys()) > 0 def __sub__(self, rvalue): ''' Subjunction operation. ''' return self.__ops(rvalue, '__sub__', '__ne__') def __and__(self, rvalue): ''' Conjunction operation. ''' return self.__ops(rvalue, '__and__', '__eq__') def __ne__(self, rvalue): return not self.__eq__(rvalue) def __eq__(self, rvalue): ''' Having nla, we are able to use it in operations like:: if nla == 'some value': ... ''' lvalue = self.getvalue() if lvalue is self: if isinstance(rvalue, type(self)): return (self - rvalue) is None if isinstance(rvalue, dict): return dict(self) == rvalue return False return lvalue == rvalue @classmethod def get_size(self): size = 0 for field in self.fields: size += struct.calcsize(field[1]) return size @classmethod def nla2name(self, name): ''' Convert NLA name into human-friendly name Example: IFLA_ADDRESS -> address Requires self.prefix to be set ''' return name[(name.find(self.prefix) + 1) * len(self.prefix) :].lower() @classmethod def name2nla(self, name): ''' Convert human-friendly name into NLA name Example: address -> IFLA_ADDRESS Requires self.prefix to be set ''' name = name.upper() if name.find(self.prefix) == -1: name = "%s%s" % (self.prefix, name) return name def decode(self): ''' Decode the message. The message should have the `buf` attribute initialized. e.g.:: data = sock.recv(16384) msg = ifinfmsg(data) If you want to customize the decoding process, override the method, but don't forget to call parent's `decode()`:: class CustomMessage(nlmsg): def decode(self): nlmsg.decode(self) ... # do some custom data tuning ''' offset = self.offset global cache_hdr global clean_cbs # Decode the header if self.header is not None: ## # ~~ self['header'][name] = struct.unpack_from(...) # # Instead of `struct.unpack()` all the NLA headers, it is # much cheaper to cache decoded values. The resulting dict # will be not much bigger than some hundreds ov values. # # The code might look ugly, but line_profiler shows here # a notable performance gain. # # The chain is: # dict.get(key, None) or dict.set(unpack(key, ...)) or dict[key] # # If there is no such key in the dict, get() returns None, and # Python executes __setitem__(), which always return None, and # then dict[key] is returned. # # If the key exists, the statement after the first `or` is not # executed. if self.is_nla: key = tuple(self.data[offset : offset + 4]) self['header'] = ( cache_hdr.get(key, None) or ( cache_hdr.__setitem__( key, dict( zip( ('length', 'type'), struct.unpack_from( 'HH', self.data, offset ), ) ), ) ) or cache_hdr[key] ) ## offset += 4 self.length = self['header']['length'] else: for name, fmt in self.header: self['header'][name] = struct.unpack_from( fmt, self.data, offset )[0] offset += struct.calcsize(fmt) # update length from header # it can not be less than 4 if 'header' in self: self.length = max(self['header']['length'], 4) # handle the array case if self._nla_array: self.setvalue([]) while offset < self.offset + self.length: cell = type(self)(data=self.data, offset=offset, parent=self) cell._nla_array = False if cell.cell_header is not None: cell.header = cell.cell_header cell.decode() self.value.append(cell) offset += (cell.length + 4 - 1) & ~(4 - 1) else: self.ft_decode(offset) if clean_cbs.__dict__: self.unregister_clean_cb() self.decoded = True def encode(self): ''' Encode the message into the binary buffer:: msg.encode() sock.send(msg.data) If you want to customize the encoding process, override the method:: class CustomMessage(nlmsg): def encode(self): ... # do some custom data tuning nlmsg.encode(self) ''' offset = self.offset diff = 0 # reserve space for the header if self.header is not None: hsize = struct.calcsize(''.join([x[1] for x in self.header])) self.data.extend([0] * hsize) offset += hsize # handle the array case if self._nla_array: header_type = 1 for value in self.getvalue(): cell = type(self)(data=self.data, offset=offset, parent=self) cell._nla_array = False if cell.cell_header is not None: cell.header = cell.cell_header cell.setvalue(value) # overwrite header type after calling setvalue cell['header']['type'] = self.header_type or ( header_type \| self._nla_flags ) header_type += 1 cell.encode() offset += (cell.length + 4 - 1) & ~(4 - 1) elif self.getvalue() is not None: offset, diff = self.ft_encode(offset) # write NLA chain if self.nla_map: offset = self.encode_nlas(offset) # calculate the size and write it if 'header' in self and self.header is not None: self.length = self['header']['length'] = ( offset - self.offset - diff ) offset = self.offset for name, fmt in self.header: default = self.defaults.get('header', {}).get(name, 0) struct.pack_into( fmt, self.data, offset, self['header'].get(name, default) ) offset += struct.calcsize(fmt) def setvalue(self, value): if isinstance(value, dict): self.update(value) if 'attrs' in value: self['attrs'] = [] for nla_tuple in value['attrs']: nlv = nlmsg_base() nlv.setvalue(nla_tuple[1]) self['attrs'].append([nla_tuple[0], nlv.getvalue()]) else: try: if value in self.value_map.values(): reverse_map = dict( [(x[1], x[0]) for x in self.value_map.items()] ) value = reverse_map.get(value, value) except TypeError: pass self['value'] = value self.value = value return self def get_encoded(self, attr, default=None): ''' Return the first encoded NLA by name ''' cells = [i[1] for i in self['attrs'] if i[0] == attr] if cells: return cells[0] def get(self, key, default=None): ''' Universal get() for a netlink message. ''' if isinstance(key, str): key = (key,) ret = self.get_nested(key) return ret if ret is not None else default def get_nested(self, keys): ''' Return nested NLA or None ''' pointer = self for attr in keys: if isinstance(pointer, nlmsg_base): # descendant nodes: NLA or fields # nla = attr if pointer.prefix: nla = pointer.name2nla(attr) else: nla = attr.upper() # try to descend to NLA value = pointer.get_attr(nla) # try to descend to a field if value is None and attr in pointer: value = pointer[attr] # replace pointer pointer = value elif isinstance(pointer, dict): # descendant nodes: dict values # pointer = pointer.get(attr) else: # stop descending; search failed return return pointer def get_attr(self, attr, default=None): ''' Return the first NLA with that name or None ''' try: attrs = self.get_attrs(attr) except KeyError: return default if attrs: return attrs[0] else: return default def get_attrs(self, attr): ''' Return attrs by name or an empty list ''' return [i[1] for i in self['attrs'] if i[0] == attr] def nla(self, attr=None, default=NotInitialized): ''' ''' if default is NotInitialized: response = nlmsg_base() del response['value'] del response['attrs'] response.value = None chain = self.get('attrs', []) if attr is not None: chain = [i.nla for i in chain if i.name == attr] else: chain = [i.nla for i in chain] if chain: for link in chain: link.chain = chain response = chain[0] return response def __getitem__(self, key): if isinstance(key, int): return self.chain[key] if key == 'value' and key not in self: return NotInitialized if key not in self: fields = dict(self.fields) if key in fields: return 0 return dict.__getitem__(self, key) def __delitem__(self, key): if key == 'value' and key not in self: return return dict.__delitem__(self, key) def __setstate__(self, state): return self.load(state) def __reduce__(self): return (type(self), (), self.dump()) def load(self, dump): ''' Load packet from a dict:: ipr = IPRoute() lo = ipr.link('dump', ifname='lo')[0] msg_type, msg_value = type(lo), lo.dump() ... lo = msg_type() lo.load(msg_value) The same methods -- `dump()`/`load()` -- implement the pickling protocol for the nlmsg class, see `__reduce__()` and `__setstate__()`. ''' if isinstance(dump, dict): for k, v in dump.items(): if k == 'header': self['header'].update(dump['header']) else: self[k] = v else: self.setvalue(dump) return self def dump_attrs(self, attrs): ret = [] for i in attrs: if isinstance(i, nlmsg_base): ret.append(i.dump()) elif isinstance(i, (set, list, tuple, nla_slot)): if isinstance(i[1], nlmsg_base): # catch nlmsg ret.append([i[0], i[1].dump()]) elif isinstance(i[1], (set, list, tuple)): # catch iterables ret.append([i[0], self.dump_attrs(i[1])]) elif isinstance(i[1], bytes): ret.append([i[0], hexdump(i[1])]) else: ret.append([i[0], i[1]]) elif isinstance(i, dict): if 'attrs' in i: i['attrs'] = self.dump_attrs(i['attrs']) ret.append(i) else: ret.append(i) return ret def dump(self): ''' Dump packet as a dict ''' a = self.getvalue() if isinstance(a, dict): ret = {} for k, v in a.items(): if k == 'header': ret['header'] = dict(a['header']) elif k == 'attrs': ret['attrs'] = self.dump_attrs(v) else: ret[k] = v elif isinstance(a, nlmsg_base): ret = a.dump() else: ret = a return ret def getvalue(self): ''' Atomic NLAs return their value in the 'value' field, not as a dictionary. Complex NLAs return whole dictionary. ''' if ( self._nla_array and len(self.value) and hasattr(self.value[0], 'getvalue') ): return [x.getvalue() for x in self.value] if self.value != NotInitialized: # value decoded by custom decoder return self.value if 'value' in self and self['value'] != NotInitialized: # raw value got by generic decoder return self.value_map.get(self['value'], self['value']) return self def compile_nla_table(self): # Bug-Url: https://github.com/svinota/pyroute2/issues/980 # Bug-Url: https://github.com/svinota/pyroute2/pull/981 if isinstance(self.nla_map, NlaMapAdapter): self.nla_map.types = self self.__class__.__t_nla_map = self.nla_map self.__class__.__r_nla_map = self.nla_map self.__class__.__compiled_nla = True return elif isinstance(self.nla_map, dict): if isinstance(self.nla_map['decode'], NlaMapAdapter): self.nla_map['decode'].types = self if isinstance(self.nla_map['encode'], NlaMapAdapter): self.nla_map['encode'].types = self self.__class__.__t_nla_map = self.nla_map['decode'] self.__class__.__r_nla_map = self.nla_map['encode'] self.__class__.__compiled_nla = True return # clean up NLA mappings t_nla_map = {} r_nla_map = {} # fix nla flags nla_map = [] for item in self.nla_map: if not isinstance(item[-1], int): item = list(item) item.append(0) nla_map.append(item) # detect, whether we have pre-defined keys if not isinstance(nla_map[0][0], int): # create enumeration nla_types = enumerate((i[0] for i in nla_map)) # that's a little bit tricky, but to reduce # the required amount of code in modules, we have # to jump over the head zipped = [ (k[1][0], k[0][0], k[0][1], k[0][2]) for k in zip(nla_map, nla_types) ] else: zipped = nla_map for key, name, nla_class, nla_flags in zipped: # it is an array if nla_class[0] == '': nla_class = nla_class[1:] nla_array = True else: nla_array = False # are there any init call in the string? lb = nla_class.find('(') rb = nla_class.find(')') if 0 < lb < rb: init = nla_class[lb + 1 : rb] nla_class = nla_class[:lb] else: init = None # lookup NLA class if nla_class == 'recursive': nla_class = type(self) elif nla_class == 'nested': nla_class = type(self) nla_flags \|= NLA_F_NESTED else: nla_class = getattr(self, nla_class) # update mappings prime = { 'class': nla_class, 'type': key, 'name': name, 'nla_flags': nla_flags, 'nla_array': nla_array, 'init': init, } t_nla_map[key] = r_nla_map[name] = prime self.__class__.__t_nla_map = t_nla_map self.__class__.__r_nla_map = r_nla_map self.__class__.__compiled_nla = True def valid_nla(self, nla): return nla in self.__class__.__r_nla_map.keys() def encode_nlas(self, offset): ''' Encode the NLA chain. Should not be called manually, since it is called from `encode()` routine. ''' r_nla_map = self.__class__.__r_nla_map for i in range(len(self['attrs'])): cell = self['attrs'][i] if cell[0] in r_nla_map: prime = r_nla_map[cell[0]] msg_class = prime['class'] # is it a class or a function? if isinstance(msg_class, types.FunctionType): # if it is a function -- use it to get the class msg_class = msg_class(self, value=cell[1]) # encode NLA nla_instance = msg_class( data=self.data, offset=offset, parent=self, init=prime['init'], ) nla_instance._nla_flags \|= prime['nla_flags'] if isinstance(cell, tuple) and len(cell) > 2: nla_instance._nla_flags \|= cell[2] nla_instance._nla_array = prime['nla_array'] nla_instance.setvalue(cell[1]) # overwrite header type after calling setvalue nla_instance['header']['type'] = ( prime['type'] \| nla_instance._nla_flags ) try: nla_instance.encode() except: raise else: nla_instance.decoded = True self['attrs'][i] = nla_slot(prime['name'], nla_instance) offset += (nla_instance.length + 4 - 1) & ~(4 - 1) return offset def decode_nlas(self, offset): ''' Decode the NLA chain. Should not be called manually, since it is called from `decode()` routine. ''' t_nla_map = self.__class__.__t_nla_map while offset - self.offset <= self.length - 4: nla_instance = None # pick the length and the type (length, base_msg_type) = struct.unpack_from( 'HH', self.data, offset ) # first two bits of msg_type are flags: msg_type = base_msg_type & ~(NLA_F_NESTED \| NLA_F_NET_BYTEORDER) # rewind to the beginning length = min(max(length, 4), (self.length - offset + self.offset)) # we have a mapping for this NLA if msg_type in t_nla_map: prime = t_nla_map[msg_type] # get the class msg_class = t_nla_map[msg_type]['class'] # is it a class or a function? if isinstance(msg_class, types.FunctionType): # if it is a function -- use it to get the class msg_class = msg_class(self, data=self.data, offset=offset) # decode NLA nla_instance = msg_class( data=self.data, offset=offset, parent=self, length=length, init=prime['init'], ) nla_instance._nla_array = prime['nla_array'] nla_instance._nla_flags = base_msg_type & ( NLA_F_NESTED \| NLA_F_NET_BYTEORDER ) name = prime['name'] else: name = 'UNKNOWN' nla_instance = nla_base( data=self.data, offset=offset, length=length ) self['attrs'].append(nla_slot(name, nla_instance)) offset += (length + 4 - 1) & ~(4 - 1) ## # 8<--------------------------------------------------------------------- # # NLMSG fields codecs, mixin classes # class nlmsg_decoder_generic(object): @staticmethod def decode_field(fmt, data, offset): global cache_fmt ## # ~~ size = struct.calcsize(efmt) # # The use of the cache gives here a tiny performance # improvement, but it is an improvement anyways # size = ( cache_fmt.get(fmt, None) or cache_fmt.__setitem__(fmt, struct.calcsize(fmt)) or cache_fmt[fmt] ) ## value = struct.unpack_from(fmt, data, offset) offset += size if len(value) == 1: return value[0], offset else: return value, offset def ft_decode(self, offset): for name, fmt in self.fields: if isinstance(fmt, str): self[name], offset = self.decode_field(fmt, self.data, offset) elif isinstance(fmt, type): self[name], offset = fmt.decode_from(self.data, offset) # read NLA chain if self.nla_map: offset = (offset + 4 - 1) & ~(4 - 1) try: self.decode_nlas(offset) except Exception as e: log.warning(traceback.format_exc()) raise NetlinkNLADecodeError(e) else: del self['attrs'] if self['value'] is NotInitialized: del self['value'] class nlmsg_decoder_string(object): def ft_decode(self, offset): (value,) = struct.unpack_from( '%is' % (self.length - 4), self.data, offset ) if self.zstring == 1: self['value'] = value.strip(b'\0') else: self['value'] = value class nlmsg_decoder_struct(object): def ft_decode(self, offset): names = [] fmt = '' for field in self.fields: names.append(field[0]) fmt += field[1] value = struct.unpack_from(fmt, self.data, offset) values = list(value) for name in names: if name[0] != '_': self[name] = values.pop(0) # read NLA chain if self.nla_map: offset = (offset + 4 - 1) & ~(4 - 1) try: self.decode_nlas(offset) except Exception as e: log.warning(traceback.format_exc()) raise NetlinkNLADecodeError(e) else: del self['attrs'] if self['value'] is NotInitialized: del self['value'] class nlmsg_encoder_generic(object): @staticmethod def encode_field(fmt, data, offset, value, zstring=0): if fmt == 's': length = len(value or '') + zstring efmt = '%is' % (length) else: length = struct.calcsize(fmt) efmt = fmt data.extend([0] length) # in python3 we should force it if isinstance(value, str): value = bytes(value, 'utf-8') elif isinstance(value, float): value = int(value) try: if fmt[-1] == 'x': struct.pack_into(efmt, data, offset) elif type(value) in (list, tuple, set): struct.pack_into(efmt, data, offset, value) elif len(fmt) > 1 and fmt[-1] == 'B' and value == 0: struct.pack_into(fmt[:-1] + 'x', data, offset) else: struct.pack_into(efmt, data, offset, value) except struct.error: log.error(''.join(traceback.format_stack())) log.error(traceback.format_exc()) log.error("error pack: %s %s %s" % (efmt, value, type(value))) raise return offset + length def ft_encode(self, offset): if hasattr(self, 'zstring'): zs = self.zstring else: zs = 0 for name, fmt in self.fields: default = self.defaults.get(name, 0) value = self[name] if self.get(name) is not None else default if isinstance(fmt, str): offset = self.encode_field(fmt, self.data, offset, value, zs) elif isinstance(fmt, type): offset = fmt.encode_into(self.data, offset, value) diff = 0 if self.align > 0: diff = ((offset + self.align - 1) & ~(self.align - 1)) - offset offset += diff self.data.extend([0] diff) return offset, diff # # 8<--------------------------------------------------------------------- ## class nla_slot(object): __slots__ = ("cell",) def __init__(self, name, value): self.cell = (name, value) def try_to_decode(self): try: cell = self.cell[1] if not cell.decoded: cell.decode() return True except Exception: log.warning("decoding %s" % (self.cell[0])) log.warning(traceback.format_exc()) return False def get_value(self): cell = self.cell[1] if self.try_to_decode(): return cell.getvalue() else: return cell.data[cell.offset : cell.offset + cell.length] def get_flags(self): if self.try_to_decode(): return self.cell[1]._nla_flags return None @property def name(self): return self.cell[0] @property def value(self): return self.get_value() @property def nla(self): self.try_to_decode() return self.cell[1] def __getitem__(self, key): if key == 1: return self.get_value() elif key == 0: return self.cell[0] elif isinstance(key, slice): s = list(self.cell.__getitem__(key)) if self.cell[1] in s: s[s.index(self.cell[1])] = self.get_value() return s else: raise IndexError(key) def __repr__(self): if self.get_flags(): return repr((self.cell[0], self.get_value(), self.get_flags())) return repr((self.cell[0], self.get_value())) ## # 8<--------------------------------------------------------------------- # # NLA base classes # class nla_header(object): __slots__ = () is_nla = True header = (('length', 'H'), ('type', 'H')) class nla_base( nla_header, nlmsg_base, nlmsg_encoder_generic, nlmsg_decoder_generic ): ''' Generic NLA base class. ''' __slots__ = () zstring = 0 class nla_base_string( nla_header, nlmsg_base, nlmsg_encoder_generic, nlmsg_decoder_string ): ''' NLA base class, string decoder. ''' __slots__ = () fields = [('value', 's')] zstring = 0 class nla_base_struct( nla_header, nlmsg_base, nlmsg_encoder_generic, nlmsg_decoder_struct ): ''' NLA base class, packed struct decoder. ''' __slots__ = () # # 8<--------------------------------------------------------------------- ## class nlmsg_atoms(object): ''' A collection of base NLA types ''' __slots__ = () class none(nla_base): ''' 'none' type is used to skip decoding of NLA. You can also use 'hex' type to dump NLA's content. ''' __slots__ = () def decode(self): nla_base.decode(self) self.value = None class flag(nla_base): ''' 'flag' type is used to denote attrs that have no payload ''' __slots__ = () fields = [] def decode(self): nla_base.decode(self) self.value = True class uint8(nla_base): __slots__ = () sql_type = 'INTEGER' fields = [('value', 'B')] class uint16(nla_base): __slots__ = () sql_type = 'INTEGER' fields = [('value', 'H')] class uint32(nla_base): __slots__ = () sql_type = 'BIGINT' fields = [('value', 'I')] class uint64(nla_base): __slots__ = () sql_type = 'BIGINT' fields = [('value', 'Q')] class int8(nla_base): __slots__ = () sql_type = 'INTEGER' fields = [('value', 'b')] class int16(nla_base): __slots__ = () sql_type = 'INTEGER' fields = [('value', 'h')] class int32(nla_base): __slots__ = () sql_type = 'BIGINT' fields = [('value', 'i')] class int64(nla_base): __slots__ = () sql_type = 'BIGINT' fields = [('value', 'q')] class be8(nla_base): __slots__ = () sql_type = 'INTEGER' fields = [('value', '>B')] class be16(nla_base): __slots__ = () sql_type = 'INTEGER' fields = [('value', '>H')] class be32(nla_base): __slots__ = () sql_type = 'BIGINT' fields = [('value', '>I')] class be64(nla_base): __slots__ = () sql_type = 'BIGINT' fields = [('value', '>Q')] class sbe8(nla_base): __slots__ = () sql_type = 'INTEGER' fields = [('value', '>b')] class sbe16(nla_base): __slots__ = () sql_type = 'INTEGER' fields = [('value', '>h')] class sbe32(nla_base): __slots__ = () sql_type = 'BIGINT' fields = [('value', '>i')] class sbe64(nla_base): __slots__ = () sql_type = 'BIGINT' fields = [('value', '>q')] class ipXaddr(nla_base_string): __slots__ = () sql_type = 'TEXT' family = None def encode(self): self['value'] = inet_pton(self.family, self.value) nla_base_string.encode(self) def decode(self): nla_base_string.decode(self) self.value = inet_ntop(self.family, self['value']) class ip4addr(ipXaddr): ''' Explicit IPv4 address type class. ''' __slots__ = () family = AF_INET class ip6addr(ipXaddr): ''' Explicit IPv6 address type class. ''' __slots__ = () family = AF_INET6 class ipaddr(nla_base_string): ''' This class is used to decode IP addresses according to the family. Socket library currently supports only two families, AF_INET and AF_INET6. We do not specify here the string size, it will be calculated in runtime. ''' __slots__ = () sql_type = 'TEXT' def ft_encode(self, offset): # use real provided family, not implicit if self.value.find(':') > -1: family = AF_INET6 else: family = AF_INET self['value'] = inet_pton(family, self.value) return nla_base_string.ft_encode(self, offset) def ft_decode(self, offset): nla_base_string.ft_decode(self, offset) # use real provided family, not implicit if self.length > 8: family = AF_INET6 else: family = AF_INET self.value = inet_ntop(family, self['value']) class target(nla_base_string): ''' A universal target class. The target type depends on the msg family: * AF_INET: IPv4 addr, string: "127.0.0.1" * AF_INET6: IPv6 addr, string: "::1" * AF_MPLS: MPLS labels, 0 .. k: [{"label": 0x20, "ttl": 16}, ...] ''' __slots__ = () sql_type = 'TEXT' family = None family_attr = None own_parent = True def __init__(self, argv, kwarg): init = kwarg.get('init', None) if init is not None: key, value = init.split(',') if key == 'family' and value.startswith('AF_'): self.family = getattr(socket, value) elif key == 'nla': self.family_attr = value super().__init__(argv, *kwarg) def get_family(self): if self.family is not None: return self.family pointer = self if self.family_attr is not None: nla = self.family_attr else: nla = 'family' while pointer.parent is not None: pointer = pointer.parent family = pointer.get(nla) if family is not None: return family return AF_UNSPEC @staticmethod def get_addrlen(family): return {AF_INET: 4, AF_INET6: 16, AF_MPLS: 4}.get(family, 4) def encode(self): family = self.get_family() if family in (AF_INET, AF_INET6): self['value'] = inet_pton(family, self.value) elif family == AF_MPLS: self['value'] = b'' if isinstance(self.value, (set, list, tuple)): labels = self.value else: if 'label' in self: labels = [ { 'label': self.get('label', 0), 'tc': self.get('tc', 0), 'bos': self.get('bos', 0), 'ttl': self.get('ttl', 0), } ] else: labels = [] for record in labels: label = ( (record.get('label', 0) << 12) \| (record.get('tc', 0) << 9) \| ((1 if record.get('bos') else 0) << 8) \| record.get('ttl', 0) ) self['value'] += struct.pack('>I', label) else: raise TypeError('socket family not supported') nla_base_string.encode(self) def decode(self): nla_base_string.decode(self) family = self.get_family() data = self['value'] if family in (AF_INET, AF_INET6): if family == AF_INET: data = data[:4] elif family == AF_INET6: data = data[:16] self.value = inet_ntop(family, data) elif family == AF_MPLS: self.value = [] for i in range(len(data) // 4): label = struct.unpack('>I', data[i 4 : i * 4 + 4])[0] record = { 'label': (label & 0xFFFFF000) >> 12, 'tc': (label & 0x00000E00) >> 9, 'bos': (label & 0x00000100) >> 8, 'ttl': label & 0x000000FF, } self.value.append(record) else: raise TypeError('socket family not supported') class mpls_target(target): __slots__ = () family = AF_MPLS class l2addr(nla_base): ''' Decode MAC address. ''' __slots__ = () sql_type = 'TEXT' fields = [('value', '=6s')] def encode(self): self['value'] = struct.pack( 'BBBBBB', [int(i, 16) for i in self.value.split(':')] ) nla_base.encode(self) def decode(self): nla_base.decode(self) self.value = ':'.join( '%02x' % (i) for i in struct.unpack('BBBBBB', self['value']) ) class lladdr(nla_base_string): ''' Decode link layer address: a MAC, IPv4 or IPv6 address. This type depends on the link layer address length: 6: MAC addr, string: "52:ff:ff:ff:ff:03" * 4: IPv4 addr, string: "127.0.0.1" * 16: IPv6 addr, string: "::1" * any other length: hex dump ''' __slots__ = () sql_type = 'TEXT' def encode(self): if ':' in self.value: if len(self.value) == 17 and '::' not in self.value: self['value'] = struct.pack( 'BBBBBB', [int(i, 16) for i in self.value.split(':')] ) else: self['value'] = inet_pton(AF_INET6, self.value) elif '.' in self.value: self['value'] = inet_pton(AF_INET, self.value) else: raise TypeError('Unsupported value {}'.format(self.value)) nla_base_string.encode(self) def decode(self): nla_base_string.decode(self) if len(self['value']) == 6: self.value = ':'.join( '%02x' % (i) for i in struct.unpack('BBBBBB', self['value']) ) elif len(self['value']) == 4: self.value = inet_ntop(AF_INET, self['value']) elif len(self['value']) == 16: self.value = inet_ntop(AF_INET6, self['value']) elif len(self['value']) == 0: self.value = '' else: # unknown / invalid lladdr # extract data for the whole message offset = self.parent.offset length = self.parent.length data = self.parent.data[offset : offset + length] # report logging.warning( 'unknown or invalid lladdr size, please report to: ' 'https://github.com/svinota/pyroute2/issues/717 \n' 'packet data: %s', hexdump(data), ) # continue with hex dump as the value self.value = hexdump(self['value']) class hex(nla_base_string): ''' Represent NLA's content with header as hex string. ''' __slots__ = () def decode(self): nla_base_string.decode(self) self.value = hexdump(self['value']) class array(nla_base_string): ''' Array of simple data type ''' __slots__ = ("_fmt",) own_parent = True @property def fmt(self): # try to get format from parent # work only with elementary types if getattr(self, "_fmt", None) is not None: return self._fmt try: fclass = getattr(self.parent, self._nla_init) self._fmt = fclass.fields[0][1] except Exception: self._fmt = self._nla_init return self._fmt def encode(self): fmt = '%s%i%s' % (self.fmt[:-1], len(self.value), self.fmt[-1:]) self['value'] = struct.pack(fmt, self.value) nla_base_string.encode(self) def decode(self): nla_base_string.decode(self) data_length = len(self['value']) element_size = struct.calcsize(self.fmt) array_size = data_length // element_size trail = (data_length % element_size) or -data_length data = self['value'][:-trail] fmt = '%s%i%s' % (self.fmt[:-1], array_size, self.fmt[-1:]) self.value = struct.unpack(fmt, data) class cdata(nla_base_string): ''' Binary data ''' __slots__ = () class string(nla_base_string): ''' UTF-8 string. ''' __slots__ = () sql_type = 'TEXT' def encode(self): if isinstance(self['value'], str) and sys.version[0] == '3': self['value'] = bytes(self['value'], 'utf-8') nla_base_string.encode(self) def decode(self): nla_base_string.decode(self) self.value = self['value'] try: self.value = self.value.decode('utf-8') except UnicodeDecodeError: pass # Failed to decode, keep undecoded value class asciiz(string): ''' Zero-terminated string. ''' __slots__ = () zstring = 1 # FIXME: support NLA_FLAG and NLA_MSECS as well. # # aliases to support standard kernel attributes: # binary = cdata # NLA_BINARY nul_string = asciiz # NLA_NUL_STRING ## # 8<--------------------------------------------------------------------- # # NLA base classes # class nla(nla_base, nlmsg_atoms): ''' Main NLA class ''' __slots__ = () def decode(self): nla_base.decode(self) del self['header'] class nla_string(nla_base_string, nlmsg_atoms): ''' NLA + string decoder ''' __slots__ = () def decode(self): nla_base_string.decode(self) del self['header'] class nla_struct(nla_base_struct, nlmsg_atoms): ''' NLA + packed struct decoder ''' __slots__ = () def decode(self): nla_base_struct.decode(self) del self['header'] # # 8<--------------------------------------------------------------------- ## class nlmsg( nlmsg_base, nlmsg_encoder_generic, nlmsg_decoder_generic, nlmsg_atoms ): ''' Main netlink message class ''' __slots__ = () header = ( ('length', 'I'), ('type', 'H'), ('flags', 'H'), ('sequence_number', 'I'), ('pid', 'I'), ) class nlmsgerr(nlmsg): ''' Extended ack error message ''' __slots__ = () fields = (('error', 'i'),) prefix = 'NLMSGERR_ATTR_' nla_map = ( ('NLMSGERR_ATTR_UNUSED', 'none'), ('NLMSGERR_ATTR_MSG', 'asciiz'), ('NLMSGERR_ATTR_OFFS', 'uint32'), ('NLMSGERR_ATTR_COOKIE', 'uint8'), ) class genlmsg(nlmsg): ''' Generic netlink message ''' __slots__ = () fields = (('cmd', 'B'), ('version', 'B'), ('reserved', 'H')) class ctrlmsg(genlmsg): ''' Netlink control message ''' __slots__ = () # FIXME: to be extended nla_map = ( ('CTRL_ATTR_UNSPEC', 'none'), ('CTRL_ATTR_FAMILY_ID', 'uint16'), ('CTRL_ATTR_FAMILY_NAME', 'asciiz'), ('CTRL_ATTR_VERSION', 'uint32'), ('CTRL_ATTR_HDRSIZE', 'uint32'), ('CTRL_ATTR_MAXATTR', 'uint32'), ('CTRL_ATTR_OPS', 'ops'), ('CTRL_ATTR_MCAST_GROUPS', 'mcast_groups'), ('CTRL_ATTR_POLICY', 'policy_nest'), ('CTRL_ATTR_OP_POLICY', 'command_nest'), ('CTRL_ATTR_OP', 'uint32'), ) class ops(nla): __slots__ = () nla_map = ( ('CTRL_ATTR_OP_UNSPEC', 'none'), ('CTRL_ATTR_OP_ID', 'uint32'), ('CTRL_ATTR_OP_FLAGS', 'uint32'), ) class mcast_groups(nla): __slots__ = () nla_map = ( ('CTRL_ATTR_MCAST_GRP_UNSPEC', 'none'), ('CTRL_ATTR_MCAST_GRP_NAME', 'asciiz'), ('CTRL_ATTR_MCAST_GRP_ID', 'uint32'), ) class policy_nest(nla): __slots__ = () nla_map = { 'decode': NlaMapAdapter( lambda x: NlaSpec('attribute_nest', x, f'POLICY({x})') ), 'encode': NlaMapAdapter( lambda x: NlaSpec('attribute_nest', int(x[7:-1]), x) ), } class attribute_nest(nla): __slots__ = () nla_map = { 'decode': NlaMapAdapter( lambda x: NlaSpec('nl_policy_type_attr', x, f'ATTR({x})') ), 'encode': NlaMapAdapter( lambda x: NlaSpec('nl_policy_type_attr', int(x[5:-1]), x) ), } class nl_policy_type_attr(nla): __slots__ = () nla_map = ( ('NL_POLICY_TYPE_ATTR_UNSPEC', 'none'), ('NL_POLICY_TYPE_ATTR_TYPE', 'uint32'), ('NL_POLICY_TYPE_ATTR_MIN_VALUE_S', 'int64'), ('NL_POLICY_TYPE_ATTR_MAX_VALUE_S', 'int64'), ('NL_POLICY_TYPE_ATTR_MIN_VALUE_U', 'int64'), ('NL_POLICY_TYPE_ATTR_MAX_VALUE_U', 'int64'), ('NL_POLICY_TYPE_ATTR_MIN_LENGTH', 'uint32'), ('NL_POLICY_TYPE_ATTR_MAX_LENGTH', 'uint32'), ('NL_POLICY_TYPE_ATTR_POLICY_IDX', 'uint32'), ('NL_POLICY_TYPE_ATTR_POLICY_MAXTYPE', 'uint32'), ('NL_POLICY_TYPE_ATTR_BITFIELD32_MASK', 'uint32'), ('NL_POLICY_TYPE_ATTR_PAD', 'uint64'), ('NL_POLICY_TYPE_ATTR_MASK', 'uint64'), ) class command_nest(nla): __slots__ = () nla_map = { 'decode': NlaMapAdapter( lambda x: NlaSpec('command_nest_attrs', x, f'OP({x})') ), 'encode': NlaMapAdapter( lambda x: NlaSpec('command_nest_attrs', int(x[3:-1]), x) ), } class command_nest_attrs(nla): __slots__ = () nla_map = ( ('CTRL_ATTR_POLICY_UNSPEC', 'none'), ('CTRL_ATTR_POLICY_DO', 'uint32'), ('CTRL_ATTR_POLICY_DUMP', 'uint32'), )	73.26 KB	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 buffer.py try: from multiprocessing import shared_memory except ImportError: shared_memory = None class Page: ''' Memory page. ''' def __init__(self, view, offset): self.view = view self.offset = offset self.is_free = True def use(self): self.is_free = False def free(self): self.is_free = True def close(self): self.view.release() class Buffer: ''' Manage the buffer memory to receive raw netlink data. ''' def __init__(self, mode='internal', size=10485760, page_size=32768): self.mode = mode self.size = size self.page_size = page_size if self.mode == 'internal': self.mem = None self.buf = bytearray(self.size) elif self.mode == 'shared': if shared_memory is None: raise ModuleNotFoundError('shared memory buffer not supported') self.mem = shared_memory.SharedMemory(create=True, size=self.size) self.buf = self.mem.buf self.view = memoryview(self.buf) self.directory = {} for index in range(size // page_size): offset = index * page_size self.directory[index] = Page( self.view[offset : offset + self.page_size], offset ) def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): self.close() def get_free_page(self): for index, page in self.directory.items(): if page.is_free: page.use() return page raise MemoryError('no free memory pages available') def close(self): for page in self.directory.values(): page.close() self.view.release() if self.mode == 'shared': self.mem.close() self.mem.unlink() def __getitem__(self, key): return self.directory[key]	1.9 KB	2025-05-25 12:26
📁 connector	-	2025-05-25 12:26
📄 core.py import asyncio import collections import ctypes import errno import logging import os import socket import struct import threading import time import warnings from contextlib import contextmanager from typing import Optional, Union from urllib import parse from pyroute2 import config, netns from pyroute2.common import AddrPool from pyroute2.netlink import NLM_F_MULTI from pyroute2.netlink.coredata import CoreConfig, CoreSocketSpec from pyroute2.statsd import StatsDClientSocket MAGIC_CLOSE = 0x42 log = logging.getLogger(__name__) Stats = collections.namedtuple('Stats', ('qsize', 'delta', 'delay')) class Average: def __init__(self) -> None: self.total: int = 0 self.count: int = 0 def add(self, value: int) -> None: self.total += value self.count += 1 def getvalue(self) -> int: return self.total // self.count if self.count > 0 else 0 class Telemetry: sock: Union[config.LocalMock, StatsDClientSocket] 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, ): address = address or config.telemetry self.timings: dict[str, Average] = {} if address is None: self.sock = config.LocalMock() return save: bool = config.mock_netns config.mock_netns = False self.sock = StatsDClientSocket(address, use_socket, flags, libc) config.mock_netns = save @contextmanager def update(self, name: str): start = time.time_ns() try: yield self finally: stop = time.time_ns() if name not in self.timings: self.timings[name] = Average() self.timings[name].add(stop - start) self.sock.put(name, 1, 'c') self.sock.put(name, self.timings[name].getvalue(), 'g') self.sock.commit() def incr(self, name: str) -> None: self.sock.incr(name) def close(self): self.sock.close() class NoClose(socket.socket): def __init__(self, sock: socket.socket): self.magic = 0 fd = os.dup(sock.fileno()) sock.close() super().__init__(fileno=fd) def close(self): if self.magic == MAGIC_CLOSE: super().close() class CoreMessageQueue: def __init__(self, event_loop): # python versions < 3.11 require the event loop # to be running on Queue().__init__(), while # python 3.12+ bind the queue on get()/put() # # so we must require the event loop on __init__() # just to provide the compatibility. # self.event_loop = event_loop self.queues = {0: asyncio.Queue()} self.root = self.queues[0] async def join(self): return await self.root.join() async def reschedule(self, target): while not self.root.empty(): target.append(await self.root.get()) self.root.task_done() for key in tuple(self.queues.keys()): del self.queues[key] async def get(self, tag): ret = await self.queues[tag].get() self.queues[tag].task_done() return ret async def put(self, tag, message): if tag not in self.queues: tag = 0 return await self.queues[tag].put(message) def ensure_tag(self, tag): if tag not in self.queues: self.queues[tag] = asyncio.Queue() def free_tag(self, tag): del self.queues[tag] def put_nowait(self, tag, message): if tag not in self.queues: tag = 0 return self.queues[tag].put_nowait(message) class CoreProtocol(asyncio.Protocol): def __init__(self, on_con_lost, enqueue, error_event, status): self.transport = None self.enqueue = enqueue self.on_con_lost = on_con_lost self.error_event = error_event self.status = status def connection_made(self, transport): self.transport = transport def enqueue_error(self, code: Optional[int]) -> None: if code is None: code = errno.ECOMM try: self.enqueue(struct.pack('IHHQIQQ', 28, 2, 0, 0, code, 0, 0), None) except AttributeError: # while closing, self.msg_queue can be removed already pass def connection_lost(self, exc: Optional[Exception]) -> None: self.on_con_lost.set_result(True) self.enqueue_error(errno.ECONNRESET) def error_received(self, exc: OSError) -> None: self.status['error'] = exc self.error_event.set() self.enqueue_error(exc.errno) class CoreStreamProtocol(CoreProtocol): def data_received(self, data): log.debug('SOCK_STREAM enqueue %s bytes' % len(data)) self.enqueue(data, None) class CoreDatagramProtocol(CoreProtocol): def datagram_received(self, data, addr): log.debug('SOCK_DGRAM enqueue %s bytes' % len(data)) self.enqueue(data, addr) class AsyncCoreSocket: '''Pyroute2 core socket class. This class implements the core socket concept for all the pyroute2 communications, both Netlink and internal RPC. The asynchronous version is the most basic. All the sync classes are built on top of it. ''' compiled = None __spec = None __marshal = None def __init__( self, target='localhost', rcvsize=16384, use_socket=None, netns=None, flags=os.O_CREAT, libc=None, groups=0, use_event_loop=None, telemetry=None, ): # 8<----------------------------------------- self.spec = CoreSocketSpec( CoreConfig( target=target, rcvsize=rcvsize, netns=netns, flags=flags, groups=groups, use_libc=libc is not None, use_socket=use_socket is not None, use_event_loop=use_event_loop is not None, telemetry=telemetry, ) ) self.status = self.spec.status self.status['eids'] = set() self.local = threading.local() self.lock = threading.Lock() self.libc = libc self.use_socket = use_socket self.use_event_loop = use_event_loop self.request_proxy = None self._tid = threading.get_ident() self._error_event = threading.Event() url = parse.urlparse(self.spec['target']) self.scheme = url.scheme if url.scheme else url.path self.callbacks = [] # [(predicate, callback, args), ...] self.addr_pool = AddrPool(minaddr=0x000000FF, maxaddr=0x0000FFFF) self.marshal = None self.buffer = [] self.telemetry = Telemetry(telemetry) self.msg_reschedule = [] self.__open_sockets = set() self.__open_resources = set() self.__msg_queues = set() def _check_tid( self, tag: str = '', level: int = logging.ERROR, message: Optional[str] = None, ): if self._tid != threading.get_ident(): if message is None: message = ( f'calling #{tag} from another thread may require' f' additional resource cleanup' ) if level == logging.ERROR: raise RuntimeError(message) elif level == logging.WARN: warnings.warn(message) def _register_loop_ref(self): self.__open_resources.add((self.event_loop, self.msg_queue)) def get_loop(self): return self.event_loop @property def spec(self): return self.__spec @spec.setter def spec(self, value): if self.__spec is None: self.__spec = value @property def marshal(self): return self.__marshal @marshal.setter def marshal(self, value): if self.__marshal is None: self.__marshal = value else: log.debug( 'preventing override of marshal %s with %s', self.__marshal, value, ) def set_marshal(self, value): self.__marshal = value # 8<-------------------------------------------------------------- # Thread local section @property def msg_queue(self): return self.local.msg_queue @property def connection_lost(self): if self._error_event.is_set(): raise self.status['error'] if not hasattr(self.local, 'connection_lost'): self.local.connection_lost = self.event_loop.create_future() return self.local.connection_lost @property def socket(self): '''socket infrastructure reconciler.''' if self._error_event.is_set(): raise self.status['error'] if self.use_socket is not None and not hasattr(self.local, 'socket'): if not hasattr(self.use_socket, '_lock'): self._check_tid(tag='use_socket', level=logging.ERROR) self.local.prime = None self.local.socket = self.use_socket if not hasattr(self.local, 'prime'): self.local.prime = self.setup_socket() if not hasattr(self.local, 'msg_queue'): self.local.msg_queue = CoreMessageQueue(event_loop=self.event_loop) self.__msg_queues.add(self.local.msg_queue) if not hasattr(self.local, 'socket'): self.local.socket = NoClose(self.local.prime) self.__open_sockets.add(self.local.socket) return self.local.socket @property def event_loop(self): if self._error_event.is_set(): raise self.status['error'] if not hasattr(self.local, 'event_loop'): if self.status['use_event_loop']: self._check_tid( level=logging.ERROR, message='Predefined event loop can not be used ' 'in another thread', ) self.local.event_loop = self.setup_event_loop() if self.local.event_loop.is_closed(): raise OSError(errno.EBADF, 'Bad file descriptor') return self.local.event_loop @property def transport(self): return self.local.transport @property def protocol(self): if not hasattr(self.local, 'protocol'): self.local.protocol = None return self.local.protocol # # 8<-------------------------------------------------------------- # def setup_event_loop(self): try: event_loop = asyncio.get_running_loop() self.status['event_loop'] = 'auto' except RuntimeError: event_loop = asyncio.new_event_loop() self.status['event_loop'] = 'new' self.status['eids'].add(id(event_loop)) return event_loop def setup_socket(self): if self.use_socket is not None: return self.use_socket sock = netns.create_socket( self.spec['netns'], socket.AF_INET, socket.SOCK_STREAM, flags=self.spec['flags'], libc=self.libc, ) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) return sock async def setup_endpoint(self): # Setup asyncio if getattr(self.local, 'transport', None) is not None: return self.local.transport, self.local.protocol = ( await self.event_loop.connect_accepted_socket( lambda: CoreStreamProtocol( self.connection_lost, self.enqueue, self._error_event, self.status, ), sock=self.socket, ) ) # 8<-------------------------------------------------------------- def __getattr__(self, attr): if attr in ( 'getsockname', 'getsockopt', 'makefile', 'setsockopt', 'setblocking', 'settimeout', 'gettimeout', 'shutdown', 'recvfrom', 'recvfrom_into', 'fileno', 'sendto', 'connect', 'listen', ): return getattr(self.socket, attr) elif attr in ('_sendto', '_recv', '_recv_into'): return getattr(self.socket, attr.lstrip("_")) raise AttributeError(attr) async def bind(self, addr): '''Bind the socket to the address.''' await self.setup_endpoint() return self.socket.bind(addr) def close(self, code=errno.ECONNRESET): '''Terminate the object.''' def send_terminator(msg_queue): msg_queue.put_nowait(0, b'') with self.lock: for event_loop, msg_queue in self.__open_resources: if not event_loop.is_closed(): event_loop.call_soon_threadsafe(send_terminator, msg_queue) if self.telemetry is not None: self.telemetry.close() if hasattr(self.local, 'transport'): self.local.transport.abort() self.local.transport.close() del self.local.transport del self.local.protocol if self.status['event_loop'] == 'new': if ( hasattr(self.local, 'event_loop') and not self.local.event_loop.is_closed() ): self.event_loop.run_until_complete( self.event_loop.shutdown_asyncgens() ) self.event_loop.stop() self.event_loop.close() del self.local.event_loop for sock in tuple(self.__open_sockets): sock.magic = MAGIC_CLOSE sock.close() def clone(self): '''Return a copy of itself with a new underlying socket. This method can not work if `use_socket` or `event_loop` was used in the object constructor.''' if self.status['use_socket'] or self.status['event_loop']: raise RuntimeError('can not clone socket') new_spec = {} for key, value in self.spec.items(): if key in self.__init__.__code__.co_varnames: new_spec[key] = value return type(self)(new_spec) def setsockopt(self, level, optname, value): return self.socket.setsockopt(level, optname, value) def getsockopt(self, level, optname): return self.socket.getsockopt(level, optname) def recv(self, buffersize, flags=0): '''Get one buffer from the socket.''' return self.socket.recv(buffersize, flags) def send(self, data, flags=0): '''Send one buffer via the socket.''' return self.socket.send(data, flags) def accept(self): if self.use_socket is not None: return (self, None) (connection, address) = self.socket.accept() new_socket = self.clone() new_socket.socket = connection return (new_socket, address) def connect(self, address): self.socket.connect(address) def enqueue(self, data, addr): return self.msg_queue.put_nowait(0, data) async def get( self, msg_seq=0, terminate=None, callback=None, noraise=False ): '''Get a conversation answer from the socket.''' await self.setup_endpoint() log.debug( "get: %s / %s / %s / %s", msg_seq, terminate, callback, noraise ) if msg_seq == -1: msg_seq = 0 enough = False started = False error = None while not enough: log.debug('await data on %s', self.msg_queue) data = await self.msg_queue.get(msg_seq) messages = tuple(self.marshal.parse(data, msg_seq, callback)) if len(messages) == 0: break for msg in messages: log.debug("message %s", msg) if msg.get('header', {}).get('error') is not None: error = msg['header']['error'] enough = True break if self.marshal.is_enough(msg): enough = True break msg['header']['target'] = self.status['target'] msg['header']['stats'] = Stats(0, 0, 0) started = True log.debug("yield %s", msg['header']) yield msg if started and ( (msg_seq == 0) or (not msg['header'].get('flags', 0) & NLM_F_MULTI) or (callable(terminate) and terminate(msg)) ): enough = True if not noraise and error: raise error async def __aenter__(self): return self async def __aexit__(self, exc_type, exc_value, traceback): self.close() def register_callback(self, callback, predicate=lambda x: True, args=None): ''' Register a callback to run on a message arrival. Callback is the function that will be called with the message as the first argument. Predicate is the optional callable object, that returns True or False. Upon True, the callback will be called. Upon False it will not. Args is a list or tuple of arguments. Simplest example, assume ipr is the IPRoute() instance:: # create a simplest callback that will print messages def cb(msg): print(msg) # register callback for any message: ipr.register_callback(cb) More complex example, with filtering:: # Set object's attribute after the message key def cb(msg, obj): obj.some_attr = msg["some key"] # Register the callback only for the loopback device, index 1: ipr.register_callback(cb, lambda x: x.get('index', None) == 1, (self, )) Please note: you do not need to register the default 0 queue to invoke callbacks on broadcast messages. Callbacks are iterated before** messages get enqueued. ''' if args is None: args = [] self.callbacks.append((predicate, callback, args)) def unregister_callback(self, callback): ''' Remove the first reference to the function from the callback register ''' cb = tuple(self.callbacks) for cr in cb: if cr[1] == callback: self.callbacks.pop(cb.index(cr)) return def register_policy(self, policy, msg_class=None): ''' Register netlink encoding/decoding policy. Can be specified in two ways: `nlsocket.register_policy(MSG_ID, msg_class)` to register one particular rule, or `nlsocket.register_policy({MSG_ID1: msg_class})` to register several rules at once. E.g.:: policy = {RTM_NEWLINK: ifinfmsg, RTM_DELLINK: ifinfmsg, RTM_NEWADDR: ifaddrmsg, RTM_DELADDR: ifaddrmsg} nlsocket.register_policy(policy) One can call `register_policy()` as many times, as one want to -- it will just extend the current policy scheme, not replace it. ''' if isinstance(policy, int) and msg_class is not None: policy = {policy: msg_class} if not isinstance(policy, dict): raise TypeError('wrong policy type') for key in policy: self.marshal.msg_map[key] = policy[key] return self.marshal.msg_map def unregister_policy(self, policy): ''' Unregister policy. Policy can be: - int -- then it will just remove one policy - list or tuple of ints -- remove all given - dict -- remove policies by keys from dict In the last case the routine will ignore dict values, it is implemented so just to make it compatible with `get_policy_map()` return value. ''' if isinstance(policy, int): policy = [policy] elif isinstance(policy, dict): policy = list(policy) if not isinstance(policy, (tuple, list, set)): raise TypeError('wrong policy type') for key in policy: del self.marshal.msg_map[key] return self.marshal.msg_map def get_policy_map(self, policy=None): ''' Return policy for a given message type or for all message types. Policy parameter can be either int, or a list of ints. Always return dictionary. ''' if policy is None: return self.marshal.msg_map if isinstance(policy, int): policy = [policy] if not isinstance(policy, (list, tuple, set)): raise TypeError('wrong policy type') ret = {} for key in policy: ret[key] = self.marshal.msg_map[key] return ret class SyncAPI: ''' Synchronous API wrapper around asynchronous classes ''' @property def marshal(self): return self.asyncore.marshal @marshal.setter def marshal(self, value): self.asyncore.marshal = value def set_marshal(self, value): return self.asyncore.set_marshal(value) def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): self.close() def __getattr__(self, key): if key in ( 'epid', 'pid', 'spec', 'status', 'send', 'recv', 'lock', 'sendto', 'setsockopt', 'getsockopt', 'register_policy', 'unregister_policy', ): return getattr(self.asyncore, key) raise AttributeError(key) @property def fileno(self): return self.asyncore.socket.fileno def _setup_transport(self): if self.asyncore.status['closed']: raise OSError(errno.EBADF, 'Bad file descriptor') if hasattr(self.asyncore.local, 'event_loop'): return self.asyncore.event_loop.run_until_complete( self.asyncore.setup_endpoint() ) def _cleanup_transport(self) -> None: if not self.asyncore.status['closed'] and hasattr( self.asyncore.local, 'keep_event_loop' ): return if hasattr(self.asyncore, 'transport'): self.asyncore.transport.abort() self.asyncore.transport.close() self.asyncore.event_loop.run_until_complete( self.asyncore.event_loop.shutdown_asyncgens() ) self.asyncore.event_loop.stop() self.asyncore.event_loop.close() if hasattr(self.asyncore.local, 'transport'): del self.asyncore.local.transport if hasattr(self.asyncore.local, 'connection_lost'): del self.asyncore.local.connection_lost if hasattr(self.asyncore.local, 'event_loop'): del self.asyncore.local.event_loop def _run_with_cleanup(self, func, cmd: str, argv, kwarg): if len(argv) > 0 and isinstance(argv[0], str): cmd = f'{cmd}-{argv[0]}' try: self._setup_transport() with self.asyncore.telemetry.update(cmd): return self.asyncore.event_loop.run_until_complete( func(argv, **kwarg) ) finally: self._cleanup_transport() def mock_data(self, data): if getattr(self.asyncore.local, 'msg_queue', None) is None: self.asyncore.local.msg_queue = CoreMessageQueue( event_loop=self.event_loop ) self.asyncore.msg_queue.put_nowait(0, data) def close(self, code=errno.ECONNRESET): '''Correctly close the socket and free all the resources.''' self.asyncore.status['closed'] = True self.asyncore.close(code) class CoreSocket(SyncAPI): def __init__( self, target='localhost', rcvsize=16384, use_socket=None, netns=None, flags=os.O_CREAT, libc=None, groups=0, ): self.asyncore = AsyncCoreSocket( target, rcvsize, use_socket, netns, flags, libc, groups ) 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')	24.51 KB	2025-05-25 12:26
📄 coredata.py import os from dataclasses import asdict, dataclass from typing import Optional, Type, Union from pyroute2 import config from pyroute2.requests.main import RequestFilter, RequestProcessor @dataclass class CoreConfig: target: str = 'localhost' flags: int = os.O_CREAT groups: int = 0 rcvsize: int = 16384 netns: Optional[str] = None tag_field: str = '' address: Optional[tuple[str, int]] = None telemetry: Optional[tuple[str, int]] = None use_socket: bool = False use_event_loop: bool = False use_libc: bool = False class CoreSocketSpec: defaults: dict[str, Union[bool, int, str, None, tuple[str, ...]]] = { 'closed': False, 'compiled': None, 'uname': config.uname, } status_filters: list[Type[RequestFilter]] = [] def __init__(self, config: CoreConfig): self.config = config self.status = RequestProcessor() for flt in self.status_filters: self.status.add_filter(flt()) self.status.update(self.defaults) self.status.update(asdict(self.config)) def __setitem__(self, key, value): setattr(self.config, key, value) self.status.update(asdict(self.config)) def __getitem__(self, key): return getattr(self.config, key) def serializable(self): return not any( ( self.config.use_socket, self.config.use_event_loop, self.config.use_libc, ) )	1.47 KB	2025-05-25 12:26
📁 devlink	-	2025-05-25 12:26
📁 diag	-	2025-05-25 12:26
📁 event	-	2025-05-25 12:26
📄 exceptions.py import os import time class NetlinkError(Exception): ''' Base netlink error ''' def __init__(self, code, msg=None): msg = msg or os.strerror(code) super(NetlinkError, self).__init__(code, msg) self.code = code self.extra_code = 0 class NetlinkDecodeError(Exception): ''' Base decoding error class. Incapsulates underlying error for the following analysis ''' def __init__(self, exception): self.exception = exception class NetlinkHeaderDecodeError(NetlinkDecodeError): ''' The error occured while decoding a header ''' pass class NetlinkDataDecodeError(NetlinkDecodeError): ''' The error occured while decoding the message fields ''' pass class NetlinkNLADecodeError(NetlinkDecodeError): ''' The error occured while decoding NLA chain ''' pass class IPSetError(NetlinkError): ''' Netlink error with IPSet special error codes. Messages are imported from errcode.c ''' pass class NetlinkDumpInterrupted(NetlinkError): ''' Raised when NLM_F_DUMP_INTR is set in the flags. ''' def __init__(self, code=-1, msg='dump interrupted'): super(NetlinkDumpInterrupted, self).__init__(code, msg) class SkipInode(Exception): def __init__(self, code=0, msg=None): super(SkipInode, self).__init__(code, msg) self.code = code class ChaoticException(Exception): def __init__(self): chaotic_id = str(time.time()) super(ChaoticException, self).__init__(chaotic_id) self.chaotic_id = chaotic_id	1.58 KB	2025-05-25 12:26
📁 generic	-	2025-05-25 12:26
📁 ipq	-	2025-05-25 12:26
📄 marshal.py import errno import struct import threading from functools import partial from typing import Callable, Generator, Optional from pyroute2.netlink import ( NLM_F_ACK_TLVS, NLMSG_DONE, NLMSG_ERROR, mtypes, nlmsg, nlmsgerr, ) from pyroute2.netlink.exceptions import ( NetlinkDecodeError, NetlinkError, NetlinkHeaderDecodeError, ) class Marshal: ''' Generic marshalling class ''' msg_map = {} seq_map = None key_offset = None key_format = None key_mask = None debug = False default_message_class = nlmsg error_type = NLMSG_ERROR def __init__(self): self.lock = threading.Lock() self.msg_map = self.msg_map.copy() self.seq_map = {} self.defragmentation = {} def parse_one_message( self, key, flags, sequence_number, data, offset, length ): msg = None error = None msg_class = self.msg_map.get(key, self.default_message_class) # ignore length for a while # get the message if (key == self.error_type) or ( key == NLMSG_DONE and flags & NLM_F_ACK_TLVS ): msg = nlmsgerr(data, offset=offset) else: msg = msg_class(data, offset=offset) try: msg.decode() except NetlinkHeaderDecodeError as e: msg = nlmsg() msg['header']['error'] = e except NetlinkDecodeError as e: msg['header']['error'] = e if isinstance(msg, nlmsgerr) and msg['error'] != 0: code = abs(msg['error']) if code == errno.ENOBUFS: error = OSError(code, msg.get_attr('NLMSGERR_ATTR_MSG')) else: error = NetlinkError(code, msg.get_attr('NLMSGERR_ATTR_MSG')) enc_type = struct.unpack_from('H', data, offset + 24)[0] enc_class = self.msg_map.get(enc_type, nlmsg) enc = enc_class(data, offset=offset + 20) enc.decode() msg['header']['errmsg'] = enc msg['header']['error'] = error return msg def get_parser(self, key, flags, sequence_number): return self.seq_map.get( sequence_number, partial(self.parse_one_message, key, flags, sequence_number), ) def parse( self, data: bytes, seq: Optional[int] = None, callback: Optional[Callable] = None, skip_alien_seq: bool = False, ) -> Generator[nlmsg, None, None]: ''' Parse string data. At this moment all transport, except of the native Netlink is deprecated in this library, so we should not support any defragmentation on that level ''' offset = 0 # there must be at least one header in the buffer, # 'IHHII' == 16 bytes while offset <= len(data) - 16: # pick type and length (length, key, flags, sequence_number) = struct.unpack_from( 'IHHI', data, offset ) if skip_alien_seq and sequence_number != seq: continue if not 0 < length <= len(data): break # support custom parser keys # see also: pyroute2.netlink.diag.MarshalDiag if self.key_format is not None: (key,) = struct.unpack_from( self.key_format, data, offset + self.key_offset ) if self.key_mask is not None: key &= self.key_mask parser = self.get_parser(key, flags, sequence_number) msg = parser(data, offset, length) offset += length if msg is None: continue if callable(callback) and seq == sequence_number: try: if callback(msg): continue except Exception: pass mtype = msg['header'].get('type', None) if mtype in (1, 2, 3, 4) and 'event' not in msg: msg['event'] = mtypes.get(mtype, 'none') self.fix_message(msg) yield msg def is_enough(self, msg): return msg['header']['type'] == NLMSG_DONE def fix_message(self, msg): pass	4.24 KB	2025-05-25 12:26
📁 nfnetlink	-	2025-05-25 12:26
📁 nl80211	-	2025-05-25 12:26
📄 nlsocket.py ''' Base netlink socket and marshal =============================== All the netlink providers are derived from the socket class, so they provide normal socket API, including `getsockopt()`, `setsockopt()`, they can be used in poll/select I/O loops etc. asynchronous I/O ---------------- To run async reader thread, one should call `NetlinkSocket.bind(async_cache=True)`. In that case a background thread will be launched. The thread will automatically collect all the messages and store into a userspace buffer. .. note:: There is no need to turn on async I/O, if you don't plan to receive broadcast messages. ENOBUF and async I/O -------------------- When Netlink messages arrive faster than a program reads then from the socket, the messages overflow the socket buffer and one gets ENOBUF on `recv()`:: ... self.recv(bufsize) error: [Errno 105] No buffer space available One way to avoid ENOBUF, is to use async I/O. Then the library not only reads and buffers all the messages, but also re-prioritizes threads. Suppressing the parser activity, the library increases the response delay, but spares CPU to read and enqueue arriving messages as fast, as it is possible. With logging level DEBUG you can notice messages, that the library started to calm down the parser thread:: DEBUG:root:Packet burst: the reader thread priority is increased, beware of delays on netlink calls Counters: delta=25 qsize=25 delay=0.1 This state requires no immediate action, but just some more attention. When the delay between messages on the parser thread exceeds 1 second, DEBUG messages become WARNING ones:: WARNING:root:Packet burst: the reader thread priority is increased, beware of delays on netlink calls Counters: delta=2525 qsize=213536 delay=3 This state means, that almost all the CPU resources are dedicated to the reader thread. It doesn't mean, that the reader thread consumes 100% CPU -- it means, that the CPU is reserved for the case of more intensive bursts. The library will return to the normal state only when the broadcast storm will be over, and then the CPU will be 100% loaded with the parser for some time, when it will process all the messages queued so far. when async I/O doesn't help --------------------------- Sometimes, even turning async I/O doesn't fix ENOBUF. Mostly it means, that in this particular case the Python performance is not enough even to read and store the raw data from the socket. There is no workaround for such cases, except of using something not Python-based. One can still play around with SO_RCVBUF socket option, but it doesn't help much. So keep it in mind, and if you expect massive broadcast Netlink storms, perform stress testing prior to deploy a solution in the production. classes ------- ''' import asyncio import errno import logging import os import random import struct from dataclasses import dataclass from socket import SO_RCVBUF, SO_SNDBUF, SOCK_DGRAM, SOL_SOCKET from typing import Optional from pyroute2 import config, netns from pyroute2.common import AddrPool, basestring, msg_done from pyroute2.config import AF_NETLINK from pyroute2.netlink import ( NETLINK_ADD_MEMBERSHIP, NETLINK_DROP_MEMBERSHIP, NETLINK_EXT_ACK, NETLINK_GENERIC, NETLINK_GET_STRICT_CHK, NETLINK_LISTEN_ALL_NSID, NLM_F_ACK, NLM_F_APPEND, NLM_F_ATOMIC, NLM_F_CREATE, NLM_F_DUMP, NLM_F_ECHO, NLM_F_EXCL, NLM_F_REPLACE, NLM_F_REQUEST, NLM_F_ROOT, SOL_NETLINK, ) from pyroute2.netlink.core import ( AsyncCoreSocket, CoreConfig, CoreDatagramProtocol, CoreSocketSpec, SyncAPI, ) from pyroute2.netlink.exceptions import ChaoticException, NetlinkError from pyroute2.netlink.marshal import Marshal from pyroute2.requests.main import RequestFilter log = logging.getLogger(__name__) class CompileContext: def __init__(self, netlink_socket): self.netlink_socket = netlink_socket self.netlink_socket.compiled = [] def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): self.close() def close(self): self.netlink_socket.compiled = None # 8<----------------------------------------------------------- # Singleton, containing possible modifiers to the NetlinkSocket # bind() call. # # Normally, you can open only one netlink connection for one # process, but there is a hack. Current PID_MAX_LIMIT is 2^22, # so we can use the rest to modify the pid field. # # See also libnl library, lib/socket.c:generate_local_port() sockets = AddrPool(minaddr=0x0, maxaddr=0x3FF, reverse=True) # 8<----------------------------------------------------------- @dataclass class NetlinkConfig(CoreConfig): family: int = NETLINK_GENERIC port: Optional[int] = None pid: Optional[int] = None fileno: Optional[int] = None rcvbuf: int = 1048576 sndbuf: int = 1048576 all_ns: bool = False ext_ack: bool = False strict_check: bool = False nlm_echo: bool = False class NetlinkSocketSpecFilter(RequestFilter): def set_target(self, context, value): if 'target' in context: return {'target': context['target']} return {'target': value} def set_netns(self, context, value): if 'target' in context: return {'netns': value} return {'target': value, 'netns': value} def set_pid(self, context, value): if value is None: return {'pid': os.getpid() & 0x3FFFFF, 'port': context['port']} elif value == 0: return {'pid': os.getpid(), 'port': 0} else: return {'pid': value, 'port': 0} def set_port(self, context, value): if isinstance(value, int): return {'port': value, 'epid': context['pid'] + (value << 22)} class NetlinkSocketSpec(CoreSocketSpec): defaults = { 'pid': 0, 'epid': 0, 'port': 0, 'closed': False, 'uname': config.uname, 'use_socket': False, 'event_loop': 'none', 'transport_mode': 'none', } status_filters = [NetlinkSocketSpecFilter] class AsyncNetlinkSocket(AsyncCoreSocket): ''' Netlink socket ''' def __init__( self, family=NETLINK_GENERIC, 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=0, nlm_echo=False, use_socket=None, netns=None, flags=os.O_CREAT, libc=None, use_event_loop=None, telemetry=None, ): # 8<----------------------------------------- self.spec = NetlinkSocketSpec( NetlinkConfig( family=family, port=port, pid=pid, fileno=fileno, sndbuf=sndbuf, rcvbuf=rcvbuf, rcvsize=rcvsize, all_ns=all_ns, target=target, ext_ack=ext_ack, strict_check=strict_check, groups=groups, nlm_echo=nlm_echo, tag_field='sequence_number', netns=netns, flags=flags, use_libc=libc is not None, use_socket=use_socket is not None, use_event_loop=use_event_loop is not None, telemetry=telemetry, ) ) # TODO: merge capabilities to self.status self.capabilities = { 'create_bridge': config.kernel > [3, 2, 0], 'create_bond': config.kernel > [3, 2, 0], 'create_dummy': True, 'provide_master': config.kernel[0] > 2, } super().__init__( libc=libc, use_socket=use_socket, use_event_loop=use_event_loop ) self.marshal = Marshal() self.request_proxy = None self.batch = None async def setup_endpoint(self): if getattr(self.local, 'transport', None) is not None: return self.local.transport, self.local.protocol = ( await self.event_loop.create_datagram_endpoint( lambda: CoreDatagramProtocol( self.connection_lost, self.enqueue, self._error_event, self.status, ), sock=self.socket, ) ) def setup_socket(self): if self.status['use_socket']: return self.use_socket sock = netns.create_socket( netns=self.spec['netns'], family=AF_NETLINK, socket_type=SOCK_DGRAM, proto=self.spec['family'], # netlink family = socket proto fileno=self.spec['fileno'], flags=self.spec['flags'], libc=self.libc, ) sock.setsockopt(SOL_SOCKET, SO_SNDBUF, self.status['sndbuf']) sock.setsockopt(SOL_SOCKET, SO_RCVBUF, self.status['rcvbuf']) if self.status['ext_ack']: sock.setsockopt(SOL_NETLINK, NETLINK_EXT_ACK, 1) if self.status['all_ns']: sock.setsockopt(SOL_NETLINK, NETLINK_LISTEN_ALL_NSID, 1) if self.status['strict_check']: sock.setsockopt(SOL_NETLINK, NETLINK_GET_STRICT_CHK, 1) return sock @property def uname(self): return self.status['uname'] @property def groups(self): return self.status['groups'] @property def pid(self): return self.status['pid'] @property def port(self): return self.status['port'] @property def epid(self): return self.status['epid'] @property def target(self): return self.status['target'] def _sync_bind(self, groups=0, pid=None, kwarg): self.spec['groups'] = groups # if we have pre-defined port, use it strictly self.spec['pid'] = pid if pid is None: for port in range(20, 200): try: self.spec['port'] = port self.socket.bind( (self.status['epid'], self.status['groups']) ) break except Exception as e: # create a new underlying socket -- on kernel 4 # one failed bind() makes the socket useless log.debug(e) else: raise KeyError('no free address available') async def bind(self, groups=0, pid=None, kwarg): ''' Bind the socket to given multicast groups, using given pid. - If pid is None, use automatic port allocation - If pid == 0, use process' pid - If pid == <int>, use the value instead of pid ''' await self.setup_endpoint() self._sync_bind(groups, pid, *kwarg) def add_membership(self, group): self.socket.setsockopt(SOL_NETLINK, NETLINK_ADD_MEMBERSHIP, group) def drop_membership(self, group): self.socket.setsockopt(SOL_NETLINK, NETLINK_DROP_MEMBERSHIP, group) def enqueue(self, data, addr): # calculate msg_seq tag = struct.unpack_from('I', data, 8)[0] return self.msg_queue.put_nowait(tag, data) def compile(self): return CompileContext(self) def make_request_type(self, command, command_map): if isinstance(command, basestring): return (lambda x: (x[0], self.make_request_flags(x[1])))( command_map[command] ) elif isinstance(command, int): return command, self.make_request_flags('create') elif isinstance(command, (list, tuple)): return command else: raise TypeError('allowed command types: int, str, list, tuple') def make_request_flags(self, mode): flags = { 'dump': NLM_F_REQUEST \| NLM_F_DUMP, 'get': NLM_F_REQUEST \| NLM_F_ACK, 'req': NLM_F_REQUEST \| NLM_F_ACK, 'put': NLM_F_REQUEST \| NLM_F_CREATE, } flags['create'] = flags['req'] \| NLM_F_CREATE \| NLM_F_EXCL flags['append'] = flags['req'] \| NLM_F_CREATE \| NLM_F_APPEND flags['change'] = flags['req'] \| NLM_F_REPLACE flags['replace'] = flags['change'] \| NLM_F_CREATE return flags[mode] \| ( NLM_F_ECHO if (self.status['nlm_echo'] and mode not in ('get', 'dump')) else 0 ) async def put( self, msg, msg_type, msg_flags=NLM_F_REQUEST, addr=(0, 0), msg_seq=0, msg_pid=None, ): request = NetlinkRequest( self, msg, msg_type=msg_type, msg_flags=msg_flags, msg_seq=msg_seq, msg_pid=msg_pid, ) await request.send() return request async def nlm_request_batch(self, msgs, noraise=False): expected_responses = [] data = b'' for message in msgs: request = NetlinkRequest(self, message) await request.prepare() data += request.msg.data if (request.msg['header']['flags'] & NLM_F_ACK) or ( request.msg['header']['flags'] & NLM_F_DUMP ): expected_responses.append(request) else: request.cleanup() self.send(data) for request in expected_responses: try: async for msg in request.response(): yield msg except NetlinkError: if not noraise: raise async def nlm_request( self, msg, msg_type, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP, terminate=None, callback=None, parser=None, ): request = NetlinkRequest( self, msg, terminate=terminate, callback=callback ) request.msg['header']['type'] = msg_type request.msg['header']['flags'] = msg_flags await request.send() return request.response() class NetlinkRequest: # request flags flags = { 'dump': NLM_F_REQUEST \| NLM_F_DUMP, 'root': NLM_F_REQUEST \| NLM_F_ROOT \| NLM_F_ATOMIC, 'get': NLM_F_REQUEST \| NLM_F_ACK, 'req': NLM_F_REQUEST \| NLM_F_ACK, } flags['create'] = flags['req'] \| NLM_F_CREATE \| NLM_F_EXCL flags['append'] = flags['req'] \| NLM_F_CREATE \| NLM_F_APPEND flags['change'] = flags['req'] \| NLM_F_REPLACE flags['replace'] = flags['change'] \| NLM_F_CREATE def __init__( self, sock, msg, command=None, command_map=None, dump_filter=None, request_filter=None, terminate=None, callback=None, parser=None, msg_type=None, msg_flags=None, msg_seq=None, msg_pid=None, ): self.sock = sock self.addr_pool = sock.addr_pool self.status = sock.status self.epid = sock.epid if msg_pid is None else msg_pid self.marshal = sock.marshal self.parser = parser # if not isinstance(msg, nlmsg): # msg_class = self.marshal.msg_map[msg_type] # msg = msg_class(msg) self.msg_seq = self.addr_pool.alloc() if msg_seq is None else msg_seq # prio 3: message object # prio 2: direct msg_type & msg_flags arguments # prio 1: command map if command_map is not None: msg_type, msg_flags = self.calculate_request_type( command, command_map, self.status['nlm_echo'] ) if msg_type is not None: msg['header']['type'] = msg_type if msg_flags is not None: msg['header']['flags'] = msg_flags # if there is no type & flags yet, set defaults # FIXME: collect usecases if msg['header'].get('type') is None: msg['header']['type'] = 0 if msg['header'].get('flags') is None: msg['header']['flags'] = NLM_F_REQUEST \| NLM_F_DUMP msg['header']['sequence_number'] = self.msg_seq msg['header']['pid'] = self.epid or os.getpid() msg.reset() # set fields if request_filter is not None: for field in msg.fields: msg[field[0]] = request_filter.get_value( field[0], default=0, mode='field' ) # attach NLAs for key, value in request_filter.items(): nla = type(msg).name2nla(key) if msg.valid_nla(nla) and value is not None: msg['attrs'].append([nla, value]) # extend with custom NLAs if 'attrs' in request_filter: msg['attrs'].extend(request_filter['attrs']) self.msg = msg self.dump_filter = dump_filter self.terminate = terminate self.callback = callback self.command = command @classmethod def calculate_request_type(cls, command, command_map, echo=False): if isinstance(command, basestring): return (lambda x: (x[0], cls.calculate_request_flags(x[1], echo)))( command_map[command] ) elif isinstance(command, int): return command, cls.calculate_request_flags('create', echo) elif isinstance(command, (list, tuple)): return command else: raise TypeError('allowed command types: int, str, list, tuple') @classmethod def calculate_request_flags(cls, mode, echo): return cls.flags[mode] \| ( NLM_F_ECHO if (echo and mode not in ('get', 'dump')) else 0 ) @staticmethod def match_one_message(dump_filter, msg): if hasattr(dump_filter, '__call__'): return dump_filter(msg) elif isinstance(dump_filter, dict): matches = [] for key in dump_filter: # get the attribute if not isinstance(key, (str, tuple)): continue value = msg.get(key) if value is not None and callable(dump_filter[key]): matches.append(dump_filter[key](value)) else: matches.append(dump_filter[key] == value) return all(matches) async def prepare(self): await self.sock.setup_endpoint() self.msg.encode() self.sock.msg_queue.ensure_tag(self.msg_seq) if self.parser is not None: self.marshal.seq_map[self.msg_seq] = self.parser def cleanup(self): self.addr_pool.free(self.msg_seq, ban=0xFF) self.sock.msg_queue.free_tag(self.msg_seq) if self.msg_seq in self.marshal.seq_map: self.marshal.seq_map.pop(self.msg_seq) async def proxy(self): if self.sock.batch is not None: self.sock.batch += self.msg.data await self.sock.msg_queue.put(self.msg_seq, msg_done(self.msg)) return True if self.sock.request_proxy is None: return False ret = self.sock.request_proxy.handle(self.msg) if ret == b'': return False await self.sock.msg_queue.put(self.msg_seq, ret) return True async def send(self): await self.prepare() if await self.proxy(): return len(self.msg.data) count = 0 exc = RuntimeError('Max attempts sending message') for count in range(30): try: return self.sock.send(self.msg.data) except NetlinkError as e: if e.code != errno.EBUSY: exc = e break log.warning(f'Error 16, retry {count}') await asyncio.sleep(0.3) except Exception as e: exc = e break self.cleanup() raise exc async def response(self): async for msg in self.sock.get( msg_seq=self.msg_seq, terminate=self.terminate, callback=self.callback, ): if self.dump_filter is not None and not self.match_one_message( self.dump_filter, msg ): continue for cr in self.sock.callbacks: try: if cr[0](msg): cr[1](msg, cr[2]) except Exception: log.warning("Callback fail: %{cr}") yield msg self.cleanup() class NetlinkSocket(SyncAPI): def __init__( self, family=NETLINK_GENERIC, 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=0, nlm_echo=False, netns=None, flags=os.O_CREAT, libc=None, use_socket=None, use_event_loop=None, telemetry=None, ): self.asyncore = AsyncNetlinkSocket( family=family, port=port, pid=pid, fileno=fileno, sndbuf=sndbuf, rcvbuf=rcvbuf, rcvsize=rcvsize, all_ns=all_ns, target=target, ext_ack=ext_ack, strict_check=strict_check, groups=groups, nlm_echo=nlm_echo, netns=netns, flags=flags, libc=libc, use_socket=use_socket, use_event_loop=use_event_loop, telemetry=telemetry, ) self.asyncore.local.keep_event_loop = True self.asyncore.status['event_loop'] = 'new' self.asyncore.event_loop.run_until_complete( self.asyncore.setup_endpoint() ) if self.asyncore.socket.fileno() == -1: raise OSError(9, 'Bad file descriptor') def bind(self, argv, kwarg): with self.lock: self.asyncore._check_tid(tag='bind', level=logging.WARN) self.asyncore.local.keep_event_loop = True self._run_with_cleanup( self.asyncore.bind, 'nl-bind', argv, *kwarg ) self.asyncore._register_loop_ref() def put( self, msg, msg_type, msg_flags=NLM_F_REQUEST, addr=(0, 0), msg_seq=None, msg_pid=None, ): if msg is None: msg_class = self.marshal.msg_map[msg_type] msg = msg_class() return self._run_with_cleanup( self.asyncore.put, 'nl-put', msg, msg_type, msg_flags, addr, msg_seq, msg_pid, ) def nlm_request_batch(self, msgs, noraise=False): async def collect_data(): return [ x async for x in self.asyncore.nlm_request_batch(msgs, noraise) ] return self._run_with_cleanup(collect_data, 'nl-req-batch') def nlm_request( self, msg, msg_type, msg_flags=NLM_F_REQUEST \| NLM_F_DUMP, terminate=None, callback=None, parser=None, ): async def collect_data(): return [ x async for x in await self.asyncore.nlm_request( msg, msg_type, msg_flags, terminate, callback, parser ) ] return self._run_with_cleanup(collect_data, 'nl-req') def get(self, msg_seq=0, terminate=None, callback=None, noraise=False): '''Sync wrapper for async_get().''' async def collect_data(): return [ i async for i in self.asyncore.get( msg_seq, terminate, callback, noraise ) ] return self._run_with_cleanup(collect_data, 'nl-get') class ChaoticNetlinkSocket(NetlinkSocket): success_rate = 1 def __init__(self, argv, *kwarg): self.success_rate = kwarg.pop('success_rate', 0.7) super(ChaoticNetlinkSocket, self).__init__(argv, *kwarg) def get(self, argv, *kwarg): if random.random() > self.success_rate: raise ChaoticException() return super(ChaoticNetlinkSocket, self).get(argv, **kwarg)	24.17 KB	2025-05-25 12:26
📁 rtnl	-	2025-05-25 12:26
📁 taskstats	-	2025-05-25 12:26
📁 uevent	-	2025-05-25 12:26

Edit File:

Terminal