全球看热讯：python-paramiko操作的封装

2023-01-31 11:04:59 来源：博客园

(资料图片)

python-paramiko操作的封装

1. python-paramiko操作的封装

案例1、封装密钥paramiko远程使用命令

#!/usr/bin/env python# _*_ coding: utf-8 _*_# Author:shichao# File: .pyimport paramikodef remoteExecCommand(ip, port, username, key ,command):    try:        # 远程连接 ip和端口        transport = paramiko.Transport(ip, port)        # 连接私钥        pkey = paramiko.RSAKey.from_private_key_file(key)        # 连接用户和密码        transport.connect(username=username, pkey=pkey)        ssh = paramiko.SSHClient()        ssh._transport = transport        # stdio 是输入， stdout 是输出，stderr错误输出        stdio, stdout, stderr = ssh.exec_command(command)        channel = stdout.channel        status = channel.recv_exit_status()        # stdout标准输出读取数据        stdout = stdout.read().decode()        # stderr标准错误数据输出读取数据        stderr = stderr.read().decode()        # 使用return 返回数据        return {"stdout":stdout , "stderr":stderr, "status":status}    except Exception as e:        print(e)    finally:        try:            ssh.close()        except Exception as e:            print(e)        try:            ssh.close()        except Exception as e:            print(e)# 进行本地测试if __name__ == "__main__":   result = remoteExecCommand("192.168.0.200", 22, "root", "/Users/admin/.ssh/id_rsa", "ls /tmp/")   print(result)

案例2、封装使用密码paramiko远程使用命令

#!/usr/bin/env python# _*_ coding: utf-8 _*_# Author:shichao# File: .pyimport paramikodef remoteExecCommand(ip, port, username, pwd ,command):    try:        # 远程连接 ip和端口        transport = paramiko.Transport(ip, port)        # 连接私钥        pkey = paramiko.RSAKey.from_private_key_file(pwd)        # 连接用户和密码        transport.connect(username=username, pkey=pkey)        ssh = paramiko.SSHClient()        ssh._transport = transport        # stdio 是输入， stdout 是输出，stderr错误输出        stdio, stdout, stderr = ssh.exec_command(command)        channel = stdout.channel        status = channel.recv_exit_status()        # stdout标准输出读取数据        stdout = stdout.read().decode()        # stderr标准错误数据输出读取数据        stderr = stderr.read().decode()        # 使用return 返回数据        return {"stdout":stdout , "stderr":stderr, "status":status}    except Exception as e:        print(e)    finally:        try:            ssh.close()        except Exception as e:            print(e)        try:            ssh.close()        except Exception as e:            print(e)# 进行本地测试if __name__ == "__main__":   result = remoteExecCommand("192.168.0.200", 22, "root", "123456", "ls /tmp/")   print(result)

案例3、封装paramiko-client远程上传文件

#!/usr/bin/env python# _*_ coding: utf-8 _*_# Author:shichao# File: .pyimport paramikodef remoteGET(ip, port, username, key, localpath, remotepath,):    try:        # 连接私钥        pkey = paramiko.RSAKey.from_private_key_file(key)        # 远程连接 ip和端口        transport = paramiko.Transport(ip, port)        # 连接用户和密码        transport.connect(username=username, pkey=pkey)        # 文件工具        sftp = paramiko.SFTPClient.from_transport(transport)        # 上传文件 localpath是本地路径，remotepath远程路径        result = sftp.put(localpath,remotepath)        # 使用return 返回数据        return result    except Exception as e:        print(e)    # 关闭连接    finally:        try:            sftp.close()        except Exception as e:            print(e)        try:            transport.close()        except Exception as e:            print(e)# 进行本地测试if __name__ == "__main__":   result = remoteGET("192.168.0.200", 22, "root", "/Users/admin/.ssh/id_rsa", "/tmp/hello.sh", "/tmp/hello.sh" )   print(result)

案例4、封装paramiko-client远程下载文件

#!/usr/bin/env python# _*_ coding: utf-8 _*_# Author:shichao# File: .pyimport paramikodef remoteGET(ip, port, username, key, remotepath, localpath,):    try:        # 连接私钥        pkey = paramiko.RSAKey.from_private_key_file(key)        # 远程连接 ip和端口        transport = paramiko.Transport(ip, port)        # 连接用户和密码        transport.connect(username=username, pkey=pkey)        # 文件工具        sftp = paramiko.SFTPClient.from_transport(transport)        # 上传文件 localpath是本地路径，remotepath远程路径        result = sftp.get(remotepath,localpath)        # 使用return 返回数据        return result    except Exception as e:        print(e)    # 关闭连接    finally:        try:            sftp.close()        except Exception as e:            print(e)        try:            transport.close()        except Exception as e:            print(e)# 进行本地测试if __name__ == "__main__":   result = remoteGET("192.168.0.200", 22, "root", "/Users/admin/.ssh/id_rsa", "/scripts/gethello.sh", "/tmp/gethello.sh",)   print(result)

关键词：上传文件错误数据关闭连接