使用python的paramiko来实现批量主机管理

python

python第三方模块paramiko，很多人都知道fabric这个linux强大的主机批量管理工具，其实底层fabric就是用到了paramiko
这边我用paramiko来实现一个linux批量主机管理

功能描述
1. 可以分组批量执行远端命令，且要支持sudo，且捕获命令执行的退出符来判断命令执行是否成功还是失败,当然还有命令执行的时间
2. 可以put上传文件到服务器组，如果文件存在可以备份
3. 可以从远端服务器组下载文件
4. 多进程
代码目录结构

├── conf
│ ├── host_list.conf # 配置文件定义主机组，主机ip，ssh用户,密码或者ssh私钥地址
├── logs
│ └── paramiko.log # 日志是必须要有的
├── ssh_client.py # 执行文件
└── ssh_lib.py # 代码lib

配置文件 host_list.conf
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
### host_list.conf 说明

###### global
    概述: 全局性配置,如果下面的主机组没有设置用户名和密码就采用global
    ssh_user: ssh用户名
    ssh_passwd : ssh密码
    ssh_key : ssh用户私钥文件路径(绝对路径) 可以不写,如果ssh_key存在将优先使用,ssh_key

###### redis|mysql|.......
    概述: 定义ssh 的group
    ssh_user: ssh用户名 ,可以不写将采用global 的设置
    ssh_passwd: ssh密码, 同上也是可以不写,采用global
    ssh_key : 私钥路径,也可以不写
    hostlist: 主机ip列表用,做分隔,这个必须要写,写主机名也可以,只要能够解析到
    如果要指定端口使用":"做分隔例如: hostlist = 192.168.2.1:22,192.168.2.2:9922

下面是配置文件

1
2
3
4
5
6
7
8
9
10
11
12
 [global]
ssh_user = seal
ssh_password = lalala
ssh_key = /Users/seal/.ssh/id_rsa

[redis]
ssh_user = seal
ssh_password = xxxxx
hostlist = 192.168.2.199,192.168.2.198

[mysql]
hostlist = 192.168.2.23:22

代码库文件ssh_lib.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
 #!/usr/bin/env python
# coding: utf-8

"""
@Version: 1.0
@Author: KnightSeal
@Contact: knightseal@yeah.net
@Name: ssh_lib.py
@Create: 16/3/16 14:18
@Desc: 一个自定义ssh客户端
"""

import time
import logging
import paramiko
import functools

CONN_TIMEOUT = 5
CMD_TIMEOUT = 10


def myprint(color,mes):
    """
    用于打印颜色
    :param color: 颜色
    :param mes: 打印内容
    :return: 没有返回
    """
    info = {'red':31, 'green': 33, 'yellow':35, 'blue':34,'dark_green':36,'default':37}
    if color in info:
        fore = info[color]
    else:
        fore = 37
    color = "\x1B[%d;%dm" % (1,fore)
    print("%s%s\x1B[0m" % (color,mes))


def time_func(func):
    """
    装饰器: 命令运行时间
    :param func: 函数
    :return:
    """
    @functools.wraps(func)
    def wapper(*args,**kwargs):
        banner_footer = "*" * 90
        myprint('green',banner_footer)
        start = time.time()
        try:
            result = func(*args,**kwargs)
            return result
        except Exception :
            myprint("red","Exception: 网络连接错误或者密码不正确")
        finally:
            end = time.time()
            myprint("red","耗时:%s 秒" % round((end-start),2))
            myprint('green',banner_footer)
    return wapper

