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 ''' General structure ================= The NDB objects are dictionary-like structures that represent network objects -- interfaces, routes, addresses etc. They support the common dict API, like item getting, setting, iteration through key and values. In addition to that, NDB object add specific calls, see the API section below. Most of the NDB object types store all attributes in a flat one level dictionary. Some, like multihop routes, implement nested structures. In addition to that, some objects like `Interface` provide views on the DB that list only related objects -- addresses, routes and neighbours. More on these topic see in the corresponding sections. NDB objects and RTNL API ======================== The dictionary fields represent RTNL messages fields and NLA names, and the objects are used as argument dictionaries to normal `IPRoute` methods like `link()` or `route()`. Thus everything described for the `IPRoute` methods is valid here as well. See also: :ref:`iproute` .. testsetup:: from pyroute2 import IPRoute, NDB from pyroute2 import config config.mock_netlink = True .. testcode:: # create a vlan interface with IPRoute eth0 = 2 with IPRoute() as ipr: ipr.link("add", ifname="vlan1108", kind="vlan", link=eth0, vlan_id=1108) # same with NDB: with NDB(log="stderr") as ndb: vlan = ndb.interfaces.create( ifname="vlan1108", kind="vlan", link="eth0", vlan_id=1108, ) vlan.commit() Slightly simplifying, if a network object doesn't exist, NDB will run an RTNL method with "add" argument, if exists -- "set", and to remove an object NDB will call the method with "del" argument. API === ''' import collections import errno import json import threading import time import traceback import weakref from functools import partial from pyroute2 import cli from pyroute2.netlink.exceptions import NetlinkError from pyroute2.requests.main import RequestProcessor from ..auth_manager import AuthManager, check_auth from ..events import InvalidateHandlerException, State from ..messages import cmsg_event from ..report import Record RSLV_IGNORE = 0 RSLV_RAISE = 1 RSLV_NONE = 2 RSLV_DELETE = 3 def fallback_add(self, idx_req, req): # ignore all set/get for objects with incomplete idx_req if set(idx_req.keys()) != set(self.kspec): self.log.debug('ignore incomplete idx_req in the fallback') return # try to set the object ( self.ndb._event_queue.put( self.sources[self['target']].api(self.api, 'set', req), source=self['target'], ) ) # try to get the object ( self.ndb._event_queue.put( self.sources[self['target']].api(self.api, 'get', idx_req), source=self['target'], ) ) # reload the collected data self.load_sql() class RTNL_Object(dict): ''' The common base class for NDB objects -- interfaces, routes, rules addresses etc. Implements common logic for all the classes, like item setting, commit/rollback, RTNL event filters, loading values from the DB backend etc. ''' view = None # (optional) view to load values for the summary etc. utable = None # table to send updates to resolve_fields = [] key_extra_fields = [] hidden_fields = [] fields_cmp = {} fields_load_transform = {} field_filter = object rollback_chain = [] fallback_for = None schema = None event_map = None state = None log = None errors = None msg_class = None reverse_update = None _table = None _apply_script = None _apply_script_snapshots = [] _key = None _replace = None _replace_on_key_change = False _init_complete = False # 8<------------------------------------------------------------ # # Documented public properties section # @property def table(self): ''' Main reference table for the object. The SQL schema of this table is used to build the object key and to verify fields. Read-write property. ''' return self._table @table.setter def table(self, value): self._table = value @property def etable(self): ''' Effective table where the object actually fetches the data from. It is not always equal `self.table`, e.g. snapshot objects fetch the data from snapshot tables. Read-only property. ''' if self.ctxid: return '%s_%s' % (self.table, self.ctxid) else: return self.table @property def key(self): ''' Key of the object, used to build SQL requests to fetch the data from the DB. Read-write property. ''' nkey = self._key or {} ret = collections.OrderedDict() for name in self.kspec: kname = self.iclass.nla2name(name) if kname in self: value = self[kname] if value is None and name in nkey: value = nkey[name] if isinstance(value, (list, tuple, dict)): value = json.dumps(value) ret[name] = value if len(ret) < len(self.kspec): for name in self.key_extra_fields: kname = self.iclass.nla2name(name) if self.get(kname): ret[name] = self[kname] return ret @key.setter def key(self, k): if not isinstance(k, dict): return for key, value in k.items(): if value is not None: dict.__setitem__(self, self.iclass.nla2name(key), value) # # 8<------------------------------------------------------------ # @classmethod def _count(cls, view): return view.ndb.task_manager.db_fetchone( 'SELECT count() FROM %s' % view.table ) @classmethod def _dump_where(cls, view): return '', [] @classmethod def _sdump(cls, view, names, fnames): req = ''' SELECT %s FROM %s AS main ''' % ( fnames, cls.table, ) yield names where, values = cls._dump_where(view) for record in view.ndb.task_manager.db_fetch(req + where, values): yield record @classmethod def summary(cls, view): return cls._sdump( view, view.ndb.schema.compiled[cls.table]['norm_idx'], view.ndb.schema.compiled[cls.table]['knames'], ) @classmethod def dump(cls, view): return cls._sdump( view, view.ndb.schema.compiled[cls.table]['norm_names'], view.ndb.schema.compiled[cls.table]['fnames'], ) @classmethod def spec_normalize(cls, spec): return spec @staticmethod def key_load_context(key, context): return key def __init__( self, view, key, iclass, ctxid=None, load=True, master=None, check=True, auth_managers=None, ): self.view = view self.ndb = view.ndb self.sources = view.ndb.sources self.master = master self.ctxid = ctxid self.schema = view.ndb.schema self.task_manager = view.ndb.task_manager self.changed = set() self.iclass = iclass self.utable = self.utable or self.table self.errors = [] self.atime = time.time() self.log = self.ndb.log.channel('rtnl_object') self.log.debug('init') if auth_managers is None: auth_managers = [AuthManager(None, self.ndb.log.channel('auth'))] self.auth_managers = auth_managers self.state = State() self.state.set('invalid') self.snapshot_deps = [] self.load_event = threading.Event() self.load_event.set() self.load_debug = False self.lock = threading.Lock() self.object_data = RequestProcessor(context=weakref.proxy(self)) self.object_data.add_filter(self.field_filter()) self.kspec = self.schema.compiled[self.table]['idx'] self.knorm = self.schema.compiled[self.table]['norm_idx'] self.spec = self.schema.compiled[self.table]['all_names'] self.names = self.schema.compiled[self.table]['norm_names'] self.lookup_fallbacks = self.schema.compiled[self.table][ 'lookup_fallbacks' ] self.names_count = [self.names.count(x) for x in self.names] self.last_save = None if self.event_map is None: self.event_map = {} self._apply_script = [] self.fallback_for = { 'add': {errno.EEXIST: fallback_add, errno.EAGAIN: None}, 'set': {errno.ENODEV: None}, 'del': { errno.ENODEV: None, # interfaces errno.ENOENT: None, # rules errno.ESRCH: None, # routes errno.EADDRNOTAVAIL: None, # addresses }, } if isinstance(key, dict): self.chain = key.pop('ndb_chain', None) create = key.pop('create', False) else: self.chain = None create = False exists = self.exists(key) ckey = self.complete_key(key) if create: if check & exists: raise KeyError('object exists') for name in key: self[self.iclass.nla2name(name)] = key[name] # FIXME -- merge with complete_key() if 'target' not in self: self.load_value('target', self.view.default_target) else: if not exists: raise KeyError('object does not exists') self.key = ckey if load: if ctxid is None: self.load_sql() else: self.load_sql(table=self.table) self._init_complete = True @classmethod def new_spec(cls, spec, context=None, localhost=None): if isinstance(spec, Record): spec = spec._as_dict() spec = cls.spec_normalize(spec) rp = RequestProcessor(context=spec) rp.add_filter(cls.field_filter()) rp.update(spec) if isinstance(context, dict): rp.update(context) if 'target' not in rp and localhost is not None: rp['target'] = localhost return rp @staticmethod def resolve(view, spec, fields, policy=RSLV_IGNORE): ''' Resolve specific fields e.g. convert port ifname into index. ''' for field in fields: reference = spec.get(field) try: if isinstance(reference, dict) and 'index' in reference: spec[field] = reference['index'] elif reference is not None and not isinstance(reference, int): spec[field] = view[reference]['index'] except (KeyError, TypeError): if policy == RSLV_RAISE: raise elif policy == RSLV_NONE: spec[field] = None elif policy == RSLV_DELETE: del spec[field] elif policy == RSLV_IGNORE: pass else: raise TypeError('unknown rslv policy') def mark_tflags(self, mark): pass def keys(self): return filter(lambda x: x not in self.hidden_fields, dict.keys(self)) def items(self): return filter( lambda x: x[0] not in self.hidden_fields, dict.items(self) ) @property def context(self): return {'target': self.get('target', self.ndb.localhost)} @classmethod def nla2name(self, name): return self.msg_class.nla2name(name) @classmethod def compare_record(self, left, right): pass @classmethod def name2nla(self, name): return self.msg_class.name2nla(name) def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): self.commit() def __hash__(self): return id(self) @check_auth('obj:read') def __getitem__(self, key): return dict.__getitem__(self, key) @check_auth('obj:modify') def __setitem__(self, key, value): for nkey, nvalue in self.object_data.filter(key, value).items(): if self.get(nkey) == nvalue: continue if self.state == 'system' and nkey in self.knorm: if self._replace_on_key_change: self.log.debug( f'prepare replace {nkey} = {nvalue} in {self.key}' ) self._replace = type(self)( self.view, self.key, auth_managers=self.auth_managers ) self.state.set('replace') else: raise ValueError( 'attempt to change a key field (%s)' % nkey ) if nkey in ('net_ns_fd', 'net_ns_pid'): self.state.set('setns') if nvalue != self.get(nkey, None): if nkey != 'target': self.changed.add(nkey) dict.__setitem__(self, nkey, nvalue) def fields(self, argv): # TODO: deprecate and move to show() Fields = collections.namedtuple('Fields', argv) return Fields([self[key] for key in argv]) def key_repr(self): return repr(self.key) @cli.change_pointer def create(self, spec): ''' Create an RTNL object of the same type, and add it to the commit chain. The spec format depends on the object. The method allows to chain creation of multiple objects sharing the same context. .. code-block:: python ( ndb.interfaces['eth0'] # 1. .set(state="up") # 2. .ipaddr # 3. .create(address='10.0.0.1', prefixlen=24) # 4. <- create() .create(address='10.0.0.2', prefixlen=24) # 5. <- create() .commit() # 6. ) Here: 1. returns an interface object `eth0` 2. sets `state="up"` and returns the object itself 3. returns an address view, that uses `eth0` as the context 4. creates an IP address, the interface lookup is done via context 5. creates another IP address -- same type, same context 6. commits the changes in the order: (interface `state="up"`; address `10.0.0.1/24`; address `10.0.0.2/24`) ''' spec['create'] = True spec['ndb_chain'] = self return self.view[spec] @cli.show_result @check_auth('obj:read') def show(self, fmt=None): ''' Return the object in a specified format. The format may be specified with the keyword argument `format` or in the `ndb.config['show_format']`. TODO: document different formats ''' fmt = fmt or self.view.ndb.config.get('show_format', 'native') if fmt == 'native': return dict(self) else: out = collections.OrderedDict() for key in sorted(self): out[key] = self[key] return '%s\n' % json.dumps(out, indent=4, separators=(',', ': ')) def set(self, argv, *kwarg): ''' Call formats: `set(key, value)` * `set(key=value)` * `set(key1=value1, key2=value2)` .. code-block:: python with ndb.interfaces["eth0"] as eth0: eth0.set( mtu=1200, state='up', address='00:11:22:33:44:55', ) ''' if argv: self[argv[0]] = argv[1] elif kwarg: for key, value in kwarg.items(): self[key] = value return self def wtime(self, itn=1): return max(min(itn * 0.1, 1), self.view.ndb._event_queue.qsize() / 10) def register(self): # # Construct a weakref handler for events. # # If the referent doesn't exist, raise the # exception to remove the handler from the # chain. # def wr_handler(wr, fname, argv): try: return getattr(wr(), fname)(argv) except Exception: # check if the weakref became invalid if wr() is None: raise InvalidateHandlerException() raise wr = weakref.ref(self) for event, fname in self.event_map.items(): # # Do not trust the implicit scope and pass the # weakref explicitly via partial # ( self.ndb.task_manager.register_handler( event, partial(wr_handler, wr, fname) ) ) @check_auth('obj:modify') def snapshot(self, ctxid=None): ''' Create and return a snapshot of the object. The method creates corresponding SQL tables for the object itself and for detected dependencies. The snapshot tables will be removed as soon as the snapshot gets collected by the GC. ''' ctxid = ctxid or self.ctxid or id(self) if self._replace is None: key = self.key else: key = self._replace.key snp = type(self)( self.view, key, ctxid=ctxid, auth_managers=self.auth_managers ) self.ndb.task_manager.db_save_deps( ctxid, weakref.ref(snp), self.iclass ) snp.changed = set(self.changed) return snp def complete_key(self, key): ''' Try to complete the object key based on the provided fields. E.g.:: >>> ndb.interfaces['eth0'].complete_key({"ifname": "eth0"}) {'ifname': 'eth0', 'index': 2, 'target': u'localhost', 'tflags': 0} It is an internal method and is not supposed to be used externally. ''' self.log.debug('complete key %s from table %s' % (key, self.etable)) fetch = [] if isinstance(key, Record): key = key._as_dict() else: key = dict(key) self.resolve( view=self.view, spec=key, fields=self.resolve_fields, policy=RSLV_DELETE, ) for name in self.kspec: if name not in key: fetch.append('f_%s' % name) if fetch: keys = [] values = [] for name, value in key.items(): nla_name = self.iclass.name2nla(name) if nla_name in self.spec: name = nla_name if value is not None and name in self.spec: keys.append('f_%s = %s' % (name, self.schema.plch)) values.append(value) spec = self.ndb.task_manager.db_fetchone( 'SELECT %s FROM %s WHERE %s' % (' , '.join(fetch), self.etable, ' AND '.join(keys)), values, ) if spec is None: self.log.debug('got none') return None for name, value in zip(fetch, spec): key[name[2:]] = value self.log.debug('got %s' % key) return key def exists(self, key): ''' Check if the object exists in the DB ''' return self.view.exists(key) @check_auth('obj:modify') def rollback(self, snapshot=None): ''' Try to rollback the object state using the snapshot provided as an argument or using `self.last_save`. ''' if self._replace is not None: self.log.debug( 'rollback replace: %s :: %s' % (self.key, self._replace.key) ) new_replace = type(self)( self.view, self.key, auth_managers=self.auth_managers ) new_replace.state.set('remove') self.state.set('replace') self.update(self._replace) self._replace = new_replace self.log.debug('rollback: %s' % str(self.state.events)) snapshot = snapshot or self.last_save if snapshot == -1: return self.remove().apply() else: snapshot.state.set(self.state.get()) snapshot.rollback_chain = self._apply_script_snapshots snapshot.apply(rollback=True) for link, snp in snapshot.snapshot_deps: link.rollback(snapshot=snp) return self def clear(self): pass @property def clean(self): return ( self.state == 'system' and not self.changed and not self._apply_script ) @check_auth('obj:modify') def commit(self): ''' Commit the pending changes. If an exception is raised during `commit()`, automatically `rollback()` to the latest saved snapshot. ''' if self.clean: return self if self.chain: self.chain.commit() self.log.debug('commit: %s' % str(self.state.events)) # Is it a new object? if self.state == 'invalid': # Save values, try to apply save = dict(self) self.last_save = -1 try: return self.apply(mode='commit') except Exception as e_i: # Save the debug info e_i.trace = traceback.format_exc() # ACHTUNG! The routine doesn't clean up the system # # Drop all the values and rollback to the initial state for key in tuple(self.keys()): del self[key] for key in save: dict.__setitem__(self, key, save[key]) raise e_i # Continue with an existing object # The snapshot tables in the DB will be dropped as soon as the GC # collects the object. But in the case of an exception the `snp` # variable will be saved in the traceback, so the tables will be # available to debug. If the traceback will be saved somewhere then # the tables will never be dropped by the GC, so you can do it # manually by `ndb.task_manager.db_purge_snapshots()` -- to invalidate # all the snapshots and to drop the associated tables. self.last_save = self.snapshot() # Apply the changes try: self.apply(mode='commit') except Exception as e_c: # Rollback in the case of any error try: self.rollback() except Exception as e_r: e_c.chain = [e_r] if hasattr(e_r, 'chain'): e_c.chain.extend(e_r.chain) e_r.chain = None raise finally: if self.last_save is not None: (self.last_save.state.set(self.state.get())) if self._replace is not None: self._replace = None return self def remove(self): ''' Set the desired state to `remove`, so the next `apply()` call will delete the object from the system. ''' with self.lock: self.state.set('remove') return self def check(self): state_map = ( ('invalid', 'system'), ('remove', 'invalid'), ('setns', 'invalid'), ('setns', 'system'), ('replace', 'system'), ) self.load_sql() self.log.debug('check: %s' % str(self.state.events)) if self.state.transition() not in state_map: self.log.debug('check state: False') return False if self.changed: self.log.debug('check changed: %s' % (self.changed)) return False self.log.debug('check: True') return True def make_req(self, prime): req = dict(prime) for key in self.changed: req[key] = self[key] return req def make_idx_req(self, prime): return prime def get_count(self): conditions = [] values = [] for name in self.kspec: conditions.append('f_%s = %s' % (name, self.schema.plch)) values.append(self.get(self.iclass.nla2name(name), None)) return ( self.ndb.task_manager.db_fetchone( ''' SELECT count() FROM %s WHERE %s ''' % (self.table, ' AND '.join(conditions)), values, ) )[0] def hook_apply(self, method, spec): pass @check_auth('obj:modify') def save_context(self): if self.state == 'invalid': self.last_save = -1 else: self.last_save = self.snapshot() return self @check_auth('obj:modify') def apply(self, rollback=False, req_filter=None, mode='apply'): ''' Apply the pending changes. If an exception is raised during `apply()`, no `rollback()` is called. No automatic snapshots are madre. In order to properly revert the changes, you have to run:: obj.save_context() try: obj.apply() except Exception: obj.rollback() ''' # Resolve the fields self.resolve( view=self.view, spec=self, fields=self.resolve_fields, policy=RSLV_RAISE, ) self.log.debug('events log: %s' % str(self.state.events)) self.log.debug('run apply') self.load_event.clear() self._apply_script_snapshots = [] # Load the current state try: self.task_manager.db_commit() except Exception: pass self.load_sql(set_state=False) if self.state == 'system' and self.get_count() == 0: state = self.state.set('invalid') else: state = self.state.get() # Create the request. prime = { x: self[x] for x in self.schema.compiled[self.table]['norm_idx'] if self.get(x) is not None } req = self.make_req(prime) idx_req = self.make_idx_req(prime) self.log.debug('apply req: %s' % str(req)) self.log.debug('apply idx_req: %s' % str(idx_req)) method = None # if state in ('invalid', 'replace'): for k, v in tuple(self.items()): if k not in req and v is not None: req[k] = v if self.master is not None: req = self.new_spec( req, self.master.context, self.ndb.localhost ) method = 'add' elif state == 'system': method = 'set' elif state == 'setns': method = 'set' elif state == 'remove': method = 'del' req = idx_req else: raise Exception('state transition not supported') self.log.debug(f'apply transition from: {state}') self.log.debug(f'apply method: {method}') if req_filter is not None: req = req_filter(req) first_call_success = False for itn in range(10): try: self.log.debug('API call %s (%s)' % (method, req)) (self.sources[self['target']].api(self.api, method, req)) first_call_success = True (self.hook_apply(method, req)) except NetlinkError as e: (self.log.debug('error: %s' % e)) if not first_call_success: self.log.debug('error on the first API call, escalate') raise ## # # FIXME: performance penalty # required now only in some NDA corner cases # must be moved to objects.neighbour # # ## if e.code in self.fallback_for[method]: self.log.debug('ignore error %s for %s' % (e.code, self)) if self.fallback_for[method][e.code] is not None: self.log.debug( 'run fallback %s (%s)' % (self.fallback_for[method][e.code], req) ) try: if isinstance( self.fallback_for[method][e.code], str ): self.sources[self['target']].api( self.api, self.fallback_for[method][e.code], req, ) else: self.fallback_for[method][e.code]( self, idx_req, req ) except NetlinkError: pass else: raise e wtime = self.wtime(itn) mqsize = self.view.ndb._event_queue.qsize() nq = self.schema.stats.get(self['target']) if nq is not None: nqsize = nq.qsize else: nqsize = 0 self.log.debug( 'stats: apply %s {' 'objid %s, wtime %s, ' 'mqsize %s, nqsize %s' '}' % (method, id(self), wtime, mqsize, nqsize) ) if self.check(): self.log.debug('checked') break self.log.debug('check failed') self.load_event.wait(wtime) self.load_event.clear() else: self.log.debug('stats: %s apply %s fail' % (id(self), method)) if not self.use_db_resync(lambda x: x, self.check): self._apply_script = [] raise Exception('could not apply the changes') self.log.debug('stats: %s pass' % (id(self))) # if state == 'replace': self._replace.remove() self._replace.apply() # if rollback: # # Iterate all the snapshot tables and collect the diff for cls in self.view.classes.values(): if issubclass(type(self), cls) or issubclass(cls, type(self)): continue table = cls.table # comprare the tables diff = self.ndb.task_manager.db_fetch( ''' SELECT FROM %s_%s EXCEPT SELECT * FROM %s ''' % (table, self.ctxid, table) ) for record in diff: record = dict( zip((self.schema.compiled[table]['all_names']), record) ) key = dict( [ x for x in record.items() if x[0] in self.schema.compiled[table]['idx'] ] ) key['create'] = True try: obj = self.view.template(key, table) except KeyError: continue obj.load_sql(ctxid=self.ctxid) obj.state.set('invalid') obj.register() try: obj.apply() except Exception as e: self.errors.append((time.time(), obj, e)) for obj in reversed(self.rollback_chain): obj.rollback() else: apply_script = self._apply_script self._apply_script = [] for op, kwarg in apply_script: kwarg['self'] = self kwarg['mode'] = mode ret = self.use_db_resync( lambda x: not isinstance(x, KeyError), op, tuple(), kwarg ) if not isinstance(ret, list): ret = [ret] for obj in ret: if isinstance(obj, Exception): raise obj elif obj is not None: self._apply_script_snapshots.append(obj) return self def use_db_resync(self, criteria, method, argv=None, kwarg=None): ret = None argv = argv or [] kwarg = kwarg or {} self.log.debug(f'criteria {criteria}') self.log.debug(f'method {method}, {argv}, {kwarg}') for attempt in range(3): ret = method(argv, kwarg) self.log.debug(f'ret {ret}') if criteria(ret): self.log.debug('criteria matched') return ret self.log.debug(f'resync the DB attempt {attempt}') self.ndb.task_manager.db_flush(self['target']) self.load_event.clear() ( self.ndb._event_queue.put( self.sources[self['target']].api('dump'), source=self['target'], ) ) ( self.ndb._event_queue.put( (cmsg_event(self['target'], self.load_event),), source=self['target'], ) ) self.load_event.wait(self.wtime(1)) self.load_event.clear() return ret def update(self, data): for key, value in data.items(): self.load_value(key, value) def update_from_sql(self, spec): ''' A bit special case: we can have several fields with non unique names. ''' for key, count, value in zip(self.names, self.names_count, spec): if count == 1 or value is not None: self.load_value(key, value) def load_direct(self, key, value): super(RTNL_Object, self).__setitem__(key, value) def load_value(self, key, value): ''' Load a value and clean up the `self.changed` set if the loaded value matches the expectation. ''' if key in self.fields_load_transform: value = self.fields_load_transform[key](value) if self.load_debug: self.log.debug('load %s: %s' % (key, value)) if key not in self.changed: dict.__setitem__(self, key, value) elif self.get(key) == value: self.changed.remove(key) elif key in self.fields_cmp and self.fields_cmp[key](self, value): self.changed.remove(key) elif self.load_debug: self.log.debug( 'discard %s: %s (expected %s)' % (key, value, self.get(key)) ) def load_sql(self, table=None, ctxid=None, set_state=True): ''' Load the data from the database. ''' if not self.key: return if table is None: if ctxid is None: table = self.etable else: table = '%s_%s' % (self.table, ctxid) keys = [] values = [] for name, value in self.key.items(): keys.append('f_%s = %s' % (name, self.schema.plch)) if isinstance(value, (list, tuple, dict)): value = json.dumps(value) values.append(value) spec = self.ndb.task_manager.db_fetchone( 'SELECT FROM %s WHERE %s' % (table, ' AND '.join(keys)), values ) self.log.debug('load_sql load: %s' % str(spec)) self.log.debug('load_sql names: %s' % str(self.names)) if set_state: with self.lock: if spec is None: if self.state != 'invalid': # No such object (anymore) self.state.set('invalid') self.changed = set() elif self.state not in ('remove', 'setns'): self.update_from_sql(spec) self.state.set('system') return spec def load_rtnlmsg(self, target, event): ''' Check if the RTNL event matches the object and load the data from the database if it does. ''' # TODO: partial match (object rename / restore) # ... # full match for norm, name in zip(self.knorm, self.kspec): value = self.get(norm) if name == 'target': if value != target: return elif name == 'tflags': continue elif value not in (event.get_attr(name), event.get(norm)): return self.log.debug('load_rtnl: %s' % str(event.get('header'))) if event['header'].get('type', 0) % 2: self.state.set('invalid') self.changed = set() else: self.load_sql() self.load_event.set()	37.08 KB	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 address.py ''' .. testsetup:: * from socket import AF_INET from pyroute2 import NDB from pyroute2 import config config.mock_netlink = True ndb = NDB() .. testcleanup:: * ndb.close() Using the global view ===================== The `addresses` view provides access to all the addresses registered in the DB, as well as methods to create and remove them: .. testcode:: eth0 = ndb.interfaces['eth0'] # create an address ndb.addresses.create( address='10.0.0.1', prefixlen=24, index=eth0['index'], ).commit() # remove it with ndb.addresses['10.0.0.1/24'] as addr: addr.remove() # list addresses for record in ndb.addresses.summary(): print(record) .. testoutput:: :hide: ('localhost', 0, 'lo', '127.0.0.1', 8) ('localhost', 0, 'eth0', '192.168.122.28', 24) Using ipaddr views ================== Interfaces also provide address views as subsets of the global address view: .. testcode:: with ndb.interfaces['eth0'] as eth0: for record in eth0.ipaddr.summary(): print(record) .. testoutput:: ('localhost', 0, 'eth0', '192.168.122.28', 24) It is possible use the same API as with the global address view: .. testcode:: with ndb.interfaces['eth0'] as eth0: eth0.ipaddr.create( address='10.0.0.1', prefixlen=24 # index is implied ).commit() for record in ndb.addresses.summary(): print(record) .. testoutput:: ('localhost', 0, 'lo', '127.0.0.1', 8) ('localhost', 0, 'eth0', '10.0.0.1', 24) ('localhost', 0, 'eth0', '192.168.122.28', 24) Using interface methods ======================= Interfaces provide also simple methods to manage addresses: .. testcode:: with ndb.interfaces['eth0'] as eth0: eth0.del_ip('192.168.122.28/24') # remove an existing address eth0.del_ip(family=AF_INET) # ... or remove all IPv4 addresses eth0.add_ip('10.0.0.1/24') # add a new IP address eth0.add_ip(address='10.0.0.2', prefixlen=24) # ... or using keywords eth0.set('state', 'up') with ndb.addresses.summary() as report: report.select_records(ifname='eth0') for address in report: print(address) .. testoutput:: ('localhost', 0, 'eth0', '10.0.0.1', 24) ('localhost', 0, 'eth0', '10.0.0.2', 24) Functions `add_ip()` and `del_ip()` return the interface object, so they can be chained as in the example above, and the final `commit()` will commit all the changes in the chain. The keywords to `del_ip()` are the same object field names that may be used in the selectors or report filters: .. testcode:: with ndb.interfaces['eth0'] as eth0: eth0.del_ip(prefixlen=24) # remove all addresses with mask /24 A match function that may be passed to the `del_ip()` is the same as for `addresses.dump().select_records()`, and it gets a named tuple as the argument. The fields are named in the same way as address objects fields. So if you want to filter addresses by a pattern or the `prefixlen` field with a match function, you may use: .. testcode:: x1 with ndb.interfaces['eth0'] as eth0: eth0.add_ip('10.0.0.1/25') with ndb.interfaces['eth0'] as eth0: eth0.del_ip(lambda x: x.address.startswith('192.168')) eth0.del_ip(lambda x: x.prefixlen == 25) An empty `del_ip()` removes all the IP addresses on the interface: .. testcode:: x2 with ndb.interfaces['eth0'] as eth0: eth0.del_ip() # flush all the IP:s Accessing one address details ============================= Access an address as a separate RTNL object: .. code:: print(ndb.addresses['192.168.122.28/24']) .. testoutput:: x3 :hide: {'target': 'localhost', 'address': '192.168.122.28', 'prefixlen': 24, \ 'tflags': 0, 'family': 2, 'index': 2, 'local': '192.168.122.28', \ 'flags': 512, 'scope': 0, 'label': 'eth0', 'broadcast': '192.168.122.255', \ 'anycast': None, 'multicast': None} Please notice that address objects are read-only, you may not change them, only remove old ones, and create new. ''' from pyroute2.netlink.rtnl.ifaddrmsg import ifaddrmsg from pyroute2.requests.address import AddressFieldFilter from ..objects import RTNL_Object def load_ifaddrmsg(schema, target, event): # # bypass # schema.load_netlink('addresses', target, event) # # last address removal should trigger routes flush # Bug-Url: https://github.com/svinota/pyroute2/issues/849 # if event['header']['type'] % 2 and event.get('index'): # # check IPv4 addresses on the interface # addresses = schema.execute( ''' SELECT * FROM addresses WHERE f_target = %s AND f_index = %s AND f_family = 2 ''' % (schema.plch, schema.plch), (target, event['index']), ).fetchmany() if not len(addresses): schema.execute( ''' DELETE FROM routes WHERE f_target = %s AND f_RTA_OIF = %s OR f_RTA_IIF = %s ''' % (schema.plch, schema.plch, schema.plch), (target, event['index'], event['index']), ) # Take care of multipath routes schema.execute( ''' DELETE FROM nh WHERE f_target = %s AND f_oif = %s ''' % (schema.plch, schema.plch), (target, event['index']), ) schema.execute( ''' DELETE FROM routes WHERE f_target = %s AND f_deps = 1 AND f_route_id NOT IN (SELECT n.f_route_id FROM nh n) ''' % (schema.plch,), (target,), ) ifaddr_spec = ( ifaddrmsg.sql_schema() .unique_index('family', 'prefixlen', 'index', 'IFA_ADDRESS', 'IFA_LOCAL') .foreign_key( 'interfaces', ('f_target', 'f_tflags', 'f_index'), ('f_target', 'f_tflags', 'f_index'), ) ) init = { 'specs': [['addresses', ifaddr_spec]], 'classes': [['addresses', ifaddrmsg]], 'event_map': {ifaddrmsg: [load_ifaddrmsg]}, } class Address(RTNL_Object): table = 'addresses' msg_class = ifaddrmsg field_filter = AddressFieldFilter api = 'addr' @classmethod def _count(cls, view): if view.chain: return view.ndb.task_manager.db_fetchone( 'SELECT count() FROM %s WHERE f_index = %s' % (view.table, view.ndb.schema.plch), [view.chain['index']], ) else: return view.ndb.task_manager.db_fetchone( 'SELECT count() FROM %s' % view.table ) @classmethod def _dump_where(cls, view): if view.chain: plch = view.ndb.schema.plch where = ''' WHERE main.f_target = %s AND main.f_index = %s ''' % ( plch, plch, ) values = [view.chain['target'], view.chain['index']] else: where = '' values = [] return (where, values) @classmethod def summary(cls, view): req = ''' SELECT main.f_target, main.f_tflags, intf.f_IFLA_IFNAME, main.f_IFA_ADDRESS, main.f_prefixlen FROM addresses AS main INNER JOIN interfaces AS intf ON main.f_index = intf.f_index AND main.f_target = intf.f_target ''' yield ('target', 'tflags', 'ifname', 'address', 'prefixlen') where, values = cls._dump_where(view) for record in view.ndb.task_manager.db_fetch(req + where, values): yield record def mark_tflags(self, mark): plch = (self.schema.plch,) * 3 self.schema.execute( ''' UPDATE interfaces SET f_tflags = %s WHERE f_index = %s AND f_target = %s ''' % plch, (mark, self['index'], self['target']), ) def __init__(self, argv, kwarg): kwarg['iclass'] = ifaddrmsg self.event_map = {ifaddrmsg: "load_rtnlmsg"} super(Address, self).__init__(argv, **kwarg) @staticmethod def compare_record(left, right): if isinstance(right, str): return right == left['address'] or right == '%s/%i' % ( left['address'], left['prefixlen'], ) @classmethod def spec_normalize(cls, spec): ''' Address key normalization:: { ... } -> { ... } "10.0.0.1/24" -> {"address": "10.0.0.1", "prefixlen": 24} ''' if isinstance(spec, str): return {'address': spec} return spec def key_repr(self): return '%s/%s %s/%s' % ( self.get('target', ''), self.get('label', self.get('index', '')), self.get('local', self.get('address', '')), self.get('prefixlen', ''), )	9.53 KB	2025-05-25 12:26
📄 interface.py ''' .. testsetup:: from pyroute2 import NDB from pyroute2 import config config.mock_netlink = True .. testsetup:: preset_1 from pyroute2 import NDB from pyroute2 import config config.mock_netlink = True ndb = NDB( sources=[ {'target': 'localhost', 'kind': 'local'}, {'target': 'worker1.sample.com', 'kind': 'local'}, {'target': 'worker2.sample.com', 'kind': 'local'}, ] ) .. testsetup:: preset_br0_1 from pyroute2 import NDB from pyroute2 import config config.mock_netlink = True ndb = NDB() ndb.interfaces.create(ifname='eth1', kind='dummy').commit() ndb.interfaces.create(ifname='br0', kind='bridge').commit() ndb.interfaces.create(ifname='bond0', kind='bond').commit() .. testsetup:: preset_br0_2 from pyroute2 import NDB from pyroute2 import config config.mock_netlink = True ndb = NDB() ndb.interfaces.create(ifname='br0', kind='bridge').commit() ndb.interfaces['br0'].add_port('eth0').commit() List interfaces =============== List interface keys: .. testcode:: with NDB(log='on') as ndb: for key in ndb.interfaces: print(key) .. testoutput:: :hide: ('localhost', 0, 0, 772, 1, 1, 0, '00:00:00:00:00:00', \ '00:00:00:00:00:00', 'lo', 65536, None, 'noqueue', None, 1000, 'UNKNOWN', 0, \ None, None, None, 0, None, 0, 1, 1, 1, 0, None, None, 0, 65535, 65536, None, \ None, None, 0, 0, None, None, None, None, None, None, 65536, None, None, \ 'up', None, None, None, None, None, None, None, None, '[]') ('localhost', 0, 0, 772, 2, 1, 0, '52:54:00:72:58:b2', \ 'ff:ff:ff:ff:ff:ff', 'eth0', 1500, None, 'fq_codel', None, 1000, 'UNKNOWN', \ 0, None, None, None, 0, None, 0, 1, 1, 1, 0, None, None, 0, 65535, 65536, \ None, None, None, 0, 0, None, None, None, None, None, None, 65536, None, \ None, 'up', None, None, None, None, None, None, None, None, '[]') NDB views support some dict methods: `items()`, `values()`, `keys()`: .. testcode:: with NDB(log='on') as ndb: for key, nic in ndb.interfaces.items(): nic.set('state', 'up') nic.commit() Get interface objects ===================== The keys may be used as selectors to get interface objects: .. testcode:: with NDB() as ndb: for key in ndb.interfaces: print(ndb.interfaces[key]) .. testoutput:: :hide: :options: +ELLIPSIS ... Also possible selector formats are `dict()` and simple string. The latter means the interface name: .. testcode:: preset_1 eth0 = ndb.interfaces['eth0'] Dict selectors are necessary to get interfaces by other properties: .. testcode:: preset_1 wrk1_eth0 = ndb.interfaces[{'target': 'worker1.sample.com', 'ifname': 'eth0'}] wrk2_eth0 = ndb.interfaces[{'target': 'worker2.sample.com', 'address': '52:54:00:72:58:b2'}] Change nic properties ===================== Changing MTU and MAC address: .. testcode:: preset_1 with ndb.interfaces['eth0'] as eth0: eth0['mtu'] = 1248 eth0['address'] = '00:11:22:33:44:55' # --> <-- eth0.commit() is called by the context manager One can change a property either using the assignment statement, or using the `.set()` routine: .. testcode:: preset_1 # same code with ndb.interfaces['eth0'] as eth0: eth0.set('mtu', 1248) eth0.set('address', '00:11:22:33:44:55') Create virtual interfaces ========================= Create a bridge and add a port, `eth0`: .. testcode:: preset_1 with ndb.interfaces.create(ifname='br0', kind='bridge') as br0: br0.add_port('eth0') Bridge and bond ports ===================== Add bridge and bond ports one can use specific API: .. code:: with ndb.interfaces['br0'] as br0: br0.add_port('eth0') br0.add_port('eth1') br0.set('br_max_age', 1024) br0.set('br_forward_delay', 1500) with ndb.interfaces['bond0'] as bond0: bond0.add_port('eth0') bond0.add_port('eth1') To remove a port: .. code:: with ndb.interfaces['br0'] as br0: br0.del_port('eth0') Or by setting the master property on a port, in the same way as with `IPRoute`: .. code:: index = ndb.interfaces['br0']['index'] # add a port to a bridge with ndb.interfaces['eth0'] as eth0: eth0.set('master', index) # remove a port from a bridge with ndb.interfaces['eth0'] as eth0: eth0.set('master', 0) ''' import errno import json import traceback from pyroute2.common import basestring from pyroute2.config import AF_BRIDGE from pyroute2.netlink.exceptions import NetlinkError from pyroute2.netlink.rtnl.ifinfmsg import ifinfmsg from pyroute2.netlink.rtnl.p2pmsg import p2pmsg from pyroute2.requests.link import LinkFieldFilter from ..auth_manager import AuthManager, check_auth from ..objects import RTNL_Object def load_ifinfmsg(schema, target, event): # # link goes down: flush all related routes # if not event['flags'] & 1: schema.execute( 'DELETE FROM routes WHERE ' 'f_target = %s AND ' 'f_RTA_OIF = %s OR f_RTA_IIF = %s' % (schema.plch, schema.plch, schema.plch), (target, event['index'], event['index']), ) # # ignore wireless updates # if event.get_attr('IFLA_WIRELESS'): return # # IFLA_PROP_LIST, IFLA_ALT_IFNAME # prop_list = event.get('IFLA_PROP_LIST') event['alt_ifname_list'] = [] if prop_list is not None: for ifname in prop_list.altnames(): event['alt_ifname_list'].append(ifname) # # AF_BRIDGE events # if event['family'] == AF_BRIDGE: # schema.load_netlink('af_bridge_ifs', target, event) try: vlans = event.get_attr('IFLA_AF_SPEC').get_attrs( 'IFLA_BRIDGE_VLAN_INFO' ) except AttributeError: # AttributeError: 'NoneType' object has no attribute 'get_attrs' # -- vlan filters not supported return # flush the old vlans info schema.execute( ''' DELETE FROM af_bridge_vlans WHERE f_target = %s AND f_index = %s ''' % (schema.plch, schema.plch), (target, event['index']), ) for v in vlans: v['index'] = event['index'] v['header'] = {'type': event['header']['type']} schema.load_netlink('af_bridge_vlans', target, v) return schema.load_netlink('interfaces', target, event) # # load ifinfo, if exists # if not event['header'].get('type', 0) % 2: linkinfo = event.get_attr('IFLA_LINKINFO') if linkinfo is not None: iftype = linkinfo.get_attr('IFLA_INFO_KIND') table = 'ifinfo_%s' % iftype if iftype == 'gre': ifdata = linkinfo.get_attr('IFLA_INFO_DATA') local = ifdata.get_attr('IFLA_GRE_LOCAL') remote = ifdata.get_attr('IFLA_GRE_REMOTE') p2p = p2pmsg() p2p['index'] = event['index'] p2p['family'] = 2 p2p['attrs'] = [('P2P_LOCAL', local), ('P2P_REMOTE', remote)] schema.load_netlink('p2p', target, p2p) elif iftype == 'veth': link = event.get_attr('IFLA_LINK') ifname = event.get_attr('IFLA_IFNAME') # for veth interfaces, IFLA_LINK points to # the peer -- but NOT in automatic updates if (not link) and ( (target,) in schema.fetch('SELECT f_target FROM SOURCES') ): schema.log.debug('reload veth %s' % event['index']) try: update = schema.sources[target].api( 'link', 'get', index=event['index'] ) update = tuple(update)[0] return schema.load_netlink( 'interfaces', target, update ) except NetlinkError as e: if e.code == errno.ENODEV: schema.log.debug(f"interface has gone: {ifname}") if table in schema.spec: ifdata = linkinfo.get_attr('IFLA_INFO_DATA') if ifdata is not None: ifdata['header'] = {} ifdata['index'] = event['index'] schema.load_netlink(table, target, ifdata) ip_tunnels = ('gre', 'gretap', 'ip6gre', 'ip6gretap', 'ip6tnl', 'sit', 'ipip') schema_ifinfmsg = ( ifinfmsg.sql_schema().push('alt_ifname_list', 'TEXT').unique_index('index') ) schema_brinfmsg = ( ifinfmsg.sql_schema() .unique_index('index') .foreign_key( 'interface', ('f_target', 'f_tflags', 'f_index'), ('f_target', 'f_tflags', 'f_index'), ) ) schema_p2pmsg = ( p2pmsg.sql_schema() .unique_index('index') .foreign_key( 'interfaces', ('f_target', 'f_tflags', 'f_index'), ('f_target', 'f_tflags', 'f_index'), ) ) schema_af_bridge_vlans = ( ifinfmsg.af_spec_bridge.vlan_info.sql_schema() .push('index', 'INTEGER') .unique_index('vid', 'index') .foreign_key( 'af_bridge_ifs', ('f_target', 'f_tflags', 'f_index'), ('f_target', 'f_tflags', 'f_index'), ) ) init = { 'specs': [ ['interfaces', schema_ifinfmsg], ['af_bridge_ifs', schema_ifinfmsg], ['af_bridge_vlans', schema_af_bridge_vlans], ['p2p', schema_p2pmsg], ], 'classes': [ ['interfaces', ifinfmsg], ['af_bridge_ifs', ifinfmsg], ['vlans', ifinfmsg], ['af_bridge_vlans', ifinfmsg.af_spec_bridge.vlan_info], ['p2p', p2pmsg], ], 'event_map': {ifinfmsg: [load_ifinfmsg]}, } ifinfo_names = ( 'bridge', 'bond', 'vlan', 'vxlan', 'gre', 'gretap', 'ip6gre', 'ip6gretap', 'ip6tnl', 'ipip', 'ipvlan', 'sit', 'macvlan', 'macvtap', 'tun', 'vrf', 'vti', 'vti6', ) supported_ifinfo = {x: ifinfmsg.ifinfo.data_map[x] for x in ifinfo_names} # # load supported ifinfo # for name, data in supported_ifinfo.items(): name = 'ifinfo_%s' % name init['classes'].append([name, data]) schema = ( data.sql_schema() .push('index', 'BIGINT') .unique_index('index') .foreign_key( 'interfaces', ('f_target', 'f_tflags', 'f_index'), ('f_target', 'f_tflags', 'f_index'), ) ) init['specs'].append([name, schema]) def _cmp_master(self, value): if self['master'] == value: return True elif self['master'] == 0 and value is None: dict.__setitem__(self, 'master', None) return True return False class Vlan(RTNL_Object): table = 'af_bridge_vlans' msg_class = ifinfmsg.af_spec_bridge.vlan_info api = 'vlan_filter' @classmethod def _count(cls, view): if view.chain: return view.ndb.task_manager.db_fetchone( 'SELECT count() FROM %s WHERE f_index = %s' % (view.table, view.ndb.schema.plch), [view.chain['index']], ) else: return view.ndb.task_manager.db_fetchone( 'SELECT count() FROM %s' % view.table ) @classmethod def _dump_where(cls, view): if view.chain: plch = view.ndb.schema.plch where = ''' WHERE main.f_target = %s AND main.f_index = %s ''' % ( plch, plch, ) values = [view.chain['target'], view.chain['index']] else: where = '' values = [] return (where, values) @classmethod def summary(cls, view): req = ''' SELECT main.f_target, main.f_tflags, main.f_vid, intf.f_IFLA_IFNAME FROM af_bridge_vlans AS main INNER JOIN interfaces AS intf ON main.f_index = intf.f_index AND main.f_target = intf.f_target ''' yield ('target', 'tflags', 'vid', 'ifname') where, values = cls._dump_where(view) for record in view.ndb.task_manager.db_fetch(req + where, values): yield record @staticmethod def compare_record(left, right): if isinstance(right, int): return right == left['vid'] def __init__(self, argv, kwarg): kwarg['iclass'] = ifinfmsg.af_spec_bridge.vlan_info if 'auth_managers' not in kwarg or kwarg['auth_managers'] is None: kwarg['auth_managers'] = [] log = argv[0].ndb.log.channel('vlan auth') kwarg['auth_managers'].append( AuthManager( {'obj:read': True, 'obj:list': True, 'obj:modify': False}, log ) ) super(Vlan, self).__init__(argv, *kwarg) def make_req(self, prime): ret = {} if 'index' in self: ret['index'] = self['index'] ret['vlan_info'] = {'vid': self['vid']} if 'flags' in self: ret['vlan_info']['flags'] = self['flags'] return ret def make_idx_req(self, prime): return self.make_req(prime) class Interface(RTNL_Object): table = 'interfaces' msg_class = ifinfmsg api = 'link' key_extra_fields = ['IFLA_IFNAME'] resolve_fields = ['vxlan_link', 'link', 'master'] fields_cmp = {'master': _cmp_master} fields_load_transform = { 'alt_ifname_list': lambda x: list(json.loads(x or '[]')) } field_filter = LinkFieldFilter old_ifname = None @classmethod def _count(cls, view): if view.chain: return view.ndb.task_manager.db_fetchone( 'SELECT count() FROM %s WHERE f_IFLA_MASTER = %s' % (view.table, view.ndb.schema.plch), [view.chain['index']], ) else: return view.ndb.task_manager.db_fetchone( 'SELECT count() FROM %s' % view.table ) @classmethod def _dump_where(cls, view): if view.chain: plch = view.ndb.schema.plch where = ''' WHERE f_target = %s AND f_IFLA_MASTER = %s ''' % ( plch, plch, ) values = [view.chain['target'], view.chain['index']] else: where = 'WHERE f_index != 0' values = [] return (where, values) @classmethod def summary(cls, view): req = ''' SELECT f_target, f_tflags, f_index, f_IFLA_IFNAME, f_IFLA_ADDRESS, f_flags, f_IFLA_INFO_KIND FROM interfaces ''' yield ( 'target', 'tflags', 'index', 'ifname', 'address', 'flags', 'kind', ) where, values = cls._dump_where(view) for record in view.ndb.task_manager.db_fetch(req + where, values): yield record def mark_tflags(self, mark): plch = (self.schema.plch,) 3 self.schema.execute( ''' UPDATE interfaces SET f_tflags = %s WHERE f_index = %s AND f_target = %s ''' % plch, (mark, self['index'], self['target']), ) def __init__(self, argv, kwarg): kwarg['iclass'] = ifinfmsg self.event_map = {ifinfmsg: "load_rtnlmsg"} self._alt_ifname_orig = set() dict.__setitem__(self, 'alt_ifname_list', list()) dict.__setitem__(self, 'state', 'unknown') warnings = [] if isinstance(argv[1], dict): if 'reuse' in argv[1]: warnings.append('ignore IPDB-specific `reuse` keyword') del argv[1]['reuse'] if argv[1].get('create') and 'ifname' not in argv[1]: raise Exception('specify at least ifname') # type specific cases if argv[1].get('kind') == 'tuntap': # translate custom tuntap format into the native tun warnings.append('translated tuntap ifinfo into tun, no flags') argv[1]['kind'] = 'tun' if argv[1].get('mode') == 'tun': argv[1]['tun_type'] = 1 elif argv[1].get('mode') == 'tap': argv[1]['tun_type'] = 2 else: raise TypeError('tun type error') del argv[1]['mode'] if 'uid' in argv[1]: argv[1]['tun_owner'] = argv[1].pop('uid') if 'gid' in argv[1]: argv[1]['tun_owner'] = argv[1].pop('gid') super(Interface, self).__init__(argv, kwarg) for line in warnings: self.log.warning(line) @property def ipaddr(self): return self.view.ndb._get_view('addresses', chain=self) @property def ports(self): return self.view.ndb._get_view('interfaces', chain=self) @property def routes(self): return self.view.ndb._get_view('routes', chain=self) @property def neighbours(self): return self.view.ndb._get_view('neighbours', chain=self) @property def vlans(self): return self.view.ndb._get_view('af_bridge_vlans', chain=self) @property def context(self): ctx = {} if self.get('target'): ctx['target'] = self['target'] if self.get('index'): ctx['index'] = self['index'] return ctx @classmethod def compare_record(self, left, right): # specific compare if isinstance(right, basestring): return right == left['ifname'] or right == left['address'] @check_auth('obj:modify') def add_vlan(self, spec): def do_add_vlan(self, mode, spec): try: method = getattr(self.vlan.create(spec), mode) return [method()] except Exception as e_s: e_s.trace = traceback.format_stack() return [e_s] self._apply_script.append((do_add_vlan, {'spec': spec})) return self @check_auth('obj:modify') def ensure_vlan(self, spec): def do_ensure_vlan(self, mode, spec): try: method = getattr(self.vlan.create(spec), mode) return [method()] except KeyError: return [] except Exception as e_s: e_s.trace = traceback.format_stack() return [e_s] self._apply_script.append((do_ensure_vlan, {'spec': spec})) return self @check_auth('obj:modify') def del_vlan(self, spec): def do_del_vlan(self, mode, spec): try: method = getattr(self.vlan[spec].remove(), mode) return [method()] except Exception as e_s: e_s.trace = traceback.format_stack() return [e_s] self._apply_script.append((do_del_vlan, {'spec': spec})) return self @check_auth('obj:modify') def add_neighbour(self, spec=None, kwarg): spec = spec or kwarg def do_add_neighbour(self, mode, spec): try: method = getattr(self.neighbours.create(spec), mode) return [method()] except Exception as e_s: e_s.trace = traceback.format_stack() return [e_s] self._apply_script.append((do_add_neighbour, {'spec': spec})) return self @check_auth('obj:modify') def ensure_neighbour(self, spec=None, kwarg): spec = spec or kwarg def do_ensure_neighbour(self, mode, spec): try: method = getattr(self.neighbours.create(spec), mode) return [method()] except KeyError: return [] except Exception as e_s: e_s.trace = traceback.format_stack() return [e_s] self._apply_script.append((do_ensure_neighbour, {'spec': spec})) return self @check_auth('obj:modify') def del_neighbour(self, spec=None, kwarg): spec = spec or dict(kwarg) def do_del_neighbour(self, mode, spec): ret = [] if isinstance(spec, basestring): specs = [spec] elif callable(spec): specs = self.ipaddr.dump() specs.select_records(spec) else: specs = self.ipaddr.dump() specs.select_records(spec) for sp in specs: try: method = getattr(self.neighbours.locate(sp).remove(), mode) ret.append(method()) except KeyError: pass except Exception as e_s: e_s.trace = traceback.format_stack() ret.append(e_s) if not ret: ret = KeyError('no neighbour records matched') return ret self._apply_script.append((do_del_neighbour, {'spec': spec})) return self @check_auth('obj:modify') def add_ip(self, spec=None, kwarg): spec = spec or kwarg def do_add_ip(self, mode, spec): try: method = getattr(self.ipaddr.create(spec), mode) return [method()] except Exception as e_s: e_s.trace = traceback.format_stack() return [e_s] self._apply_script.append((do_add_ip, {'spec': spec})) return self @check_auth('obj:modify') def ensure_ip(self, spec=None, kwarg): spec = spec or kwarg def do_ensure_ip(self, mode, spec): try: method = getattr(self.ipaddr.create(spec), mode) return [method()] except KeyError: return [] except Exception as e_s: e_s.trace = traceback.format_stack() return [e_s] self._apply_script.append((do_ensure_ip, {'spec': spec})) return self @check_auth('obj:modify') def del_ip(self, spec=None, kwarg): spec = spec or kwarg def do_del_ip(self, mode, spec): ret = [] if isinstance(spec, basestring): specs = [spec] elif callable(spec): specs = self.ipaddr.dump() specs.select_records(spec) else: specs = self.ipaddr.dump() specs.select_records(spec) for sp in specs: try: method = getattr(self.ipaddr.locate(sp).remove(), mode) ret.append(method()) except KeyError: pass except Exception as e_s: e_s.trace = traceback.format_stack() ret.append(e_s) if not ret: ret = KeyError('no address records matched') return ret self._apply_script.append((do_del_ip, {'spec': spec})) return self @check_auth('obj:modify') def add_port(self, spec): def do_add_port(self, mode, spec): try: port = self.view[spec] if port['target'] != self['target']: raise ValueError('target must be the same') port['master'] = self['index'] getattr(port, mode)() return [port] except Exception as e_s: e_s.trace = traceback.format_stack() return [e_s] self._apply_script.append((do_add_port, {'spec': spec})) return self @check_auth('obj:modify') def del_port(self, spec): def do_del_port(self, mode, spec): try: port = self.view[spec] if port['master'] != self['index']: raise ValueError('wrong port master index') if port['target'] != self['target']: raise ValueError('target must be the same') port['master'] = 0 getattr(port, mode)() return [port] except Exception as e_s: e_s.trace = traceback.format_stack() return [e_s] self._apply_script.append((do_del_port, {'spec': spec})) return self @check_auth('obj:modify') def add_altname(self, ifname): new_list = set(self['alt_ifname_list']) new_list.add(ifname) self['alt_ifname_list'] = list(new_list) return self @check_auth('obj:modify') def del_altname(self, ifname): new_list = set(self['alt_ifname_list']) new_list.remove(ifname) self['alt_ifname_list'] = list(new_list) return self @check_auth('obj:modify') def __setitem__(self, key, value): if key == 'peer': dict.__setitem__(self, key, value) elif key == 'ifname': if value in self['alt_ifname_list']: self.del_altname(value) if key in self and self.old_ifname is None: self.old_ifname = self[key] super(Interface, self).__setitem__(key, value) elif key == 'target' and self.state == 'invalid': dict.__setitem__(self, key, value) elif key == 'net_ns_fd' and self.state == 'invalid': dict.__setitem__(self, 'target', value) elif ( key == 'target' and self.get('target') and self['target'] != value ): super(Interface, self).__setitem__('net_ns_fd', value) else: super(Interface, self).__setitem__(key, value) @classmethod def spec_normalize(cls, spec): ''' Interface key normalization:: { ... } -> { ... } "eth0" -> {"ifname": "eth0", ...} 1 -> {"index": 1, ...} ''' if isinstance(spec, basestring): return {'ifname': spec} elif isinstance(spec, int): return {'index': spec} elif isinstance(spec, ifinfmsg): ret = {'ifname': spec.get('ifname')} if spec.get('index') != 0: ret['index'] = spec.get('index') return ret return spec def complete_key(self, key): if isinstance(key, dict): ret_key = key else: ret_key = {'target': self.ndb.localhost} if isinstance(key, basestring): ret_key['ifname'] = key elif isinstance(key, int): ret_key['index'] = key return super(Interface, self).complete_key(ret_key) def is_peer(self, other): '''Evaluate whether the given interface "points at" this one.''' if other['kind'] == 'vlan': return ( other['target'] == self['target'] and other['link'] == self['index'] ) elif other['kind'] == 'vxlan': return ( other['target'] == self['target'] and other['vxlan_link'] == self['index'] ) elif other['kind'] == self['kind'] == 'veth': other_link = other.get('link') if other_link != self['index']: return False other_link_netnsid = other.get('link_netnsid') if other_link_netnsid is not None: self_source = self.sources[self['target']] other_source = other.sources[other['target']] info = other_source.api( 'get_netnsid', pid=self_source.api('get_pid'), target_nsid=other_link_netnsid, ) return info['current_nsid'] == other_link_netnsid return self['target'] == other['target'] def set_xdp_fd(self, fd): self.sources[self['target']].api( 'link', 'set', index=self['index'], xdp_fd=fd ) def snapshot(self, ctxid=None): # 1. make own snapshot snp = super(Interface, self).snapshot(ctxid=ctxid) # 2. collect dependencies and store in self.snapshot_deps for spec in self.ndb.interfaces.getmany( {'IFLA_MASTER': self['index']} ): # bridge ports link = type(self)( self.view, spec, auth_managers=self.auth_managers ) snp.snapshot_deps.append((link, link.snapshot())) for spec in self.ndb.interfaces.getmany({'IFLA_LINK': self['index']}): link = type(self)( self.view, spec, auth_managers=self.auth_managers ) # vlans & veth if self.is_peer(link) and not link.is_peer(self): snp.snapshot_deps.append((link, link.snapshot())) # return the root node return snp def make_req(self, prime): req = super(Interface, self).make_req(prime) # # --> link('set', ...) if self.state == 'system': req['master'] = self['master'] # # FIXME: make type plugins? kind = self['kind'] if kind in ip_tunnels: req['kind'] = kind for key in self: if ( key.startswith(f'{kind}_') and key not in req and self[key] ): req[key] = self[key] return req @check_auth('obj:modify') def apply_altnames( self, alt_ifname_setup, alt_ifname_current, old_ifname=None ): if 'alt_ifname_list' in self.changed: self.changed.remove('alt_ifname_list') if alt_ifname_current is None: # load the current state self.load_from_system() self.load_sql(set_state=False) alt_ifname_current = set(self['alt_ifname_list']) alt_ifname_remove = alt_ifname_current - alt_ifname_setup alt_ifname_add = alt_ifname_setup - alt_ifname_current for ifname in alt_ifname_remove: self.sources[self['target']].api( 'link', 'property_del', index=self['index'], altname=ifname ) for ifname in alt_ifname_add: self.sources[self['target']].api( 'link', 'property_add', index=self['index'], altname=ifname ) # reload alt ifnames from the system to check the state self.load_from_system() self.load_sql(set_state=False) if old_ifname is not None and old_ifname in self['alt_ifname_list']: alt_ifname_setup.add(old_ifname) if set(self['alt_ifname_list']) != alt_ifname_setup: raise Exception('could not setup alt ifnames') @check_auth('obj:modify') def apply(self, rollback=False, req_filter=None, mode='apply'): # translate string link references into numbers for key in ('link', 'master'): if key in self and isinstance(self[key], basestring): self[key] = self.ndb.interfaces[self[key]]['index'] setns = self.state.get() == 'setns' remove = self.state.get() == 'remove' alt_ifname_setup = set(self['alt_ifname_list']) old_ifname = self.old_ifname if 'ifname' in self.changed else None try: if 'index' in self and ( self.old_ifname or 'alt_ifname_list' in self.changed ): self.apply_altnames(alt_ifname_setup, None) if 'alt_ifname_list' in self.changed: self.changed.remove('alt_ifname_list') super(Interface, self).apply(rollback, req_filter, mode) if setns and self['net_ns_fd'] in self.sources: self.load_value('target', self['net_ns_fd']) dict.__setitem__(self, 'net_ns_fd', None) for link in self.sources[self['target']].api( 'link', 'get', ifname=self['ifname'] ): # after interface move the name is the same, # but the index may change # # in this case .load_sql() will not update # the object, and the engine will try to apply # the interface's attributes to another interface # with the same index as our old one # # so resync the index first # # Bug-Url: https://github.com/svinota/pyroute2/issues/1181 # self.load_value('index', link['index']) break spec = self.load_sql() if spec: self.state.set('system') if not remove and self.state != 'invalid': self.apply_altnames( alt_ifname_setup, set(self['alt_ifname_list']), old_ifname ) except NetlinkError as e: if ( e.code == 95 and self.get('master') is not None and self.get('master') > 0 and self.state == 'invalid' ): # # on some old kernels it is impossible to create # interfaces with master set; attempt to do it in # two steps def req_filter(req): return dict( [ x for x in req.items() if not x[0].startswith('master') ] ) self.apply(rollback, req_filter, mode) self.apply(rollback, None, mode) elif ( e.code == 95 and self.get('br_vlan_filtering') is not None and self.get('br_vlan_filtering') == 0 ): # # if vlan filtering is not enabled, then the parameter # is reported by netlink, but not accepted upon bridge # creation, so simply strip it def req_filter(req): return dict( [ x for x in req.items() if not x[0].startswith('br_vlan_') ] ) self.apply(rollback, req_filter, mode) else: raise finally: self.old_ifname = None if ('net_ns_fd' in self.get('peer', {})) and ( self['peer']['net_ns_fd'] in self.view.ndb.sources ): # wait for the peer in net_ns_fd, only if the netns # is connected to the NDB instance self.view.wait( target=self['peer']['net_ns_fd'], ifname=self['peer']['ifname'], timeout=5, ) return self def hook_apply(self, method, spec): if method == 'set': if self['kind'] == 'bridge': keys = filter(lambda x: x.startswith('br_'), self.changed) if keys: req = { 'index': self['index'], 'kind': 'bridge', 'family': AF_BRIDGE, } for key in keys: req[key] = self[key] self.sources[self['target']].api(self.api, method, *req) # FIXME: make a reasonable shortcut for this self.load_from_system() elif self['kind'] in ip_tunnels and self['state'] == 'down': # force reading attributes for tunnels in the down state self.load_from_system() elif method == 'add': if self['kind'] == 'tun': self.load_sql() self.load_event.wait(0.1) if 'index' not in self: raise NetlinkError(errno.EAGAIN) update = self.sources[self['target']].api( self.api, 'get', index=self['index'] ) self.ndb._event_queue.put(update) def load_from_system(self): self.load_event.clear() ( self.ndb._event_queue.put( self.sources[self['target']].api( self.api, 'get', index=self['index'] ) ) ) self.load_event.wait(1) def load_sql(self, argv, *kwarg): spec = super(Interface, self).load_sql(argv, *kwarg) if spec: tname = 'ifinfo_%s' % self['kind'] if tname in self.schema.compiled: names = self.schema.compiled[tname]['norm_names'] spec = self.ndb.task_manager.db_fetchone( 'SELECT from %s WHERE f_index = %s' % (tname, self.schema.plch), (self['index'],), ) if spec: self.update(dict(zip(names, spec))) return spec def load_rtnlmsg(self, argv, kwarg): super(Interface, self).load_rtnlmsg(argv, **kwarg) def key_repr(self): return '%s/%s' % ( self.get('target', ''), self.get('ifname', self.get('index', '')), )	37.48 KB	2025-05-25 12:26
📄 neighbour.py import errno from pyroute2.common import basestring from pyroute2.config import AF_BRIDGE from pyroute2.netlink.rtnl.ndmsg import ndmsg from pyroute2.requests.neighbour import NeighbourFieldFilter from ..events import RescheduleException from ..objects import RTNL_Object def load_ndmsg(schema, target, event): # # ignore events with ifindex == 0 # if event['ifindex'] == 0: return # if event.get_attr('NDA_IFINDEX') is None: event['attrs'].append(('NDA_IFINDEX', event['ifindex'])) # # AF_BRIDGE events # if event['family'] == AF_BRIDGE: # # bypass for now # try: schema.load_netlink('af_bridge_fdb', target, event, propagate=True) except Exception: raise RescheduleException() else: schema.load_netlink('neighbours', target, event) ndmsg_schema = ( ndmsg.sql_schema() .unique_index('ifindex', 'NDA_DST', 'NDA_VLAN') .constraint('NDA_DST', "NOT NULL DEFAULT ''") .constraint('NDA_VLAN', "NOT NULL DEFAULT 0") .foreign_key( 'interfaces', ('f_target', 'f_tflags', 'f_ifindex'), ('f_target', 'f_tflags', 'f_index'), ) ) brmsg_schema = ( ndmsg.sql_schema() .unique_index('ifindex', 'flags', 'NDA_DST', 'NDA_LLADDR', 'NDA_VLAN') .constraint('NDA_LLADDR', "NOT NULL DEFAULT ''") .constraint('NDA_DST', "NOT NULL DEFAULT ''") .constraint('NDA_VLAN', "NOT NULL DEFAULT 0") .foreign_key( 'interfaces', ('f_target', 'f_tflags', 'f_ifindex'), ('f_target', 'f_tflags', 'f_index'), ) ) init = { 'specs': [['neighbours', ndmsg_schema], ['af_bridge_fdb', brmsg_schema]], 'classes': [['neighbours', ndmsg], ['af_bridge_fdb', ndmsg]], 'event_map': {ndmsg: [load_ndmsg]}, } def fallback_add(self, idx_req, req): ( self.ndb._event_queue.put( self.sources[self['target']].api(self.api, 'dump'), source=self['target'], ) ) self.load_sql() class Neighbour(RTNL_Object): table = 'neighbours' msg_class = ndmsg field_filter = NeighbourFieldFilter api = 'neigh' @classmethod def _count(cls, view): if view.chain: return view.ndb.task_manager.db_fetchone( 'SELECT count() FROM %s WHERE f_ifindex = %s' % (view.table, view.ndb.schema.plch), [view.chain['index']], ) else: return view.ndb.task_manager.db_fetchone( 'SELECT count() FROM %s' % view.table ) @classmethod def _dump_where(cls, view): if view.chain: plch = view.ndb.schema.plch where = ''' WHERE main.f_target = %s AND main.f_ifindex = %s ''' % ( plch, plch, ) values = [view.chain['target'], view.chain['index']] else: where = '' values = [] return (where, values) @classmethod def summary(cls, view): req = f''' SELECT main.f_target, main.f_tflags, intf.f_IFLA_IFNAME, main.f_NDA_LLADDR, main.f_NDA_DST FROM {cls.table} AS main INNER JOIN interfaces AS intf ON main.f_ifindex = intf.f_index AND main.f_target = intf.f_target ''' yield ('target', 'tflags', 'ifname', 'lladdr', 'dst') where, values = cls._dump_where(view) for record in view.ndb.task_manager.db_fetch(req + where, values): yield record def __init__(self, argv, kwarg): kwarg['iclass'] = ndmsg self.event_map = {ndmsg: "load_rtnlmsg"} super(Neighbour, self).__init__(argv, **kwarg) self.fallback_for['add'][errno.EEXIST] = fallback_add def complete_key(self, key): if isinstance(key, dict): ret_key = key else: ret_key = {'target': self.ndb.localhost} if isinstance(key, basestring): ret_key['NDA_DST'] = key return super(Neighbour, self).complete_key(ret_key) def make_req(self, prime): req = super(Neighbour, self).make_req(prime) if 'vlan' in req and req['vlan'] == 0: req.pop('vlan') return req class FDBRecord(Neighbour): table = 'af_bridge_fdb' msg_class = ndmsg api = 'fdb' @classmethod def summary(cls, view): for record in super(FDBRecord, cls).summary(view): yield record[:-1] def make_idx_req(self, prime): req = super(FDBRecord, self).make_req(prime) if 'NDA_VLAN' in req and req['NDA_VLAN'] == 0: req.pop('NDA_VLAN') return req	4.79 KB	2025-05-25 12:26
📄 netns.py import warnings from pyroute2 import netns from pyroute2.common import basestring from pyroute2.netlink.rtnl.nsinfmsg import nsinfmsg from pyroute2.requests.netns import NetNSFieldFilter from ..objects import RTNL_Object def load_nsinfmsg(schema, target, event): # # check if there is corresponding source # netns_path = event.get_attr('NSINFO_PATH') if netns_path is None: schema.log.debug('ignore %s %s' % (target, event)) return if schema.config['auto_netns']: warnings.warn( 'automatic netns sourcing is being refactored', DeprecationWarning ) schema.load_netlink('netns', target, event) schema = nsinfmsg.sql_schema().unique_index('NSINFO_PATH') init = { 'specs': [['netns', schema]], 'classes': [['netns', nsinfmsg]], 'event_map': {nsinfmsg: [load_nsinfmsg]}, } class NetNS(RTNL_Object): table = 'netns' msg_class = nsinfmsg table_alias = 'n' api = 'netns' field_filter = NetNSFieldFilter def __init__(self, argv, kwarg): kwarg['iclass'] = nsinfmsg self.event_map = {nsinfmsg: "load_rtnlmsg"} super(NetNS, self).__init__(argv, **kwarg) @classmethod def spec_normalize(cls, spec): if isinstance(spec, basestring): spec = {'path': spec} path = netns._get_netnspath(spec['path']) # on Python3 _get_netnspath() returns bytes, not str, so # we have to decode it here in order to avoid issues with # cache keys and DB inserts if hasattr(path, 'decode'): path = path.decode('utf-8') spec['path'] = path return spec def __setitem__(self, key, value): if self.state == 'system': raise ValueError('attempt to change a readonly object') if key == 'path': value = netns._get_netnspath(value).decode('utf-8') return super(NetNS, self).__setitem__(key, value)	1.9 KB	2025-05-25 12:26
📄 probe.py ''' Run a network probe =================== A successful network probe neither creates real network objects like interfaces or addresses, nor database records. The only important thing it does -- it raises no exception. On the contrary, an unsuccessful network probe raises a `NetlinkError` exception, cancelling the whole transaction. A network probe is always run from the corresponding netlink target: a local system, a remote system, a network namespace, a container. An example scenario: * target alpha, set up eth0 10.0.0.2/24 * target beta, set up eth0 10.0.0.4/24 * ping 10.0.0.4 (beta) from target alpha * ping 10.0.0.2 (alpha) from target beta The code below sets up the addresses and checks ICMP responses. If any step fails, the whole transaction will be rolled back automatically:: with NDB(log='debug') as ndb: ndb.sources.add(kind='remote', hostname='alpha', username='root') ndb.sources.add(kind='remote', hostname='beta', username='root') with ndb.begin() as trx: trx.push( (ndb .interfaces[{'target': 'alpha', 'ifname': 'eth0'}] .set(state='up') .add_ip(address='10.0.0.2', prefixlen=24) ), (ndb .interfaces[{'target': 'beta', 'ifname': 'eth0'}] .set(state='up') .add_ip(address='10.0.0.4', prefixlen=24) ), (ndb .probes .create(target='alpha', kind='ping', dst='10.0.0.4') ), (ndb .probes .create(target='beta', kind='ping', dst='10.0.0.2') ), ) ''' from pyroute2.netlink.rtnl.probe_msg import probe_msg from ..objects import RTNL_Object def load_probe_msg(schema, target, event): pass schema = probe_msg.sql_schema().unique_index() init = { 'specs': [['probes', schema]], 'classes': [['probes', probe_msg]], 'event_map': {probe_msg: [load_probe_msg]}, } class Probe(RTNL_Object): table = 'probes' msg_class = probe_msg api = 'probe' def __init__(self, argv, kwarg): kwarg['iclass'] = probe_msg self.event_map = {probe_msg: 'load_probe_msg'} super().__init__(argv, **kwarg) def check(self): return True	2.35 KB	2025-05-25 12:26
📄 route.py ''' .. testsetup:: from pyroute2 import NDB from pyroute2 import config config.mock_netlink = True ndb = NDB() .. testsetup:: tables from pyroute2 import NDB from pyroute2 import config config.mock_netlink = True ndb = NDB() ndb.routes.create( dst='1.1.1.1/32', gateway='127.0.0.10', oif=1, table=101 ).commit() ndb.routes.create( dst='1.1.1.2/32', gateway='127.0.0.10', oif=1, table=5001 ).commit() ndb.routes.create( dst='1.1.1.3/32', gateway='127.0.0.10', oif=1, table=5002 ).commit() .. testsetup:: metrics from pyroute2 import NDB from pyroute2 import config config.mock_netlink = True ndb = NDB() ndb.routes.create( dst='10.0.0.0/24', gateway='127.0.0.10' ).commit() Simple routes ============= Ordinary routes management is really simple: .. testcode:: # create a route ndb.routes.create( dst='10.0.0.0/24', gateway='192.168.122.1' ).commit() # retrieve a route and change it with ndb.routes['10.0.0.0/24'] as route: route.set(gateway='192.168.122.10') # remove a route with ndb.routes['10.0.0.0/24'] as route: route.remove() Multiple routing tables ======================= But Linux systems have more than one routing table: .. doctest:: tables >>> set((x.table for x in ndb.routes.summary())) {101, 5001, 5002, 254, 255} The main routing table is 254. All the routes people mostly work with are in that table. To address routes in other routing tables, you can use dict specs: .. testcode:: ndb.routes.create( dst='10.0.0.0/24', gateway='192.168.122.1', table=101 ).commit() with ndb.routes[{'table': 101, 'dst': '10.0.0.0/24'}] as route: route.set('gateway', '192.168.122.10') route.set('priority', 500) with ndb.routes[{'table': 101, 'dst': '10.0.0.0/24'}] as route: route.remove() Route metrics ============= `route['metrics']` attribute provides a dictionary-like object that reflects route metrics like hop limit, mtu etc: .. testcode:: metrics # set up all metrics from a dictionary with ndb.routes['10.0.0.0/24'] as route: route.set('metrics', {'mtu': 1500, 'hoplimit': 20}) # fix individual metrics with ndb.routes['10.0.0.0/24']['metrics'] as metrics: metrics.set('mtu', 1500) metrics.set('hoplimit', 20) MPLS routes =========== See here: :ref:`mpls` ''' import json import struct import time import uuid from collections import OrderedDict from functools import partial from socket import AF_INET, inet_pton from pyroute2.common import AF_MPLS, basestring from pyroute2.netlink.rtnl.rtmsg import LWTUNNEL_ENCAP_MPLS, nh, rtmsg from pyroute2.requests.common import MPLSTarget from pyroute2.requests.route import RouteFieldFilter from ..auth_manager import check_auth from ..objects import RTNL_Object from ..report import Record _dump_rt = ['main.f_%s' % x[0] for x in rtmsg.sql_schema()][:-2] _dump_nh = ['nh.f_%s' % x[0] for x in nh.sql_schema()][:-2] F_RTA_MULTIPATH = 1 F_RTA_ENCAP = 2 F_RTA_METRICS = 4 def get_route_id(schema, target, event): keys = ['f_target = %s' % schema.plch] values = [target] for key in schema.indices['routes']: keys.append('f_%s = %s' % (key, schema.plch)) values.append(event.get(key) or event.get_attr(key)) # spec = 'WHERE %s' % ' AND '.join(keys) s_req = 'SELECT f_route_id FROM routes %s' % spec # # get existing route_id for route_id in schema.execute(s_req, values).fetchall(): # # if exists return route_id[0][0] # # or create a new route_id return str(uuid.uuid4()) def load_rtmsg(schema, target, event): route_id = None post = [] # fix RTA_TABLE rta_table = event.get_attr('RTA_TABLE', -1) if rta_table == -1: event['attrs'].append(['RTA_TABLE', 254]) # # manage gc marks on related routes # # only for automatic routes: # - table 254 (main) # - proto 2 (kernel) # - scope 253 (link) elif ( (event.get_attr('RTA_TABLE') == 254) and (event['proto'] == 2) and (event['scope'] == 253) and (event['family'] == AF_INET) ): evt = event['header']['type'] # # set f_gc_mark = timestamp for "del" events # and clean it for "new" events # try: rtmsg_gc_mark( schema, target, event, int(time.time()) if (evt % 2) else None ) except Exception as e: schema.log.error('gc_mark event: %s' % (event,)) schema.log.error('gc_mark: %s' % (e,)) # # only for RTM_NEWROUTE events # if not event['header']['type'] % 2: event['deps'] = 0 # # RTA_MULTIPATH # mp = event.get_attr('RTA_MULTIPATH') if mp: # # create key route_id = route_id or get_route_id(schema, target, event) # # load multipath for idx in range(len(mp)): mp[idx]['header'] = {} # for load_netlink() mp[idx]['route_id'] = route_id # set route_id on NH mp[idx]['nh_id'] = idx # add NH number post.append( partial( schema.load_netlink, 'nh', target, mp[idx], 'routes' ) ) event['deps'] \|= F_RTA_MULTIPATH # # RTA_ENCAP # encap = event.get_attr('RTA_ENCAP') encap_type = event.get_attr('RTA_ENCAP_TYPE') if encap_type == LWTUNNEL_ENCAP_MPLS: route_id = route_id or get_route_id(schema, target, event) # encap['header'] = {} encap['route_id'] = route_id post.append( partial( schema.load_netlink, 'enc_mpls', target, encap, 'routes' ) ) event['deps'] \|= F_RTA_ENCAP # # RTA_METRICS # metrics = event.get_attr('RTA_METRICS') if metrics: # # create key route_id = route_id or get_route_id(schema, target, event) # metrics['header'] = {} metrics['route_id'] = route_id post.append( partial( schema.load_netlink, 'metrics', target, metrics, 'routes' ) ) event['deps'] \|= F_RTA_METRICS # if route_id is not None: event['route_id'] = route_id schema.load_netlink('routes', target, event) # for procedure in post: procedure() def rtmsg_gc_mark(schema, target, event, gc_mark=None): # if gc_mark is None: gc_clause = ' AND f_gc_mark IS NOT NULL' else: gc_clause = '' # # select all routes for that OIF where f_gc_mark is not null # key_fields = ','.join(['f_%s' % x for x in schema.indices['routes']]) key_query = ' AND '.join( ['f_%s = %s' % (x, schema.plch) for x in schema.indices['routes']] ) routes = schema.execute( ''' SELECT %s,f_RTA_GATEWAY FROM routes WHERE f_target = %s AND f_RTA_OIF = %s AND f_RTA_GATEWAY IS NOT NULL %s AND f_family = 2 ''' % (key_fields, schema.plch, schema.plch, gc_clause), (target, event.get_attr('RTA_OIF')), ).fetchmany() # # get the route's RTA_DST and calculate the network # addr = event.get_attr('RTA_DST') net = struct.unpack('>I', inet_pton(AF_INET, addr))[0] & ( 0xFFFFFFFF << (32 - event['dst_len']) ) # # now iterate all the routes from the query above and # mark those with matching RTA_GATEWAY # for route in routes: # get route GW gw = route[-1] try: gwnet = struct.unpack('>I', inet_pton(AF_INET, gw))[0] & net if gwnet == net: ( schema.execute( 'UPDATE routes SET f_gc_mark = %s ' 'WHERE f_target = %s AND %s' % (schema.plch, schema.plch, key_query), (gc_mark, target) + route[:-1], ) ) except Exception as e: schema.log.error('gc_mark event: %s' % (event,)) schema.log.error('gc_mark: %s : %s' % (e, route)) rt_schema = ( rtmsg.sql_schema() .push('route_id', 'TEXT UNIQUE') .push('gc_mark', 'INTEGER') .push('deps', 'INTEGER') .unique_index( 'family', 'dst_len', 'tos', 'scope', 'type', 'RTA_DST', 'RTA_OIF', 'RTA_PRIORITY', 'RTA_TABLE', 'RTA_VIA', 'RTA_NEWDST', ) .foreign_key( 'interfaces', ('f_target', 'f_tflags', 'f_RTA_OIF'), ('f_target', 'f_tflags', 'f_index'), ) .foreign_key( 'interfaces', ('f_target', 'f_tflags', 'f_RTA_IIF'), ('f_target', 'f_tflags', 'f_index'), ) ) nh_schema = ( nh.sql_schema() .push('route_id', 'TEXT') .push('nh_id', 'INTEGER') .unique_index('route_id', 'nh_id') .foreign_key('routes', ('f_route_id',), ('f_route_id',)) .foreign_key( 'interfaces', ('f_target', 'f_tflags', 'f_oif'), ('f_target', 'f_tflags', 'f_index'), ) ) metrics_schema = ( rtmsg.metrics.sql_schema() .push('route_id', 'TEXT') .unique_index('route_id') .foreign_key('routes', ('f_route_id',), ('f_route_id',)) ) mpls_enc_schema = ( rtmsg.mpls_encap_info.sql_schema() .push('route_id', 'TEXT') .unique_index('route_id') .foreign_key('routes', ('f_route_id',), ('f_route_id',)) ) init = { 'specs': [ ['routes', rt_schema], ['nh', nh_schema], ['metrics', metrics_schema], ['enc_mpls', mpls_enc_schema], ], 'classes': [ ['routes', rtmsg], ['nh', nh], ['metrics', rtmsg.metrics], ['enc_mpls', rtmsg.mpls_encap_info], ], 'event_map': {rtmsg: [load_rtmsg]}, } class Via(OrderedDict): def __init__(self, prime=None): super(OrderedDict, self).__init__() if prime is None: prime = {} elif not isinstance(prime, dict): raise TypeError() self['family'] = prime.get('family', AF_INET) self['addr'] = prime.get('addr', '0.0.0.0') def __eq__(self, right): return ( isinstance(right, (dict)) and self['family'] == right.get('family', AF_INET) and self['addr'] == right.get('addr', '0.0.0.0') ) def __repr__(self): return repr(dict(self)) class Route(RTNL_Object): table = 'routes' msg_class = rtmsg hidden_fields = ['route_id'] api = 'route' field_filter = RouteFieldFilter _replace_on_key_change = True @classmethod def _count(cls, view): if view.chain: return view.ndb.task_manager.db_fetchone( 'SELECT count() FROM %s WHERE f_RTA_OIF = %s' % (view.table, view.ndb.schema.plch), [view.chain['index']], ) else: return view.ndb.task_manager.db_fetchone( 'SELECT count() FROM %s' % view.table ) @classmethod def _dump_where(cls, view): if view.chain: plch = view.ndb.schema.plch where = ''' WHERE main.f_target = %s AND main.f_RTA_OIF = %s ''' % ( plch, plch, ) values = [view.chain['target'], view.chain['index']] else: where = '' values = [] return (where, values) @classmethod def summary(cls, view): req = ''' WITH main AS (SELECT nr.f_target, nr.f_tflags, nr.f_RTA_TABLE, nr.f_RTA_DST, nr.f_dst_len, CASE WHEN nh.f_oif > nr.f_RTA_OIF THEN nh.f_oif ELSE nr.f_RTA_OIF END AS f_RTA_OIF, CASE WHEN nh.f_RTA_GATEWAY IS NOT NULL THEN nh.f_RTA_GATEWAY ELSE nr.f_RTA_GATEWAY END AS f_RTA_GATEWAY FROM routes AS nr LEFT JOIN nh ON nr.f_route_id = nh.f_route_id AND nr.f_target = nh.f_target) SELECT main.f_target, main.f_tflags, main.f_RTA_TABLE, intf.f_IFLA_IFNAME, main.f_RTA_DST, main.f_dst_len, main.f_RTA_GATEWAY FROM main INNER JOIN interfaces AS intf ON main.f_rta_oif = intf.f_index AND main.f_target = intf.f_target ''' yield ( 'target', 'tflags', 'table', 'ifname', 'dst', 'dst_len', 'gateway', ) where, values = cls._dump_where(view) for record in view.ndb.task_manager.db_fetch(req + where, values): yield record @classmethod def dump(cls, view): req = ''' SELECT main.f_target,main.f_tflags,%s FROM routes AS main LEFT JOIN nh AS nh ON main.f_route_id = nh.f_route_id AND main.f_target = nh.f_target ''' % ','.join( ['%s' % x for x in _dump_rt + _dump_nh + ['main.f_route_id']] ) header = ( ['target', 'tflags'] + [rtmsg.nla2name(x[7:]) for x in _dump_rt] + ['nh_%s' % nh.nla2name(x[5:]) for x in _dump_nh] + ['metrics', 'encap'] ) yield header plch = view.ndb.schema.plch where, values = cls._dump_where(view) for record in view.ndb.task_manager.db_fetch(req + where, values): route_id = record[-1] record = list(record[:-1]) if route_id is not None: # # fetch metrics metrics = tuple( view.ndb.task_manager.db_fetch( ''' SELECT * FROM metrics WHERE f_route_id = %s ''' % (plch,), (route_id,), ) ) if metrics: ret = {} names = view.ndb.schema.compiled['metrics']['norm_names'] for k, v in zip(names, metrics[0]): if v is not None and k not in ( 'target', 'route_id', 'tflags', ): ret[k] = v record.append(json.dumps(ret)) else: record.append(None) # # fetch encap enc_mpls = tuple( view.ndb.task_manager.db_fetch( ''' SELECT * FROM enc_mpls WHERE f_route_id = %s ''' % (plch,), (route_id,), ) ) if enc_mpls: record.append(enc_mpls[0][2]) else: record.append(None) else: record.extend((None, None)) yield record @classmethod def spec_normalize(cls, spec): if isinstance(spec, basestring): return {'dst': spec} return spec @classmethod def compare_record(self, left, right): if isinstance(right, str): return right == f'{left["dst"]}/{left["dst_len"]}' def _cmp_target(key, self, right): right = [MPLSTarget(x) for x in json.loads(right)] return all([x[0] == x[1] for x in zip(self[key], right)]) def _cmp_via(self, right): return self['via'] == Via(json.loads(right)) def _cmp_encap(self, right): return all([x[0] == x[1] for x in zip(self.get('encap', []), right)]) fields_cmp = { 'dst': partial(_cmp_target, 'dst'), 'src': partial(_cmp_target, 'src'), 'newdst': partial(_cmp_target, 'newdst'), 'encap': _cmp_encap, 'via': _cmp_via, } def mark_tflags(self, mark): plch = (self.schema.plch,) * 4 self.schema.execute( ''' UPDATE interfaces SET f_tflags = %s WHERE (f_index = %s OR f_index = %s) AND f_target = %s ''' % plch, (mark, self['iif'], self['oif'], self['target']), ) def __init__(self, argv, kwarg): kwarg['iclass'] = rtmsg self.event_map = {rtmsg: "load_rtnlmsg"} dict.__setitem__(self, 'multipath', []) dict.__setitem__(self, 'metrics', MetricsStub(self)) dict.__setitem__(self, 'deps', 0) super(Route, self).__init__(argv, *kwarg) def complete_key(self, key): ret_key = {} if isinstance(key, basestring): ret_key['dst'] = key elif isinstance(key, (Record, tuple, list)): return super(Route, self).complete_key(key) elif isinstance(key, dict): ret_key.update(key) else: raise TypeError('unsupported key type') if 'target' not in ret_key: ret_key['target'] = self.ndb.localhost ## # previously here was a code that injected the default # table == 254 into the key: # # table = ret_key.get('table', ret_key.get('RTA_TABLE', 254)) # if 'table' not in ret_key: # ret_key['table'] = table # # the issue with the code is that self.exists() didn't use # it, thus it was possible to get self.exists() == True and # at the same time loading from the DB resulted in an empty # record # # probably more correct behaviour would be to raise KeyError # if a route spec has no table defined, and the route is # in another table than 254; but for now routes['ipaddr/mask'] # returns records even outside of the main table if isinstance(ret_key.get('dst_len'), basestring): ret_key['dst_len'] = int(ret_key['dst_len']) if isinstance(ret_key.get('dst'), basestring): if ret_key.get('dst') == 'default': ret_key['dst'] = '' ret_key['dst_len'] = 0 elif '/' in ret_key['dst']: ret_key['dst'], ret_key['dst_len'] = ret_key['dst'].split('/') if ret_key.get('family', 0) == AF_MPLS: for field in ('dst', 'src', 'newdst', 'via'): value = ret_key.get(field, key.get(field, None)) if isinstance(value, (list, tuple, dict)): ret_key[field] = json.dumps(value) return super(Route, self).complete_key(ret_key) @property def clean(self): clean = True for s in (self['metrics'],) + tuple(self['multipath']): if hasattr(s, 'changed'): clean &= len(s.changed) == 0 return clean & super(Route, self).clean def make_req(self, prime): req = dict(prime) for key in self.changed: req[key] = self[key] if self['multipath']: req['multipath'] = self['multipath'] if self['metrics']: req['metrics'] = self['metrics'] if self.get('encap') and self.get('encap_type'): req['encap'] = { 'type': self['encap_type'], 'labels': self['encap'], } if self.get('gateway'): req['gateway'] = self['gateway'] return req @check_auth('obj:modify') def __setitem__(self, key, value): if key == 'route_id': raise ValueError('route_id is read only') elif key == 'multipath': super(Route, self).__setitem__('multipath', []) for mp in value: mp = dict(mp) if self.state == 'invalid': mp['create'] = True obj = NextHop( self, self.view, mp, auth_managers=self.auth_managers ) obj.state.set(self.state.get()) self['multipath'].append(obj) if key in self.changed: self.changed.remove(key) elif key == 'metrics': value = dict(value) if not isinstance(self['metrics'], Metrics): value['create'] = True obj = Metrics( self, self.view, value, auth_managers=self.auth_managers ) obj.state.set(self.state.get()) super(Route, self).__setitem__('metrics', obj) if key in self.changed: self.changed.remove(key) elif self.get('family', 0) == AF_MPLS and key in ( 'dst', 'src', 'newdst', ): if isinstance(value, (dict, int)): value = [value] na = [] target = None for label in value: target = MPLSTarget(label) target['bos'] = 0 na.append(target) target['bos'] = 1 super(Route, self).__setitem__(key, na) else: super(Route, self).__setitem__(key, value) @check_auth('obj:modify') def apply(self, rollback=False, req_filter=None, mode='apply'): if ( (self.get('table') == 255) and (self.get('family') == 10) and (self.get('proto') == 2) ): # skip automatic ipv6 routes with proto kernel return self else: if self.get('family', AF_INET) == AF_MPLS and not self.get('dst'): dict.__setitem__(self, 'dst', [MPLSTarget()]) return super(Route, self).apply(rollback, req_filter, mode) def load_sql(self, argv, *kwarg): super(Route, self).load_sql(argv, *kwarg) # transform MPLS if self.get('family', 0) == AF_MPLS: for field in ('newdst', 'dst', 'src', 'via'): value = self.get(field, None) if isinstance(value, basestring) and value != '': if field == 'via': na = json.loads(value) else: na = [MPLSTarget(x) for x in json.loads(value)] dict.__setitem__(self, field, na) # # fetch encap deps if self['deps'] & F_RTA_ENCAP: for _ in range(5): enc = tuple( self.task_manager.db_fetch( 'SELECT FROM enc_mpls WHERE f_route_id = %s' % (self.schema.plch,), (self['route_id'],), ) ) if enc: na = [MPLSTarget(x) for x in json.loads(enc[0][2])] self.load_value('encap', na) break time.sleep(0.1) else: self.log.error('no encap loaded for %s' % (self['route_id'],)) # # if not self.load_event.is_set(): return # # fetch metrics if self['deps'] & F_RTA_METRICS: for _ in range(5): metrics = tuple( self.task_manager.db_fetch( 'SELECT * FROM metrics WHERE f_route_id = %s' % (self.schema.plch,), (self['route_id'],), ) ) if metrics: self['metrics'] = Metrics( self, self.view, {'route_id': self['route_id']}, auth_managers=self.auth_managers, ) break time.sleep(0.1) else: self.log.error( 'no metrics loaded for %s' % (self['route_id'],) ) # # fetch multipath # # FIXME: use self['deps'] if 'nh_id' not in self and self.get('route_id') is not None: nhs = self.task_manager.db_fetch( 'SELECT * FROM nh WHERE f_route_id = %s' % (self.schema.plch,), (self['route_id'],), ) flush = False idx = 0 for nexthop in tuple(self['multipath']): if not isinstance(nexthop, NextHop): flush = True if not flush: try: spec = next(nhs) except StopIteration: flush = True for key, value in zip(nexthop.names, spec): if key in nexthop and value is None: continue else: nexthop.load_value(key, value) if flush: self['multipath'].pop(idx) continue idx += 1 for nexthop in nhs: key = {'route_id': self['route_id'], 'nh_id': nexthop[-1]} ( self['multipath'].append( NextHop( self, self.view, key, auth_managers=self.auth_managers, ) ) ) class RouteSub: def apply(self, rollback=False, req_filter=None, mode='apply'): return self.route.apply(rollback, req_filter, mode) def commit(self): return self.route.commit() def set(self, key, value): self[key] = value def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): self.commit() class NextHop(RouteSub, RTNL_Object): msg_class = nh table = 'nh' hidden_fields = ('route_id', 'target') def mark_tflags(self, mark): plch = (self.schema.plch,) * 4 self.schema.execute( ''' UPDATE interfaces SET f_tflags = %s WHERE (f_index = %s OR f_index = %s) AND f_target = %s ''' % plch, (mark, self.route['iif'], self.route['oif'], self.route['target']), ) def __init__(self, route, argv, kwarg): self.route = route kwarg['iclass'] = nh kwarg['check'] = False super(NextHop, self).__init__(argv, *kwarg) class MetricsStub(RouteSub, dict): def __init__(self, route): self.route = route def __setitem__(self, key, value): # This assignment forces the Metrics object to replace # MetricsStub; it is the MetricsStub object end of life self.route['metrics'] = {key: value} def __getitem__(self, key): raise KeyError('metrics not initialized for this route') class Metrics(RouteSub, RTNL_Object): msg_class = rtmsg.metrics table = 'metrics' hidden_fields = ('route_id', 'target') def mark_tflags(self, mark): plch = (self.schema.plch,) 4 self.schema.execute( ''' UPDATE interfaces SET f_tflags = %s WHERE (f_index = %s OR f_index = %s) AND f_target = %s ''' % plch, (mark, self.route['iif'], self.route['oif'], self.route['target']), ) def __init__(self, route, argv, kwarg): self.route = route kwarg['iclass'] = rtmsg.metrics kwarg['check'] = False super(Metrics, self).__init__(argv, **kwarg)	28.2 KB	2025-05-25 12:26
📄 rule.py from collections import OrderedDict from pyroute2.common import get_address_family from pyroute2.netlink.rtnl.fibmsg import fibmsg from ..objects import RTNL_Object schema = fibmsg.sql_schema().unique_index( 'family', 'dst_len', 'src_len', 'tos', 'action', 'flags', 'FRA_DST', 'FRA_SRC', 'FRA_IIFNAME', 'FRA_GOTO', 'FRA_PRIORITY', 'FRA_FWMARK', 'FRA_FLOW', 'FRA_TUN_ID', 'FRA_SUPPRESS_IFGROUP', 'FRA_SUPPRESS_PREFIXLEN', 'FRA_TABLE', 'FRA_FWMASK', 'FRA_OIFNAME', 'FRA_L3MDEV', 'FRA_UID_RANGE', 'FRA_PROTOCOL', 'FRA_IP_PROTO', 'FRA_SPORT_RANGE', 'FRA_DPORT_RANGE', ) init = { 'specs': [['rules', schema]], 'classes': [['rules', fibmsg]], 'event_map': {fibmsg: ['rules']}, } class Rule(RTNL_Object): table = 'rules' msg_class = fibmsg api = 'rule' _replace_on_key_change = True @classmethod def summary(cls, view): req = ''' SELECT f_target, f_tflags, f_family, f_FRA_PRIORITY, f_action, f_FRA_TABLE FROM rules ''' yield ('target', 'tflags', 'family', 'priority', 'action', 'table') for record in view.ndb.task_manager.db_fetch(req): yield record def __init__(self, argv, kwarg): kwarg['iclass'] = fibmsg self._fields = [x[0] for x in fibmsg.fields] self.event_map = {fibmsg: "load_rtnlmsg"} key = argv[1] families = set( [ get_address_family(x) for x in (key.get('src'), key.get('dst')) if x != '' and x is not None ] ) if len(families) > 1: raise TypeError('src and dst must be of the same IP family') elif len(families) == 1: key['family'] = families.pop() super(Rule, self).__init__(argv, *kwarg) def load_sql(self, argv, *kwarg): spec = super(Rule, self).load_sql(argv, **kwarg) if spec is None: return nkey = OrderedDict() for name_norm, name_raw, value in zip(self.names, self.spec, spec): if name_raw in self.kspec: nkey[name_raw] = value if name_norm not in self._fields and value in (0, ''): dict.__setitem__(self, name_norm, None) self._key = nkey return spec	2.38 KB	2025-05-25 12:26

Terminal