shell脚本批量部署ssh概述今天主要分享一个批量ssh免密脚本，仅供参考。需求管理端有多台服务器，维护几百台服务器的时候需配置ssh免密，但密码很多特殊字符，如果用expect是很难处理的，故python脚本实现。".formatloggingbreakelse:print("%s管理端生成ssh证书成功！"%master.ip)msg="{}-管理端已存在ssh证书！
shell脚本批量部署ssh?今天主要分享一个批量ssh免密脚本，仅供参考，现在小编就来说说关于shell脚本批量部署ssh?下面内容希望能帮助到你，我们来一起看看吧!
shell脚本批量部署ssh
概述今天主要分享一个批量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 hostexcept 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)continuehost_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)breakelse:print("%s 管理端生成ssh证书成功！"%master.ip)msg = "{} - 管理端生成ssh证书成功！".format(master.ip)logging(msg)continueelse:print("%s 管理端已存在ssh证书！"%master.ip)msg = "{} - 管理端已存在ssh证书！".format(master.ip)logging(msg)continuehost.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)continueelse:print("%s - 证书分发 - %s - 失败！"%(master.ip,node.ip))msg = "{} - 证书分发 - {} - 失败！".format(master.ip,node.ip)logging(msg)print(result)continuehost.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 Falsenode_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主机证书到nodessh_to_other(master_result,node_result)if __name__ == '__main__':main()
执行结果
至于验证能不能ssh免密登录就不截图了...
后面会分享更多devops和DBA方面内容，感兴趣的朋友可以关注下！
