用 Python 開嗎主動式裝置連線狀態監控系統

身為網路管理者，隨時掌握網路裝置的連線狀態至關重要。本文將引導您使用 Python 建立一個主動式監控系統，結合 Docker 的便利性與 Sendmail 的通知功能，實作高效的網路管理。

Python 網路自動化的優勢

Python 擁有豐富的網路函式庫，例如 socket 和 smtplib，簡化了網路程式設計的複雜度，同時具備跨平台特性，讓您的監控程式碼在不同作業系統上都能順暢執行。

實戰演練：建構網路監控程式

以下步驟將逐步引導您在 Docker 環境中建立 Python 網路監控程式。

步驟一：設定 Docker 環境

首先，請確認您的系統已安裝 Docker，接著啟動一個 Ubuntu 20.04 的 Docker 容器：

docker run -it --name pynetauto_ubuntu20sendmail -p 20-22:20-22 -p 25:25 -p 12020-12025:12020-12025 pynetauto/pynetauto_ubuntu20 /bin/bash

步驟二：建立專案目錄

在 Docker 容器中，建立並進入 monitoring 目錄：

mkdir /home/monitoring && cd /home/monitoring

步驟三：建立 Python 檔案

在 monitoring 目錄下，建立 monitor.py 和 email_notifier.py 兩個檔案：

touch monitor.py email_notifier.py

步驟四：編寫連線檢查程式碼 (monitor.py)

首先，建立 monitor.py，用於檢查目標裝置的 SSH 連線埠 (port 22) 是否開啟：

import socket
import time
from datetime import datetime

TARGET_IP = '192.168.1.10'  # 目標裝置 IP 位址
TARGET_PORT = 22  # SSH 連線埠
TIMEOUT = 3  # 連線逾時時間（秒）
MAX_FAILURES = 10 # 最大連線失敗次數

def check_connection(ip, port):
    try:
        with socket.create_connection((ip, port), timeout=TIMEOUT) as s:
            return True
    except OSError:
        return False

def monitor_device():
    failure_count = 0
    while True:
        with open('monitoring_log.txt', 'a') as log_file:
            current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            if check_connection(TARGET_IP, TARGET_PORT):
                log_file.write(f"{current_time} 連線埠 {TARGET_PORT} 已開啟\n")
                print(f"連線埠 {TARGET_PORT} 已開啟")
                failure_count = 0
            else:
                failure_count += 1
                log_file.write(f"{current_time} 連線埠 {TARGET_PORT} 已關閉\n")
                print(f"連線埠 {TARGET_PORT} 已關閉")
                if failure_count >= MAX_FAILURES:
                    send_alert()  # 傳送警示通知
                    failure_count = 0 # 重置計數器
            time.sleep(3) # 間隔三秒檢查

def send_alert():
    from email_notifier import send_email
    send_email("您的裝置連線中斷！", "請立即檢查您的裝置連線狀態。") # 傳送警示郵件

if __name__ == "__main__":
    monitor_device()

此程式碼使用 socket.create_connection() 函式檢查連線狀態，並將結果記錄至 monitoring_log.txt 檔案。若連線失敗次數達到 MAX_FAILURES，則呼叫 send_alert() 函式傳送警示。

步驟五：編寫電子郵件通知程式碼 (email_notifier.py)

接著，建立 email_notifier.py，負責傳送電子郵件警示：

import smtplib
from email.mime.text import MIMEText

def send_email(subject, message):
    sender_email = "your_sender_email@gmail.com"  # 寄件者 email
    receiver_email = "your_receiver_email@example.com"  # 收件者 email
    password = "your_email_password"  # 您的 email 密碼或應用程式密碼

    msg = MIMEText(message)
    msg["Subject"] = subject
    msg["From"] = sender_email
    msg["To"] = receiver_email

    try:
        with smtplib.SMTP_SSL("smtp.gmail.com", 465) as server:
            server.login(sender_email, password)
            server.send_message(msg)
        print("已傳送電子郵件通知")
    except Exception as e:
        print(f"傳送電子郵件失敗: {e}")

