python運維自動化Paramiko的實現(xiàn)示例

更新時間：2025年08月25日 11:26:05 作者：禿了也弱了。

Paramiko是Python的SSH庫,提供SSHClient和SFTPClient類,用于遠程命令執(zhí)行及文件傳輸,支持連接、執(zhí)行、上傳下載,適用于自動化部署等場景,下面就來詳細介紹一下Paramiko的使用

一、Paramiko

1、安裝

pip install paramiko

2、常用的類

SSHClient: 最常用的類，表示一個SSH客戶端連接。
SFTPClient: 用于SFTP傳輸操作。
Transport: 低級別的模塊，可用來實現(xiàn)SSH2協(xié)議。
SSHClient主要方法
connect(hostname, port=22, username=None, password=None, pkey=None): 用于連接到遠程服務器?？梢赃x擇使用用戶名和密碼或密鑰進行認證。
exec_command(command): 在遠程服務器上執(zhí)行指令。
open_sftp(): 返回一個SFTPClient對象，可用于文件的上傳與下載。
SFTPClient主要方法
get(remotepath, localpath, callback=None): 下載遠程文件。
put(localpath, remotepath, callback=None, confirm=True): 上傳本地文件。
listdir(path=“.”): 列出遠程目錄的內容。

3、使用

（1）SSH執(zhí)行遠程命令

import paramiko

# 創(chuàng)建SSH客戶端
client = paramiko.SSHClient()

# 自動添加未知的服務器密鑰及策略
'''
AutoAddPolicy：自動添加主機名及主機密鑰到本地的known_hosts，不依賴load_system_host_key的配置。即新建立ssh連接時不需要再輸入yes或no進行確認。最為常用。
WarningPolicy 用于記錄一個未知的主機密鑰的python警告。并接受，功能上和AutoAddPolicy類似，但是會提示是新連接。
RejectPolicy 自動拒絕未知的主機名和密鑰，依賴load_system_host_key的配置。此為默認選項
'''
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# 連接SSH服務端
'''
hostname(str類型)，連接的目標主機地址；
port(int類型)，連接目標主機的端口，默認為22；
username(str類型)，校驗的用戶名(默認為當前的本地用戶名)；
password(str類型)，密碼用于身份校驗或解鎖私鑰；
pkey(Pkey類型)，私鑰方式用于身份驗證；
key_filename(str or list(str)類型)，一個文件名或文件名列表，用于私鑰的身份驗證；
timeout(float類型)，一個可選的超時時間(以秒為單位)的TCP連接；
allow_agent(bool類型)，設置為False時用于禁用連接到SSH代理；
look_for_keys(bool類型)，設置為False時用于來禁用在～/.ssh中搜索私鑰文件；
compress(bool類型)，設置為True時打開壓縮。
'''
client.connect('192.168.56.10', port=22, username='root', password='vagrant')

# 執(zhí)行命令
stdin, stdout, stderr = client.exec_command('ls -l')

# 獲取命令執(zhí)行結果
result = stdout.read()

print(result.decode())

# 關閉連接
client.close()

（2）SFTP上傳/下載文件

import paramiko

# 創(chuàng)建SSH客戶端
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('192.168.56.10', port=22, username='root', password='vagrant')

# 創(chuàng)建SFTP會話
sftp = client.open_sftp()

# 上傳文件
sftp.put('localfile.txt', '/remote/path/remote.txt')

# 下載文件
sftp.get('/remote/path/remote.txt', 'localfile.txt')

'''
mkdir,在SFTP服務端創(chuàng)建目錄，如sftp.mkdir("/home/userdir",mode=0777),默認模式是0777(八進制)，在某些系統(tǒng)上，mode被忽略。在使用它的地方，當前的umask值首先被屏蔽掉。
remove,刪除SFTP服務端指定目錄，如sftp.remove("/home/userdir")。
rename,重命名SFTP服務端文件或目錄，如sftp.rename("/home/test.sh","/home/testfile.sh")
stat,獲取遠程SFTP服務端指定文件信息，如sftp.stat("/home/testfile.sh")。
listdir,獲取遠程SFTP服務端指定目錄列表，以Python的列表(List)形式返回，如sftp.listdir("/home")。
'''


# 關閉SFTP會話和SSH連接
sftp.close()
client.close()

（3）SSH簡單封裝

import paramiko

