Antoine Van Elstraete
b0d571ae26
- Docstrings for all modules, classes and methods - README.md with installation and usage instructions - Update CLAUDE.md with dns.py Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
"""
|
|
Checker SNMP.
|
|
|
|
Vérifie la disponibilité d'un équipement réseau via SNMP et peut
|
|
récupérer sa température.
|
|
"""
|
|
|
|
import asyncio
|
|
import time
|
|
|
|
from pysnmp.hlapi.v1arch.asyncio import (
|
|
Slim,
|
|
ObjectIdentity,
|
|
ObjectType,
|
|
)
|
|
|
|
from .base import BaseChecker, CheckResult
|
|
|
|
|
|
class SnmpChecker(BaseChecker):
|
|
"""
|
|
Vérifie la disponibilité d'un équipement via SNMP.
|
|
|
|
Utilise SNMPv2c pour interroger un OID et optionnellement
|
|
récupérer la température de l'équipement.
|
|
|
|
Configuration YAML:
|
|
host: Adresse IP de l'équipement (obligatoire).
|
|
port: Port SNMP (défaut: 161).
|
|
community: Communauté SNMP (défaut: public).
|
|
oid: OID à interroger (défaut: sysDescr).
|
|
temperature_oid: OID de la température (optionnel).
|
|
Note: Utiliser l'OID complet (feuille, pas branche).
|
|
Exemples:
|
|
- Mikrotik: 1.3.6.1.4.1.14988.1.1.3.100.1.3.52.0
|
|
- Synology: 1.3.6.1.4.1.6574.1.2.0
|
|
timeout: Délai d'attente en secondes (défaut: 5).
|
|
"""
|
|
|
|
def check(self) -> CheckResult:
|
|
"""
|
|
Exécute la vérification SNMP.
|
|
|
|
Wrapper synchrone autour de _async_check() pour compatibilité
|
|
avec l'interface BaseChecker.
|
|
|
|
Returns:
|
|
CheckResult avec success=True si l'équipement répond.
|
|
"""
|
|
return asyncio.run(self._async_check())
|
|
|
|
async def _async_check(self) -> CheckResult:
|
|
"""
|
|
Exécute la requête SNMP de manière asynchrone.
|
|
|
|
Interroge l'OID principal et optionnellement l'OID de température
|
|
en une seule requête SNMP GET.
|
|
|
|
Returns:
|
|
CheckResult contenant le résultat et la température si configurée.
|
|
"""
|
|
host = self.config["host"]
|
|
port = self.config.get("port", 161)
|
|
community = self.config.get("community", "public")
|
|
oid = self.config.get("oid", "1.3.6.1.2.1.1.1.0") # sysDescr
|
|
temperature_oid = self.config.get("temperature_oid")
|
|
timeout_val = self.config.get("timeout", 5)
|
|
|
|
start = time.time()
|
|
try:
|
|
with Slim() as slim:
|
|
# Construit la liste des OIDs à interroger
|
|
oids = [ObjectType(ObjectIdentity(oid))]
|
|
if temperature_oid:
|
|
oids.append(ObjectType(ObjectIdentity(temperature_oid)))
|
|
|
|
error_indication, error_status, error_index, var_binds = await slim.get(
|
|
community,
|
|
host,
|
|
port,
|
|
*oids,
|
|
timeout=timeout_val,
|
|
retries=1
|
|
)
|
|
|
|
response_time = (time.time() - start) * 1000 # ms
|
|
|
|
if error_indication:
|
|
return CheckResult(
|
|
success=False,
|
|
message=f"SNMP error: {error_indication}",
|
|
response_time=None
|
|
)
|
|
elif error_status:
|
|
return CheckResult(
|
|
success=False,
|
|
message=f"SNMP error: {error_status.prettyPrint()}",
|
|
response_time=None
|
|
)
|
|
else:
|
|
# Inclut uniquement l'OID principal dans les détails
|
|
details = {str(var_binds[0][0]): str(var_binds[0][1])}
|
|
|
|
# Extrait la température si configurée (deuxième OID)
|
|
if temperature_oid and len(var_binds) >= 2:
|
|
try:
|
|
details["temperature"] = int(var_binds[1][1])
|
|
except (ValueError, TypeError):
|
|
pass # Ignore si la valeur n'est pas un entier
|
|
|
|
return CheckResult(
|
|
success=True,
|
|
message="SNMP response OK",
|
|
response_time=response_time,
|
|
details=details
|
|
)
|
|
except Exception as e:
|
|
return CheckResult(
|
|
success=False,
|
|
message=f"SNMP error: {e}",
|
|
response_time=None
|
|
)
|