SSH密钥对批量认证Python脚本

Posted 2020-10-01

用于集群中管理机器上的密钥对分发

版本：Python2.7

模块：paramiko、multiprocessing

备注：注意在脚本70行左右获取远程服务器秘钥指纹信息的时候不同的操作系统可能会有出入，如果使用本脚本务必在本地测试通过

# -*- coding:utf-8 -*-

import sys
reload(sys), sys.setdefaultencoding(‘utf-8‘)
import os
import time
import paramiko
import multiprocessing


# 此脚本用于进行秘钥对分发
# 需要配合nmap使用，所以需要在linux环境下

# 密码，不同服务器的不同密码可写在这一个列表中
password = [‘19820909‘]

# ip地址，多个写在一个列表中，具体列表请自行循环生成
ip_list = [‘10.10.10.139‘,‘10.10.10.140‘]

# 端口,不能有多个端口,需要统一
port = 22

# 本地用户的家目录和用户名 保持默认即执行脚本的用户
load_username = os.environ["USER"]
load_userhome = os.environ["HOME"]

# 远程服务器的用户名，可以指定用户名。默认和 load_username 一致
long_username = load_username
#long_username = "zouri"



try:
    # 获取本地公钥字符串
    keys_str = open("%s/.ssh/id_rsa.pub" % load_userhome).read().split("\n")[0]
except:
    print ‘Attempt command: \033[1;31mssh-keygen -t rsa -P "" -f ~/.ssh/id_rsa \033[0m ‘
    sys.exit(0)

def network_test(ip):
    # 使用 nmap 测试网络的连通性
    sss = os.popen("nmap -sP %s|grep Host|grep up|awk ‘{print $3}‘" % ip)
    sun = sss.read()
    if ‘up‘ in sun:
        return True
    else:
        return False



def ssh_test(ip, password):
    sun = False

    s = paramiko.SSHClient()
    s.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        # 建立ssh链接
        s.connect(ip, port, long_username, password)

        a, b, c = s.exec_command("echo $HOME")
        long_range_home = b.read().replace("\n", "")

        s.exec_command("mkdir %s/.ssh && chmod 700 %s/.ssh" % (long_range_home, long_range_home))
        s.exec_command("echo %s >> %s/.ssh/authorized_keys" % (keys_str, long_range_home))
        s.exec_command("chmod 600 %s/.ssh/authorized_keys" % (long_range_home))
        s.close()
        sun = True

        # 获取远程服务器的 秘钥指纹信息 写入本地
        key_1 = os.popen("ssh-keyscan -p %d %s" % (port,ip)).read().split("\n")[0]
        know_file = open(load_userhome + "/.ssh/known_hosts","a+")
        know_file.write(key_1 + "\n")
        know_file.close()

    except "Authentication":
        pass
    finally:
        return sun

def ssh_run(ip):  # 执行函数
    sun = 1
    if not network_test(ip):
        print "\033[1;31m%s\033[0m is no,Maybe the Network" % ip
    else:
        for i_in in password:
            if ssh_test(ip, i_in):
                print "%s is ok,Keys has been added" % ip
                sun = 0
                break
        if sun == 1:
            print "\033[1;31m%s\033[0m is no,Maybe the Password" % ip


def sun_run(x):
    # 开启多线程运行 x代表最多开多少线程
    sun_pool = multiprocessing.Pool(processes=x)
    for i in ip_list:
        sun_pool.apply_async(ssh_run, (i,))
    sun_pool.close()
    sun_pool.join()


if "__main__" == __name__:
    sun_time_01 = int(time.time())
    sun_run(30)
    sun_time_02 = int(time.time())
    print "共耗时%s秒" % str(sun_time_02 - sun_time_01)

本文出自 “马走日” 博客，谢绝转载！