在雲服務測試中,每每須要咱們進入雲服務內容進行相關內容的測試。這測試可使用平臺自身的noVNC、外部輔助xshell等工具鏈接到雲服務內部進行測試。
可是在如此反覆的測試操做中,就須要用到自動化測試方法去解決這方面的需求。
在python中咱們能夠經過第三方庫paramiko,對linux的雲服務器進行操做。
以下命令先行安裝python
pip install paramiko
paramiko包含兩個核心組件:SSHClient和SFTPClient。linux
SSHClient的做用相似於Linux的ssh命令,是對SSH會話的封裝,該類封裝了傳輸(Transport),通道(Channel)及SFTPClient創建的方法(open_sftp),一般用於執行遠程命令。 SFTPClient的做用相似與Linux的sftp命令,是對SFTP客戶端的封裝,用以實現遠程文件操做,如文件上傳、下載、修改文件權限等操做。
Paramiko中的幾個基礎名詞:shell
Channel:一種類Socket,一種安全的SSH傳輸通道; Transport:一種加密的會話,使用時會同步建立了一個加密的Tunnels(通道),這個Tunnels叫作Channel; Session:client與Server保持鏈接的對象,用connect()/start_client()/start_server()開始會話。
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對象。
利用SSHClient對象的open_sftp()方法,能夠直接返回一個基於當前鏈接的sftp對象,能夠進行文件的上傳等操做。ssh
sftp = client.open_sftp() sftp.put('test.txt','text.txt')
import paramiko # 實例化SSHClient client = paramiko.SSHClient() # 自動添加策略,保存服務器的主機名和密鑰信息,若是不添加,那麼再也不本地know_hosts文件中記錄的主機將沒法鏈接 client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # 鏈接SSH服務端,以用戶名和密碼進行認證 client.connect(hostname='192.168.1.105', port=22, username='root', password='123456') # 打開一個Channel並執行命令 stdin, stdout, stderr = client.exec_command('df -h ') # stdout 爲正確輸出,stderr爲錯誤輸出,同時是有1個變量有值 # 打印執行結果 print(stdout.read().decode('utf-8')) # 關閉SSHClient client.close()
import paramiko class Paramiko: def __init__(self, ip, user, passwd, port): self.user = user self.ip = ip self.passwd = passwd self.port = port
# 登陸要測試的主機(linux主機) def type_login_root(self): self.ssh = paramiko.SSHClient() self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) self.ssh.connect(self.ip, self.port, self.user, self.passwd) # 執行linux命令 def exec_commands(self, cmd): stdin, stdout, stderr = self.ssh.exec_command(cmd) results = str(stdout.read()) return results # 批量順序執行 def CMD(self, cmd_list=[]): CmdDict = {} for c in cmd_list: CmdMes = self.exec_commands(c) print(CmdMes) CmdDict[c] = CmdMes return CmdDict
睿江雲官網連接:https://www.eflycloud.com/home?from=RJ0035tcp