tft每日頭條

 > 圖文

 > shell腳本批量部署ssh

shell腳本批量部署ssh

圖文 更新时间:2025-02-07 21:58:42

shell腳本批量部署ssh?今天主要分享一個批量ssh免密腳本,僅供參考,現在小編就來說說關于shell腳本批量部署ssh?下面内容希望能幫助到你,我們來一起看看吧!

shell腳本批量部署ssh(分享一個實用腳本)1

shell腳本批量部署ssh

概述

今天主要分享一個批量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方面内容,感興趣的朋友可以關注下!

更多精彩资讯请关注tft每日頭條,我们将持续为您更新最新资讯!

查看全部

相关圖文资讯推荐

热门圖文资讯推荐

网友关注

Copyright 2023-2025 - www.tftnews.com All Rights Reserved