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 ''' CLI provides a simple syntax to manipulate NDB. The syntax is the same for the console and the http versions. The first level of the hierarchy represents NDB views: * interfaces -- network interfaces * addresses -- IP addresses * routes -- IP and MPLS routes, one record per NH * neighbours -- ARP cache * rules -- RPDB rules * netns -- network namespaces * vlans -- bridge VLAN filters CLI supports indentation, though it is optional. A level up in the indentation means a level up in the object hierarchy. Same effect with `..` command. An example script to create a bridge and a port, no indentation:: ; comments start with ; interfaces create ifname br0, kind bridge address 00:11:22:33:44:55 state up add_ip 192.168.123.21/24 add_ip 192.168.123.22/24 commit .. ; level up to the interfaces view create ifname br0p0, kind dummy state up master br0 commit Same script with indentation, no `..` needed:: interfaces create ifname br0, kind bridge address 00:11:22:33:44:55 state up add_ip 192.168.123.21/24 add_ip 192.168.123.22/24 commit create ifname br0p0, kind dummy state up master br0 commit Select objects:: ; by name interfaces br0 ; ... ; by spec interfaces {target netns01, ifname eth0} ; ... {address 00:11:22:33:44:55} ; ... Manage interfaces ----------------- Create:: interfaces create ifname br0.100, kind vlan, vlan_id 100, link br0 commit create ifname v0, kind veth, peer v0p commit Change mac address:: interfaces br0 address 00:11:22:33:44:55 commit Change netns and rename:: sources add netns test01 interfaces v0p net_ns_fd test01 commit {target test01, ifname v0p} ifname eth0 commit summary \| filter kind veth \| select target, ifname \| format json Manage addresses ---------------- ... Manage routes ------------- ... Generate reports ---------------- It is possible to modify the output of the dump or summary commands:: interfaces ; print index, ifname and MAC address for UP interfaces dump \| filter state up \| select index, ifname, address routes ; output in JSON format summary \| format json The `format` command can only be the last in the sentence. Available output filters: filter { } -- filter out records select { } -- output only selected record fields format { } -- change the output format, possible values: csv, json Wait for events --------------- ... ''' t_stmt = 1 t_dict = 2 t_comma = 3 t_pipe = 4 t_end_of_dict = 7 t_end_of_sentence = 8 t_end_of_stream = 9 def change_pointer(f): f.__cli_cptr__ = True return f def show_result(f): f.__cli_publish__ = True return f	2.92 KB	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📁 auth	-	2025-05-25 12:26
📄 console.py import code import getpass import socket import sys from pyroute2.cli.session import Session from pyroute2.ndb.main import NDB class Console(code.InteractiveConsole): def __init__(self, stdout=None, log=None, sources=None): self.db = NDB(log=log, sources=sources) self.db.config.update( {'show_format': 'json', 'recordset_pipe': 'true'} ) self.stdout = stdout or sys.stdout self.session = Session(self.db, self.stdout, self.set_prompt) self.matches = [] self.isatty = sys.stdin.isatty() self.prompt = '' self.set_prompt() code.InteractiveConsole.__init__(self) def close(self): self.db.close() def help(self): self.session.lprint( "Built-in commands: \n" "exit\t-- exit cli\n" "ls\t-- list current namespace\n" ".\t-- print the current object\n" ".. or Ctrl-D\t-- one level up\n" ) def set_prompt(self, prompt=None): if self.isatty: if prompt is not None: self.prompt = '%s > ' % (prompt) else: self.prompt = '%s > ' % (self.session.ptr.__class__.__name__) self.prompt = '%s@%s : %s' % ( getpass.getuser(), (socket.gethostname().split('.')[0]), self.prompt, ) def loadrc(self, fname): with open(fname, 'r') as f: self.session.handle(f.read()) def interact(self, readfunc=None): if self.isatty and readfunc is None: self.session.lprint("pyroute2 cli prototype") if readfunc is None: readfunc = self.raw_input indent = 0 while True: try: text = readfunc(self.prompt) except EOFError: if self.session.stack: self.session.stack_pop() continue else: self.close() break except Exception: self.close() break try: indent = self.session.handle(text, indent) except SystemExit: self.close() return except: self.showtraceback() continue def set_completer(self, readline): readline.parse_and_bind('tab: complete') readline.set_completer(self.completer) readline.set_completion_display_matches_hook(self.display) def completer(self, text, state): if state == 0: d = [x for x in dir(self.session.ptr) if x.startswith(text)] if isinstance(self.session.ptr, dict): keys = [str(y) for y in self.session.ptr.keys()] d.extend([x for x in keys if x.startswith(text)]) self.matches = d try: return self.matches[state] except: pass def display(self, line, matches, length): self.session.lprint() self.session.lprint(matches) self.session.lprint('%s%s' % (self.prompt, line), end='') if __name__ == '__main__': Console().interact()	3.15 KB	2025-05-25 12:26
📄 parser.py import re import shlex from pyroute2.cli import ( t_comma, t_dict, t_end_of_dict, t_end_of_sentence, t_end_of_stream, t_pipe, t_stmt, ) from pyroute2.common import basestring class Token(object): def __init__(self, lex, expect=(), prohibit=(), leaf=False): self.lex = lex self.leaf = leaf self.kind = 0 self.name = None self.argv = [] self.kwarg = {} self.parse() if expect and self.kind not in expect: raise SyntaxError('expected %s, got %s' % (expect, self.kind)) if prohibit and self.kind in prohibit: raise SyntaxError('unexpected %s' % (self.name,)) def convert(self, arg): if re.match('^[0-9]+$', arg): return int(arg) else: return arg def parse(self): # triage first = self.lex.get_token() self.name = first ## # no token # if first == '': self.kind = t_end_of_stream ## # dict, e.g. # # resource spec, function arguments:: # {arg1, arg2} # {key1 value1, key2 value2} # {key {skey1 value}} # elif first == '{': arg_name = None while True: nt = Token( self.lex, expect=(t_stmt, t_dict, t_comma, t_end_of_dict) ) if arg_name is None: if nt.kind == t_dict: self.argv.append(nt.kwarg) elif nt.kind == t_comma: continue elif nt.kind == t_stmt: arg_name = nt.name else: if nt.kind in (t_end_of_dict, t_comma): self.argv.append(arg_name) elif nt.kind == t_stmt: self.kwarg[arg_name] = nt.name elif nt.kind == t_dict: self.kwarg[arg_name] = nt.kwarg arg_name = None if nt.kind == t_end_of_dict: self.kind = t_dict self.name = '%s %s' % (self.argv, self.kwarg) return ## # end of dict # elif first == '}': self.kind = t_end_of_dict ## # end of sentence # elif first == ';': self.kind = t_end_of_sentence ## # end of dict entry # elif first == ',': self.kind = t_comma ## # pipe # elif first == '\|': self.kind = t_pipe elif first == '=': lookahead = self.lex.get_token() if lookahead == '>': self.name = '=>' self.kind = t_pipe else: self.lex.push_token(lookahead) self.kind = t_stmt ## # simple statement # # object name:: # name # # function call:: # func # func {arg1, arg2} # func {key1 value1, key2 value2} # else: self.name = self.convert(first) self.kind = t_stmt class Sentence(object): def __init__(self, text, indent=0, master=None): self.offset = 0 self.statements = [] self.text = text self.lex = shlex.shlex(text) self.lex.wordchars += '.:/' self.lex.commenters = '#!' self.lex.debug = False self.indent = indent if master: self.chain = master.chain else: self.chain = [] self.parse() def __iter__(self): for stmt in self.statements: yield stmt def parse(self): sentence = self while True: nt = Token(self.lex) if nt.kind == t_end_of_sentence: sentence = Sentence(None, self.indent, master=self) elif nt.kind == t_end_of_stream: return else: sentence.statements.append(nt) if sentence not in self.chain: self.chain.append(sentence) def __repr__(self): ret = '----\n' for s in self.statements: ret += '%i [%s] %s\n' % (self.indent, s.kind, s.name) ret += '\targv: %s\n' % (s.argv) ret += '\tkwarg: %s\n' % (s.kwarg) return ret class Parser(object): def __init__(self, stream): self.stream = stream self.indent = None self.sentences = [] self.parse() def parse(self): if hasattr(self.stream, 'readlines'): for text in self.stream.readlines(): self.parse_string(text) elif isinstance(self.stream, basestring): self.parse_string(self.stream) else: raise ValueError('unsupported stream') self.parsed = True def parse_string(self, text): # 1. get indentation indent = re.match(r'^([ \t]*)', text).groups(0)[0] spaces = [] # 2. sort it if indent: spaces = list(set(indent)) if len(spaces) > 1: raise SyntaxError('mixed indentation') if self.indent is None: self.indent = spaces[0] if self.indent != spaces[0]: raise SyntaxError('mixed indentation') sentence = Sentence(text, len(indent)) self.sentences.extend(sentence.chain)	5.46 KB	2025-05-25 12:26
📄 server.py import json from http.server import BaseHTTPRequestHandler from http.server import HTTPServer as HTTPServer from pyroute2.cli.session import Session from pyroute2.ndb.main import NDB class ProxyEncoder(object): def __init__(self, wfile): self.wfile = wfile def write(self, data): self.wfile.write(data.encode('utf-8')) def flush(self): self.wfile.flush() class Handler(BaseHTTPRequestHandler): def do_error(self, code, reason): self.send_error(code, reason) self.end_headers() def do_POST(self): # # sanity checks: # # * path if self.path != '/v1/': return self.do_error(404, 'url not found') # * content length if 'Content-Length' not in self.headers: return self.do_error(411, 'Content-Length') # * content type if 'Content-Type' not in self.headers: return self.do_error(400, 'Content-Type') # content_length = int(self.headers['Content-Length']) content_type = self.headers['Content-Type'] data = self.rfile.read(content_length) if content_type == 'application/json': try: request = json.loads(data) except ValueError: return self.do_error(400, 'Incorrect JSON input') elif content_type == 'text/plain': request = {'commands': data.decode('utf-8').split(';')} else: self.do_error(400, 'Incorrect content type') # auth plugins if 'X-Auth-Mech' in self.headers: auth_plugin = self.server.auth_plugins.get( self.headers['X-Auth-Mech'] ) if auth_plugin is None: return self.do_error(501, 'Authentication mechanism not found') try: am = auth_plugin(self.headers) except Exception: return self.do_error(401, 'Authentication failed') ndb = self.server.ndb.auth_proxy(am) elif self.server.auth_strict: return self.do_error(401, 'Authentication required') else: ndb = self.server.ndb session = Session( ndb=ndb, stdout=ProxyEncoder(self.wfile), builtins=('ls', '.', '..', 'version'), ) self.send_response(200) self.end_headers() for cmd in request['commands']: session.handle(cmd) class Server(HTTPServer): def __init__( self, address='localhost', port=8080, sources=None, ndb=None, log=None, auth_strict=False, auth_plugins=None, ): self.sessions = {} self.auth_strict = auth_strict self.auth_plugins = auth_plugins or {} if ndb is not None: self.ndb = ndb else: self.ndb = NDB(sources=sources, log=log) self.ndb.config.update( {'show_format': 'json', 'recordset_pipe': 'true'} ) HTTPServer.__init__(self, (address, port), Handler)	3.03 KB	2025-05-25 12:26
📄 session.py from __future__ import print_function import sys import traceback from collections import namedtuple from pyroute2 import config from pyroute2.cli import t_comma, t_dict, t_pipe, t_stmt from pyroute2.cli.parser import Parser from pyroute2.common import basestring class Session(object): def __init__(self, ndb, stdout=None, ptrname_callback=None, builtins=None): self.db = ndb self.ptr = self.db self._ptrname = None self._ptrname_callback = ptrname_callback self.stack = [] self.errors = 0 self.indent_stack = set() self.prompt = '' self.stdout = stdout or sys.stdout self.builtins = builtins or ( 'ls', '.', '..', 'version', 'exit', ':stack', ) @property def ptrname(self): return self._ptrname @ptrname.setter def ptrname(self, name): self._ptrname = name if self._ptrname_callback is not None: self._ptrname_callback(name) def stack_pop(self): self.ptr, self.ptrname = self.stack.pop() return (self.ptr, self.ptrname) def lprint(self, text='', end='\n'): if not isinstance(text, basestring): text = str(text) self.stdout.write(text) if end: self.stdout.write(end) self.stdout.flush() def handle_statement(self, stmt, token): obj = None if stmt.kind == t_dict: obj = self.ptr[stmt.kwarg] elif stmt.kind == t_stmt: obj = getattr(self.ptr, stmt.name, None) if obj is None and isinstance(self.ptr, dict): try: obj = self.ptr.get(stmt.name, None) except KeyError: pass if hasattr(obj, '__call__'): try: nt = next(token) except StopIteration: nt = namedtuple('Token', ('kind', 'argv', 'kwarg'))( t_dict, [], {} ) if nt.kind == t_dict: args = nt try: pipe = next(token) if pipe.kind != t_pipe: raise TypeError('pipe expected') except StopIteration: pipe = None elif nt.kind == t_stmt: argv = [] kwarg = {} arg_name = nt.name pipe = None for nt in token: if arg_name is None: if nt.kind == t_stmt: arg_name = nt.name elif nt.kind == t_comma: continue elif nt.kind == t_pipe: pipe = nt break else: raise TypeError('stmt expected') else: if nt.kind == t_comma: argv.append(arg_name) elif nt.kind == t_stmt: kwarg[arg_name] = nt.name elif nt.kind == t_pipe: pipe = nt break else: raise TypeError('stmt or comma expected') arg_name = None if arg_name is not None: argv.append(arg_name) args = namedtuple('Token', ('kind', 'argv', 'kwarg'))( t_dict, argv, kwarg ) elif nt.kind == t_pipe: args = namedtuple('Token', ('kind', 'argv', 'kwarg'))( t_dict, [], {} ) pipe = nt else: raise TypeError('dict, stmt or comma expected') # at this step we have # args -- arguments # pipe -- pipe or None try: ret = obj(args.argv, *args.kwarg) # if pipe is not None: ptr = self.ptr self.ptr = ret try: stmt = next(token) except StopIteration: raise TypeError('statement expected') if stmt.kind != t_stmt: raise TypeError('statement expected') try: self.handle_statement(stmt, token) except Exception: pass self.ptr = ptr return if hasattr(obj, '__cli_cptr__'): obj = ret elif hasattr(obj, '__cli_publish__'): if hasattr(ret, 'generator') or hasattr(ret, 'next'): for line in ret: if isinstance(line, basestring): self.lprint(line) else: self.lprint(repr(line)) else: self.lprint(ret) return elif isinstance(ret, (bool, basestring, int, float)): self.lprint(ret) return else: return except Exception: self.errors += 1 traceback.print_exc() return else: if isinstance(self.ptr, dict) and not isinstance(obj, dict): try: nt = next(token) if nt.kind == t_stmt: self.ptr[stmt.name] = nt.name elif nt.kind == t_dict and nt.argv: self.ptr[stmt.name] = nt.argv elif nt.kind == t_dict and nt.kwarg: self.ptr[stmt.name] = nt.kwarg else: raise TypeError('failed setting a key/value pair') return except NotImplementedError: raise KeyError() except StopIteration: pass if obj is None: raise KeyError() elif isinstance(obj, (basestring, int, float)): self.lprint(obj) else: return obj def handle_sentence(self, sentence, indent): if sentence.indent < indent: while max(self.indent_stack) > sentence.indent: self.indent_stack.remove(max(self.indent_stack)) if self.stack: self.ptr, self.ptrname = self.stack.pop() else: self.indent_stack.add(sentence.indent) indent = sentence.indent iterator = iter(sentence) obj = None save_ptr = self.ptr save_ptrname = self.ptrname try: for stmt in iterator: if stmt.name in self.builtins: if stmt.name == 'exit': raise SystemExit() elif stmt.name == 'ls': self.lprint(dir(self.ptr)) elif stmt.name == ':stack': self.lprint('stack:') for item in self.stack: self.lprint(item) self.lprint('end') elif stmt.name == '.': self.lprint(repr(self.ptr)) elif stmt.name == '..': if self.stack: save_ptr, save_ptrname = self.stack.pop() elif stmt.name == 'version': try: self.lprint(config.version.__version__) except: self.lprint('unknown') break else: try: obj = self.handle_statement(stmt, iterator) if obj is not None: self.ptr = obj if hasattr(obj, 'key_repr'): self.ptrname = obj.key_repr() else: self.ptrname = stmt.name except KeyError: self.lprint('object not found') self.errors += 1 return indent except: self.errors += 1 traceback.print_exc() except SystemExit: raise finally: if obj is not None: self.stack.append((save_ptr, save_ptrname)) else: self.ptr, self.ptrname = save_ptr, save_ptrname return indent def handle(self, text, indent=0): parser = Parser(text) for sentence in parser.sentences: indent = self.handle_sentence(sentence, indent) return indent	8.94 KB	2025-05-25 12:26

Terminal