此程式碼使用 smtplib 函式庫傳送電子郵件通知。請將 sender_email、receiver_email 和 password 替換為您的實際資訊。

  graph LR
    B[B]
    D[D]
    No[No]
    Yes[Yes]
    A[monitor.py] --> B{連線正常?};
    B -- Yes --> C[持續監控];
    B -- No --> D{連續失敗次數 >= 10?};
    D -- Yes --> E[email_notifier.py];
    E --> F[傳送 Email 通知];
    D -- No --> C;

圖表説明: 此流程圖展示了監控程式的運作邏輯，清晰呈現了程式如何根據連線狀態執行不同操作。

這個案例示範瞭如何使用 Python 建立一個主動式網路裝置連線狀態監控系統。透過結合 Docker 和 Sendmail，您可以輕鬆地佈署和管理您的監控系統。

這個架構可以進一步擴充套件，例如整合其他監控指標，例如 CPU 使用率、記憶體使用量等，並根據不同的警示級別傳送不同型別的通知，例如 SMS 或 Slack 訊息。透過持續的改進和最佳化，您可以開發一個更加完善和強大的網路監控系統。

import subprocess
import re
import time
from twilio.rest import Client
from easysnmp import Session

# --- 設定區段 ---
ROUTER_IP = "192.168.1.1"  # 路由器 IP 位址，請修改為你的裝置 IP
SNMP_COMMUNITY = "public"  # SNMP Community 字串，請修改為你的裝置設定
CPU_THRESHOLD = 90  # CPU 使用率閾值 (%)
CHECK_INTERVAL = 300  # 檢查間隔 (秒)

# Twilio 設定 (請替換為你的帳號資訊)
TWILIO_ACCOUNT_SID = "ACxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
TWILIO_AUTH_TOKEN = "your_auth_token"
TWILIO_PHONE_NUMBER = "+11234567890"
MY_PHONE_NUMBER = "+886912345678"

# --- SNMP 查詢函式 ---
def get_cpu_utilization(router_ip, snmp_community):
    try:
        session = Session(hostname=router_ip, community=snmp_community, version=2)
        cpu_oid = '1.3.6.1.4.1.9.9.109.1.1.1.1.5' # 五分鐘平均 CPU 使用率 OID
        cpu_utilization = int(session.get(cpu_oid).value)
        return cpu_utilization
    except Exception as e:
        print(f"SNMP 查詢錯誤: {e}")
        return None

# --- Twilio 簡訊傳送函式 ---
def send_sms_alert(message):
    try:
        client = Client(TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN)
        message = client.messages.create(
            body=message,
            from_=TWILIO_PHONE_NUMBER,
            to=MY_PHONE_NUMBER
        )
        print(f"已傳送簡訊警示 (SID: {message.sid})")
    except Exception as e:
        print(f"簡訊傳送失敗: {e}")

# --- CPU 監控主迴圈 ---
if __name__ == "__main__":
    while True:
        cpu_usage = get_cpu_utilization(ROUTER_IP, SNMP_COMMUNITY)

        if cpu_usage is not None:
            current_time = time.strftime("%Y-%m-%d %H:%M:%S")
            print(f"{current_time} - 路由器 {ROUTER_IP} CPU 使用率: {cpu_usage}%")

            if cpu_usage > CPU_THRESHOLD:
                alert_message = f"警告：路由器 {ROUTER_IP} CPU 使用率過高 ({cpu_usage}%)！"
                send_sms_alert(alert_message)

        time.sleep(CHECK_INTERVAL)

此程式碼整合了 SNMP 和 Twilio 的功能，實作了 CPU 負載監控和告警。它利用 easysnmp 函式庫簡化了 SNMP 查詢的過程，並使用 twilio 函式庫傳送 SMS 警示。程式碼中包含了錯誤處理機制，以應對 SNMP 查詢或簡訊傳送失敗的情況。設定區段允許使用者自定義路由器 IP、SNMP Community、CPU 閾值、檢查間隔以及 Twilio 帳號資訊。主迴圈持續監控 CPU 使用率，並在超過閾值時傳送警示。

  graph LR
    A[啟動監控程式] --> B{SNMP 查詢 CPU 使用率}
    B -- 成功 --> C{CPU 使用率 > 閾值?}
    C -- 是 --> D[傳送 Twilio SMS 警示]
    C -- 否 --> E[等待下一次檢查]
    B -- 失敗 --> F[記錄錯誤]
    D --> E
    F --> E

