乐趣区

关于运维:不到200行代码使用python写一个超简单的跳板机工具

jiezi

[toc]

在日常工作中，少不了须要上服务器执行一些操作或者命令，即便在当初云时代，然而一旦呈现问题，还是须要上机排查，所以写了一个小小的跳板机工具

在写这个小工具之前，我就在想肯定要非常简单，足够简略，所以这里用了不到 200 行代码，写了一个，能够说是非常简单。

1. 应用模块介绍

  • pexpect

就这一个就完了！

简略介绍一下这个模块：
pexpect 是 expect 的一个 python 实现，用来进行人机交互，比方执行某一段程序时要求输出用户名和明码，或者是 yes/no 这种状况时，应用 pexpect 便能够通过捕捉这种关键字进行输出一些必要信息来持续操作程序。

pexpect 的应用范畴很广，能够用来实现与 ssh、ftp、telnet 等程序的主动交互

1.1 应用形式

pexpect 的应用基本上分为三个步骤:

  1. 首先用 spawn 来执行一条命令或者一段程序
  2. 而后 expect 捕捉关键字
  3. 捕捉到指定关键字后，执行 send 指令发送必要的内容来持续操作程序

1.1.1 spawn类

spawn是 pexpect 的一个次要类，用来执行一个程序，并返回一个操作这个程序的句柄，前面的一系列操作都能够通过这个句柄来实现，上面是它的构造函数的定义：

class spawn(command, args=[], timeout=30, maxread=2000,
                 searchwindowsize=None, logfile=None, cwd=None,env=None,
                 ignore_sighup=False, echo=True, preexec_fn=None,
                 encoding=None, codec_errors='strict', dimensions=None,
                 use_poll=False)
  • command是任意的命令
child = pexpect.spawn('/usr/bin/ftp')
child = pexpect.spawn('/usr/bin/ssh user@example.com')
child = pexpect.spawn('ls -latr /tmp')

然而当蕴含一些特殊字符时(>, |, or *)，就必须启动一个 shell 来执行了，例如：

child = pexpect.spawn('/bin/bash -c"ls -l | grep LOG > logs.txt"')
child.expect(pexpect.EOF)

也能够这样写，定一个变量，将要执行的命令通过这个变量来接管

shell_cmd = 'ls -l | grep LOG > logs.txt'
child = pexpect.spawn('/bin/bash', ['-c', shell_cmd])
child.expect(pexpect.EOF)
  • args=[] 执行程序时传入所需的参数
child = pexpect.spawn('/usr/bin/ftp', [])
child = pexpect.spawn('/usr/bin/ssh', ['user@example.com'])
child = pexpect.spawn('ls', ['-latr', '/tmp'])
  • timeout=30 设置超时工夫
  • maxread=2000 pexpect 从终端控制台一次读取的最大字节数
  • searchwindowsize 匹配缓冲区字符串的地位，默认是从开始地位匹配

不过有时候须要打印下执行的后果，也就是将输入打印到规范输入里，形式如下：

import pexpect
import sys

child = pexpect.spawn("df -h", encoding='utf-8')
child.logfile = sys.stdout
child.expect(pexpect.EOF)

必须要加上 encoding='utf-8'，否则会报TypeError: write() argument must be str, not bytes
当然，也能够应用 logfile 这个参数，例如：

child = pexpect.spawn("df -h", logfile=sys.stdout, encoding='utf-8')
child.expect(pexpect.EOF)

1.1.2 expect办法

expect最初会返回 0 示意匹配到了所需的关键字，如果定义了一个关键字列表的话，就会返回一个数字示意匹配到了列表中第几个关键字，从 0 开始计算，也就是该关键字所在的索引号

expect(pattern, timeout=-1, searchwindowsize=-1, async_=False, **kw)

pattern 能够是 StringType, EOF，正则表达式，或者这些类型的列表
如果当 pattern 是一个列表时，且列表里有多个可被匹配到的值时，最初只能返回第一个被匹配到的索引，例如：

