shell腳本批量部署ssh?今天主要分享一個批量ssh免密腳本,僅供參考,現在小編就來說說關于shell腳本批量部署ssh?下面内容希望能幫助到你,我們來一起看看吧!
今天主要分享一個批量ssh免密腳本,僅供參考。
管理端有多台服務器,維護幾百台服務器的時候需配置ssh免密,但密碼很多特殊字符,如果用expect是很難處理的,故python腳本實現。
1、安裝fabric
python3環境
pip install fabric
2、準備管理端和節點密碼文件(文件名分别為master.txt和node.txt)
格式為:IP 端口 賬号 密碼
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方面内容,感興趣的朋友可以關注下!
更多精彩资讯请关注tft每日頭條,我们将持续为您更新最新资讯!