一, 简介
Paramiko 模块是基于 Python 实现的 SSH 远程安全连接, 用于 SSH 远程执行命令, 文件传输等功能.
安装模块
默认 Python 没有自带, 需要手动安装:
pip3 install paramiko
二, 上传文件
- #!/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
测试一下上传, 完整代码如下:
- #!/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 = '[email protected]'
- 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)
- View Code
执行输出:
上传文件: 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
测试一下, 下载文件功能, 完整代码如下:
- #!/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 = '[email protected]'
- 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)
- View Code
执行输出:
下载文件: /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
测试一下, 远程执行命令功能, 完整代码如下:
- #!/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 = '[email protected]'
- 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)
- View Code
执行输出:
操作系统为: 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
- 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 或者其他较长的值即可解决这个问题.
参考:
-
- 腾讯云 https://www.qcloud.com/community/article/945339
本文参考链接:
来源: http://www.bubuko.com/infodetail-3109174.html