File "check_fortinet.py"
Full path: /opt/updates.edris.fr/NDS/Plugins/check_fortinet.py
File size: 38.94 KB
MIME-type: text/x-python
Charset: utf-8
Download Open Edit Advanced Editor Back
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
check_fortinet.py
Plugin de supervision Centreon / Nagios des FortiGate et FortiSwitch (SNMPv3).
Remplace check_fortinet.sh (mêmes périmètres, mêmes sorties, mêmes noms de
perfdata) avec une empreinte minimale :
- une seule session SNMPv3 par contrôle (pysnmp) : découverte du moteur et
dérivation des clés une seule fois, aucune MIB chargée, aucun fichier
d'état net-snmp réécrit ;
- les tables sont lues colonne par colonne en GETBULK (une poignée de
requêtes) au lieu d'un snmpwalk par valeur ;
- aucun fichier temporaire, aucun ping, aucune requête MySQL, aucun
processus externe (curl, jq, ansible-vault).
Install :
read -rsp 'Mot de passe proxy : ' PROXY_PASS; echo
PROXY_ENC=$(python3 -c 'import urllib.parse; print(urllib.parse.quote(input(), safe=""))' <<< "$PROXY_PASS")
export HTTPS_PROXY="http://SVCPCENTREONPROXY:${PROXY_ENC}@proxy.matmut.fr:80" HTTP_PROXY="http://SVCPCENTREONPROXY:${PROXY_ENC}@proxy.matmut.fr:80"
/opt/edris/plugins-venv/bin/pip install --trusted-host files.pythonhosted.org pysnmp==7.1.30 cryptography
install -m 755 -o centreon-engine -g centreon-engine check_fortinet.py /usr/lib/centreon/plugins/
Usage :
check_fortinet.py -A <perimetre> -B <adresse>
--snmp-username <user> --authpassphrase <pass> --privpassphrase <pass>
[--authprotocol SHA] [--privprotocol AES] [--hostname <nom_centreon>]
[--model-source ipam|snmp] [--ipam-username <user> --ipam-password <pass>]
/opt/edris/plugins-venv/bin/python $USER1$/check_fortinet.py -A cpu -B $HOSTADDRESS$ \
--snmp-username '$_HOSTSNMPV3USER$' --authpassphrase '$_HOSTSNMPV3AUTHPASS$' \
--privpassphrase '$_HOSTSNMPV3PRIVPASS$' --hostname '$HOSTNAME$' \
--ipam-username '$_HOSTIPAMUSER$' --ipam-password '$_HOSTIPAMPASS$'
Modèle : lu dans l'IPAM (type de l'équipement, comme le script bash) ou, avec
--model-source snmp, dans entPhysicalDescr.1 (même session SNMP, pas d'appel
HTTPS supplémentaire).
Périmètres :
FortiSwitch : cpu, uptime, poe, temperature
FortiGate : cpu, memory, session, state-cluster, master, cluster,
sla, sla-o-bvpn-hz, sla-o-bvpn-ren, sla-o-int_obs,
sla-o-int_sfr, sla-HC_Google, sla-HC_Google_Agences, mos,
temperature, fan, psu, tunnels, hub-tunnels, uptime,
uptime-interfaces, bgp
sla-HC_Google* interrogent l'API REST du FortiGate (compte --api-username) ; le
mot de passe est lu dans le coffre ansible-vault (déchiffré en Python) ou passé
par --api-password.
Dépendances : pysnmp >= 7.1 (Python >= 3.10), cryptography.
Codes retour : 0 = OK, 1 = WARNING, 2 = CRITICAL, 3 = UNKNOWN
"""
import argparse
import asyncio
import base64
import binascii
import hashlib
import hmac
import http.cookiejar
import json
import re
import ssl
import sys
import urllib.error
import urllib.parse
import urllib.request
OK, WARNING, CRITICAL, UNKNOWN = 0, 1, 2, 3
try:
from pysnmp.hlapi.v3arch.asyncio import (
USM_AUTH_HMAC96_MD5, USM_AUTH_HMAC96_SHA, USM_AUTH_HMAC128_SHA224,
USM_AUTH_HMAC192_SHA256, USM_AUTH_HMAC256_SHA384, USM_AUTH_HMAC384_SHA512,
USM_AUTH_NONE, USM_PRIV_CBC56_DES, USM_PRIV_CBC168_3DES,
USM_PRIV_CFB128_AES, USM_PRIV_CFB192_AES, USM_PRIV_CFB192_AES_BLUMENTHAL,
USM_PRIV_CFB256_AES, USM_PRIV_CFB256_AES_BLUMENTHAL, USM_PRIV_NONE,
ContextData, ObjectIdentity, ObjectType, SnmpEngine, UdpTransportTarget,
UsmUserData, bulk_cmd, get_cmd,
)
from pysnmp.proto.rfc1905 import EndOfMibView, NoSuchInstance, NoSuchObject
except ImportError as exc:
print("UNKNOWN - module pysnmp (>= 7.1) introuvable : %s" % exc)
sys.exit(UNKNOWN)
AUTH_PROTOCOLS = {
"MD5": USM_AUTH_HMAC96_MD5,
"SHA": USM_AUTH_HMAC96_SHA,
"SHA224": USM_AUTH_HMAC128_SHA224,
"SHA256": USM_AUTH_HMAC192_SHA256,
"SHA384": USM_AUTH_HMAC256_SHA384,
"SHA512": USM_AUTH_HMAC384_SHA512,
}
# Mêmes noms que net-snmp (AES192/AES256 = Blumenthal, suffixe C = Reeder/Cisco).
PRIV_PROTOCOLS = {
"DES": USM_PRIV_CBC56_DES,
"3DES": USM_PRIV_CBC168_3DES,
"AES": USM_PRIV_CFB128_AES,
"AES128": USM_PRIV_CFB128_AES,
"AES192": USM_PRIV_CFB192_AES_BLUMENTHAL,
"AES256": USM_PRIV_CFB256_AES_BLUMENTHAL,
"AES192C": USM_PRIV_CFB192_AES,
"AES256C": USM_PRIV_CFB256_AES,
}
# Nombre max de varbinds demandés par réponse GETBULK.
MAX_VARBINDS = 50
# Durée max d'un contrôle, sous le timeout de commande Centreon (60 s par défaut).
CHECK_DEADLINE = 50
# Timeout des appels HTTPS (IPAM, API FortiOS).
HTTP_TIMEOUT = 10
IPAM_URL = "https://10.131.212.97"
VAULT_PASSWORD_FILE = "/etc/ansible/vars/.vault_pass.txt"
# --- OID ------------------------------------------------------------------
ENT_PHYSICAL_DESCR_1 = "1.3.6.1.2.1.47.1.2.1.1.2.1"
SYS_UPTIME = "1.3.6.1.2.1.1.3.0"
IF_DESCR = "1.3.6.1.2.1.2.2.1.2"
IF_OPER_STATUS = "1.3.6.1.2.1.2.2.1.8"
IF_LAST_CHANGE = "1.3.6.1.2.1.2.2.1.9"
BGP_PEER_ENTRY = "1.3.6.1.2.1.15.3.1" # bgpPeerEntry
ENT_SENSOR_VALUE_3 = "1.3.6.1.2.1.99.1.1.1.4.3"
PETH_MAIN_PSE_POWER_1 = "1.3.6.1.2.1.105.1.3.1.1.2.1"
PETH_MAIN_PSE_CONSUMPTION_1 = "1.3.6.1.2.1.105.1.3.1.1.4.1"
FSW_CPU = "1.3.6.1.4.1.12356.106.4.1.2.0"
FGT_CPU = "1.3.6.1.4.1.12356.101.4.1.3.0"
FGT_MEMORY = "1.3.6.1.4.1.12356.101.4.1.4.0"
FGT_SESSIONS = "1.3.6.1.4.1.12356.101.4.1.8.0"
FGT_UPTIME = "1.3.6.1.4.1.12356.101.4.1.22.0"
FGT_SENSOR_ENTRY = "1.3.6.1.4.1.12356.101.4.3.2.1" # fgHwSensorEntry
FGT_HEALTHCHECK_ENTRY = "1.3.6.1.4.1.12356.101.4.9.2.1" # fgVWLHealthCheckLinkEntry
FGT_VPN_UP_COUNT = "1.3.6.1.4.1.12356.101.12.1.1.0" # fgVpnTunnelUpCount
FGT_VPN_TUNNEL_ENTRY = "1.3.6.1.4.1.12356.101.12.2.2.1" # fgVpnTunEntry
FGT_HA_ENTRY = "1.3.6.1.4.1.12356.101.13.2.1.1" # fgHaStatsEntry
# Colonnes des tables
SENSOR_NAME, SENSOR_VALUE, SENSOR_ALARM = 2, 3, 4
HC_STATE, HC_LATENCY, HC_JITTER, HC_PACKET_LOSS, HC_IFNAME, HC_MOS = 4, 5, 6, 9, 14, 19
VPN_NAME, VPN_STATUS = 2, 20
HA_INDEX, HA_SERIAL, HA_HOSTNAME, HA_SYNC, HA_MASTER_SERIAL = 1, 2, 11, 12, 16
BGP_STATE, BGP_ESTABLISHED_TIME = 2, 16
# --- Seuils par modèle ------------------------------------------------------
# Numéro de modèle cherché dans le type (FortiGate-40F, fgt40F, ...).
MODELS = ("40", "100", "600", "3000")
# warning, critical, max
SESSION_LIMITS = {
"40": (560000, 630000, 700000),
"100": (1200000, 1350000, 1500000),
"600": (6400000, 7200000, 8000000),
"3000": (56000000, 63000000, 70000000),
}
TEMP_MAX = 110
def first_word(name):
"""Nommage historique des capteurs 40F : premier mot, coupé au premier '.'."""
return name.split(" ", 1)[0].split(".", 1)[0]
def underscored(name):
return name.replace(" ", "_")
# filtre sur le nom du capteur, fonction de nommage, warning, critical
TEMP_SENSORS = {
"40": (None, first_word, 80, 90),
"100": (r"temp", underscored, 90, 100),
"600": (r"temp|tb|ts", underscored, 80, 85),
"3000": (r"temp|tb|ts|th", underscored, 80, 85),
}
SLA_LINKS = r"O-INT|O-BVPN|O-BES2"
SLA_SINGLE_LINK = {
"sla-o-bvpn-hz": "BVPN-HZ",
"sla-o-bvpn-ren": "BVPN-RE",
"sla-o-int_obs": "INT_OBS",
"sla-o-int_sfr": "INT_SFR",
}
# Membre du health-check HC_Google, coffre ansible et clé du mot de passe API.
HC_GOOGLE = {
"sla-HC_Google": (
"Int_Vlan_599", "/etc/ansible/vars/Pwd_sites_centreon.yml", "pwd_sites_centreon"),
"sla-HC_Google_Agences": (
"lan1", "/etc/ansible/vars/Pwd_agences_centreon.yml", "pwd_agences_centreon"),
}
# Tunnel de secours par modèle et renommage des noms de phase 1 tronqués.
TUNNEL_BACKUP = {
"40": ("BVPN", r"O-BVPN-RE(?!N)", "O-BVPN-REN"),
"100": ("BES2", r"O-BES2-RE(?!N)", "O-BES2-REN"),
}
UPTIME_INTERFACES = r"bvpn|bes2|sfr|obs|inet"
BGP_PEERS = {
"100.80.0.1": "BVPN-HZ",
"100.81.0.1": "BVPN-REN",
"100.82.0.1": "BES2-HZ",
"100.83.0.1": "BES2-REN",
"100.84.0.1": "BES2_OC-HZ",
"100.85.0.1": "BES2_OC-REN",
"100.86.0.1": "OBS",
"100.87.0.1": "SFR",
"100.90.0.1": "INT_PRA-OBS",
}
BGP_ESTABLISHED = 6
BGP_STATES = {BGP_ESTABLISHED: "Established", 3: "Active"}
# MOS (échelle 1 à 5) : alerte en dessous des seuils.
MOS_WARNING, MOS_CRITICAL = 3, 2
class PluginExit(Exception):
def __init__(self, code, output):
Exception.__init__(self, output)
self.code = code
self.output = output
def finish(code, text, perfdata=None):
raise PluginExit(code, "%s|%s" % (text, perfdata) if perfdata is not None else text)
# --------------------------------------------------------------------------
# SNMP
# --------------------------------------------------------------------------
class SnmpError(Exception):
pass
def oid_tuple(oid):
return tuple(int(x) for x in oid.strip(".").split("."))
def is_missing(value):
return value is None or isinstance(value, (NoSuchObject, NoSuchInstance, EndOfMibView))
def as_str(value):
if is_missing(value):
return ""
if hasattr(value, "asOctets"):
return value.asOctets().decode("utf-8", "replace")
return str(int(value))
def as_number(value):
"""Valeur numérique (entier SNMP ou chaîne '12.34'), None si absente."""
if is_missing(value):
return None
if hasattr(value, "asOctets"):
try:
return float(as_str(value).strip().strip('"'))
except ValueError:
return None
return int(value)
class Snmp(object):
def __init__(self, args):
self.host = args.host
self.port = args.port
self.timeout = args.timeout
self.retries = args.retries
self.engine = SnmpEngine()
self.context = ContextData(contextName=args.snmp_context)
if args.authpassphrase and args.privpassphrase:
self.auth = UsmUserData(
args.snmp_username,
authKey=args.authpassphrase, privKey=args.privpassphrase,
authProtocol=AUTH_PROTOCOLS[args.authprotocol],
privProtocol=PRIV_PROTOCOLS[args.privprotocol])
elif args.authpassphrase:
self.auth = UsmUserData(
args.snmp_username, authKey=args.authpassphrase,
authProtocol=AUTH_PROTOCOLS[args.authprotocol], privProtocol=USM_PRIV_NONE)
else:
self.auth = UsmUserData(args.snmp_username, authProtocol=USM_AUTH_NONE,
privProtocol=USM_PRIV_NONE)
self.target = None
async def open(self):
try:
self.target = await UdpTransportTarget.create(
(self.host, self.port), timeout=self.timeout, retries=self.retries)
except Exception as exc:
raise SnmpError("adresse invalide %s : %s" % (self.host, exc))
def close(self):
self.engine.close_dispatcher()
@staticmethod
def _check(error_indication, error_status, error_index):
if error_indication:
raise SnmpError(str(error_indication))
if error_status:
raise SnmpError("%s (index %s)" % (error_status.prettyPrint(), error_index))
async def get(self, *oids):
"""Un seul GET pour plusieurs OID : liste des valeurs (None si absente)."""
result = await get_cmd(
self.engine, self.auth, self.target, self.context,
*[ObjectType(ObjectIdentity(oid)) for oid in oids], lookupMib=False)
self._check(*result[:3])
return [None if is_missing(value) else value for _, value in result[3]]
async def walk(self, *roots):
"""
Parcourt plusieurs sous-arbres en parallèle dans les mêmes requêtes
GETBULK : {racine: {suffixe_oid: valeur}} dans l'ordre des OID.
"""
roots = [oid_tuple(root) for root in roots]
result = dict((root, {}) for root in roots)
cursor = dict((root, root) for root in roots)
pending = list(roots)
while pending:
repetitions = max(1, MAX_VARBINDS // len(pending))
response = await bulk_cmd(
self.engine, self.auth, self.target, self.context, 0, repetitions,
*[ObjectType(ObjectIdentity(cursor[root])) for root in pending],
lookupMib=False)
self._check(*response[:3])
var_binds = response[3]
if not var_binds:
break
done = set()
for position, (name, value) in enumerate(var_binds):
root = pending[position % len(pending)]
if root in done:
continue
oid = tuple(name)
if (isinstance(value, EndOfMibView) or oid[:len(root)] != root
or oid <= cursor[root]):
done.add(root)
continue
result[root][oid[len(root):]] = value
cursor[root] = oid
pending = [root for root in pending if root not in done]
return result
async def table(self, entry, columns):
"""Colonnes d'une table : {index_ligne: {colonne: valeur}} dans l'ordre."""
base = oid_tuple(entry)
walked = await self.walk(*["%s.%d" % (entry, col) for col in columns])
rows = {}
for col in columns:
for index, value in walked[base + (col,)].items():
rows.setdefault(index, {})[col] = value
return dict(sorted(rows.items()))
# --------------------------------------------------------------------------
# HTTPS (IPAM, API FortiOS) et coffre ansible
# --------------------------------------------------------------------------
def https_opener(*handlers):
"""Certificats non vérifiés (curl -k) et sans proxy (no_proxy / --noproxy)."""
return urllib.request.build_opener(
urllib.request.ProxyHandler({}),
urllib.request.HTTPSHandler(context=ssl._create_unverified_context()),
*handlers)
def ipam_model(args):
"""Type de l'équipement dans l'IPAM (paramètre 'type' de la classe)."""
if not args.ipam_username or not args.ipam_password:
finish(UNKNOWN, "UNKNOWN - --ipam-username / --ipam-password obligatoires "
"(ou --model-source snmp)")
# Requête identique à celle du script bash.
url = "%s/rest/iplnetdev_list?WHERE=iplnetdev_hostaddr%%LIKE%%'%s'" % (
args.ipam_url.rstrip("/"), args.host)
request = urllib.request.Request(url, headers={
"cache-control": "no-cache",
"x-ipm-username": base64.b64encode(args.ipam_username.encode()).decode(),
"x-ipm-password": base64.b64encode(args.ipam_password.encode()).decode(),
})
try:
body = https_opener().open(request, timeout=HTTP_TIMEOUT).read()
devices = json.loads(body) if body.strip() else []
except (urllib.error.URLError, OSError, ValueError) as exc:
finish(UNKNOWN, "UNKNOWN - IPAM injoignable pour %s : %s" % (args.host, exc))
types = []
for device in devices if isinstance(devices, list) else []:
params = device.get("iplnetdev_class_parameters") or ""
match = re.search(r"(?:^|&)type=([^&]+)", params)
if match:
types.append(urllib.parse.unquote_plus(match.group(1)))
return "\n".join(types)
def vault_decrypt(path, password):
"""Déchiffre un fichier ansible-vault (format 1.1 / 1.2, AES256)."""
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
with open(path) as fh:
lines = fh.read().splitlines()
header = lines[0].strip().split(";") if lines else []
if len(header) < 3 or header[0] != "$ANSIBLE_VAULT" or header[2] != "AES256":
raise ValueError("format de coffre non supporté")
salt, expected, ciphertext = (
binascii.unhexlify(part.strip())
for part in binascii.unhexlify("".join(line.strip() for line in lines[1:])).split(b"\n", 2))
derived = hashlib.pbkdf2_hmac("sha256", password, salt, 10000, 80)
key, hmac_key, iv = derived[:32], derived[32:64], derived[64:]
if not hmac.compare_digest(hmac.new(hmac_key, ciphertext, hashlib.sha256).digest(), expected):
raise ValueError("mot de passe du coffre incorrect")
decryptor = Cipher(algorithms.AES(key), modes.CTR(iv)).decryptor()
plain = decryptor.update(ciphertext) + decryptor.finalize()
return plain[:-plain[-1]].decode("utf-8")
def vault_secret(vault_file, key, password_file):
"""Valeur 'key: valeur' du coffre (équivalent de ansible-vault view | grep)."""
try:
with open(password_file, "rb") as fh:
content = vault_decrypt(vault_file, fh.read().strip())
except (IOError, OSError, ValueError, IndexError, binascii.Error) as exc:
finish(UNKNOWN, "UNKNOWN - lecture du coffre %s impossible : %s" % (vault_file, exc))
for line in content.splitlines():
if key in line and ": " in line:
value = line.split(": ", 1)[1].strip()
if len(value) > 1 and value[0] == value[-1] and value[0] in "'\"":
value = value[1:-1]
return value
finish(UNKNOWN, "UNKNOWN - clé %s absente du coffre %s" % (key, vault_file))
def fortios_health_check_status(args, health_check, member, password):
"""Statut d'un membre de health-check SD-WAN via l'API REST FortiOS."""
jar = http.cookiejar.CookieJar()
opener = https_opener(urllib.request.HTTPCookieProcessor(jar))
base = "https://%s:%d" % (args.host, args.api_port)
try:
login = urllib.parse.urlencode({"username": args.api_username,
"secretkey": password}).encode()
opener.open(base + "/logincheck", data=login, timeout=HTTP_TIMEOUT).read()
token = next((cookie.value.strip('"') for cookie in jar if "token" in cookie.name), "")
request = urllib.request.Request(
base + "/api/v2/monitor/virtual-wan/health-check",
headers={"Content-Type": "application/json", "X-CSRFTOKEN": token})
data = json.load(opener.open(request, timeout=HTTP_TIMEOUT))
return ((data.get("results") or {}).get(health_check) or {}).get(member, {}).get("status")
except (urllib.error.URLError, OSError, ValueError, AttributeError):
return None
finally:
try:
opener.open(base + "/logout", timeout=HTTP_TIMEOUT).read()
except (urllib.error.URLError, OSError):
pass
# --------------------------------------------------------------------------
# Helpers communs
# --------------------------------------------------------------------------
def model_key(model):
for key in MODELS:
if re.search(r"(?<!\d)%s(?!\d)" % key, model):
return key
return None
def need(value, what):
if value is None:
finish(UNKNOWN, "UNKNOWN - %s non disponible" % what)
return value
def percent_usage(label, value):
"""Sortie commune CPU / mémoire : warning > 70, critical > 80."""
value = int(need(as_number(value), label))
code = CRITICAL if value > 80 else WARNING if value > 70 else OK
finish(code, "Current usage : %d%%" % value, "%s=%d;70;80;0;100" % (label, value))
def uptime_result(value):
"""TimeTicks (ou chaîne 'N days, ...') -> jours, sortie historique."""
if is_missing(value):
finish(UNKNOWN, "UNKNOWN - uptime non disponible")
if hasattr(value, "asOctets"):
match = re.search(r"(\d+)\s*day", as_str(value))
days = int(match.group(1)) if match else 0
else:
days = int(value) // 8640000
if days:
finish(OK, "OK - Uptime (in day): %d" % days, "uptime=%dday(s)" % days)
finish(OK, "OK - Uptime : 0", "uptime=0day")
def duration(seconds):
return "%dd%dh%dm" % (seconds // 86400, seconds % 86400 // 3600, seconds % 3600 // 60)
def rounded_days(seconds):
return (seconds + 43200) // 86400
def label_of(name):
return name.replace(" ", "_").replace("-", "_")
async def sensors(snmp):
return await snmp.table(FGT_SENSOR_ENTRY, (SENSOR_NAME, SENSOR_VALUE, SENSOR_ALARM))
async def health_checks(snmp, columns=(HC_LATENCY, HC_JITTER, HC_PACKET_LOSS)):
rows = await snmp.table(FGT_HEALTHCHECK_ENTRY, (HC_STATE, HC_IFNAME) + tuple(columns))
links = []
for row in rows.values():
links.append({
"name": as_str(row.get(HC_IFNAME)).strip('"'),
"state": as_number(row.get(HC_STATE)),
"latency": as_str(row.get(HC_LATENCY)).strip('"'),
"jitter": as_str(row.get(HC_JITTER)).strip('"'),
"loss": as_str(row.get(HC_PACKET_LOSS)).strip('"'),
"mos": as_str(row.get(HC_MOS)).strip('"'),
})
return links
def sla_level(link):
"""Niveau d'un lien SLA : latence/gigue 60/100 ms, perte 5/10 %."""
def num(text):
try:
return float(text)
except ValueError:
return 0.0
latency, jitter, loss = num(link["latency"]), num(link["jitter"]), num(link["loss"])
if latency > 100 or jitter > 100 or loss > 10:
return CRITICAL
if latency > 60 or jitter > 60 or loss > 5:
return WARNING
return OK
def status_lists(rows, name_of, keep):
"""Sépare les éléments (fan, PSU) en OK / DOWN selon fgHwSensorEntAlarmStatus."""
up, down = [], []
for row in rows.values():
name = as_str(row.get(SENSOR_NAME))
if not keep(name):
continue
status = as_number(row.get(SENSOR_ALARM))
if status == 0:
up.append("%s : OK" % name_of(name))
elif status == 1:
down.append("%s : DOWN" % name_of(name))
return ", ".join(up), ", ".join(down)
def up_down_result(up, down, label):
if down and not up:
finish(CRITICAL, down)
if down and up:
finish(WARNING, "%s - %s" % (down, up))
if up:
finish(OK, up)
finish(UNKNOWN, "Informations %s manquantes" % label)
# --------------------------------------------------------------------------
# FortiSwitch
# --------------------------------------------------------------------------
async def fsw_cpu(snmp, args, model):
percent_usage("CPU", (await snmp.get(FSW_CPU))[0])
async def fsw_uptime(snmp, args, model):
uptime_result((await snmp.get(SYS_UPTIME))[0])
async def fsw_poe(snmp, args, model):
usage, power_max = await snmp.get(PETH_MAIN_PSE_CONSUMPTION_1, PETH_MAIN_PSE_POWER_1)
power_max = int(need(as_number(power_max), "puissance PoE max"))
usage = int(need(as_number(usage), "consommation PoE"))
warn, crit = int(0.8 * power_max), int(0.9 * power_max)
percentage = int(usage * 100 / power_max) if power_max else 0
code = CRITICAL if usage > crit else WARNING if usage > warn else OK
finish(code, "PoE : %d/%dW (%d%%)" % (usage, power_max, percentage),
"PoE=%d;%d;%d;0;%d" % (usage, warn, crit, power_max))
async def fsw_temperature(snmp, args, model):
value = int(need(as_number((await snmp.get(ENT_SENSOR_VALUE_3))[0]), "température"))
warn, crit, maximum = 65, 70, 80
code = CRITICAL if value >= crit else WARNING if value >= warn else OK
finish(code, "%s : %d°C" % (("OK", "WARNING", "CRITICAL")[code], value),
"Temperature=%d;%d;%d;0;%d" % (value, warn, crit, maximum))
# --------------------------------------------------------------------------
# FortiGate
# --------------------------------------------------------------------------
async def fgt_cpu(snmp, args, model):
percent_usage("CPU", (await snmp.get(FGT_CPU))[0])
async def fgt_memory(snmp, args, model):
percent_usage("MEMORY", (await snmp.get(FGT_MEMORY))[0])
async def fgt_session(snmp, args, model):
"""Seuils par modèle pour l'alerte, perfdata sans seuils (comme le bash)."""
session = int(need(as_number((await snmp.get(FGT_SESSIONS))[0]), "nombre de sessions"))
perfdata = "Sessions=%d;;;0;" % session
limits = SESSION_LIMITS.get(model_key(model))
if limits is None:
finish(OK, "Sessions actives : %d (seuils non définis pour %s)" % (session, model),
perfdata)
warn, crit, maximum = limits
code = CRITICAL if session > crit else WARNING if session > warn else OK
finish(code, "Sessions actives : %d (%d%%)" % (session, session * 100 // maximum), perfdata)
async def fgt_master(snmp, args, model):
"""Actif si ce membre (nom Centreon) est le master du cluster HA."""
if not args.hostname:
finish(UNKNOWN, "UNKNOWN - --hostname obligatoire pour le périmètre %s" % args.perimetre)
rows = await snmp.table(FGT_HA_ENTRY, (HA_SERIAL, HA_HOSTNAME, HA_MASTER_SERIAL))
if not rows:
finish(OK, "Actif")
for row in rows.values():
if args.hostname.lower() in as_str(row.get(HA_HOSTNAME)).lower():
is_master = as_str(row.get(HA_MASTER_SERIAL)) == as_str(row.get(HA_SERIAL))
finish(OK, "Actif" if is_master else "Passif")
finish(UNKNOWN, "UNKNOWN - membre HA '%s' introuvable" % args.hostname)
async def fgt_cluster(snmp, args, model):
rows = await snmp.table(FGT_HA_ENTRY, (HA_INDEX, HA_HOSTNAME, HA_SYNC))
if len(rows) < 2:
finish(OK, "Standalone")
synced, unsynced = [], []
for row in rows.values():
name = as_str(row.get(HA_HOSTNAME))
status = as_number(row.get(HA_SYNC))
if status == 1:
synced.append("%s : synchronized" % name)
elif status == 0:
unsynced.append("%s : unsynchronized" % name)
synced, unsynced = ", ".join(synced), ", ".join(unsynced)
if unsynced and not synced:
finish(CRITICAL, "Nodes unsynchronized : %s" % unsynced)
if unsynced:
finish(WARNING, "Nodes unsynchronized : %s - %s" % (unsynced, synced))
if synced:
finish(OK, "HA CLUSTER OK : %s" % synced)
finish(UNKNOWN, "Informations HA Sync manquantes")
async def fgt_sla(snmp, args, model):
"""Tous les liens O-INT / O-BVPN / O-BES2 vivants (les liens morts sont ignorés)."""
texts = {OK: [], WARNING: [], CRITICAL: []}
perfdata = []
for link in await health_checks(snmp):
if not re.search(SLA_LINKS, link["name"]) or link["state"] == 1:
continue
name = link["name"]
texts[sla_level(link)].append(
"%s_Latency=%sms ; %s_Jitter=%sms ; %s_Packet Loss=%s%%"
% (name, link["latency"], name, link["jitter"], name, link["loss"]))
perfdata.append(
"%s_Latency=%s;60;100;0;200 %s_Jitter=%s;60;100;0;200 %s_Packet_Loss=%s;5;10;0;100"
% (name, link["latency"], name, link["jitter"], name, link["loss"]))
if not perfdata:
finish(UNKNOWN, "UNKNOWN - aucun lien SLA actif (%s)" % SLA_LINKS)
code = CRITICAL if texts[CRITICAL] else WARNING if texts[WARNING] else OK
parts = [" ".join(texts[level]) for level in (CRITICAL, WARNING, OK) if texts[level]]
finish(code, " - ".join(parts), " ".join(perfdata))
async def fgt_sla_single(snmp, args, model):
"""Premier lien correspondant au périmètre (sla-o-*)."""
pattern = SLA_SINGLE_LINK[args.perimetre]
for link in await health_checks(snmp):
if not re.search(pattern, link["name"], re.IGNORECASE):
continue
if link["state"] == 1:
finish(UNKNOWN, "%s : DOWN" % link["name"])
finish(sla_level(link),
"Latency = %sms ; Jitter = %sms ; Packet Loss = %s%%"
% (link["latency"], link["jitter"], link["loss"]),
"Latency=%s;60;100;0;200 Jitter=%s;60;100;0;200 Packet_Loss=%s;5;10;0;100"
% (link["latency"], link["jitter"], link["loss"]))
finish(UNKNOWN, "UNKNOWN - aucun lien %s trouvé" % pattern)
async def fgt_mos(snmp, args, model):
"""MOS des liens O-INT / O-BVPN / O-BES2 vivants : warning < 3, critical < 2."""
code, texts, perfdata = OK, [], []
for link in await health_checks(snmp, (HC_MOS,)):
if not re.search(SLA_LINKS, link["name"]) or link["state"] == 1:
continue
try:
mos = float(link["mos"])
except ValueError:
texts.append("%s_MOS=N/A" % link["name"])
continue
if mos < MOS_CRITICAL:
code = CRITICAL
elif mos < MOS_WARNING:
code = max(code, WARNING)
texts.append("%s_MOS=%s" % (link["name"], link["mos"]))
perfdata.append("%s_MOS=%s;%s:;%s:;0;5"
% (link["name"], link["mos"], MOS_WARNING, MOS_CRITICAL))
if not texts:
finish(UNKNOWN, "UNKNOWN - aucun lien SLA actif (%s)" % SLA_LINKS)
finish(code, " ".join(texts), " ".join(perfdata))
async def fgt_hc_google(snmp, args, model):
"""Statut du health-check SD-WAN HC_Google (API REST FortiOS)."""
member, vault_file, vault_key = HC_GOOGLE[args.perimetre]
password = args.api_password or vault_secret(vault_file, vault_key, args.vault_password_file)
status = fortios_health_check_status(args, "HC_Google", member, password)
if status == "up":
finish(OK, "HC_Google = up", "HC_Google=1;0;0;0;1")
if status == "down":
finish(CRITICAL, "HC_Google = down", "HC_Google=0;0;0;0;1")
finish(UNKNOWN, "Aucune information", "HC_Google=0;0;0;0;1")
async def fgt_temperature(snmp, args, model):
config = TEMP_SENSORS.get(model_key(model))
if config is None:
finish(UNKNOWN, "UNKNOWN - modèle %s non géré pour les températures" % model)
name_filter, name_of, warn, crit = config
texts = {OK: [], WARNING: [], CRITICAL: []}
perfdata = []
for row in (await sensors(snmp)).values():
raw_name = as_str(row.get(SENSOR_NAME))
if name_filter and not re.search(name_filter, raw_name, re.IGNORECASE):
continue
value = as_number(row.get(SENSOR_VALUE))
if value is None:
continue
name, value = name_of(raw_name), int(value)
level = CRITICAL if value > crit else WARNING if value > warn else OK
texts[level].append("%s=%d°C" % (name, value))
perfdata.append("%s=%d;%d;%d;0;%d" % (name, value, warn, crit, TEMP_MAX))
if not perfdata:
finish(UNKNOWN, "UNKNOWN - aucun capteur de température trouvé")
code = CRITICAL if texts[CRITICAL] else WARNING if texts[WARNING] else OK
parts = [" - ".join(texts[CRITICAL]), " - ".join(texts[WARNING]), " ".join(texts[OK])]
finish(code, " - ".join(p for p in parts if p), " ".join(perfdata))
async def fgt_fan(snmp, args, model):
up, down = status_lists(await sensors(snmp), lambda n: n,
lambda n: "fan" in n.lower())
up_down_result(up, down, "FAN")
async def fgt_psu(snmp, args, model):
key = model_key(model)
if key == "100":
def keep(name):
return "PS" in name
def name_of(name):
return name.replace("]", "", 1).replace("[", "", 1).replace(" ", "", 1)
elif key in ("600", "3000"):
def keep(name):
return "PS" in name and "status" in name.lower()
def name_of(name):
return name
else:
finish(UNKNOWN, "Informations PSU manquantes")
up, down = status_lists(await sensors(snmp), name_of, keep)
up_down_result(up, down, "PSU")
async def fgt_tunnels(snmp, args, model):
"""
Mode nominal = INT down + secours (BVPN/BES2) up. Les deux montés en même
temps n'est pas normal -> WARNING.
"""
config = TUNNEL_BACKUP.get(model_key(model))
if config is None:
finish(UNKNOWN, "UNKNOWN - modèle %s non géré pour les tunnels" % model)
backup, rename_from, rename_to = config
rows = await snmp.table(FGT_VPN_TUNNEL_ENTRY, (VPN_NAME, VPN_STATUS))
up, down = [], []
for row in rows.values():
name = re.sub(rename_from, rename_to, as_str(row.get(VPN_NAME)))
status = as_number(row.get(VPN_STATUS))
if status == 2:
up.append(name)
elif status == 1:
down.append(name)
def has(names, word):
return any(word in name for name in names)
total = len(rows)
up_text = ", ".join("%s : up" % name for name in up)
down_text = ", ".join("%s : down" % name for name in down)
if has(down, "INT") and has(down, backup):
finish(CRITICAL, "%d/%d tunnel monté : %s - %s" % (len(up), total, up_text, down_text))
if has(up, "INT") and has(up, backup):
finish(WARNING, "%d/%d tunnels montés : %s - %s" % (len(up), total, up_text, down_text))
if has(up, "INT") and has(down, backup):
finish(WARNING, "Mode Secours = %d/%d : %s - %s" % (len(up), total, up_text, down_text))
if has(down, "INT") and has(up, backup):
finish(OK, "Mode Nominal = %d/%d : %s - %s" % (len(up), total, up_text, down_text))
finish(UNKNOWN, "Informations manquantes")
async def fgt_hub_tunnels(snmp, args, model):
tunnels = int(need(as_number((await snmp.get(FGT_VPN_UP_COUNT))[0]), "nombre de tunnels"))
finish(OK, "Nombre de tunnels UP : %d" % tunnels, "Tunnels_UP=%d;;;0;800" % tunnels)
async def fgt_uptime(snmp, args, model):
uptime_result((await snmp.get(FGT_UPTIME))[0])
async def fgt_uptime_interfaces(snmp, args, model):
"""Durée depuis le dernier changement d'état des interfaces WAN (toujours OK)."""
descriptions = (await snmp.walk(IF_DESCR))[oid_tuple(IF_DESCR)]
interfaces = [(index[0], as_str(value)) for index, value in descriptions.items()
if re.search(UPTIME_INTERFACES, as_str(value), re.IGNORECASE)]
if not interfaces:
finish(UNKNOWN, "UNKNOWN - aucune interface %s trouvée" % UPTIME_INTERFACES)
oids = [SYS_UPTIME]
for index, _ in interfaces:
oids += ["%s.%d" % (IF_OPER_STATUS, index), "%s.%d" % (IF_LAST_CHANGE, index)]
values = await snmp.get(*oids)
sys_uptime = int(need(as_number(values[0]), "sysUpTime"))
texts, perfdata = [], []
for position, (index, name) in enumerate(interfaces):
oper_status = as_number(values[1 + 2 * position])
last_change = as_number(values[2 + 2 * position]) or 0
days = 0
if oper_status != 1:
status = "DOWN"
elif sys_uptime >= last_change:
seconds = (sys_uptime - last_change) // 100
status = "Up (%s)" % duration(seconds)
days = rounded_days(seconds)
else:
status = "Up (Stable)"
texts.append("%s: %s" % (name, status))
perfdata.append("%s=%d;;;;" % (label_of(name), days))
finish(OK, "OK : %s " % " - ".join(texts), " ".join(perfdata))
async def fgt_bgp(snmp, args, model):
"""État et durée d'établissement des voisins BGP : CRITICAL si un voisin n'est pas Established."""
rows = await snmp.table(BGP_PEER_ENTRY, (BGP_STATE, BGP_ESTABLISHED_TIME))
if not rows:
finish(UNKNOWN, "UNKNOWN : Aucun voisin BGP trouvé")
code, texts, perfdata = OK, [], []
for index, row in rows.items():
peer = ".".join(str(part) for part in index[:4])
name = BGP_PEERS.get(peer, peer)
state_value = as_number(row.get(BGP_STATE))
if state_value != BGP_ESTABLISHED:
code = CRITICAL
seconds = int(as_number(row.get(BGP_ESTABLISHED_TIME)) or 0)
texts.append("%s: %s (%s)" % (name, BGP_STATES.get(state_value, "Down"),
duration(seconds)))
perfdata.append("%s=%dd;;;;" % (label_of(name), rounded_days(seconds)))
finish(code, "%s " % " - ".join(texts), " ".join(perfdata))
FORTISWITCH = {
"cpu": fsw_cpu,
"uptime": fsw_uptime,
"poe": fsw_poe,
"temperature": fsw_temperature,
}
FORTIGATE = {
"uptime-interfaces": fgt_uptime_interfaces,
"bgp": fgt_bgp,
"cpu": fgt_cpu,
"state-cluster": fgt_master,
"master": fgt_master,
"memory": fgt_memory,
"session": fgt_session,
"sla": fgt_sla,
"sla-o-bvpn-hz": fgt_sla_single,
"sla-o-bvpn-ren": fgt_sla_single,
"sla-o-int_obs": fgt_sla_single,
"sla-o-int_sfr": fgt_sla_single,
"sla-HC_Google": fgt_hc_google,
"sla-HC_Google_Agences": fgt_hc_google,
"cluster": fgt_cluster,
"temperature": fgt_temperature,
"fan": fgt_fan,
"psu": fgt_psu,
"tunnels": fgt_tunnels,
"hub-tunnels": fgt_hub_tunnels,
"uptime": fgt_uptime,
"mos": fgt_mos,
}
# --------------------------------------------------------------------------
# Main
# --------------------------------------------------------------------------
def device_family(model):
lowered = model.lower()
if "fortiswitch" in lowered:
return FORTISWITCH
if "fortigate" in lowered or "fgt" in lowered or "fgmodel" in lowered:
return FORTIGATE
finish(UNKNOWN, "Modele non reconnu")
async def run(args):
snmp = Snmp(args)
try:
await snmp.open()
if args.model_source == "ipam":
model = ipam_model(args)
else:
model = as_str((await snmp.get(ENT_PHYSICAL_DESCR_1))[0])
check = device_family(model).get(args.perimetre)
if check is None:
finish(UNKNOWN, "Perimètre non reconnu")
await check(snmp, args, model)
finish(UNKNOWN, "UNKNOWN - aucun résultat pour le périmètre %s" % args.perimetre)
except SnmpError as exc:
finish(UNKNOWN, "No response from remote host %s (SNMP : %s)" % (args.host, exc))
finally:
snmp.close()
def parse_args():
parser = argparse.ArgumentParser(description="Supervision FortiGate / FortiSwitch (SNMPv3).")
parser.add_argument("-A", dest="perimetre", required=True, help="périmètre à contrôler")
parser.add_argument("-B", dest="host", required=True, help="adresse de l'équipement")
# Options historiques du script bash, acceptées et ignorées.
for letter in "CDEFGHI":
parser.add_argument("-" + letter, dest="ignored_" + letter, help=argparse.SUPPRESS)
parser.add_argument("--hostname", help="nom Centreon du host (state-cluster / master)")
snmp = parser.add_argument_group("SNMPv3")
snmp.add_argument("--snmp-username", required=True)
snmp.add_argument("--authpassphrase")
snmp.add_argument("--privpassphrase")
snmp.add_argument("--authprotocol", default="SHA", type=str.upper,
choices=sorted(AUTH_PROTOCOLS))
snmp.add_argument("--privprotocol", default="AES", type=str.upper,
choices=sorted(PRIV_PROTOCOLS))
snmp.add_argument("--snmp-context", default="", help="contexte SNMPv3 (vide par défaut)")
snmp.add_argument("--port", type=int, default=161)
snmp.add_argument("--timeout", type=float, default=2, help="timeout par requête (s)")
snmp.add_argument("--retries", type=int, default=1)
model = parser.add_argument_group("modèle")
model.add_argument("--model-source", choices=("ipam", "snmp"), default="ipam",
help="ipam (défaut, comme le bash) ou snmp (entPhysicalDescr.1)")
model.add_argument("--ipam-url", default=IPAM_URL)
model.add_argument("--ipam-username")
model.add_argument("--ipam-password")
api = parser.add_argument_group("API FortiOS (sla-HC_Google*)")
api.add_argument("--api-username", default="centreon")
api.add_argument("--api-password", help="sinon lu dans le coffre ansible-vault")
api.add_argument("--api-port", type=int, default=443)
api.add_argument("--vault-password-file", default=VAULT_PASSWORD_FILE)
return parser.parse_args()
def main():
args = parse_args()
try:
asyncio.run(asyncio.wait_for(run(args), CHECK_DEADLINE))
except PluginExit as result:
print(result.output)
sys.exit(result.code)
except asyncio.TimeoutError:
print("UNKNOWN - délai de %d s dépassé sur %s" % (CHECK_DEADLINE, args.host))
sys.exit(UNKNOWN)
if __name__ == "__main__":
main()