圖表説明: 此流程圖描述了 CPU 監控程式的執行流程。程式首先透過 SNMP 查詢 CPU 使用率。如果查詢成功，則檢查 CPU 使用率是否超過設定的閾值。如果超過閾值，則傳送 Twilio SMS 警示。無論查詢成功與否，程式都會等待一段時間後再次執行檢查。

這個程式碼範例提供了一個更完善與易於理解的 CPU 監控解決方案，並包含了詳細的註解和錯誤處理機制，更貼近實際應用場景。

這個版本更精簡與功能更強大，並符合玄貓的程式碼風格和註解規範。

import subprocess
import os
import time
from twilio.rest import Client  # 匯入 Twilio 模組

# 設定 Twilio 帳戶 SID 和 Auth Token
account_sid = "ACxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"  # 替換為您的帳戶 SID
auth_token = "your_auth_token"  # 替換為您的 Auth Token
twilio_phone_number = "+11234567890"  # 替換為您的 Twilio 電話號碼
recipient_phone_number = "+886912345678"  # 替換為接收簡訊的手機號碼


class NetworkDevice:  # 基礎類別，定義網路裝置的共同屬性
    def __init__(self, ip_address, hostname, device_type):
        self.ip_address = ip_address
        self.hostname = hostname
        self.device_type = device_type

    def connect(self):  # 連線至裝置
        print(f"連線至 {self.device_type}: {self.hostname} ({self.ip_address})")

    def get_cpu_utilization(self):  # 取得 CPU 使用率
        try:
            output = subprocess.check_output(["snmpwalk", "-v", "2c", "-c", "public", self.ip_address, "iso.3.6.1.4.1.9.9.109.1.1.1.1.5.1"], stderr=subprocess.STDOUT, text=True, timeout=5)  # 設定 timeout
            cpu_util = int(output.split("Gauge32: ")[1].strip())
            return cpu_util
        except subprocess.TimeoutExpired:
            return "Timeout: No Response from {}".format(self.ip_address)
        except (subprocess.CalledProcessError, IndexError, ValueError):
            return "Error: Could not retrieve CPU utilization from {}".format(self.ip_address)


class Router(NetworkDevice):  # 路由器類別，繼承自 NetworkDevice
    def __init__(self, ip_address, hostname, model, routing_protocol):
        super().__init__(ip_address, hostname, "路由器")  # 呼叫父類別的建構子
        self.model = model
        self.routing_protocol = routing_protocol

    def configure_routing(self):
        print(f"設定路由協定: {self.routing_protocol}")


class Switch(NetworkDevice):  # 交換器類別，繼承自 NetworkDevice
    def __init__(self, ip_address, hostname, model, vlan):
        super().__init__(ip_address, hostname, "交換器")  # 呼叫父類別的建構子
        self.model = model
        self.vlan = vlan

    def configure_vlan(self):
        print(f"設定 VLAN: {self.vlan}")



def send_sms_alert(message):  # 使用 Twilio API 傳送 SMS 告警
    try:
        client = Client(account_sid, auth_token)
        message = client.messages.create(
            body=message,
            from_=twilio_phone_number,
            to=recipient_phone_number
        )
        print(f"已傳送 SMS 告警：{message.sid}")
    except Exception as e:
        print(f"傳送 SMS 告警失敗：{e}")



# 建立路由器物件
r1 = Router("192.168.1.1", "R1", "Cisco 8000v", "OSPF")
r1.connect()


# 監控 CPU 使用率並傳送告警
while True:
    cpu_utilization = r1.get_cpu_utilization()
    if isinstance(cpu_utilization, int):
        print(f"CPU 使用率：{cpu_utilization}%")
        if cpu_utilization > 90:
            alert_message = f"{r1.hostname} ({r1.ip_address}) CPU 使用率過高：{cpu_utilization}%"
            send_sms_alert(alert_message)
    else:
        print(cpu_utilization)  # 輸出錯誤訊息
        alert_message = f"{r1.hostname} ({r1.ip_address}) CPU 使用率取得失敗：{cpu_utilization}"
        send_sms_alert(alert_message)

    time.sleep(60)  # 每 60 秒檢查一次

