apppublic/appPublic/sshx.py
2023-12-17 12:21:17 +08:00

333 lines
7.8 KiB
Python

import os
import sys
import time
import shlex
from functools import partial
from threading import Thread
from appPublic.myTE import tmpTml
import asyncio, asyncssh, sys
class SSHNode:
def __init__(self, host,
username='root',
port=22,
password=None,
jumpers=[]):
self.server2 = {
"host":host,
"username":username,
"password":password,
"port":port
}
print(self.server2)
self.jumpers = jumpers
self.conn = None
self.jumper_conns = []
self.batch_cmds = []
def info(self):
d = {
"jumpers":self.jumpers,
}
d.update(self.server2)
return d
def asjumper(self):
a = self.jumpers.copy()
a.append(self.server2)
return a
def set_jumpers(self, jumpers):
self.jumpers = jumpers
async def connect(self):
refconn = None
for j in self.jumpers:
host = j['host']
username = j.get('username', 'root')
port = j.get('port',22)
password= j.get('password', None)
if refconn:
refconn = await refconn.connect_ssh(host,
username=username,
password=password,
port=port)
else:
refconn = await asyncssh.connect(host,
username=username,
password=password,
port=port)
self.jumper_conns.append(refconn)
host = self.server2['host']
username = self.server2.get('username', 'root')
port = self.server2.get('port',22)
password = self.server2.get('password', None)
if refconn:
self.conn = await refconn.connect_ssh(host,
username=username,
port=port,
password=password,
known_hosts=None)
else:
self.conn = await asyncssh.connect(host,
username=username,
port=port)
def close(self):
self.conn.close()
cnt = len(self.jumper_conns)
cnt -= 1
while cnt >= 0:
self.jumper_conns[cnt].close()
cnt -= 1
self.jumper_conns = []
self.conn = None
async def _l2r(self, lf, rf):
x = await asyncssh.scp(lf, (self.conn, rf),
preserve=True, recurse=True)
return x
async def _process(self, *args, **kw):
a = await self.conn.create_process(*args, **kw)
return a
async def _r2l(self, rf, lf):
x = await asyncssh.scp((self.conn, rf), lf,
preserve=True, recurse=True)
return x
async def _cmd(self, cmd, input=None, stdin=None, stdout=None):
return await self.conn.run(cmd, input=input, stdin=stdin, stdout=stdout)
async def _xcmd(self, cmd, xmsgs=[], ns={},
show_input=None,
show_stdout=None):
proc = await self._process(cmd, term_type='xterm',
term_size=(80,24),
encoding='utf-8'
)
keyin = False
def feed_data(xmsgs, debug_input):
if len(xmsgs) == 0:
print('#####++##### xmsgs has zero elements')
return
keyin = True
a = xmsgs.pop(0)
while True:
if a[1] is None:
proc.stdin.write_eof()
self.running = False
else:
s = a[1].format(**ns)
proc.stdin.write(s)
if len(xmsgs) == 0 or xmsgs[0][0]:
break
a = xmsgs.pop(0)
already_output = False
callee = None
loop = asyncio.get_event_loop()
self.running = True
while self.running:
if keyin:
keyin = False
await proc.stdin.drain()
if proc.stdout.at_eof():
break
tup = proc.collect_output()
x = tup[0]
if x!='' and show_stdout:
if x is None:
break
if callee:
callee.cancel()
callee = None
show_stdout(x)
else:
if callee is None:
if len(xmsgs) > 0:
f = partial(feed_data, xmsgs, show_input)
t = xmsgs[0][0] or 0
callee = loop.call_later(t, f)
await asyncio.sleep(0.05)
print('##########fininshed##########')
async def _run(self, cmd, input=None, stdin=None, stdout=None):
if cmd.startswith('l2r'):
args = shlex.split(cmd)
if len(args) == 3:
x = await self._l2r(args[1], args[2])
return x
if cmd.startswith('r2l'):
args = shlex.split(cmd)
if len(args) == 3:
x = await self._r2l(args[1], args[2])
return x
return await self._cmd(cmd, input=input, stdin=stdin, stdout=stdout)
def show_result(self, x):
if isinstance(x, Exception):
print('Exception:',e)
else:
print('stdout:', x.stdout)
print('stderr:', x.stderr)
async def run(self, cmd, input=None, stdin=None, stdout=None):
await self.connect()
result = await self._run(cmd, input=input,
stdin=stdin, stdout=stdout)
self.close()
return result
class SSHNodes:
def __init__(self, nodes, usernmae='root', port=22, jumpers=[]):
self.nodes = [ Node(n, username=username, port=port, jumpers=jumpers) for n in nodes ]
self.batch_cmds = []
def append_cmd(self, cmd, stdin=None, stdout=None):
self.batch_cmds.append({
"cmd":cmd,
"stdin":stdin,
"stdout":stdout})
def show_result(self, result, i=0):
if isinstance(result, Exception):
print(f'Task {i} failed:{result}')
elif result.exit_status != 0:
print(f'Task {i} exit {result.exit_status}')
print(result.stderr, end='')
else:
print(f'Task {i} successed:')
print(result.stdout, end='')
async def run(self, cmd, stdin=None, stdout=None):
tasks = [ n.run(cmd, stdin=stdin, stdout=stdout) for n in self.nodes ]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
async def exe_batch(self):
tasks = [ n.exe_batch(self.batch_cmds) for n in self.nodes ]
results = await asyncio.gather(*tasks, return_excetion=True)
return results
for i, result in enumerate(results):
self.show_result(result,i)
async def main():
if len(sys.argv) < 3:
print(f'{sys.argv[0]} cmd host1 host2 ....')
sys.exit(1)
cmd = sys.argv[1]
jumpor = {
"host":"glib.cc",
"username":"ceni",
"port":10022
}
hosts = sys.argv[2:]
mn = SSHNodes(hosts, jumpers=[jumpor])
while True:
print('input command:')
cmd = input()
print('input stdin:')
stdin = input()
if stdin == '':
stdin = None
print('input stdout:(default is stdout)')
stdout = input()
if stdout == '':
stdout = None
x = await mn.run(cmd, stdin=stdin, stdout=stdout)
for r in x:
if isinstance(r, Exception):
print(r)
else:
print(r.stdout)
class SSHBash:
def __init__(self, node, loop=None):
if loop is None:
loop = asyncio.get_event_loop()
self.node = node
self.loop = loop
self.conn = None
self.stdin_need = False
self.subloop = asyncio.new_event_loop()
self.subthread = Thread(target=self.start_thread_loop)
self.subthread.setDaemon(True)
self.subthread.start()
def start_thread_loop(self):
asyncio.set_event_loop(self.subloop)
self.subloop.run_forever()
def exit(self):
if self.conn:
self.node.close(self.conn)
self.p_obj.close()
self.subloop.stop()
self.loop.stop()
async def feed_stdin(self, f):
self.stdin_need = False
x = await f(65535)
if x is None:
self.exit()
self.p_obj.stdin.write(x)
await self.p_obj.stdin.drain()
self.stdin_need = True
async def run(self, read_co, write_co):
await self.node.connect()
self.p_obj = await self.node._process('bash',
term_type='vt100',
term_size=(80,24),
encoding=None)
if isinstance(self.p_obj, Exception):
print('Excetion:', self.p_obj)
self.exit()
return
if self.p_obj is None:
print('self.p_obj is None')
self.exit()
return
# self.loop.add_reader(sys.stdin.fileno(), self.read_input)
self.stdin_need = True
while True:
if self.stdin_need:
asyncio.run_coroutine_threadsafe(self.feed_stdin(read_co), self.subloop)
if self.p_obj.stdout.at_eof():
self.exit()
break
x = await self.p_obj.stdout.read(1024)
await write_co(x)
if __name__ == '__main__':
async def sysstdin_read():
return os.read(sys.stdin.fileno(), 65535)
async def sysstdout_write(x):
sys.stdout.write(x.decode('utf-8'))
async def test_sshbash():
jp = {
"host":"glib.cc",
"username":"ceni",
"port":10022
}
jn = SSHNode('k3', jumpers=[jp])
bash = SSHBash(jn)
await bash.run(sysstdin_read, sysstdout_write)
loop = asyncio.get_event_loop()
loop.run_until_complete(test_sshbash())