def show_help():
    """
    显示帮助
    :return:
    """
    help_info = {
            '-h': '打印帮助',
            '-G': '指定服务器分组,逗号分隔多个组,将读取conf/host_list.conf进行解析,例如 -G redis',
            '-k': '运行命令是否使用sudo',
            '-c': '运行命令,使用\"双引号包裹命令',
            '-p': 'put上传到远端服务器,使用\"双引号包裹,|分隔,例如: -p "localfile|remotefile" ',
            '-b': 'put上传到远端服务器是否备份源文件',
            '-g': 'get下载到本地,使用\"双引号包裹,|分隔 例如: -g "remotefile|localfile" ',
            'example\n' :
                        '\t执行redis服务器命令:\tpython ssh_client.py -G redis -c "cat /etc/hosts|grep -v "127.0.0.1" "\n'
                        "\t批量下载redis服务器/etc/hosts文件到本地/data/hosts:\tpython ssh_client.py -G redis -g /etc/hosts|/data/hosts\n"
                        "\t批量上传本地文件test.log到远端redis服务器/tmp/test.log[如果加上-b将对远端文件进行备份]:\tpython ssh_client.py -G redis -p test.log|/tmp/test.log\n"
            }
    for item in sorted(help_info.items(),key=lambda x:x[0]):
        myprint("green","%s %s" % item)


@time_func
def run_cmd(username,password,ip,cmd,port=22,key=None,sudo=False,logfile="logs/paramiko.log"):
    """
    运行远端命令:
    :param username: ssh用户名
    :param password: ssh密码
    :param ip: 主机ip,或者主机名
    :param cmd: 命令
    :param port: ssh 端口
    :param key: 私钥路径地址
    :param sudo: 是否使用sudo
    :param logfile: 日志文件路径
    """

    #日志对象
    logger = logging.getLogger()
    hdlr = logging.FileHandler(logfile)
    formatter = logging.Formatter('myssh ' +'%(asctime)s %(levelname)s %(message)s')
    hdlr.setFormatter(formatter)
    logger.addHandler(hdlr)
    logger.setLevel(logging.INFO)
    host_info = "* 主机:\t%s : 运行命令:[%s]" % (ip,cmd)
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    if key and key.endswith('rsa'):
        key = paramiko.RSAKey.from_private_key_file(key)
    elif key and key.endswith('dsa'):
        key = paramiko.DSSKey.from_private_key_file(key)

    ssh.connect(hostname=ip,port=port,username=username,password=password,pkey=key,\
           timeout=CONN_TIMEOUT,compress=True)

    # 如果需要用sudo
    if sudo:
        cmd = "sudo %s" % cmd
        stdin,stdout,stderr = ssh.exec_command(command=cmd,timeout=CMD_TIMEOUT,get_pty =True)
        # 使用sudo会提示密码输入,防止密码被打印在屏幕
        time.sleep(0.01)
        # 输入sudo密码
        stdin.write(password + "\n")
        stdin.flush()
        # 如果有错误
        if len(stderr.read()) > 0:
            stderr = "* %s" % (''.join(stderr.readlines()[1:]))
            stdout = stderr
            cmd_run_status = 255
        else:
            # 获取命令运行的退出是否是0
            cmd_run_status = stdout.channel.recv_exit_status()
            stdout = '* %s' % ('* '.join(stdout.readlines()[1:]))

        #输出结果
        banner_footer = "*" * 90
        #myprint('green',banner_footer)
        myprint('green',host_info)
        if cmd_run_status == 0:
            myprint("green","* 运行:\t[OK]")
        else:
            myprint("red","* 运行: [ERRO]")
        myprint("green",stdout)
        #myprint('green',banner_footer)

    # 不用sudo
    else:
        stdin,stdout,stderr =ssh.exec_command(command=cmd,timeout=CMD_TIMEOUT,get_pty =True)
        # 如果有错误
        if len(stderr.read()) > 0:
            stderr = "* %s" % (''.join(stderr.readlines()[1:]))
            stdout = stderr
            cmd_run_status = 255
        else:
            # 获取命令运行的退出是否是0
            cmd_run_status = stdout.channel.recv_exit_status()
            stdout = '* %s' % ('* '.join(stdout.readlines()))

        #输出结果
        banner_footer = "*" * 90
        #myprint('green',banner_footer)
        myprint('green',host_info)
        if cmd_run_status == 0:
            myprint("green","* 运行:\t[OK]")
        else:
            myprint("red","* 运行: [ERRO]")
        myprint("green",stdout)
        #myprint('green',banner_footer)
    log = "用户:%s, HOST: %s, CMD: %s, CMD_EXIT_CODE: %s" % (username,ip,cmd,cmd_run_status)
    logger.info(log)
    ssh.close()