這個程式碼版本進行了以下改進：

物件導向程式設計： 使用類別 NetworkDevice、Router 和 Switch 來組織程式碼，並利用繼承和多型等 OOP 概念。
錯誤處理： get_cpu_utilization 方法加入了 try...except 區塊來處理潛在的錯誤，例如 SNMP 超時、命令執行錯誤、資料解析錯誤等。並提供更具體的錯誤訊息。
Timeout 設定: 在 subprocess.check_output 中加入 timeout 引數，避免程式無限期等待 SNMP 回應。
Twilio 整合： 加入 send_sms_alert 函式，使用 Twilio API 傳送 SMS 告警。請記得替換程式碼中的佔位符為您的實際 Twilio 帳戶資訊和電話號碼。
迴圈監控： 使用 while True 迴圈持續監控 CPU 使用率，並設定 time.sleep(60) 每 60 秒檢查一次。

這個版本更具結構性、可讀性和可維護性，並且能夠更有效地處理錯誤和傳送告警。

  sequenceDiagram
    participant User
    participant Python Script
    participant Router

    User->>Python Script: 執行指令碼
    activate Python Script
    Python Script->>Router: SNMP 查詢 CPU 使用率
    activate Router
    Router-->>Python Script: CPU 使用率資料
    deactivate Router
    Python Script->>Python Script: 判斷 CPU 使用率
    Python Script-->>User: 顯示 CPU 使用率
    alt CPU 使用率 > 90%
        Python Script->>Twilio: 傳送 SMS 告警
        activate Twilio
        Twilio-->>Python Script: 告警傳送成功
        deactivate Twilio
        Python Script-->>User: 顯示告警訊息
    end
    deactivate Python Script

圖表説明： 這個序列圖描述了 Python 指令碼如何與路由器互動並透過 Twilio 傳送 SMS 告警的流程。

注意： 請確保已安裝必要的 Python 模組：twilio、pysnmp。可以使用 pip 安裝：pip install twilio pysnmp。

這個修改後的版本更符合玄貓的風格，並包含更詳細的程式碼説明和圖表。

from getpass import getpass

def get_credentials():
    username1 = input("請輸入網路管理員 1 的 ID：")
    password1 = None
    while not password1:
        password1 = getpass("請輸入網路管理員 1 的密碼：")
        password1_verify = getpass("請確認網路管理員 1 的密碼：")
        if password1 != password1_verify:
            print("密碼不符，請重新輸入。")
            password1 = None

    print("使用者名稱 1：", username1)  # 移除密碼顯示

    yes_or_no = input("網路管理員 2 的憑證是否與網路管理員 1 相同？（Yes/No）：").lower()
    expected_response = ['yes', 'y', 'no', 'n']
    while yes_or_no not in expected_response:
        yes_or_no = input("請輸入 yes 或 no：").lower()

    if yes_or_no in ('yes', 'y'):
        username2 = username1
        password2 = password1
        print("使用者名稱 2：", username2) # 移除密碼顯示
    else:
        username2 = input("請輸入網路管理員 2 的 ID：")
        password2 = None
        while not password2:
            password2 = getpass("請輸入網路管理員 2 的密碼：")
            password2_verify = getpass("請確認網路管理員 2 的密碼：")
            if password2 != password2_verify:
                print("密碼不符，請重新輸入。")
                password2 = None
        print("使用者名稱 2：", username2) # 移除密碼顯示

get_credentials()

此程式碼強化了使用者憑證收集的安全性，加入了密碼驗證機制。使用者需要輸入兩次密碼，確保輸入一致。同時，為了安全起見，程式碼不再顯示使用者輸入的密碼。

  graph LR
    B[B]
    C[C]
    D[D]
    G[G]
    H[H]
    No[No]
    Yes[Yes]
    A[輸入使用者名稱 1] --> B{輸入密碼 1};
    B --> C{確認密碼 1};
    C -- 密碼相符 --> D{詢問管理員 2 憑證是否相同};
    C -- 密碼不符 --> B;
    D -- Yes --> E[使用相同憑證];
    D -- No --> F[輸入使用者名稱 2];
    F --> G{輸入密碼 2};
    G --> H{確認密碼 2};
    H -- 密碼相符 --> I[完成];
    H -- 密碼不符 --> G;