child = pexpect.spawn("echo'hello world'", logfile=sys.stdout, encoding='utf-8')
index = child.expect(['hahaha', 'hello', 'hello world'])
print(index)  # 后果为 1，'hello' 的索引编号

留神：这里肯定要晓得, 是拿 pattern 里的内容来匹配 spawn 中的关键字

1.1.3 send办法

send就是发送关键字给程序，这里举一个示例，写一个简略的 shell 脚本, 这个脚本会接管一个 username 的参数，并把该变量的值写道一个文件里

# test.sh
#!/bin/bash
read -p "Input your name:" username
echo "$username" > name.txt

而后通过 expect 进行捕捉关键字，最初应用 send 发送关键字

child = pexpect.spawn("sh test.sh", logfile=sys.stdout, encoding='utf-8')
index = child.expect("Input your name")
if index == 0:
    child.send("dogfei")  # 执行完还要再敲一次回车

下面执行完之后，还要再敲下回车，如果想不必敲回车，能够应用 sendline
后果当前目录下会有一个 name.txt 文件，并且外面的内容就是刚传进去的变量的值

ok, 说了这么多，基本上对 pexpect 这个模块有了一个理解，那么接下来就直奔主题！

实现近程 ssh 连贯到指标主机

既然是跳板机，那必定是通过执行 ssh 来近程登录，那么就要有服务器地址、用户名、明码、端口等信息，而后近程登录的时候，个别会有几种状况，如下：

  • 第一次 ssh 连贯时，个别会提醒：Are you sure you want to continue connecting (yes/no)，要求输出yes/no
  • 如果 ssh 连贯时，ssh 的端口不正确，个别会提醒：Connection refused
  • 如果网络问题，个别会连贯超时
  • 如果之前 ssh 连贯过，那再次连贯时，就会提醒：password:，要求输出明码
  • 输出明码且明码正确，个别会提醒：Last login
  • 明码谬误，个别会提醒：Permission denied, please try again

晓得了，这些，那么咱们在写的时候，就比拟容易了。

import pexpect
import os

def run_cmd(cmd, patterns):
    child = pexpect.spawn(cmd, encoding='utf-8')
    child.setwinsize(lines, columns)
    index = child.expect(patterns, timeout=10)
    return [index, child]

该函数返回一个列表，别离是捕捉关键字的索引编号，操作程序的一个句柄，而后上面是针对近程 ssh 时的各种状况，用到了一个递归，这样防止应用嵌套 while 循环

def sshclient(host, user, port, passwd):
    ssh_newkey = "continue"
    ssh_passwd = "assword:"
    ssh_confirm = "yes"
    ssh_refuse = "Connection refused"
    ssh_login = "Last login:"
    ssh_repeat_passwd = "Permission denied, please try again"
    ssh_noroutetohost = "No route to host"
    ssh_conntimeout = "Connection timed out"
    # 近程 ssh 时的残缺命令
    ssh_cmd = "ssh {u}@{h} -p {p}".format(u=user, h=host, p=port)
    # 初始化一个句柄，并获取索引号
    index, child = run_cmd(ssh_cmd, [
        ssh_newkey,
        ssh_passwd,
        ssh_refuse,
        ssh_login,
        ssh_noroutetohost,
        ssh_conntimeout,
        pexpect.EOF,
        pexpect.TIMEOUT])
    try:
        if index == 0:
            child.sendline(ssh_confirm)
            # 个别第一次 ssh 时，会让你输出 yes/no 之类的，所以匹配到这个的时候，就做一次递归
            return sshclient(host, user, port, passwd)
        elif index == 1:
            print("Begin Load Password...")
            child.sendline(passwd)
            result = child.expect([
                ssh_repeat_passwd,
                ssh_login,
            ])
            if result == 1:
                print("{} login success (-_-)".format(host))
                child.interact()
                return
            elif result == 0:
                # 阐明明码谬误，须要从新输出明码，并进行递归
                passwd = input('Passwd:').strip()
                return sshclient(host, user, port, passwd)
        elif index == 2:
            print("Connect refused, Pls check ssh port.")
            return
        elif index == 3:
            print("Login success")
            child.interact()
            return
        elif index == 4:
            print("The host %s connected faild: No route to host" % host)
            return
        elif index == 5:
            print("The host %s connected faild: Connection timeout" % host)
            return
        elif index == 6:
            print("Abnormal exit")
            return
        elif index == 7:
            print("Timeout for connect host %s, pls check network" % host)
            return
        return
    except Exception as e:
        raise e

