Superviser les serveurs Dell PowerEdge avec Python, iDRAC et SNMPv3

Windows

Pour installer Python sous Windows, rendez-vous sur la page officielle de téléchargement de Python et téléchargez la dernière version de Python Install Manager. Lancez son programme d’installation, puis utilisez ce gestionnaire pour installer Python.

• Une fois Python installé, ouvrez un terminal Windows et installez les paquets pysnmp et cryptography nécessaires :

C:\Users\user> py -m pip install pysnmp cryptography

Installer Python sous Debian

La procédure suivante repose sur Debian 13, mais elle devrait également fonctionner sur la plupart des distributions Linux, moyennant quelques adaptations.

• Installez le paquet python3-venv pour permettre la création d’environnements virtuels Python :

root@host:~# apt update && apt install python3-venv

• Avec un compte utilisateur standard, créez un environnement virtuel Python dédié au script SnmpTDell :

john@host:~$ python3 -m venv .snmptdell

• Activez l’environnement virtuel Python :

john@host:~$ source .snmptdell/bin/activate

• Installez les paquets pysnmp et cryptography nécessaires dans l’environnement virtuel :

(.snmptdell) john@host:~$ python3 -m pip install pysnmp cryptography

Les OID SNMP de Dell iDRAC

Comme expliqué dans l’introduction, le script Python SnmpTDell utilise SNMPv3 pour récupérer des informations matérielles auprès des interfaces Dell iDRAC. Chaque information est associée à un identifiant d’objet (OID, Object Identifier), qui permet au script d’interroger un composant précis et d’en récupérer l’état.

Le tableau suivant répertorie les OID SNMP référencés dans le script : informations système, BIOS, batteries, disques physiques et alimentations.

Pour en savoir plus sur ces OID et leurs valeurs possibles, consultez le guide de référence SNMP officiel de Dell iDRAC.

SnmpTDell.py : script Python de supervision SNMPv3

Copiez le script Python de supervision ci-dessous dans un fichier nommé SnmpTDell.py, sur la machine Windows ou Linux qui communiquera avec les interfaces iDRAC des serveurs Dell PowerEdge à superviser.

#! /usr/bin/python
import asyncio
from pysnmp.hlapi.v3arch.asyncio import *
import re
# Modules used to test host reachability with ping
import subprocess
import platform

# Script version
v_script = "20261110"
# Device information
auth_proto = usmHMACSHAAuthProtocol  # or usmHMACMD5AuthProtocol
priv_proto = usmAesCfb128Protocol      # or usmAesCfb128Protocol, usmAesCfb192Protocol, etc.

# Supported SNMP authentication and privacy protocols
SNMP_PROTOCOLS = [
    ("SHA512", "AES256", usmHMAC384SHA512AuthProtocol, usmAesCfb256Protocol),
    ("SHA1",   "AES128", usmHMACSHAAuthProtocol, usmAesCfb128Protocol),
    ("MD5",    "DES",    usmHMACMD5AuthProtocol,       usmDESPrivProtocol),
    ("SHA256", "AES256", usmHMAC192SHA256AuthProtocol, usmAesCfb256Protocol),
    ("SHA256", "AES128", usmHMAC192SHA256AuthProtocol, usmAesCfb128Protocol),
]

# Device information
liste_oid = { "servicetag": '1.3.6.1.4.1.674.10892.5.1.3.2.0', "nom_serveur": '1.3.6.1.4.1.674.10892.5.1.3.1.0', "servicecode": '1.3.6.1.4.1.674.10892.5.1.3.3.0', "modele": '1.3.6.1.4.1.674.10892.5.1.3.12.0', "statut_global": '1.3.6.1.4.1.674.10892.5.2.1.0', "status_battery": '1.3.6.1.4.1.674.10892.5.5.1.20.130.15.1.4.1', "bios_version": '1.3.6.1.4.1.674.10892.5.4.300.50.1.8.1.1', "bios_releasedate": '1.3.6.1.4.1.674.10892.5.4.300.50.1.7.1.1', "cmos_battery": '1.3.6.1.4.1.674.10892.5.4.200.10.1.52.1' }
liste_oid_disques = { "nom_disque": '1.3.6.1.4.1.674.10892.5.5.1.20.130.4.1.6', "emplacement_disque": '1.3.6.1.4.1.674.10892.5.5.1.20.130.4.1.1', "status_disques": '1.3.6.1.4.1.674.10892.5.5.1.20.130.4.1.24', "status_smart": '1.3.6.1.4.1.674.10892.5.5.1.20.130.4.1.31', "status_vd": '1.3.6.1.4.1.674.10892.5.5.1.20.140.1.1.4', "sn_disques": '1.3.6.1.4.1.674.10892.5.5.1.20.130.4.1.7', "modele_disques": '1.3.6.1.4.1.674.10892.5.5.1.20.130.4.1.8' }
liste_oid_alim = { "ps_status": '1.3.6.1.4.1.674.10892.5.4.600.12.1.5' }
resultats = []

