簡介:
paramiko 是 python 下對 ssh(v2) 協議封裝的一個庫, 可以用於實現客戶端或者服務器端的一些功能。本章節主要講述如何實現客戶端功能
安裝:
pip install paramiko
常用組件:
Channel 實現 ssh 通道建立和維護功能
Client 實現 ssh 客戶端功能
SFTP 實現 sftp 功能
Transport 實現 ssh 客戶端和服務器端數據傳輸功能
Client:
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 示例:
#!/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()
pkey 對象:
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 指紋
SFTP:
組件:
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)
Channel:
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() # 斷開和服務器的連接
Transport:
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 示例:
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 下運行)
#!/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 + paramiko + websocket 實現 webssh: 項目地址: https://github.com/huyuan1999/django-webssh