class SSHClient:
    def __init__(self, hostname, port=22, username=None, password=None, pkey=None, use_proxy=False, proxy_hostname=None,
                 proxy_port=None, proxy_username=None, proxy_password=None, timeout=5):
        """
        初始化SSH客戶端。

        :param hostname: 主機名或IP地址
        :param port: 端口號，默認是22
        :param username: 用戶名
        :param password: 密碼
        :param pkey: 私鑰文件對象，用于密鑰認證
        :param use_proxy: 是否使用代理標志
        :param proxy_hostname: 代理的主機名稱
        :param proxy_port: 代理的端口號
        :param proxy_username: 代理的用戶名
        :param proxy_password: 代理的密碼
        """
        self.hostname = hostname
        self.port = port
        self.username = username
        self.password = password
        self.pkey = pkey
        self.use_proxy = use_proxy
        self.proxy_hostname = proxy_hostname
        self.proxy_port = proxy_port
        self.proxy_username = proxy_username
        self.proxy_password = proxy_password
        self.client = None
        self.timeout = timeout

    def _setup_proxy(self):
        """
        配置代理。
        """
        if self.use_proxy:
            proxy_ssh = paramiko.SSHClient()
            proxy_ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            proxy_ssh.connect(hostname=self.proxy_hostname, port=self.proxy_port, username=self.proxy_username,
                              password=self.proxy_password, timeout=self.timeout)
            vm_transport = proxy_ssh.get_transport()
            remote_address = (self.hostname, self.port)
            local_address = (self.proxy_hostname, self.proxy_port)
            vm_channel = vm_transport.open_channel("direct-tcpip", remote_address, local_address)
            return vm_channel
        else:
            return None

    def _connect(self):
        """
        創(chuàng)建SSH連接。
        """
        self.client = paramiko.SSHClient()
        self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        proxy = self._setup_proxy()
        self.client.connect(self.hostname, port=self.port, username=self.username, password=self.password,
                            pkey=self.pkey, sock=proxy, timeout=self.timeout)

    def exec_command(self, command):
        """
        在遠程服務器上執(zhí)行命令。
        :param command: 要執(zhí)行的命令字符串
        :return: 命令輸出結果的標準輸出和標準錯誤
        """
        if self.client is None:
            self._connect()
        stdin, stdout, stderr = self.client.exec_command(command)
        return stdout.read().decode(), stderr.read().decode()

    def close(self):
        """
        關閉SSH連接。
        """
        if self.client:
            self.client.close()
            self.client = None


# 使用示例
if __name__ == "__main__":
    # 直接連接單個服務器
    '''
    ssh = SSHClient(hostname='192.168.56.10', port=22, username='root', password='vagrant')
    stdout, stderr = ssh.exec_command('ls -l')
    print(stdout)
    ssh.close()
    '''

    # 通過代理連接
    '''
    ssh_with_proxy = SSHClient(hostname='hostname', port=22, username='username', password='password', use_proxy=True,
                               proxy_hostname='proxy_hostname', proxy_port=22, proxy_username='proxy_username',
                               proxy_password='proxy_password')
    stdout, stderr = ssh_with_proxy.exec_command('ls -l')
    print(stdout)
    ssh_with_proxy.close()
    '''




    # 連接多個服務器執(zhí)行多個命令
    servers = {
        '192.168.56.10': {
            'port': 22,
            'username': 'root',
            'password': '123'
        },
        '192.168.56.11': {
            'port': 22,
            'username': 'root',
            'password': '123'
        },
        '192.168.56.12': {
            'port': 22,
            'username': 'root',
            'password': '123'
        },
    }
    
    commands = [
        'ls -l',
        'cd xxx',
    ]

    for ip, info in servers.items():
        ssh = SSHClient(hostname=ip, port=info.get('port'), username=info.get('username'), password=info.get('password'))
        for command in commands:
            stdout, stderr = ssh.exec_command(command)
            print(stdout)
        
        ssh.close()

（4）SFTP簡單封裝

#!/usr/bin/env python
# coding:utf-8
from stat import *
import os
import paramiko