class Disques(object):
        "Disk information container"

class Alimentations(object):
        "Power supply information container"

# Server class definition
class Serveur(object):
        def __init__(self, disques, alimentations, servicetag):
                self.disques = disques
                self.alimentations = alimentations
                self.servicetag = servicetag

        def afficher_info(self):
                print(f"Brand: {self.marque}, Model: {self.modele}")


# Detect the supported SNMP authentication and privacy protocol combination
async def detect_snmp_protocols(host, user, authkey, privkey, test_oid='1.3.6.1.4.1.674.10892.5.1.3.2.0', port=161):
    for auth_name, priv_name, auth_proto, priv_proto in SNMP_PROTOCOLS:
        #print(f"Test {auth_name} + {auth_proto} + {priv_name} + {priv_proto} on {host}...")

        try:
            errorIndication, errorStatus, errorIndex, varBinds = await get_cmd(
                SnmpEngine(),
                UsmUserData(
                    user,
                    authKey=authkey,
                    privKey=privkey,
                    authProtocol=auth_proto,
                    privProtocol=priv_proto,
                ),
                await UdpTransportTarget.create(
                    (host, port),
                    timeout=2,
                    retries=0
                ),
                ContextData(),
                ObjectType(ObjectIdentity(test_oid))
            )

            if errorIndication:
                #print(f"  Failed: {errorIndication}")
                continue

            if errorStatus:
                print(f"  SNMP error: {errorStatus.prettyPrint()}")
                continue

            #print(f"  OK : {auth_name} + {auth_proto} + {priv_name} + {priv_proto}")

            return auth_proto, priv_proto

        except Exception as e:
            print(f"  Exception : {e}")

    return None, None

# Execute SNMP GET and BULK WALK requests
async def snmp(host, port, user, authkey, privkey, auth_proto, priv_proto, start_oid, type="walk"):
    resultats = [] # Reset the results list
    if type=="get":
        #async def run_snmp_get():
            iterator = await get_cmd(
            SnmpEngine(),
            UsmUserData(
                user,
                authKey=authkey,
                privKey=privkey,
                authProtocol=auth_proto,
                privProtocol=priv_proto,
            ),
            await UdpTransportTarget.create((host, port)),
            ContextData(),
            ObjectType(ObjectIdentity(start_oid)),
                        lookupMib=False, # Return numeric OIDs (e.g. 1.3.6.1.4.1.674) instead of MIB names
            )

            errorIndication, errorStatus, errorIndex, varBinds = iterator

            if errorIndication:
                print(errorIndication)

            elif errorStatus:
                print(
                "{} at {}".format(
                errorStatus.prettyPrint(),
                errorIndex and varBinds[int(errorIndex) - 1][0] or "?",
                )
            )

            else:
                for varBind in varBinds:
                    #print(" = ".join([x.prettyPrint() for x in varBind]))
                    return (" = ".join([x.prettyPrint() for x in varBind]))

    if type=="walk":
            snmpEngine = SnmpEngine()
            errorIndication, errorStatus, errorIndex, varBindTable = await bulk_cmd(
                    snmpEngine,
                    UsmUserData(user,
                    authKey=authkey,
                    privKey=privkey,
                    authProtocol=auth_proto,
                    privProtocol=priv_proto,
                    ),
                    await UdpTransportTarget.create((host, port)),
                    ContextData(),
                    0,
                    150,
                    #*varBinds,
                    ObjectType(ObjectIdentity(start_oid)),
                    lookupMib=False, # Return numeric OIDs (e.g. 1.3.6.1.4.1.674) instead of MIB names
            )

            if errorIndication:
                    print(errorIndication)
            elif errorStatus:
                    print(
                        f"{errorStatus.prettyPrint()} at {varBinds[int(errorIndex) - 1][0] if errorIndex else '?'}"
                    )
            else:
                    for varBind in varBindTable:
                        # Workaround: filter BULK results by the requested OID prefix to avoid unrelated entries
                        TEST = str(start_oid + '.').split('10892')[1] # Extract the part after the enterprise branch 10892
                        if TEST in str(" = ".join([x.prettyPrint() for x in varBind])): # Keep only entries matching the requested OID branch
                           #print(" = ".join([x.prettyPrint() for x in varBind]))
                           #return varBind
                           #return (" = ".join([x.prettyPrint() for x in varBind]))
                           resultats.append(" = ".join([x.prettyPrint() for x in varBind]))
            return resultats

            varBinds = varBindTable

