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

#!/usr/bin/env python3
# coding: utf-8
import paramikodef 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 Falseif __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)

#!/usr/bin/env python3
# coding: utf-8
import paramikodef 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 Falseif __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)

#!/usr/bin/env python3
# coding: utf-8
import paramikodef 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 Falseif __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)