概述
今天主要分享一个批量ssh免密脚本，仅供参考。
需求
管理端有多台服务器，维护几百台服务器的时候需配置ssh免密，但密码很多特殊字符，如果用expect是很难处理的，故python脚本实现。
环境准备
1、安装fabric
python3环境
pip install fabric
2、准备管理端和节点密码文件（文件名分别为master.txt和node.txt）
格式为：IP 端口 账号 密码
批量ssh免密脚本
import argparse
import collections
import subprocess
import os
from datetime import datetime
from fabric import Connection
from invoke import Responder
import 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 False
def 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_list
def 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免密登录就不截图了…
后面会分享更多devops和DBA方面内容，感兴趣的朋友可以关注下！