到这里，咱们就能应用这个程序来进行近程操作了，拿本机做下测试：

if __name__ == "__main__":
    sshclient('127.0.0.1','dogfei',22, '123456')

提醒如下：

$ python3 test_jp.py
Begin Load Password...
127.0.0.1 login success (-_-)
 Sun Jun 20 19:43:32 2021 from 127.0.0.1

但这里仅实现了能近程连贯到近程主机，可是跳板机的话，是有很多台机器，并且有很多类型，每台机器都有本人的编号，所以咱们要实现这样的性能，淦！

繁难跳板机的实现

既然有很多主机，这些主机又划分了很多类型，也就是标签之类的，还有可能每台主机的明码都不一样，又或者雷同类型的机器明码一样，又或者不能应用 root 登录等等等等，所以咱们肯定要做一个既简略又灵便的，机器明码不统一和用户名不统一这个能够通过定义一个残缺的登录命令来解决，而对主机划分类型、设置明码（默认明码），用户名这些信息，就要通过一个非常灵活的数据库表构造来实现了，这里我通过一个本地的配置文件来解决，形式如下：

global:
  user: root
  port: 22
  passwd: 123456
jumpserver:
  - name: k8s
    hostList:
      - 192.168.1.1
      - 192.168.1.2
      - 192.168.1.3

该配置文件有一个全局的配置，global 下的为全局配置，如果 jumpserver 局部中没有指定，则会拿全局配置，如果所有的机器用户名、明码、端口都一样，那么就能够这样写.

如果用户名，明码不一样，那就能够这样写：

global:
  user: root
  port: 22
  passwd: 123456
jumpserver:
  - name: k8s
    hostList:
      - 192.168.1.1
      - 192.168.1.2
      - 192.168.1.3
  - name: mysql
    hostList:
      - host: 192.168.1.4
        user: dogfei
      - host: 192.168.1.5
        user: db
        passwd: 111111

通过这样设计当前，那么咱们的代码就变成了这个样子：

import yaml

def parseYaml(yamlfile, parse_list=None):
    if parse_list is None:
        parse_list = []
    with open(yamlfile, 'r', encoding='utf-8') as fr:
        yaml_to_dict = yaml.safe_load(fr)
        global_user = yaml_to_dict['global']['user']
        global_passwd = yaml_to_dict['global']['passwd']
        global_port = int(yaml_to_dict['global']['port'])
        for detail in yaml_to_dict['jumpserver']:
            tag = detail['name']
            get_hostList = detail['hostList']
            if isinstance(get_hostList[0], dict):
                for ssh in get_hostList:
                    sshDetail = {
                        'tag': tag,
                        'host': ssh['host'],
                        'user': ssh['user'] if 'user' in ssh else global_user,
                        'port': int(ssh['port']) if 'port' in ssh else global_port,
                        'passwd': ssh['passwd'] if 'passwd' in ssh else global_passwd
                    }
                    parse_list.append(sshDetail)
            elif isinstance(get_hostList[0], str):
                for h in get_hostList:
                    sshDetail = {
                        'tag': tag,
                        'host': h,
                        'user': global_user,
                        'port': global_port,
                        'passwd': global_passwd
                    }
                    parse_list.append(sshDetail)
        return parse_list

