分享一个实用脚本—管理端批量SSH免密，值得收藏

概述

需求

管理端有多台服务器，维护几百台服务器的时候需配置ssh免密，但密码很多特殊字符，如果用expect是很难处理的，故python脚本实现。

环境准备

1、安装fabric

python3环境

pip install fabric

2、准备管理端和节点密码文件（文件名分别为master.txt和node.txt）

格式为：IP 端口 账号 密码

批量ssh免密脚本

import argparseimport collectionsimport subprocessimport osfrom datetime import datetimefrom fabric import Connectionfrom invoke import Responderimport itertools# 存放管理端和目标服务器的密码文件(IP 端口 用户名 密码)MASTER_FILE = 'master.txt'NODE_FILE = 'node.txt'# 日志文件FILE_NAME = 'ssh_{}.log'.format(datetime.now().strftime('%Y-%m-%d'))# 全局变量,我这使用的是 命名元祖HOST_RESULT = collections.namedtuple('Host',['ip','port','user','passwd'])def logging(msg): ''' 记录错误信息的函数 :param msg: 错误信息 :return: ''' base_dir = os.getcwd() full_filename = os.path.join(base_dir,FILE_NAME) command = 'echo '{}' >> {} '.format(msg,full_filename) subprocess.check_call(command,shell=True)def ssh_connect(ip,port='22',user='root',passwd=None): ''' 使用 ssh 连接服务器 :param : IP 端口 用户名 密码 :return: ''' if passwd: try: host = Connection(ip,port=port,user=user,connect_kwargs={'password':passwd,'timeout':60}) host.run('ls',hide=True,warn=True) if host.is_connected: return host except Exception as e: return Falsedef check_host_passwd(iplist): ''' 检测密码是否可用 :param iplist: IP 端口 用户名 密码列表 :return: 有效的IP 端口 用户名 密码列表 ''' host_list = [] for inform in iplist: ipaddr=inform[0] port=inform[1] username=inform[2] passwd=inform[3] host = ssh_connect(ipaddr,port,username,passwd) if not host: msg = '{} - 登录失败！'.format(ipaddr) logging(msg) continue host_info = HOST_RESULT(ip=ipaddr,port=port,user=username,passwd=passwd) host_list.append(host_info) return host_listdef gen_master_ssh_key(master_list): '''生成秘钥 :param master_list: 管理端的ip、端口、用户名和密码组成的元祖类型 :return: ''' print('*********************************管理端生成密钥*********************************') for master in master_list: host = ssh_connect(master.ip,port=master.port,user=master.user,passwd=master.passwd) if not host: return False,'{}.master主机登录失败'.format(master.ip) # 执行 Linux命令，判断是否存在 id_rsa.pub文件 command = 'find /root/.ssh/ -name 'id_rsa.pub'' result = host.run(command,hide=True,warn=True) if len(result.stdout.strip()) == 0: id_rsa = Responder( pattern = r'/root/.ssh/id_rsa', response = '/root/.ssh/id_rsa\n' ) passphrase = Responder( pattern = r'passphrase', response = '\r\n' ) yes = Responder( pattern = r'(y/n)', response = 'y\n' ) # 执行Linux 的 生成秘钥的命令 result = host.run('ssh-keygen -t rsa',hide=True,warn=True,watchers=[id_rsa,passphrase,yes],timeout=10) if not result.ok: print('%s 管理端生成ssh证书失败！'%master.ip) msg = '{} - 管理端生成ssh证书失败！'.format(master.ip) logging(msg) break else: print('%s 管理端生成ssh证书成功！'%master.ip) msg = '{} - 管理端生成ssh证书成功！'.format(master.ip) logging(msg) continue else: print('%s 管理端已存在ssh证书！'%master.ip) msg = '{} - 管理端已存在ssh证书！'.format(master.ip) logging(msg) continue host.close() return True,'管理端生成证书完成'def ssh_to_other(master_list,node_list): ''' 把生成的证书分发给下面的免密的服务器 :param master_list: 管理服务器 元祖 :param node_list: 节点列表 :return: ''' print('**********************************证书文件分发**********************************') for master in master_list: host = ssh_connect(master.ip,port=master.port,user=master.user,passwd=master.passwd) if not host: return False,'{}.master主机登录失败'.format(master.ip) for node in node_list: passwd = Responder( pattern=r'password', response=node.passwd + '\n' ) yes = Responder( pattern = r'(yes/no)', response = 'yes\n' ) # 清除 known_hosts文件 clean_command = 'ssh-keygen -f '/root/.ssh/known_hosts' -R {}'.format(node.ip) result = host.run(clean_command,hide=True,warn=True,timeout=30) # if result.ok: # return 'known_hosts 记录清理' # else: # return 'konwn_hosts 无需清理' # 分发证书的 Linux命令 scp_crt_command = 'ssh-copy-id -i /root/.ssh/id_rsa.pub {}@{}'.format(node.user,node.ip) #权限拒绝一般需要chattr -ia /root/.ssh/authorized_keys #pty=True 这个参数会影响是否能分发成功 result = host.run(scp_crt_command,pty=True,watchers=[passwd,yes],hide=True,warn=True,timeout=60) if result.ok: print('%s - 证书分发 - %s - 成功！'%(master.ip,node.ip)) msg = '{} - 证书分发 - {} - 成功！'.format(master.ip,node.ip) logging(msg) continue else: print('%s - 证书分发 - %s - 失败！'%(master.ip,node.ip)) msg = '{} - 证书分发 - {} - 失败！'.format(master.ip,node.ip) logging(msg) print(result) continue host.close() return True,'管理端已完成证书分发！'def check_ssh_login(master_list,node_list): ''' 测试免密登录是否实现的函数 :param master: 主服务器 :param nodes: 节点服务器 :return: ''' # 主服务器的连接 host = ssh_connect(master.ip,passwd=master.passwd) if not host: return False,'{} - master 登录失败'.format(master.ip) # 遍历节点服务器列表，对每一个ip进行测试 for node in nodes: ssh_command = 'ssh {} echo 'ok' '.format(node) try: result = host.run(ssh_command,pty=True,hide=True,warn=True,timeout=5) if result.ok: return True,'{} 登录{}成功'.format(master.ip,node) else: msg = '{} 登录{}失败--{}'.format(master.ip,node,result.stdout) logging(msg) except Exception as e: msg = '{} - master登录{} 失败--{}'.format(master.ip,node,e) logging(msg) return True,''def main(): ''' 运行函数 :return: ''' base_dir = os.getcwd() master_path = os.path.join(base_dir,MASTER_FILE) node_path = os.path.join(base_dir,NODE_FILE) with open(master_path,'r') as m: masterlist = [line.split() for line in m] with open(node_path,'r') as f: nodelist = [line.split() for line in f] # 登录节点并返回有效的主机列表 master_result = check_host_passwd(masterlist) if len(master_result) == 0: print('无可用管理节点') return False node_result = check_host_passwd(nodelist) if len(node_result) == 0: print('无可用节点') return False #生成管理端主机密钥 status,msg = gen_master_ssh_key(master_result) if not status: logging(msg) # 分发所有master主机证书到node ssh_to_other(master_result,node_result)if __name__ == '__main__': main()

执行结果

至于验证能不能ssh免密登录就不截图了...

赞 (0)

相关推荐