Python Paramiko实现sftp文件上传下载以及远程执行命令

电玩女神 2021-12-19 23:13 775阅读 0赞

# 一、简介 #

Paramiko模块是基于Python实现的SSH远程安全连接，用于SSH远程执行命令、文件传输等功能。

## 安装模块 ##

默认Python没有自带，需要手动安装：

pip3 install paramiko

# 二、上传文件 #

测试一下上传，完整代码如下：

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

#!/usr/bin/env python3
    # coding: utf-8
    import paramiko
    
    
    def sftp_upload_file(host, user, password, server_path, local_path, timeout=10):
        """
        上传文件，注意：不支持文件夹
        :param host: 主机名
        :param user: 用户名
        :param password: 密码
        :param server_path: 远程路径，比如：/home/sdn/tmp.txt
        :param local_path: 本地路径，比如：D:/text.txt
        :param timeout: 超时时间(默认)，必须是int类型
        :return: bool
        """
        try:
            t = paramiko.Transport((host, 22))
            t.banner_timeout = timeout
            t.connect(username=user, password=password)
            sftp = paramiko.SFTPClient.from_transport(t)
            sftp.put(local_path, server_path)
            t.close()
            return True
        except Exception as e:
            print(e)
            return False
    
    
    if __name__ == '__main__':
        host = '192.168.10.1'
        user = 'xiao'
        password = 'xiao@1234'
    
        server_path = '/tmp/tmp.txt'
        local_path = 'D:/text.txt'
        res = sftp_upload_file(host, user, password, server_path, local_path)
        if not res:
            print("上传文件: %s 失败"%local_path)
        else:
            print("上传文件: %s 成功" % local_path)

执行输出：

上传文件: D:/text.txt 成功

# 三、下载文件 #

def sftp_down_file(host,user,password,server_path, local_path,timeout=10):
        """
        下载文件，注意：不支持文件夹
        :param host: 主机名
        :param user: 用户名
        :param password: 密码
        :param server_path: 远程路径，比如：/home/sdn/tmp.txt
        :param local_path: 本地路径，比如：D:/text.txt
        :param timeout: 超时时间(默认)，必须是int类型
        :return: bool
        """
        try:
            t = paramiko.Transport((host,22))
            t.banner_timeout = timeout
            t.connect(username=user,password=password)
            sftp = paramiko.SFTPClient.from_transport(t)
            sftp.get(server_path, local_path)
            t.close()
            return True
        except Exception as e:
            print(e)
            return False

测试一下，下载文件功能，完整代码如下：

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

#!/usr/bin/env python3
    # coding: utf-8
    import paramiko
    
    
    def sftp_down_file(host,user,password,server_path, local_path,timeout=10):
        """
        下载文件，注意：不支持文件夹
        :param host: 主机名
        :param user: 用户名
        :param password: 密码
        :param server_path: 远程路径，比如：/home/sdn/tmp.txt
        :param local_path: 本地路径，比如：D:/text.txt
        :param timeout: 超时时间(默认)，必须是int类型
        :return: bool
        """
        try:
            t = paramiko.Transport((host,22))
            t.banner_timeout = timeout
            t.connect(username=user,password=password)
            sftp = paramiko.SFTPClient.from_transport(t)
            sftp.get(server_path, local_path)
            t.close()
            return True
        except Exception as e:
            print(e)
            return False
    
    
    if __name__ == '__main__':
        host = '192.168.10.1'
        user = 'xiao'
        password = 'xiao@1234'
    
        server_path = '/tmp/tmp.txt'
        local_path = 'D:/text.txt'
        res = sftp_down_file(host, user, password, server_path, local_path)
        if not res:
            print("下载文件: %s 失败"%server_path)
        else:
            print("下载文件: %s 成功" % server_path)

执行输出：

下载文件: /tmp/tmp.txt 成功

# 四、远程执行命令 #

def ssh_exec_command(host,user,password, cmd,timeout=10):
        """
        使用ssh连接远程服务器执行命令
        :param host: 主机名
        :param user: 用户名
        :param password: 密码
        :param cmd: 执行的命令
        :param seconds: 超时时间(默认)，必须是int类型
        :return: dict
        """
        result = {
        'status': 1, 'data': None}  # 返回结果
        try:
            ssh = paramiko.SSHClient()  # 创建一个新的SSHClient实例
            ssh.banner_timeout = timeout
            # 设置host key,如果在"known_hosts"中没有保存相关的信息, SSHClient 默认行为是拒绝连接, 会提示yes/no
            ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            ssh.connect(host, 22, user, password, timeout=timeout)  # 连接远程服务器,超时时间1秒
            stdin, stdout, stderr = ssh.exec_command(cmd,get_pty=True,timeout=timeout)  # 执行命令
            out = stdout.readlines()    # 执行结果,readlines会返回列表
            # 执行状态,0表示成功，1表示失败
            channel = stdout.channel
            status = channel.recv_exit_status()
            ssh.close()  # 关闭ssh连接
    
            # 修改返回结果
            result['status'] = status
            result['data'] = out
            return result
        except Exception as e:
            print(e)
            print("错误, 登录服务器失败! ip: {}".format(host))
            return False

