initial

config.yaml

@@ -0,0 +1,10 @@
+phpipam:
+  url: "http://ipam.canalias.fr"   # URL de base de phpIPAM (sans slash final)
+  app_id: "dnsmasq"                       # App ID déclaré dans Administration > API
+  token: "gBF4_BdN0F9k-rAb03d8t_zsUEULgLby"           # App token (onglet App security = Token ou None)
+  verify_ssl: false                      # Mettre false pour les certificats auto-signés
+
+dnsmasq:
+  output_file: "/tmp/reservations.conf"  # Fichier de sortie
+  # Optionnel : commentaire d'en-tête dans le fichier généré
+  header_comment: "Généré automatiquement par dnsmasq_sync.py — NE PAS MODIFIER"

dnsmasq_sync.py

@@ -0,0 +1,279 @@
+#!/usr/bin/env python3
+"""
+dnsmasq_sync.py — Génère une configuration dnsmasq (dhcp-host) à partir de phpIPAM.
+
+Usage:
+    python dnsmasq_sync.py [--config config.yaml] [--dry-run]
+
+Prérequis :
+    - phpIPAM avec API activée (Administration > phpIPAM API)
+    - App_security = "Token" (ou "None" sans token requis)
+    - Droits lecture sur les adresses / sous-réseaux
+"""
+
+import argparse
+import ipaddress
+import logging
+import sys
+from datetime import datetime
+from pathlib import Path
+
+import requests
+import yaml
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+log = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Config
+# ---------------------------------------------------------------------------
+
+def load_config(path: str) -> dict:
+    config_path = Path(path)
+    if not config_path.exists():
+        log.error("Fichier de configuration introuvable : %s", path)
+        sys.exit(1)
+    with config_path.open() as fh:
+        cfg = yaml.safe_load(fh)
+    _validate_config(cfg)
+    return cfg
+
+
+def _validate_config(cfg: dict) -> None:
+    required = {
+        "phpipam": ["url", "app_id"],
+        "dnsmasq": ["output_file"],
+    }
+    for section, keys in required.items():
+        if section not in cfg:
+            log.error("Section '%s' manquante dans la configuration.", section)
+            sys.exit(1)
+        for key in keys:
+            if key not in cfg[section]:
+                log.error("Clé '%s.%s' manquante dans la configuration.", section, key)
+                sys.exit(1)
+
+
+# ---------------------------------------------------------------------------
+# phpIPAM API client
+# ---------------------------------------------------------------------------
+
+class PhpIpamClient:
+    def __init__(self, url: str, app_id: str, token: str | None, verify_ssl: bool = True):
+        self.base_url = url.rstrip("/")
+        self.app_id = app_id
+        self.session = requests.Session()
+        self.session.verify = verify_ssl
+        if token:
+            # phpIPAM attend le token dans l'en-tête "token"
+            self.session.headers.update({"token": token})
+
+    def _get(self, endpoint: str) -> list | dict:
+        url = f"{self.base_url}/api/{self.app_id}/{endpoint.lstrip('/')}"
+        try:
+            resp = self.session.get(url, timeout=30)
+            resp.raise_for_status()
+        except requests.exceptions.SSLError:
+            log.error(
+                "Erreur SSL. Si vous utilisez un certificat auto-signé, "
+                "passez verify_ssl: false dans la config."
+            )
+            sys.exit(1)
+        except requests.exceptions.ConnectionError as exc:
+            log.error("Impossible de joindre phpIPAM (%s) : %s", url, exc)
+            sys.exit(1)
+        except requests.exceptions.HTTPError as exc:
+            log.error("Erreur HTTP %s sur %s : %s", resp.status_code, url, exc)
+            sys.exit(1)
+
+        data = resp.json()
+        if not data.get("success"):
+            # 404 "No addresses found" n'est pas une erreur fatale
+            if data.get("code") == 404:
+                return []
+            log.warning("Réponse phpIPAM non-success pour %s : %s", endpoint, data.get("message"))
+            return []
+        return data.get("data") or []
+
+    def get_sections(self) -> list[dict]:
+        return self._get("sections/")
+
+    def get_subnets_for_section(self, section_id: int) -> list[dict]:
+        return self._get(f"sections/{section_id}/subnets/")
+
+    def get_addresses_for_subnet(self, subnet_id: int) -> list[dict]:
+        return self._get(f"subnets/{subnet_id}/addresses/")
+
+
+# ---------------------------------------------------------------------------
+# Collecte des hôtes avec MAC
+# ---------------------------------------------------------------------------
+
+def collect_hosts(client: PhpIpamClient) -> list[dict]:
+    """
+    Parcourt toutes les sections → sous-réseaux → adresses
+    et retourne les hôtes ayant une adresse MAC définie.
+    """
+    hosts = []
+    seen_ids: set[str] = set()
+
+    sections = client.get_sections()
+    log.info("%d section(s) trouvée(s).", len(sections))
+
+    for section in sections:
+        section_id = int(section["id"])
+        section_name = section.get("name", f"section-{section_id}")
+        subnets = client.get_subnets_for_section(section_id)
+        log.info("  Section '%s' : %d sous-réseau(x).", section_name, len(subnets))
+
+        for subnet in subnets:
+            subnet_id = int(subnet["id"])
+            addresses = client.get_addresses_for_subnet(subnet_id)
+
+            for addr in addresses:
+                addr_id = str(addr.get("id", ""))
+                mac = _normalize_mac(addr.get("mac", ""))
+                ip = addr.get("ip", "").strip()
+
+                # Ignorer si pas de MAC, pas d'IP, ou déjà traité
+                if not mac or not ip or addr_id in seen_ids:
+                    continue
+
+                # Valider l'IP
+                try:
+                    ipaddress.ip_address(ip)
+                except ValueError:
+                    log.warning("IP invalide ignorée : %s (id=%s)", ip, addr_id)
+                    continue
+
+                seen_ids.add(addr_id)
+                hosts.append({
+                    "mac": mac,
+                    "ip": ip,
+                    "hostname": _sanitize_hostname(addr.get("hostname") or addr.get("description") or ""),
+                })
+
+    return hosts
+
+
+def _normalize_mac(raw: str) -> str:
+    """Normalise une adresse MAC en xx:xx:xx:xx:xx:xx minuscule."""
+    if not raw:
+        return ""
+    mac = raw.strip().lower().replace("-", ":").replace(".", ":")
+    # Supprimer les séparateurs puis reformater
+    digits = mac.replace(":", "")
+    if len(digits) != 12 or not all(c in "0123456789abcdef" for c in digits):
+        log.warning("MAC ignorée (format invalide) : %s", raw)
+        return ""
+    return ":".join(digits[i:i+2] for i in range(0, 12, 2))
+
+
+def _sanitize_hostname(name: str) -> str:
+    """Retourne un hostname valide ou une chaîne vide."""
+    if not name:
+        return ""
+    # Remplacer les espaces et caractères non-ASCII par '-'
+    sanitized = "".join(c if c.isalnum() or c in "-." else "-" for c in name.strip())
+    # dnsmasq n'accepte pas les noms commençant/finissant par '-'
+    return sanitized.strip("-")
+
+
+# ---------------------------------------------------------------------------
+# Génération du fichier dnsmasq
+# ---------------------------------------------------------------------------
+
+def generate_config(hosts: list[dict], header_comment: str) -> str:
+    lines = [
+        f"# {header_comment}",
+        f"# Généré le : {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
+        f"# Entrées : {len(hosts)}",
+        "",
+    ]
+
+    for host in sorted(hosts, key=lambda h: tuple(int(x) for x in h["ip"].split("."))):
+        if host["hostname"]:
+            line = f"dhcp-host={host['mac']},{host['ip']},{host['hostname']}"
+        else:
+            line = f"dhcp-host={host['mac']},{host['ip']}"
+        lines.append(line)
+
+    lines.append("")
+    return "\n".join(lines)
+
+
+def write_config(content: str, output_file: str, dry_run: bool) -> None:
+    if dry_run:
+        print("=== DRY-RUN — contenu qui serait écrit dans", output_file, "===")
+        print(content)
+        return
+
+    out = Path(output_file)
+    out.parent.mkdir(parents=True, exist_ok=True)
+    out.write_text(content, encoding="utf-8")
+    log.info("Fichier écrit : %s", output_file)
+
+
+# ---------------------------------------------------------------------------
+# Point d'entrée
+# ---------------------------------------------------------------------------
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Synchronise phpIPAM → configuration dnsmasq (dhcp-host)."
+    )
+    parser.add_argument(
+        "--config", default="config.yaml",
+        help="Chemin vers le fichier de configuration YAML (défaut: config.yaml)"
+    )
+    parser.add_argument(
+        "--dry-run", action="store_true",
+        help="Affiche la config générée sans écrire le fichier"
+    )
+    parser.add_argument(
+        "--debug", action="store_true",
+        help="Active les logs de niveau DEBUG"
+    )
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+
+    if args.debug:
+        logging.getLogger().setLevel(logging.DEBUG)
+
+    cfg = load_config(args.config)
+    phpipam_cfg = cfg["phpipam"]
+    dnsmasq_cfg = cfg["dnsmasq"]
+
+    client = PhpIpamClient(
+        url=phpipam_cfg["url"],
+        app_id=phpipam_cfg["app_id"],
+        token=phpipam_cfg.get("token"),
+        verify_ssl=phpipam_cfg.get("verify_ssl", True),
+    )
+
+    log.info("Connexion à phpIPAM : %s (app_id=%s)", phpipam_cfg["url"], phpipam_cfg["app_id"])
+    hosts = collect_hosts(client)
+    log.info("%d hôte(s) avec adresse MAC trouvé(s).", len(hosts))
+
+    if not hosts:
+        log.warning("Aucun hôte trouvé — le fichier de sortie ne sera pas modifié.")
+        sys.exit(0)
+
+    header = dnsmasq_cfg.get(
+        "header_comment",
+        "Généré automatiquement par dnsmasq_sync.py — NE PAS MODIFIER"
+    )
+    content = generate_config(hosts, header)
+    write_config(content, dnsmasq_cfg["output_file"], dry_run=args.dry_run)
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,2 @@
+requests>=2.31.0
+PyYAML>=6.0