概述

今天主要分享一个批量ssh免密脚本,仅供参考。


需求

管理端有多台服务器,维护几百台服务器的时候需配置ssh免密,但密码很多特殊字符,如果用expect是很难处理的,故python脚本实现。


环境准备

1、安装fabric

python3环境

pip install fabric

 

分享一个实用脚本—管理端批量SSH免密,值得收藏

 

2、准备管理端和节点密码文件(文件名分别为master.txt和node.txt)

格式为:IP 端口 账号 密码

 

分享一个实用脚本—管理端批量SSH免密,值得收藏

 


批量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免密,值得收藏

 

分享一个实用脚本—管理端批量SSH免密,值得收藏

 

分享一个实用脚本—管理端批量SSH免密,值得收藏

 

分享一个实用脚本—管理端批量SSH免密,值得收藏

 


执行结果

至于验证能不能ssh免密登录就不截图了...

分享一个实用脚本—管理端批量SSH免密,值得收藏

 转自https://www.toutiao.com/i6907968863619908104/?tt_from=weixin&utm_campaign=client_share&wxshare_count=1&timestamp=1608858003&app=news_article&utm_source=weixin&utm_medium=toutiao_android&use_new_style=1&req_id=202012250900030101941000190315C8DF&group_id=6907968863619908104

另外据说ansible也是一款很强大的运维工具,下次是有时间再分享下


评论关闭
IT序号网

微信公众号号:IT虾米 (左侧二维码扫一扫)欢迎添加!

wxPython:pythonGUI库