测试一下，远程执行命令功能，完整代码如下：

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

#!/usr/bin/env python3
    # coding: utf-8
    import paramiko
    
    
    def ssh_exec_command(host,user,password, cmd,timeout=10):
        """
        使用ssh连接远程服务器执行命令
        :param host: 主机名
        :param user: 用户名
        :param password: 密码
        :param cmd: 执行的命令
        :param seconds: 超时时间(默认)，必须是int类型
        :return: dict
        """
        result = {
         'status': 1, 'data': None}  # 返回结果
        try:
            ssh = paramiko.SSHClient()  # 创建一个新的SSHClient实例
            ssh.banner_timeout = timeout
            # 设置host key,如果在"known_hosts"中没有保存相关的信息, SSHClient 默认行为是拒绝连接, 会提示yes/no
            ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            ssh.connect(host, 22, user, password, timeout=timeout)  # 连接远程服务器,超时时间1秒
            stdin, stdout, stderr = ssh.exec_command(cmd,get_pty=True,timeout=timeout)  # 执行命令
            out = stdout.readlines()    # 执行结果,readlines会返回列表
            # 执行状态,0表示成功，1表示失败
            channel = stdout.channel
            status = channel.recv_exit_status()
            ssh.close()  # 关闭ssh连接
    
            # 修改返回结果
            result['status'] = status
            result['data'] = out
            return result
        except Exception as e:
            print(e)
            print("错误, 登录服务器失败! ip: {}".format(host))
            return False
    
    
    if __name__ == '__main__':
        host = '192.168.10.1'
        user = 'xiao'
        password = 'xiao@1234'
    
        cmd = "cat /etc/issue | awk '{print $1,$2,$3}'"
        res = ssh_exec_command(host, user, password, cmd)
        # print(res)
        if not res or not res['data'] or res['status'] != 0:
            print("错误, ip: {} 执行命令: {} 失败".format(host, cmd), "red")
            exit()
    
        value = res['data'][0].strip()  # 获取实际值
        print("操作系统为: %s"%value)

执行输出：

操作系统为: Ubuntu 16.04.2 LTS

# 五、错误集锦 #

**1. EllipticCurvePublicKey.public\_bytes**

Please use EllipticCurvePublicKey.public_bytes to obtain both compressed and uncompressed point encoding.

### 原因 ###

paramiko 2.4.2 依赖 cryptography，而最新的cryptography==2.5里有一些弃用的API。

### 解决 ###

删掉cryptography，安装2.4.2，就不会报错了。

pip uninstall cryptography
    pip install cryptography==2.4.2

本文参考链接：

[https://blog.51cto.com/wangfeng7399/2376115][https_blog.51cto.com_wangfeng7399_2376115]

**2. Error reading SSH protocol banner**

Traceback (most recent call last):
      File "/python3/lib/python3.5/site-packages/paramiko/transport.py", line 1966, in run
        self._check_banner()
      File "/python3/lib/python3.5/site-packages/paramiko/transport.py", line 2143, in _check_banner
        "Error reading SSH protocol banner" + str(e)
    paramiko.ssh_exception.SSHException: Error reading SSH protocol banner
    
    Error reading SSH protocol banner

要解决这个问题， 需要将paramiko的响应等待时间调长。   
修改paramiko/transport.py文件中的

self.banner_timeout

值， 将其设为300或者其他较长的值即可解决这个问题。

参考：   
\- [stackoverflow][]   
\- [腾讯云][Link 1]

[https://blog.csdn.net/qq\_30513141/article/details/78201078][https_blog.csdn.net_qq_30513141_article_details_78201078]

本文参考链接：

[https://blog.csdn.net/Temanm/article/details/50607741][https_blog.csdn.net_Temanm_article_details_50607741]

[ContractedBlock.gif]: https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif
[ExpandedBlockStart.gif]: /images/20211219/3faa60b318de4104a813620f82dad82b.png
[https_blog.51cto.com_wangfeng7399_2376115]: https://blog.51cto.com/wangfeng7399/2376115
[stackoverflow]: https://stackoverflow.com/questions/25609153/paramiko-error-reading-ssh-protocol-banner%20stackover%20flow
[Link 1]: https://www.qcloud.com/community/article/945339
[https_blog.csdn.net_qq_30513141_article_details_78201078]: https://blog.csdn.net/qq_30513141/article/details/78201078
[https_blog.csdn.net_Temanm_article_details_50607741]: https://blog.csdn.net/Temanm/article/details/50607741