介绍
Paramiko 一个第三方包，需要单独安装我们知道远程批量主机管理，比如ansible、Fabric，不需要安装客户端的远程执行命令等，这些都是基于Python原生的SSH，相当于模拟了一个SSH客户端。其实用的就是paramiko。其实想Fabric这种东西你自己也可以写，其实特别简单。paramiko 就是模拟了SSH的客户端，然后通过和SSH服务器交互就可以登录，执行命令等操作。
代码实例
简单的SSH登录实现
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Author: rex.cheny
# E-mail: rex.cheny@outlook.com import paramiko def main():
# 实例化SSH客户端
ssh = paramiko.SSHClient()
# 允许连接不在 known_hosts 文件的IP
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器
ssh.connect(hostname="172.16.42.136", port=22, username="root", password="12qwaszx!") # 执行命令，它返回三个结果，stdin 是你输入的，stdout是它返回给你的，stderr是错误信息
stdin, stdout, stderr = ssh.exec_command("ls /")
result = stdout.read()
print(result.decode(encoding="utf-8"))
ssh.close() if __name__ == '__main__':
main()
简单的SFTP实现
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Author: rex.cheny
# E-mail: rex.cheny@outlook.com import paramiko def main():
try:
# 实例化SSH客户端
ssh = paramiko.SSHClient()
# 允许连接不在 known_hosts 文件的IP
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器
ssh.connect(hostname="172.16.42.136", port=22, username="root", password="12qwaszx!")
transport = ssh.get_transport() """
上面代码也可以这样写
transport = paramiko.Transport((host, port))
transport.connect(username=username, password=password)
""" sftp = paramiko.SFTPClient.from_transport(transport)
sftp.put("./readme", "/tmp")
transport.close()
except Exception as err:
print(err) if __name__ == '__main__':
main()
包含SFTP和SSH功能的代码实例
#!/usr/bin/env python
# -*- coding: utf-8 -*- import sys
import os
import paramiko
import configparser class paramikoTools(object): def __init__(self, hostname, port, username, password, privateKeyFile=None, privateKeyFilePwd=None):
""" :param hostname: SSH服务器地址
:param port: 端口
:param username: 用户名
:param password: 密码
:param privateKeyFile: 私钥路径，如果没有则不用填写
:param privateKeyFilePwd: 私钥解压密码如果私钥加密了请务必输入否则登陆将会失败
:return:
"""
self._hostname = hostname
self._port = port
self._username = username
self._password = password
self._privateKeyFile = privateKeyFile
self._privateKeyFilePwd = privateKeyFilePwd # 实例化一个SSH客户端对象
self._sshClient = paramiko.SSHClient()
# 第一次SSH登录服务器需要输入一个Yes，这个就是帮你自动输入yes
self._sshClient.set_missing_host_key_policy(paramiko.AutoAddPolicy())
#
self._transport = None
#
self._sftpClient = None # 用于记录是否创建了SFTP客户端实例
self._isSftpSessionOpened = False
# 用于记录当前是否已经登录
self._isLogined = False def login(self):
if self._privateKeyFile:
if os.path.exists(self._privateKeyFile):
self._isLogined = self._loginWithPrivateKey()
else:
print("私钥路径不存在。")
else:
self._isLogined = self._loginWithPassword() # 用户名和密码登陆方法
def _loginWithPassword(self):
try:
# 连接远程服务器
self._sshClient.connect(hostname=self._hostname, port=self._port, username=self._username,
password=self._password)
print("密码登陆成功。")
except Exception as err:
# print err
print("连接失败，请检查主机名、端口、用户名或者密码是否正确。")
return False
else:
return True # 私钥登陆方法
def _loginWithPrivateKey(self):
try:
if self._privateKeyFilePwd:
# 取出私钥，有时私钥加密了这里就需要设置私钥的密码，用于对私钥解密。
key = paramiko.RSAKey.from_private_key_file(self._privateKeyFile, password=self._privateKeyFilePwd)
else:
key = paramiko.RSAKey.from_private_key_file(self._privateKeyFile) self._sshClient.connect(hostname=self._hostname, port=self._port, username=self._username,
password=self._password, pkey=key)
print("秘钥登陆成功。")
except Exception as err:
# print err
print("连接失败，请检查主机名、端口、用户名、密码、私钥是否正确。")
return False
else:
return True @property
def isLogined(self):
return self._isLogined # 创建SFTP客户端
def _createSftpSession(self):
"""
我这里建立SFTP连接的方式和网上略有不同，我这里是基于现有SSH连接来建立的SFTP会话。网上很多都是单纯的建立SFTP会话
但是即便单纯建立SFTP会话其背后也要建立套接字和身份验证，可能网上的其他形式如下：
t = paramiko.Transport((host, port))
t.connect(username=username, password=password)
sftp = paramiko.SFTPClient.from_transport(t)
"""
if self._isLogined:
try:
# 使用现有SSH连接获取一个 transport
self._transport = self._sshClient.get_transport()
# 把 transport 传递进去建立SFTP客户端连接
self._sftpClient = paramiko.SFTPClient.from_transport(self._transport)
except Exception as err:
print(err)
else:
self._isSftpSessionOpened = True
else:
print("ssh会话已经关闭或者没有建立，无法创建SFTP客户端连接。")
self._isSftpSessionOpened = False # 路径检查
def _pathChect(self, localPath, remotePath):
# F 表示文件； D 表示目录
CHECK_CODE = {"CODE": "", "LOCALPATHTYPE": "F", "REMOTEPATHTYPE": "F"}
if os.path.exists(localPath):
return CHECK_CODE
else:
# 本地路径检查失败，不存在。
CHECK_CODE = {"CODE": "", "LOCALPATHTYPE": "", "REMOTEPATHTYPE": ""}
return CHECK_CODE # 获取错误码对信息
def _getCheckCodeDesc(self, temp_code):
CHECK_CODE_DICT= {"": "检查通过", "": "本地路径检查失败可能不存在"}
DESC = ""
if CHECK_CODE_DICT.has_key(temp_code):
pass
else:
DESC = "错误码不存在"
return DESC # 关闭ssh连接
def sshConnectLogOut(self):
self._sshClient.close()
self._transport.close()
self._isLogined = False
self._isSftpSessionOpened = False # 执行命令
def execCommand(self, cmd):
result = ''
if self._isLogined:
try:
# 远程执行命令，会返回元祖，里面包含输入、输出、错误
stdin, stdout, stderr = self._sshClient.exec_command(cmd)
except Exception as err:
print(err)
else:
# 输出的就是字符串内容
print(stdout.read())
# return result
else:
print("当前没有建立SSH的有效连接。") # 上传文件
def upLoadFile(self, localPath, remotePath):
count = 0
# 如果SFTP客户端会话没有建立则自动建立，尝试三次
for times in range(3):
if not self._isSftpSessionOpened:
count = times + 1
print("第 %d 次尝试建立SFTP客户端会话，如果失败将再尝试 %d 次" % (count, 3 - count))
self._createSftpSession()
if count == 3 and not self._isSftpSessionOpened:
return None
else:
continue
else:
print("SFTP客户端会话建立成功。")
break # 目前这里是直接上传，没有区分文件还是目录，所以默认只能支持文件，本地是文件路径、远程也是文件路径
try:
CODE = self._pathChect(localPath, remotePath)["CODE"]
if CODE == "":
# 路径检查都成功后如何处理
print("开始上传文件。")
self._sftpClient.put(localPath, remotePath)
print("上传文件完毕。")
else:
print(self._getCheckCodeDesc(CODE))
except Exception as err:
print(err)
else:
pass # 下载文件
def downLoad(self, remotePath, localPath):
pass def main():
# 建立ini文件读取实例
config = configparser.ConfigParser()
# 设置配置文件路径
config_file_path = '/Users/rex.chen/PycharmProjects/Study/paramikoTest/config'
# 读取文件
config.read(config_file_path)
# 获取内容，第一个参数是区，第二个参数是 键
hostname = config.get("ssh", "hostname")
port = config.getint("ssh", "port")
username = config.get("ssh", "username")
password = config.get("ssh", "password")
private_key_path = config.get("ssh", "private_key_path") a = paramikoTools(hostname, port, username, password, private_key_path, password)
a.login()
# a.execCommand("df")
# a.upLoadFile('/Users/rex.chen/Downloads/mapi-webapp.war', '/home/*/update/mapi-webapp.war')
print(a.isLogined) if __name__ == "__main__":
main()
同目录的配置文件如下：
# paramiko连接使用的配置文件，不是必须的，只是这样可以避免过多的硬编码 [ssh]
hostname=
port=
username=
password=
private_key_path=