def check_host(host, user, authkey, privkey, port=161, auth_proto=usmHMACSHAAuthProtocol, priv_proto=usmAesCfb128Protocol):
    "Check iDRAC reachability with ping, then SNMP."
    # Ping test   
    # Prepare the ping command
    system = platform.system().lower() # Detect the operating system
    ip = host
    if system == "windows":
        cmd = ["ping", "-n", "1", "-w", "100", ip]
    else:
        cmd = ["ping", "-c", "1", "-W", "1", ip]
    
    try:
        result = subprocess.run(
            cmd,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            timeout=2
        )
    except subprocess.TimeoutExpired:
        print("IP address is not responding; iDRAC is unreachable")
        return "pingko" # Return immediately
    else:
        # Check for a failed response
        if result.returncode != 0:
            print("IP address is not responding; iDRAC is unreachable")
            return "pingko" # Return immediately

    # Check the SNMP connection
    try:
        test_conn=asyncio.run(snmp(host, 161, user, authkey, privkey, auth_proto, priv_proto, liste_oid["servicetag"], type="get"))
    except:
        print("SNMP is not responding; iDRAC is unreachable")
        return "snmpko" # Return immediately
    else:
        # Check for an empty SNMP response
        if test_conn is None:
            print("SNMP is not responding; iDRAC is unreachable")
            return "snmpko" # Return immediately