def sftp_transfer(username,password,ip,local,remote,port=22,key=None,transfer_type="get",backup=True,logfile="logs/paramiko.log"):
    """
    sftp 文件传输,
    :param username: ssh用户名
    :param password: ssh密码
    :param ip: 主机ip,或者主机名
    :param port: ssh 端口
    :param local: 本地路径
    :param remote: 远端路径
    :param transfer_type: 传输类型 get|下载,put上传
    :param backup: 文件进行覆盖前,是否进行backup
    :param logfile: 日志文件路径
    :return:
    """
    logger = logging.getLogger()
    hdlr = logging.FileHandler(logfile)
    formatter = logging.Formatter('myssh ' +'%(asctime)s %(levelname)s %(message)s')
    hdlr.setFormatter(formatter)
    logger.addHandler(hdlr)
    logger.setLevel(logging.INFO)
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    if key and key.endswith('rsa'):
        key = paramiko.RSAKey.from_private_key_file(key)
    elif key and key.endswith('dsa'):
        key = paramiko.DSSKey.from_private_key_file(key)
    # 连接对象
    ssh.connect(hostname=ip,port=port,username=username,password=password,pkey=key,\
           timeout=CONN_TIMEOUT,compress=True)
    tansport =ssh.get_transport()
    sftp = paramiko.SFTPClient.from_transport(tansport)
    # 进行下载
    if transfer_type == 'get':
        try:
            sftp.stat(remote)
            # 为了保证远程下载多个服务器同名文件,就在文件名后面以ip地址为后缀
            download_file = "%s.%s" % (local,ip)
            sftp.get(remote,download_file)
            status = 'OK'
            myprint("green","从%s下载:%s:到本地,文件为:%s OK" % (ip,remote,download_file))
        except Exception as e:
            myprint("red","从%s下载: %s ERRO :%s" % (ip,remote,str(e)))
            status = str(e)
        finally:
            sftp.close()
            log = "用户: %s, HOST: %s, CMD: 下载文件, LOCAL: %s, REMOTE: %s, STATUS: %s" % (username,ip,local,remote,status)
            logger.info(log)
            tansport.close()
    # 上传
    else:
        # 是否需要开启备份模式,会先对远程已经存在的文件进行备份
        if backup:
            timestamp = int(time.time())
            backup_file = "%s.backup.%s" % (remote,timestamp)
            try:
                sftp.stat(remote) #文件是否存在,不存在出现FileNotFoundError
                sftp.rename(remote,backup_file) # 已经存在则进行备份
                myprint("red","%s 已经备份为: %s" % (remote,backup_file))
            except FileNotFoundError:
                pass
            except OSError:
                pass
        # 进行上传
        try:
            sftp.put(local,remote)
            status = 'OK'
            myprint("green","本地文件:%s,上传到:%s:%s OK" % (local,ip,remote))
        except Exception as e:
            myprint("red","ERRO: %s" % str(e))
            status = str(e)
        finally:
            log = "用户: %s, HOST: %s, CMD: 上传文件, LOCAL: %s, REMOTE: %s, STATUS: %s" % (username,ip,local,remote,status)
            logger.info(log)
            tansport.close()
            sftp.close()


# 单元测试
if __name__ == '__main__':
    run_cmd('seal','123456','192.168.2.199','ls',port=22,key=None,sudo=False,)
执行文件ssh_client.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
#!/usr/bin/env python
# coding: utf-8

"""
@Version: 1.0
@Author: KnightSeal
@Contact: knightseal@yeah.net
@Name: ssh_client.py
@Create: 16/3/16 11:13
@Desc: ssh客户端执行文件
"""
import sys
import getopt
import configparser
from  multiprocessing import Pool,cpu_count
from ssh_lib import *