if __name__ == '__main__':
    print(parseYaml('ip.yaml'))

最终返回的是一个列表，列表里的元素是一个字典，如下所示：

[{'tag': 'k8s', 'host': '192.168.1.1', 'user': 'root', 'port': 22, 'passwd': 123456},
    {'tag': 'k8s', 'host': '192.168.1.2', 'user': 'root', 'port': 22, 'passwd': 123456},
    {'tag': 'k8s', 'host': '192.168.1.3', 'user': 'root', 'port': 22, 'passwd': 123456},
    {'tag': 'mysql', 'host': '192.168.1.4', 'user': 'dogfei', 'port': 22, 'passwd': 123456},
    {'tag': 'mysql', 'host': '192.168.1.5', 'user': 'db', 'port': 22, 'passwd': 111111}
]

拿到这一堆数据字后，就能够进行丑化了，如下所示：

def list_info(originList):
    try:
        print("******\033[1;30;43mIP 信息如下，请抉择对应的编号进行登陆 \033[0m******\n")
        print("\033[0;32m{:<5}\033[0m{:<19}{}".format("编号", "IP 地址", "标签"))
        sshList = []
        sshDict = {}
        for info in originList:
            id = originList.index(info) + 1
            host = info['host']
            tag = info['tag']
            user = info['user']
            port = int(info['port'])
            passwd = info['passwd']
            sshDict[id] = [host, user, passwd, port]
            print("{:<5}{:<22}{}".format(id, host, tag))
        return sshDict
    except Exception as e:
        raise e

这段代码的执行后果如图所示：

而后上面就是一堆循环，次要是实现打印主机信息、退出、进入主机等性能，如下所示：

def login(yamlfile):
    try:
        print("\033[1;30;47m{:^50}\033[0m\n".format("繁难跳板机"))
        outer_flag = False
        while not outer_flag:
            print("\033[5;35;46m{:<}\033[0m\n".format("请抉择"))
            print("\033[0;32m 输出'p/P'打印所有主机信息 \033[0m")
            print("\033[0;31m 输出'q/quit'退出 \033[0m\n")
            input_x = input(">>>>>:").strip().lower()
            if input_x == 'p':
                os.system("clear")
                ip_info = list_info(yamlfile)
                print("\n")
                print("\033[0;32m 输出' 编号 '进入对应的服务器 \033[0m")
                print("\033[0;32m 输出'q'退出 \033[0m")
                print("\033[0;32m 输出'b'返回 \033[0m")
                inner_flag = False
                while not inner_flag:
                    act = input("\033[0;32m>>>>>: \033[0m").strip().lower()
                    if act.isdigit():
                        ip_id = int(act)
                        if ip_id in ip_info.keys():
                            host = ip_info[ip_id][0]
                            user = ip_info[ip_id][1]
                            passwd = str(ip_info[ip_id][2])
                            port = int(ip_info[ip_id][3])
                            sshclient(
                                host=host,
                                user=user,
                                port=port,
                                passwd=passwd
                            )
                            inner_flag = True
                        else:
                            print("\033[0;31m 编号不存在，请从新输出，退出请输出任意字符 \033[0m")
                            continue
                    else:
                        if act == 'q' or act == 'quit':
                            print("\033[0;31m 告辞 !!!\033[0m")
                            inner_flag = True
                            outer_flag = True
                        elif act == 'b' or act == 'back':
                            inner_flag = True
            elif input_x == 'q' or input_x == 'quit':
                print("\033[0;31m 告辞 !!!\033[0m")
                outer_flag = True
            else:
                print("\033[0;31m 请输出指定的内容 !!!\033[0m")
                continue
    except Exception as e:
        raise e

而后上面看下效果图：

上面是动静演示：

总共加上空行，一共 185 行，真是太实用了啊！

残缺代码可去我的集体博客或者关注公众号进行查看：https://www.dogfei.cn

欢送各位朋友关注我的公众号，来一起学习提高哦

退出移动版