DnsMasqSync/dnsmasq_sync.py

#!/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()