• 1. 使用模块介绍
• 1.1 使用方式
• 1.1.1 `spawn`类
• 1.1.2 `expect`方法
• 1.1.3 `send`方法
• 实现远程 ssh 连接到目标主机
• 简易跳板机的实现

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

在写这个小工具之前，我就在想一定要非常简单，足够简单，所以这里用了不到 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')

提示如下：

nbsp;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