SSH連接主機

目的

使用python實現(xiàn)ssh連接特定主機，可以發(fā)送命令和接收命令回顯，并具備日志功能。

需求

根據(jù)目的可分解為4個需求

1. ssh連接特定主機
2. 發(fā)送命令
3. 回顯命令—處理輸出以及異常
4. 日志功能

實現(xiàn)

Ssh.py

維護一個Ssh類

import paramiko
from lib.common.Logger import Logger
import time
import re

class ClassCli:
    def __init__(self, ip: str, username: str, passwd: str):
        self.ip = ip
        self.username = username
        self.passwd = passwd
        self.SSHConnect = self.__connect()
        self.logger = Logger(self.__class__.__name__)
        self.shell = None

    def __connect(self):
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())  # 自動添加主機密鑰
        ssh.connect(self.ip, username=self.username, password=self.passwd)
        return ssh


    def open_shell(self):
        # 模擬交互式終端
        self.shell = self.SSHConnect.invoke_shell()
        time.sleep(1)  # 等待終端初始化完成

        # 讀取并記錄登錄信息
        login_info = ""
        while self.shell.recv_ready():
            login_info += self.shell.recv(1024).decode('utf-8', 'ignore')
        self.logger.info("Login Information:\n" + login_info)

    def exec_command(self, cmd: str, waitstr: str = r'[#$]', timeout: int = 10):
        if not self.shell:
            self.open_shell()  # 如果shell未創(chuàng)建，則創(chuàng)建并記錄登錄信息

        # 發(fā)送命令
        self.shell.send(f"{cmd}\n")
        # 等待特定字符串出現(xiàn)
        start_time = time.time()
        output = ""
        while True:
            if self.shell.recv_ready():
                output += self.shell.recv(1024).decode('utf-8', 'ignore')
                if re.search(waitstr, output):
                    break
            if time.time() - start_time > timeout:
                raise TimeoutError(f"Command execution timed out after {timeout} seconds")
            time.sleep(0.1)  # 稍作等待，避免CPU占用過高

        self.logger.info(output)
        return output

    def close(self):
        # 關閉SSH連接
        self.SSHConnect.close()

Logger.py

通過logging庫建立一個日志記錄器

import logging

class Logger:
    def  __init__(self, __name__):
        # 獲取日志記錄器
        self.logger = logging.getLogger(__name__)
        self.logger.setLevel(logging.DEBUG)
        # 獲取控制臺處理器
        self.handler = logging.StreamHandler()
        self.handler.setLevel(logging.DEBUG)
        self.formatter = logging.Formatter('[%(levelname)s]:%(asctime)s - %(filename)s:%(lineno)s: %(message)s')
        self.handler.setFormatter(self.formatter)
        self.logger.addHandler(self.handler)
        pass

    def info(self, str):
        return self.logger.info(str)
    
    def error(self, str):
        return self.logger.error(str)

    def debug(self, str):
        return self.logger.debug(str)    

main.py

測試

from Logger import Logger
from Ssh  import ClassCli

logger = Logger(__name__)

node = ClassCli("192.168.137.131", "root", "123456")
node.exec_command("ls")
logger.info("cmd is : ls")
logger.error("cmd is : ls")
logger.debug("cmd is : ls")
node.close()