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 import builtins import json import struct from pyroute2.netlink import nlmsg from pyroute2.netlink.nlsocket import Marshal Tversion = 100 Rversion = 101 Tauth = 102 Rauth = 103 Tattach = 104 Rattach = 105 Terror = 106 # illegal Rerror = 107 Tflush = 108 Rflush = 109 Twalk = 110 Rwalk = 111 Topen = 112 Ropen = 113 Tcreate = 114 Rcreate = 115 Tread = 116 Rread = 117 Twrite = 118 Rwrite = 119 Tclunk = 120 Rclunk = 121 Tremove = 122 Rremove = 123 Tstat = 124 Rstat = 125 Twstat = 126 Rwstat = 127 Topenfd = 98 Ropenfd = 99 # 9P2000.pr2 extensions Tcall = 80 Rcall = 81 class Plan9Exit(Exception): pass def array(kind, header='H'): class CustomArray: @staticmethod def decode_count(data, offset): (count,) = struct.unpack_from(header, data, offset) return count, offset + struct.calcsize(header) @staticmethod def decode_from(data, offset): count, offset = CustomArray.decode_count(data, offset) ret = [] for _ in range(count): value, offset = kind.decode_from(data, offset) ret.append(value) return ret, offset @staticmethod def encode_into(data, offset, value): if not isinstance(value, (tuple, list)): value = [] data.extend([0] * struct.calcsize(header)) struct.pack_into(header, data, offset, len(value)) offset += struct.calcsize(header) for item in value: offset = kind.encode_into(data, offset, item) return offset return CustomArray class Qid(dict): length = 13 def __init__(self, qtype, vers, path): self['type'] = qtype self['vers'] = vers self['path'] = path @staticmethod def decode_from(data, offset): return ( dict( zip( ('type', 'vers', 'path'), struct.unpack_from('=BIQ', data, offset), ) ), offset + struct.calcsize('=BIQ'), ) @staticmethod def encode_into(data, offset, value): data.extend([0] * Qid.length) struct.pack_into( '=BIQ', data, offset, value['type'], value['vers'], value['path'] ) return offset + Qid.length class Stat(dict): header_fmt = 'H' def __init__(self): self['size'] = 58 self['type'] = 0 self['dev'] = 0 self['qid.type'] = 0 self['qid.vers'] = 0 self['qid.path'] = 0 self['mode'] = 0 self['atime'] = 0 self['mtime'] = 0 self['length'] = 0 self['name'] = '' self['uid'] = '' self['gid'] = '' self['muid'] = '' @staticmethod def decode_from(data, offset=0): ret = dict( zip( ( 'size', 'type', 'dev', 'qid.type', 'qid.vers', 'qid.path', 'mode', 'atime', 'mtime', 'length', ), struct.unpack_from('=HHIBIQIIIQ', data, offset), ) ) offset += 41 for key in ('name', 'uid', 'gid', 'muid'): ret[key], offset = String.decode_from(data, offset) return ret, offset @staticmethod def encode_into(data, offset, value): data.extend([0] * 41) # size of all the data except uint16 `size` header # 41 + 2 + name + 2 + uid + 2 + gid + 2 + muid value['size'] = ( 47 + len(value['name']) + len(value['uid']) + len(value['gid']) + len(value['muid']) ) struct.pack_into( '=HHIBIQIIIQ', data, offset, value['size'], value['type'], value['dev'], value['qid.type'], value['qid.vers'], value['qid.path'], value['mode'], value['atime'], value['mtime'], value['length'], ) offset += 41 for key in ('name', 'uid', 'gid', 'muid'): offset = String.encode_into(data, offset, value[key]) return offset class WStat(Stat): @staticmethod def decode_from(data, offset=0): # just ignore plength for now return Stat.decode_from(data, offset + 2) @staticmethod def encode_into(data, offset, value): data.extend([0] * 2) new_offset = Stat.encode_into(data, offset + 2, value) # size of all the data except uint16 `plength` header struct.pack_into('H', data, offset, new_offset - offset - 2) return new_offset class CData: header_fmt = 'I' @staticmethod def decode_from(data, offset=0): (length,) = struct.unpack_from(CData.header_fmt, data, offset) offset += struct.calcsize(CData.header_fmt) return bytearray(data[offset : offset + length]), offset + length @staticmethod def encode_into(data, offset, value): length = len(value) if isinstance(value, str): value = value.encode('utf-8') data.extend([0] * (length + struct.calcsize(CData.header_fmt))) struct.pack_into( f'{CData.header_fmt}{length}s', data, offset, length, value ) return offset + length + struct.calcsize(CData.header_fmt) class String: header_fmt = 'H' @staticmethod def decode_from(data, offset=0): (length,) = struct.unpack_from(String.header_fmt, data, offset) offset += struct.calcsize(String.header_fmt) (value,) = struct.unpack_from(f'{length}s', data, offset) value = value.decode('utf-8') return value, offset + length @staticmethod def encode_into(data, offset, value): length = len(value) data.extend([0] * (length + struct.calcsize(String.header_fmt))) struct.pack_into( f'{String.header_fmt}{length}s', data, offset, length, value.encode('utf-8'), ) return offset + length + struct.calcsize(String.header_fmt) class msg_base(nlmsg): align = 0 header = (('length', '=I'), ('type', 'B'), ('tag', 'H')) class msg_terror(msg_base): defaults = {'header': {'type': Terror}} class msg_rerror(msg_base): defaults = {'header': {'type': Rerror}} fields = (('ename', String),) class msg_tversion(msg_base): defaults = {'header': {'type': Tversion}} fields = (('msize', 'I'), ('version', String)) class msg_rversion(msg_base): defaults = {'header': {'type': Rversion}} fields = (('msize', 'I'), ('version', String)) class msg_tauth(msg_base): defaults = {'header': {'type': Tauth}} fields = (('afid', 'I'), ('uname', String), ('aname', String)) class msg_rauth(msg_base): defaults = {'header': {'type': Rauth}} fields = (('aqid', '13B'),) class msg_tattach(msg_base): defaults = {'header': {'type': Tattach}} fields = ( ('fid', 'I'), ('afid', 'I'), ('uname', String), ('aname', String), ) class msg_rattach(msg_base): defaults = {'header': {'type': Rattach}} fields = (('qid', Qid),) class msg_twalk(msg_base): defaults = {'header': {'type': Twalk}} fields = (('fid', 'I'), ('newfid', 'I'), ('wname', array(String))) class msg_rwalk(msg_base): defaults = {'header': {'type': Rwalk}} fields = (('wqid', array(Qid)),) class msg_tstat(msg_base): defaults = {'header': {'type': Tstat}} fields = (('fid', 'I'),) class msg_rstat(msg_base): defaults = {'header': {'type': Rstat}} fields = (('stat', WStat),) class msg_twstat(msg_base): defaults = {'header': {'type': Twstat}} fields = (('fid', 'I'), ('stat', WStat)) class msg_rwstat(msg_base): defaults = {'header': {'type': Rwstat}} class msg_tclunk(msg_base): defaults = {'header': {'type': Tclunk}} fields = (('fid', 'I'),) class msg_rclunk(msg_base): defaults = {'header': {'type': Rclunk}} pass class msg_topen(msg_base): defaults = {'header': {'type': Topen}} fields = (('fid', 'I'), ('mode', 'B')) class msg_ropen(msg_base): defaults = {'header': {'type': Ropen}} fields = (('qid', Qid), ('iounit', 'I')) class msg_tread(msg_base): defaults = {'header': {'type': Tread}} fields = (('fid', 'I'), ('offset', 'Q'), ('count', 'I')) class msg_rread(msg_base): defaults = {'header': {'type': Rread}, 'data': b''} fields = (('data', CData),) class msg_twrite(msg_base): defaults = {'header': {'type': Twrite}} fields = (('fid', 'I'), ('offset', 'Q'), ('data', CData)) class msg_rwrite(msg_base): defaults = {'header': {'type': Rwrite}} fields = (('count', 'I'),) class msg_tcall(msg_base): defaults = {'header': {'type': Tcall}} fields = (('fid', 'I'), ('text', String), ('data', CData)) class msg_rcall(msg_base): defaults = {'header': {'type': Rcall}} fields = (('err', 'H'), ('text', String), ('data', CData)) class Marshal9P(Marshal): default_message_class = msg_rerror msg_map = { Tversion: msg_tversion, Rversion: msg_rversion, Tauth: msg_tauth, Rauth: msg_rauth, Tattach: msg_tattach, Rattach: msg_rattach, Rerror: msg_rerror, Twalk: msg_twalk, Rwalk: msg_rwalk, Topen: msg_topen, Ropen: msg_ropen, Tread: msg_tread, Rread: msg_rread, Tclunk: msg_tclunk, Rclunk: msg_rclunk, Tstat: msg_tstat, Rstat: msg_rstat, Twstat: msg_twstat, Rwstat: msg_rwstat, Twrite: msg_twrite, Rwrite: msg_rwrite, Tcall: msg_tcall, Rcall: msg_rcall, Tcreate: msg_base, Rcreate: msg_base, Tremove: msg_base, Rremove: msg_base, } def parse(self, data, seq=None, callback=None, skip_alien_seq=False): offset = 0 while offset <= len(data) - 5: (length, key, tag) = struct.unpack_from('=IBH', data, offset) if skip_alien_seq and tag != seq: continue if not 0 < length <= len(data): break parser = self.get_parser(key, 0, tag) msg = parser(data, offset, length) if key == Rerror: spec = json.loads(msg['ename']) if spec['class'] in dir(builtins): cls = getattr(builtins, spec['class']) elif spec['class'] == 'Plan9Exit': cls = Plan9Exit else: cls = Exception if not spec.get('argv'): spec['argv'] = [spec['str']] raise cls(*spec['argv']) offset += length if msg is None: continue yield msg	10.82 KB	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 client.py import json import os import pwd import struct from pyroute2.common import AddrPool from pyroute2.netlink.core import ( AsyncCoreSocket, CoreMessageQueue, CoreStreamProtocol, ) from pyroute2.netlink.coredata import CoreConfig, CoreSocketSpec from pyroute2.plan9 import ( Marshal9P, msg_tattach, msg_tcall, msg_tread, msg_tversion, msg_twalk, msg_twrite, ) class Plan9ClientSocket(AsyncCoreSocket): '''9p2000 client. * address -- `('address', port)` to listen on * use_socket -- alternatively, provide a connected SOCK_STRAM socket ''' def __init__(self, address=None, use_socket=None, use_event_loop=None): self.spec = CoreSocketSpec( CoreConfig( tag_field='tag', target='localhost', netns=None, address=address, use_socket=use_socket is not None, use_event_loop=use_event_loop is not None, ) ) self.marshal = Marshal9P() self.wnames = {'': 0} self.cwd = 0 self.fid_pool = AddrPool(minaddr=0x00000001, maxaddr=0x0000FFFF) super().__init__(use_socket=use_socket, use_event_loop=use_event_loop) def enqueue(self, data, addr): tag = struct.unpack_from('H', data, 5)[0] return self.msg_queue.put_nowait(tag, data) async def setup_endpoint(self, loop=None): if getattr(self.local, 'transport', None) is not None: return self.local.msg_queue = CoreMessageQueue(event_loop=self.event_loop) if self.status['use_socket']: address = {'sock': self.use_socket} else: address = { 'host': self.status['address'][0], 'port': self.status['address'][1], } self.local.transport, self.local.protocol = ( await self.event_loop.create_connection( lambda: CoreStreamProtocol( self.connection_lost, self.enqueue, self._error_event, self.status, ), *address, ) ) async def start_session(self): '''Initiate 9p2000 session. One must await this routine before running any other requests. ''' await self.setup_endpoint() await self.version() await self.auth() await self.attach() async def request(self, msg, tag=0): await self.setup_endpoint() if tag == 0: tag = self.addr_pool.alloc() try: msg['header']['tag'] = tag msg.reset() msg.encode() self.msg_queue.ensure_tag(tag) self.transport.write(msg.data) return [x async for x in self.get(msg_seq=tag)][0] finally: self.addr_pool.free(tag, ban=0xFF) async def version(self): '''`Tverion` request. No arguments required.''' m = msg_tversion() m['header']['tag'] = 0xFFFF m['msize'] = 8192 m['version'] = '9P2000' return await self.request(m, tag=0xFFFF) async def auth(self): pass async def attach(self, aname=''): '''`Tattach` request. `aname` (optional) -- aname to attach to ''' m = msg_tattach() m['fid'] = 0 m['afid'] = 0xFFFFFFFF m['uname'] = pwd.getpwuid(os.getuid()).pw_name m['aname'] = aname return await self.request(m) async def walk(self, path, newfid=None, fid=None): '''`Twalk` request. * `path` -- string path to the file * `newfid` (optional) -- use this fid to store the info * `fid` (optional) -- use this fid to walk from, otherwise walk from the current directory for this client session ''' m = msg_twalk() m['fid'] = self.cwd if fid is None else fid m['newfid'] = newfid if newfid is not None else self.fid_pool.alloc() m['wname'] = path.split(os.path.sep) self.wnames[path] = m['newfid'] return await self.request(m) async def fid(self, path): '''Walk the path and return `fid` to the required file. * `path` -- string path to the file ''' if path not in self.wnames: newfid = self.fid_pool.alloc() await self.walk(path, newfid) self.wnames[path] = newfid return self.wnames[path] async def read(self, fid, offset=0, count=8192): '''`Tread` request. * `fid` -- fid of the file to read from * `offset` (optional, default 0) -- read offset * `count` (optional, default 8192) -- read count ''' m = msg_tread() m['fid'] = fid m['offset'] = offset m['count'] = count return await self.request(m) async def write(self, fid, data, offset=0): '''`Twrite` request. * `fid` -- fid of the file to write to * `data` -- bytes to write * `offset` (optional, default 0) -- write offset ''' m = msg_twrite() m['fid'] = fid m['offset'] = 0 m['data'] = data return await self.request(m) async def call( self, fid, argv=None, kwarg=None, data=b'', data_arg='data', loader=json.loads, ): '''`Tcall` request. * `fid` -- fid of the file that represents a registered function * `argv` (optional) -- positional arguments as an iterable * `kwarg` (optional) -- keyword arguments as a dictionary * `data` (opional) -- optional binary data * `data_arg` (optional) -- name of the argument to use with the binary data * `loader` (optional, default `json.loads`) -- loader for the response data ''' spec = { 'argv': argv if argv is not None else [], 'kwarg': kwarg if kwarg is not None else {}, 'data_arg': data_arg, } m = msg_tcall() m['fid'] = fid m['text'] = json.dumps(spec) m['data'] = data response = await self.request(m) return loader(response['data'])	6.13 KB	2025-05-25 12:26
📄 filesystem.py import grp import io import json import os import pwd import time from dataclasses import dataclass from functools import partial from pyroute2.plan9 import Plan9Exit, Qid, Stat, Tcall, Tread, Twrite def _publish_function_w(session, inode, request, response): inode.metadata.dirty = True inode.data.seek(0) inode.data.truncate() inode.data.write(request['data']) response['count'] = len(request['data']) return response def _publish_function_r( func, loader, dumper, session, inode, request, response ): if request['offset'] == 0 and inode.metadata.has_new_data(): try: kwarg = loader(inode.data.getvalue()) ret = func(*kwarg) inode.metadata.dirty = False except Plan9Exit: raise except Exception as e: ret = e inode.data.seek(0) inode.data.truncate() inode.data.write(dumper(ret)) inode.data.seek(request['offset']) response['data'] = inode.data.read(request['count']) return response def _publish_function_c( func, loader, dumper, session, inode, request, response ): spec = loader(request['text']) if 'argv' not in spec: spec['argv'] = [] if 'kwarg' not in spec: spec['kwarg'] = {} if 'data_arg' not in spec: spec['data_arg'] = 'data' data = request['data'] if data: spec['kwarg'][spec['data_arg']] = data ret = func(spec['argv'], *spec['kwarg']) response['err'] = 0 response['text'] = '' response['data'] = dumper(ret) return response @dataclass class InodeMetadata: call_on_read: bool = False dirty: bool = False def has_new_data(self) -> bool: return self.dirty or self.call_on_read class Inode: children = None parents = None qid = None stat = None data = None callbacks = None def __init__( self, name, path, data='', qtype=0, mode=0o600, uid=None, gid=None, parents=None, children=None, ): self.data = io.BytesIO(data.encode('utf-8')) self.parents = parents if parents is not None else set() self.children = children if children is not None else set() self.callbacks = {} self.metadata = InodeMetadata() self.stat = Stat() self.qid = Qid(qtype, 0, path) self.stat['uid'] = ( uid if uid is not None else pwd.getpwuid(os.getuid()).pw_name ) self.stat['gid'] = ( gid if gid is not None else grp.getgrgid(os.getgid()).gr_name ) self.stat['muid'] = self.stat['uid'] self.stat['qid.type'] = self.qid['type'] self.stat['qid.vers'] = self.qid['vers'] self.stat['qid.path'] = self.qid['path'] # shift qid.type 3 bytes left and OR with mode self.stat['mode'] = (self.qid['type'] << (8 3)) \| mode self.stat['type'] = self.qid['path'] self.stat['dev'] = 0 self.stat['mtime'] = int(time.time()) self.stat['atime'] = int(time.time()) self.stat['name'] = name self.sync() def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): pass def sync(self): self.stat['length'] = len(self.data.getvalue()) def get_parent(self, name=None): return tuple(self.parents)[0] def get_child(self, name): for child in self.children: if child.stat['name'] == name: return child raise KeyError('file not found') def register_function( self, func, loader=json.loads, dumper=lambda x: x if isinstance(x, bytes) else str(x).encode('utf-8'), ): self.add_callback(Twrite, _publish_function_w) self.add_callback( Tread, partial(_publish_function_r, func, loader, dumper) ) self.add_callback( Tcall, partial(_publish_function_c, func, loader, dumper) ) def add_callback(self, call, f): self.callbacks[call] = f return self def add_parent(self, inode): return self.parents.add(inode) def add_child(self, inode): inode.add_parent(self) return self.children.add(inode) def del_parent(self, inode, force=False): if force or len(self.parents) > 1: return self.parents.remove(inode) raise KeyError('can not remove the only parent') def del_child(self, inode): inode.del_parent(self, force=True) return self.children.remove(inode) def __id__(self): return self.qid['path'] class Filesystem: inodes = {} def __init__(self): self.__path = 255 # create the root inode path = 0 self.inodes[path] = root = Inode('/', path, qtype=0x80, mode=0o755) root.add_parent(root) @property def new_path(self): self.__path += 1 return self.__path def walk(self, wname): inode = self.inodes[0] for name in wname.split(os.path.sep): if name == '': continue inode = inode.get_child(name) return inode def create(self, wname, qtype=0, mode=None, data=''): if mode is None: mode = 0o750 if qtype & 0x80 else 0o640 pname, name = os.path.split(wname) parent = self.walk(pname) path = self.new_path inode = Inode(name, path, qtype=qtype, mode=mode, data=data) parent.add_child(inode) return inode def get_inode(self, path): return self.inodes[path] def set_inode(self, path, inode): self.inodes[path] = inode class Session: fid_table = {} def __init__(self, filesystem): self.filesystem = filesystem @property def root(self): return self.filesystem.inodes[0] def set_fid(self, fid, inode): self.fid_table[fid] = inode def get_fid(self, fid): return self.fid_table[fid] def create(self, name, mode): pass	5.97 KB	2025-05-25 12:26
📄 ipc.py import collections import multiprocessing from socket import socketpair from pyroute2.netlink.nlsocket import NetlinkSocket from pyroute2.plan9.client import Plan9ClientSocket IPCSocketPair = collections.namedtuple('IPCSocketPair', ('server', 'client')) class IPCSocket(NetlinkSocket): def setup_socket(self): # create socket pair sp = IPCSocketPair(*socketpair()) # start the server self.socket = sp self.p9server = multiprocessing.Process(target=self.ipc_server) self.p9server.daemon = True self.p9server.start() # create and init the client self.p9client = Plan9ClientSocket(use_socket=sp.client) self.p9client.init() return sp def ipc_server(self): raise NotImplementedError() def recv(self, buffersize, flags=0): ret = self.p9client.call( fid=self.p9client.fid('call'), fname='recv', kwarg={'buffersize': buffersize, 'flags': flags}, ) return ret['data'] def send(self, data, flags=0): return self.p9client.call( fid=self.p9client.fid('call'), fname='send', kwarg={'flags': flags}, data=data, ) def bind(self): return self.p9client.call(fid=self.p9client.fid('call'), fname='bind') def close(self): self.socket.client.close() self.socket.server.close() self.p9server.wait()	1.43 KB	2025-05-25 12:26
📄 plan9socket.py from pyroute2.netlink.core import CoreSocket from pyroute2.plan9 import Marshal9P class Plan9Socket(CoreSocket): def __init__(self, use_socket=None): super().__init__(use_socket=use_socket) self.marshal = Marshal9P() self.spec['tag_field'] = 'tag'	278 B	2025-05-25 12:26
📄 server.py import asyncio import json from pyroute2.netlink.core import AsyncCoreSocket, CoreMessageQueue from pyroute2.netlink.coredata import CoreConfig, CoreSocketSpec from pyroute2.plan9 import ( Marshal9P, Plan9Exit, Stat, Tattach, Tauth, Tcall, Tclunk, Tcreate, Topen, Tread, Tremove, Tstat, Tversion, Twalk, Twrite, Twstat, msg_rattach, msg_rcall, msg_rclunk, msg_rerror, msg_ropen, msg_rread, msg_rstat, msg_rversion, msg_rwalk, msg_rwrite, msg_rwstat, ) from pyroute2.plan9.filesystem import Filesystem, Session data = str(dir()) def get_exception_args(exc): args = [] if hasattr(exc, 'errno'): args.append(exc.errno) args.append(exc.strerror) return args def route(rtable, request, state): def decorator(f): rtable[request] = f return f return decorator class Plan9ServerProtocol(asyncio.Protocol): rtable = {} def __init__(self, on_con_lost, marshal, filesystem): self.transport = None self.session = None self.filesystem = filesystem self.marshal = marshal self.on_con_lost = on_con_lost @route(rtable, request=Tversion, state=(None,)) def t_version(self, req): m = msg_rversion() m['header']['tag'] = 0xFFFF m['msize'] = req['msize'] m['version'] = '9P2000' return m @route(rtable, request=Tauth, state=(Tversion,)) def t_auth(self, req): m = msg_rerror() m['ename'] = 'no authentication required' return m @route(rtable, request=Tattach, state=(Tauth,)) def t_attach(self, req): m = msg_rattach() root = self.session.filesystem.inodes[0] self.session.set_fid(req['fid'], root) m['qid'] = root.qid return m @route(rtable, request=Twalk, state=(Tattach,)) def t_walk(self, req): m = msg_rwalk() inode = self.session.get_fid(req['fid']) wqid = [] if len(req['wname']) == 0: self.session.set_fid(req['newfid'], inode) else: for name in req['wname']: if name == '.': continue elif name == '..': inode = inode.get_parent() else: inode = inode.get_child(name) wqid.append(inode.qid) m['wqid'] = wqid self.session.set_fid(req['newfid'], inode) return m @route(rtable, request=Tstat, state=(Twalk,)) def t_stat(self, req): m = msg_rstat() inode = self.session.get_fid(req['fid']) inode.sync() m['stat'] = inode.stat return m @route(rtable, request=Twstat, state=(Twalk,)) def t_wstat(self, req): m = msg_rwstat() return m @route(rtable, request=Topen, state=(Twalk, Tstat)) def t_open(self, req): m = msg_ropen() m['qid'] = self.session.get_fid(req['fid']).qid m['iounit'] = 8192 return m @route(rtable, request=Tcall, state=(Twalk, Topen, Tstat)) def t_call(self, req): m = msg_rcall() inode = self.session.get_fid(req['fid']) m['err'] = 255 if Tcall in inode.callbacks: m = inode.callbacks[Tcall](self.session, inode, req, m) return m @route(rtable, request=Twrite, state=(Topen,)) def t_write(self, req): m = msg_rwrite() inode = self.session.get_fid(req['fid']) if Twrite in inode.callbacks: return inode.callbacks[Twrite](self.session, inode, req, m) if inode.qid['type'] & 0x80: raise TypeError('can not call write() on dir') inode.data.seek(req['offset']) m['count'] = inode.data.write(req['data']) return m @route(rtable, request=Tread, state=(Topen,)) def t_read(self, req): m = msg_rread() inode = self.session.get_fid(req['fid']) if Tread in inode.callbacks: return inode.callbacks[Tread](self.session, inode, req, m) if inode.qid['type'] & 0x80: data = bytearray() offset = 0 for child in inode.children: offset = Stat.encode_into(data, offset, child.stat) data = data[req['offset'] : req['offset'] + req['count']] else: inode.data.seek(req['offset']) data = inode.data.read(req['count']) m['data'] = data return m @route(rtable, request=Tclunk, state=(Topen, Tstat, Twalk, Tread)) def t_clunk(self, req): return msg_rclunk() @route(rtable, request=Tcreate, state=(Twalk,)) def t_create(self, req): return self.permission_denied(req) @route(rtable, request=Tremove, state=(Twalk,)) def t_remove(self, req): return self.permission_denied(req) def permission_denied(self, req): r_message = msg_rerror() r_message['ename'] = 'permission denied' r_message['header']['tag'] = req['header']['tag'] return r_message def error(self, e, tag=0): r_message = msg_rerror() spec = { 'class': e.__class__.__name__, 'argv': get_exception_args(e), 'str': str(e), } r_message['ename'] = json.dumps(spec) r_message['header']['tag'] = tag r_message.encode() self.transport.write(r_message.data) def data_received(self, data): for t_message in self.marshal.parse(data): tag = t_message['header']['tag'] try: r_message = self.rtable[t_message['header']['type']]( self, t_message ) r_message['header']['tag'] = tag r_message.encode() except Plan9Exit as e: self.error(e, tag) self.transport.abort() self.transport.close() return except Exception as e: self.error(e, tag) return self.transport.write(r_message.data) def connection_made(self, transport): self.transport = transport self.session = Session(self.filesystem) class Plan9ServerSocket(AsyncCoreSocket): '''9p2000 server. Requires either an IP address to listen on, or an open `SOCK_STREAM` socket to operate. An IP example, suitable to establish IPC between processes in one network: .. testcode:: from pyroute2 import Plan9ClientSocket, Plan9ServerSocket address = ('localhost', 8149) p9server = Plan9ServerSocket(address=address) p9client = Plan9ClientSocket(address=address) Server/client running on a `socketpair()` suitable for internal API within one process, or between parent/child processes: .. testcode:: from socket import socketpair from pyroute2 import Plan9ClientSocket, Plan9ServerSocket server, client = socketpair() p9server = Plan9ServerSocket(use_socket=server) p9client = Plan9ClientSocket(use_socket=client) ''' def __init__(self, address=None, use_socket=None, use_event_loop=None): self.spec = CoreSocketSpec( CoreConfig( tag_field='tag', target='localhost', netns=None, address=address, use_socket=use_socket is not None, use_event_loop=use_event_loop is not None, ) ) self.filesystem = Filesystem() self.marshal = Marshal9P() super().__init__(use_socket=use_socket, use_event_loop=use_event_loop) def register_function( self, func, inode, loader=json.loads, dumper=lambda x: json.dumps(x).encode('utf-8'), ): '''Register a function to an file. The file usage: * `write()`: write arguments for the call as a json dictionary of keyword arguments to the file data. * `read()`: 1. if the arguments were written to the data, call the function and write the result to the file data 2. read the file data and return to the client * `call()`: protocol extension, `Tcall` = 80, `Rcall` = 81, make this in one turn. .. testcode:: :hide: from pyroute2.plan9 import Tcall, Rcall assert Tcall == 80 assert Rcall == 81 Registering a function: .. testcode:: def communicate(a, b): return a + b def example_register(): fd = p9server.filesystem.create('test_func') p9server.register_function(communicate, fd) Communication using Twrite/Tread: .. testcode:: import json async def example_write(): fid = await p9client.fid('test_func') await p9client.write( fid, json.dumps({"a": 17, "b": 25}) ) msg = await p9client.read(fid) response = json.loads(msg['data']) assert response == 42 Same, using a command line 9p client from plan9port:: $ echo '{"a": 17, "b": 25}' \| 9p -a localhost:8149 write test_func $ 9p -a localhost:8149 read test_func 42 And using a mounted file system via FUSE client from plan9port:: $ 9pfuse localhost:8149 mnt $ echo '{"a": 17, "b": 25}' >mnt/test_func $ cat mnt/test_func 42 And the same, but using Tcall: .. testcode:: async def example_call(): fid = await p9client.fid('test_func') response = await p9client.call(fid, argv=(17, 25)) assert response == 42 And finnaly run this code: .. testcode:: async def main(): server_task = await p9server.async_run() example_register() await p9client.start_session() await example_write() await example_call() server_task.cancel() asyncio.run(main()) ''' return inode.register_function(func, loader, dumper) async def setup_endpoint(self): if getattr(self.local, 'server', None) is not None: return self.local.msg_queue = CoreMessageQueue(event_loop=self.event_loop) if self.status['use_socket']: self.local.server = None self.local.transport, self.local.protocol = ( await self.event_loop.create_connection( lambda: Plan9ServerProtocol( self.connection_lost, self.marshal, self.filesystem ), sock=self.use_socket, ) ) else: self.local.transport = None self.local.protocol = None self.local.server = await self.event_loop.create_server( lambda: Plan9ServerProtocol( self.connection_lost, self.marshal, self.filesystem ), *self.status['address'] ) async def async_run(self): '''Return the server asyncio task. Using this task one can stop the server: .. testcode:: async def main(): server = Plan9ServerSocket(address=('localhost', 8149)) server_task = await server.async_run() # ... server is running here server_task.cancel() # ... server is stopped asyncio.run(main()) To forcefully close all client connections and stop the server immediately from a registered function, one can pass this task to the function, cancel it, and raise `Plan9Exit()` exception: .. testcode:: import functools from pyroute2.plan9 import Plan9Exit server_sock, client_sock = socketpair() def test_exit_func(context): if 'server_task' in context: context['server_task'].cancel() raise Plan9Exit('server stopped upon client request') return 'server starting, please wait' async def server(): p9server = Plan9ServerSocket(use_socket=server_sock) context = {} inode = p9server.filesystem.create('stop') p9server.register_function( functools.partial(test_exit_func, context), inode ) context['server_task'] = await p9server.async_run() try: await context['server_task'] except asyncio.exceptions.CancelledError: pass assert context['server_task'].cancelled() .. testcode:: :hide: async def client(): p9client = Plan9ClientSocket(use_socket=client_sock) await p9client.start_session() fid = await p9client.fid('stop') try: await p9client.call(fid) except Plan9Exit: pass async def main(): await asyncio.gather(server(), client()) asyncio.run(main()) ''' await self.setup_endpoint() if self.status['use_socket']: return self.protocol.on_con_lost else: return asyncio.create_task(self.local.server.serve_forever()) def run(self): '''A simple synchronous runner. Uses `event_loop.run_forever()`. ''' self.event_loop.create_task(self.async_run()) self.event_loop.run_forever()	13.66 KB	2025-05-25 12:26

Terminal