jumpserver_core
Posted 台灯不太亮
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了jumpserver_core相关的知识，希望对你有一定的参考价值。
一、标准版
#!/usr/bin/env python # -*- coding:utf-8 -*- """ 功能： 实现了跳板ssh功能 缺点： 1、用户的输入的cmd只有在回车后，才会被发送到远端 server；造成不能使用补全功能 2、每次回显消息时候，都会附带这执行该条消息的cmd """ import paramiko import sys import select tran = paramiko.Transport((‘192.168.10.206‘, 22,), ) # 基于创建transport 对象 tran.start_client() # 基于transport 创建一个SSH session 作为 client tran.auth_password(‘root‘, ‘redhat‘) # 向server 进行连接认证 chan = tran.open_session() # 向server请求一个新的会话Chanlen，可定义窗口大小，最大包大小，超时时间 chan.get_pty() # 基于该session 通道，向server请求一个伪终端 chan.invoke_shell() # 开启一个交互shell：通过伪终端 while True: read, write, error = select.select([chan, sys.stdin, ], [], [], 1)# 若chan 发生变化，表示远端server发送了数据过来 if chan in read: try: data = chan.recv(1024).decode() if len(data) == 0: # 链路异常，会受到空消息 print(‘\r\n *** END *** \r\n‘) break sys.stdout.write(data) sys.stdout.flush() except Exception as ex: print(ex)# 若stdin有变化，表示用户有cmd需要送给远端server if sys.stdin in read: inp = sys.stdin.readline() chan.sendall(inp) chan.close() tran.close()
二、适用版
#!/usr/bin/env python # -*- coding:utf-8 -*- """ 功能： 实现了跳板ssh功能 实现了Tab补全，实时发送cmd（每输入一个字符都会发送） 缺点： 体验不好 """ import paramiko import sys import select import termios import tty tran = paramiko.Transport((‘192.168.10.206‘, 22,), ) tran.start_client() tran.auth_password(‘root‘, ‘redhat‘) chan = tran.open_session() chan.get_pty() chan.invoke_shell() oldtty = termios.tcgetattr(sys.stdin) # 获取源终端属性 try: # 为当前终端设置新属性 tty.setraw(sys.stdin.fileno()) tty.setcbreak(sys.stdin.fileno()) chan.settimeout(0.0) while True: read, write, error = select.select([chan, sys.stdin, ], [], [], 1) # 接受消息 if chan in read: try: data = chan.recv(1024).decode() if len(data) == 0: print(‘\r\n *** END *** \r\n‘) break sys.stdout.write(data) sys.stdout.flush() except Exception as ex: print(ex) # 发送命令 if sys.stdin in read: inp = sys.stdin.read(1) # 注意这里一定是一个字符，若为readline则会卡主； chan.sendall(inp) finally: # 将终端恢复为原始属性 termios.tcsetattr(sys.stdin, termios.TCSADRAIN, oldtty) chan.close() tran.close()
三、豪华版
#!/usr/bin/env python # -*- coding:utf-8 -*- import os import sys import paramiko import select import getpass import termios import tty def pre_init(): # 获取用户名 default_username = getpass.getuser() username = input(‘username, defult is {}: ‘.format(default_username)).strip() if len(username) == 0: username = default_username # 获取目标服务器 hostname = input(‘hostname: ‘).strip() if len(hostname) == 0: print(‘hostname required‘.center(40, ‘*‘)) exit(1) return (username, hostname) class JumpClient: def __init__(self, username, hostname): self.username = username self.hostname = hostname self.tran = paramiko.Transport((self.hostname, 22)) self.tran.start_client() try: self.init_connection() except Exception as ex: print(ex) else: self.chan = self.tran.open_session() self.chan.get_pty() self.chan.invoke_shell() def init_connection(self): # 获取验证类型；默认为密码认证 default_auth_type = ‘p‘ auth_type = input(‘Auth type [password(p)/key(k)], defulat is [p]: ‘).strip() if len(auth_type) == 0: auth_type = default_auth_type # 载入私钥；判断位置、判断是否需要密码 if auth_type == ‘k‘: default_key = os.path.join(os.environ.get(‘HOME‘), ‘.ssh‘, ‘id_rsa‘) auth_key = input(‘Auth key path, defulat is {}: ‘.format(default_key)) if len(auth_key) == 0: auth_key = default_key # 若私钥没有密码，则直接载入；若有密码，则输入密码 try: key = paramiko.RSAKey.from_private_key_file(auth_key) except paramiko.PasswordRequiredException: passwd = getpass.getpass(‘RSA key\‘s passwd: ‘).strip() key = paramiko.RSAKey.from_private_key_file(auth_key, password=passwd) finally: self.tran.auth_publickey(self.username, key) # 链接服务器 if auth_type == ‘p‘: server_passwd = getpass.getpass(‘password: ‘).strip() self.tran.auth_password(self.username, server_passwd) def open_terminal(self): # 获取原先 tty终端属性 oldtty = termios.tcgetattr(sys.stdin) try: # 设置新的 tty终端属性 tty.setraw(sys.stdin.fileno()) tty.setcbreak(sys.stdin.fileno()) self.chan.settimeout(0.0) while True: r_lst, w_lst, e_lst = select.select([self.chan, sys.stdin, ], [], [], 1) if self.chan in r_lst: try: data = self.chan.recv(1024).decode() if len(data) == 0: print(‘\r\n*** EOF ***\r\n‘) break else: sys.stdout.write(data) # 不用print，因为末尾会自动追加换行 sys.stdout.flush() except Exception as ex: print(ex) if sys.stdin in r_lst: data = sys.stdin.read(1) # 读取一个字符就发送 self.chan.sendall(data) finally: # 恢复原先 tty终端属性 termios.tcsetattr(sys.stdin, termios.TCSADRAIN, oldtty) def close_terminal(self): self.chan.close() self.tran.close() if __name__ == ‘__main__‘: user, host = pre_init() obj = JumpClient(user, host) obj.open_terminal() obj.close_terminal()
四、至尊版
#!/usr/bin/env python # -*- coding:utf-8 -*- """ 支持的客户端：1、windows（不支持补全、字符立即发送） 2、linux（支持补全等，但vim工具有问题）""" import os import sys import paramiko import select import getpass import threading try: import termios import tty linux_client = True except ImportError: linux_client = False def pre_init(): # 获取目标服务器 hostname = input(‘hostname: ‘).strip() if len(hostname) == 0: print(‘hostname required‘.center(40, ‘*‘)) exit(1) # 获取用户名 default_username = getpass.getuser() username = input(‘username: ‘).strip() if len(username) == 0: username = default_username return (username, hostname) class JumpClient: def __init__(self, username, hostname): self.username = username self.hostname = hostname self.tran = paramiko.Transport((self.hostname, 22)) self.tran.start_client()# 判断，若为linux 则调用linux初始化方法；若为windows 则调用windows初始化方法 try: if linux_client: self.linux_init() else: self.windows_init() except Exception as ex: print(ex) else: self.chan = self.tran.open_session() self.chan.get_pty() self.chan.invoke_shell() def linux_init(self): # 获取验证类型；默认为密码认证 default_auth_type = ‘p‘ auth_type = input(‘Auth type [password(p)/key(k)], defulat is [p]: ‘).strip() if len(auth_type) == 0: auth_type = default_auth_type # 载入私钥；判断位置、判断是否需要密码 if auth_type == ‘k‘: default_key = os.path.join(os.environ.get(‘HOME‘), ‘.ssh‘, ‘id_rsa‘) auth_key = input(‘Auth key path, defulat is {}: ‘.format(default_key)) if len(auth_key) == 0: auth_key = default_key # 若私钥没有密码，则直接载入；若有密码，则输入密码 try: key = paramiko.RSAKey.from_private_key_file(auth_key) except paramiko.PasswordRequiredException: passwd = getpass.getpass(‘RSA key\‘s passwd: ‘).strip() key = paramiko.RSAKey.from_private_key_file(auth_key, password=passwd) finally: self.tran.auth_publickey(self.username, key) if auth_type == ‘p‘: server_passwd = getpass.getpass(‘password: ‘).strip() self.tran.auth_password(self.username, server_passwd) def linux_run(self): # 获取原先 tty终端属性 oldtty = termios.tcgetattr(sys.stdin) try: # 设置新的 tty终端属性 tty.setraw(sys.stdin.fileno()) tty.setcbreak(sys.stdin.fileno()) self.chan.settimeout(0.0) while True: r_lst, w_lst, e_lst = select.select([self.chan, sys.stdin, ], [], [], 1) if self.chan in r_lst: try: data = self.chan.recv(1024).decode() if len(data) == 0: print(‘\r\n*** EOF ***\r\n‘) break else: sys.stdout.write(data) # 不用print，因为末尾会自动追加换行 sys.stdout.flush() except Exception as ex: print(ex) if sys.stdin in r_lst: data = sys.stdin.read(1) # 读取一个字符就发送 self.chan.sendall(data) finally: # 恢复原先 tty终端属性 termios.tcsetattr(sys.stdin, termios.TCSADRAIN, oldtty) def windows_init(self): user_pass = input(‘password: ‘).strip() self.tran.auth_password(self.username, user_pass) def windows_run(self): # 子线程的任务：从传入的socket对象中，不断接受数据（如果有数据的话）并打印出来 def function(sk): while True: try: ret = sk.recv(1024).decode() if len(ret) == 0: break else: sys.stdout.write(ret) sys.stdout.flush() except Exception as ex: print(ex) # 接受数据：开启一个子线程，并指定任务 t = threading.Thread(target=function, args=(self.chan,)) t.start() # 发送数据：主线程本身去完成 try: while True: cmd = sys.stdin.read(1) self.chan.send(cmd) except Exception as ex: print(ex) def close_terminal(self): self.chan.close() self.tran.close()if __name__ == ‘__main__‘: user, host = pre_init() # 获取要登录的目标服务器 obj = JumpClient(user, host) if linux_client: obj.linux_run() else: obj.windows_run()# 关闭链接 obj.close_terminal()
以上是关于jumpserver_core的主要内容，如果未能解决你的问题，请参考以下文章