网站首页 > 博客 > 正文

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

bbbsonic 博客 2024-05-22 1 0

一、简介

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 = '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)

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 = '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)

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(ip,cmd))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(ip,cmd))

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)

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或者其他较长的值即可解决这个问题。

参考： - stackoverflow - 腾讯云

https://blog.csdn.net/qq_30513141/article/details/78201078

本文参考链接：

https://blog.csdn.net/Temanm/article/details/50607741

精彩链接

评论可见，请评论后查看内容，谢谢！！！评论后请刷新页面。

本文由用户于 2024-05-22 发布在夸智网，如有疑问，请联系我们。
本文链接：https://www.kuazhi.com/post/714068559.html

夸智网

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

promethus监控nginx

Pycharm专业版配置远程服务器并自动同步代码

发表评论取消回复

夸智网

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

promethus监控nginx

Pycharm专业版配置远程服务器并自动同步代码

相关文章

发表评论取消回复