构建ssh免密登录步骤（分享一个实用脚本）
构建ssh免密登录步骤（分享一个实用脚本）pip install fabric2、准备管理端和节点密码文件（文件名分别为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
概述今天主要分享一个批量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方面内容，感兴趣的朋友可以关注下！