class scripts_run(object):
    """
    运行
    """
    def __init__(self,groups,do_type=False,config_file='conf/host_list.conf',cmd=False,sudo=False,local=False,remote=False,backup=True):
        """
        构造函数
        :param groups: 用户输入的组名
        :param config_file: 主机配置文件
        :param do_type: 执行类型
        :param sudo: 是否使用sudo 运行命令
        :param localdir: 本地文件路径
        :param remote: 远端文件路径
        :param backup: 上传是否备份
        :return:
        """
        self.cf = configparser.ConfigParser()
        self.cf.read(config_file)

        self.conf_groups = self.cf.sections()
        del self.conf_groups[self.conf_groups.index('global')]
        self.username = self.cf.get("global",'ssh_user',fallback=False)
        self.password = self.cf.get("global",'ssh_password',fallback=False)
        self.key = self.cf.get("global",'ssh_key',fallback=None)
        # 用户输入的组合配置文件的组进行交际,得到真正需要运行的组
        self.groups = list(set(groups.split(",")) & set(self.conf_groups))
        self.config_file = config_file
        self.cmd = cmd
        self.sudo = sudo
        self.local  = local
        self.remote = remote
        self.transfer = do_type
        self.upload_backup = backup
        # 解析配置文件
        myprint("green","将要进行操作的服务器组是: %s" % ' '.join(self.groups))

    def do(self):
        """
        开始进行操作
        :return:
        """
        for group in self.groups:
            myprint("green","GROUPS: %s" % group)
            # 获得服务器组用户名和密码
            self.username = self.cf.get(group,"ssh_user",fallback=self.username)
            self.password = self.cf.get(group,"ssh_password",fallback=self.password)
            self.key = self.cf.get(group,"ssh_key",fallback=self.key)

            # 获取服务器列表
            host_list = self.cf.get(group,"hostlist").split(',')
            # 主机数量小于5,进程池为3
            if len(host_list) <= 5:
                self.pool = Pool(3)
            else: #大于5 进程池是主机核心数
                self.pool = Pool(cpu_count())
            for i in host_list:
                if ":" in i:
                    host,port = i.split(":")
                    port = int(port)
                else:
                    host = i
                    port = 22

                # 执行命令
                if self.cmd:
                    self.pool.apply_async(func=run_cmd,args=(self.username,self.password,host,self.cmd,port,self.key,self.sudo))
                elif self.transfer:
                    self.pool.apply_async(func=sftp_transfer,args=(self.username,self.password,host,self.local,self.remote,port,self.key,self.transfer,self.upload_backup))
                    pass

            self.pool.close()
            self.pool.join()

if __name__ == '__main__':
    argv = sys.argv[1:]
    # 得到命令参数和列表
    args = dict(getopt.getopt(argv,('hG:H:kc:p:g:f:b'))[0])
    if '-h' in args:
        show_help()
    elif '-g' in args and '-p' in args:
        myprint("red","不要闹了到底是上传还是下载啊!")
    elif '-G' not in args:
        myprint("red","请使用-G指定服务器组")
    else:
        # 执行命令
        if '-G' in args and '-c' in args:
            hosts_list = args['-G']
            cmd = args['-c']
            if '-k' in args:
                sudo =True
            else:
                sudo = False
            x = scripts_run(groups=hosts_list,cmd=cmd,sudo=sudo)
            x.do()

        # 进行上传
        elif '-G' in args and '-p' in args:
            hosts_list = args['-G']
            local,remote = args['-p'].split('|')
            if '-b' in args:
                backup = True
            else:
                backup = False
            x = scripts_run(groups=hosts_list,local=local,remote=remote,backup=backup,do_type='put')
            x.do()

        # 进行下载
        elif '-G' in args and '-g' in args:
            hosts_list = args['-G']
            print(args['-g'])
            remote,local = args['-g'].split('|')
            x = scripts_run(groups=hosts_list,local=local,remote=remote,do_type='get')
            x.do()

    #x = scripts_run(groups='mysql',do_type='put',local='test2.log',remote='/tmp/123',backup=False)
    #x = scripts_run(groups='mysql',cmd='ls')
    #x.do()
截图示例


文章目录
  1. 1. 功能描述
  2. 2. 代码目录结构
  3. 3. 配置文件 host_list.conf
  4. 4. 代码库文件ssh_lib.py
  5. 5. 执行文件ssh_client.py
    1. 5.1. 截图示例