圖表説明： 此流程圖描述了使用者憑證收集的流程，包含密碼驗證和憑證複用選項。

透過逐步完善 yes/no 輸入和使用者憑證收集的程式碼，我們展示瞭如何構建更強健、更安全的 Python 應用程式。妥善處理使用者輸入和驗證，是開發任何互動式應用程式的重要環節。

手動升級 IOS XE：逐步解析

在開始自動化之前，先了解手動升級的步驟非常重要。這有助於理解整個流程，並為後續的自動化指令碼編寫打下基礎。

準備工作： 首先，您需要準備好 Cisco Catalyst 8000v 的 .ova 檔案和升級用的 .bin 映像檔。這些檔案可以透過 Cisco 官方管道取得。
佈署虛擬機器： 使用 .ova 檔案在 VMware Workstation 上佈署虛擬機器。設定網路介面，確保虛擬路由器可以連線到網路。
連線到路由器： 使用終端軟體，例如 PuTTY 或 SecureCRT，連線到虛擬路由器。
上傳映像檔： 使用 copy 指令將 .bin 映像檔上傳到路由器的快閃記憶體中。例如：copy tftp://<TFTP 伺服器 IP 位址>/<映像檔名稱> flash:
驗證映像檔： 使用 verify /md5 flash:<映像檔名稱> 指令驗證上傳的映像檔完整性。
設定啟動組態： 使用 configure terminal 進入全域組態模式，並使用 boot system flash:<映像檔名稱> 指令設定路由器在下次啟動時使用新的映像檔。
儲存組態： 使用 copy running-config startup-config 指令儲存組態。
重新啟動路由器： 使用 reload 指令重新啟動路由器，使新的 IOS XE 映像檔生效。

  graph LR
    A[準備 .ova 和 .bin 檔案] --> B(佈署虛擬機器)
    B --> C(連線到路由器)
    C --> D(上傳映像檔)
    D --> E(驗證映像檔)
    E --> F(設定啟動組態)
    F --> G(儲存組態)
    G --> H(重新啟動路由器)

以上流程圖展示了手動升級 IOS XE 的步驟。每個步驟都至關重要，確保升級過程順利完成。

Python 自動化升級：提升效率

手動升級過程繁瑣與容易出錯。使用 Python 指令碼可以自動化這些步驟，提高效率並減少錯誤。以下是一個 Python 指令碼範例，使用 netmiko 函式庫與路由器互動：

from netmiko import ConnectHandler

def upgrade_ios_xe(device, image_file, tftp_server):
    try:
        with ConnectHandler(**device) as ssh:
            ssh.enable()
            ssh.send_command_timing(f"copy tftp://{tftp_server}/{image_file} flash:") # 複製映像檔
            ssh.send_command_timing("verify /md5 flash:" + image_file) # 驗證映像檔
            ssh.send_config_set(["boot system flash:" + image_file]) # 設定啟動組態
            ssh.send_command_timing("copy running-config startup-config") # 儲存組態
            ssh.send_command_timing("reload") # 重新啟動路由器
            print("IOS XE 升級完成！")
    except Exception as e:
        print(f"升級過程中發生錯誤：{e}")

# 裝置資訊
device = {
    "device_type": "cisco_ios",
    "host": "<路由器 IP 位址>",
    "username": "<使用者名稱>",
    "password": "<密碼>",
}

# 映像檔和 TFTP 伺服器資訊
image_file = "<映像檔名稱>"
tftp_server = "<TFTP 伺服器 IP 位址>"

# 執行升級
upgrade_ios_xe(device, image_file, tftp_server)

這個 Python 指令碼使用 netmiko 函式庫建立與路由器的 SSH 連線，並自動執行上傳映像檔、驗證、設定啟動組態、儲存組態和重新啟動等步驟。

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。