Wala shell

Advanced PHP File Manager

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

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:26
📄 __init__.py	0 B	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 address.py import ipaddress from socket import AF_INET, AF_INET6 from pyroute2.common import dqn2int, get_address_family, getbroadcast from .common import Index, IPRouteFilter, NLAKeyTransform class AddressFieldFilter(Index, NLAKeyTransform): _nla_prefix = 'IFA_' def get_flags(self, context, mode): if mode == 'field': return 0 return context['flags'] def set_prefixlen(self, context, value): if isinstance(value, str): if '.' in value: value = dqn2int(value) value = int(value) return {'prefixlen': value} def set_address(self, context, value): if not value: return {} ret = {'address': value} if isinstance(value, str): addr_spec = value.split('/') ret['address'] = addr_spec[0] if len(addr_spec) > 1: ret.update(self.set_prefixlen(context, addr_spec[1])) if ':' in ret['address']: ret['address'] = ipaddress.ip_address( ret['address'] ).compressed return ret def set_local(self, context, value): if not value: return {} return {'local': value} def set_mask(self, context, value): return {'prefixlen': value} def _cacheinfo(self, key, value): return {key: value, 'cacheinfo': {}} def set_preferred_lft(self, context, value): return self._cacheinfo('preferred', value) def set_preferred(self, context, value): return self._cacheinfo('preferred', value) def set_valid_lft(self, context, value): return self._cacheinfo('valid', value) def set_valid(self, context, value): return self._cacheinfo('valid', value) class AddressIPRouteFilter(IPRouteFilter): def set_cacheinfo(self, context, value): cacheinfo = value.copy() if self.command != 'dump': for i in ('preferred', 'valid'): cacheinfo[f'ifa_{i}'] = context.get(i, pow(2, 32) - 1) return {'cacheinfo': cacheinfo} return {} def set_broadcast(self, context, value): ret = {} if self.command != 'dump' and isinstance(value, bool): keys = set(context.keys()) if value and 'address' in keys and 'prefixlen' in keys: if 'family' in keys: family = context['family'] else: ret['family'] = family = get_address_family( context['address'] ) ret['broadcast'] = getbroadcast( context['address'], context['prefixlen'], family ) else: ret['broadcast'] = value return ret def finalize(self, context): if self.command != 'dump': if 'cacheinfo' in context: context.pop('preferred', None) context.pop('valid', None) if ( context['cacheinfo']['ifa_valid'] < context['cacheinfo']['ifa_preferred'] ): raise ValueError('preferred_lft is greater than valid_lft') if 'family' not in context and 'address' in context: context['family'] = get_address_family(context['address']) if 'prefixlen' not in context: if context['family'] == AF_INET: context['prefixlen'] = 32 elif context['family'] == AF_INET6: context['prefixlen'] = 128 if ( 'local' not in context and 'address' in context and context['family'] == AF_INET ): # inject IFA_LOCAL, if family is AF_INET and # IFA_LOCAL is not set context['local'] = context['address']	3.82 KB	2025-05-25 12:26
📄 bridge.py from functools import reduce from pyroute2.config import AF_BRIDGE from pyroute2.netlink.rtnl.ifinfmsg import ifinfmsg, protinfo_bridge from .common import Index, IPRouteFilter, NLAKeyTransform PREFIXLEN = len(protinfo_bridge.prefix) class BridgeFieldFilter(Index, NLAKeyTransform): _nla_prefix = ifinfmsg.prefix class BridgeIPRouteFilter(IPRouteFilter): def build_vlan_info_spec(self, orig_spec): range_vids = [int(i) for i in str(orig_spec['vid']).split('-')] if len(range_vids) == 2: if 0 < int(range_vids[0]) < range_vids[1] < 4095: new_spec = [] new_spec.append( { 'vid': range_vids[0], 'flags': self.convert_flags('range_begin'), } ) new_spec.append( { 'vid': range_vids[1], 'flags': self.convert_flags('range_end'), } ) return new_spec elif len(range_vids) == 1: if 0 < range_vids[0] < 4095: # PVID? if 'pvid' in orig_spec.keys(): if orig_spec['pvid']: orig_spec['flags'] = self.convert_flags( ['pvid', 'untagged'] ) del orig_spec['pvid'] # Make sure the vid is an int. orig_spec['vid'] = range_vids[0] return [orig_spec] return [] def build_vlan_tunnel_info_spec(self, orig_spec): # vlan_tunnel_info uses the same format as vlan_info, # just adds tunnel_id. vlan_info_spec = self.build_vlan_info_spec(orig_spec) range_ids = [int(i) for i in str(orig_spec['id']).split('-')] if len(range_ids) == 2 and len(vlan_info_spec) == 2: if 0 < range_ids[0] < range_ids[1] < 16777215: # vid to id mapping range must be the same length if ( vlan_info_spec[1]['vid'] - vlan_info_spec[0]['vid'] == range_ids[1] - range_ids[0] ): vlan_info_spec[0]['id'] = range_ids[0] vlan_info_spec[1]['id'] = range_ids[1] return [ self.create_nla_spec(vlan_info_spec[0]), self.create_nla_spec(vlan_info_spec[1]), ] elif len(range_ids) == 1 and len(vlan_info_spec) == 1: if 0 < range_ids[0] < 4095: vlan_info_spec[0]['id'] = range_ids[0] # Delete flags because vlan_tunnel_info doesn't seem # to use them, except for the RANGE. try: del vlan_info_spec[0]['flags'] except KeyError: pass return [self.create_nla_spec(vlan_info_spec[0])] return [] def create_nla_spec(self, spec): attrs = [] for key in spec.keys(): nla = ifinfmsg.af_spec_bridge.vlan_tunnel_info.name2nla(key) attrs.append([nla, spec[key]]) return {'attrs': attrs} def convert_flags(self, flags): if isinstance(flags, int): return flags elif isinstance(flags, str): return ifinfmsg.af_spec_bridge.vlan_info.names2flags([flags]) elif isinstance(flags, list): return ifinfmsg.af_spec_bridge.vlan_info.names2flags(flags) return 0 def build_spec(self, orig_spec): if 'vid' in orig_spec.keys(): if 'id' in orig_spec.keys(): return self.build_vlan_tunnel_info_spec(orig_spec) else: return self.build_vlan_info_spec(orig_spec) return [] def finalize(self, context): if self.command != 'dump': if 'IFLA_AF_SPEC' not in context: context['IFLA_AF_SPEC'] = {'attrs': []} for key in ('vlan_info', 'vlan_tunnel_info'): if key in context: nla = ifinfmsg.af_spec_bridge.name2nla(key) new_spec = self.build_spec(context[key]) for spec in new_spec: context['IFLA_AF_SPEC']['attrs'].append([nla, spec]) try: del context[key] except KeyError: pass for key in ('mode', 'vlan_flags'): if key in context: nla = ifinfmsg.af_spec_bridge.name2nla(key) context['IFLA_AF_SPEC']['attrs'].append( [nla, context[key]] ) try: del context[key] except KeyError: pass class BridgePortFieldFilter(IPRouteFilter): _nla_prefix = ifinfmsg.prefix _nla_names = [x[0] for x in protinfo_bridge.nla_map] # allow lowercase names w/o prefixes _nla_names = reduce( lambda x, y: x + y, [[x, x[PREFIXLEN:].lower()] for x in _nla_names] ) # allow tuple ids for the dump_filter lookup _nla_names = reduce( lambda x, y: x + y, [[x, ('protinfo', x)] for x in _nla_names] ) allowed = list(_nla_names) allowed.append('attrs') allowed.append('family') allowed.append('index') def finalize(self, context): context['family'] = AF_BRIDGE keys = tuple(context.keys()) if self.command in ('show', 'dump'): for key in keys: if isinstance(key, str) and key in self._nla_names: value = context[key] if isinstance(value, bool): value = int(value) context[('protinfo', key)] = value del context[key] return attrs = [] for key in keys: attrs.append((protinfo_bridge.name2nla(key), context[key])) context['attrs'] = [('IFLA_PROTINFO', {'attrs': attrs}, 0x8000)]	5.99 KB	2025-05-25 12:26
📄 common.py import ipaddress import json from collections import OrderedDict from socket import AF_INET, AF_INET6 from pyroute2.common import AF_MPLS, dqn2int, get_address_family class MPLSTarget(OrderedDict): def __init__(self, prime=None): super(MPLSTarget, self).__init__() if prime is None: prime = {} elif isinstance(prime, str): prime = {'label': int(prime)} elif isinstance(prime, int): prime = {'label': prime} elif isinstance(prime, dict): pass else: raise TypeError() self['label'] = prime.get('label', 16) self['tc'] = prime.get('tc', 0) self['bos'] = prime.get('bos', 1) self['ttl'] = prime.get('ttl', 0) def __eq__(self, right): return ( isinstance(right, (dict, MPLSTarget)) and self['label'] == right.get('label', 16) and self['tc'] == right.get('tc', 0) and self['bos'] == right.get('bos', 1) and self['ttl'] == right.get('ttl', 0) ) def __repr__(self): return repr(dict(self)) class IPTargets: add_defaults = True def parse_target(self, key, context, value): ret = {key: value} set_full_mask = False if isinstance(value, int): return {'family': AF_MPLS, key: MPLSTarget(value)} if isinstance(value, (list, tuple)): targets = [] target = None for spec in value: target = MPLSTarget(spec) target['bos'] = 0 targets.append(target) if target: target['bos'] = 1 return {'family': AF_MPLS, key: targets} else: return {} if isinstance(value, dict): if value.get('label'): return {'family': AF_MPLS, key: MPLSTarget(value)} # do not overwrite message family for IP VIA return {key: value} try: return self.parse_target(key, context, json.loads(value)) except (json.JSONDecodeError, TypeError): pass if isinstance(value, str): if value == '': return {} labels = value.split('/') if len(labels) > 1: # MPLS label stack simple syntax? e.g.: 16/24, 200/300 etc. try: return self.parse_target(key, context, labels) except ValueError: pass # only simple IP targets are left value, prefixlen = labels ret[key] = value if self.add_defaults: if prefixlen.find('.') > 0: ret[f'{key}_len'] = dqn2int(prefixlen, AF_INET) elif prefixlen.find(':') >= 0: ret[f'{key}_len'] = dqn2int(prefixlen, AF_INET6) else: ret[f'{key}_len'] = int(prefixlen) else: if ( self.add_defaults and key in ('dst', 'src') and f'{key}_len' not in context ): set_full_mask = True if self.add_defaults: ret['family'] = get_address_family(value) if ret['family'] == AF_INET6: ret[key] = ipaddress.ip_address(value).compressed if set_full_mask: if ret['family'] == AF_INET6: ret[f'{key}_len'] = 128 elif ret['family'] == AF_INET: ret[f'{key}_len'] = 32 return ret def set_dst(self, context, value): if value in ('', 'default'): return {'dst': ''} elif value in ('0', '0.0.0.0'): return {'dst': '', 'family': AF_INET} elif value in ('::', '::/0'): return {'dst': '', 'family': AF_INET6} return self.parse_target('dst', context, value) def set_src(self, context, value): return self.parse_target('src', context, value) def set_via(self, context, value): return self.parse_target('via', context, value) def set_newdst(self, context, value): return self.parse_target('newdst', context, value) def set_gateway(self, context, value): return self.parse_target('gateway', context, value) class Index: def set_index(self, context, value): if isinstance(value, (list, tuple)): value = value[0] return {'index': value} class IPRouteFilter: def __init__(self, command): self.command = command def policy(self, key): if self.command == 'add' and key in ('tso_max_segs', 'tso_max_size'): return False return True class NLAKeyTransform: _nla_prefix = '' def key_transform(self, key): if isinstance(key, str) and key.startswith(self._nla_prefix): key = key[len(self._nla_prefix) :].lower() return key	4.96 KB	2025-05-25 12:26
📄 link.py from pyroute2.netlink.rtnl.ifinfmsg import IFF_NOARP, IFF_UP, ifinfmsg from pyroute2.netlink.rtnl.ifinfmsg.plugins.vlan import flags as vlan_flags from .common import Index, IPRouteFilter, NLAKeyTransform class LinkFieldFilter(Index, NLAKeyTransform): _nla_prefix = 'IFLA_' def _link(self, key, context, value): if isinstance(value, dict): return {key: value['index']} return {key: value} def set_vxlan_link(self, context, value): return self._link('vxlan_link', context, value) def set_link(self, context, value): return self._link('link', context, value) def set_master(self, context, value): return self._link('master', context, value) def set_address(self, context, value): if isinstance(value, str): # lower the case if not value.islower(): value = value.lower() # convert xxxx.xxxx.xxxx to xx:xx:xx:xx:xx:xx if len(value) == 14 and value[4] == value[9] == '.': value = ':'.join( [':'.join((x[:2], x[2:])) for x in value.split('.')] ) return {'address': value} def set_carrier(self, context, value): return {} def set_carrier_changes(self, context, value): return {} def set_info_slave_kind(self, context, value): return {} def set_mask(self, context, value): return {'change': value} def set_info_kind(self, context, value): return {'kind': value} class LinkIPRouteFilter(IPRouteFilter): def set_altname(self, context, value): if self.command in ('property_add', 'property_del'): if not isinstance(value, (list, tuple, set)): value = [value] return { 'IFLA_PROP_LIST': { 'attrs': [ ('IFLA_ALT_IFNAME', alt_ifname) for alt_ifname in value ] } } else: return {'IFLA_ALT_IFNAME': value} def set_xdp_fd(self, context, value): return {'xdp': {'attrs': [('IFLA_XDP_FD', value)]}} def set_vf(self, context, value): return {'IFLA_VFINFO_LIST': self.get_vf(value)} def set_state(self, context, value): ret = {} if self.command == 'dump': return {'state': value} if value == 'up': ret['flags'] = (context.get('flags', 0) or 0) \| IFF_UP ret['change'] = (context.get('change', 0) or 0) \| IFF_UP return ret def set_arp(self, context, value): ret = {} if not value: ret['flags'] = (context.get('flags', 0) or 0) \| IFF_NOARP ret['change'] = (context.get('change', 0) or 0) \| IFF_NOARP return ret def set_noarp(self, context, value): ret = {} if value: ret['flags'] = (context.get('flags', 0) or 0) \| IFF_NOARP ret['change'] = (context.get('change', 0) or 0) \| IFF_NOARP return ret def finalize(self, context): # set interface type specific attributes self.kind = context.pop('kind', None) if self.kind is None: return # load specific NLA names self.specific = {} cls = ifinfmsg.ifinfo.data_map.get(self.kind, None) if cls is not None: prefix = cls.prefix or 'IFLA_' for nla, _ in cls.nla_map: self.specific[nla] = nla self.specific[nla[len(prefix) :].lower()] = nla if self.command == 'dump': context[('linkinfo', 'kind')] = self.kind for key, value in tuple(context.items()): if key in self.specific: context[('linkinfo', 'data', key)] = value try: del context[key] except KeyError: pass return # get common ifinfmsg NLAs self.common = [] for key, _ in ifinfmsg.nla_map: self.common.append(key) self.common.append(key[len(ifinfmsg.prefix) :].lower()) self.common.append('family') self.common.append('ifi_type') self.common.append('index') self.common.append('flags') self.common.append('change') for key in ('index', 'change', 'flags'): if key not in context: context[key] = 0 linkinfo = {'attrs': []} self.linkinfo = linkinfo['attrs'] self._info_data = None self._info_slave_data = None context['IFLA_LINKINFO'] = linkinfo self.linkinfo.append(['IFLA_INFO_KIND', self.kind]) # flush deferred NLAs for key, value in tuple(context.items()): if self.push_specific(key, value): try: del context[key] except KeyError: pass def push_specific(self, key, value): # FIXME: vlan hack if self.kind == 'vlan': if key == 'vlan_flags': if isinstance(value, (list, tuple)): if len(value) == 2 and all( (isinstance(x, int) for x in value) ): value = {'flags': value[0], 'mask': value[1]} else: ret = 0 for x in value: ret \|= vlan_flags.get(x, 1) value = {'flags': ret, 'mask': ret} elif isinstance(value, int): value = {'flags': value, 'mask': value} elif isinstance(value, str): value = vlan_flags.get(value, 1) value = {'flags': value, 'mask': value} elif not isinstance(value, dict): raise ValueError() elif key in ('vlan_egress_qos', 'vlan_ingress_qos'): if isinstance(value, dict) and {'from', 'to'} == value.keys(): value = {'attrs': (('IFLA_VLAN_QOS_MAPPING', value),)} # the kind is known: lookup the NLA if key in self.specific: # FIXME: slave hack if self.kind.endswith('_slave'): self.info_slave_data.append((self.specific[key], value)) else: self.info_data.append((self.specific[key], value)) return True elif key == 'peer' and self.kind == 'veth': # FIXME: veth hack if isinstance(value, dict): attrs = [] for k, v in value.items(): attrs.append([ifinfmsg.name2nla(k), v]) else: attrs = [['IFLA_IFNAME', value]] nla = ['VETH_INFO_PEER', {'attrs': attrs}] self.info_data.append(nla) return True elif key == 'mode': # FIXME: ipvlan / tuntap / bond hack if self.kind == 'tuntap': nla = ['IFTUN_MODE', value] else: nla = ['IFLA_%s_MODE' % self.kind.upper(), value] self.info_data.append(nla) return True return False @property def info_data(self): if self._info_data is None: info_data = ('IFLA_INFO_DATA', {'attrs': []}) self._info_data = info_data[1]['attrs'] self.linkinfo.append(info_data) return self._info_data @property def info_slave_data(self): if self._info_slave_data is None: info_slave_data = ('IFLA_INFO_SLAVE_DATA', {'attrs': []}) self._info_slave_data = info_slave_data[1]['attrs'] self.linkinfo.append(info_slave_data) return self._info_slave_data def get_vf(self, spec): vflist = [] if not isinstance(spec, (list, tuple)): spec = (spec,) for vf in spec: vfcfg = [] # pop VF index vfid = vf.pop('vf') # mandatory # pop VLAN spec vlan = vf.pop('vlan', None) # optional if isinstance(vlan, int): vfcfg.append(('IFLA_VF_VLAN', {'vf': vfid, 'vlan': vlan})) elif isinstance(vlan, dict): vlan['vf'] = vfid vfcfg.append(('IFLA_VF_VLAN', vlan)) elif isinstance(vlan, (list, tuple)): vlist = [] for vspec in vlan: vspec['vf'] = vfid vlist.append(('IFLA_VF_VLAN_INFO', vspec)) vfcfg.append(('IFLA_VF_VLAN_LIST', {'attrs': vlist})) # pop rate spec rate = vf.pop('rate', None) # optional if rate is not None: rate['vf'] = vfid vfcfg.append(('IFLA_VF_RATE', rate)) # create simple VF attrs for attr in vf: vfcfg.append( ( ifinfmsg.vflist.vfinfo.name2nla(attr), {'vf': vfid, attr: vf[attr]}, ) ) vflist.append(('IFLA_VF_INFO', {'attrs': vfcfg})) return {'attrs': vflist}	8.99 KB	2025-05-25 12:26
📄 main.py ''' General request and RTNL object data filters. ''' import abc import weakref from collections import ChainMap class RequestFilter(abc.ABC): pass class RequestProcessor(dict): field_filters = tuple() mark = tuple() context = None combined = None parameters = None def __init__(self, context=None, prime=None): self.reset_filters() self.reset_mark() self.parameters = {} prime = {} if prime is None else prime self.context = ( context if isinstance(context, (dict, weakref.ProxyType)) else {} ) self.combined = ChainMap(self, self.context) if isinstance(prime, dict): self.update(prime) def __setitem__(self, key, value): if value is None: return new_data = self.filter(key, value) op = 'set' # if isinstance(new_data, tuple): op, new_data = new_data elif not isinstance(new_data, dict): raise ValueError('invalid new_data type') # if op == 'set': if key in self: super().__delitem__(key) for nkey, nvalue in new_data.items(): if nkey in self: super().__delitem__(nkey) super().__setitem__(nkey, nvalue) return # if op == 'patch': for nkey, nvalue in new_data.items(): if nkey not in self: self[nkey] = [] self[nkey].extend(nvalue) return # raise RuntimeError('error applying new values') def reset_filters(self): self.field_filters = [] def set_parameter(self, name, value): self.parameters[name] = value def reset_mark(self): self.mark = [] def items(self): for key, value in super().items(): if key not in self.mark: yield key, value def get_value(self, key, default=None, mode=None): for field_filter in self.field_filters: getter = getattr(field_filter, f'get_{key}', None) if getter is not None: return getter(self, mode) self.mark.append(key) return self.get(key, default) def filter(self, key, value): job = {key: value} ret = None for field_filter in self.field_filters: for k, v in tuple(job.items()): if hasattr(field_filter, 'key_transform'): k = field_filter.key_transform(k) if ( hasattr(field_filter, 'allowed') and k not in field_filter.allowed ): return {} if hasattr(field_filter, 'policy') and not field_filter.policy( k ): return {} setter = getattr(field_filter, f'set_{k}', None) if setter is not None: if ret is None: ret = {} ret.update(setter(ChainMap(self.combined, ret), v)) if ret is not None: job = ret return ret if ret is not None else {key: value} def update(self, prime): for key, value in tuple(prime.items()): self[key] = value def add_filter(self, field_filter): self.field_filters.append(field_filter) return self def finalize(self): self.update(self) for field_filter in self.field_filters: if hasattr(field_filter, 'finalize'): field_filter.finalize(self.combined) return self	3.6 KB	2025-05-25 12:26
📄 neighbour.py from pyroute2.common import get_address_family from pyroute2.netlink.rtnl import ndmsg from .common import Index, IPRouteFilter class NeighbourFieldFilter(Index): def set_index(self, context, value): return { 'ifindex': super(NeighbourFieldFilter, self).set_index( context, value )['index'] } def set_ifindex(self, context, value): return self.set_index(context, value) def _state(self, value): if isinstance(value, str): value = ndmsg.states_a2n(value) return {'state': value} def set_nud(self, context, value): return self._state(value) def set_state(self, context, value): return self._state(value) def set_dst(self, context, value): if value: return {'dst': value} else: return {} class NeighbourIPRouteFilter(IPRouteFilter): def finalize(self, context): if self.command not in ('dump', 'get'): if 'state' not in context: context['state'] = ndmsg.NUD_PERMANENT if 'dst' in context and 'family' not in context: context['family'] = get_address_family(context['dst'])	1.18 KB	2025-05-25 12:26
📄 netns.py from .common import NLAKeyTransform class NetNSFieldFilter(NLAKeyTransform): _nla_prefix = 'NSINFO_'	107 B	2025-05-25 12:26
📄 probe.py from socket import AF_INET from .common import NLAKeyTransform class ProbeFieldFilter(NLAKeyTransform): _nla_prefix = 'PROBE_' def finalize(self, context): if 'family' not in context: context['family'] = AF_INET if 'proto' not in context: context['proto'] = 1 if 'port' not in context: context['port'] = 0 if 'dst_len' not in context: context['dst_len'] = 32 if 'kind' not in context: context['kind'] = 'ping' if 'num' not in context: context['num'] = 1 if 'timeout' not in context: context['timeout'] = 1	658 B	2025-05-25 12:26
📄 route.py from socket import AF_INET, AF_INET6 from pyroute2.common import AF_MPLS from pyroute2.netlink.rtnl import encap_type, rt_proto, rt_scope, rt_type from pyroute2.netlink.rtnl.rtmsg import IP6_RT_PRIO_USER, LWTUNNEL_ENCAP_MPLS from pyroute2.netlink.rtnl.rtmsg import nh as nh_header from pyroute2.netlink.rtnl.rtmsg import rtmsg from .common import IPRouteFilter, IPTargets, MPLSTarget, NLAKeyTransform encap_types = {'mpls': 1, AF_MPLS: 1, 'seg6': 5, 'bpf': 6, 'seg6local': 7} class RouteFieldFilter(IPTargets, NLAKeyTransform): _nla_prefix = 'RTA_' def __init__(self, add_defaults=True): self.add_defaults = add_defaults def index(self, key, context, value): if isinstance(value, (list, tuple)): value = value[0] return {key: value} def set_oif(self, context, value): return self.index('oif', context, value) def set_iif(self, context, value): return self.index('iif', context, value) def set_family(self, context, value): if value == AF_MPLS: return {'family': AF_MPLS, 'dst_len': 20, 'table': 254, 'type': 1} return {'family': value} def set_priority(self, context, value): ''' In the kernel: .. code-block:: c static int inet6_rtm_newroute(...) { ... if (cfg.fc_metric == 0) cfg.fc_metric = IP6_RT_PRIO_USER; ... } ''' if context.get('family') == AF_INET6 and value == 0: return {'priority': IP6_RT_PRIO_USER} return {'priority': value} def set_flags(self, context, value): if context.get('family') == AF_MPLS: return {} if isinstance(value, (list, tuple, str)): return {'flags': rtmsg.names2flags(value)} return {'flags': value} def set_encap(self, context, value): # FIXME: planned for the next refactoring cycle if isinstance(value, dict) and value.get('type') == 'mpls': na = [] target = None labels = value.get('labels', []) if isinstance(labels, (dict, int)): labels = [labels] if isinstance(labels, str): labels = labels.split('/') for label in labels: target = MPLSTarget(label) target['bos'] = 0 na.append(target) target['bos'] = 1 return {'encap_type': LWTUNNEL_ENCAP_MPLS, 'encap': na} return {'encap': value} def set_scope(self, context, value): if isinstance(value, str): return {'scope': rt_scope[value]} return {'scope': value} def set_proto(self, context, value): if isinstance(value, str): return {'proto': rt_proto[value]} return {'proto': value} def set_encap_type(self, context, value): if isinstance(value, str): return {'encap_type': encap_type[value]} return {'encap_type': value} def set_type(self, context, value): if isinstance(value, str): return {'type': rt_type[value]} return {'type': value} class RouteIPRouteFilter(IPRouteFilter): def get_table(self, context, mode): table = context.get('table', 0) if mode == 'field': if context.parameters['strict_check']: return 0 return table if 0 < table < 255 else 254 return table def set_metrics(self, context, value): if value and 'attrs' not in value: metrics = {'attrs': []} for name, metric in value.items(): rtax = rtmsg.metrics.name2nla(name) if metric is not None: metrics['attrs'].append([rtax, metric]) if metrics['attrs']: return {'metrics': metrics} return {} def set_multipath(self, context, value): ret = {} if value: hops = [] for v in value: if 'attrs' in v: hops.append(v) continue nh = {'attrs': []} nh_fields = [x[0] for x in nh_header.fields] for name in nh_fields: nh[name] = v.get(name, 0) for name in v: if name in nh_fields or v[name] is None: continue if name == 'encap' and isinstance(v[name], dict): if ( v[name].get('type', None) is None or v[name].get('labels', None) is None ): continue nh['attrs'].append( [ 'RTA_ENCAP_TYPE', encap_types.get( v[name]['type'], v[name]['type'] ), ] ) nh['attrs'].append( ['RTA_ENCAP', self.encap_header(v[name])] ) elif name == 'newdst': nh['attrs'].append( ['RTA_NEWDST', self.mpls_rta(v[name])] ) else: rta = rtmsg.name2nla(name) nh['attrs'].append([rta, v[name]]) hops.append(nh) if hops: ret = {'multipath': hops} if context.get('family') is None: # autodetect and propagate family hop = ret[0] attrs = hop.get('attrs', []) for attr in attrs: if attr[0] == 'RTA_GATEWAY': ret['family'] = ( AF_INET6 if attr[1].find(':') >= 0 else AF_INET ) break return ret def set_encap(self, context, value): if ( isinstance(value, (list, tuple)) and context.get('encap_type') == LWTUNNEL_ENCAP_MPLS ): return {'encap': {'attrs': [['MPLS_IPTUNNEL_DST', value]]}} elif isinstance(value, dict): # human-friendly form: # # 'encap': {'type': 'mpls', # 'labels': '200/300'} # # 'type' is mandatory if 'type' in value and 'labels' in value: return { 'encap_type': encap_types.get( value['type'], value['type'] ), 'encap': self.encap_header(value), } # human-friendly form: # # 'encap': {'type': 'seg6', # 'mode': 'encap' # 'segs': '2000::5,2000::6'} # # 'encap': {'type': 'seg6', # 'mode': 'inline' # 'segs': '2000::5,2000::6' # 'hmac': 1} # # 'encap': {'type': 'seg6', # 'mode': 'encap' # 'segs': '2000::5,2000::6' # 'hmac': 0xf} # # 'encap': {'type': 'seg6', # 'mode': 'inline' # 'segs': ['2000::5', '2000::6']} # # 'type', 'mode' and 'segs' are mandatory if 'type' in value and 'mode' in value and 'segs' in value: return { 'encap_type': encap_types.get( value['type'], value['type'] ), 'encap': self.encap_header(value), } elif 'type' in value and ( 'in' in value or 'out' in value or 'xmit' in value ): return { 'encap_type': encap_types.get( value['type'], value['type'] ), 'encap': self.encap_header(value), } # human-friendly form: # # 'encap': {'type': 'seg6local', # 'action': 'End'} # # 'encap': {'type': 'seg6local', # 'action': 'End.DT6', # 'table': '10'} # # 'encap': {'type': 'seg6local', # 'action': 'End.DT4', # 'vrf_table': 10} # # 'encap': {'type': 'seg6local', # 'action': 'End.DT46', # 'vrf_table': 10} # # 'encap': {'type': 'seg6local', # 'action': 'End.DX6', # 'nh6': '2000::5'} # # 'encap': {'type': 'seg6local', # 'action': 'End.B6' # 'srh': {'segs': '2000::5,2000::6', # 'hmac': 0xf}} # # 'type' and 'action' are mandatory elif 'type' in value and 'action' in value: return { 'encap_type': encap_types.get( value['type'], value['type'] ), 'encap': self.encap_header(value), } return {} def finalize(self, context): # cleanup extra NLA for AF_MPLS if context.get('family') == AF_MPLS: for key in tuple(context.keys()): if key not in ( 'family', 'proto', 'type', 'dst', 'newdst', 'via', 'multipath', 'oif', ): context.pop(key, None) # cleanup empty strings for key in context: if context[key] in ('', None): try: del context[key] except KeyError: pass def mpls_rta(self, value): # FIXME: planned for the next refactoring cycle ret = [] if not isinstance(value, (list, tuple, set)): value = (value,) for label in value: ret.append(MPLSTarget(label)) if ret: ret[-1]['bos'] = 1 return ret def encap_header(self, header): ''' Encap header transform. Format samples: {'type': 'mpls', 'labels': '200/300'} {'type': AF_MPLS, 'labels': (200, 300)} {'type': 'mpls', 'labels': 200} {'type': AF_MPLS, 'labels': [{'bos': 0, 'label': 200, 'ttl': 16}, {'bos': 1, 'label': 300, 'ttl': 16}]} ''' if isinstance(header['type'], int) or ( header['type'] in ('mpls', AF_MPLS, LWTUNNEL_ENCAP_MPLS) ): ret = [] override_bos = True labels = header['labels'] if isinstance(labels, str): labels = labels.split('/') if not isinstance(labels, (tuple, list, set)): labels = (labels,) for label in labels: if isinstance(label, dict): # dicts append intact override_bos = False ret.append(label) else: # otherwise construct label dict if isinstance(label, str): label = int(label) ret.append({'bos': 0, 'label': label}) # the last label becomes bottom-of-stack if override_bos: ret[-1]['bos'] = 1 return {'attrs': [['MPLS_IPTUNNEL_DST', ret]]} ''' Seg6 encap header transform. Format samples: {'type': 'seg6', 'mode': 'encap', 'segs': '2000::5,2000::6'} {'type': 'seg6', 'mode': 'encap' 'segs': '2000::5,2000::6', 'hmac': 1} ''' if header['type'] == 'seg6': # Init step ret = {} # Parse segs segs = header['segs'] # If they are in the form in_addr6,in_addr6 if isinstance(segs, str): # Create an array with the splitted values temp = segs.split(',') # Init segs segs = [] # Iterate over the values for seg in temp: # Discard empty string if seg != '': # Add seg to segs segs.append(seg) # Retrieve mode mode = header['mode'] # hmac is optional and contains the hmac key hmac = header.get('hmac', None) # Construct the new object ret = {'mode': mode, 'segs': segs} # If hmac is present convert to u32 if hmac: # Add to ret the hmac key ret['hmac'] = hmac & 0xFFFFFFFF # Done return the object return {'attrs': [['SEG6_IPTUNNEL_SRH', ret]]} ''' BPF encap header transform. Format samples: {'type': 'bpf', 'in': {'fd':4, 'name':'firewall'}} {'type': 'bpf', 'in' : {'fd':4, 'name':'firewall'}, 'out' : {'fd':5, 'name':'stats'}, 'xmit': {'fd':6, 'name':'vlan_push', 'headroom':4}} ''' if header['type'] == 'bpf': attrs = {} for key, value in header.items(): if key not in ['in', 'out', 'xmit']: continue obj = [ ['LWT_BPF_PROG_FD', value['fd']], ['LWT_BPF_PROG_NAME', value['name']], ] if key == 'in': attrs['LWT_BPF_IN'] = {'attrs': obj} elif key == 'out': attrs['LWT_BPF_OUT'] = {'attrs': obj} elif key == 'xmit': attrs['LWT_BPF_XMIT'] = {'attrs': obj} if 'headroom' in value: attrs['LWT_BPF_XMIT_HEADROOM'] = value['headroom'] return {'attrs': list(attrs.items())} ''' Seg6 encap header transform. Format samples: {'type': 'seg6local', 'action': 'End.DT6', 'table': '10'} {'type': 'seg6local', 'action': 'End.DT4', 'vrf_table': 10} {'type': 'seg6local', 'action': 'End.DT46', 'vrf_table': 10} {'type': 'seg6local', 'action': 'End.B6', 'table': '10' 'srh': {'segs': '2000::5,2000::6'}} ''' if header['type'] == 'seg6local': # Init step ret = {} table = None nh4 = None nh6 = None iif = None # Actually not used oif = None srh = {} segs = [] hmac = None prog_fd = None prog_name = None vrf_table = None # Parse segs if srh: segs = header['srh']['segs'] # If they are in the form in_addr6,in_addr6 if isinstance(segs, str): # Create an array with the splitted values temp = segs.split(',') # Init segs segs = [] # Iterate over the values for seg in temp: # Discard empty string if seg != '': # Add seg to segs segs.append(seg) # hmac is optional and contains the hmac key hmac = header.get('hmac', None) # Retrieve action action = header['action'] if action == 'End.X': # Retrieve nh6 nh6 = header['nh6'] elif action == 'End.T': # Retrieve table and convert to u32 table = header['table'] & 0xFFFFFFFF elif action == 'End.DX2': # Retrieve oif and convert to u32 oif = header['oif'] & 0xFFFFFFFF elif action == 'End.DX6': # Retrieve nh6 nh6 = header['nh6'] elif action == 'End.DX4': # Retrieve nh6 nh4 = header['nh4'] elif action == 'End.DT6': # Retrieve table table = header['table'] elif action == 'End.DT4': # Retrieve vrf_table vrf_table = header['vrf_table'] elif action == 'End.DT46': # Retrieve vrf_table vrf_table = header['vrf_table'] elif action == 'End.B6': # Parse segs segs = header['srh']['segs'] # If they are in the form in_addr6,in_addr6 if isinstance(segs, str): # Create an array with the splitted values temp = segs.split(',') # Init segs segs = [] # Iterate over the values for seg in temp: # Discard empty string if seg != '': # Add seg to segs segs.append(seg) # hmac is optional and contains the hmac key hmac = header.get('hmac', None) srh['segs'] = segs # If hmac is present convert to u32 if hmac: # Add to ret the hmac key srh['hmac'] = hmac & 0xFFFFFFFF srh['mode'] = 'inline' elif action == 'End.B6.Encaps': # Parse segs segs = header['srh']['segs'] # If they are in the form in_addr6,in_addr6 if isinstance(segs, str): # Create an array with the splitted values temp = segs.split(',') # Init segs segs = [] # Iterate over the values for seg in temp: # Discard empty string if seg != '': # Add seg to segs segs.append(seg) # hmac is optional and contains the hmac key hmac = header.get('hmac', None) srh['segs'] = segs if hmac: # Add to ret the hmac key srh['hmac'] = hmac & 0xFFFFFFFF srh['mode'] = 'encap' elif action == 'End.BPF': prog_fd = header['bpf']['fd'] prog_name = header['bpf']['name'] # Construct the new object ret = [] ret.append(['SEG6_LOCAL_ACTION', {'value': action}]) if table: # Add the table to ret ret.append(['SEG6_LOCAL_TABLE', {'value': table}]) if vrf_table: # Add the vrf_table to ret ret.append(['SEG6_LOCAL_VRFTABLE', {'value': vrf_table}]) if nh4: # Add the nh4 to ret ret.append(['SEG6_LOCAL_NH4', {'value': nh4}]) if nh6: # Add the nh6 to ret ret.append(['SEG6_LOCAL_NH6', {'value': nh6}]) if iif: # Add the iif to ret ret.append(['SEG6_LOCAL_IIF', {'value': iif}]) if oif: # Add the oif to ret ret.append(['SEG6_LOCAL_OIF', {'value': oif}]) if srh: # Add the srh to ret ret.append(['SEG6_LOCAL_SRH', srh]) if prog_fd and prog_name: # Add the prog_fd and prog_name to ret ret.append( [ 'SEG6_LOCAL_BPF', { 'attrs': [ ['LWT_BPF_PROG_FD', prog_fd], ['LWT_BPF_PROG_NAME', prog_name], ] }, ] ) # Done return the object return {'attrs': ret}	20.35 KB	2025-05-25 12:26
📄 rule.py import socket from pyroute2.netlink.rtnl.fibmsg import FR_ACT_NAMES, fibmsg from .common import Index, IPRouteFilter, IPTargets, NLAKeyTransform DEFAULT_FRA_PRIORITY = 32000 class RuleFieldFilter(IPTargets, Index, NLAKeyTransform): _nla_prefix = fibmsg.prefix class RuleIPRouteFilter(IPRouteFilter): def get_table(self, context, mode): table = context.get('table', 0) if mode == 'field': return table if 0 <= table <= 255 else 252 return table def set_action(self, context, value): if isinstance(value, str): return { 'action': FR_ACT_NAMES.get( value, FR_ACT_NAMES.get('FR_ACT_' + value.upper(), value) ) } return {'action': value} def finalize(self, context): if self.command != 'dump': if 'family' not in context: context['family'] = socket.AF_INET if 'priority' not in context: context['priority'] = DEFAULT_FRA_PRIORITY if 'table' in context and 'action' not in context: context['action'] = 'to_tbl' for key in ('src_len', 'dst_len'): if ( context.get(key, None) is None and context.get(key[:3], None) is not None ): context[key] = {socket.AF_INET6: 128, socket.AF_INET: 32}[ context['family'] ]	1.46 KB	2025-05-25 12:26
📄 tc.py from pyroute2.netlink.rtnl import TC_H_ROOT from pyroute2.netlink.rtnl.tcmsg import plugins as tc_plugins from .common import IPRouteFilter class TcRequestFilter: def transform_handle(self, key, context, handle): if isinstance(handle, str): (major, minor) = [ int(x if x else '0', 16) for x in handle.split(':') ] handle = (major << 8 * 2) \| minor return {key: handle} def set_handle(self, context, value): return self.transform_handle('handle', context, value) def set_root(self, context, value): ret = {} if value: ret['parent'] = 0xFFFFFFFF if 'handle' not in context: ret['handle'] = 0x10000 return ret def set_target(self, context, value): return self.transform_handle('target', context, value) def set_parent(self, context, value): return self.transform_handle('parent', context, value) def set_default(self, context, value): return self.transform_handle('default', context, value) class TcIPRouteFilter(IPRouteFilter): def finalize(self, context): if self.command.startswith('dump'): context.pop('kind', None) return if 'index' not in context: context['index'] = 0 if 'handle' not in context: context['handle'] = 0 # get & run the plugin if 'kind' in context: if context['kind'] in tc_plugins: plugin = tc_plugins[context['kind']] context['parent'] = context.get( 'parent', getattr(plugin, 'parent', 0) ) if set(context.keys()) > set(('kind', 'index', 'handle')): get_parameters = None if self.command[-5:] == 'class': get_parameters = getattr( plugin, 'get_class_parameters', None ) else: get_parameters = getattr( plugin, 'get_parameters', None ) if get_parameters is not None: context['options'] = get_parameters(dict(context)) if hasattr(plugin, 'fix_request'): plugin.fix_request(context) else: context['parent'] = TC_H_ROOT	2.39 KB	2025-05-25 12:26

Terminal