This commit is contained in:
yumoqing 2024-10-10 10:24:48 +08:00
parent 82dacf619e
commit 1d874946a3
4 changed files with 109 additions and 25 deletions

View File

@ -6,7 +6,7 @@ import codecs
from urllib.parse import quote from urllib.parse import quote
import json import json
import asyncio import asyncio
from traceback import fromat_exc from traceback import format_exc
import random import random
import time import time

View File

@ -1,5 +1,6 @@
import time import time
import uuid import uuid
from traceback import print_exc
from aiohttp_auth import auth from aiohttp_auth import auth
from aiohttp_auth.auth.ticket_auth import TktAuthentication from aiohttp_auth.auth.ticket_auth import TktAuthentication
from aiohttp_session.redis_storage import RedisStorage from aiohttp_session.redis_storage import RedisStorage
@ -150,6 +151,8 @@ class AuthAPI:
except Exception as e: except Exception as e:
t3 = time.time() t3 = time.time()
info(f'Exception=client({ip}) {user} access {path} cost {t3-t1}, ({t2-t1}), except={e}') info(f'Exception=client({ip}) {user} access {path} cost {t3-t1}, ({t2-t1}), except={e}')
print(f'Exception=client({ip}) {user} access {path} cost {t3-t1}, ({t2-t1}), except={e}')
print_exc()
raise e raise e
if user is None: if user is None:

View File

@ -6,6 +6,7 @@ import codecs
from urllib.parse import quote from urllib.parse import quote
import json import json
import asyncio import asyncio
from traceback import fromat_exc
import random import random
import time import time
@ -62,6 +63,10 @@ async def save_file(str_or_bytes, filename):
r = await fs.save(filename, str_or_bytes) r = await fs.save(filename, str_or_bytes)
return r return r
def realpath(path):
fs = FileStorage()
return fs.realPath(path)
class FileOutZone(Exception): class FileOutZone(Exception):
def __init__(self,fp,*args,**kwargs): def __init__(self,fp,*args,**kwargs):
super(FileOutZone,self).__init__(*args,**kwargs) super(FileOutZone,self).__init__(*args,**kwargs)
@ -199,6 +204,8 @@ def initEnv():
g.async_sleep = asyncio.sleep g.async_sleep = asyncio.sleep
g.quotedstr = quotedstr g.quotedstr = quotedstr
g.save_file = save_file g.save_file = save_file
g.realpath = realpath
g.format_exc = format_exc
def set_builtins(): def set_builtins():
all_builtins = [ i for i in dir(builtins) if not i.startswith('_')] all_builtins = [ i for i in dir(builtins) if not i.startswith('_')]

View File