def get_info(host, user, authkey, privkey, port=161, auth_proto=usmHMACSHAAuthProtocol, priv_proto=usmAesCfb128Protocol):
    "Retrieve server hardware information through SNMP."
    dst, subj = "", "SERVER STATUS OK"
    index_alims = []
    index_disques = []
    #########
    # SERVER #
    #########
    for nom, oid in liste_oid.items():
        res = asyncio.run(snmp(host, port, user, authkey, privkey, auth_proto, priv_proto, oid, type="get"))
        #snmp_walk(host, port, user, authkey, privkey, auth_proto, priv_proto, oid)
        for element in res:
            valeur = str(res).replace(' = ','=').split('=')[1]
            #oid = str(res).replace(' = ','=').split('=')[0]
            if nom == "servicetag":
                ObjDisques = Disques() # Create the disk object
                ObjAlims = Alimentations() # Create the power supply object
                ObjServeur = Serveur(ObjDisques, ObjAlims, servicetag=valeur) # Create the server object
            else:
                setattr(ObjServeur, nom, valeur)


    #########
    # DISKS  #
    #########
    res = asyncio.run(snmp(host, port, user, authkey, privkey, auth_proto, priv_proto, liste_oid_disques["emplacement_disque"]))
    for disque_id in res:
        emplacement_disque = str(disque_id).replace(' = ','=').split('=')[1] # Retrieve the disk identifier for each entry
        index_disques.append(emplacement_disque)
        for i in "nom_disque", "status_disques", "status_smart", "modele_disques", "sn_disques":
            res = asyncio.run(snmp(host, port, user, authkey, privkey, auth_proto, priv_proto, liste_oid_disques[i] + "." + emplacement_disque, type="get"))
            setattr(ObjServeur.disques, i + "_" + emplacement_disque, str(res).replace(' = ','=').split('=')[1])

     
    messages = {
        6: "WARNING",
        5: "ERROR",
        4: "WARNING",
        3: "OK",
        2: "WARNING",
        1: "UNKNOWN"
    }
    smart = {
        0: "OK",
        1: "ERROR"
    }
    batterie = {
        1: "UNKNOWN",
        2: "OK",
        3: "ERROR"
    }
    cmos = {
        1: "OTHER",
        2: "UNKNOWN",
        3: "OK",
        5: "ERROR"
    }
    print("SnmpTDell version:", v_script)
    print("Server", ObjServeur.nom_serveur + " " + ObjServeur.modele, "health status:", messages.get(int(ObjServeur.statut_global), "UNKNOWN"))
    subj = "Server " + ObjServeur.nom_serveur + " " + ObjServeur.modele + " health status: " + messages.get(int(ObjServeur.statut_global), "UNKNOWN")
    print("* Service Tag : " + ObjServeur.servicetag.replace(" ",""))
    print("* Service Code : " + ObjServeur.servicecode.replace(" ",""))
    print("* BIOS version : " + ObjServeur.bios_version.replace(" ",""))
    print("* BIOS release date: " + ObjServeur.bios_releasedate.replace(" ",""))
    if ObjServeur.cmos_battery == "No Such Instance currently exists at this OID":
        print("* CMOS battery status: Not available")
    else:
        print("* CMOS battery status: " + cmos.get(int(ObjServeur.cmos_battery.replace(" ","")), "UNKNOWN"))
    if ObjServeur.status_battery == "No Such Instance currently exists at this OID":
        print("* Storage controller battery status: No battery detected on this server")
    else:
        print("* Storage controller battery status: " + batterie.get(int(ObjServeur.status_battery.replace(" ","")), "UNKNOWN"))


    #######
    # PSUs  #
    #######
    res = asyncio.run(snmp(host, port, user, authkey, privkey, auth_proto, priv_proto, liste_oid_alim["ps_status"], type="walk"))
    for alim_id in res:
        emplacement_alim = str(alim_id).replace(' = ','=').split('=')[1] # Retrieve the power supply status
        index_alims.append(str(alim_id).replace(' = ','=').split('=')[0][-1]) # Store the power supply identifiers
        setattr(ObjServeur.alimentations, "ps_status_" + str(alim_id).replace(' = ','=').split('=')[0][-1], emplacement_alim)

    print("* POWER SUPPLIES:")
    for alim_id in index_alims:
        print(" - Power supply " + alim_id + ": " + messages.get(int(getattr(ObjServeur.alimentations, f"ps_status_{alim_id}")), "UNKNOWN"))
        #getattr(ObjServeur.alimentations, f"ps_status_{alim_id}"))

    #########
    # DISKS  #
    #########
    print("* DISKS:")
    for disque_id in index_disques:
        print(" - Disk " + disque_id + ":", 
        getattr(ObjServeur.disques, f"nom_disque_{disque_id}") + ",",
        "SN :",
        getattr(ObjServeur.disques, f"sn_disques_{disque_id}") + ",",
        "Model:",
        getattr(ObjServeur.disques, f"modele_disques_{disque_id}") + ",",
        "SMART status:",
        #getattr(ObjServeur.disques, f"status_smart_{disque_id}") + ",",
        smart.get(int(getattr(ObjServeur.disques, f"status_smart_{disque_id}")), "UNKNOWN") + ",", \
        "Health status:", \
        messages.get(int(getattr(ObjServeur.disques, f"status_disques_{disque_id}")), "UNKNOWN"))
        #getattr(ObjServeur.disques, f"status_disques_{disque_id}"))
        getattr(ObjServeur.disques, f"nom_disque_{disque_id}") + ", " + \
        "SN: " + getattr(ObjServeur.disques, f"sn_disques_{disque_id}") + ", " + \
        "Model: " + getattr(ObjServeur.disques, f"modele_disques_{disque_id}") + ", " + \
        "SMART status: " + smart.get(int(getattr(ObjServeur.disques, f"status_smart_{disque_id}")), "UNKNOWN") + ", " + \
        "Health status: " + messages.get(int(getattr(ObjServeur.disques, f"status_disques_{disque_id}")), "UNKNOWN") + "\n"
    return subj

