347 lines
13 KiB
Python
347 lines
13 KiB
Python
"""Modele de donnees d'un projet.
|
|
|
|
Un projet est ce que l'utilisateur construit dans l'editeur : des commandes de
|
|
controle, des modeles de service, des modeles d'hote, et facultativement des
|
|
hotes. Les objets se referencent par leur nom, comme dans Centreon.
|
|
|
|
Le projet est serialisable en JSON : c'est ce que le navigateur conserve, ce
|
|
qu'on exporte, ce qu'on partage et ce que la CI rejoue.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, field
|
|
from typing import Any
|
|
|
|
SCHEMA_VERSION = 1
|
|
|
|
# Versions de Centreon proposees dans l'editeur et ce qu'elles impliquent.
|
|
CENTREON_VERSIONS: dict[str, dict[str, Any]] = {
|
|
"24.x": {"label": "Centreon 24.x", "macro_fields": 5},
|
|
"23.x": {"label": "Centreon 23.x", "macro_fields": 5},
|
|
"22.x": {"label": "Centreon 22.x", "macro_fields": 5},
|
|
"21.x": {"label": "Centreon 21.x", "macro_fields": 5},
|
|
"20.04": {"label": "Centreon 20.04", "macro_fields": 5},
|
|
"legacy": {"label": "Centreon anterieur a 20.04", "macro_fields": 3},
|
|
}
|
|
|
|
COMMAND_TYPES = {"check": "Controle", "notif": "Notification", "misc": "Divers", "discovery": "Decouverte"}
|
|
|
|
FORBIDDEN = ";"
|
|
|
|
|
|
def _clean(value: Any) -> str:
|
|
return "" if value is None else str(value).strip()
|
|
|
|
|
|
@dataclass
|
|
class Macro:
|
|
name: str = ""
|
|
value: str = ""
|
|
description: str = ""
|
|
is_password: bool = False
|
|
|
|
@classmethod
|
|
def from_dict(cls, raw: dict[str, Any]) -> "Macro":
|
|
return cls(
|
|
name=_clean(raw.get("name")).upper(),
|
|
value="" if raw.get("value") is None else str(raw.get("value")),
|
|
description=_clean(raw.get("description")),
|
|
is_password=bool(raw.get("is_password")),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Command:
|
|
name: str = ""
|
|
type: str = "check"
|
|
line: str = ""
|
|
comment: str = ""
|
|
|
|
@classmethod
|
|
def from_dict(cls, raw: dict[str, Any]) -> "Command":
|
|
return cls(
|
|
name=_clean(raw.get("name")),
|
|
type=_clean(raw.get("type")) or "check",
|
|
line=_clean(raw.get("line")),
|
|
comment=_clean(raw.get("comment")),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class ServiceTemplate:
|
|
name: str = ""
|
|
alias: str = ""
|
|
parent: str = "generic-active-service-custom"
|
|
command: str = ""
|
|
command_args: str = ""
|
|
check_interval: int = 5
|
|
retry_interval: int = 1
|
|
max_attempts: int = 3
|
|
active_checks: int = 1
|
|
passive_checks: int = 0
|
|
host_templates: list[str] = field(default_factory=list)
|
|
macros: list[Macro] = field(default_factory=list)
|
|
|
|
@classmethod
|
|
def from_dict(cls, raw: dict[str, Any]) -> "ServiceTemplate":
|
|
def as_int(key: str, default: int) -> int:
|
|
try:
|
|
return int(raw.get(key, default))
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
return cls(
|
|
name=_clean(raw.get("name")),
|
|
alias=_clean(raw.get("alias")),
|
|
parent=_clean(raw.get("parent")),
|
|
command=_clean(raw.get("command")),
|
|
command_args=_clean(raw.get("command_args")),
|
|
check_interval=as_int("check_interval", 5),
|
|
retry_interval=as_int("retry_interval", 1),
|
|
max_attempts=as_int("max_attempts", 3),
|
|
active_checks=as_int("active_checks", 1),
|
|
passive_checks=as_int("passive_checks", 0),
|
|
host_templates=[_clean(h) for h in raw.get("host_templates") or [] if _clean(h)],
|
|
macros=[Macro.from_dict(m) for m in raw.get("macros") or []],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class HostTemplate:
|
|
name: str = ""
|
|
alias: str = ""
|
|
parents: list[str] = field(default_factory=list)
|
|
check_command: str = ""
|
|
check_command_args: str = ""
|
|
snmp_version: str = ""
|
|
snmp_community: str = ""
|
|
timezone: str = ""
|
|
macros: list[Macro] = field(default_factory=list)
|
|
|
|
@classmethod
|
|
def from_dict(cls, raw: dict[str, Any]) -> "HostTemplate":
|
|
return cls(
|
|
name=_clean(raw.get("name")),
|
|
alias=_clean(raw.get("alias")),
|
|
parents=[_clean(p) for p in raw.get("parents") or [] if _clean(p)],
|
|
check_command=_clean(raw.get("check_command")),
|
|
check_command_args=_clean(raw.get("check_command_args")),
|
|
snmp_version=_clean(raw.get("snmp_version")),
|
|
snmp_community=_clean(raw.get("snmp_community")),
|
|
timezone=_clean(raw.get("timezone")),
|
|
macros=[Macro.from_dict(m) for m in raw.get("macros") or []],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Host:
|
|
name: str = ""
|
|
alias: str = ""
|
|
address: str = ""
|
|
templates: list[str] = field(default_factory=list)
|
|
poller: str = "Central"
|
|
hostgroups: str = ""
|
|
macros: list[Macro] = field(default_factory=list)
|
|
|
|
@classmethod
|
|
def from_dict(cls, raw: dict[str, Any]) -> "Host":
|
|
return cls(
|
|
name=_clean(raw.get("name")),
|
|
alias=_clean(raw.get("alias")),
|
|
address=_clean(raw.get("address")),
|
|
templates=[_clean(t) for t in raw.get("templates") or [] if _clean(t)],
|
|
poller=_clean(raw.get("poller")) or "Central",
|
|
hostgroups=_clean(raw.get("hostgroups")),
|
|
macros=[Macro.from_dict(m) for m in raw.get("macros") or []],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Project:
|
|
name: str = "Nouveau projet"
|
|
centreon_version: str = "24.x"
|
|
prefix: str = ""
|
|
plugins_dir: str = "/usr/lib/centreon/plugins"
|
|
poller: str = "Central"
|
|
apply_config: bool = True
|
|
commands: list[Command] = field(default_factory=list)
|
|
service_templates: list[ServiceTemplate] = field(default_factory=list)
|
|
host_templates: list[HostTemplate] = field(default_factory=list)
|
|
hosts: list[Host] = field(default_factory=list)
|
|
schema: int = SCHEMA_VERSION
|
|
|
|
@property
|
|
def macro_fields(self) -> int:
|
|
return CENTREON_VERSIONS.get(self.centreon_version, {}).get("macro_fields", 5)
|
|
|
|
@classmethod
|
|
def from_dict(cls, raw: dict[str, Any] | None) -> "Project":
|
|
raw = raw or {}
|
|
version = _clean(raw.get("centreon_version")) or "24.x"
|
|
if version not in CENTREON_VERSIONS:
|
|
version = "24.x"
|
|
return cls(
|
|
name=_clean(raw.get("name")) or "Nouveau projet",
|
|
centreon_version=version,
|
|
prefix=_clean(raw.get("prefix")),
|
|
plugins_dir=_clean(raw.get("plugins_dir")) or "/usr/lib/centreon/plugins",
|
|
poller=_clean(raw.get("poller")) or "Central",
|
|
apply_config=bool(raw.get("apply_config", True)),
|
|
commands=[Command.from_dict(c) for c in raw.get("commands") or []],
|
|
service_templates=[ServiceTemplate.from_dict(s) for s in raw.get("service_templates") or []],
|
|
host_templates=[HostTemplate.from_dict(h) for h in raw.get("host_templates") or []],
|
|
hosts=[Host.from_dict(h) for h in raw.get("hosts") or []],
|
|
)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
def merge(self, other: "Project") -> list[str]:
|
|
"""Ajoute les objets d'un autre projet, en ignorant les noms deja pris."""
|
|
skipped: list[str] = []
|
|
|
|
def add(target: list[Any], items: list[Any], label: str) -> None:
|
|
existing = {obj.name for obj in target}
|
|
for item in items:
|
|
if item.name in existing:
|
|
skipped.append(f"{label} deja present, ignore : {item.name}")
|
|
continue
|
|
target.append(item)
|
|
existing.add(item.name)
|
|
|
|
add(self.commands, other.commands, "Commande")
|
|
add(self.service_templates, other.service_templates, "Modele de service")
|
|
add(self.host_templates, other.host_templates, "Modele d'hote")
|
|
add(self.hosts, other.hosts, "Hote")
|
|
return skipped
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Validation
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def validate(project: Project) -> list[dict[str, str]]:
|
|
"""Retourne la liste des problemes, du bloquant au simple avertissement."""
|
|
|
|
issues: list[dict[str, str]] = []
|
|
|
|
def error(message: str) -> None:
|
|
issues.append({"level": "error", "message": message})
|
|
|
|
def warn(message: str) -> None:
|
|
issues.append({"level": "warning", "message": message})
|
|
|
|
def check_names(items: list[Any], label: str) -> None:
|
|
seen: set[str] = set()
|
|
for index, item in enumerate(items, start=1):
|
|
if not item.name:
|
|
error(f"{label} n{chr(176)}{index} : le nom est vide.")
|
|
continue
|
|
if item.name in seen:
|
|
error(f"{label} : nom en double, {item.name}")
|
|
seen.add(item.name)
|
|
if FORBIDDEN in item.name:
|
|
error(f"{label} {item.name} : le point-virgule est le separateur CLAPI.")
|
|
|
|
check_names(project.commands, "Commande")
|
|
check_names(project.service_templates, "Modele de service")
|
|
check_names(project.host_templates, "Modele d'hote")
|
|
check_names(project.hosts, "Hote")
|
|
|
|
command_names = {c.name for c in project.commands}
|
|
htpl_names = {h.name for h in project.host_templates}
|
|
|
|
for command in project.commands:
|
|
if not command.line:
|
|
error(f"Commande {command.name} : la ligne de commande est vide.")
|
|
elif FORBIDDEN in command.line:
|
|
error(
|
|
f"Commande {command.name} : la ligne contient un point-virgule, "
|
|
"ce qui casserait le fichier d'import."
|
|
)
|
|
|
|
for stpl in project.service_templates:
|
|
if not stpl.command:
|
|
warn(f"Modele de service {stpl.name} : aucune commande de controle associee.")
|
|
elif stpl.command not in command_names:
|
|
warn(
|
|
f"Modele de service {stpl.name} : la commande {stpl.command} n'est pas dans "
|
|
"le projet, elle devra exister deja sur le central."
|
|
)
|
|
if not stpl.host_templates:
|
|
warn(
|
|
f"Modele de service {stpl.name} : rattache a aucun modele d'hote, "
|
|
"il ne sera deploye sur aucun hote."
|
|
)
|
|
for htpl in stpl.host_templates:
|
|
if htpl not in htpl_names:
|
|
warn(f"Modele de service {stpl.name} : modele d'hote inconnu, {htpl}")
|
|
_check_macros(stpl.macros, f"Modele de service {stpl.name}", issues)
|
|
|
|
for htpl in project.host_templates:
|
|
_check_macros(htpl.macros, f"Modele d'hote {htpl.name}", issues)
|
|
|
|
for host in project.hosts:
|
|
if not host.address:
|
|
error(f"Hote {host.name} : l'adresse est vide.")
|
|
for template in host.templates:
|
|
if template not in htpl_names:
|
|
warn(f"Hote {host.name} : modele d'hote inconnu, {template}")
|
|
_check_macros(host.macros, f"Hote {host.name}", issues)
|
|
|
|
if not project.commands and not project.service_templates and not project.host_templates:
|
|
warn("Le projet est vide.")
|
|
|
|
return issues
|
|
|
|
|
|
def _check_macros(macros: list[Macro], label: str, issues: list[dict[str, str]]) -> None:
|
|
seen: set[str] = set()
|
|
for macro in macros:
|
|
if not macro.name:
|
|
issues.append({"level": "error", "message": f"{label} : une macro n'a pas de nom."})
|
|
continue
|
|
if macro.name in seen:
|
|
issues.append({"level": "error", "message": f"{label} : macro en double, {macro.name}"})
|
|
seen.add(macro.name)
|
|
if FORBIDDEN in macro.value or FORBIDDEN in macro.description:
|
|
issues.append(
|
|
{
|
|
"level": "error",
|
|
"message": f"{label}, macro {macro.name} : le point-virgule est interdit.",
|
|
}
|
|
)
|
|
if macro.is_password and macro.value:
|
|
issues.append(
|
|
{
|
|
"level": "warning",
|
|
"message": (
|
|
f"{label}, macro {macro.name} : un mot de passe est renseigne. "
|
|
"Il apparaitra en clair dans les fichiers generes."
|
|
),
|
|
}
|
|
)
|
|
|
|
|
|
def detect_macros(line: str) -> dict[str, list[str]]:
|
|
"""Repere les macros personnalisees utilisees dans une ligne de commande."""
|
|
import re
|
|
|
|
service = sorted(set(re.findall(r"\$_SERVICE([A-Z0-9_]+)\$", line)))
|
|
host = sorted(set(re.findall(r"\$_HOST([A-Z0-9_]+)\$", line)))
|
|
return {"service": service, "host": host}
|
|
|
|
|
|
def strip_secrets(project: Project) -> Project:
|
|
"""Vide toutes les macros marquees mot de passe.
|
|
|
|
Applique avant tout stockage cote serveur : partage, projet enregistre,
|
|
soumission de pack. Le site n'a aucune raison de detenir ces valeurs.
|
|
"""
|
|
for holder in (*project.service_templates, *project.host_templates, *project.hosts):
|
|
for macro in holder.macros:
|
|
if macro.is_password:
|
|
macro.value = ""
|
|
return project
|