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的主要内容,如果未能解决你的问题,请参考以下文章

VSCode自定义代码片段——CSS选择器

谷歌浏览器调试jsp 引入代码片段,如何调试代码片段中的js

片段和活动之间的核心区别是啥?哪些代码可以写成片段?

VSCode自定义代码片段——.vue文件的模板

VSCode自定义代码片段6——CSS选择器

VSCode自定义代码片段——声明函数