### READ CONFIGURATION FILE
with open('./conf.ini', 'r') as fichier:
    host = user = authkey = privkey = ""
    flag_mail = False
    for ligne in fichier:
        #print(ligne)
        if (re.search(r"\[.*\]", ligne) and flag_mail == False): # A new iDRAC IP address was found
            host = str(re.search(r"\[(.*?)\]", ligne).group(1))
            print("---------------------- " + host + " ----------------------")
            flag_mail = True
            continue # Continue to the next line
        if (re.search(r"\[.*\]", ligne) and flag_mail == True): # A new iDRAC IP address was found (multiple servers in the configuration)
            #resultat = check_host(host, user, authkey, privkey) # Check ping and SNMP connectivity
            auth_proto, priv_proto = asyncio.run(detect_snmp_protocols(host=host,user=user,authkey=authkey,privkey=privkey)) # Detect authentication and privacy protocols
            resultat = check_host(host, user, authkey, privkey, auth_proto=auth_proto, priv_proto=priv_proto)
            if resultat not in ("snmpko","pingko"):
                #corps, subj = get_info(host=host,user=user,authkey=authkey,privkey=privkey)
                subj = get_info(host=host,user=user,authkey=authkey,privkey=privkey, auth_proto=auth_proto, priv_proto=priv_proto)
                #print("BODY: " + corps + " SUBJECT: " + subj)
            if resultat == "snmpko":
                print("iDRAC unreachable")
            if resultat == "pingko":
                print("iDRAC unreachable")
            host = str(re.search(r"\[(.*?)\]", ligne).group(1))
            print("---------------------- " + host + " ----------------------")
        if re.search(r".* = .*", ligne):
            # Read the client name
            if str(ligne.replace(' = ','=').split('=')[0]) == 'client':
                client = str(re.findall(r"'(.*?)'", str(ligne.replace(' = ','=').split('=')[1]))[0])
            # Read the authentication key
            if str(ligne.replace(' = ','=').split('=')[0]) == 'authkey':
                authkey = str(re.findall(r"'(.*?)'", str(ligne.replace(' = ','=').split('=')[1]))[0])
            # Read the privacy (encryption) key
            if str(ligne.replace(' = ','=').split('=')[0]) == 'privkey':
                privkey = str(re.findall(r"'(.*?)'", str(ligne.replace(' = ','=').split('=')[1]))[0])
            # Read the SNMP username
            if str(ligne.replace(' = ','=').split('=')[0]) == 'user':
                user = str(re.findall(r"'(.*?)'", str(ligne.replace(' = ','=').split('=')[1]))[0])
            # Read the email settings
            if str(ligne.replace(' = ','=').split('=')[0]) == 'mail':
                mail = str(re.findall(r"'(.*?)'", str(ligne.replace(' = ','=').split('=')[1]))[0])
            # Read the email settings
            if str(ligne.replace(' = ','=').split('=')[0]) == 'smail':
                sender = str(re.findall(r"'(.*?)'", str(ligne.replace(' = ','=').split('=')[1]))[0])

    auth_proto, priv_proto = asyncio.run(detect_snmp_protocols(host=host,user=user,authkey=authkey,privkey=privkey)) # Detect authentication and privacy protocols
    resultat = check_host(host, user, authkey, privkey, auth_proto=auth_proto, priv_proto=priv_proto) # Check ping and SNMP connectivity
    if resultat not in ("snmpko","pingko"):
        subj = get_info(host=host,user=user,authkey=authkey,privkey=privkey, auth_proto=auth_proto, priv_proto=priv_proto)
    if resultat == "snmpko":
        print("iDRAC unreachable")
    if resultat == "pingko":
        print("iDRAC unreachable")

• Le script lit un fichier de configuration conf.ini contenant les adresses IP des interfaces iDRAC et les identifiants SNMPv3 des serveurs Dell PowerEdge à superviser. Chaque serveur possède sa propre section de configuration, ce qui permet de superviser plusieurs serveurs avec un seul script :

[192.168.1.100]
authkey = 'd0ntus31t' # Authentication passphrase
privkey = 'd0ntus31t' # Encryption passphrase
user = 'snmptdell' # SNMPv3 username

[192.168.1.101]
authkey = 'SnmpTDellR0CKS' # Authentication passphrase
privkey = 'SnmpTDellR0CKS' # Encryption passphrase
user = 'snmptdell' # SNMPv3 username

[192.168.1.102]
authkey = 'd0nuts31t' # Authentication passphrase
privkey = 'd0nuts31t' # Encryption passphrase
user = 'snmptdell' # SNMPv3 username

Windows

• Ouvrez un terminal Windows, placez-vous dans le répertoire contenant le script SnmpTDell.py et le fichier de configuration conf.ini, puis exécutez la commande suivante :

C:\Users\user> py SnmpTDell.py

Linux

• Activez l’environnement virtuel Python créé précédemment :

john@host:~$ source .snmptdell/bin/activate

• Placez-vous dans le répertoire contenant le script SnmpTDell.py et le fichier de configuration conf.ini, puis lancez le script :

(.snmptdell) john@host:~$ python3 SnmpTDell.py

Résultats de la supervision

Que vous exécutiez SnmpTDell sous Windows ou sous Linux, vous devriez obtenir un résultat similaire à celui ci-dessous. Il présente l’état matériel des serveurs Dell PowerEdge, notamment les informations du BIOS, des batteries, des alimentations et des disques physiques :

---------------------- 192.168.1.100 ----------------------
SnmpTDell version: 20261110
Server SERVER01 PowerEdge R740 health status: OK
* Service Tag : DEADBEE5
* Service Code : 45142555000
* BIOS version : 2.22.1
* BIOS release date : ...
* CMOS battery status: OK
* Storage controller battery status: OK
* POWER SUPPLIES:
 - Power supply 1: OK
 - Power supply 2: OK
* DISKS:
 - Disk 1: ..., SN: ..., Model: ..., SMART status: OK, Health status: OK

---------------------- 192.168.1.101 ----------------------
IP address is not responding; iDRAC is unreachable
iDRAC unreachable

---------------------- 192.168.1.102 ----------------------
[…]