@ -7,6 +7,8 @@ from aiohttp import web
import aiohttp_cors import aiohttp_cors
from traceback import print_exc from traceback import print_exc
from appPublic.sshx import SSHNode from appPublic.sshx import SSHNode
from appPublic.dictObject import DictObject
from appPublic.log import info, debug, warning, error, exception, critical
from .baseProcessor import BaseProcessor, PythonScriptProcessor from .baseProcessor import BaseProcessor, PythonScriptProcessor
class XtermProcessor(PythonScriptProcessor): class XtermProcessor(PythonScriptProcessor):
@ -81,8 +83,7 @@ class XtermProcessor(PythonScriptProcessor):
self.p_obj.close() self.p_obj.close()
async def ws_send(ws:web.WebSocketResponse, data): async def ws_send(ws:web.WebSocketResponse, data):
if not isinstance(data, str): info(f'data={data} {ws=}')
data = json.dumps(data)
d = { d = {
"type":1, "type":1,
"data":data "data":data
@ -91,54 +92,121 @@ async def ws_send(ws:web.WebSocketResponse, data):
try: try:
return await ws.send_str(d) return await ws.send_str(d)
except Exception as e: except Exception as e:
print('ws.send_str() error:', e) exception(f'ws.send_str() error: {e=}')
print_exc()
return False return False
class WsSession:
def __init__(self, session):
self.session = session
self.nodes = {}
def join(node):
self.nodes[node.id] = node
def leave(node):
self.nodes = {k:v for k,v in self.nodes.items() if k != node.id}
class WsData:
def __init__(self):
self.nodes = {}
self.sessions = {}
def add_node(self, node):
self.nodes[node.id] = node
def del_node(self, node):
self.nodes = {k:v for k,v in self.nodes.items() if k!=node.id}
def get_nodes(self):
return self.nodes
def get_node(self, id):
return self.nodes.get(id)
def add_session(self, session):
self.sessions[session.sessionid] = session
def del_session(self, session):
self.sessions = {k:v for k,v in self.sessions.items() if k != session.sessionid}
def get_session(self, id):
return self.sessions.get(id)
class WsPool: class WsPool:
def __init__(self, ws, ws_path, app): def __init__(self, ws, ip, ws_path, app):
self.app = app self.app = app
self.ip = ip
self.id = None self.id = None
self.ws = ws self.ws = ws
print(f'-----------{self.id=}-----------')
self.ws_path = ws_path self.ws_path = ws_path
def get_data(self): def get_data(self):
return self.app.get_data(self.ws_path) r = self.app.get_data(self.ws_path)
if r is None:
r = WsData()
self.set_data(r)
return r
def set_data(self, data): def set_data(self, data):
self.app.set_data(self.ws_path, data) self.app.set_data(self.ws_path, data)
def is_online(self, userid): def is_online(self, userid):
data = self.get_data() data = self.get_data()
ws = data.get(userid) node = data.get_node(userid)
if ws is None: if node is None:
return False return False
return True return True
def register(self, id): def register(self, id):
self.id = id iddata = DictObject()
self.add_mine() iddata.id = id
self.add_me(iddata)
def add_mine(self): def add_me(self, iddata):
data = self.get_data() data = self.get_data()
if data is None: if data is None:
data = {} data = {}
data.update({self.id:self.ws}) iddata.ws = self.ws
iddata.ip = self.ip
self.id = iddata.id
data.add_node(ddata)
self.set_data(data) self.set_data(data)
def delete_mine(self): def delete_id(self, id):
data = self.get_data() data = self.get_data()
if not data.get(self.id): node = data.get_node(id)
return if node:
data = {k:v for k,v in self.data.items() if k != self.id } data.del_node(node)
self.set_data(data) self.set_data(data)
def delete_me(self):
self.delete_id(self.id)
def add_session(self, session):
data = self.get_data()
data.add_session(session)
self.set_data(data)
def del_session(self, session):
data = self.get_data()
data.del_session(session)
self.set_data(data)
def get_session(self, sessionid):
data = self.get_data()
return data.get_session(sessionid)
async def sendto(self, data, id=None): async def sendto(self, data, id=None):
if id is None: if id is None:
return await ws_send(self.ws, data) return await ws_send(self.ws, data)
d = self.get_data() d = self.get_data()
ws = d.get(id) iddata = d.get_node(id)
return await ws_send(ws, data) ws = iddata.ws
try:
return await ws_send(ws, data)
except:
self.delete_id(id)
class WebsocketProcessor(PythonScriptProcessor): class WebsocketProcessor(PythonScriptProcessor):
@classmethod @classmethod
@ -151,32 +219,38 @@ class WebsocketProcessor(PythonScriptProcessor):
lenv.update(params) lenv.update(params)
params_kw = lenv.params_kw params_kw = lenv.params_kw
userid = lenv.params_kw.userid or await lenv.get_user() userid = lenv.params_kw.userid or await lenv.get_user()
debug(f'========== debug ===========')
del lenv['request'] del lenv['request']
txt = await self.loadScript(self.real_path) txt = await self.loadScript(self.real_path)
ws = web.WebSocketResponse() ws = web.WebSocketResponse()
debug(f'========== debug ===========')
try: try:
await ws.prepare(request) await ws.prepare(request)
except Exception as e: except Exception as e:
print('--------except:', e, type(e)) exception(f'--------except: {e}')
print_exc() print_exc()
raise e raise e
ws_pool = WsPool(ws, request.path, request.app) debug(f'========== debug ===========')
ws_pool = WsPool(ws, request['client_ip'], request.path, request.app)
debug(f'========== debug ===========')
async for msg in ws: async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT: if msg.type == aiohttp.WSMsgType.TEXT:
debug(f'========== {msg=} ===========')
if msg.data == 'exit': if msg.data == 'exit':
break break
debug(f'WS:msg from client={msg}')
lenv['ws_data'] = msg.data lenv['ws_data'] = msg.data
lenv['ws_pool'] = ws_pool lenv['ws_pool'] = ws_pool
exec(txt,lenv,lenv) exec(txt,lenv,lenv)
func = lenv['myfunc'] func = lenv['myfunc']
resp = await func(request,**lenv) resp = await func(request,**lenv)
# await self.ws_sendstr(ws, resp)
elif msg.type == aiohttp.WSMsgType.ERROR: elif msg.type == aiohttp.WSMsgType.ERROR:
# print('ws connection closed with exception %s' % ws.exception()) error('ws connection closed with exception %s' % ws.exception())
break break
else: else:
print('datatype error', msg.type) info('datatype error', msg.type)
ws_pool.delete_mine() debug(f'========== ws connection end ===========')
ws_pool.delete_me()
self.retResponse = ws self.retResponse = ws
await ws.close() await ws.close()
return ws return ws