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 import types from pyroute2.netlink import nla, nlmsg from pyroute2.netlink.rtnl.tcmsg import ( cls_basic, cls_flow, cls_fw, cls_matchall, cls_u32, sched_bpf, sched_cake, sched_choke, sched_clsact, sched_codel, sched_drr, sched_fq_codel, sched_hfsc, sched_htb, sched_ingress, sched_netem, sched_pfifo, sched_pfifo_fast, sched_plug, sched_sfq, sched_tbf, sched_template, ) plugins = { 'plug': sched_plug, 'sfq': sched_sfq, 'clsact': sched_clsact, 'codel': sched_codel, 'fq_codel': sched_fq_codel, 'hfsc': sched_hfsc, 'htb': sched_htb, 'bpf': sched_bpf, 'tbf': sched_tbf, 'netem': sched_netem, 'fw': cls_fw, 'u32': cls_u32, 'matchall': cls_matchall, 'basic': cls_basic, 'flow': cls_flow, 'ingress': sched_ingress, 'pfifo': sched_pfifo, 'pfifo_fast': sched_pfifo_fast, 'choke': sched_choke, 'drr': sched_drr, 'prio': sched_pfifo_fast, 'cake': sched_cake, } class tcmsg(nlmsg): prefix = 'TCA_' fields = ( ('family', 'B'), ('pad1', 'B'), ('pad2', 'H'), ('index', 'i'), ('handle', 'I'), ('parent', 'I'), ('info', 'I'), ) nla_map = ( ('TCA_UNSPEC', 'none'), ('TCA_KIND', 'asciiz'), ('TCA_OPTIONS', 'get_options'), ('TCA_STATS', 'stats'), ('TCA_XSTATS', 'get_xstats'), ('TCA_RATE', 'hex'), ('TCA_FCNT', 'hex'), ('TCA_STATS2', 'get_stats2'), ('TCA_STAB', 'hex'), ) class stats(nla): fields = ( ('bytes', 'Q'), ('packets', 'I'), ('drop', 'I'), ('overlimits', 'I'), ('bps', 'I'), ('pps', 'I'), ('qlen', 'I'), ('backlog', 'I'), ) def get_plugin(self, plug, argv, kwarg): # get the plugin name kind = self.get_attr('TCA_KIND') # get the plugin implementation or the default one p = plugins.get(kind, sched_template) # get the interface interface = getattr(p, plug, getattr(sched_template, plug)) # if it is a method, run and return the result if isinstance(interface, types.FunctionType): return interface(self, argv, *kwarg) else: return interface @staticmethod def get_stats2(self, argv, *kwarg): return self.get_plugin('stats2', argv, *kwarg) @staticmethod def get_xstats(self, argv, *kwarg): return self.get_plugin('stats', argv, *kwarg) @staticmethod def get_options(self, argv, *kwarg): return self.get_plugin('options', argv, **kwarg)	2.68 KB	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 act_bpf.py from pyroute2.netlink import nla from pyroute2.netlink.rtnl.tcmsg.common import tc_actions def fix_request(request): if 'rate' in request: del request['rate'] class options(nla): nla_map = ( ('TCA_ACT_BPF_UNSPEC', 'none'), ('TCA_ACT_BPF_TM,', 'none'), ('TCA_ACT_BPF_PARMS', 'tca_act_bpf_parms'), ('TCA_ACT_BPF_OPS_LEN', 'uint16'), ('TCA_ACT_BPF_OPS', 'hex'), ('TCA_ACT_BPF_FD', 'uint32'), ('TCA_ACT_BPF_NAME', 'asciiz'), ) class tca_act_bpf_parms(nla): fields = ( ('index', 'I'), ('capab', 'I'), ('action', 'i'), ('refcnt', 'i'), ('bindcnt', 'i'), ) def get_parameters(kwarg): ret = {'attrs': []} if 'fd' in kwarg: ret['attrs'].append(['TCA_ACT_BPF_FD', kwarg['fd']]) if 'name' in kwarg: ret['attrs'].append(['TCA_ACT_BPF_NAME', kwarg['name']]) a = tc_actions[kwarg.get('action', 'drop')] ret['attrs'].append(['TCA_ACT_BPF_PARMS', {'action': a}]) return ret	1.04 KB	2025-05-25 12:26
📄 act_connmark.py from pyroute2.netlink import NLA_F_NESTED, nla from pyroute2.netlink.rtnl.tcmsg.common import tc_actions """ connmark - netfilter connmark retriever action see tc-connmark(8) This filter restores the connection mark into the packet mark. Connection marks are typically handled by the CONNMARK iptables module. See iptables-extensions(8). There is no mandatory parameter, but you can specify the action, which defaults to 'pipe', and the conntrack zone (see the manual). """ class options(nla): nla_flags = NLA_F_NESTED nla_map = ( ('TCA_CONNMARK_UNSPEC', 'none'), ('TCA_CONNMARK_PARMS', 'tca_connmark_parms'), ('TCA_CONNMARK_TM', 'none'), ) class tca_connmark_parms(nla): fields = ( ('index', 'I'), ('capab', 'I'), ('action', 'i'), ('refcnt', 'i'), ('bindcnt', 'i'), ('zone', 'H'), ('__padding', 'H'), # XXX is there a better way to do this ? ) def get_parameters(kwarg): ret = {'attrs': []} parms = { 'action': tc_actions[kwarg.get('action', 'pipe')], 'zone': kwarg.get('zone', 0), } ret['attrs'].append(['TCA_CONNMARK_PARMS', parms]) return ret	1.21 KB	2025-05-25 12:26
📄 act_gact.py from pyroute2.netlink import NLA_F_NESTED, nla from pyroute2.netlink.rtnl.tcmsg.common import tc_actions class options(nla): nla_flags = NLA_F_NESTED nla_map = ( ('TCA_GACT_UNSPEC', 'none'), ('TCA_GACT_TM', 'none'), ('TCA_GACT_PARMS', 'tca_gact_parms'), ('TCA_GACT_PROB', 'none'), ) class tca_gact_parms(nla): fields = ( ('index', 'I'), ('capab', 'I'), ('action', 'i'), ('refcnt', 'i'), ('bindcnt', 'i'), ) def get_parameters(kwarg): ret = {'attrs': []} a = tc_actions[kwarg.get('action', 'drop')] ret['attrs'].append(['TCA_GACT_PARMS', {'action': a}]) return ret	709 B	2025-05-25 12:26
📄 act_mirred.py from pyroute2.netlink import NLA_F_NESTED, nla from pyroute2.netlink.rtnl.tcmsg.common import tc_actions """ Mirred - mirror/redirect action see tc-mirred(8) Use like any other action, with the following parameters available: - direction (mandatory): ingress or egress - action (mandatory): mirror or redirect - ifindex (mandatory): destination interface for mirrored or redirected packets - index: explicit index for this action """ # see tc_mirred.h MIRRED_EACTIONS = { ("egress", "redirect"): 1, # redirect packet to egress ("egress", "mirror"): 2, # mirror packet to egress ("ingress", "redirect"): 3, # redirect packet to ingress ("ingress", "mirror"): 4, # mirror packet to ingress } class options(nla): nla_flags = NLA_F_NESTED nla_map = ( ('TCA_MIRRED_UNSPEC', 'none'), ('TCA_MIRRED_TM', 'none'), ('TCA_MIRRED_PARMS', 'tca_mirred_parms'), ) class tca_mirred_parms(nla): fields = ( ('index', 'I'), ('capab', 'I'), ('action', 'i'), ('refcnt', 'i'), ('bindcnt', 'i'), ('eaction', 'i'), ('ifindex', 'I'), ) def get_parameters(kwarg): ret = {'attrs': []} # direction, action and ifindex are mandatory parms = { 'eaction': MIRRED_EACTIONS[(kwarg['direction'], kwarg['action'])], 'ifindex': kwarg['ifindex'], } if 'index' in kwarg: parms['index'] = int(kwarg['index']) # From m_mirred.c if kwarg['action'] == 'redirect': parms['action'] = tc_actions['stolen'] else: # mirror parms['action'] = tc_actions['pipe'] ret['attrs'].append(['TCA_MIRRED_PARMS', parms]) return ret	1.68 KB	2025-05-25 12:26
📄 act_police.py from pyroute2.netlink.rtnl.tcmsg.common import ( get_rate_parameters, nla_plus_rtab, ) actions = { 'unspec': -1, # TC_POLICE_UNSPEC 'ok': 0, # TC_POLICE_OK 'reclassify': 1, # TC_POLICE_RECLASSIFY 'shot': 2, # TC_POLICE_SHOT 'drop': 2, # TC_POLICE_SHOT 'pipe': 3, } # TC_POLICE_PIPE class options(nla_plus_rtab): nla_map = ( ('TCA_POLICE_UNSPEC', 'none'), ('TCA_POLICE_TBF', 'police_tbf'), ('TCA_POLICE_RATE', 'rtab'), ('TCA_POLICE_PEAKRATE', 'ptab'), ('TCA_POLICE_AVRATE', 'uint32'), ('TCA_POLICE_RESULT', 'uint32'), ) class police_tbf(nla_plus_rtab.parms): fields = ( ('index', 'I'), ('action', 'i'), ('limit', 'I'), ('burst', 'I'), ('mtu', 'I'), ('rate_cell_log', 'B'), ('rate___reserved', 'B'), ('rate_overhead', 'H'), ('rate_cell_align', 'h'), ('rate_mpu', 'H'), ('rate', 'I'), ('peak_cell_log', 'B'), ('peak___reserved', 'B'), ('peak_overhead', 'H'), ('peak_cell_align', 'h'), ('peak_mpu', 'H'), ('peak', 'I'), ('refcnt', 'i'), ('bindcnt', 'i'), ('capab', 'I'), ) class nla_plus_police(object): class police(options): pass def get_parameters(kwarg): # if no limit specified, set it to zero to make # the next call happy kwarg['limit'] = kwarg.get('limit', 0) tbfp = get_rate_parameters(kwarg) # create an alias -- while TBF uses 'buffer', rate # policy uses 'burst' tbfp['burst'] = tbfp['buffer'] # action resolver tbfp['action'] = actions[kwarg.get('action', 'reclassify')] return {'attrs': [['TCA_POLICE_TBF', tbfp], ['TCA_POLICE_RATE', True]]}	1.82 KB	2025-05-25 12:26
📄 act_skbedit.py ''' skbedit +++++++ Usage:: from pyroute2 import IPRoute # Assume you are working with eth1 interface IFNAME = "eth1" ipr = IPRoute() ifindex = ipr.link_lookup(ifname=IFNAME) # First create parent qdisc ipr.tc("add", "htb", index=ifindex, handle=0x10000) # Then add a matchall filter with skbedit action # Simple action example action = {"kind": "skbedit", "priority": 0x10001 # Also known as "1:1" in TC format } ipr.tc("add-filter", "matchall", index=ifindex, parent=0x10000, prio=1, action=action) # Extended action example action = {"kind": "skbedit", "priority": 0x10001, # Also known as "1:1" in TC format "mark": 0x1337, "mask": 0xFFFFFFFF, "ptype": "host" } ipr.tc("add-filter", "matchall", index=ifindex, parent=0x10000, prio=1, action=action) NOTES: Here is the list of all supported options:: - mark: integer - mask: integer - priority: integer - ptype: "host", "otherhost", "broadcast" or "multicast" - queue: integer ''' from pyroute2.netlink import nla from pyroute2.netlink.rtnl.tcmsg.common import tc_actions # Packet types defined in if_packet.h PACKET_HOST = 0 PACKET_BROADCAST = 1 PACKET_MULTICAST = 2 PACKET_OTHERHOST = 3 def convert_ptype(value): types = { 'host': PACKET_HOST, 'otherhost': PACKET_OTHERHOST, 'broadcast': PACKET_BROADCAST, 'multicast': PACKET_MULTICAST, } res = types.get(value.lower()) if res is not None: return res raise ValueError( 'Invalid ptype specified! See tc-skbedit man ' 'page for valid values.' ) def get_parameters(kwarg): ret = {'attrs': []} attrs_map = ( ('priority', 'TCA_SKBEDIT_PRIORITY'), ('queue', 'TCA_SKBEDIT_QUEUE_MAPPING'), ('mark', 'TCA_SKBEDIT_MARK'), ('ptype', 'TCA_SKBEDIT_PTYPE'), ('mask', 'TCA_SKBEDIT_MASK'), ) # Assign TCA_SKBEDIT_PARMS first parms = {} parms['action'] = tc_actions['pipe'] ret['attrs'].append(['TCA_SKBEDIT_PARMS', parms]) for k, v in attrs_map: r = kwarg.get(k, None) if r is not None: if k == 'ptype': r = convert_ptype(r) ret['attrs'].append([v, r]) return ret class options(nla): nla_map = ( ('TCA_SKBEDIT_UNSPEC', 'none'), ('TCA_SKBEDIT_TM', 'tca_parse_tm'), ('TCA_SKBEDIT_PARMS', 'tca_parse_parms'), ('TCA_SKBEDIT_PRIORITY', 'uint32'), ('TCA_SKBEDIT_QUEUE_MAPPING', 'uint16'), ('TCA_SKBEDIT_MARK', 'uint32'), ('TCA_SKBEDIT_PAD', 'hex'), ('TCA_SKBEDIT_PTYPE', 'uint16'), ('TCA_SKBEDIT_MASK', 'uint32'), ('TCA_SKBEDIT_FLAGS', 'uint64'), ) class tca_parse_parms(nla): # As described in tc_mpls.h, it uses # generic TC action fields fields = ( ('index', 'I'), ('capab', 'I'), ('action', 'i'), ('refcnt', 'i'), ('bindcnt', 'i'), ) class tca_parse_tm(nla): # See struct tcf_t fields = ( ('install', 'Q'), ('lastuse', 'Q'), ('expires', 'Q'), ('firstuse', 'Q'), )	3.26 KB	2025-05-25 12:26
📄 act_vlan.py from socket import htons from pyroute2.netlink import nla from pyroute2.netlink.rtnl.tcmsg.common import tc_actions v_actions = {'pop': 1, 'push': 2, 'modify': 3} class options(nla): nla_map = ( ('TCA_VLAN_UNSPEC', 'none'), ('TCA_VLAN_TM', 'none'), ('TCA_VLAN_PARMS', 'tca_vlan_parms'), ('TCA_VLAN_PUSH_VLAN_ID', 'uint16'), ('TCA_VLAN_PUSH_VLAN_PROTOCOL', 'uint16'), ('TCA_VLAN_PAD', 'none'), ('TCA_VLAN_PUSH_VLAN_PRIORITY', 'uint8'), ) class tca_vlan_parms(nla): fields = ( ('index', 'I'), ('capab', 'I'), ('action', 'i'), ('refcnt', 'i'), ('bindcnt', 'i'), ('v_action', 'i'), ) def get_parameters(kwarg): ret = {'attrs': []} parms = {'v_action': v_actions[kwarg['v_action']]} parms['action'] = tc_actions[kwarg.get('action', 'pipe')] ret['attrs'].append(['TCA_VLAN_PARMS', parms]) # Vlan id compulsory for "push" and "modify" if kwarg['v_action'] in ['push', 'modify']: ret['attrs'].append(['TCA_VLAN_PUSH_VLAN_ID', kwarg['id']]) if 'priority' in kwarg: ret['attrs'].append(['TCA_VLAN_PUSH_VLAN_PRIORITY', kwarg['priority']]) if kwarg.get('protocol', '802.1Q') == '802.1ad': ret['attrs'].append(['TCA_VLAN_PUSH_VLAN_PROTOCOL', htons(0x88A8)]) else: ret['attrs'].append(['TCA_VLAN_PUSH_VLAN_PROTOCOL', htons(0x8100)]) return ret	1.43 KB	2025-05-25 12:26
📄 cls_basic.py ''' basic +++++ Basic filter has multiple types supports. Examples with ipset matches:: # Prepare a simple match on an ipset at index 0 src # (the first ipset name that appears when running `ipset list`) match = [{"kind": "ipset", "index": 0, "mode": "src"}] ip.tc("add-filter", "basic", ifb0, parent=0x10000, classid=0x10010, match=match) # The same match but inverted, simply add inverse flag match = [{"kind": "ipset", "index": 0, "mode": "src", "inverse": True}] ip.tc("add-filter", "basic", ifb0, parent=0x10000, classid=0x10010, match=match) # Still one ipset but with multiple dimensions: # comma separated list of modes match = [{"kind": "ipset", "index": 0, "mode": "src,dst"}] ip.tc("add-filter", "basic", ifb0, parent=0x10000, classid=0x10010, match=match) # Now let's add multiple expressions (ipset 0 src and ipset 1 src) match = [{"kind": "ipset", "index": 0, "mode": "src", "relation": "and"}, {"kind": "ipset", "index": 1, "mode": "src"}] ip.tc("add-filter", "basic", ifb0, parent=0x10000, classid=0x10010, match=match) # The same works with OR (ipset 0 src or ipset 1 src) match = [{"kind": "ipset", "index": 0, "mode": "src", "relation": "OR"}, {"kind": "ipset", "index": 1, "mode": "src"}] ip.tc("add-filter", "basic", ifb0, parent=0x10000, classid=0x10010, match=match) Examples with cmp matches:: # Repeating the example given in the man page match = [{"kind": "cmp", "layer": 2, "opnd": "gt", "align": "u16", "offset": 3, "mask": 0xff00, "value": 20}] ip.tc("add-filter", "basic", ifb0, parent=0x10000, classid=0x10010, match=match) # Now, the same example but with variations # - use layer name instead of enum # - use operand sign instead of name match = [{"kind": "cmp", "layer": "transport", "opnd": ">","align": "u16", "offset": 3, "mask": 0xff00, "value": 20}] ip.tc("add-filter", "basic", ifb0, parent=0x10000, classid=0x10010, match=match) # Again, the same example with all possible keywords even if they are # ignored match = [{"kind": "cmp", "layer": "tcp", "opnd": ">", "align": "u16", "offset": 3, "mask": 0xff00, "value": 20, "trans": False}] ip.tc("add-filter", "basic", ifb0, parent=0x10000, classid=0x10010, match=match) # Another example, we want to work at the link layer # and filter incoming packets matching hwaddr 00:DE:AD:C0:DE:00 # OSI model tells us that the source hwaddr is at offset 0 of # the link layer. # Size of hwaddr is 6-bytes in length, so I use an u32 then an u16 # to do the complete match match = [{"kind": "cmp", "layer": "link", "opnd": "eq", "align": "u32", "offset": 0, "mask": 0xffffffff, "value": 0x00DEADC0, "relation": "and"}, {"kind": "cmp", "layer": "link", "opnd": "eq", "align": "u16", "offset": 4, "mask": 0xffff, "value": 0xDE00}] ip.tc("add-filter", "basic", ifb0, parent=0x10000, classid=0x10010, match=match) # As the man page says, here are the different key-value pairs you can use: # "layer": "link" or "eth" or 0 # "layer": "network" or "ip" or 1 # "layer": "transport" or "tcp" or 2 # "opnd": "eq" or "=" or 0 # "opnd": "gt" or ">" or 1 # "opnd": "lt" or "<" or 2 # "align": "u8" or "u16" or "u32" # "trans": True or False # "offset", "mask" and "value": any integer Examples with meta matches:: # Repeating the example given in the man page match = [{"kind": "meta", "object":{"kind": "nfmark", "opnd": "gt"}, "value": 24, "relation": "and"}, {"kind": "meta", "object":{"kind": "tcindex", "opnd": "eq"}, "value": 0xf0, "mask": 0xf0}] ip.tc("add-filter", "basic", ifb0, parent=0x10000, classid=0x10010, match=match) # Now, the same example but with variations # - use operand sign instead of name match = [{"kind": "meta", "object":{"kind": "nfmark", "opnd": ">"}, "value": 24, "relation": "and"}, {"kind": "meta", "object":{"kind": "tcindex", "opnd": "="}, "value": 0xf0, "mask": 0xf0}] ip.tc("add-filter", "basic", ifb0, parent=0x10000, classid=0x10010, match=match) # Another example given by the tc helper # meta(indev shift 1 eq "ppp") match = [{"kind": "meta", "object":{"kind": "dev", "opnd": "eq", "shift": 1}, "value": "ppp"}] ip.tc("add-filter", "basic", ifb0, parent=0x10000, classid=0x10010, match=match) # Another example, drop every packets arriving on ifb0 match = [{"kind": "meta", "object":{"kind": "dev", "opnd": "eq"}, "value": "ifb0"}] ip.tc("add-filter", "basic", ifb0, parent=0x10000, classid=0x10010, match=match, action="drop") # As the man page says, here are the different key-value pairs you can use: # "opnd": "eq" or "=" or 0 # "opnd": "gt" or ">" or 1 # "opnd": "lt" or "<" or 2 # "shift": any integer between 0 and 255 included # "kind" object: see `tc filter add dev iface basic match 'meta(list)'` result # "value": any string if kind matches 'dev' or 'sk_bound_if', # any integer otherwise NOTES: When not specified, `inverse` flag is set to False. Do not specify `relation` keyword on the last expression or if there is only one expression. `relation` can be written using multiple format: "and", "AND", "&&", "or", "OR", "\|\|" You can combine multiple different types of ematch. Here is an example:: match = [{"kind": "cmp", "layer": 2, "opnd": "eq", "align": "u32", "offset": 0, "value": 32, "relation": "&&"}, {"kind": "meta", "object":{"kind": "vlan_tag", "opnd": "eq"}, "value": 100, "relation": "\|\|"}, {"kind": "ipset", "index": 0, "mode": "src", "inverse": True} ] ''' import struct from socket import htons from pyroute2 import protocols from pyroute2.netlink import nla from pyroute2.netlink.rtnl.tcmsg.common_act import get_tca_action, tca_act_prio from pyroute2.netlink.rtnl.tcmsg.common_ematch import ( get_tcf_ematches, nla_plus_tcf_ematch_opt, ) def fix_request(request): if 'rate' in request: del request['rate'] request['info'] = htons( request.get('protocol', protocols.ETH_P_ALL) & 0xFFFF ) \| ((request.get('prio', 0) << 16) & 0xFFFF0000) def get_parameters(kwarg): ret = {'attrs': []} attrs_map = (('classid', 'TCA_BASIC_CLASSID'),) if kwarg.get('match'): ret['attrs'].append(['TCA_BASIC_EMATCHES', get_tcf_ematches(kwarg)]) if kwarg.get('action'): ret['attrs'].append(['TCA_BASIC_ACT', get_tca_action(kwarg)]) for k, v in attrs_map: r = kwarg.get(k, None) if r is not None: ret['attrs'].append([v, r]) return ret class options(nla): nla_map = ( ('TCA_BASIC_UNSPEC', 'none'), ('TCA_BASIC_CLASSID', 'uint32'), ('TCA_BASIC_EMATCHES', 'parse_basic_ematch_tree'), ('TCA_BASIC_ACT', 'tca_act_prio'), ('TCA_BASIC_POLICE', 'hex'), ) class parse_basic_ematch_tree(nla): nla_map = ( ('TCA_EMATCH_TREE_UNSPEC', 'none'), ('TCA_EMATCH_TREE_HDR', 'tcf_parse_header'), ('TCA_EMATCH_TREE_LIST', '*tcf_parse_list'), ) class tcf_parse_header(nla): fields = (('nmatches', 'H'), ('progid', 'H')) class tcf_parse_list(nla, nla_plus_tcf_ematch_opt): fields = ( ('matchid', 'H'), ('kind', 'H'), ('flags', 'H'), ('pad', 'H'), ('opt', 's'), ) def decode(self): nla.decode(self) size = 0 for field in self.fields + self.header: if 'opt' in field: # Ignore this field as it a hack used to brain encoder continue size += struct.calcsize(field[1]) start = self.offset + size end = self.offset + self.length data = self.data[start:end] self['opt'] = self.parse_ematch_options(self, data) tca_act_prio = tca_act_prio	8.34 KB	2025-05-25 12:26
📄 cls_flow.py ''' flow ++++ Flow filter supports two types of modes:: - map - hash # Prepare a Qdisc with fq-codel ip.tc("add", "fq_codel", ifb0, parent=0x10001, handle=0x10010) # Create flow filter with hash mode # Single: keys = "src" # Multi (comma separated list of keys): keys = "src,nfct-src" ip.tc("add-filter", "flow", ifb0, mode="hash", keys=keys, divisor=1024, perturb=60, handle=0x10, baseclass=0x10010, parent=0x10001) # Create flow filter with map mode # Simple map dst with no OP: ip.tc("add-filter", "flow", ifb0, mode="map", key="dst", divisor=1024, handle=10 baseclass=0x10010) # Same filter with xor OP: ops = [{"op": "xor", "num": 0xFF}] ip.tc("add-filter", "flow", ifb0, mode="map", key="dst", divisor=1024, handle=10 baseclass=0x10010, ops=ops) # Complex one with addend OP (incl. minus support): ops = [{"op": "addend", "num": '-192.168.0.0'}] ip.tc("add-filter", "flow", ifb0, mode="map", key="dst", divisor=1024, handle=10 baseclass=0x10010, ops=ops) # Example with multiple OPS: ops = [{"op": "and", "num": 0xFF}, {"op": "rshift", "num": 4}] ip.tc("add-filter", "flow", ifb0, mode="map", key="dst", divisor=1024, handle=10 baseclass=0x10010, ops=ops) NOTES: When using `map` mode, use the keyword `key` to pass a key. When using `hash` mode, use the keyword `keys` to pass a key even if there is only one key. In `map` mode, the `num` parameter in `OPS` is always an integer unless if you use the OP `addend`, which can be a string IPv4 address. You can also add a minus sign at the beginning of the `num` value even if it is an IPv4 address. ''' from socket import htons from pyroute2 import protocols from pyroute2.netlink import nla from pyroute2.netlink.rtnl.tcmsg.common import ( get_tca_keys, get_tca_mode, get_tca_ops, tc_flow_keys, tc_flow_modes, ) from pyroute2.netlink.rtnl.tcmsg.common_act import get_tca_action, tca_act_prio def fix_request(request): if 'rate' in request: del request['rate'] request['info'] = htons( request.get('protocol', protocols.ETH_P_ALL) & 0xFFFF ) \| ((request.get('prio', 0) << 16) & 0xFFFF0000) def get_parameters(kwarg): ret = {'attrs': []} attrs_map = ( ('baseclass', 'TCA_FLOW_BASECLASS'), ('divisor', 'TCA_FLOW_DIVISOR'), ('perturb', 'TCA_FLOW_PERTURB'), ) if kwarg.get('mode'): ret['attrs'].append(['TCA_FLOW_MODE', get_tca_mode(kwarg)]) if kwarg.get('mode') == 'hash': ret['attrs'].append(['TCA_FLOW_KEYS', get_tca_keys(kwarg, 'keys')]) if kwarg.get('mode') == 'map': ret['attrs'].append(['TCA_FLOW_KEYS', get_tca_keys(kwarg, 'key')]) # Check for OPS presence if 'ops' in kwarg: get_tca_ops(kwarg, ret['attrs']) if kwarg.get('action'): ret['attrs'].append(['TCA_FLOW_ACT', get_tca_action(kwarg)]) for k, v in attrs_map: r = kwarg.get(k, None) if r is not None: ret['attrs'].append([v, r]) return ret class options(nla): nla_map = ( ('TCA_FLOW_UNSPEC', 'none'), ('TCA_FLOW_KEYS', 'tca_parse_keys'), ('TCA_FLOW_MODE', 'tca_parse_mode'), ('TCA_FLOW_BASECLASS', 'uint32'), ('TCA_FLOW_RSHIFT', 'uint32'), ('TCA_FLOW_ADDEND', 'uint32'), ('TCA_FLOW_MASK', 'uint32'), ('TCA_FLOW_XOR', 'uint32'), ('TCA_FLOW_DIVISOR', 'uint32'), ('TCA_FLOW_ACT', 'tca_act_prio'), ('TCA_FLOW_POLICE', 'hex'), ('TCA_FLOW_EMATCHES', 'hex'), ('TCA_FLOW_PERTURB', 'uint32'), ) class tca_parse_mode(nla): fields = (('flow_mode', 'I'),) def decode(self): nla.decode(self) for key, value in tc_flow_modes.items(): if self['flow_mode'] == value: self['flow_mode'] = key break def encode(self): self['flow_mode'] = self['value'] nla.encode(self) class tca_parse_keys(nla): fields = (('flow_keys', 'I'),) def decode(self): nla.decode(self) keys = '' for key, value in tc_flow_keys.items(): if value & self['flow_keys']: keys = '{0},{1}'.format(keys, key) self['flow_keys'] = keys.strip(',') def encode(self): self['flow_keys'] = self['value'] nla.encode(self) tca_act_prio = tca_act_prio	4.57 KB	2025-05-25 12:26
📄 cls_fw.py from socket import htons from pyroute2 import protocols from pyroute2.netlink import nla from pyroute2.netlink.rtnl.tcmsg.act_police import ( get_parameters as ap_parameters, ) from pyroute2.netlink.rtnl.tcmsg.act_police import nla_plus_police from pyroute2.netlink.rtnl.tcmsg.common_act import get_tca_action, tca_act_prio def fix_request(request): if 'rate' in request: del request['rate'] request['info'] = htons( request.get('protocol', protocols.ETH_P_ALL) & 0xFFFF ) \| ((request.get('prio', 0) << 16) & 0xFFFF0000) def get_parameters(kwarg): ret = {'attrs': []} attrs_map = ( ('classid', 'TCA_FW_CLASSID'), # ('police', 'TCA_FW_POLICE'), # Handled in ap_parameters ('indev', 'TCA_FW_INDEV'), ('mask', 'TCA_FW_MASK'), ) if kwarg.get('rate'): ret['attrs'].append(['TCA_FW_POLICE', ap_parameters(kwarg)]) if kwarg.get('action'): ret['attrs'].append(['TCA_FW_ACT', get_tca_action(kwarg)]) for k, v in attrs_map: r = kwarg.get(k, None) if r is not None: ret['attrs'].append([v, r]) return ret class options(nla, nla_plus_police): nla_map = ( ('TCA_FW_UNSPEC', 'none'), ('TCA_FW_CLASSID', 'uint32'), ('TCA_FW_POLICE', 'police'), # TODO string? ('TCA_FW_INDEV', 'hex'), # TODO string ('TCA_FW_ACT', 'tca_act_prio'), ('TCA_FW_MASK', 'uint32'), ) tca_act_prio = tca_act_prio	1.46 KB	2025-05-25 12:26
📄 cls_matchall.py from socket import htons from pyroute2 import protocols from pyroute2.netlink import nla from pyroute2.netlink.rtnl.tcmsg.common_act import get_tca_action, tca_act_prio def fix_request(request): if 'rate' in request: del request['rate'] request['info'] = htons( request.get('protocol', protocols.ETH_P_ALL) & 0xFFFF ) \| ((request.get('prio', 0) << 16) & 0xFFFF0000) def get_parameters(kwarg): ret = {'attrs': []} attrs_map = ( ('classid', 'TCA_MATCHALL_CLASSID'), ('flags', 'TCA_MATCHALL_FLAGS'), ) if kwarg.get('action'): ret['attrs'].append(['TCA_MATCHALL_ACT', get_tca_action(kwarg)]) for k, v in attrs_map: r = kwarg.get(k, None) if r is not None: ret['attrs'].append([v, r]) return ret class options(nla): nla_map = ( ('TCA_MATCHALL_UNSPEC', 'none'), ('TCA_MATCHALL_CLASSID', 'be32'), ('TCA_MATCHALL_ACT', 'tca_act_prio'), ('TCA_MATCHALL_FLAGS', 'be32'), ) tca_act_prio = tca_act_prio	1.02 KB	2025-05-25 12:26
📄 cls_u32.py ''' u32 +++ Filters can take an `action` argument, which affects the packet behavior when the filter matches. Currently the gact, bpf, and police action types are supported, and can be attached to the u32 and bpf filter types:: # An action can be a simple string, which translates to a gact type action = "drop" # Or it can be an explicit type (these are equivalent) action = dict(kind="gact", action="drop") # There can also be a chain of actions, which depend on the return # value of the previous action. action = [ dict(kind="bpf", fd=fd, name=name, action="ok"), dict(kind="police", rate="10kbit", burst=10240, limit=0), dict(kind="gact", action="ok"), ] # Add the action to a u32 match-all filter ip.tc("add", "htb", eth0, 0x10000, default=0x200000) ip.tc("add-filter", "u32", eth0, parent=0x10000, prio=10, protocol=protocols.ETH_P_ALL, target=0x10020, keys=["0x0/0x0+0"], action=action) # Add two more filters: One to send packets with a src address of # 192.168.0.1/32 into 1:10 and the second to send packets with a # dst address of 192.168.0.0/24 into 1:20 ip.tc("add-filter", "u32", eth0, parent=0x10000, prio=10, protocol=protocols.ETH_P_IP, target=0x10010, keys=["0xc0a80001/0xffffffff+12"]) # 0xc0a800010 = 192.168.0.1 # 0xffffffff = 255.255.255.255 (/32) # 12 = Source network field bit offset ip.tc("add-filter", "u32", eth0, parent=0x10000, prio=10, protocol=protocols.ETH_P_IP, target=0x10020, keys=["0xc0a80000/0xffffff00+16"]) # 0xc0a80000 = 192.168.0.0 # 0xffffff00 = 255.255.255.0 (/24) # 16 = Destination network field bit offset ''' import struct from socket import htons from pyroute2.netlink import nla, nlmsg from pyroute2.netlink.rtnl.tcmsg.act_police import ( get_parameters as ap_parameters, ) from pyroute2.netlink.rtnl.tcmsg.act_police import nla_plus_police from pyroute2.netlink.rtnl.tcmsg.common_act import get_tca_action, tca_act_prio def fix_request(request): if 'rate' in request: del request['rate'] request['info'] = htons(request.get('protocol', 0) & 0xFFFF) \| ( (request.get('prio', 0) << 16) & 0xFFFF0000 ) def get_parameters(kwarg): ret = {'attrs': []} if kwarg.get('rate'): ret['attrs'].append(['TCA_U32_POLICE', ap_parameters(kwarg)]) elif kwarg.get('action'): ret['attrs'].append(['TCA_U32_ACT', get_tca_action(kwarg)]) ret['attrs'].append(['TCA_U32_CLASSID', kwarg['target']]) ret['attrs'].append(['TCA_U32_SEL', {'keys': kwarg['keys']}]) return ret class options(nla, nla_plus_police): nla_map = ( ('TCA_U32_UNSPEC', 'none'), ('TCA_U32_CLASSID', 'uint32'), ('TCA_U32_HASH', 'uint32'), ('TCA_U32_LINK', 'hex'), ('TCA_U32_DIVISOR', 'uint32'), ('TCA_U32_SEL', 'u32_sel'), ('TCA_U32_POLICE', 'police'), ('TCA_U32_ACT', 'tca_act_prio'), ('TCA_U32_INDEV', 'hex'), ('TCA_U32_PCNT', 'u32_pcnt'), ('TCA_U32_MARK', 'u32_mark'), ) tca_act_prio = tca_act_prio class u32_sel(nla): fields = ( ('flags', 'B'), ('offshift', 'B'), ('nkeys', 'B'), ('__align', 'x'), ('offmask', '>H'), ('off', 'H'), ('offoff', 'h'), ('hoff', 'h'), ('hmask', '>I'), ) class u32_key(nlmsg): header = None fields = ( ('key_mask', '>I'), ('key_val', '>I'), ('key_off', 'i'), ('key_offmask', 'i'), ) def encode(self): ''' Key sample:: 'keys': ['0x0006/0x00ff+8', '0x0000/0xffc0+2', '0x5/0xf+0', '0x10/0xff+33'] => 00060000/00ff0000 + 8 05000000/0f00ffc0 + 0 00100000/00ff0000 + 32 ''' def cut_field(key, separator): ''' split a field from the end of the string ''' field = '0' pos = key.find(separator) new_key = key if pos > 0: field = key[pos + 1 :] new_key = key[:pos] return (new_key, field) # 'header' array to pack keys to header = [(0, 0) for i in range(256)] keys = [] # iterate keys and pack them to the 'header' for key in self['keys']: # TODO tags: filter (key, nh) = cut_field(key, '@') # FIXME: do not ignore nh (key, offset) = cut_field(key, '+') offset = int(offset, 0) # a little trick: if you provide /00ff+8, that # really means /ff+9, so we should take it into # account (key, mask) = cut_field(key, '/') if mask[:2] == '0x': mask = mask[2:] while True: if mask[:2] == '00': offset += 1 mask = mask[2:] else: break mask = '0x' + mask mask = int(mask, 0) value = int(key, 0) bits = 24 if mask == 0 and value == 0: key = self.u32_key(data=self.data) key['key_off'] = offset key['key_mask'] = mask key['key_val'] = value keys.append(key) for bmask in struct.unpack('4B', struct.pack('>I', mask)): if bmask > 0: bvalue = (value & (bmask << bits)) >> bits header[offset] = (bvalue, bmask) offset += 1 bits -= 8 # recalculate keys from 'header' key = None value = 0 mask = 0 for offset in range(256): (bvalue, bmask) = header[offset] if bmask > 0 and key is None: key = self.u32_key(data=self.data) key['key_off'] = offset key['key_mask'] = 0 key['key_val'] = 0 bits = 24 if key is not None and bits >= 0: key['key_mask'] \|= bmask << bits key['key_val'] \|= bvalue << bits bits -= 8 if bits < 0 or offset == 255: keys.append(key) key = None if not keys: raise ValueError('no keys specified') self['nkeys'] = len(keys) # FIXME: do not hardcode flags :) self['flags'] = 1 nla.encode(self) offset = self.offset + 20 # 4 bytes header + 16 bytes fields for key in keys: key.offset = offset key.encode() offset += 16 # keys haven't header self.length = offset - self.offset struct.pack_into('H', self.data, self.offset, offset - self.offset) def decode(self): nla.decode(self) offset = self.offset + 16 self['keys'] = [] nkeys = self['nkeys'] while nkeys: key = self.u32_key(data=self.data, offset=offset) key.decode() offset += 16 self['keys'].append(key) nkeys -= 1 class u32_mark(nla): fields = (('val', 'I'), ('mask', 'I'), ('success', 'I')) class u32_pcnt(nla): fields = (('rcnt', 'Q'), ('rhit', 'Q'), ('kcnts', 'Q'))	7.91 KB	2025-05-25 12:26
📄 common.py import logging import os import re import struct from math import log as logfm from socket import inet_aton from pyroute2 import config from pyroute2.common import ( basestring, rate_suffixes, size_suffixes, time_suffixes, ) from pyroute2.netlink import nla, nla_string log = logging.getLogger(__name__) LINKLAYER_UNSPEC = 0 LINKLAYER_ETHERNET = 1 LINKLAYER_ATM = 2 ATM_CELL_SIZE = 53 ATM_CELL_PAYLOAD = 48 TCA_ACT_MAX_PRIO = 32 TIME_UNITS_PER_SEC = 1000000 try: with open('/proc/net/psched', 'r') as psched: [t2us, us2t, clock_res, wee] = [ int(i, 16) for i in psched.read().split() ] clock_factor = float(clock_res) / TIME_UNITS_PER_SEC tick_in_usec = float(t2us) / us2t * clock_factor except IOError as e: if config.uname[0] == 'Linux': log.warning("tcmsg: %s", e) log.warning("the tc subsystem functionality is limited") clock_res = 0 clock_factor = 1 tick_in_usec = 1 wee = 1000 _first_letter = re.compile('[^0-9]+') def get_hz(): if clock_res == 1000000: return wee else: return os.environ.get('HZ', 1000) def get_by_suffix(value, default, func): if not isinstance(value, basestring): return value pos = _first_letter.search(value) if pos is None: suffix = default else: pos = pos.start() value, suffix = value[:pos], value[pos:] value = int(value) return func(value, suffix) def get_size(size): return get_by_suffix(size, 'b', lambda x, y: x * size_suffixes[y]) def get_time(lat): return get_by_suffix( lat, 'ms', lambda x, y: (x * TIME_UNITS_PER_SEC) / time_suffixes[y] ) def get_rate(rate): return get_by_suffix(rate, 'bit', lambda x, y: (x * rate_suffixes[y]) / 8) def time2tick(time): # The code is ported from tc utility return int(time) * tick_in_usec def calc_xmittime(rate, size): # The code is ported from tc utility return int(time2tick(TIME_UNITS_PER_SEC * (float(size) / rate))) def percent2u32(pct): '''xlate a percentage to an uint32 value 0% -> 0 100% -> 232 - 1''' return int((232 - 1) * pct / 100) def red_eval_ewma(qmin, burst, avpkt): # The code is ported from tc utility wlog = 1 W = 0.5 a = float(burst) + 1 - float(qmin) / avpkt if a < 1: raise ValueError(f'wrong a = {a}') while wlog < 32: wlog += 1 W /= 2 if a <= (1 - pow(1 - W, burst)) / W: return wlog return -1 def red_eval_P(qmin, qmax, probability): # The code is ported from tc utility i = qmax - qmin if i <= 0: raise ValueError(f'qmax - qmin must be > 0 (got {i})') probability /= i for i in range(32): if probability > 1: break probability = 2 return i def red_eval_idle_damping(Wlog, avpkt, bps): # The code is ported from tc utility xmit_time = calc_xmittime(bps, avpkt) lW = -logfm(1.0 - 1.0 / (1 << Wlog)) / xmit_time maxtime = 31.0 / lW sbuf = [] for clog in range(32): if maxtime / (1 << clog) < 512: break if clog >= 32: return -1, sbuf for i in range(255): sbuf.append((i << clog) lW) if sbuf[i] > 31: sbuf[i] = 31 sbuf.append(31) return clog, sbuf def get_rate_parameters(kwarg): # rate and burst are required rate = get_rate(kwarg['rate']) burst = kwarg['burst'] # if peak, mtu is required peak = get_rate(kwarg.get('peak', 0)) mtu = kwarg.get('mtu', 0) if peak: assert mtu # limit OR latency is required limit = kwarg.get('limit', None) latency = get_time(kwarg.get('latency', None)) assert limit is not None or latency is not None # calculate limit from latency if limit is None: rate_limit = rate * float(latency) / TIME_UNITS_PER_SEC + burst if peak: peak_limit = peak * float(latency) / TIME_UNITS_PER_SEC + mtu if rate_limit > peak_limit: rate_limit = peak_limit limit = rate_limit return { 'rate': int(rate), 'mtu': mtu, 'buffer': calc_xmittime(rate, burst), 'limit': int(limit), } tc_flow_keys = { 'src': 0x01, 'dst': 0x02, 'proto': 0x04, 'proto-src': 0x08, 'proto-dst': 0x10, 'iif': 0x20, 'priority': 0x40, 'mark': 0x80, 'nfct': 0x0100, 'nfct-src': 0x0200, 'nfct-dst': 0x0400, 'nfct-proto-src': 0x0800, 'nfct-proto-dst': 0x1000, 'rt-classid': 0x2000, 'sk-uid': 0x4000, 'sk-gid': 0x8000, 'vlan-tag': 0x010000, 'rxhash': 0x020000, } def get_tca_keys(kwarg, name): if name not in kwarg: raise ValueError('Missing attribute: {0}'.format(name)) res = 0 keys = kwarg[name] if name == 'hash': keys = keys.split(',') for key, value in tc_flow_keys.items(): if key in keys: res \|= value return res tc_flow_modes = {'map': 0, 'hash': 1} def get_tca_mode(kwarg): if 'mode' not in kwarg: raise ValueError('Missing attribute: mode') for key, value in tc_flow_modes.items(): if key == kwarg['mode']: return value raise ValueError('Unknown flow mode {0}'.format(kwarg['mode'])) def get_tca_ops(kwarg, attrs): xor_value = 0 mask_value = 0 addend_value = 0 rshift_value = 0 for elem in kwarg['ops']: op = elem['op'] num = elem['num'] if op == 'and': mask_value = num attrs.append(['TCA_FLOW_XOR', xor_value]) attrs.append(['TCA_FLOW_MASK', mask_value]) elif op == 'or': if mask_value == 0: mask_value = (~num + 1) & 0xFFFFFFFF xor_value = num attrs.append(['TCA_FLOW_XOR', xor_value]) attrs.append(['TCA_FLOW_MASK', mask_value]) elif op == 'xor': if mask_value == 0: mask_value = 0xFFFFFFFF xor_value = num attrs.append(['TCA_FLOW_XOR', xor_value]) attrs.append(['TCA_FLOW_MASK', mask_value]) elif op == 'rshift': rshift_value = num attrs.append(['TCA_FLOW_RSHIFT', rshift_value]) elif op == 'addend': # Check if an IP was specified if isinstance(num, str) and len(num.split('.')) == 4: if num.startswith('-'): inverse = True else: inverse = False ip = num.strip('-') # Convert IP to uint32 ip = inet_aton(ip) ip = struct.unpack('>I', ip)[0] if inverse: ip = (~ip + 1) & 0xFFFFFFFF addend_value = ip else: addend_value = num attrs.append(['TCA_FLOW_ADDEND', addend_value]) tc_actions = { 'unspec': -1, # TC_ACT_UNSPEC 'ok': 0, # TC_ACT_OK 'reclassify': 1, # TC_ACT_RECLASSIFY 'shot': 2, # TC_ACT_SHOT 'drop': 2, # TC_ACT_SHOT 'pipe': 3, # TC_ACT_PIPE 'stolen': 4, # TC_ACT_STOLEN 'queued': 5, # TC_ACT_QUEUED 'repeat': 6, # TC_ACT_REPEAT 'redirect': 7, # TC_ACT_REDIRECT } class nla_plus_rtab(nla): class parms(nla): def adjust_size(self, size, mpu, linklayer): # The current code is ported from tc utility if size < mpu: size = mpu if linklayer == LINKLAYER_ATM: cells = size / ATM_CELL_PAYLOAD if size % ATM_CELL_PAYLOAD > 0: cells += 1 size = cells * ATM_CELL_SIZE return size def calc_rtab(self, kind): # The current code is ported from tc utility rtab = [] mtu = self.get('mtu', 0) or 1600 cell_log = self['%s_cell_log' % (kind)] mpu = self['%s_mpu' % (kind)] rate = self.get(kind, 'rate') # calculate cell_log if cell_log == 0: while (mtu >> cell_log) > 255: cell_log += 1 # fill up the table for i in range(256): size = self.adjust_size( (i + 1) << cell_log, mpu, LINKLAYER_ETHERNET ) rtab.append(calc_xmittime(rate, size)) self['%s_cell_align' % (kind)] = -1 self['%s_cell_log' % (kind)] = cell_log return rtab def encode(self): self.rtab = None self.ptab = None if self.get('rate', False): self.rtab = self.calc_rtab('rate') if self.get('peak', False): self.ptab = self.calc_rtab('peak') if self.get('ceil', False): self.ctab = self.calc_rtab('ceil') nla.encode(self) class rtab(nla_string): own_parent = True def encode(self): parms = ( self.parent.get_encoded('TCA_TBF_PARMS') or self.parent.get_encoded('TCA_HTB_PARMS') or self.parent.get_encoded('TCA_POLICE_TBF') ) if parms is not None: self.value = getattr(parms, self.__class__.__name__) self['value'] = struct.pack( 'I' * 256, (int(x) for x in self.value) ) nla_string.encode(self) def decode(self): nla_string.decode(self) parms = ( self.parent.get_attr('TCA_TBF_PARMS') or self.parent.get_attr('TCA_HTB_PARMS') or self.parent.get_attr('TCA_POLICE_TBF') ) if parms is not None: rtab = struct.unpack( 'I' (len(self['value']) / 4), self['value'] ) self.value = rtab setattr(parms, self.__class__.__name__, rtab) class ptab(rtab): pass class ctab(rtab): pass class stats2(nla): nla_map = ( ('TCA_STATS_UNSPEC', 'none'), ('TCA_STATS_BASIC', 'basic'), ('TCA_STATS_RATE_EST', 'rate_est'), ('TCA_STATS_QUEUE', 'queue'), ('TCA_STATS_APP', 'stats_app'), ) class basic(nla): fields = (('bytes', 'Q'), ('packets', 'I')) class rate_est(nla): fields = (('bps', 'I'), ('pps', 'I')) class queue(nla): fields = ( ('qlen', 'I'), ('backlog', 'I'), ('drops', 'I'), ('requeues', 'I'), ('overlimits', 'I'), ) class stats_app(nla.hex): pass	10.48 KB	2025-05-25 12:26
📄 common_act.py from pyroute2.netlink import nla from pyroute2.netlink.rtnl.tcmsg import ( act_bpf, act_connmark, act_gact, act_mirred, act_police, act_skbedit, act_vlan, ) from pyroute2.netlink.rtnl.tcmsg.common import TCA_ACT_MAX_PRIO, stats2 plugins = { 'gact': act_gact, 'bpf': act_bpf, 'police': act_police, 'mirred': act_mirred, 'connmark': act_connmark, 'vlan': act_vlan, 'skbedit': act_skbedit, } class nla_plus_tca_act_opt(object): @staticmethod def get_act_options(self, argv, *kwarg): kind = self.get_attr('TCA_ACT_KIND') if kind in plugins: return plugins[kind].options return self.hex class tca_act_prio(nla): nla_map = tuple( [('TCA_ACT_PRIO_%i' % x, 'tca_act') for x in range(TCA_ACT_MAX_PRIO)] ) class tca_act(nla, nla_plus_tca_act_opt): nla_map = ( ('TCA_ACT_UNSPEC', 'none'), ('TCA_ACT_KIND', 'asciiz'), ('TCA_ACT_OPTIONS', 'get_act_options'), ('TCA_ACT_INDEX', 'hex'), ('TCA_ACT_STATS', 'stats2'), ) stats2 = stats2 def get_act_parms(kwarg): if 'kind' not in kwarg: raise Exception('action requires "kind" parameter') if kwarg['kind'] in plugins: return plugins[kwarg['kind']].get_parameters(kwarg) else: return [] # All filters can use any act type, this is a generic parser for all def get_tca_action(kwarg): ret = {'attrs': []} act = kwarg.get('action', 'drop') # convert simple action='..' to kwarg style if isinstance(act, str): act = {'kind': 'gact', 'action': act} # convert single dict action to first entry in a list of actions acts = act if isinstance(act, list) else [act] for i, act in enumerate(acts, start=1): opt = { 'attrs': [ ['TCA_ACT_KIND', act['kind']], ['TCA_ACT_OPTIONS', get_act_parms(act)], ] } ret['attrs'].append(['TCA_ACT_PRIO_%d' % i, opt]) return ret	2.01 KB	2025-05-25 12:26
📄 common_ematch.py from pyroute2.netlink.rtnl.tcmsg import em_cmp, em_ipset, em_meta plugins = { # 0: em_container, 1: em_cmp, # 2: em_nbyte, # 3: em_u32, 4: em_meta, # 5: em_text, # 6: em_vlan, # 7: em_canid, 8: em_ipset, # 9: em_ipt, } plugins_translate = { 'container': 0, 'cmp': 1, 'nbyte': 2, 'u32': 3, 'meta': 4, 'text': 5, 'vlan': 6, 'canid': 7, 'ipset': 8, 'ipt': 9, } TCF_EM_REL_END = 0 TCF_EM_REL_AND = 1 TCF_EM_REL_OR = 2 TCF_EM_INVERSE_MASK = 4 RELATIONS_DICT = { 'and': TCF_EM_REL_AND, 'AND': TCF_EM_REL_AND, '&&': TCF_EM_REL_AND, 'or': TCF_EM_REL_OR, 'OR': TCF_EM_REL_OR, '\|\|': TCF_EM_REL_OR, } class nla_plus_tcf_ematch_opt(object): @staticmethod def parse_ematch_options(self, argv, *kwarg): if 'kind' not in self: raise ValueError('ematch requires "kind" parameter') kind = self['kind'] if kind in plugins: ret = plugins[kind].data(data=argv[0]) ret.decode() return ret return self.hex def get_ematch_parms(kwarg): if 'kind' not in kwarg: raise ValueError('ematch requires "kind" parameter') if kwarg['kind'] in plugins: return plugins[kwarg['kind']].get_parameters(kwarg) else: return [] def get_tcf_ematches(kwarg): ret = {'attrs': []} matches = [] header = {'nmatches': 0, 'progid': 0} # Get the number of expressions expr_count = len(kwarg['match']) header['nmatches'] = expr_count # Load plugin and transfer data for i in range(0, expr_count): match = {'matchid': 0, 'kind': None, 'flags': 0, 'pad': 0, 'opt': None} cur_match = kwarg['match'][i] # Translate string kind into numeric kind kind = plugins_translate[cur_match['kind']] match['kind'] = kind data = plugins[kind].data() data.setvalue(cur_match) data.encode() # Add ematch encoded data match['opt'] = data.data # Safety check if i == expr_count - 1 and 'relation' in cur_match: raise ValueError('Could not set a relation to the last expression') if i < expr_count - 1 and 'relation' not in cur_match: raise ValueError( 'You must specify a relation for every expression' ' except the last one' ) # Set relation to flags if 'relation' in cur_match: relation = cur_match['relation'] if relation in RELATIONS_DICT: match['flags'] \|= RELATIONS_DICT.get(relation) else: raise ValueError('Unknown relation {0}'.format(relation)) else: match['flags'] = TCF_EM_REL_END # Handle inverse flag if 'inverse' in cur_match: if cur_match['inverse']: match['flags'] \|= TCF_EM_INVERSE_MASK # Append new match to list of matches matches.append(match) ret['attrs'].append(['TCA_EMATCH_TREE_HDR', header]) ret['attrs'].append(['TCA_EMATCH_TREE_LIST', matches]) return ret	3.07 KB	2025-05-25 12:26
📄 em_cmp.py from pyroute2.netlink import nla TCF_EM_OPND_EQ = 0 TCF_EM_OPND_GT = 1 TCF_EM_OPND_LT = 2 OPERANDS_DICT = { TCF_EM_OPND_EQ: ('eq', '='), TCF_EM_OPND_GT: ('gt', '>'), TCF_EM_OPND_LT: ('lt', '<'), } # align types TCF_EM_ALIGN_U8 = 1 TCF_EM_ALIGN_U16 = 2 TCF_EM_ALIGN_U32 = 4 ALIGNS_DICT = { TCF_EM_ALIGN_U8: 'u8', TCF_EM_ALIGN_U16: 'u16', TCF_EM_ALIGN_U32: 'u32', } # layer types TCF_LAYER_LINK = 0 TCF_LAYER_NETWORK = 1 TCF_LAYER_TRANSPORT = 2 LAYERS_DICT = { TCF_LAYER_LINK: ('link', 'eth'), TCF_LAYER_NETWORK: ('network', 'ip'), TCF_LAYER_TRANSPORT: ('transport', 'tcp'), } # see tc_em_cmp.h TCF_EM_CMP_TRANS = 1 class data(nla): fields = ( ('val', 'I'), ('mask', 'I'), ('off', 'H'), ('align_flags', 'B'), ('layer_opnd', 'B'), ) def decode(self): self.header = None self.length = 24 nla.decode(self) self['align'] = self['align_flags'] & 0x0F self['flags'] = (self['align_flags'] & 0xF0) >> 4 self['layer'] = self['layer_opnd'] & 0x0F self['opnd'] = (self['layer_opnd'] & 0xF0) >> 4 del self['layer_opnd'] del self['align_flags'] # Perform translation for readability with nldecap self['layer'] = 'TCF_LAYER_{}'.format( LAYERS_DICT[self['layer']][0] ).upper() self['align'] = 'TCF_EM_ALIGN_{}'.format( ALIGNS_DICT[self['align']] ).upper() self['opnd'] = 'TCF_EM_OPND_{}'.format( OPERANDS_DICT[self['opnd']][0] ).upper() def encode(self): # Set default values self['layer_opnd'] = 0 self['align_flags'] = 0 # Build align_flags byte if 'trans' in self: self['align_flags'] = TCF_EM_CMP_TRANS << 4 for k, v in ALIGNS_DICT.items(): if self['align'].lower() == v: self['align_flags'] \|= k break # Build layer_opnd byte if isinstance(self['opnd'], int): self['layer_opnd'] = self['opnd'] << 4 else: for k, v in OPERANDS_DICT.items(): if self['opnd'].lower() in v: self['layer_opnd'] = k << 4 break # Layer code if isinstance(self['layer'], int): self['layer_opnd'] \|= self['layer'] else: for k, v in LAYERS_DICT.items(): if self['layer'].lower() in v: self['layer_opnd'] \|= k break self['off'] = self.get('offset', 0) self['val'] = self.get('value', 0) nla.encode(self) # Patch NLA structure self['header']['length'] -= 4 self.data = self.data[4:]	2.71 KB	2025-05-25 12:26
📄 em_ipset.py from pyroute2.netlink import nlmsg_base, nlmsg_encoder_generic # see em_ipset.c IPSET_DIM = { 'IPSET_DIM_ZERO': 0, 'IPSET_DIM_ONE': 1, 'IPSET_DIM_TWO': 2, 'IPSET_DIM_THREE': 3, 'IPSET_DIM_MAX': 6, } TCF_IPSET_MODE_DST = 0 TCF_IPSET_MODE_SRC = 2 def get_parameters(kwarg): ret = {'attrs': []} attrs_map = ( ('matchid', 'TCF_EM_MATCHID'), ('kind', 'TCF_EM_KIND'), ('flags', 'TCF_EM_FLAGS'), ('pad', 'TCF_EM_PAD'), ) for k, v in attrs_map: r = kwarg.get(k, None) if r is not None: ret['attrs'].append([v, r]) return ret class data(nlmsg_base, nlmsg_encoder_generic): fields = ( ('ip_set_index', 'H'), ('ip_set_dim', 'B'), ('ip_set_flags', 'B'), ) def encode(self): flags, dim = self._get_ip_set_parms() self['ip_set_index'] = self['index'] self['ip_set_dim'] = dim self['ip_set_flags'] = flags nlmsg_base.encode(self) def _get_ip_set_parms(self): flags = 0 dim = 0 mode = self['mode'] # Split to get dimension modes = mode.split(',') dim = len(modes) if dim > IPSET_DIM['IPSET_DIM_MAX']: raise ValueError( 'IPSet dimension could not be greater than {0}'.format( IPSET_DIM['IPSET_DIM_MAX'] ) ) for i in range(0, dim): if modes[i] == 'dst': flags \|= TCF_IPSET_MODE_DST << i elif modes[i] == 'src': flags \|= TCF_IPSET_MODE_SRC << i else: raise ValueError('Unknown IP set mode "{0}"'.format(modes[i])) return (flags, dim)	1.7 KB	2025-05-25 12:26
📄 em_meta.py from struct import pack, unpack from pyroute2.netlink import nla TCF_EM_OPND_EQ = 0 TCF_EM_OPND_GT = 1 TCF_EM_OPND_LT = 2 OPERANDS_DICT = { TCF_EM_OPND_EQ: ('eq', '='), TCF_EM_OPND_GT: ('gt', '>'), TCF_EM_OPND_LT: ('lt', '<'), } # meta types TCF_META_TYPE_VAR = 0 TCF_META_TYPE_INT = 1 TCF_META_ID_MASK = 0x7FF TCF_META_TYPE_MASK = 0xF << 12 # see tc_em_meta.h META_ID = { 'value': 0, 'random': 1, 'loadavg_0': 2, 'loadavg_1': 3, 'loadavg_2': 4, 'dev': 5, 'priority': 6, 'protocol': 7, 'pkttype': 8, 'pktlen': 9, 'datalen': 10, 'maclen': 11, 'nfmark': 12, 'tcindex': 13, 'rtclassid': 14, 'rtiif': 15, 'sk_family': 16, 'sk_state': 17, 'sk_reuse': 18, 'sk_bound_if': 19, 'sk_refcnt': 20, 'sk_shutdown': 21, 'sk_proto': 22, 'sk_type': 23, 'sk_rcvbuf': 24, 'sk_rmem_alloc': 25, 'sk_wmem_alloc': 26, 'sk_omem_alloc': 27, 'sk_wmem_queued': 28, 'sk_rcv_qlen': 29, 'sk_snd_qlen': 30, 'sk_err_qlen': 31, 'sk_forward_allocs': 32, 'sk_sndbuf': 33, 'sk_allocs': 34, 'sk_route_caps': 35, 'sk_hash': 36, 'sk_lingertime': 37, 'sk_ack_backlog': 38, 'sk_max_ack_backlog': 39, 'sk_prio': 40, 'sk_rcvlowat': 41, 'sk_rcvtimeo': 42, 'sk_sndtimeo': 43, 'sk_sendmsg_off': 44, 'sk_write_pending': 45, 'vlan_tag': 46, 'rxhash': 47, } strings_meta = ('dev', 'sk_bound_if') class data(nla): nla_map = ( ('TCA_EM_META_UNSPEC', 'none'), ('TCA_EM_META_HDR', 'tca_em_meta_header_parse'), ('TCA_EM_META_LVALUE', 'uint32'), ('TCA_EM_META_RVALUE', 'hex'), ) def decode(self): self.header = None self.length = 24 nla.decode(self) # Patch to have a better view in nldecap attrs = dict(self['attrs']) rvalue = attrs.get('TCA_EM_META_RVALUE') meta_hdr = attrs.get('TCA_EM_META_HDR') meta_id = meta_hdr['id'] rvalue = bytearray.fromhex(rvalue.replace(':', '')) if meta_id == 'TCF_META_TYPE_VAR': rvalue.decode('utf-8') if meta_id == 'TCF_META_TYPE_INT': rvalue = unpack('<I', rvalue)[0] self['attrs'][2] = ('TCA_EM_META_RVALUE', rvalue) def encode(self): if 'object' not in self: raise ValueError('An object definition must be given!') if 'value' not in self: raise ValueError('A value must be given!') # The value can either be a string or an int depending # on the selected meta kind. # It is really important to distinct them otherwise # the kernel won't be nice with us... value = self['value'] kind = self['object'].get('kind') if not kind: raise ValueError('Not meta kind specified!') else: if kind in strings_meta: if not isinstance(value, str): raise ValueError( '{} kinds have to use string value!'.format( ' and '.join(strings_meta) ) ) else: value = value.encode('utf-8') else: if not isinstance(value, int): raise ValueError( 'Invalid value specified, it must ' 'be an integer' ) else: value = pack('<I', value) self['attrs'].append(['TCA_EM_META_HDR', self['object']]) self['attrs'].append(['TCA_EM_META_LVALUE', self.get('mask', 0)]) self['attrs'].append(['TCA_EM_META_RVALUE', value]) nla.encode(self) # Patch NLA structure self['header']['length'] -= 4 self.data = self.data[4:] class tca_em_meta_header_parse(nla): fields = ( ('kind', 'H'), ('shift', 'B'), ('opnd', 'B'), ('id', 'H'), ('pad', 'H'), ) def decode(self): nla.decode(self) self['id'] = (self['kind'] & TCF_META_TYPE_MASK) >> 12 if self['id'] == TCF_META_TYPE_VAR: self['id'] = 'TCF_META_TYPE_VAR' elif self['id'] == TCF_META_TYPE_INT: self['id'] = 'TCF_META_TYPE_INT' else: pass self['kind'] &= TCF_META_ID_MASK for k, v in META_ID.items(): if self['kind'] == v: self['kind'] = 'TCF_META_ID_{}'.format(k.upper()) fmt = 'TCF_EM_OPND_{}'.format( OPERANDS_DICT[self['opnd']][0].upper() ) self['opnd'] = fmt del self['pad'] def encode(self): if not isinstance(self['kind'], str): raise ValueError("kind' keywords must be set!") kind = self['kind'].lower() if kind in strings_meta: self['id'] = TCF_META_TYPE_VAR else: self['id'] = TCF_META_TYPE_INT self['id'] <<= 12 for k, v in META_ID.items(): if kind == k: self['kind'] = self['id'] \| v break if isinstance(self['opnd'], str): for k, v in OPERANDS_DICT.items(): if self['opnd'].lower() in v: self['opnd'] = k break # Perform sanity checks on 'shift' value if isinstance(self['shift'], str): # If it fails, it will raise a ValueError # which is what we want self['shift'] = int(self['shift']) if not 0 <= self['shift'] <= 255: raise ValueError( "'shift' value must be between" "0 and 255 included!" ) nla.encode(self)	5.79 KB	2025-05-25 12:26
📄 sched_bpf.py ''' ''' from socket import htons from pyroute2.netlink import NLA_F_NESTED, nla from pyroute2.netlink.rtnl import TC_H_ROOT from pyroute2.netlink.rtnl.tcmsg.act_police import ( get_parameters as ap_parameters, ) from pyroute2.netlink.rtnl.tcmsg.act_police import nla_plus_police from pyroute2.netlink.rtnl.tcmsg.common import TCA_ACT_MAX_PRIO, stats2 from pyroute2.netlink.rtnl.tcmsg.common_act import ( get_tca_action, nla_plus_tca_act_opt, ) from pyroute2.protocols import ETH_P_ALL parent = TC_H_ROOT TCA_BPF_FLAG_ACT_DIRECT = 1 def fix_request(request): if 'rate' in request: del request['rate'] if 'info' not in request: request['info'] = htons( request.pop('protocol', ETH_P_ALL) & 0xFFFF ) \| ((request.pop('prio', 0) << 16) & 0xFFFF0000) def get_parameters(kwarg): ret = {'attrs': []} attrs_map = ( # ('action', 'TCA_BPF_ACT'), # ('police', 'TCA_BPF_POLICE'), ('classid', 'TCA_BPF_CLASSID'), ('fd', 'TCA_BPF_FD'), ('name', 'TCA_BPF_NAME'), ('flags', 'TCA_BPF_FLAGS'), ) act = kwarg.get('action') if act: ret['attrs'].append(['TCA_BPF_ACT', get_tca_action(kwarg)]) if kwarg.get('rate'): ret['attrs'].append(['TCA_BPF_POLICE', ap_parameters(kwarg)]) kwarg['flags'] = kwarg.get('flags', 0) if kwarg.get('direct_action', False): kwarg['flags'] \|= TCA_BPF_FLAG_ACT_DIRECT for k, v in attrs_map: r = kwarg.get(k, None) if r is not None: ret['attrs'].append([v, r]) return ret class options(nla, nla_plus_police): nla_map = ( ('TCA_BPF_UNSPEC', 'none'), ('TCA_BPF_ACT', 'bpf_act'), ('TCA_BPF_POLICE', 'police'), ('TCA_BPF_CLASSID', 'uint32'), ('TCA_BPF_OPS_LEN', 'uint32'), ('TCA_BPF_OPS', 'uint32'), ('TCA_BPF_FD', 'uint32'), ('TCA_BPF_NAME', 'asciiz'), ('TCA_BPF_FLAGS', 'uint32'), ) class bpf_act(nla): nla_flags = NLA_F_NESTED nla_map = tuple( [ ('TCA_ACT_PRIO_%i' % x, 'tca_act_bpf') for x in range(TCA_ACT_MAX_PRIO) ] ) class tca_act_bpf(nla, nla_plus_tca_act_opt): nla_map = ( ('TCA_ACT_UNSPEC', 'none'), ('TCA_ACT_KIND', 'asciiz'), ('TCA_ACT_OPTIONS', 'get_act_options'), ('TCA_ACT_INDEX', 'hex'), ('TCA_ACT_STATS', 'get_stats2'), ) @staticmethod def get_stats2(self, argv, *kwarg): return stats2	2.57 KB	2025-05-25 12:26
📄 sched_cake.py ''' cake ++++ Usage: # Imports from pyroute2 import IPRoute # Add cake with 2048kbit of bandwidth capacity with IPRoute() as ipr: # Get interface index index = ipr.link_lookup(ifname='lo') ipr.tc('add', kind='cake', index=index, bandwidth='2048kbit') # Same with 15mbit of bandwidth capacity with IPRoute() as ipr: # Get interface index index = ipr.link_lookup(ifname='lo') ipr.tc('add', kind='cake', index=index, bandwidth='15mbit') # If you don't know the bandwidth capacity, use autorate with IPRoute() as ipr: # Get interface index index = ipr.link_lookup(ifname='lo') ipr.tc('add', kind='cake', index=index, bandwidth='unlimited', autorate=True) # If you want to tune ATM properties use: # atm_mode=False # For no ATM tuning # atm_mode=True # For ADSL tuning # atm_mode='ptm' # For VDSL2 tuning with IPRoute() as ipr: # Get interface index index = ipr.link_lookup(ifname='lo') ipr.tc('add', kind='cake', index=index, bandwidth='unlimited', autorate=True, atm_mode=True) # Complex example which has no-sense with IPRoute() as ipr: # Get interface index index = ipr.link_lookup(ifname='lo') ipr.tc('add', kind='cake', index=index, bandwidth='unlimited', autorate=True, nat=True, rtt='interplanetary', target=10000, flow_mode='dsthost', diffserv_mode='precedence', fwmark=0x1337) NOTES: Here is the list of all supported options with their values: - ack_filter: False, True or 'aggressive' (False by default) - atm_mode: False, True or 'ptm' (False by default) - autorate: False or True (False by default) - bandwidth: any integer, 'N[kbit\|mbit\|gbit]' or 'unlimited' - diffserv_mode: 'diffserv3', 'diffserv4', 'diffserv8', 'besteffort', 'precedence' ('diffserv3' by default) - ingress: False or True (False by default) - overhead: any integer between -64 and 256 inclusive (0 by default) - flow_mode: 'flowblind', 'srchost', 'dsthost', 'hosts', 'flows', 'dual-srchost', 'dual-dsthost', 'triple-isolate' ('triple-isolate' by default) - fwmark: any integer (0 by default) - memlimit: any integer (by default, calculated based on the bandwidth and RTT settings) - mpu: any integer between 0 and 256 inclusive (0 by default) - nat: False or True (False by default) - raw: False or True (True by default) - rtt: any integer or 'datacentre', 'lan', 'metro', 'regional', 'internet', 'oceanic', 'satellite', 'interplanetary' ('internet' by default) - split_gso: False or True (True by default) - target: any integer (5000 by default) - wash: False or True (False by default) ''' from pyroute2.netlink import nla from pyroute2.netlink.rtnl import TC_H_ROOT # Defines from pkt_sched.h CAKE_FLOW_NONE = 0 CAKE_FLOW_SRC_IP = 1 CAKE_FLOW_DST_IP = 2 CAKE_FLOW_HOSTS = 3 CAKE_FLOW_FLOWS = 4 CAKE_FLOW_DUAL_SRC = 5 CAKE_FLOW_DUAL_DST = 6 CAKE_FLOW_TRIPLE = 7 CAKE_DIFFSERV_DIFFSERV3 = 0 CAKE_DIFFSERV_DIFFSERV4 = 1 CAKE_DIFFSERV_DIFFSERV8 = 2 CAKE_DIFFSERV_BESTEFFORT = 3 CAKE_DIFFSERV_PRECEDENCE = 4 CAKE_ACK_NONE = 0 CAKE_ACK_FILTER = 1 CAKE_ACK_AGGRESSIVE = 2 CAKE_ATM_NONE = 0 CAKE_ATM_ATM = 1 CAKE_ATM_PTM = 2 TCA_CAKE_MAX_TINS = 8 def fix_request(request): if 'rate' in request: del request['rate'] if 'parent' not in request or request['parent'] == 0: request['parent'] = TC_H_ROOT def convert_bandwidth(value): types = [('kbit', 1000), ('mbit', 1000000), ('gbit', 1000000000)] if 'unlimited' == value: return 0 try: # Value is passed as an int x = int(value) return x >> 3 except ValueError: value = value.lower() for t, mul in types: if len(value.split(t)) == 2: x = int(value.split(t)[0]) * mul return x >> 3 raise ValueError( 'Invalid bandwidth value. Specify either an integer, ' '"unlimited" or a value with "kbit", "mbit" or ' '"gbit" appended' ) def convert_rtt(value): types = { 'datacentre': 100, 'lan': 1000, 'metro': 10000, 'regional': 30000, 'internet': 100000, 'oceanic': 300000, 'satellite': 1000000, 'interplanetary': 3600000000, } try: # Value is passed as an int x = int(value) return x except ValueError: rtt = types.get(value.lower()) if rtt is not None: return rtt raise ValueError( 'Invalid rtt value. Specify either an integer (us), ' 'or datacentre, lan, metro, regional, internet, ' 'oceanic or interplanetary.' ) def convert_atm(value): if isinstance(value, bool): if not value: return CAKE_ATM_NONE else: return CAKE_ATM_ATM else: if value == 'ptm': return CAKE_ATM_PTM raise ValueError('Invalid ATM value!') def convert_flowmode(value): modes = { 'flowblind': CAKE_FLOW_NONE, 'srchost': CAKE_FLOW_SRC_IP, 'dsthost': CAKE_FLOW_DST_IP, 'hosts': CAKE_FLOW_HOSTS, 'flows': CAKE_FLOW_FLOWS, 'dual-srchost': CAKE_FLOW_DUAL_SRC, 'dual-dsthost': CAKE_FLOW_DUAL_DST, 'triple-isolate': CAKE_FLOW_TRIPLE, } res = modes.get(value.lower()) if res: return res raise ValueError( 'Invalid flow mode specified! See tc-cake man ' 'page for valid values.' ) def convert_diffserv(value): modes = { 'diffserv3': CAKE_DIFFSERV_DIFFSERV3, 'diffserv4': CAKE_DIFFSERV_DIFFSERV4, 'diffserv8': CAKE_DIFFSERV_DIFFSERV8, 'besteffort': CAKE_DIFFSERV_BESTEFFORT, 'precedence': CAKE_DIFFSERV_PRECEDENCE, } res = modes.get(value.lower()) if res is not None: return res raise ValueError( 'Invalid diffserv mode specified! See tc-cake man ' 'page for valid values.' ) def convert_ackfilter(value): if isinstance(value, bool): if not value: return CAKE_ACK_NONE else: return CAKE_ACK_FILTER else: if value == 'aggressive': return CAKE_ACK_AGGRESSIVE raise ValueError('Invalid ACK filter!') def check_range(name, value, start, end): if not isinstance(value, int): raise ValueError('{} value must be an integer'.format(name)) if not start <= value <= end: raise ValueError( '{0} value must be between {1} and {2} ' 'inclusive.'.format(name, start, end) ) def get_parameters(kwarg): ret = {'attrs': []} attrs_map = ( ('ack_filter', 'TCA_CAKE_ACK_FILTER'), ('atm_mode', 'TCA_CAKE_ATM'), ('autorate', 'TCA_CAKE_AUTORATE'), ('bandwidth', 'TCA_CAKE_BASE_RATE64'), ('diffserv_mode', 'TCA_CAKE_DIFFSERV_MODE'), ('ingress', 'TCA_CAKE_INGRESS'), ('overhead', 'TCA_CAKE_OVERHEAD'), ('flow_mode', 'TCA_CAKE_FLOW_MODE'), ('fwmark', 'TCA_CAKE_FWMARK'), ('memlimit', 'TCA_CAKE_MEMORY'), ('mpu', 'TCA_CAKE_MPU'), ('nat', 'TCA_CAKE_NAT'), ('raw', 'TCA_CAKE_RAW'), ('rtt', 'TCA_CAKE_RTT'), ('split_gso', 'TCA_CAKE_SPLIT_GSO'), ('target', 'TCA_CAKE_TARGET'), ('wash', 'TCA_CAKE_WASH'), ) for k, v in attrs_map: r = kwarg.get(k, None) if r is not None: if k == 'bandwidth': r = convert_bandwidth(r) elif k == 'rtt': r = convert_rtt(r) elif k == 'atm_mode': r = convert_atm(r) elif k == 'flow_mode': r = convert_flowmode(r) elif k == 'diffserv_mode': r = convert_diffserv(r) elif k == 'ack_filter': r = convert_ackfilter(r) elif k == 'mpu': check_range(k, r, 0, 256) elif k == 'overhead': check_range(k, r, -64, 256) ret['attrs'].append([v, r]) return ret class options(nla): nla_map = ( ('TCA_CAKE_UNSPEC', 'none'), ('TCA_CAKE_PAD', 'uint64'), ('TCA_CAKE_BASE_RATE64', 'uint64'), ('TCA_CAKE_DIFFSERV_MODE', 'uint32'), ('TCA_CAKE_ATM', 'uint32'), ('TCA_CAKE_FLOW_MODE', 'uint32'), ('TCA_CAKE_OVERHEAD', 'int32'), ('TCA_CAKE_RTT', 'uint32'), ('TCA_CAKE_TARGET', 'uint32'), ('TCA_CAKE_AUTORATE', 'uint32'), ('TCA_CAKE_MEMORY', 'uint32'), ('TCA_CAKE_NAT', 'uint32'), ('TCA_CAKE_RAW', 'uint32'), ('TCA_CAKE_WASH', 'uint32'), ('TCA_CAKE_MPU', 'uint32'), ('TCA_CAKE_INGRESS', 'uint32'), ('TCA_CAKE_ACK_FILTER', 'uint32'), ('TCA_CAKE_SPLIT_GSO', 'uint32'), ('TCA_CAKE_FWMARK', 'uint32'), ) def encode(self): # Set default Auto-Rate value if not self.get_attr('TCA_CAKE_AUTORATE'): self['attrs'].append(['TCA_CAKE_AUTORATE', 0]) nla.encode(self) class stats2(nla): nla_map = ( ('TCA_STATS_UNSPEC', 'none'), ('TCA_STATS_BASIC', 'basic'), ('TCA_STATS_RATE_EST', 'rate_est'), ('TCA_STATS_QUEUE', 'queue'), ('TCA_STATS_APP', 'stats_app'), ) class basic(nla): fields = (('bytes', 'Q'), ('packets', 'I')) class rate_est(nla): fields = (('bps', 'I'), ('pps', 'I')) class queue(nla): fields = ( ('qlen', 'I'), ('backlog', 'I'), ('drops', 'I'), ('requeues', 'I'), ('overlimits', 'I'), ) class stats_app(nla): nla_map = ( ('__TCA_CAKE_STATS_INVALID', 'none'), ('TCA_CAKE_STATS_PAD', 'hex'), ('TCA_CAKE_STATS_CAPACITY_ESTIMATE64', 'uint64'), ('TCA_CAKE_STATS_MEMORY_LIMIT', 'uint32'), ('TCA_CAKE_STATS_MEMORY_USED', 'uint32'), ('TCA_CAKE_STATS_AVG_NETOFF', 'uint32'), ('TCA_CAKE_STATS_MAX_NETLEN', 'uint32'), ('TCA_CAKE_STATS_MAX_ADJLEN', 'uint32'), ('TCA_CAKE_STATS_MIN_NETLEN', 'uint32'), ('TCA_CAKE_STATS_MIN_ADJLEN', 'uint32'), ('TCA_CAKE_STATS_TIN_STATS', 'tca_parse_tins'), ('TCA_CAKE_STATS_DEFICIT', 'uint32'), ('TCA_CAKE_STATS_COBALT_COUNT', 'uint32'), ('TCA_CAKE_STATS_DROPPING', 'uint32'), ('TCA_CAKE_STATS_DROP_NEXT_US', 'uint32'), ('TCA_CAKE_STATS_P_DROP', 'uint32'), ('TCA_CAKE_STATS_BLUE_TIMER_US', 'uint32'), ) class tca_parse_tins(nla): nla_map = tuple( [ ('TCA_CAKE_TIN_STATS_%i' % x, 'tca_parse_tin_stats') for x in range(TCA_CAKE_MAX_TINS) ] ) class tca_parse_tin_stats(nla): nla_map = ( ('__TCA_CAKE_TIN_STATS_INVALID', 'none'), ('TCA_CAKE_TIN_STATS_PAD', 'hex'), ('TCA_CAKE_TIN_STATS_SENT_PACKETS', 'uint32'), ('TCA_CAKE_TIN_STATS_SENT_BYTES64', 'uint64'), ('TCA_CAKE_TIN_STATS_DROPPED_PACKETS', 'uint32'), ('TCA_CAKE_TIN_STATS_DROPPED_BYTES64', 'uint64'), ('TCA_CAKE_TIN_STATS_ACKS_DROPPED_PACKETS', 'uint32'), ('TCA_CAKE_TIN_STATS_ACKS_DROPPED_BYTES64', 'uint64'), ('TCA_CAKE_TIN_STATS_ECN_MARKED_PACKETS', 'uint32'), ('TCA_CAKE_TIN_STATS_ECN_MARKED_BYTES64', 'uint64'), ('TCA_CAKE_TIN_STATS_BACKLOG_PACKETS', 'uint32'), ('TCA_CAKE_TIN_STATS_BACKLOG_BYTES', 'uint32'), ('TCA_CAKE_TIN_STATS_THRESHOLD_RATE64', 'uint64'), ('TCA_CAKE_TIN_STATS_TARGET_US', 'uint32'), ('TCA_CAKE_TIN_STATS_INTERVAL_US', 'uint32'), ('TCA_CAKE_TIN_STATS_WAY_INDIRECT_HITS', 'uint32'), ('TCA_CAKE_TIN_STATS_WAY_MISSES', 'uint32'), ('TCA_CAKE_TIN_STATS_WAY_COLLISIONS', 'uint32'), ('TCA_CAKE_TIN_STATS_PEAK_DELAY_US', 'uint32'), ('TCA_CAKE_TIN_STATS_AVG_DELAY_US', 'uint32'), ('TCA_CAKE_TIN_STATS_BASE_DELAY_US', 'uint32'), ('TCA_CAKE_TIN_STATS_SPARSE_FLOWS', 'uint32'), ('TCA_CAKE_TIN_STATS_BULK_FLOWS', 'uint32'), ('TCA_CAKE_TIN_STATS_UNRESPONSIVE_FLOWS', 'uint32'), ('TCA_CAKE_TIN_STATS_MAX_SKBLEN', 'uint32'), ('TCA_CAKE_TIN_STATS_FLOW_QUANTUM', 'uint32'), )	12.62 KB	2025-05-25 12:26
📄 sched_choke.py ''' choke +++++ Parameters: * `limit` (required) -- int * `bandwith` (required) -- str/int * `min` -- int * `max` -- int * `avpkt` -- str/int, packet size * `burst` -- int * `probability` -- float * `ecn` -- bool Example:: ip.tc('add', 'choke', interface, limit=5500, bandwith="10mbit", ecn=True) ''' import logging import struct from pyroute2.netlink import nla, nla_string from pyroute2.netlink.rtnl import TC_H_ROOT from pyroute2.netlink.rtnl.tcmsg.common import ( get_rate, get_size, red_eval_ewma, red_eval_idle_damping, red_eval_P, ) from pyroute2.netlink.rtnl.tcmsg.common import stats2 as c_stats2 log = logging.getLogger(__name__) parent = TC_H_ROOT def get_parameters(kwarg): # The code is ported from iproute2 avpkt = 1000 probability = 0.02 opt = { 'limit': kwarg['limit'], # required 'qth_min': kwarg.get('min', 0), 'qth_max': kwarg.get('max', 0), 'Wlog': 0, 'Plog': 0, 'Scell_log': 0, 'flags': 1 if kwarg.get('ecn') else 0, } rate = get_rate(kwarg['bandwith']) # required burst = kwarg.get('burst', 0) avpkt = get_size(kwarg.get('avpkt', 1000)) probability = kwarg.get('probability', 0.02) if not opt['qth_max']: opt['qth_max'] = opt['limit'] // 4 if not opt['qth_min']: opt['qth_min'] = opt['qth_max'] // 3 if not burst: burst = (2 * opt['qth_min'] + opt['qth_max']) // 3 if opt['qth_max'] > opt['limit']: raise Exception('max is larger than limit') if opt['qth_min'] >= opt['qth_max']: raise Exception('min is not smaller than max') # Wlog opt['Wlog'] = red_eval_ewma(opt['qth_min'] * avpkt, burst, avpkt) if opt['Wlog'] < 0: raise Exception('failed to calculate EWMA') elif opt['Wlog'] > 10: log.warning('choke: burst %s seems to be too large' % burst) # Plog opt['Plog'] = red_eval_P( opt['qth_min'] * avpkt, opt['qth_max'] * avpkt, probability ) if opt['Plog'] < 0: raise Exception('choke: failed to calculate probability') # Scell_log, stab opt['Scell_log'], stab = red_eval_idle_damping(opt['Wlog'], avpkt, rate) if opt['Scell_log'] < 0: raise Exception('choke: failed to calculate idle damping table') return { 'attrs': [ ['TCA_CHOKE_PARMS', opt], ['TCA_CHOKE_STAB', stab], ['TCA_CHOKE_MAX_P', int(probability * pow(2, 32))], ] } class options(nla): nla_map = ( ('TCA_CHOKE_UNSPEC', 'none'), ('TCA_CHOKE_PARMS', 'qopt'), ('TCA_CHOKE_STAB', 'stab'), ('TCA_CHOKE_MAX_P', 'uint32'), ) class qopt(nla): fields = ( ('limit', 'I'), ('qth_min', 'I'), ('qth_max', 'I'), ('Wlog', 'B'), ('Plog', 'B'), ('Scell_log', 'B'), ('flags', 'B'), ) class stab(nla_string): def encode(self): self['value'] = struct.pack( 'B' * 256, *(int(x) for x in self.value) ) nla_string.encode(self) class stats(nla): fields = ( ('early', 'I'), ('pdrop', 'I'), ('other', 'I'), ('marked', 'I'), ('matched', 'I'), ) class stats2(c_stats2): class stats_app(stats): pass	3.35 KB	2025-05-25 12:26
📄 sched_clsact.py ''' clsact ++++++ The clsact qdisc provides a mechanism to attach integrated filter-action classifiers to an interface, either at ingress or egress, or both. The use case shown here is using a bpf program (implemented elsewhere) to direct the packet processing. The example also uses the direct-action feature to specify what to do with each packet (pass, drop, redirect, etc.). BPF ingress/egress example using clsact qdisc:: # open_bpf_fd is outside the scope of pyroute2 #fd = open_bpf_fd() eth0 = ip.get_links(ifname="eth0")[0] ip.tc("add", "clsact", eth0) # add ingress clsact ip.tc("add-filter", "bpf", idx, ":1", fd=fd, name="myprog", parent="ffff:fff2", classid=1, direct_action=True) # add egress clsact ip.tc("add-filter", "bpf", idx, ":1", fd=fd, name="myprog", parent="ffff:fff3", classid=1, direct_action=True) ''' from pyroute2.netlink.rtnl import TC_H_CLSACT parent = TC_H_CLSACT def fix_request(request): if 'rate' in request: del request['rate'] request['handle'] = 0xFFFF0000	1.04 KB	2025-05-25 12:26
📄 sched_codel.py import logging from pyroute2.netlink import nla from pyroute2.netlink.rtnl import TC_H_ROOT from pyroute2.netlink.rtnl.tcmsg.common import get_time from pyroute2.netlink.rtnl.tcmsg.common import stats2 as c_stats2 log = logging.getLogger(__name__) parent = TC_H_ROOT def get_parameters(kwarg): # # ACHTUNG: experimental code # # Parameters naming scheme WILL be changed in next releases # ret = {'attrs': []} transform = { 'cdl_limit': lambda x: x, 'cdl_ecn': lambda x: x, 'cdl_target': get_time, 'cdl_ce_threshold': get_time, 'cdl_interval': get_time, } for key in transform: if key in kwarg: log.warning( 'codel parameters naming will be changed ' 'in next releases (%s)' % key ) ret['attrs'].append( ['TCA_CODEL_%s' % key[4:].upper(), transform[key](kwarg[key])] ) return ret class options(nla): nla_map = ( ('TCA_CODEL_UNSPEC', 'none'), ('TCA_CODEL_TARGET', 'uint32'), ('TCA_CODEL_LIMIT', 'uint32'), ('TCA_CODEL_INTERVAL', 'uint32'), ('TCA_CODEL_ECN', 'uint32'), ('TCA_CODEL_CE_THRESHOLD', 'uint32'), ) class stats(nla): fields = ( ('maxpacket', 'I'), ('count', 'I'), ('lastcount', 'I'), ('ldelay', 'I'), ('drop_next', 'I'), ('drop_overlimit', 'I'), ('ecn_mark', 'I'), ('dropping', 'I'), ('ce_mark', 'I'), ) class stats2(c_stats2): class stats_app(stats): pass	1.57 KB	2025-05-25 12:26
📄 sched_drr.py ''' drr +++ The qdisc doesn't accept any parameters, but the class accepts `quantum` parameter:: ip.tc('add', 'drr', interface, '1:') ip.tc('add-class', 'drr', interface, '1:10', quantum=1600) ip.tc('add-class', 'drr', interface, '1:20', quantum=1600) ''' from pyroute2.netlink import nla from pyroute2.netlink.rtnl import TC_H_ROOT from pyroute2.netlink.rtnl.tcmsg.common import stats2 as c_stats2 parent = TC_H_ROOT def get_class_parameters(kwarg): return {'attrs': [['TCA_DRR_QUANTUM', kwarg.get('quantum', 0)]]} class options(nla): nla_map = (('TCA_DRR_UNSPEC', 'none'), ('TCA_DRR_QUANTUM', 'uint32')) class stats(nla): fields = (('deficit', 'I'),) class stats2(c_stats2): class stats_app(stats): pass	755 B	2025-05-25 12:26
📄 sched_fq_codel.py import logging from pyroute2.netlink import nla from pyroute2.netlink.rtnl import RTM_DELQDISC, RTM_NEWQDISC, TC_H_ROOT from pyroute2.netlink.rtnl.tcmsg.common import get_time, stats2 log = logging.getLogger(__name__) parent = TC_H_ROOT def get_parameters(kwarg): # # ACHTUNG: experimental code # # Parameters naming scheme WILL be changed in next releases # ret = {'attrs': []} transform = { 'fqc_limit': lambda x: x, 'fqc_flows': lambda x: x, 'fqc_quantum': lambda x: x, 'fqc_ecn': lambda x: x, 'fqc_target': get_time, 'fqc_ce_threshold': get_time, 'fqc_interval': get_time, } for key in transform: if key in kwarg: log.warning( 'fq_codel parameters naming will be changed ' 'in next releases (%s)' % key ) ret['attrs'].append( [ 'TCA_FQ_CODEL_%s' % key[4:].upper(), transform[key](kwarg[key]), ] ) return ret class options(nla): nla_map = ( ('TCA_FQ_CODEL_UNSPEC', 'none'), ('TCA_FQ_CODEL_TARGET', 'uint32'), ('TCA_FQ_CODEL_LIMIT', 'uint32'), ('TCA_FQ_CODEL_INTERVAL', 'uint32'), ('TCA_FQ_CODEL_ECN', 'uint32'), ('TCA_FQ_CODEL_FLOWS', 'uint32'), ('TCA_FQ_CODEL_QUANTUM', 'uint32'), ('TCA_FQ_CODEL_CE_THRESHOLD', 'uint32'), ('TCA_FQ_CODEL_DROP_BATCH_SIZE', 'uint32'), ('TCA_FQ_CODEL_MEMORY_LIMIT', 'uint32'), ) class qdisc_stats(nla): fields = ( ('type', 'I'), ('maxpacket', 'I'), ('drop_overlimit', 'I'), ('ecn_mark', 'I'), ('new_flow_count', 'I'), ('new_flows_len', 'I'), ('old_flows_len', 'I'), ('ce_mark', 'I'), ('memory_usage', 'I'), ('drop_overmemory', 'I'), ) class class_stats(nla): fields = ( ('type', 'I'), ('deficit', 'i'), ('ldelay', 'I'), ('count', 'I'), ('lastcount', 'I'), ('dropping', 'I'), ('drop_next', 'i'), ) class qdisc_stats2(stats2): class stats_app(qdisc_stats): pass class class_stats2(stats2): class stats_app(class_stats): pass def stats2(msg, argv, kwarg): if msg['header']['type'] in (RTM_NEWQDISC, RTM_DELQDISC): return qdisc_stats2 else: return class_stats2 # To keep the compatibility with TCA_XSTATS def stats(msg, argv, **kwarg): if msg['header']['type'] in (RTM_NEWQDISC, RTM_DELQDISC): return qdisc_stats else: return class_stats	2.6 KB	2025-05-25 12:26
📄 sched_hfsc.py ''' hfsc ++++ Simple HFSC example:: eth0 = ip.get_links(ifname="eth0")[0] ip.tc("add", "hfsc", eth0, handle="1:", default="1:1") ip.tc("add-class", "hfsc", eth0, handle="1:1", parent="1:0" rsc={"m2": "5mbit"}) HFSC curve nla types: * `rsc`: real-time curve * `fsc`: link-share curve * `usc`: upper-limit curve ''' from pyroute2.netlink import nla from pyroute2.netlink.rtnl import RTM_DELQDISC, RTM_NEWQDISC, TC_H_ROOT from pyroute2.netlink.rtnl.tcmsg.common import get_rate, get_time from pyroute2.netlink.rtnl.tcmsg.common import stats2 as c_stats2 parent = TC_H_ROOT def get_parameters(kwarg): defcls = kwarg.get('default', kwarg.get('defcls', 0x10)) defcls &= 0xFFFF return {'defcls': defcls} def get_class_parameters(kwarg): ret = {'attrs': []} for key in ('rsc', 'fsc', 'usc'): if key in kwarg: ret['attrs'].append( [ 'TCA_HFSC_%s' % key.upper(), { 'm1': get_rate(kwarg[key].get('m1', 0)), 'd': get_time(kwarg[key].get('d', 0)), 'm2': get_rate(kwarg[key].get('m2', 0)), }, ] ) return ret class options_hfsc(nla): fields = (('defcls', 'H'),) # default class class options_hfsc_class(nla): nla_map = ( ('TCA_HFSC_UNSPEC', 'none'), ('TCA_HFSC_RSC', 'hfsc_curve'), # real-time curve ('TCA_HFSC_FSC', 'hfsc_curve'), # link-share curve ('TCA_HFSC_USC', 'hfsc_curve'), ) # upper-limit curve class hfsc_curve(nla): fields = ( ('m1', 'I'), # slope of the first segment in bps ('d', 'I'), # x-projection of the first segment in us ('m2', 'I'), ) # slope of the second segment in bps def options(msg, argv, *kwarg): if msg['header']['type'] in (RTM_NEWQDISC, RTM_DELQDISC): return options_hfsc else: return options_hfsc_class class stats2(c_stats2): class stats_app(nla): fields = ( ('work', 'Q'), # total work done ('rtwork', 'Q'), # total work done by real-time criteria ('period', 'I'), # current period ('level', 'I'), ) # class level in hierarchy	2.3 KB	2025-05-25 12:26
📄 sched_htb.py ''' htb +++ TODO: list parameters An example with htb qdisc, lets assume eth0 == 2:: # u32 --> +--> htb 1:10 --> sfq 10:0 # \| \| # \| \| # eth0 -- htb 1:0 -- htb 1:1 # \| \| # \| \| # u32 --> +--> htb 1:20 --> sfq 20:0 eth0 = 2 # add root queue 1:0 ip.tc("add", "htb", eth0, 0x10000, default=0x200000) # root class 1:1 ip.tc("add-class", "htb", eth0, 0x10001, parent=0x10000, rate="256kbit", burst=1024 * 6) # two branches: 1:10 and 1:20 ip.tc("add-class", "htb", eth0, 0x10010, parent=0x10001, rate="192kbit", burst=1024 * 6, prio=1) ip.tc("add-class", "htb", eht0, 0x10020, parent=0x10001, rate="128kbit", burst=1024 * 6, prio=2) # two leaves: 10:0 and 20:0 ip.tc("add", "sfq", eth0, 0x100000, parent=0x10010, perturb=10) ip.tc("add", "sfq", eth0, 0x200000, parent=0x10020, perturb=10) # two filters: one to load packets into 1:10 and the # second to 1:20 ip.tc("add-filter", "u32", eth0, parent=0x10000, prio=10, protocol=socket.AF_INET, target=0x10010, keys=["0x0006/0x00ff+8", "0x0000/0xffc0+2"]) ip.tc("add-filter", "u32", eth0, parent=0x10000, prio=10, protocol=socket.AF_INET, target=0x10020, keys=["0x5/0xf+0", "0x10/0xff+33"]) ''' from pyroute2.netlink import nla from pyroute2.netlink.rtnl import RTM_DELQDISC, RTM_NEWQDISC, TC_H_ROOT from pyroute2.netlink.rtnl.tcmsg.common import ( calc_xmittime, get_hz, get_rate, nla_plus_rtab, stats2, ) parent = TC_H_ROOT def get_class_parameters(kwarg): prio = kwarg.get('prio', 0) mtu = kwarg.get('mtu', 1600) mpu = kwarg.get('mpu', 0) overhead = kwarg.get('overhead', 0) quantum = kwarg.get('quantum', 0) rate = get_rate(kwarg.get('rate', None)) ceil = get_rate(kwarg.get('ceil', 0)) or rate rate64 = 0 ceil64 = 0 burst = ( kwarg.get('burst', None) or kwarg.get('maxburst', None) or kwarg.get('buffer', None) ) if rate is not None: if burst is None: burst = rate / get_hz() + mtu burst = calc_xmittime(rate, burst) cburst = ( kwarg.get('cburst', None) or kwarg.get('cmaxburst', None) or kwarg.get('cbuffer', None) ) if ceil is not None: if cburst is None: cburst = ceil / get_hz() + mtu cburst = calc_xmittime(ceil, cburst) if rate is not None and rate >= 1 << 32: rate64 = rate rate = (1 << 32) - 1 if ceil is not None and ceil >= 1 << 32: ceil64 = ceil ceil = (1 << 32) - 1 ret = { 'attrs': [ [ 'TCA_HTB_PARMS', { 'buffer': burst, 'cbuffer': cburst, 'quantum': quantum, 'prio': prio, 'rate': rate, 'ceil': ceil, 'ceil_overhead': overhead, 'rate_overhead': overhead, 'rate_mpu': mpu, 'ceil_mpu': mpu, }, ], ['TCA_HTB_RTAB', True], ['TCA_HTB_CTAB', True], ] } if rate64 > 0: ret['attrs'].append(['TCA_HTB_RATE64', rate64]) if ceil64 > 0: ret['attrs'].append(['TCA_HTB_CEIL64', ceil64]) return ret def get_parameters(kwarg): rate2quantum = kwarg.get('r2q', 0xA) version = kwarg.get('version', 3) defcls = kwarg.get('default', 0x10) return { 'attrs': [ [ 'TCA_HTB_INIT', { 'debug': 0, 'defcls': defcls, 'direct_pkts': 0, 'rate2quantum': rate2quantum, 'version': version, }, ] ] } def fix_request(request): if not request: opts = get_parameters({}) request['attrs'].append(['TCA_OPTIONS', opts]) if 'options' not in request: request['options'] = get_parameters({}) if 'rate' in request: del request['rate'] # The tokens and ctokens are badly defined in the kernel structure # as unsigned int instead of signed int. (cf net/sched/sch_htb.c # in linux source) class stats(nla): fields = ( ('lends', 'I'), ('borrows', 'I'), ('giants', 'I'), ('tokens', 'i'), ('ctokens', 'i'), ) class qdisc_stats2(stats2): nla_map = ( ('TCA_STATS_UNSPEC', 'none'), ('TCA_STATS_BASIC', 'basic'), ('TCA_STATS_RATE_EST', 'rate_est'), ('TCA_STATS_QUEUE', 'queue'), ) class class_stats2(stats2): class stats_app(stats): pass def stats2(msg, argv, *kwarg): if msg['header']['type'] in (RTM_NEWQDISC, RTM_DELQDISC): return qdisc_stats2 else: return class_stats2 class options(nla_plus_rtab): nla_map = ( ('TCA_HTB_UNSPEC', 'none'), ('TCA_HTB_PARMS', 'htb_parms'), ('TCA_HTB_INIT', 'htb_glob'), ('TCA_HTB_CTAB', 'ctab'), ('TCA_HTB_RTAB', 'rtab'), ('TCA_HTB_DIRECT_QLEN', 'uint32'), ('TCA_HTB_RATE64', 'uint64'), ('TCA_HTB_CEIL64', 'uint64'), ('TCA_HTB_PAD', 'hex'), ('TCA_HTB_OFFLOAD', 'hex'), ) class htb_glob(nla): fields = ( ('version', 'I'), ('rate2quantum', 'I'), ('defcls', 'I'), ('debug', 'I'), ('direct_pkts', 'I'), ) class htb_parms(nla_plus_rtab.parms): fields = ( ('rate_cell_log', 'B'), ('rate___reserved', 'B'), ('rate_overhead', 'H'), ('rate_cell_align', 'h'), ('rate_mpu', 'H'), ('rate', 'I'), ('ceil_cell_log', 'B'), ('ceil___reserved', 'B'), ('ceil_overhead', 'H'), ('ceil_cell_align', 'h'), ('ceil_mpu', 'H'), ('ceil', 'I'), ('buffer', 'I'), ('cbuffer', 'I'), ('quantum', 'I'), ('level', 'I'), ('prio', 'I'), )	6.3 KB	2025-05-25 12:26
📄 sched_ingress.py from pyroute2.netlink import nla from pyroute2.netlink.rtnl import TC_H_INGRESS parent = TC_H_INGRESS def fix_request(request): if 'rate' in request: del request['rate'] request['handle'] = 0xFFFF0000 class options(nla): fields = (('value', 'I'),)	273 B	2025-05-25 12:26
📄 sched_netem.py from pyroute2.netlink import nla from pyroute2.netlink.rtnl import TC_H_ROOT from pyroute2.netlink.rtnl.tcmsg.common import get_rate, percent2u32, time2tick parent = TC_H_ROOT def get_parameters(kwarg): delay = time2tick(kwarg.get('delay', 0)) # in microsecond limit = kwarg.get('limit', 1000) # fifo limit (packets) see netem.c:230 loss = percent2u32(kwarg.get('loss', 0)) # int percentage gap = kwarg.get('gap', 0) duplicate = percent2u32(kwarg.get('duplicate', 0)) # int percentage jitter = time2tick(kwarg.get('jitter', 0)) # in microsecond opts = { 'delay': delay, 'limit': limit, 'loss': loss, 'gap': gap, 'duplicate': duplicate, 'jitter': jitter, 'attrs': [], } # correlation (delay, loss, duplicate) delay_corr = percent2u32(kwarg.get('delay_corr', 0)) loss_corr = percent2u32(kwarg.get('loss_corr', 0)) dup_corr = percent2u32(kwarg.get('dup_corr', 0)) if delay_corr or loss_corr or dup_corr: # delay_corr requires that both jitter and delay are != 0 if delay_corr and not (delay and jitter): raise Exception( 'delay correlation requires delay' ' and jitter to be set' ) # loss correlation and loss if loss_corr and not loss: raise Exception('loss correlation requires loss to be set') # duplicate correlation and duplicate if dup_corr and not duplicate: raise Exception( 'duplicate correlation requires ' 'duplicate to be set' ) opts['attrs'].append( [ 'TCA_NETEM_CORR', { 'delay_corr': delay_corr, 'loss_corr': loss_corr, 'dup_corr': dup_corr, }, ] ) # reorder (probability, correlation) prob_reorder = percent2u32(kwarg.get('prob_reorder', 0)) corr_reorder = percent2u32(kwarg.get('corr_reorder', 0)) if prob_reorder != 0: # gap defaults to 1 if equal to 0 if gap == 0: opts['gap'] = gap = 1 opts['attrs'].append( [ 'TCA_NETEM_REORDER', {'prob_reorder': prob_reorder, 'corr_reorder': corr_reorder}, ] ) else: if gap != 0: raise Exception('gap can only be set when prob_reorder is set') elif corr_reorder != 0: raise Exception( 'corr_reorder can only be set when ' 'prob_reorder is set' ) # corrupt (probability, correlation) prob_corrupt = percent2u32(kwarg.get('prob_corrupt', 0)) corr_corrupt = percent2u32(kwarg.get('corr_corrupt', 0)) if prob_corrupt: opts['attrs'].append( [ 'TCA_NETEM_CORRUPT', {'prob_corrupt': prob_corrupt, 'corr_corrupt': corr_corrupt}, ] ) elif corr_corrupt != 0: raise Exception( 'corr_corrupt can only be set when ' 'prob_corrupt is set' ) # rate (rate, packet_overhead, cell_size, cell_overhead) rate = get_rate(kwarg.get('rate', None)) packet_overhead = kwarg.get('packet_overhead', 0) cell_size = kwarg.get('cell_size', 0) cell_overhead = kwarg.get('cell_overhead', 0) if rate is not None: opts['attrs'].append( [ 'TCA_NETEM_RATE', { 'rate': rate, 'packet_overhead': packet_overhead, 'cell_size': cell_size, 'cell_overhead': cell_overhead, }, ] ) elif packet_overhead != 0 or cell_size != 0 or cell_overhead != 0: raise Exception( 'packet_overhead, cell_size and cell_overhead' 'can only be set when rate is set' ) # TODO # delay distribution (dist_size, dist_data) return opts class options(nla): nla_map = ( ('TCA_NETEM_UNSPEC', 'none'), ('TCA_NETEM_CORR', 'netem_corr'), ('TCA_NETEM_DELAY_DIST', 'none'), ('TCA_NETEM_REORDER', 'netem_reorder'), ('TCA_NETEM_CORRUPT', 'netem_corrupt'), ('TCA_NETEM_LOSS', 'none'), ('TCA_NETEM_RATE', 'netem_rate'), ) fields = ( ('delay', 'I'), ('limit', 'I'), ('loss', 'I'), ('gap', 'I'), ('duplicate', 'I'), ('jitter', 'I'), ) class netem_corr(nla): '''correlation''' fields = (('delay_corr', 'I'), ('loss_corr', 'I'), ('dup_corr', 'I')) class netem_reorder(nla): '''reorder has probability and correlation''' fields = (('prob_reorder', 'I'), ('corr_reorder', 'I')) class netem_corrupt(nla): '''corruption has probability and correlation''' fields = (('prob_corrupt', 'I'), ('corr_corrupt', 'I')) class netem_rate(nla): '''rate''' fields = ( ('rate', 'I'), ('packet_overhead', 'i'), ('cell_size', 'I'), ('cell_overhead', 'i'), )	5.02 KB	2025-05-25 12:26
📄 sched_pfifo.py from pyroute2.netlink import nla from pyroute2.netlink.rtnl import TC_H_ROOT parent = TC_H_ROOT class options(nla): fields = (('limit', 'i'),) def get_parameters(kwarg): return kwarg	196 B	2025-05-25 12:26
📄 sched_pfifo_fast.py from pyroute2.netlink import nla from pyroute2.netlink.rtnl import TC_H_ROOT parent = TC_H_ROOT class options(nla): fields = (('bands', 'i'), ('priomap', '16B')) def get_parameters(kwarg): return kwarg	215 B	2025-05-25 12:26
📄 sched_plug.py from pyroute2.netlink import nla from pyroute2.netlink.rtnl import TC_H_ROOT parent = TC_H_ROOT actions = { 'TCQ_PLUG_BUFFER': 0, 'TCQ_PLUG_RELEASE_ONE': 1, 'TCQ_PLUG_RELEASE_INDEFINITE': 2, 'TCQ_PLUG_LIMIT': 3, } def get_parameters(kwarg): return { 'action': actions.get(kwarg.get('action', 0), 0), 'limit': kwarg.get('limit', 0), } class options(nla): fields = (('action', 'i'), ('limit', 'I'))	446 B	2025-05-25 12:26
📄 sched_sfq.py from pyroute2.netlink import nla from pyroute2.netlink.rtnl import TC_H_ROOT from pyroute2.netlink.rtnl.tcmsg.common import ( get_size, red_eval_ewma, red_eval_P, ) parent = TC_H_ROOT TC_RED_ECN = 1 TC_RED_HARDDROP = 2 TC_RED_ADAPTATIVE = 4 def get_parameters(kwarg): kwarg['quantum'] = get_size(kwarg.get('quantum', 0)) kwarg['perturb_period'] = kwarg.get('perturb', 0) or kwarg.get( 'perturb_period', 0 ) limit = kwarg['limit'] = kwarg.get('limit', 0) or kwarg.get( 'redflowlimit', 0 ) qth_min = kwarg.get('min', 0) qth_max = kwarg.get('max', 0) avpkt = kwarg.get('avpkt', 1000) probability = kwarg.get('probability', 0.02) ecn = kwarg.get('ecn', False) harddrop = kwarg.get('harddrop', False) kwarg['flags'] = kwarg.get('flags', 0) if ecn: kwarg['flags'] \|= TC_RED_ECN if harddrop: kwarg['flags'] \|= TC_RED_HARDDROP if kwarg.get('redflowlimit'): qth_max = qth_max or limit / 4 qth_min = qth_min or qth_max / 3 kwarg['burst'] = kwarg['burst'] or (2 * qth_min + qth_max) / ( 3 * avpkt ) if limit <= qth_max: raise ValueError('limit must be > qth_max') if qth_max <= qth_min: raise ValueError('qth_max must be > qth_min') kwarg['qth_min'] = qth_min kwarg['qth_max'] = qth_max kwarg['Wlog'] = red_eval_ewma(qth_min, kwarg['burst'], avpkt) kwarg['Plog'] = red_eval_P(qth_min, qth_max, probability) if kwarg['Wlog'] < 0: raise ValueError('Wlog must be > 0') if kwarg['Plog'] < 0: raise ValueError('Plog must be > 0') kwarg['max_P'] = int(probability * pow(2, 23)) return kwarg class options_sfq_v0(nla): fields = ( ('quantum', 'I'), ('perturb_period', 'i'), ('limit', 'I'), ('divisor', 'I'), ('flows', 'I'), ) class options_sfq_v1(nla): fields = ( ('quantum', 'I'), ('perturb_period', 'i'), ('limit_v0', 'I'), ('divisor', 'I'), ('flows', 'I'), ('depth', 'I'), ('headdrop', 'I'), ('limit_v1', 'I'), ('qth_min', 'I'), ('qth_max', 'I'), ('Wlog', 'B'), ('Plog', 'B'), ('Scell_log', 'B'), ('flags', 'B'), ('max_P', 'I'), ('prob_drop', 'I'), ('forced_drop', 'I'), ('prob_mark', 'I'), ('forced_mark', 'I'), ('prob_mark_head', 'I'), ('forced_mark_head', 'I'), ) def options(argv, *kwarg): if kwarg.get('length', 0) >= options_sfq_v1.get_size(): return options_sfq_v1 else: return options_sfq_v0	2.65 KB	2025-05-25 12:26
📄 sched_tbf.py from pyroute2.netlink.rtnl import TC_H_ROOT from pyroute2.netlink.rtnl.tcmsg.common import ( get_rate_parameters, nla_plus_rtab, ) parent = TC_H_ROOT def fix_request(request): if 'rate' in request: del request['rate'] def get_parameters(kwarg): parms = get_rate_parameters(kwarg) # fill parameters return {'attrs': [['TCA_TBF_PARMS', parms], ['TCA_TBF_RTAB', True]]} class options(nla_plus_rtab): nla_map = ( ('TCA_TBF_UNSPEC', 'none'), ('TCA_TBF_PARMS', 'tbf_parms'), ('TCA_TBF_RTAB', 'rtab'), ('TCA_TBF_PTAB', 'ptab'), ) class tbf_parms(nla_plus_rtab.parms): fields = ( ('rate_cell_log', 'B'), ('rate___reserved', 'B'), ('rate_overhead', 'H'), ('rate_cell_align', 'h'), ('rate_mpu', 'H'), ('rate', 'I'), ('peak_cell_log', 'B'), ('peak___reserved', 'B'), ('peak_overhead', 'H'), ('peak_cell_align', 'h'), ('peak_mpu', 'H'), ('peak', 'I'), ('limit', 'I'), ('buffer', 'I'), ('mtu', 'I'), )	1.14 KB	2025-05-25 12:26
📄 sched_template.py ''' Template sched file. All the tcmsg plugins should be registered in `__init__.py`, see the `plugins` dict. All the methods, variables and classes are optional, but the naming scheme is fixed. ''' from pyroute2.netlink import nla from pyroute2.netlink.rtnl import TC_H_ROOT from pyroute2.netlink.rtnl.tcmsg import common # if you define the `parent` variable, it will be used # as the default parent value if no other value is # provided in the call options parent = TC_H_ROOT def fix_request(request): ''' This method it called for all types -- classes, qdiscs and filters. Can be used to fix some request fields. ''' pass def get_parameters(kwarg): ''' Called for qdiscs and filters. Should return the structure to be embedded as the qdisc parameters (`TCA_OPTIONS`). ''' return None def get_class_parameters(kwarg): ''' The same as above, but called only for classes. ''' return None class options(nla.hex): ''' The `TCA_OPTIONS` struct, by default not decoded. ''' pass class stats(nla.hex): ''' The struct to decode `TCA_XSTATS`. ''' pass class stats2(common.stats2): ''' The struct to decode `TCA_STATS2`. ''' pass	1.22 KB	2025-05-25 12:26

Edit File:

Terminal