簡介:
paramiko 是 python 下對 ssh(v2) 協議封裝的一個庫, 能夠用於實現客戶端或者服務器端的一些功能。本章節主要講述如何實現客戶端功能python
安裝:
pip install paramikoios
經常使用組件:
Channel 實現 ssh 通道創建和維護功能
Client 實現 ssh 客戶端功能
SFTP 實現 sftp 功能
Transport 實現 ssh 客戶端和服務器端數據傳輸功能git
Client 組件主要有如下幾個 class: SSHClient 對 Transport 的高級封裝, 實現了傳輸、通道、SFTP 等方法 如下幾個 class 實現 ssh 新主機密鑰管理策略: AutoAddPolicy 自動添加主機名和祕鑰到 HostKeys 對象(默認爲 known_hosts) RejectPolicy 缺省值, 自動拒絕未知的主機和祕鑰 WarningPolicy 相似於 AutoAddPolicy 但會發出一個警告 paramiko.client.SSHClient 經常使用方法: load_system_host_keys() # 加載 known_hosts 文件默認爲 ~/.ssh/known_hosts set_missing_host_key_policy() # 設置新主機和密碼管理策略, 接受一個參數, 值能夠是 AutoAddPolicy、RejectPolicy、WarningPolicy, 默認爲 RejectPolicy connect() # 創建服務器和客戶端之間的鏈接 參數: hostname # 遠程主機的地址 port=22 # 遠程主機 sshd 監聽的端口, 默認 22 username=None # 要進行身份驗證的用戶名(默認爲當前系統用戶的用戶名) password=None # 用戶名對應的密碼, 使用祕鑰登陸時若是 passphrase 沒有給定值而 password 給定了值時, password 將做爲 key 的解密密碼 passphrase=None # ssh key 的解密密碼 pkey=None # 指定 ssh 祕鑰路徑(若是該選項不爲 None , 將啓用 ssh 祕鑰認證) key_filename=None # 嘗試進行身份驗證的可選祕鑰文件或者文件列表 timeout=None # TCP 鏈接超時時間 allow_agent=True # 是否容許鏈接到 ssh 代理, 默認容許 look_for_keys=True # 是否容許 paramiko 在當前系統的 ~/.ssh/ 中搜索祕鑰用於嘗試認證, 默認容許 compress=False # 是否打開壓縮 banner_timeout=None # 等待顯示 ssh 標題的超時時間 auth_timeout=None # 等待身份認證的超時時間 exec_command() # 執行遠程命令 參數: command # 執行指定的 shell 命令 timeout # 設置命令超時時間 environment # 設置 shell 環境變量, 字典類型 返回值: stdint # 標準輸入 stdout # 標準輸出 stderr # 標準錯誤 get_transport() # 獲取底層的 transport 對象, 用於執行低級的任務 返回值: 返回 Transport 對象 invoke_shell() # 打開一個 ssh 交互式會話 參數: term # 模擬終端的類型 width # 終端窗口的寬度(以字符爲單位) height # 終端窗口的高度(以字符爲單位) width_pixels # 終端窗口的寬度(以像素爲單位) height_pixels # 終端窗口的高度(以像素爲單位) environment # 設置 shell 環境變量, 字典類型 open_sftp() # 向 ssh 服務器申請打開 sftp close() # 斷開和服務器的鏈接 異常: BadHostKeyException 沒法驗證服務器的主機密鑰 AuthenticationException 認證失敗 SSHException 鏈接或創建 SSH 會話時出現任何其餘錯誤 socket.error 鏈接時發生套接字錯誤
Client 示例:github
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Author : HuYuan # @File : paramiko_ssh_client.py import paramiko class UseSSHClient: def __init__(self, hostname, username, port=22, password=None, pkey=None, timeout=30, passphrase=None): self.hostname = hostname self.username = username self.port = port self.password = password self.pkey = pkey self.timeout = timeout self.passphrase = passphrase self._open_ssh() def get_key_obj(self, pkeyobj, pkey_file=None, pkey_obj=None, password=None): if pkey_file: with open(pkey_file) as fo: try: pkey = pkeyobj.from_private_key(fo, password=password) return pkey except: pass else: try: pkey = pkeyobj.from_private_key(pkey_obj, password=password) return pkey except: pkey_obj.seek(0) def _open_ssh(self): sshclient = paramiko.client.SSHClient() sshclient.set_missing_host_key_policy(paramiko.client.AutoAddPolicy) if self.password: sshclient.connect(hostname=self.hostname, username=self.username, port=self.port, password=self.password, timeout=self.timeout) else: # 解析 key pkey = get_key_obj(paramiko.RSAKey, pkey_file=self.pkey) or \ get_key_obj(paramiko.DSSKey, pkey_file=self.pkey) or \ get_key_obj(paramiko.ECDSAKey, pkey_file=self.pkey) or \ get_key_obj(paramiko.Ed25519Key, pkey_file=self.pkey) sshclient.connect(hostname=self.hostname, username=self.username, pkey=pkey, port=self.port, timeout=self.timeout) self.ssh = sshclient def exec_comd(self, cmd): result = self.ssh.exec_command(cmd) return result def sftp_get(self, server_path, local_path): sftp = self.ssh.open_sftp() sftp.get(server_path, local_path) def sftp_put(self, server_path, local_path): sftp = self.ssh.open_sftp() sftp.put(local_path, server_path) def close(self): self.ssh.close() if __name__ == '__main__': hostname = '192.168.1.100' username = 'root' password = '123.com' sshClient = UseSSHClient(hostname=hostname, username=username, password=password) sshClient.sftp_get('/etc/fstab', 'fstab') sshClient.sftp_put('/tmp/fstab', 'fstab') stdin, stdout, stderr = sshClient.exec_comd('ls /tmp/fstab') print(stdout.read()) sshClient.close()
paramiko 目前支持 4 總 key 認證, 分別爲: RSAKey --> rsa 加密 --> 實現 class paramiko.rsakey.RSAKey DSSKey --> dsa 加密 --> 實現 class paramiko.dsskey.DSSKey ECDSAKey --> ecdsa 加密 --> 實現 class paramiko.ecdsakey.ECDSAKey Ed25519Key --> ed25519 加密 --> 實現 class paramiko.ed25519key.Ed25519Key 他們擁有相同的方法和屬性: from_private_key() # 從文件(或類文件) 對象中讀取私鑰來建立密鑰對象 參數: file_obj # 文件或類文件對象 password=None # 若是 password 不爲 None, 則 password 值做爲私鑰的解密密碼 返回值: 返回密鑰對象, 傳遞給相似 connect() 的參數, 進行身份驗證 from_private_key_file() # 從文件中讀取私鑰來建立密鑰對象 參數: filename # 密鑰文件 password=None # 若是 password 不爲 None, 則 password 值做爲私鑰的解密密碼 返回值: 返回密鑰對象, 傳遞給相似 connect() 的參數, 進行身份驗證 write_private_key() # 將私鑰內容寫入文件(或類文件)對象 參數: file_obj # 文件或類文件對象 password=None # 若是 password 不爲 None, 則在寫入以前使用 password 指定的密碼對密鑰進行加密 write_private_key_file() # 將私鑰內容寫入文件 參數: filename # 密鑰文件 password=None # 若是 password 不爲 None, 則在寫入以前使用 password 指定的密碼對密鑰進行加密 get_name() # 獲取密鑰文件的名稱 get_base64() # 獲取密鑰公共部分的 base64 字符串 get_bits() # 返回此鍵中的有效位數, 這對於判斷密鑰的相對安全性頗有用 get_fingerprint() # 獲取密鑰公共部分的 MD5 指紋
組件: paramiko.sftp_client.SFTPClient 用於實現 sftp client 功能 paramiko.sftp_client.SFTP SFTPClient 用於向後兼容的別名 經常使用方法: from_transport() # 從 Transport 對象中打開 sftp 會話 put() # 上傳文件 參數: remotepath # 遠程路徑 localpath # 本地路徑 callback # 回掉函數,獲取已接收的字節數和總字節數 confirm # 文件傳輸完成以後是否調用stat()函數 以確認文件大小,默認爲True get() # 下載文件 參數: remotepath # 遠程路徑 localpath # 本地路徑 callback # 回掉函數,獲取已接收的字節數和總字節數 getfo() # 下載文件, 本地打開一個文件句柄用於寫入遠程服務的數據 參數: remotepath # 遠程路徑 fl # 本地文件句柄 callback putfo() # 上傳文件, 參數相似於 getfo() getcwd() # 獲取當前 sftp 會話所在目錄 listdir() # 列出指定路徑下的全部文件目錄列表(包括以隱藏文件), 默認 sftp 會話所在目錄 參數: path # 指定遠程路徑 listdir_attr() # 以相似於 ls -l 的格式列出指定路徑下的全部文件目錄列表(包括以隱藏文件), 默認 sftp 會話所在目錄 參數: path # 指定遠程路徑 mkdir() # 在遠程服務器上建立目錄 參數: path # 要建立的目錄路徑和名稱 mode # 目錄權限, 數字格式, 默認爲 o777 open() # 在遠程服務器上打開文件, 參數和 python 的 open() 函數相同 readlink() # 返回符號連接的原始路徑 參數: path # 指定符號連接 symlink() # 建立符號連接 參數: source # 符號連接的原始路徑 dest # 符號連接的目標路徑 remove() # 刪除指定的文件(不能對目錄進行刪除) 參數: path # 指定須要刪除的文件 rmdir() # 刪除指定目錄, 參數和 remove() 相同 rename() # 重命名文件或目錄 參數: oldpath # 指定須要重命名的文件 newpath # 重命名以後的名稱 stat() # 檢查遠程系統上的指定文件的信息 參數: path # 須要檢查的文件 utime() # 修改文件的 atime 和 mtime 參數: path # 指定的文件 times # 修改後的時間元組, 格式 (atime, mtime)
paramiko.channel.Channel 對 paramiko 的底層 class 之一, 實現通道創建和維護等功能 經常使用方法: exec_command() # 在遠程服務器上執行命令 fileno() # 返回 OS 級文件描述符, 可用於輪詢, 但不能用於讀取或寫入, 這主要是爲了讓 Python 的 select 模塊可以工做(此方法將致使 Channel 讀取效率下降) get_id() # 獲取通道 ID set_name() # 設置通道名稱 get_name() # 獲取通道名稱 get_pty() # 向服務器請求一個 pty 終端 參數: term="vt100" # 終端類型 width # 終端窗口的寬度(以字符爲單位) height # 終端窗口的高度(以字符爲單位) width_pixels # 終端窗口的寬度(以像素爲單位) height_pixels # 終端窗口的高度(以像素爲單位) get_transport() # 獲取底層的 transport 對象, 用於執行低級的任務 getpeername() # 獲取服務器地址 settimeout() # 設置通道超時時間 gettimeout() # 查看通道超時時間 invoke_shell() # 在此 channel 上請求交互式 shell 會話(一般會和 get_pty() 一塊兒使用) invoke_subsystem() # 請求服務器上的子系統 參數: subsystem # 子系統的名稱, 好比: sftp recv() # 接收遠程服務器返回的數據 參數: nbytes # 設置一次獲取的最大字節數 recv_exit_status() # 獲取服務上進程返回的退出狀態, 在使用 exec_command 執行命令獲取返回值時有用 request_forward_agent() # 請求轉發 ssh 代理 request_x11() # 請求 x11 會話 resize_pty() # 從新設置 pty 的大小, 用於更改 get_pty() 設置的大小 參數: width # 終端窗口的寬度(以字符爲單位) height # 終端窗口的高度(以字符爲單位) width_pixels # 終端窗口的寬度(以像素爲單位) height_pixels # 終端窗口的高度(以像素爲單位) send() # 向服務器發送數據, 檢查數據是否發送完畢, 若是僅傳輸了部分數據, 則嘗試傳送剩餘數據 參數: s # 要發送的數據 sendall() # 向服務器發送數據, 直到全部數據都已發送或發生錯誤 參數: s # 要發送的數據 set_environment_variable() # 設置環境變量 參數: name # 變量名 value # 變量值 update_environment() # 更新環境變量 參數: environment # 須要更新的環境變量的值, 類型爲 dict setblocking() # 設置通道的阻塞模式: 參數: blocking # 若是 blocking 爲 0, 則將通道設置爲非阻塞模式; 不然設置爲阻止模式, 默認爲阻塞模式 settimeout() # 設置阻塞讀寫的超時時間, 若是 setblocking() 設置爲 0, 那麼至關於 settimeout(0) 參數: timeout # 超時時間 shutdown() # 關閉通道 how # 怎麼樣關閉通道, 0: 中止接收數據、1: 中止發送數據、2: 中止接收和發送數據 shutdown_read() # shutdown(0) 的簡寫 shutdown_write() # shutdown(1) 的簡寫 close() # 斷開和服務器的鏈接
paramiko.transport.Transport paramiko 的核心主件, 實現了一系列對 ssh 底層的操做 經常使用方法: connect() # 協商會話, 並可選的進行身份認證 參數: username="" # 要進行身份驗證的用戶名 password=None # 用戶名對應的密碼 pkey=None # 使用祕鑰認證 # 若是 connect() 方法只是進行了會話協商而沒有進行身份驗證時, 則能夠使用如下方法進行身份驗證 auth_password() # 使用密碼認證 參數: username # 進行身份驗證的用戶 password # 用戶密碼 auth_publickey() # 使用祕鑰認證 參數: username # 進行身份驗證的用戶 key # 用戶祕鑰 auth_none() # 嘗試使用空密碼登陸, 因爲 Linux 的密碼策略因此該方法幾乎都是失敗 參數: username # 用戶名 close() # 斷開和服務器的鏈接 get_username() # 獲取登陸用戶的用戶名 getpeername() # 獲取服務器地址 is_active() # 若是當前會話處於活動狀態則返回 True, 反之則返回 false is_authenticated() # 若是當前會話處於活動狀態且已經過身份驗證則返回 True, 反之則返回 false open_channel() # 向服務器請求打開新的 channel 參數: kind # 打開通道的類型(session, forwarded-tcpip, direct-tcpip, x11) dest_addr=None # 端口轉發的目標地址(ip, port), 只有當通道類型爲 forwarded-tcpip 和 direct-tcpip 時生效 src_addr=None # 端口轉發的源地址, 只有當通道類型爲 forwarded-tcpip、direct-tcpip 或 x11 時生效 window_size=None # 會話的窗口大小 max_packet_size=None # 此會話的最大數據包大小 timeout=None # 打開通道的超時時間, 默認問 1h open_forwarded_tcpip_channel() # open_channel() forwarded-tcpip 的簡寫 open_session() # open_channel() session 的簡寫 open_x11_channel() # open_channel() x11 的簡寫 open_sftp_client() # 打開 sftp 通道 set_keepalive() # 打開/關閉 keepalive 數據包(默認爲關閉), 若是設置了此值, 在 interval 指定的時間內沒有數據傳輸將發送 keepalive 數據包 參數: interval # 指定多長時間沒有數據傳輸後開始發送 keepalive use_compression() # 打開或關閉壓縮, 建議關閉, 由於它會對交互式會話產生負面影響 參數: compress # 是否打開壓縮, true/false
Transport 示例:web
import paramiko hostname = '192.168.1.100' username = 'root' password = '123.com' port = 22 transport = paramiko.transport.Transport(sock=(hostname, port)) transport.connect() transport.auth_password('root','123.com') sftp = transport.open_sftp_client() sftp.get('/etc/fstab', 'fstab-transport')
示例: 實現 ssh 客戶端(只能在 Linux 下運行)shell
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Author : HuYuan # @File : python_shell.py import paramiko import sys import socket import termios import tty import select def posix_shell(chan): local_tty = termios.tcgetattr(sys.stdin) try: tty.setraw(sys.stdin.fileno()) tty.setcbreak(sys.stdin.fileno()) chan.settimeout(0.0) while True: read, write, error = select.select([chan, sys.stdin], [], []) if chan in read: try: recv = chan.recv(1024) if not len(recv): break sys.stdout.write(recv.decode()) sys.stdout.flush() except socket.timeout: pass if sys.stdin in read: content = sys.stdin.read(1) if not len(content): break chan.send(content) finally: termios.tcsetattr(sys.stdin, termios.TCSADRAIN, local_tty) def open_ssh_client(): hostname = '192.168.1.100' username = 'root' password = '123.com' port = 22 try: ssh_client = paramiko.client.SSHClient() ssh_client.set_missing_host_key_policy(paramiko.client.AutoAddPolicy()) ssh_client.connect(hostname=hostname, username=username, password=password, port=port, timeout=10) tran = ssh_client.get_transport() chan = tran.open_session() chan.get_pty() chan.invoke_shell() posix_shell(chan) tran.close() except socket.timeout as e: print('鏈接超時', e) exit(10) except Exception as e: print(e) if __name__ == '__main__': open_ssh_client()
相關項目:django
django + paramiko + websocket 實現 webssh: 項目地址: https://github.com/huyuan1999/django-webssh