paramiko介绍
paramiko是一个用于做远程控制的模块，使用该模块可以对远程服务器进行命令或文件操作，值得一说的是，fabric和ansible内部的远程管理就是使用的paramiko来现实。其实它的底层是对ssh的上层代码的一个封装。
ssh是一个协议，OpenSSH是其中一个开源实现，paramiko是Python的一个库，实现了SSHv2协议(底层使用cryptography)。
有了Paramiko以后，我们就可以在Python代码中直接使用SSH协议对远程服务器执行操作，而不是通过ssh命令对远程服务器进行操作。
安装
paramiko属于第三方库，依赖pycrypto库：
pip3 install pycrypto # Windows安装失败，安装pycrypto?dome 或者 pycryptodomex.pycryptodomex 一个独立于旧 PyCrypto 的库，所有模块都安装在Cryptodome包下。旧的 PyCrypto 和 PyCryptodome 可以共存
pip3 install pycryptodomex
pip3 install paramiko
paramiko核心组件-SSHClient和SFTPClient
- SSHClient的作用类似于Linux的ssh命令，是对SSH会话的封装，该类封装了传输(Transport)，通道(Channel)及SFTPClient建立的方法(open_sftp)，通常用于执行远程命令。
- SFTPClient的作用类似与Linux的sftp命令，是对SFTP客户端的封装，用以实现远程文件操作，如文件上传、下载、修改文件权限等操作。
SSHClient常用的方法介绍
- connect()：实现远程服务器的连接与认证，对于该方法只有hostname是必传参数。
常用参数：
hostname 连接的目标主机
port=SSH_PORT 指定端口
username=None 验证的用户名
password=None 验证的用户密码
pkey=None 私钥方式用于身份验证
key_filename=None 一个文件名或文件列表，指定私钥文件
timeout=None 可选的tcp连接超时时间
allow_agent=True, 是否允许连接到ssh代理，默认为True 允许
look_for_keys=True 是否在~/.ssh中搜索私钥文件，默认为True 允许
compress=False, 是否打开压缩- set_missing_host_key_policy()：设置远程服务器没有在know_hosts文件中记录时的应对策略。目前支持三种策略：
设置连接的远程主机没有本地主机密钥或HostKeys对象时的策略，目前支持三种：
- AutoAddPolicy 自动添加主机名及主机密钥到本地HostKeys对象，不依赖load_system_host_key的配置。即新建立ssh连接时不需要再输入yes或no进行确认；
- WarningPolicy 用于记录一个未知的主机密钥的python警告。并接受，功能上和AutoAddPolicy类似，但是会提示是新连接；
- RejectPolicy 自动拒绝未知的主机名和密钥，依赖load_system_host_key的配置。此为默认选项。
- exec_command()：在远程服务器执行Linux命令的方法。
- open_sftp()：在当前ssh会话的基础上创建一个sftp会话。该方法会返回一个SFTPClient对象。
- invoke_shell(): 用于创建一个子shell进程，这样所有的操作都可以在该子shell中进行，su切换用户不受影响，但该方法没有exec_command那种方便的ChannelFile对象，所有的标准输出和标准错误内容都通过invoke_shell返回对象的recv方法来获取，每一次调用recv只会从上一次返回的地方开始返回，也没有直接获取命令退出状态
SFTPClient常用方法介绍
SFTPCLient作为一个sftp的客户端对象，根据ssh传输协议的sftp会话，实现远程文件操作，如上传、下载、权限、状态
- from_transport(cls,t) 创建一个已连通的SFTP客户端通道
- put(localpath, remotepath, callback=None, confirm=True) 将本地文件上传到服务器 参数confirm：是否调用stat()方法检查文件状态，返回ls -l的结果
- get(remotepath, localpath, callback=None) 从服务器下载文件到本地
- mkdir() 在服务器上创建目录
- remove() 在服务器上删除目录
- rename() 在服务器上重命名目录
- stat() 查看服务器文件状态
- listdir() 列出服务器目录下的文件
具体示例方法
# 定义一个类，表示一台远端linux主机
class SSHLinux(object):
# 通过IP, 用户名，密码，超时时间初始化一个远程Linux主机
def __init__(self, host, username, password, timeout=30, port=22):
self.host = host
self.username = username
self.password = password
self.timeout = timeout
self.port = port
self.ssh = None
self.trans = None
self.sftp = None
# 链接失败的重试次数
self.try_times = 3
def connect(self):
while True:
try:
# 基于用户名和密码的 transport 方式登录
self.trans = transport.Transport(self.host, self.port) # 创建一个通道
self.trans.connect(username=self.username, password=self.password) # 连接
self.ssh = client.SSHClient() # 实例化SSHClient
self.ssh.load_system_host_keys() # 加载系统HostKeys密钥
self.ssh._transport = self.trans
# 自动添加策略，保存服务器的主机名和密钥信息，如果不添加，那么不再本地know_hosts文件中记录的主机将无法连接
self.ssh.set_missing_host_key_policy(client.AutoAddPolicy())
self.sftp = sftp_client.SFTPClient.from_transport(self.trans) # 实例化一个SFTPClient对象
return True
except Exception as e:
if self.try_times != 0:
print(f'连接{self.host}失败: {str(e)}，进行重试...')
self.try_times -= 1
else:
print(f'重试3次失败，结束程序')
return False
def send(self, cmd):
# get_pty=True 解决 nohup 执行了但是 进程没有启动成功
# nohup 执行成功了，但是python程序阻塞了，无法停止，解决方式 nohup xx.xx & sleep 1 &后增加sleep 1
ssh_stdin, ssh_stdout, ssh_stderr = self.ssh.exec_command(cmd, get_pty=True)
try:
logger.info(f"ssh stdout:{ssh_stdout.read()}")
except:
logger.info(f"ssh stderr:{ssh_stderr.read()}")
def send_invoke_shell_msg(self, password, cmd):
""" 切换为invoke_shell模式执行命令
:param password:
:param cmd:
:return:
"""
try:
channel = self.ssh.invoke_shell()
time.sleep(0.1)
channel.send(f"sudo su \n") # 切换为root
buff = ''
while not buff.endswith(f'[sudo] password for {self.username}: '):
resp = channel.recv(9999)
buff += resp.decode('utf-8')
channel.send(password)
channel.send('\n')
buff = ''
while not buff.endswith('# '): # 当指令执行结束后，Linux窗口会显示#，等待下条指令，所以可以用作识别全部输出结束的标志。
resp = channel.recv(9999)
buff += resp.decode('utf-8')
print("------end------")
# 查看是否切换成功
channel.send(cmd)
channel.send("\n")
buff = ''
while not buff.endswith('# '):
resp = channel.recv(9999)
buff += resp.decode('utf-8')
print(buff)
except paramiko.ssh_exception.AuthenticationException:
print('Failed to login. ip username or password not correct.')
exit(-1)
def close(self):
# 关闭连接
self.trans.close()
self.ssh.close()
self.sftp.close()
# SFTPClient
def upload_file(self, local_path, remote_path):
""" 文件上传
:param local_path: 本地文件地址
:param remote_path: 远程地址
:return:
"""
try:
self.sftp.put(local_path, remote_path)
logger.info(f"服务器:{self.host}, 文件: {remote_path.rsplit('/', 1)[-1]}上传成功.")
return True
except Exception as e:
logger.error(f"文件:{remote_path.rsplit('/', 1)[-1]} 上传失败:{e}")
return False
def download_file(self, local_path, remote_path):
""" 文件下载
:param local_path: 本地要下载的目录
:param remote_path: 远程文件地址
:return:
"""
try:
self.sftp.get(remote_path, local_path)
logger.info(f"服务器:{self.host}, 文件: {remote_path.rsplit('/', 1)[-1]}下载成功.")
return True
except Exception as e:
logger.error(f"文件:{remote_path.rsplit('/', 1)[-1]} 下载失败:{e}")
return False
def update_chmod(self, path, mode=777):
""" 文件权限修改 """
try:
self.sftp.chmod(path, mode)
logger.info(f"服务器:{self.host}, 文件: {path}权限变更成功.")
return True
except Exception as e:
logger.error(f"服务器:{self.host}, 文件: {path}权限变更失败: {e}.")
return False
def update_chown(self, path, uid, gid):
"""文件属主修改"""
try:
self.sftp.chown(path, uid, gid)
logger.info(f"服务器:{self.host}, 文件: {path}属主变更成功.")
return True
except Exception as e:
logger.error(f"服务器:{self.host}, 文件: {path}属主更新失败:{e}")
return False
def create_dir(self, path, mode=777):
"""新建文件夹"""
if not os.path.exists(path):
try:
self.sftp.mkdir(path)
logger.info(f"服务器:{self.host}, 文件夹: {path}创建成功.")
return True
except Exception as e:
logger.error(f"服务器:{self.host}, 文件夹: {path}创建失败:{e}")
return False
def rm_dir(self, path):
"""删除文件夹"""
try:
self.sftp.rmdir(path)
logger.info(f"服务器:{self.host}, 文件夹: {path}删除成功.")
return True
except Exception as e:
logger.error(f"服务器:{self.host}, 文件夹: {path}删除失败:{e}")
return False
def remove_file(self, file_path):
"""删除文件"""
try:
self.sftp.remove(file_path)
logger.info(f"服务器:{self.host}, 文件: {file_path}删除成功.")
return True
except Exception as e:
logger.error(f"服务器:{self.host}, 文件: {file_path}删除失败:{e}")
return False
def rename_file(self, old_path, new_path):
"""文件重命名"""
try:
self.sftp.rename(old_path, new_path)
logger.info(f"服务器:{self.host}, 文件: {new_path}重命名成功.")
return True
except Exception as e:
logger.error(f"服务器:{self.host}, 文件: {new_path}重命名失败:{e}")
return False
def get_file(self, file_path, mode="r", bufsize=-1):
"""获取文件"""
try:
return self.sftp.file(file_path, mode=mode, bufsize=bufsize)
except Exception as e:
logger.error(f"获取文件失败:{e}")
return False
def open_file(self, file_path, mode="r", bufsize=-1):
"""打开文件"""
try:
return self.sftp.open(file_path, mode=mode, bufsize=bufsize)
except Exception as e:
logger.error(f"获取文件失败:{e}")
return False
def stat_path(self, file_path):
"""查看文件属性"""
try:
self.sftp.stat(file_path)
return True
except Exception as e:
logger.error(f"文件不存在或stat失败:{e}")
return False总结
近几天在写链路测试平台，正好涉及到远程文件上传和执行，记录一下paramiko的使用过程！