Wala shell

Advanced PHP File Manager

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

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:26
📄 __init__.py	0 B	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 auth_manager.py ''' AAA concept ----------- AAA refers to Authentication, Authorization and Accounting. NDB provides a minimalistic API to integrate Authorization routines, leaving the rest -- Authentication and Accounting -- to the user. Some of NDB routines and RTNL object methods are guarded with a parametrized decorator. The decorator takes the only parameter `tag`:: @check_auth('obj:read') def __getitem__(self, key): ... @check_auth('obj:modify') def __setitem__(self, key, value): ... AuthManager ----------- The tag is checked by `AuthManager.check(...)` routine. The routine is the only method that must be provided by AuthManager-compatible objects, and must be defined as:: def check(self, obj, tag): # -> True: grant access to the tag # -> False: reject access # -> raise Exception(): reject access with a specific exception ... NDB module provides an example AuthManager:: from pyroute2 import NDB from pyroute2.ndb.auth_manager import AuthManager ndb = NDB(log='debug') am = AuthManager({'obj:list': False, # deny dump(), summary() 'obj:read': True, # permit reading RTNL attributes 'obj:modify': True}, # permit add_ip(), commit() etc. ndb.log.channel('auth')) ap = ndb.auth_proxy(am) ap.interfaces.summary() # <-- fails with PermissionError You can implement custom AuthManager classes, the only requirement -- they must provide `.check(self, obj, tag)` routine, which returns `True` or `False` or raises an exception. ''' class check_auth(object): def __init__(self, tag): self.tag = tag def __call__(self, f): def guard(obj, argv, kwarg): if not getattr(obj, '_init_complete', True): return f(obj, argv, *kwarg) if not obj.auth_managers: raise PermissionError('access rejected') if all([x.check(obj, self.tag) for x in obj.auth_managers]): return f(obj, argv, **kwarg) raise PermissionError('access rejected') guard.__doc__ = f.__doc__ return guard class AuthManager(object): def __init__(self, auth, log, policy=False): self.auth = auth self.log = log self.policy = policy self.exception = PermissionError def check(self, obj, tag): ret = self.policy if isinstance(self.auth, dict): ret = self.auth.get(tag, self.policy) if not ret and self.exception: raise self.exception('%s access rejected' % (tag,)) return ret	2.58 KB	2025-05-25 12:26
📄 cli.py #!/usr/bin/env python import argparse import json import sys from pyroute2.cli.console import Console from pyroute2.cli.server import Server try: from pyroute2.cli.auth.auth_keystone import OSAuthManager except ImportError: OSAuthManager = None try: from pyroute2.cli.auth.auth_radius import RadiusAuthManager except ImportError: RadiusAuthManager = None try: import readline except ImportError: readline = None def run(): argp = argparse.ArgumentParser() for spec in ( ('-a', '[S] IP address to listen on'), ('-c', '[C] Command line to run'), ('-l', '[C,S] Log spec'), ('-m', 'Set mode (C,S)'), ('-p', '[S] Port to listen on'), ('-r', '[C] Load rc file'), ('-s', '[C,S] Load sources from a json file'), ('-x', '[S] Strict auth'), ): argp.add_argument(spec[0], help=spec[1]) argp.add_argument('script', nargs='*', help='script to run') args = argp.parse_args() commands = [] sources = None if args.s: with open(args.s, 'r') as f: sources = json.loads(f.read()) if args.m in ('S', 'server'): if args.p: port = int(args.p) else: port = 8080 auth_plugins = {} if OSAuthManager is not None: auth_plugins['keystone'] = OSAuthManager if RadiusAuthManager is not None: auth_plugins['radius:cleartext'] = RadiusAuthManager server = Server( address=args.a or 'localhost', port=port, log=args.l, sources=sources, auth_strict=args.x, auth_plugins=auth_plugins, ) server.serve_forever() return 0 else: console = Console(log=args.l, sources=sources) if readline is not None: console.set_completer(readline) if args.r: console.loadrc(args.r) for script in args.script: console.loadrc(script) if args.c: commands.append(args.c) console.interact(readfunc=lambda x: commands.pop(0)) elif not args.script: console.interact() return 1 if console.session.errors > 0 else 0 if __name__ == '__main__': rcode = run() sys.exit(rcode)	2.25 KB	2025-05-25 12:26
📄 cluster.py import json import socket from pyroute2.common import basestring from .main import NDB from .transport import Messenger, Transport def init(config): if isinstance(config, basestring): config = json.loads(config) else: config = json.load(config) hostname = config['local'].get('hostname', socket.gethostname()) messenger = Messenger( config['local']['id'], Transport(config['local']['address'], config['local']['port']), ) for target in config['local'].get('targets', []): messenger.targets.add(target) if not messenger.targets: messenger.targets.add(hostname) for peer in config.get('peers', []): messenger.add_peer(*peer) sources = config['local'].get('sources') if sources is None: sources = [{'target': hostname, 'kind': 'local'}] return NDB( log=config.get('log', 'debug'), sources=sources, localhost=sources[0]['target'], messenger=messenger, )	1003 B	2025-05-25 12:26
📄 compat.py def ipdb_interfaces_view(ndb): '''Provide read-only interfaces view with IPDB layout. In addition to standard NDB fields provides some IPDB specific fields. The method returns a simple dict structure, no background updates or system changes are supported. Please open a ticket on the project page if you are missing any attribute used in your project: https://github.com/svinota/pyroute2/issues ''' ret = {} for record in ndb.interfaces.dump(): interface = record._as_dict() interface['ipdb_scope'] = 'system' interface['ipdb_priority'] = 0 try: interface['ipaddr'] = tuple( ( (x.address, x.prefixlen) for x in ( ndb.addresses.dump().select_records(index=record.index) ) ) ) except: with ndb.addresses.summary() as report: report.select_records(ifname=f"{record.ifname}") interface['ipaddr'] = tuple( ((x.address, x.prefixlen) for x in report) ) try: interface['ports'] = tuple( ( x.index for x in ( ndb.interfaces.dump().select_records( master=record.index ) ) ) ) except: with ndb.interfaces.dump() as report: report.select_records(ifname=f"{record.ifname}") interface['ports'] = tuple((x.index for x in report)) try: interface['neighbours'] = tuple( ( x.dst for x in ( ndb.neighbours.dump().select_records( ifindex=record.index ) ) ) ) except: with ndb.neighbours.dump() as report: report.select_records(ifindex=record.index) interface['neighbours'] = tuple((x.dst for x in report)) ret[record.ifname] = interface return ret	2.2 KB	2025-05-25 12:26
📄 events.py import threading import time class SyncStart(Exception): pass class SchemaFlush(Exception): pass class SchemaReadLock(Exception): pass class SchemaReadUnlock(Exception): pass class SchemaGenericRequest(object): def __init__(self, response, argv, kwarg): self.response = response self.argv = argv self.kwarg = kwarg class MarkFailed(Exception): pass class DBMExitException(Exception): pass class ShutdownException(Exception): pass class RescheduleException(Exception): pass class InvalidateHandlerException(Exception): pass class State(object): events = None def __init__(self, prime=None, log=None, wait_list=None): wait_list = wait_list or [] self.events = [] self.log = log self.wait_list = {x: threading.Event() for x in wait_list} if prime is not None: self.load(prime) def wait(self, state, argv, *kwarg): return self.wait_list[state].wait(argv, **kwarg) def load(self, prime): self.events = [] for state in prime.events: self.events.append(state) def transition(self): if len(self.events) < 2: return None return (self.events[-2][1], self.events[-1][1]) def get(self): if not self.events: return None return self.events[-1][1] def set(self, state): for key in self.wait_list: if key == state: self.wait_list[key].set() else: self.wait_list[key].clear() if self.log is not None: self.log.debug(state) if self.events and self.events[-1][1] == state: self.events.pop() self.events.append((time.time(), state)) return state def __eq__(self, other): if not self.events: return False return self.events[-1][1] == other def __ne__(self, other): if not self.events: return True return self.events[-1][1] != other	2.02 KB	2025-05-25 12:26
📄 main.py ''' .. testsetup:: * from pyroute2 import config from pyroute2 import NDB config.mock_netlink = True ndb = NDB() .. testcleanup:: * for key, value in tuple(globals().items()): if key.startswith('ndb') and hasattr(value, 'close'): value.close() NDB is a high level network management module. IT allows to manage interfaces, routes, addresses etc. of connected systems, containers and network namespaces. In a nutshell, NDB collects and aggregates netlink events in an SQL database, provides Python objects to reflect the system state, and applies changes back to the system. The database expects updates only from the sources, no manual SQL updates are expected normally. .. aafig:: :scale: 80 :textual: +----------------------------------------------------------------+ +----------------------------------------------------------------+ \| +----------------------------------------------------------------+ \| \| \| \| \| \| \| kernel \| \|-+ \| \|-+ +----------------------------------------------------------------+ \| \| ^ \| ^ \| `netlink events` \| \| \| \| \| `inotify events` \| \| \| \| \| `...` \| \| \| \| v v \| v \| +--------------+ +--------------+ +--------------+ \| source \| \| source \| \| source \|<--\\ +--------------+ +--------------+ +--------------+ \| \| \| \| \| \| \| \| \| \\-----------------------+-----------------------/ \| \| \| parsed netlink events \| `NDB._event_queue` \| \| \| v \| +------------------------+ \| \| `NDB.__dbm__()` thread \| \| +------------------------+ \| \| \| v \| +-----------------------------+ \| \| `NDB.schema.load_netlink()` \| \| \| `NDB.objects..load()` \| \| +-----------------------------+ \| \| \| v \| +----------------------+ \| \| SQL database \| \| \| `SQLite` \| \| \| `PostgreSQL` \| \| +----------------------+ \| \| \| \| \| V \| +---------------+ \| +---------------+ \| \| +---------------+ \| \| `RTNL_Object.apply()` \| \| NDB object: \| \| \|-------------------------/ \| `interface` \| \| \| \| `address` \| \| \| \| `route` \| \|-+ \| `...` \|-+ +---------------+ .. container:: aafig-caption object names on the diagram are clickable The goal of NDB is to provide an easy access to RTNL info and entities via Python objects, like `pyroute2.ndb.objects.interface` (see also: :ref:`ndbinterfaces`), `pyroute2.ndb.objects.route` (see also: :ref:`ndbroutes`) etc. These objects do not only reflect the system state for the time of their instantiation, but continuously monitor the system for relevant updates. The monitoring is done via netlink notifications, thus no polling. Also the objects allow to apply changes back to the system and rollback the changes. On the other hand it's too expensive to create Python objects for all the available RTNL entities, e.g. when there are hundreds of interfaces and thousands of routes. Thus NDB creates objects only upon request, when the user calls `.create()` to create new objects or runs `ndb.<view>[selector]` (e.g. `ndb.interfaces['eth0']`) to access an existing object. To list existing RTNL entities NDB uses objects of the class `RecordSet` that `yield` individual `Record` objects for every entity (see also: :ref:`ndbreports`). An object of the `Record` class is immutable, doesn't monitor any updates, doesn't contain any links to other objects and essentially behaves like a simple named tuple. .. aafig:: :scale: 80 :textual: +---------------------+ \| \| \| \| \| `NDB() instance` \| \| \| \| \| +---------------------+ \| \| +-------------------+ +-------------------+ \| +-------------------+ \| \|-----------+--------------------------+ \| \| \| \| \| \| \| \| \| \| \| \| \| `View()` \| \| \| \| \| \| \| \|-+ \| \| \| \|-+ \| \| +-------------------+ \| \| +------------------+ +------------------+ \| \| \| \| \| \| \| \| \| `.dump()` \| \| `.create()` \| \| `.summary()` \| \| `.__getitem__()` \| \| \| \| \| \| \| \| \| +------------------+ +------------------+ \| \| \| \| v v +-------------------+ +------------------+ \| \| +------------------+ \| \| \| +------------------+ \| \| \| `RecordSet()` \| \| `Interface()` \| \| \| \| \| \| `Address()` \| \| \| \| \| \| `Route()` \| \| \| +-------------------+ \| `Neighbour()` \| \| \| \| \| `Rule()` \| \|-+ \| \| ... \|-+ v +------------------+ +-------------------+ +-------------------+ \| +-------------------+ \| \| \| `filter()` \| \| \| \| `select()` \| \| \| \| `transform()` \| \| \| \| `join()` \| \|-+ \| ... \|-+ +-------------------+ \| v +-------------------+ +-------------------+ \| +-------------------+ \| \| \| \| \| \| \| \| \| \| \| `Record()` \| \| \| \| \| \|-+ \| \|-+ +-------------------+ .. container:: aafig-caption object names on the diagram are clickable Here are some simple NDB usage examples. More info see in the reference documentation below. Print all the interface names on the system, assume we have an NDB instance `ndb`: .. testcode:: for interface in ndb.interfaces.dump(): print(interface.ifname) .. testoutput:: lo eth0 Print the routing information in the CSV format: .. testcode:: for record in ndb.routes.summary().format('csv'): print(record) .. testoutput:: 'target','tflags','table','ifname','dst','dst_len','gateway' 'localhost',0,254,'eth0','',0,'192.168.122.1' 'localhost',0,254,'eth0','192.168.122.0',24, 'localhost',0,255,'lo','127.0.0.0',8, 'localhost',0,255,'lo','127.0.0.1',32, 'localhost',0,255,'lo','127.255.255.255',32, 'localhost',0,255,'eth0','192.168.122.28',32, 'localhost',0,255,'eth0','192.168.122.255',32, .. note:: More on report filtering and formatting: :ref:`ndbreports` Print IP addresses of interfaces in several network namespaces as: .. testcode:: netns nslist = ['netns01', 'netns02', 'netns03'] for nsname in nslist: ndb.sources.add(netns=nsname) report = ndb.addresses.summary() report.select_records(target=lambda x: x.startswith('netns')) report.select_fields('address', 'ifname', 'target') for line in report.format('json'): print(line) .. testoutput:: netns [ { "address": "127.0.0.1", "ifname": "lo", "target": "netns01" }, { "address": "127.0.0.1", "ifname": "lo", "target": "netns02" }, { "address": "127.0.0.1", "ifname": "lo", "target": "netns03" } ] Add an IP address on an interface: .. testcode:: with ndb.interfaces['eth0'] as eth0: eth0.add_ip('10.0.0.1/24') # ---> <--- NDB waits until the address setup Change an interface property: .. testcode:: with ndb.interfaces['eth0'] as eth0: eth0.set( state='up', address='00:11:22:33:44:55', ) # ---> <--- NDB waits here for the changes to be applied # the commit() is called automatically by the # context manager's __exit__() ''' import atexit import ctypes import ctypes.util import logging import logging.handlers import queue import threading from urllib.parse import urlparse from pyroute2.common import basestring ## # NDB stuff from .auth_manager import AuthManager from .events import ShutdownException from .messages import cmsg from .schema import DBProvider from .task_manager import TaskManager from .transaction import Transaction from .view import SourcesView, View log = logging.getLogger(__name__) NDB_VIEWS_SPECS = ( ('interfaces', 'interfaces'), ('addresses', 'addresses'), ('routes', 'routes'), ('neighbours', 'neighbours'), ('af_bridge_fdb', 'fdb'), ('rules', 'rules'), ('netns', 'netns'), ('probes', 'probes'), ('af_bridge_vlans', 'vlans'), ) class Log: def __init__(self, log_id=None): self.logger = None self.state = False self.log_id = log_id or id(self) self.logger = logging.getLogger('pyroute2.ndb.%s' % self.log_id) self.main = self.channel('main') def __call__(self, target=None, level=logging.INFO): if target is None: return self.logger is not None if self.logger is not None: for handler in tuple(self.logger.handlers): self.logger.removeHandler(handler) if target in ('off', False): if self.state: self.logger.setLevel(0) self.logger.addHandler(logging.NullHandler()) return if target in ('on', 'stderr'): handler = logging.StreamHandler() elif target == 'debug': handler = logging.StreamHandler() level = logging.DEBUG elif isinstance(target, basestring): url = urlparse(target) if not url.scheme and url.path: handler = logging.FileHandler(url.path) elif url.scheme == 'syslog': handler = logging.handlers.SysLogHandler( address=url.netloc.split(':') ) else: raise ValueError('logging scheme not supported') else: handler = target # set formatting only for new created logging handlers if handler is not target: fmt = '%(asctime)s %(levelname)8s %(name)s: %(message)s' formatter = logging.Formatter(fmt) handler.setFormatter(formatter) self.logger.addHandler(handler) self.logger.setLevel(level) @property def on(self): self.__call__(target='on') @property def off(self): self.__call__(target='off') def close(self): manager = self.logger.manager name = self.logger.name # the loggerDict can be huge, so don't # cache all the keys -- cache only the # needed ones purge_list = [] for logger in manager.loggerDict.keys(): if logger.startswith(name): purge_list.append(logger) # now shoot them one by one for logger in purge_list: del manager.loggerDict[logger] # don't force GC, leave it to the user del manager del name del purge_list def channel(self, name): return logging.getLogger('pyroute2.ndb.%s.%s' % (self.log_id, name)) def debug(self, argv, kwarg): return self.main.debug(argv, *kwarg) def info(self, argv, *kwarg): return self.main.info(argv, *kwarg) def warning(self, argv, *kwarg): return self.main.warning(argv, *kwarg) def error(self, argv, *kwarg): return self.main.error(argv, *kwarg) def critical(self, argv, *kwarg): return self.main.critical(argv, *kwarg) class DeadEnd: def put(self, argv, *kwarg): raise ShutdownException('shutdown in progress') class EventQueue: def __init__(self, argv, *kwarg): self._bypass = self._queue = queue.Queue(argv, *kwarg) def put(self, msg, source=None): return self._queue.put((source, msg)) def shutdown(self): self._queue = DeadEnd() def bypass(self, msg, source=None): return self._bypass.put((source, msg)) def get(self, argv, *kwarg): return self._bypass.get(argv, *kwarg) def qsize(self): return self._bypass.qsize() class AuthProxy: def __init__(self, ndb, auth_managers): self._ndb = ndb self._auth_managers = auth_managers for vtable, vname in NDB_VIEWS_SPECS: view = View(self._ndb, vtable, auth_managers=self._auth_managers) setattr(self, vname, view) class NDB: @property def nsmanager(self): return '%s/nsmanager' % self.localhost def __init__( self, sources=None, localhost='localhost', db_provider='sqlite3', db_spec=':memory:', db_cleanup=True, rtnl_debug=False, log=False, auto_netns=False, libc=None, ): if db_provider == 'postgres': db_provider = 'psycopg2' self.localhost = localhost self.schema = None self.libc = libc or ctypes.CDLL( ctypes.util.find_library('c'), use_errno=True ) self.log = Log(log_id=id(self)) self._db = None self._dbm_thread = None self._dbm_ready = threading.Event() self._dbm_shutdown = threading.Event() self._global_lock = threading.Lock() self._event_queue = EventQueue(maxsize=100) self.messenger = None # if log: if isinstance(log, basestring): self.log(log) elif isinstance(log, (tuple, list)): self.log(log) elif isinstance(log, dict): self.log(**log) else: raise TypeError('wrong log spec format') # # fix sources prime if sources is None: sources = [ {'target': self.localhost, 'kind': 'local', 'nlm_generator': 1} ] elif not isinstance(sources, (list, tuple)): raise ValueError('sources format not supported') for spec in sources: if 'target' not in spec: spec['target'] = self.localhost break am = AuthManager( {'obj:list': True, 'obj:read': True, 'obj:modify': True}, self.log.channel('auth'), ) self.sources = SourcesView(self, auth_managers=[am]) self._call_registry = {} self._nl = sources atexit.register(self.close) self._dbm_ready.clear() self._dbm_error = None self.config = { 'provider': str(DBProvider(db_provider)), 'spec': db_spec, 'rtnl_debug': rtnl_debug, 'db_cleanup': db_cleanup, 'auto_netns': auto_netns, 'recordset_pipe': 'false', } self.task_manager = TaskManager(self) self._dbm_thread = threading.Thread( target=self.task_manager.run, name='NDB main loop' ) self._dbm_thread.daemon = True self._dbm_thread.start() self._dbm_ready.wait() if self._dbm_error is not None: raise self._dbm_error for vtable, vname in NDB_VIEWS_SPECS: view = View(self, vtable, auth_managers=[am]) setattr(self, vname, view) # self.query = Query(self.schema) def _get_view(self, table, chain=None, auth_managers=None): return View(self, table, chain, auth_managers) def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): self.close() def begin(self): return Transaction(self.log.channel('transaction')) def readonly(self): return self.auth_proxy( AuthManager( {'obj:list': True, 'obj:read': True, 'obj:modify': False}, self.log.channel('auth'), ) ) def auth_proxy(self, auth_manager): return AuthProxy(self, [auth_manager]) def close(self): with self._global_lock: if self._dbm_shutdown.is_set(): return else: self._dbm_shutdown.set() if hasattr(atexit, 'unregister'): atexit.unregister(self.close) else: try: atexit._exithandlers.remove((self.close, (), {})) except ValueError: pass # shutdown the _dbm_thread self._event_queue.shutdown() self._event_queue.bypass((cmsg(None, ShutdownException()),)) self._dbm_thread.join() # shutdown the logger -- free the resources self.log.close() def backup(self, spec): self.task_manager.db_backup(spec) def reload(self, kinds=None): for source in self.sources.values(): if kinds is not None and source.kind in kinds: source.restart()	20.07 KB	2025-05-25 12:26
📄 messages.py class cmsg(dict): def __init__(self, target, payload=None): self['header'] = {'target': target} self.payload = payload class cmsg_event(cmsg): pass class cmsg_failed(cmsg): pass class cmsg_sstart(cmsg): pass	246 B	2025-05-25 12:26
📁 objects	-	2025-05-25 12:26
📄 query.py from .report import RecordSet class Query(object): def __init__(self, schema, fmt='raw'): self._schema = schema self._fmt = fmt def _formatter(self, cursor, fmt=None, header=None, transform=None): fmt = fmt or self._fmt if fmt == 'csv': if header: yield ','.join(header) for record in cursor: if transform: record = transform(record) if isinstance(record, (list, tuple)): yield ','.join([str(x) for x in record]) else: yield record elif fmt == 'raw': if header: yield header for record in cursor: if transform: record = transform(record) yield record else: raise TypeError('format not supported') def nodes(self, fmt=None): ''' List all the nodes within the cluster. ''' header = ('nodename',) return RecordSet( self._formatter( self._schema.fetch( ''' SELECT DISTINCT f_target FROM interfaces ''' ), fmt, header, ) ) def p2p_edges(self, fmt=None): ''' List point to point edges within the cluster, like GRE or PPP interfaces. ''' header = ('left_node', 'right_node') return RecordSet( self._formatter( self._schema.fetch( ''' SELECT DISTINCT l.f_target, r.f_target FROM p2p AS l INNER JOIN p2p AS r ON l.f_p2p_local = r.f_p2p_remote AND l.f_target != r.f_target ''' ), fmt, header, ) ) def l2_edges(self, fmt=None): ''' List l2 links within the cluster, reconstructed from the ARP caches on the nodes. Works as follows: 1. for every node take the ARP cache 2. for every record in the cache reconstruct two triplets: * the interface index -> the local interface name * the neighbour lladdr -> the remote node and interface name Issues: does not filter out fake lladdr, so CARP interfaces produce fake l2 edges within the cluster. ''' header = ( 'left_node', 'left_ifname', 'left_lladdr', 'right_node', 'right_ifname', 'right_lladdr', ) return RecordSet( self._formatter( self._schema.fetch( ''' SELECT DISTINCT j.f_target, j.f_IFLA_IFNAME, j.f_IFLA_ADDRESS, d.f_target, d.f_IFLA_IFNAME, j.f_NDA_LLADDR FROM (SELECT n.f_target, i.f_IFLA_IFNAME, i.f_IFLA_ADDRESS, n.f_NDA_LLADDR FROM neighbours AS n INNER JOIN interfaces AS i ON n.f_target = i.f_target AND i.f_IFLA_ADDRESS != '00:00:00:00:00:00' AND n.f_ifindex = i.f_index) AS j INNER JOIN interfaces AS d ON j.f_NDA_LLADDR = d.f_IFLA_ADDRESS AND j.f_target != d.f_target ''' ), fmt, header, ) ) def l3_edges(self, fmt=None): ''' List l3 edges. For every address on every node look if it is used as a gateway on remote nodes. Such cases are reported as l3 edges. Issues: does not report routes (edges) via point to point connections like GRE where local addresses are used as gateways. To be fixed. ''' header = ( 'source_node', 'gateway_node', 'gateway_address', 'dst', 'dst_len', ) return RecordSet( self._formatter( self._schema.fetch( ''' SELECT DISTINCT r.f_target, a.f_target, a.f_IFA_ADDRESS, r.f_RTA_DST, r.f_dst_len FROM addresses AS a INNER JOIN routes AS r ON r.f_target != a.f_target AND r.f_RTA_GATEWAY = a.f_IFA_ADDRESS AND r.f_RTA_GATEWAY NOT IN (SELECT f_IFA_ADDRESS FROM addresses WHERE f_target = r.f_target) ''' ), fmt, header, ) )	4.75 KB	2025-05-25 12:26
📄 report.py ''' .. note:: New in verision 0.5.11 .. testsetup:: from pyroute2 import config from pyroute2 import NDB config.mock_netlink = True ndb = NDB() .. testcleanup:: for key, value in tuple(globals().items()): if key.startswith('ndb') and hasattr(value, 'close'): value.close() Filtering example: .. testcode:: report = ndb.interfaces.dump() report.select_fields('index', 'ifname', 'address', 'state') report.transform_fields( address=lambda r: '%s%s.%s%s.%s%s' % tuple(r.address.split(':')) ) for record in report.format('csv'): print(record) .. testoutput:: 'index','ifname','address','state' 1,'lo','0000.0000.0000','up' 2,'eth0','5254.0072.58b2','up' ''' import json import warnings from itertools import chain from pyroute2 import cli MAX_REPORT_LINES = 10000 deprecation_notice = ''' RecordSet API is deprecated, pls refer to: https://docs.pyroute2.org/ndb_reports.html ''' def format_json(dump, headless=False): buf = [] fnames = None yield '[' for record in dump: if fnames is None: if headless: fnames = record._names else: fnames = record continue if buf: buf[-1] += ',' for line in buf: yield line buf = [] lines = json.dumps(dict(zip(fnames, record)), indent=4).split('\n') buf.append(' {') for line in sorted(lines[1:-1]): if line[-1] == ',': line = line[:-1] buf.append(' %s,' % line) buf[-1] = buf[-1][:-1] buf.append(' }') for line in buf: yield line yield ']' def format_csv(dump, headless=False): def dump_record(rec): row = [] for field in rec: if isinstance(field, int): row.append('%i' % field) elif field is None: row.append('') else: row.append("'%s'" % field) return row fnames = None for record in dump: if fnames is None and headless: fnames = True yield ','.join(dump_record(record._names)) yield ','.join(dump_record(record)) class Record: def __init__(self, names, values, ref_class=None): self._names = tuple(names) self._values = tuple(values) if len(self._names) != len(self._values): raise ValueError('names and values must have the same length') self._ref_class = ref_class def __getitem__(self, key): idx = len(self._names) for i in reversed(self._names): idx -= 1 if i == key: return self._values[idx] def __setitem__(self, argv, kwarg): raise TypeError('immutable object') def __getattribute__(self, key): if key.startswith('_'): return object.__getattribute__(self, key) else: return self[key] def __setattr__(self, key, value): if not key.startswith('_'): raise TypeError('immutable object') return object.__setattr__(self, key, value) def __iter__(self): return iter(self._values) def __repr__(self): return repr(self._values) def __len__(self): return len(self._values) def _select_fields(self, fields): return Record(fields, map(lambda x: self[x], fields), self._ref_class) def _transform_fields(self, spec): data = self._as_dict() for key, func in spec.items(): data[key] = func(self) return Record(data.keys(), data.values(), self._ref_class) def _match(self, f=None, spec): if callable(f): return f(self) for key, value in spec.items(): if not ( value(self[key]) if callable(value) else (self[key] == value) ): return False return True def _as_dict(self): ret = {} for key, value in zip(self._names, self._values): ret[key] = value return ret def __eq__(self, right): if hasattr(right, '_names'): n = all(x[0] == x[1] for x in zip(self._names, right._names)) v = all(x[0] == x[1] for x in zip(self._values, right._values)) return n and v elif isinstance(right, dict): for key, value in right.items(): if value != self[key]: break else: return True return False elif self._ref_class is not None and isinstance(right, (str, int)): return self._ref_class.compare_record(self, right) else: return all(x[0] == x[1] for x in zip(self._values, right)) class BaseRecordSet(object): def __init__(self, generator, ellipsis='(...)'): self.generator = generator self.ellipsis = ellipsis def __iter__(self): return self def __next__(self): return next(self.generator) def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): pass def __repr__(self): counter = 0 ret = [] for record in self: if isinstance(record, str): ret.append(record) else: ret.append(repr(record)) ret.append('\n') counter += 1 if self.ellipsis and counter > MAX_REPORT_LINES: ret.append(self.ellipsis) break if ret: ret.pop() return ''.join(ret) class RecordSetConfig(dict): def __init__(self, prime): if isinstance(prime, dict): for key, value in prime.items(): self[key] = value else: raise ValueError('only dict allowed') def __setitem__(self, key, value): if isinstance(value, str): value = json.loads(value) return super().__setitem__(key, value) class RecordSet(BaseRecordSet): ''' NDB views return objects of this class with `summary()` and `dump()` methods. RecordSet objects are generator-based, they do not store the data in the memory, but transform them on the fly. RecordSet filters also return objects of this class, thus making possible to make chains of filters. ''' def __init__(self, generator, config=None, ellipsis=True): super().__init__(generator, ellipsis) self.filters = [] self.config = RecordSetConfig(config) if config is not None else {} def __next__(self): while True: record = next(self.generator) for f in self.filters: record = f(record) if record is None: break else: return record @cli.show_result def select_fields(self, fields): ''' Select only chosen fields for every record: .. testcode:: report = ndb.interfaces.dump() report.select_fields('index', 'ifname') for line in report.format('csv'): print(line) .. testoutput:: 'index','ifname' 1,'lo' 2,'eth0' ''' self.filters.append(lambda x: x._select_fields(fields)) if self.config.get('recordset_pipe'): return RecordSet(self, config=self.config) @cli.show_result def select_records(self, f=None, spec): ''' Select records based on a function f() or a spec match. A spec is dictionary of pairs `field: constant` or `field: callable`: .. testcode:: report = ndb.addresses.summary() report.select_records(ifname=lambda x: x.startswith('eth')) for line in report.format('csv'): print(line) .. testoutput:: 'target','tflags','ifname','address','prefixlen' 'localhost',0,'eth0','192.168.122.28',24 ''' self.filters.append(lambda x: x if x._match(f, spec) else None) if self.config.get('recordset_pipe'): return RecordSet(self, config=self.config) @cli.show_result def transform_fields(self, kwarg): ''' Transform fields with a function. Function must accept the record as the only argument: .. testcode:: report = ndb.addresses.summary() report.transform_fields( address=lambda r: f'{r.address}/{r.prefixlen}' ) report.select_fields('ifname', 'address') for line in report.format('csv'): print(line) .. testoutput:: 'ifname','address' 'lo','127.0.0.1/8' 'eth0','192.168.122.28/24' ''' self.filters.append(lambda x: x._transform_fields(kwarg)) if self.config.get('recordset_pipe'): return RecordSet(self, config=self.config) @cli.show_result def transform(self, kwarg): warnings.warn(deprecation_notice, DeprecationWarning) def g(): for record in self.generator: if isinstance(record, Record): values = [] names = record._names for name, value in zip(names, record._values): if name in kwarg: value = kwarg[name](value) values.append(value) record = Record(names, values, record._ref_class) yield record return RecordSet(g()) @cli.show_result def filter(self, f=None, kwarg): warnings.warn(deprecation_notice, DeprecationWarning) def g(): for record in self.generator: m = True for key in kwarg: if kwarg[key] != getattr(record, key): m = False if m: if f is None: yield record elif f(record): yield record return RecordSet(g()) @cli.show_result def select(self, argv): warnings.warn(deprecation_notice, DeprecationWarning) return self.fields(argv) @cli.show_result def fields(self, fields): warnings.warn(deprecation_notice, DeprecationWarning) def g(): for record in self.generator: yield record._select_fields(fields) return RecordSet(g()) @cli.show_result def join(self, right, condition=lambda r1, r2: True, prefix=''): warnings.warn(deprecation_notice, DeprecationWarning) # fetch all the records from the right # ACHTUNG it may consume a lot of memory right = tuple(right) def g(): for r1 in self.generator: for r2 in right: if condition(r1, r2): n = tuple( chain( r1._names, ['%s%s' % (prefix, x) for x in r2._names], ) ) v = tuple(chain(r1._values, r2._values)) yield Record(n, v, r1._ref_class) return RecordSet(g()) @cli.show_result def format(self, kind): ''' Return an iterator over text lines in the chosen format. Supported formats: 'json', 'csv'. ''' if kind == 'json': return BaseRecordSet(format_json(self, headless=True)) elif kind == 'csv': return BaseRecordSet(format_csv(self, headless=True)) else: raise ValueError() def count(self): ''' Return number of records. The method exhausts the generator. ''' counter = 0 for record in self: counter += 1 return counter def __getitem__(self, key): return list(self)[key]	11.87 KB	2025-05-25 12:26
📄 schema.py ''' Backends -------- NDB stores all the records in an SQL database. By default it uses the SQLite3 module, which is a part of the Python stdlib, so no extra packages are required:: # SQLite3 -- simple in-memory DB ndb = NDB() # SQLite3 -- same as above with explicit arguments ndb = NDB(db_provider='sqlite3', db_spec=':memory:') # SQLite3 -- file DB ndb = NDB(db_provider='sqlite3', db_spec='test.db') It is also possible to use a PostgreSQL database via psycopg2 module:: # PostgreSQL -- local DB ndb = NDB(db_provider='psycopg2', db_spec={'dbname': 'test'}) # PostgreSQL -- remote DB ndb = NDB(db_provider='psycopg2', db_spec={'dbname': 'test', 'host': 'db1.example.com'}) Database backup --------------- Built-in database backup is implemented now only for SQLite3 backend. For the PostgresSQL backend you have to use external utilities like `pg_dump`:: # create an NDB instance ndb = NDB() # the defaults: db_provider='sqlite3', db_spec=':memory:' ... # dump the DB to a file ndb.backup('backup.db') SQL schema ---------- By default NDB deletes the data from the DB upon exit. In order to preserve the data, use `NDB(db_cleanup=False, ...)` Here is an example schema (may be changed with releases):: List of relations Schema \| Name \| Type \| Owner --------+------------------+-------+------- public \| addresses \| table \| root public \| af_bridge_fdb \| table \| root public \| af_bridge_ifs \| table \| root public \| af_bridge_vlans \| table \| root public \| enc_mpls \| table \| root public \| ifinfo_bond \| table \| root public \| ifinfo_bridge \| table \| root public \| ifinfo_gre \| table \| root public \| ifinfo_gretap \| table \| root public \| ifinfo_ip6gre \| table \| root public \| ifinfo_ip6gretap \| table \| root public \| ifinfo_ip6tnl \| table \| root public \| ifinfo_ipip \| table \| root public \| ifinfo_ipvlan \| table \| root public \| ifinfo_macvlan \| table \| root public \| ifinfo_macvtap \| table \| root public \| ifinfo_sit \| table \| root public \| ifinfo_tun \| table \| root public \| ifinfo_vlan \| table \| root public \| ifinfo_vrf \| table \| root public \| ifinfo_vti \| table \| root public \| ifinfo_vti6 \| table \| root public \| ifinfo_vxlan \| table \| root public \| interfaces \| table \| root public \| metrics \| table \| root public \| neighbours \| table \| root public \| netns \| table \| root public \| nh \| table \| root public \| p2p \| table \| root public \| routes \| table \| root public \| rules \| table \| root public \| sources \| table \| root public \| sources_options \| table \| root (33 rows) rtnl=# select f_index, f_ifla_ifname from interfaces; f_index \| f_ifla_ifname ---------+--------------- 1 \| lo 2 \| eth0 28 \| ip_vti0 31 \| ip6tnl0 32 \| ip6_vti0 36445 \| br0 11434 \| dummy0 3 \| eth1 (8 rows) rtnl=# select f_index, f_ifla_br_stp_state from ifinfo_bridge; f_index \| f_ifla_br_stp_state ---------+--------------------- 36445 \| 0 (1 row) Database upgrade ---------------- There is no DB schema upgrade from release to release. All the data stored in the DB is being fetched from the OS in the runtime, thus no persistence required. If you're using a PostgreSQL DB or a file based SQLite, simply drop all the tables from the DB, and NDB will create them from scratch on startup. ''' import enum import json import random import sqlite3 import sys import time import traceback from collections import OrderedDict from functools import partial from pyroute2 import config from pyroute2.common import basestring, uuid32 from pyroute2.netlink import NLM_F_REPLACE # from .objects import address, interface, neighbour, netns, probe, route, rule try: import psycopg2 except ImportError: psycopg2 = None # # the order is important # plugins = [interface, address, neighbour, route, netns, rule, probe] MAX_ATTEMPTS = 5 class DBProvider(enum.Enum): sqlite3 = 'sqlite3' psycopg2 = 'psycopg2' def __eq__(self, r): return str(self) == r def publish(f): if isinstance(f, str): def decorate(m): m.publish = f return m return decorate f.publish = True return f class DBDict(dict): def __init__(self, schema, table): self.schema = schema self.table = table @publish('get') def __getitem__(self, key): for (record,) in self.schema.fetch( f''' SELECT f_value FROM {self.table} WHERE f_key = {self.schema.plch} ''', (key,), ): return json.loads(record) raise KeyError(f'key {key} not found') @publish('set') def __setitem__(self, key, value): del self[key] self.schema.execute( f''' INSERT INTO {self.table} VALUES ({self.schema.plch}, {self.schema.plch}) ''', (key, json.dumps(value)), ) @publish('del') def __delitem__(self, key): self.schema.execute( f''' DELETE FROM {self.table} WHERE f_key = {self.schema.plch} ''', (key,), ) @publish def keys(self): for (key,) in self.schema.fetch(f'SELECT f_key FROM {self.table}'): yield key @publish def items(self): for key, value in self.schema.fetch( f'SELECT f_key, f_value FROM {self.table}' ): yield key, json.loads(value) @publish def values(self): for (value,) in self.schema.fetch(f'SELECT f_value FROM {self.table}'): yield json.loads(value) class DBSchema: connection = None event_map = None key_defaults = None snapshots = None # <table_name>: <obj_weakref> spec = OrderedDict() classes = {} # # OBS: field names MUST go in the same order as in the spec, # that's for the load_netlink() to work correctly -- it uses # one loop to fetch both index and row values # indices = {} foreign_keys = {} def __init__(self, config, sources, event_map, log_channel): global plugins self.sources = sources self.config = DBDict(self, 'config') self.stats = {} self.connection = None self.cursor = None self.log = log_channel self.snapshots = {} self.key_defaults = {} self.event_map = {} # cache locally these variables so they will not be # loaded from SQL for every incoming message; this # means also that these variables can not be changed # in runtime self.rtnl_log = config['rtnl_debug'] self.provider = config['provider'] # for plugin in plugins: # # 1. spec # for name, spec in plugin.init['specs']: self.spec[name] = spec.as_dict() self.indices[name] = spec.index self.foreign_keys[name] = spec.foreign_keys # # 2. classes # for name, cls in plugin.init['classes']: self.classes[name] = cls # self.initdb(config) # for plugin in plugins: # emap = plugin.init['event_map'] # for etype, ehndl in emap.items(): handlers = [] for h in ehndl: if isinstance(h, basestring): handlers.append(partial(self.load_netlink, h)) else: handlers.append(partial(h, self)) self.event_map[etype] = handlers self.gctime = self.ctime = time.time() def initdb(self, config): if self.connection is not None: self.close() if config['provider'] == DBProvider.sqlite3: self.connection = sqlite3.connect(config['spec']) self.plch = '?' self.connection.execute('PRAGMA foreign_keys = ON') elif config['provider'] == DBProvider.psycopg2: self.connection = psycopg2.connect(*config['spec']) self.plch = '%s' else: raise TypeError('DB provider not supported') self.cursor = self.connection.cursor() # # compile request lines # self.compiled = {} for table in self.spec.keys(): self.compiled[table] = self.compile_spec( table, self.spec[table], self.indices[table] ) self.create_table(table) # # service tables # self.execute( ''' DROP TABLE IF EXISTS sources_options ''' ) self.execute( ''' DROP TABLE IF EXISTS sources ''' ) self.execute( ''' DROP TABLE IF EXISTS config ''' ) self.execute( ''' CREATE TABLE config (f_key TEXT PRIMARY KEY, f_value TEXT NOT NULL) ''' ) self.execute( ''' CREATE TABLE IF NOT EXISTS sources (f_target TEXT PRIMARY KEY, f_kind TEXT NOT NULL) ''' ) self.execute( ''' CREATE TABLE IF NOT EXISTS sources_options (f_target TEXT NOT NULL, f_name TEXT NOT NULL, f_type TEXT NOT NULL, f_value TEXT NOT NULL, FOREIGN KEY (f_target) REFERENCES sources(f_target) ON UPDATE CASCADE ON DELETE CASCADE) ''' ) for key, value in config.items(): self.config[key] = value def merge_spec(self, table1, table2, table, schema_idx): spec1 = self.compiled[table1] spec2 = self.compiled[table2] names = spec1['names'] + spec2['names'][:-1] all_names = spec1['all_names'] + spec2['all_names'][2:-1] norm_names = spec1['norm_names'] + spec2['norm_names'][2:-1] idx = ('target', 'tflags') + schema_idx f_names = ['f_%s' % x for x in all_names] f_set = ['f_%s = %s' % (x, self.plch) for x in all_names] f_idx = ['f_%s' % x for x in idx] f_idx_match = ['%s.%s = %s' % (table2, x, self.plch) for x in f_idx] plchs = [self.plch] len(f_names) return { 'names': names, 'all_names': all_names, 'norm_names': norm_names, 'idx': idx, 'fnames': ','.join(f_names), 'plchs': ','.join(plchs), 'fset': ','.join(f_set), 'knames': ','.join(f_idx), 'fidx': ' AND '.join(f_idx_match), } def compile_spec(self, table, schema_names, schema_idx): # e.g.: index, flags, IFLA_IFNAME # names = [] # # same + two internal fields # all_names = ['target', 'tflags'] # # norm_names = ['target', 'tflags'] bclass = self.classes.get(table) for name in schema_names: names.append(name[-1]) all_names.append(name[-1]) iclass = bclass if len(name) > 1: for step in name[:-1]: imap = dict(iclass.nla_map) iclass = getattr(iclass, imap[step]) norm_names.append(iclass.nla2name(name[-1])) # # escaped names: f_index, f_flags, f_IFLA_IFNAME # # the reason: words like "index" are keywords in SQL # and we can not use them; neither can we change the # C structure # f_names = ['f_%s' % x for x in all_names] # # set the fields # # e.g.: f_flags = ?, f_IFLA_IFNAME = ? # # there are different placeholders: # ? -- SQLite3 # %s -- PostgreSQL # so use self.plch here # f_set = ['f_%s = %s' % (x, self.plch) for x in all_names] # # the set of the placeholders to use in the INSERT statements # plchs = [self.plch] * len(f_names) # # the index schema; use target and tflags in every index # idx = ('target', 'tflags') + schema_idx # # the same, escaped: f_target, f_tflags etc. # f_idx = ['f_%s' % x for x in idx] # # normalized idx names # norm_idx = [iclass.nla2name(x) for x in idx] # # match the index fields, fully qualified # # interfaces.f_index = ?, interfaces.f_IFLA_IFNAME = ? # # the same issue with the placeholders # f_idx_match = ['%s.%s = %s' % (table, x, self.plch) for x in f_idx] return { 'names': names, 'all_names': all_names, 'norm_names': norm_names, 'idx': idx, 'norm_idx': norm_idx, 'fnames': ','.join(f_names), 'plchs': ','.join(plchs), 'fset': ','.join(f_set), 'knames': ','.join(f_idx), 'fidx': ' AND '.join(f_idx_match), 'lookup_fallbacks': iclass.lookup_fallbacks, } @publish def add_nl_source(self, target, kind, spec): ''' A temprorary method, to be moved out ''' # flush self.execute( ''' DELETE FROM sources_options WHERE f_target = %s ''' % self.plch, (target,), ) self.execute( ''' DELETE FROM sources WHERE f_target = %s ''' % self.plch, (target,), ) # add self.execute( ''' INSERT INTO sources (f_target, f_kind) VALUES (%s, %s) ''' % (self.plch, self.plch), (target, kind), ) for key, value in spec.items(): vtype = 'int' if isinstance(value, int) else 'str' self.execute( ''' INSERT INTO sources_options (f_target, f_name, f_type, f_value) VALUES (%s, %s, %s, %s) ''' % (self.plch, self.plch, self.plch, self.plch), (target, key, vtype, value), ) def execute(self, argv, kwarg): try: # # FIXME: add logging # for _ in range(MAX_ATTEMPTS): try: self.cursor.execute(argv, *kwarg) break except (sqlite3.InterfaceError, sqlite3.OperationalError) as e: self.log.debug('%s' % e) # # Retry on: # -- InterfaceError: Error binding parameter ... # -- OperationalError: SQL logic error # pass else: raise Exception('DB execute error: %s %s' % (argv, kwarg)) except Exception: raise finally: self.connection.commit() # no performance optimisation yet return self.cursor @publish def fetchone(self, argv, *kwarg): for row in self.fetch(argv, *kwarg): return row return None @publish def fetch(self, argv, *kwarg): self.execute(argv, *kwarg) while True: row_set = self.cursor.fetchmany() if not row_set: return for row in row_set: yield row @publish def backup(self, spec): if self.provider == DBProvider.sqlite3: backup_connection = sqlite3.connect(spec) self.connection.backup(backup_connection) backup_connection.close() else: raise NotImplementedError() @publish def export(self, f='stdout'): close = False if f in ('stdout', 'stderr'): f = getattr(sys, f) elif isinstance(f, basestring): f = open(f, 'w') close = True try: for table in self.spec.keys(): f.write('\ntable %s\n' % table) for record in self.execute('SELECT FROM %s' % table): f.write(' '.join([str(x) for x in record])) f.write('\n') if self.rtnl_log: f.write('\ntable %s_log\n' % table) for record in self.execute('SELECT * FROM %s_log' % table): f.write(' '.join([str(x) for x in record])) f.write('\n') finally: if close: f.close() def close(self): if self.config['spec'] != ':memory:': # simply discard in-memory sqlite db on exit self.purge_snapshots() self.connection.commit() self.connection.close() @publish def commit(self): self.connection.commit() def create_table(self, table): req = ['f_target TEXT NOT NULL', 'f_tflags BIGINT NOT NULL DEFAULT 0'] fields = [] self.key_defaults[table] = {} for field in self.spec[table].items(): # # Why f_? # 'Cause there are attributes like 'index' and such # names may not be used in SQL statements # field = (field[0][-1], field[1]) fields.append('f_%s %s' % field) req.append('f_%s %s' % field) if field[1].strip().startswith('TEXT'): self.key_defaults[table][field[0]] = '' else: self.key_defaults[table][field[0]] = 0 if table in self.foreign_keys: for key in self.foreign_keys[table]: spec = ( '(%s)' % ','.join(key['fields']), '%s(%s)' % (key['parent'], ','.join(key['parent_fields'])), ) req.append( 'FOREIGN KEY %s REFERENCES %s ' 'ON UPDATE CASCADE ' 'ON DELETE CASCADE ' % spec ) # # make a unique index for compound keys on # the parent table # # https://sqlite.org/foreignkeys.html # if len(key['fields']) > 1: idxname = 'uidx_%s_%s' % ( key['parent'], '_'.join(key['parent_fields']), ) self.execute( 'CREATE UNIQUE INDEX ' 'IF NOT EXISTS %s ON %s' % (idxname, spec[1]) ) req = ','.join(req) req = 'CREATE TABLE IF NOT EXISTS ' '%s (%s)' % (table, req) self.execute(req) index = ','.join( ['f_target', 'f_tflags'] + ['f_%s' % x for x in self.indices[table]] ) req = 'CREATE UNIQUE INDEX IF NOT EXISTS ' '%s_idx ON %s (%s)' % ( table, table, index, ) self.execute(req) # # create table for the transaction buffer: there go the system # updates while the transaction is not committed. # # w/o keys (yet) # # req = ['f_target TEXT NOT NULL', # 'f_tflags INTEGER NOT NULL DEFAULT 0'] # req = ','.join(req) # self.execute('CREATE TABLE IF NOT EXISTS ' # '%s_buffer (%s)' % (table, req)) # # create the log table, if required # if self.rtnl_log: req = [ 'f_tstamp BIGINT NOT NULL', 'f_target TEXT NOT NULL', 'f_event INTEGER NOT NULL', ] + fields req = ','.join(req) self.execute( 'CREATE TABLE IF NOT EXISTS ' '%s_log (%s)' % (table, req) ) def mark(self, target, mark): for table in self.spec: self.execute( ''' UPDATE %s SET f_tflags = %s WHERE f_target = %s ''' % (table, self.plch, self.plch), (mark, target), ) @publish def flush(self, target): for table in self.spec: self.execute( ''' DELETE FROM %s WHERE f_target = %s ''' % (table, self.plch), (target,), ) @publish def save_deps(self, ctxid, weak_ref, iclass): uuid = uuid32() obj = weak_ref() obj_k = obj.key idx = self.indices[obj.table] conditions = [] values = [] for key in idx: conditions.append('f_%s = %s' % (key, self.plch)) if key in obj_k: values.append(obj_k[key]) else: values.append(obj.get(iclass.nla2name(key))) # # save the old f_tflags value # tflags = self.execute( ''' SELECT f_tflags FROM %s WHERE %s ''' % (obj.table, ' AND '.join(conditions)), values, ).fetchone()[0] # # mark tflags for obj # obj.mark_tflags(uuid) # # f_tflags is used in foreign keys ON UPDATE CASCADE, so all # related records will be marked # for table in self.spec: self.log.debug('create snapshot %s_%s' % (table, ctxid)) # # create the snapshot table # self.execute( ''' CREATE TABLE IF NOT EXISTS %s_%s AS SELECT * FROM %s WHERE f_tflags IS NULL ''' % (table, ctxid, table) ) # # copy the data -- is it possible to do it in one step? # self.execute( ''' INSERT INTO %s_%s SELECT * FROM %s WHERE f_tflags = %s ''' % (table, ctxid, table, self.plch), [uuid], ) # # unmark all the data # obj.mark_tflags(tflags) for table in self.spec: self.execute( ''' UPDATE %s_%s SET f_tflags = %s ''' % (table, ctxid, self.plch), [tflags], ) self.snapshots['%s_%s' % (table, ctxid)] = weak_ref @publish def purge_snapshots(self): for table in tuple(self.snapshots): for _ in range(MAX_ATTEMPTS): try: if self.provider == DBProvider.sqlite3: self.execute('DROP TABLE %s' % table) elif self.provider == DBProvider.psycopg2: self.execute('DROP TABLE %s CASCADE' % table) self.connection.commit() del self.snapshots[table] break except sqlite3.OperationalError: # # Retry on: # -- OperationalError: database table is locked # time.sleep(random.random()) else: raise Exception('DB snapshot error') @publish def get(self, table, spec): # # Retrieve info from the DB # # ndb.interfaces.get({'ifname': 'eth0'}) # conditions = [] values = [] cls = self.classes[table] cspec = self.compiled[table] for key, value in spec.items(): if key not in cspec['all_names']: key = cls.name2nla(key) if key not in cspec['all_names']: raise KeyError('field name not found') conditions.append('f_%s = %s' % (key, self.plch)) values.append(value) req = 'SELECT * FROM %s WHERE %s' % (table, ' AND '.join(conditions)) for record in self.fetch(req, values): yield dict(zip(self.compiled[table]['all_names'], record)) def log_netlink(self, table, target, event, ctable=None): # # RTNL Logs # fkeys = self.compiled[table]['names'] fields = ','.join( ['f_tstamp', 'f_target', 'f_event'] + ['f_%s' % x for x in fkeys] ) pch = ','.join([self.plch] * (len(fkeys) + 3)) values = [ int(time.time() * 1000), target, event.get('header', {}).get('type', 0), ] for field in fkeys: value = event.get_attr(field) or event.get(field) if value is None and field in self.indices[ctable or table]: value = self.key_defaults[table][field] if isinstance(value, (dict, list, tuple, set)): value = json.dumps(value) values.append(value) self.execute( 'INSERT INTO %s_log (%s) VALUES (%s)' % (table, fields, pch), values, ) def load_netlink(self, table, target, event, ctable=None, propagate=False): # if self.rtnl_log: self.log_netlink(table, target, event, ctable) # # Update metrics # if 'stats' in event['header']: self.stats[target] = event['header']['stats'] # # Periodic jobs # if time.time() - self.gctime > config.gc_timeout: self.gctime = time.time() # clean dead snapshots after GC timeout for name, wref in tuple(self.snapshots.items()): if wref() is None: del self.snapshots[name] try: self.execute('DROP TABLE %s' % name) except Exception as e: self.log.debug( 'failed to remove table %s: %s' % (name, e) ) # clean marked routes self.execute( 'DELETE FROM routes WHERE ' '(f_gc_mark + 5) < %s' % self.plch, (int(time.time()),), ) # # The event type # if event['header'].get('type', 0) % 2: # # Delete an object # conditions = ['f_target = %s' % self.plch] values = [target] for key in self.indices[table]: conditions.append('f_%s = %s' % (key, self.plch)) value = event.get(key) or event.get_attr(key) if value is None: value = self.key_defaults[table][key] if isinstance(value, (dict, list, tuple, set)): value = json.dumps(value) values.append(value) self.execute( 'DELETE FROM %s WHERE' ' %s' % (table, ' AND '.join(conditions)), values, ) else: # # Create or set an object # # field values values = [target, 0] # index values ivalues = [target, 0] compiled = self.compiled[table] # a map of sub-NLAs nodes = {} # replace r_conditions = [] r_values = [] # Check route replace if ( event['header'].get('flags', 0) == NLM_F_REPLACE and event['event'] == 'RTM_NEWROUTE' ): # Replace existing route r_conditions = [table + '.f_target = %s' % self.plch] r_values = [target] for key in self.indices[table]: if key not in [ 'RTA_DST', 'dst_len', 'table', 'RTA_PRIORITY', ]: continue r_conditions.append( table + '.f_%s = %s' % (key, self.plch) ) value = event.get(key) or event.get_attr(key) if value is None: value = self.key_defaults[table][key] if isinstance(value, (dict, list, tuple, set)): value = json.dumps(value) r_values.append(value) # fetch values (exc. the first two columns) for fname, ftype in self.spec[table].items(): node = event # if the field is located in a sub-NLA if len(fname) > 1: # see if we tried to get it already if fname[:-1] not in nodes: # descend for steg in fname[:-1]: node = node.get_attr(steg) if node is None: break nodes[fname[:-1]] = node # lookup the sub-NLA in the map node = nodes[fname[:-1]] # the event has no such sub-NLA if node is None: values.append(None) continue # NLA have priority value = node.get_attr(fname[-1]) if value is None: value = node.get(fname[-1]) if value is None and fname[-1] in self.compiled[table]['idx']: value = self.key_defaults[table][fname[-1]] node['attrs'].append((fname[-1], value)) if isinstance(value, (dict, list, tuple, set)): value = json.dumps(value) if fname[-1] in compiled['idx']: ivalues.append(value) values.append(value) try: w_fidx = compiled['fidx'] w_ivalues = ivalues if r_conditions: w_fidx = ' AND '.join(r_conditions) w_ivalues = r_values if self.provider == DBProvider.psycopg2: # # run UPSERT -- the DB provider must support it # ( self.execute( 'INSERT INTO %s (%s) VALUES (%s) ' 'ON CONFLICT (%s) ' 'DO UPDATE SET %s WHERE %s' % ( table, compiled['fnames'], compiled['plchs'], compiled['knames'], compiled['fset'], w_fidx, ), (values + values + w_ivalues), ) ) # elif self.provider == DBProvider.sqlite3: # # SQLite3 >= 3.24 actually has UPSERT, but ... # # We can not use here INSERT OR REPLACE as well, since # it drops (almost always) records with foreign key # dependencies. Maybe a bug in SQLite3, who knows. # count = ( self.execute( ''' SELECT count(*) FROM %s WHERE %s ''' % (table, w_fidx), w_ivalues, ).fetchone() )[0] if count == 0: self.execute( ''' INSERT INTO %s (%s) VALUES (%s) ''' % (table, compiled['fnames'], compiled['plchs']), values, ) else: self.execute( ''' UPDATE %s SET %s WHERE %s ''' % (table, compiled['fset'], w_fidx), (values + w_ivalues), ) else: raise NotImplementedError() # except Exception as e: # if propagate: raise e # # A good question, what should we do here self.log.debug( 'load_netlink: %s %s %s' % (table, target, event) ) self.log.error('load_netlink: %s' % traceback.format_exc())	33.37 KB	2025-05-25 12:26
📄 source.py ''' Local RTNL ---------- Local RTNL source is a simple `IPRoute` instance. By default NDB starts with one local RTNL source names `localhost`:: >>> ndb = NDB() >>> ndb.sources.summary().format("json") [ { "name": "localhost", "spec": "{'target': 'localhost', 'nlm_generator': 1}", "state": "running" }, { "name": "localhost/nsmanager", "spec": "{'target': 'localhost/nsmanager'}", "state": "running" } ] >>> ndb.sources['localhost'] [running] <IPRoute {'target: 'localhost', 'nlm_generator': 1}> The `localhost` RTNL source starts an additional async cache thread. The `nlm_generator` option means that instead of collections the `IPRoute` object returns generators, so `IPRoute` responses will not consume memory regardless of the RTNL objects number:: >>> ndb.sources['localhost'].nl.link('dump') <generator object RTNL_API.filter_messages at 0x7f61a99a34a0> See also: :ref:`iproute` Network namespaces ------------------ There are two ways to connect additional sources to an NDB instance. One is to specify sources when creating an NDB object:: ndb = NDB(sources=[{'target': 'localhost'}, {'netns': 'test01'}]) Another way is to call `ndb.sources.add()` method:: ndb.sources.add(netns='test01') This syntax: `{target': 'localhost'}` and `{'netns': 'test01'}` is the short form. The full form would be:: {'target': 'localhost', # the label for the DB 'kind': 'local', # use IPRoute class to start the source 'nlm_generator': 1} # {'target': 'test01', # the label 'kind': 'netns', # use NetNS class 'netns': 'test01'} # See also: :ref:`netns` ''' import errno import importlib import queue import socket import struct import sys import threading import time import uuid from pyroute2.common import basestring from pyroute2.iproute import IPRoute from pyroute2.netlink.exceptions import NetlinkError from pyroute2.netlink.nlsocket import NetlinkSocket from pyroute2.netlink.rtnl.ifinfmsg import ifinfmsg from .events import ShutdownException, State from .messages import cmsg_event, cmsg_failed, cmsg_sstart if sys.platform.startswith('linux'): from pyroute2 import netns from pyroute2.iproute.linux import NetNS from pyroute2.netns.manager import NetNSManager else: NetNS = None NetNSManager = None SOURCE_FAIL_PAUSE = 1 SOURCE_MAX_ERRORS = 3 class SourceProxy(object): def __init__(self, ndb, target): self.ndb = ndb self.events = queue.Queue() self.target = target def api(self, name, argv, kwarg): call_id = str(uuid.uuid4().hex) self.ndb._call_registry[call_id] = event = threading.Event() event.clear() ( self.ndb.messenger.emit( { 'type': 'api', 'target': self.target, 'call_id': call_id, 'name': name, 'argv': argv, 'kwarg': kwarg, } ) ) event.wait() response = self.ndb._call_registry.pop(call_id) if 'return' in response: return response['return'] elif 'exception' in response: raise response['exception'] class Source(dict): ''' The RNTL source. The source that is used to init the object must comply to IPRoute API, must support the async_cache. If the source starts additional threads, they must be joined in the source.close() ''' table_alias = 'src' dump_header = None summary_header = None view = None table = 'sources' vmap = {'local': IPRoute, 'netns': NetNS, 'nsmanager': NetNSManager} def __init__(self, ndb, spec): self.th = None self.nl = None self.ndb = ndb self.evq = self.ndb._event_queue # the target id -- just in case self.target = spec['target'] self.kind = spec.pop('kind', 'local') self.max_errors = spec.pop('max_errors', SOURCE_MAX_ERRORS) self.event = spec.pop('event') # RTNL API self.nl_prime = self.get_prime(self.kind) self.nl_kwarg = spec # if self.ndb.messenger is not None: self.ndb.messenger.targets.add(self.target) # self.errors_counter = 0 self.shutdown = threading.Event() self.started = threading.Event() self.lock = threading.RLock() self.shutdown_lock = threading.RLock() self.started.clear() self.log = ndb.log.channel('sources.%s' % self.target) self.state = State(log=self.log, wait_list=['running']) self.state.set('init') self.ndb.task_manager.db_add_nl_source(self.target, self.kind, spec) self.load_sql() @classmethod def _count(cls, view): return view.ndb.task_manager.db_fetchone( "SELECT count() FROM %s" % view.table ) @property def must_restart(self): if self.max_errors < 0 or self.errors_counter <= self.max_errors: return True return False @property def bind_arguments(self): return dict( filter( lambda x: x[1] is not None, ( ('async_cache', True), ('clone_socket', True), ('groups', self.nl_kwarg.get('groups')), ), ) ) def set_ready(self): try: if self.event is not None: self.evq.put( (cmsg_event(self.target, self.event),), source=self.target ) else: self.evq.put((cmsg_sstart(self.target),), source=self.target) except ShutdownException: self.state.set('stop') return False return True @classmethod def defaults(cls, spec): ret = dict(spec) defaults = {} if 'hostname' in spec: defaults['kind'] = 'remote' defaults['protocol'] = 'ssh' defaults['target'] = spec['hostname'] if 'netns' in spec: defaults['kind'] = 'netns' defaults['target'] = spec['netns'] ret['netns'] = netns._get_netnspath(spec['netns']) for key in defaults: if key not in ret: ret[key] = defaults[key] return ret def __repr__(self): if isinstance(self.nl_prime, NetlinkSocket): name = self.nl_prime.__class__.__name__ elif isinstance(self.nl_prime, type): name = self.nl_prime.__name__ return '[%s] <%s %s>' % (self.state.get(), name, self.nl_kwarg) @classmethod def nla2name(cls, name): return name @classmethod def name2nla(cls, name): return name @classmethod def summary(cls, view): yield ('state', 'name', 'spec') for key in view.keys(): yield (view[key].state.get(), key, '%s' % (view[key].nl_kwarg,)) @classmethod def dump(cls, view): return cls.summary(view) @classmethod def compare_record(self, left, right): # specific compare if isinstance(right, basestring): return right == left['name'] def get_prime(self, name): return self.vmap.get(self.kind, None) or getattr( importlib.import_module('pyroute2'), self.kind ) def api(self, name, argv, kwarg): for _ in range(100): # FIXME make a constant with self.lock: try: self.log.debug(f'source api run {name} {argv} {kwarg}') return getattr(self.nl, name)(argv, kwarg) except ( NetlinkError, AttributeError, ValueError, KeyError, TypeError, socket.error, struct.error, ): raise except Exception as e: # probably the source is restarting self.errors_counter += 1 self.log.debug(f'source api error: <{e}>') time.sleep(1) raise RuntimeError('api call failed') def fake_zero_if(self): url = 'https://github.com/svinota/pyroute2/issues/737' zero_if = ifinfmsg() zero_if['index'] = 0 zero_if['state'] = 'up' zero_if['flags'] = 1 zero_if['family'] = 0 zero_if['header']['flags'] = 2 zero_if['header']['type'] = 16 zero_if['header']['target'] = self.target zero_if['event'] = 'RTM_NEWLINK' zero_if['attrs'] = [ ('IFLA_IFNAME', url), ('IFLA_ADDRESS', '00:00:00:00:00:00'), ] zero_if.encode() self.evq.put([zero_if], source=self.target) def receiver(self): # # The source thread routine -- get events from the # channel and forward them into the common event queue # # The routine exists on an event with error code == 104 # while self.state.get() != 'stop': if self.shutdown.is_set(): break with self.lock: if self.nl is not None: try: self.nl.close(code=0) except Exception as e: self.log.warning('source restart: %s' % e) try: self.state.set('connecting') if isinstance(self.nl_prime, type): spec = {} spec.update(self.nl_kwarg) if self.kind in ('nsmanager',): spec['libc'] = self.ndb.libc self.nl = self.nl_prime(spec) else: raise TypeError('source channel not supported') self.state.set('loading') # self.nl.bind(*self.bind_arguments) # # Initial load -- enqueue the data # try: self.ndb.task_manager.db_flush(self.target) if self.kind in ('local', 'netns', 'remote'): self.fake_zero_if() self.evq.put(tuple(self.nl.dump()), source=self.target) finally: pass self.errors_counter = 0 except Exception as e: self.errors_counter += 1 self.started.set() self.state.set(f'failed, counter {self.errors_counter}') self.log.error(f'source error: {type(e)} {e}') try: self.evq.put( (cmsg_failed(self.target),), source=self.target ) except ShutdownException: self.state.set('stop') break if self.must_restart: self.log.debug('sleeping before restart') self.state.set('restart') self.shutdown.wait(SOURCE_FAIL_PAUSE) if self.shutdown.is_set(): self.log.debug('source shutdown') self.state.set('stop') break else: return self.set_ready() continue with self.lock: if self.state.get() == 'loading': if not self.set_ready(): break self.started.set() self.shutdown.clear() self.state.set('running') while self.state.get() not in ('stop', 'restart'): try: msg = tuple(self.nl.get()) self.log.debug(f'received message {msg}') except Exception as e: self.errors_counter += 1 self.log.error('source error: %s %s' % (type(e), e)) msg = None if self.must_restart: self.state.set('restart') else: self.state.set('stop') break code = 0 if msg and msg[0]['header']['error']: code = msg[0]['header']['error'].code if not msg or code == errno.ECONNRESET: self.state.set('stop') break try: self.evq.put(msg, source=self.target) except ShutdownException: self.state.set('stop') break # thus we make sure that all the events from # this source are consumed by the main loop # in __dbm__() routine try: self.sync() self.log.debug('flush DB for the target') self.ndb.task_manager.db_flush(self.target) except ShutdownException: self.log.debug('shutdown handled by the main thread') pass self.state.set('stopped') def sync(self): self.log.debug('sync') sync = threading.Event() self.evq.put((cmsg_event(self.target, sync),), source=self.target) sync.wait() def start(self): # # Start source thread with self.lock: self.log.debug('starting the source') if (self.th is not None) and self.th.is_alive(): raise RuntimeError('source is running') self.th = threading.Thread( target=self.receiver, name='NDB event source: %s' % (self.target), ) self.th.start() return self def close(self, code=errno.ECONNRESET, sync=True): with self.shutdown_lock: if self.shutdown.is_set(): self.log.debug('already stopped') return self.log.debug('source shutdown') self.shutdown.set() if self.nl is not None: try: self.nl.close(code=code) except Exception as e: self.log.error('source close: %s' % e) if sync: if self.th is not None: self.th.join() self.th = None else: self.log.debug('receiver thread missing') def restart(self, reason='unknown'): with self.lock: with self.shutdown_lock: self.log.debug('restarting the source, reason <%s>' % (reason)) self.started.clear() try: self.close() if self.th: self.th.join() self.shutdown.clear() self.start() finally: pass self.started.wait() def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): self.close() def load_sql(self): # spec = self.ndb.task_manager.db_fetchone( ''' SELECT FROM sources WHERE f_target = %s ''' % self.ndb.schema.plch, (self.target,), ) self['target'], self['kind'] = spec for spec in self.ndb.task_manager.db_fetch( ''' SELECT * FROM sources_options WHERE f_target = %s ''' % self.ndb.schema.plch, (self.target,), ): f_target, f_name, f_type, f_value = spec self[f_name] = int(f_value) if f_type == 'int' else f_value	15.91 KB	2025-05-25 12:26
📄 task_manager.py import inspect import logging import queue import threading import time import traceback from functools import partial from pyroute2 import config from . import schema from .events import ( DBMExitException, InvalidateHandlerException, RescheduleException, ShutdownException, ) from .messages import cmsg, cmsg_event, cmsg_failed, cmsg_sstart log = logging.getLogger(__name__) def Events(argv): for sequence in argv: if sequence is not None: for item in sequence: yield item class NDBConfig(dict): def __init__(self, task_manager): self.task_manager = task_manager def __getitem__(self, key): return self.task_manager.config_get(key) def __setitem__(self, key, value): return self.task_manager.config_set(key, value) def __delitem__(self, key): return self.task_manager.config_del(key) def keys(self): return self.task_manager.config_keys() def items(self): return self.task_manager.config_items() def values(self): return self.task_manager.config_values() class TaskManager: def __init__(self, ndb): self.ndb = ndb self.log = ndb.log self.event_map = {} self.event_queue = ndb._event_queue self.thread = None self.ctime = self.gctime = time.time() def register_handler(self, event, handler): if event not in self.event_map: self.event_map[event] = [] self.event_map[event].append(handler) def unregister_handler(self, event, handler): self.event_map[event].remove(handler) def default_handler(self, target, event): if isinstance(getattr(event, 'payload', None), Exception): raise event.payload log.debug('unsupported event ignored: %s' % type(event)) def check_sources_started(self, _locals, target, event): _locals['countdown'] -= 1 if _locals['countdown'] == 0: self.ndb._dbm_ready.set() def wrap_method(self, method): # # this wrapper will be published in the DBM thread # def _do_local_generator(target, request): try: for item in method(request.argv, *request.kwarg): request.response.put(item) request.response.put(StopIteration()) except Exception as e: request.response.put(e) def _do_local_single(target, request): try: (request.response.put(method(request.argv, *request.kwarg))) except Exception as e: (request.response.put(e)) # # this class will be used to map the requests # class cmsg_req(cmsg): def __init__(self, response, argv, *kwarg): self['header'] = {'target': None} self.response = response self.argv = argv self.kwarg = kwarg # # this method will proxy the original one # def _do_dispatch_generator(self, argv, *kwarg): if self.thread == id(threading.current_thread()): # same thread, run method locally for item in method(argv, *kwarg): yield item else: # another thread, run via message bus response = queue.Queue() request = cmsg_req(response, argv, *kwarg) self.event_queue.put((request,)) while True: item = response.get() if isinstance(item, StopIteration): return elif isinstance(item, Exception): raise item else: yield item def _do_dispatch_single(self, argv, *kwarg): if self.thread == id(threading.current_thread()): # same thread, run method locally return method(argv, *kwarg) else: # another thread, run via message bus response = queue.Queue(maxsize=1) request = cmsg_req(response, argv, kwarg) self.event_queue.put((request,)) ret = response.get() if isinstance(ret, Exception): raise ret else: return ret # # return the method spec to be announced # handler = _do_local_single proxy = _do_dispatch_single if inspect.isgeneratorfunction(method): handler = _do_local_generator proxy = _do_dispatch_generator return (cmsg_req, handler, proxy) def register_api(self, api_obj, prefix=''): for name in dir(api_obj): method = getattr(api_obj, name, None) if hasattr(method, 'publish'): if isinstance(method.publish, str): name = method.publish name = f'{prefix}{name}' event, handler, proxy = self.wrap_method(method) setattr(self, name, partial(proxy, self)) self.event_map[event] = [handler] def run(self): _locals = {'countdown': len(self.ndb._nl)} self.thread = id(threading.current_thread()) # init the events map event_map = { cmsg_event: [lambda t, x: x.payload.set()], cmsg_failed: [lambda t, x: (self.ndb.schema.mark(t, 1))], cmsg_sstart: [partial(self.check_sources_started, _locals)], } self.event_map = event_map try: self.ndb.schema = schema.DBSchema( self.ndb.config, self.ndb.sources, self.event_map, self.log.channel('schema'), ) self.register_api(self.ndb.schema, 'db_') self.register_api(self.ndb.schema.config, 'config_') self.ndb.config = NDBConfig(self) except Exception as e: self.ndb._dbm_error = e self.ndb._dbm_ready.set() return for spec in self.ndb._nl: spec['event'] = None self.ndb.sources.add(spec) for event, handlers in self.ndb.schema.event_map.items(): for handler in handlers: self.register_handler(event, handler) stop = False source = None reschedule = [] while not stop: source, events = self.event_queue.get() events = Events(events, reschedule) reschedule = [] try: for event in events: handlers = event_map.get( event.__class__, [self.default_handler] ) for handler in tuple(handlers): try: target = event['header']['target'] handler(target, event) except RescheduleException: if 'rcounter' not in event['header']: event['header']['rcounter'] = 0 if event['header']['rcounter'] < 3: event['header']['rcounter'] += 1 self.log.debug('reschedule %s' % (event,)) reschedule.append(event) else: self.log.error('drop %s' % (event,)) except InvalidateHandlerException: try: handlers.remove(handler) except Exception: self.log.error( 'could not invalidate ' 'event handler:\n%s' % traceback.format_exc() ) except ShutdownException: stop = True break except DBMExitException: return except Exception: self.log.error( 'could not load event:\n%s\n%s' % (event, traceback.format_exc()) ) if time.time() - self.gctime > config.gc_timeout: self.gctime = time.time() except Exception as e: self.log.error(f'exception <{e}> in source {source}') # restart the target try: self.log.debug(f'requesting source {source} restart') self.ndb.sources[source].state.set('restart') except KeyError: self.log.debug(f'key error for {source}') pass # release all the sources for target in tuple(self.ndb.sources.cache): source = self.ndb.sources.remove(target, sync=False) if source is not None and source.th is not None: self.log.debug(f'closing source {source}') source.close() if self.ndb.schema.config['db_cleanup']: self.log.debug('flush DB for the target %s' % target) self.ndb.schema.flush(target) else: self.log.debug('leave DB for debug') # close the database self.ndb.schema.commit() self.ndb.schema.close() # close the logging for handler in self.log.logger.handlers: handler.close()	9.58 KB	2025-05-25 12:26
📄 transaction.py ''' One object ---------- All the changes done using one object are applied in the order defined by the corresponding object class. .. code-block:: python eth0 = ndb.interfaces["eth0"] eth0.add_ip(address="10.0.0.1", prefixlen=24) eth0.set(state="up") eth0.set(mtu=1400) eth0.commit() In the example above first the interface attributes like state, mtu, ifname etc. will be applied, and only then IP addresses, bridge ports and like that, regardless the order they are referenced before the `commit()` call. The order is ok for most of cases. But if not, one can control it by calling `commit()` in the required places, breaking one transaction into several sequential transactions. And since RTNL object methods return the object itself, it is possible to write chains with multiple `commit()`: .. code-block:: python ( ndb.interfaces .create(ifname="test", kind="dummy") .add_ip(address="10.0.0.1", prefixlen=24) .commit() .set(state="up") .commit() ) Here the order is forced by explicit commits. Multiple objects ---------------- An important functionality of NDB are rollbacks. And there is a way to batch changes on multiple objects so one failure will trigger rollback of all the changes on all the objects. .. code-block:: python ctx = ndb.begin() ctx.push( # first set up a bridge ( ndb.interfaces .create(ifname="br0", kind="bridge") .add_port("eth0") .add_port("eth1") .set(state="up") .add_ip("10.0.0.2/24") ), # and only then create a route ( ndb.routes .create( dst="192.168.0.0", dst_len=24, gateway="10.0.0.1" ) ) ) ctx.commit() # if something goes wrong, the whole batch will be reverted Ping a remote host ------------------ The simplest usecase for external checks is to test if a remote IP is still reachable after the changes are applied: .. code-block:: python from pyroute2.ndb.transaction import PingAddress ctx = ndb.begin() ctx.push( ndb.routes.create(dst="10.0.0.0", dst_len=24, gateway="172.16.0.1"), PingAddress("10.0.0.1") ) ctx.commit() # the route will be removed if ping fails Or on the contrary, don't run transaction if a remote IP is reachable: .. code-block:: python from pyroute2.ndb.transaction import Not, PingAddress ctx = ndb.begin() ctx.push( Not(PingAddress("10.0.0.1")), ndb.routes.create(dst="10.0.0.0", dst_len=24, gateway="172.16.0.1") ) try: ctx.commit() except CheckProcessException: pass In this example, the route will be added only if `10.0.0.1` is not reachable. The default ping timeout is set to 1, but it is possible to customize it: .. code-block:: python PingAddress("10.0.0.1", timeout=10) Check an external processes --------------------------- A more generic type of check is CheckProcess: .. code-block:: python from pyroute2.ndb.transaction import CheckProcess with ndb.begin() as ctx: ctx.push(ndb.routes.create( dst="10.0.0.0", dst_len=24, gateway="172.16.0.1" )) ctx.push(CheckProcess('/path/to/script.sh')) # # --> <-- the route will be removed if the script fails `CheckProcess` is `subprocess.Popen` based, is not a shell call, thus no pipes or other shell syntax are allowed. `CheckProcess` also accepts `timeout` argument: .. code-block:: python CheckProcess('/path/to/script.sh', timeout=10).commit() If the subprocess doens't finish within the timeout, it will be terminated with SIGTERM. SIGKILL is not used. Logging and debug ----------------- `CheckProcess` and `PingAddress` accept log as an argument: .. code-block:: python PingAddress("10.0.0.1", log=ndb.log.channel("util")).commit() CheckProcess("/path/to/script.sh", log=ndb.log.channel("util")).commit() The check objects are thread safe and reusable, it is possible to run `commit()` on them multiple times. The subprocess' stdout and stderr will be both logged and saved: .. code-block:: python check = CheckProcess("/path/to/script.sh") while True: check.commit() # periodic check, the loop breaks on failure print(f'stdout: {check.out}') print(f'stderr: {check.err}') print(f'return code: {check.return_code}') time.sleep(10) Check negation -------------- It is possible to negate the check for `CheckProcess` and child classes .. code-block:: python from pyroute2.ndb.transaction import Not, CheckProcess check = Not(CheckProcess('/path/to/script.sh')) check.commit() API --- ''' import logging import shlex import shutil import subprocess import threading global_log = logging.getLogger(__name__) class CheckProcessException(Exception): pass class CheckProcess: ''' Run an external process on `commit()` and raise `CheckProcessException` if the return code is not 0. Objects of this class are thread safe and reusable. ''' def __init__(self, command, log=None, timeout=None): if not isinstance(command, str): raise TypeError('command must be a non empty string') if not len(command) > 0: raise TypeError('command must be a non empty string') self.log = log or global_log self.command = command self.args = shlex.split(command) self.timeout = timeout self.return_code = None self.out = None self.err = None self.lock = threading.Lock() def commit(self): with self.lock: self.args[0] = shutil.which(self.args[0]) if self.args[0] is None: raise FileNotFoundError() process = subprocess.Popen( self.args, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) try: self.log.debug(f'process check {self.args}') self.out, self.err = process.communicate(timeout=self.timeout) self.log.debug(f'process output: {self.out}') self.log.debug(f'process stderr: {self.err}') except subprocess.TimeoutExpired: self.log.debug('process timeout expired') process.terminate() process.stdout.close() process.stderr.close() finally: self.return_code = process.wait() if self.return_code != 0: raise CheckProcessException('CheckProcess failed') def rollback(self): pass def __repr__(self): return f'[{self.command}]' class PingAddress(CheckProcess): def __init__(self, address, log=None, timeout=1): super(PingAddress, self).__init__( f'ping -c 1 -W {timeout} {address}', log=log ) class Not: ''' Negate the `CheckProcess` results. If `CheckProcess.commit()` succeeds, raise CheckProcessException, and vice versa, if `CheckProcess.commit()` fails, return success. ''' def __init__(self, transaction): self.tx = transaction def commit(self): success = True try: self.tx.commit() except Exception: success = False if success: raise CheckProcessException(f'{self.tx} succeeded') def rollback(self): pass class Transaction: ''' `Transaction` class is an independent utility class. Being designed to be used with NDB object transactions, it may be used with any object implementing commit/rollback protocol, see `commit()` method. The class supports the context manager protocol and `Transaction` objects may be used in `with` statements: .. code-block:: python with Transaction() as tx: tx.push(obj0) # enqueue objects tx.push(obj1) # --> <-- run commit() for every object in self.queue # # if any commit() fails, run rollback() for every # executed commit() in the reverse order NDB provides a utility method to create `Transaction` objects: .. code-block:: python with ndb.begin() as tx: tx.push(ndb.interfaces["eth0"].set(state="up")) tx.push(ndb.interfaces["eth1"].set(state="up")) ''' def __init__(self, log=None): self.queue = [] self.event = threading.Event() self.event.clear() self.log = global_log or log self.log.debug('begin transaction') def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): if exc_type is None: self.commit() def push(self, argv): ''' Push objects to the transaction queue. One may use any number of positional arguments: .. code-block:: python tx.push(obj0) tx.push(obj0, obj1, obj2) tx.push([obj0, obj1, obj2]) ''' for obj in argv: self.log.debug('queue %s' % type(obj)) self.queue.append(obj) return self def append(self, obj): ''' Append one object to the queue. ''' self.log.debug('queue %s' % type(obj)) self.push(obj) return self def pop(self, index=-1): ''' Pop an object from the queue. If an index is not specified, pop the last (the rightmost) object. ''' self.log.debug('pop %s' % index) self.queue.pop(index) return self def insert(self, index, obj): ''' Insert an object into the queue. The position index is required. ''' self.log.debug('insert %i %s' % (index, type(obj))) self.queue.insert(index, obj) return self def cancel(self): ''' Cancel the transaction and empty the queue. ''' self.log.debug('cancel transaction') self.queue = [] return self def wait(self, timeout=None): ''' Wait until the transaction to be successfully committed. ''' return self.event.wait(timeout) def done(self): ''' Check if the done event is set. ''' return self.event.is_set() def commit(self): ''' Execute `commit()` for every queued object. If an execution fails, execute `rollback()` for every executed commit. All objects in the queue that follows the failed one will remain intact. Raises the original exception of the failed `commit()`. All the `rollback()` exceptions are ignored. ''' self.log.debug('commit') rollbacks = [] for obj in self.queue: rollbacks.append(obj) try: obj.commit() except Exception: for rb in reversed(rollbacks): try: rb.rollback() except Exception as e: self.log.warning('ignore rollback exception: %s' % e) raise self.event.set() return self	11.05 KB	2025-05-25 12:26
📄 transport.py import pickle import select import socket import struct import time import uuid class IdCache(dict): def invalidate(self): current_time = time.time() collect_time = current_time - 60 for mid, meta in tuple(self.items()): if meta < collect_time: self.pop(mid) def __setitem__(self, key, value): if len(self) > 100: self.invalidate() dict.__setitem__(self, key, value) class Peer(object): def __init__(self, remote_id, local_id, address, port, cache): self.address = address self.port = port self.socket = None self.remote_id = remote_id self.local_id = local_id self.cache = cache self.version = 0 self.last_exception_time = 0 @property def connected(self): return self.socket is not None def __repr__(self): if self.connected: connected = 'not connected' else: connected = 'connected' return '[%s-%s] %s:%s [%s]' % ( self.local_id, self.remote_id, self.address, self.port, connected, ) def hello(self): while True: message_id = str(uuid.uuid4().hex) if message_id not in self.cache: self.cache[message_id] = time.time() break data = pickle.dumps( {'type': 'system', 'id': message_id, 'data': 'HELLO'} ) self.send(data) def send(self, data): length = len(data) data = struct.pack('III', length, self.version, self.local_id) + data if self.socket is None: if time.time() - self.last_exception_time < 5: return self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: self.socket.connect((self.address, self.port)) self.hello() except Exception: self.last_exception_time = time.time() self.socket = None return try: self.socket.send(data) except Exception: try: self.socket.close() except Exception: pass self.socket = None def close(self): self.socket.close() class Transport(object): def __init__(self, address, port): self.peers = [] self.address = address self.port = port self.version = 0 self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1048576) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.socket.bind((self.address, self.port)) self.socket.listen(16) self.stream_endpoints = [] def add_peer(self, peer): self.peers.append(peer) def send(self, data, exclude=None): exclude = exclude or [] ret = [] for peer in self.peers: if peer.remote_id not in exclude: ret.append(peer.send(data)) return ret def get(self): while True: fds = [self.socket] + self.stream_endpoints [rlist, wlist, xlist] = select.select(fds, [], fds) for fd in xlist: if fd in self.stream_endpoints: ( self.stream_endpoints.pop( self.stream_endpoints.index(fd) ) ) for fd in rlist: if fd == self.socket: new_fd, raddr = self.socket.accept() self.stream_endpoints.append(new_fd) else: data = fd.recv(8) if len(data) == 0: ( self.stream_endpoints.pop( self.stream_endpoints.index(fd) ) ) continue length, version, remote_id = struct.unpack('III', data) if version != self.version: continue data = b'' while len(data) < length: data += fd.recv(length - len(data)) return data, remote_id def close(self): self.socket.close() class Messenger(object): def __init__(self, local_id, transport=None): self.local_id = local_id self.transport = transport or Transport('127.0.0.1', 5680) self.targets = set() self.id_cache = IdCache() def __iter__(self): return self def __next__(self): while True: msg = self.handle() if msg is not None: return msg def handle(self): data, remote_id = self.transport.get() message = pickle.loads(data) if message['id'] in self.id_cache: # discard message return None if message['type'] == 'system': # forward system messages self.transport.send(data, exclude=[remote_id]) return message self.id_cache[message['id']] = time.time() if ( message['type'] == 'transport' and message['target'] in self.targets ): # ignore DB updates with the same target message = None elif ( message['type'] == 'api' and message['target'] not in self.targets ): # ignore API messages with other targets message = None self.transport.send(data, exclude=[remote_id]) return message def emit(self, message): while True: message_id = '%s-%s' % ( message.get('target', '-'), uuid.uuid4().hex, ) if message_id not in self.id_cache: self.id_cache[message_id] = time.time() break message['id'] = message_id return self.transport.send(pickle.dumps(message)) def add_peer(self, remote_id, address, port): peer = Peer(remote_id, self.local_id, address, port, self.id_cache) self.transport.add_peer(peer)	6.2 KB	2025-05-25 12:26
📄 view.py ''' Accessing objects ================= NDB objects are grouped into "views": * interfaces * addresses * routes * neighbours * rules * netns * ... Views are dictionary-like objects that accept strings or dict selectors:: # access eth0 ndb.interfaces["eth0"] # access eth0 in the netns test01 ndb.sources.add(netns="test01") ndb.interfaces[{"target": "test01", "ifname": "eth0"}] # access a route to 10.4.0.0/24 ndb.routes["10.4.0.0/24"] # same with a dict selector ndb.routes[{"dst": "10.4.0.0", "dst_len": 24}] Objects cache ============= NDB create objects on demand, it doesn't create thousands of route objects for thousands of routes by default. The object is being created only when accessed for the first time, and stays in the cache as long as it has any not committed changes. To inspect cached objects, use views' `.cache`:: >>> ndb.interfaces.cache.keys() [(('target', u'localhost'), ('tflags', 0), ('index', 1)), # lo (('target', u'localhost'), ('tflags', 0), ('index', 5))] # eth3 There is no asynchronous cache invalidation, the cache is being cleaned up every time when an object is accessed. API === ''' import errno import gc import json import queue import threading import time from collections import OrderedDict from functools import partial from pyroute2 import cli, config from pyroute2.common import basestring ## # NDB stuff from .auth_manager import check_auth from .objects import RSLV_DELETE from .objects.address import Address from .objects.interface import Interface, Vlan from .objects.neighbour import FDBRecord, Neighbour from .objects.netns import NetNS from .objects.probe import Probe from .objects.route import Route from .objects.rule import Rule from .report import Record, RecordSet from .source import Source, SourceProxy class TmpHandler: def __init__(self, ndb, event, handler): self.ndb = ndb self.event = event self.handler = handler def __enter__(self): self.ndb.task_manager.register_handler( self.ndb.schema.classes[self.event], self.handler ) return self def __exit__(self, exc_type, exc_value, traceback): self.ndb.task_manager.unregister_handler( self.ndb.schema.classes[self.event], self.handler ) class View(dict): ''' The View() object returns RTNL objects on demand:: ifobj1 = ndb.interfaces['eth0'] ifobj2 = ndb.interfaces['eth0'] # ifobj1 != ifobj2 ''' def __init__(self, ndb, table, chain=None, auth_managers=None): self.ndb = ndb self.log = ndb.log.channel('view.%s' % table) self.table = table self.event = table # FIXME self.chain = chain self.cache = {} if auth_managers is None: auth_managers = [] if chain: auth_managers += chain.auth_managers self.auth_managers = auth_managers self.constraints = {} self.classes = OrderedDict() self.classes['interfaces'] = Interface self.classes['addresses'] = Address self.classes['neighbours'] = Neighbour self.classes['af_bridge_fdb'] = FDBRecord self.classes['routes'] = Route self.classes['rules'] = Rule self.classes['netns'] = NetNS self.classes['probes'] = Probe self.classes['af_bridge_vlans'] = Vlan def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): pass @property def default_target(self): if self.table == 'netns': return self.ndb.nsmanager else: return self.ndb.localhost @property def context(self): if self.chain is not None: return self.chain.context else: return {} def getmany(self, spec, table=None): return self.ndb.task_manager.db_get(table or self.table, spec) def getone(self, spec, table=None): for obj in self.getmany(spec, table): return obj @cli.change_pointer @check_auth('obj:read') def get(self, spec=None, table=None, *kwarg): spec = spec or kwarg try: return self.__getitem__(spec, table) except KeyError: return None def template(self, key, table=None): if self.chain: context = self.chain.context else: context = {} iclass = self.classes[table or self.table] spec = iclass.new_spec(key, context, self.default_target) return iclass( self, spec, load=False, master=self.chain, auth_managers=self.auth_managers, ) @cli.change_pointer @check_auth('obj:modify') def create(self, argspec, *kwspec): iclass = self.classes[self.table] if self.chain: context = self.chain.context else: context = {} spec = iclass.new_spec( kwspec or argspec[0], context, self.default_target ) if self.chain: spec['ndb_chain'] = self.chain spec['create'] = True return self[spec] @cli.change_pointer @check_auth('obj:modify') def ensure(self, argspec, kwspec): try: obj = self.locate(kwspec) except KeyError: obj = self.create(*kwspec) for key, value in kwspec.items(): obj[key] = value return obj @cli.change_pointer @check_auth('obj:modify') def add(self, argspec, *kwspec): self.log.warning( '''\n The name add() will be removed in future releases, use create() instead. If you believe that the idea to rename is wrong, please file your opinion to the project's bugtracker. The reason behind the rename is not to confuse interfaces.add() with bridge and bond port operations, that don't create any new interfaces but work on existing ones. ''' ) return self.create(argspec, kwspec) @check_auth('obj:read') def wait(self, spec): ret = None timeout = spec.pop('timeout', -1) action = spec.pop('action', 'add') ctime = time.time() # install a limited events queue -- for a possible immediate reaction evq = queue.Queue(maxsize=100) def handler(evq, target, event): # ignore the "queue full" exception # # if we miss some events here, nothing bad happens: we just # load them from the DB after a timeout, falling back to # the DB polling # # the most important here is not to allocate too much memory try: evq.put_nowait((target, event)) except queue.Full: pass with TmpHandler(self.ndb, self.event, partial(handler, evq)): while True: ret = self.get(spec) if (ret and action == 'add') or ( ret is None and action == 'remove' ): return ret try: target, msg = evq.get(timeout=1) except queue.Empty: pass if timeout > -1: if ctime + timeout < time.time(): raise TimeoutError() @check_auth('obj:read') def locate(self, spec=None, table=None, *kwarg): ''' This method works like `__getitem__()`, but the important difference is that it uses only key fields to locate the object in the DB, ignoring all other keys. It is useful to locate objects that may change attributes during request, like an interface may come up/down, or an address may become primary/secondary, so plain `__getitem__()` will not match while the object still exists. ''' if isinstance(spec, Record): spec = spec._as_dict() spec = spec or kwarg if not spec: raise TypeError('got an empty spec') table = table or self.table iclass = self.classes[table] spec = iclass.new_spec(spec) kspec = self.ndb.schema.compiled[table]['norm_idx'] lookup_fallbacks = self.ndb.schema.compiled[table]['lookup_fallbacks'] request = {} for name in kspec: name = iclass.nla2name(name) if name in spec: request[name] = spec[name] elif name in lookup_fallbacks: fallback = lookup_fallbacks[name] if fallback in spec: request[fallback] = spec[fallback] if not request: raise KeyError('got an empty key') return self[request] @check_auth('obj:read') def __getitem__(self, key, table=None): ret = self.template(key, table) # rtnl_object.key() returns a dictionary that can not # be used as a cache key. Create here a tuple from it. # The key order guaranteed by the dictionary. cache_key = tuple(ret.key.items()) rtime = time.time() # Iterate all the cache to remove unused and clean # (without any started transaction) objects. for ckey in tuple(self.cache): # Skip the current cache_key to avoid extra # cache del/add records in the logs if ckey == cache_key: continue # 1. Remove only expired items # 2. The number of changed rtnl_object fields must # be 0 which means that no transaction is started # 3. The number of referrers must be > 1, the first # one is the cache itself <- this op is expensive! if ( rtime - self.cache[ckey].atime > config.cache_expire and self.cache[ckey].clean and gc.get_referrers(self.cache[ckey]) ): self.log.debug('cache del %s' % (ckey,)) self.cache.pop(ckey, None) if cache_key in self.cache: self.log.debug('cache hit %s' % (cache_key,)) # Explicitly get rid of the created object del ret # The object from the cache has already # registered callbacks, simply return it ret = self.cache[cache_key] ret.atime = rtime return ret else: # Cache only existing objects if self.exists(key): ret.load_sql() self.log.debug('cache add %s' % (cache_key,)) self.cache[cache_key] = ret ret.register() return ret def exists(self, key, table=None): ''' Check if the specified object exists in the database:: ndb.interfaces.exists('eth0') ndb.interfaces.exists({'ifname': 'eth0', 'target': 'localhost'}) ndb.addresses.exists('127.0.0.1/8') ''' if self.chain: context = self.chain.context else: context = {} iclass = self.classes[self.table] key = iclass.new_spec(key, context, self.default_target) iclass.resolve( view=self, spec=key, fields=iclass.resolve_fields, policy=RSLV_DELETE, ) table = table or self.table schema = self.ndb.schema task_manager = self.ndb.task_manager names = schema.compiled[self.table]['all_names'] self.log.debug('check if the key %s exists in table %s' % (key, table)) keys = [] values = [] for name, value in key.items(): nla_name = iclass.name2nla(name) if nla_name in names: name = nla_name if value is not None and name in names: keys.append('f_%s = %s' % (name, schema.plch)) if isinstance(value, (dict, list, tuple, set)): value = json.dumps(value) values.append(value) spec = task_manager.db_fetchone( 'SELECT FROM %s WHERE %s' % (self.table, ' AND '.join(keys)), values, ) if spec is not None: self.log.debug('exists') return True else: self.log.debug('not exists') return False def __setitem__(self, key, value): raise NotImplementedError() def __delitem__(self, key): raise NotImplementedError() def __iter__(self): return self.keys() def __contains__(self, key): return key in self.dump() @check_auth('obj:list') def keys(self): for record in self.dump(): yield record @check_auth('obj:list') def values(self): for key in self.keys(): yield self[key] @check_auth('obj:list') def items(self): for key in self.keys(): yield (key, self[key]) @cli.show_result def count(self): return self.classes[self.table]._count(self)[0] def __len__(self): return self.count() def _keys(self, iclass): return ['target', 'tflags'] + self.ndb.schema.compiled[ iclass.view or iclass.table ]['names'] def _native(self, dump): fnames = next(dump) for record in dump: yield Record(fnames, record, self.classes[self.table]) @cli.show_result @check_auth('obj:list') def dump(self): iclass = self.classes[self.table] return RecordSet( self._native(iclass.dump(self)), config={ 'recordset_pipe': self.ndb.config.get( 'recordset_pipe', 'false' ) }, ) @cli.show_result @check_auth('obj:list') def summary(self): iclass = self.classes[self.table] return RecordSet( self._native(iclass.summary(self)), config={ 'recordset_pipe': self.ndb.config.get( 'recordset_pipe', 'false' ) }, ) def __repr__(self): if self.chain and 'ifname' in self.chain: parent = f'{self.chain["ifname"]}/' else: parent = '' return f''' NDB view for {parent}{self.table} Number of objects: {self.count()} to list objects use .summary() or .dump() -> RecordSet (generator) -> Record key: Union[Record, dict, spec] to get objects use ...[key] / .__getitem__(key) -> RTNL_Object ''' class SourcesView(View): def __init__(self, ndb, auth_managers=None): super(SourcesView, self).__init__(ndb, 'sources') self.classes['sources'] = Source self.cache = {} self.proxy = {} self.lock = threading.Lock() if auth_managers is None: auth_managers = [] self.auth_managers = auth_managers def async_add(self, spec): spec = dict(Source.defaults(spec)) self.cache[spec['target']] = Source(self.ndb, spec).start() return self.cache[spec['target']] def add(self, spec): spec = dict(Source.defaults(spec)) target = spec['target'] if target in self: raise KeyError(f'source {target} exists') if 'event' not in spec: sync = True spec['event'] = threading.Event() else: sync = False self.cache[spec['target']] = Source(self.ndb, spec).start() if sync: self.cache[spec['target']].event.wait() return self.cache[spec['target']] def remove(self, target, code=errno.ECONNRESET, sync=True): if target not in self: raise KeyError(f'source {target} does not exist') with self.lock: if target in self.cache: source = self.cache[target] source.close(code=code, sync=sync) return self.cache.pop(target) @check_auth('obj:list') def keys(self): for key in self.cache: yield key def _keys(self, iclass): return ['target', 'kind'] def wait(self, *spec): raise NotImplementedError() def _summary(self, argv, *kwarg): return self._dump(argv, **kwarg) def __getitem__(self, key, table=None): if isinstance(key, basestring): target = key elif isinstance(key, dict) and 'target' in key.keys(): target = key['target'] else: raise KeyError() if target in self.cache: return self.cache[target] elif target in self.proxy: return self.proxy[target] else: proxy = SourceProxy(self.ndb, target) self.proxy[target] = proxy return proxy	16.57 KB	2025-05-25 12:26

Terminal