class RemotLHost(object):
    '''封裝一個遠程Linux主機類，并將paramiko遠程上傳下載單個文件或目錄的行為封裝為其相應的方法'''

    # 通過IP, 用戶名，密碼，超時時間初始化一個遠程Linux主機
    def __init__(self, ip, username, password, timeout=30):
        self.ip = ip
        self.username = username
        self.password = password
        self.timeout = timeout
        # transport
        self.t = None
        # 鏈接失敗的重試次數(shù)
        self.conn_times = 3

    # get單個文件
    def sftp_get(self, remotefile, localfile):
        t = paramiko.Transport((self.ip, 22))
        t.connect(username=self.username, password=self.password)
        sftp = paramiko.SFTPClient.from_transport(t)
        sftp.get(remotefile, localfile)
        t.close()

    # put單個文件
    def sftp_put(self, localfile, remotefile):
        t = paramiko.Transport((self.ip, 22))
        t.connect(username=self.username, password=self.password)
        sftp = paramiko.SFTPClient.from_transport(t)
        sftp.put(localfile, remotefile)
        t.close()

    # 獲取遠端linux主機上指定目錄及其子目錄下的所有文件
    def __get_all_files_in_remote_dir(self, sftp, remote_dir):
        # 保存所有文件的列表
        all_files = list()

        # 去掉路徑字符串最后的字符'/'，如果有的話
        if remote_dir[-1] == '/':
            remote_dir = remote_dir[0:-1]

        # 獲取當前指定目錄下的所有目錄及文件，包含屬性值
        files = sftp.listdir_attr(remote_dir)
        for x in files:
            
            # remote_dir目錄中每一個文件或目錄的完整路徑
            filename = remote_dir + '/' + x.filename
            print("remote file is "+ filename)
            # 如果是目錄，則遞歸處理該目錄，這里用到了stat庫中的S_ISDIR方法，與linux中的宏的名字完全一致
            if S_ISDIR(x.st_mode):
                all_files.extend(
                    self.__get_all_files_in_remote_dir(sftp, filename))
            else:
                all_files.append(filename)
        return all_files

    def sftp_get_dir(self, remote_dir, local_dir):
        t = paramiko.Transport((self.ip, 22))
        t.connect(username=self.username, password=self.password)
        sftp = paramiko.SFTPClient.from_transport(t)

        # 獲取遠端linux主機上指定目錄及其子目錄下的所有文件
        all_files = self.__get_all_files_in_remote_dir(sftp, remote_dir)
        # 依次get每一個文件
        for x in all_files:
            filename = x.replace(remote_dir, "")
            if os.name=="nt":
                filename = filename.replace('/',os.sep).lstrip(os.sep)
            local_filename = os.path.join(local_dir, filename)
            filepath=os.sep.join(local_filename.split(os.sep)[0:-1])

            if not os.path.exists(filepath):
                os.mkdir(filepath)
           
            print(u'Get  <------- %s'% x)
            sftp.get(x, local_filename)

    #獲取本地指定目錄及其子目錄下的所有文件
    def __get_all_files_in_local_dir(self, local_dir):
        # 保存所有文件的列表
        all_files = list()

        # 獲取當前指定目錄下的所有目錄及文件，包含屬性值
        files = os.listdir(local_dir)
        for x in files:
            # local_dir目錄中每一個文件或目錄的完整路徑
            filename = os.path.join(local_dir, x)
            # 如果是目錄，則遞歸處理該目錄
            if os.path.isdir(filename):
                all_files.extend(self.__get_all_files_in_local_dir(filename))
            else:
                all_files.append(filename)
        return all_files

    def sftp_put_dir(self, local_dir, remote_dir):
        if remote_dir[-1] != '/':
            remote_dir = remote_dir + "/"
        t = paramiko.Transport((self.ip, 22))
        t.connect(username=self.username, password=self.password)
        sftp = paramiko.SFTPClient.from_transport(t)
        for root, dirs, files in os.walk(local_dir):
            for filespath in files:
                local_file = os.path.join(root, filespath)
                a=local_file.replace(local_dir, '')
                if os.name=="nt":
                    a = a.replace('\\','/').lstrip('/')

                remote_file = os.path.join(remote_dir, a)
                try:
                    sftp.put(local_file,remote_file)
                except Exception as e:
                    sftp.mkdir(os.path.split(remote_file)[0])
                    sftp.put(local_file,remote_file)
                print("Put %s to remote %s" % (local_file, remote_file))
            for name in dirs:
                local_path = os.path.join(root, name)
                
                a = local_path.replace(local_dir, '')
                if os.name=="nt":
                    a = a.replace('\\','/').lstrip('/')

                remote_path = os.path.join(remote_dir, a)
                try:
                    sftp.mkdir(remote_path)
                    print("mkdir path %s" % remote_path)
                except Exception as e:
                    print(e)

        t.close()

if __name__ == '__main__':


    host = RemotLHost('192.168.56.10', 'root', 'vagrant')

    remote_path = '/root/doc'
    local_path = 'D:\\test'

    # 將遠端remote_path目錄中的所有文件get到本地local_path目錄
    host.sftp_get_dir(remote_path, local_path)
    # 將本地local_path目錄中的所有文件put到遠端remote_path目錄
    local_path= 'D:\ltest'
    remote_path = '/root/doc2'
    host.sftp_put_dir(local_path, remote_path)

4、工具類

（1）自動部署

import sys
import time

from ssh_client import SSHClient
from sftp_client import RemotLHost

if __name__ == '__main__':
    servers = {
        '192.168.1.1': {
            'port': 22,
            'username': 'root',
            'password': ''
        },
    }

    for ip, info in servers.items():
        print('begin============================' + ip + '============================')
        ssh = SSHClient(hostname=ip, port=info.get('port'), username=info.get('username'),
                        password=info.get('password'))
        sftp = RemotLHost(ip=ip, username=info.get('username'), password=info.get('password'))



        stdout, stderr = ssh.exec_command('cd /data;sh stop.sh')
        print(stdout)
        print('stop success')
        time.sleep(1)

        remote_path = '/data/app.jar'
        local_path = 'E:\\app.jar'
        # 將遠端remote_path目錄中的所有文件get到本地local_path目錄
        sftp.sftp_put(local_path, remote_path)
        print('put success')
        time.sleep(1)


        stdout, stderr = ssh.exec_command('cd /data;sh start.sh')
        print(stdout)
        print('start success')
        time.sleep(10)

        stdout, stderr = ssh.exec_command('curl http://127.0.0.1:8081/check/version')
        print(stdout)


        print('end============================' + ip + '============================')

        ssh.close()

到此這篇關于python運維自動化Paramiko的實現(xiàn)示例的文章就介紹到這了,更多相關python Paramiko內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: