#!/usr/bin/env python3 """Tests de la couche pure de kea-sync-from-db.py (génération de config). Lancement, depuis kea-sync-from-db/ : python3.12 test_kea_sync.py Aucune dépendance externe : mysql.connector est remplacé par un stub, aucune connexion MySQL ni fichier de config n'est nécessaire, et rien n'est écrit sur disque (l'écriture des secrets vit dans write_config, pas dans build_config). Couvre la migration Kea 3.1.8+ (ARM §18.8) : suppression de kea-ctrl-agent, canal HTTP porté par kea-dhcp4, et refus par Kea >= 3.2 de 'user'/'password' en clair. """ import configparser import importlib.util import json import os import sys import types # mysql-connector n'est pas nécessaire pour tester la génération de config : # on neutralise l'import de tête de module. _mysql = types.ModuleType("mysql") _mysql.connector = types.ModuleType("mysql.connector") sys.modules["mysql"] = _mysql sys.modules["mysql.connector"] = _mysql.connector _here = os.path.dirname(os.path.abspath(__file__)) _spec = importlib.util.spec_from_file_location("ksfd", os.path.join(_here, "kea-sync-from-db.py")) ksfd = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(ksfd) FAILURES: list[str] = [] USER_FILE = "/usr/local/etc/kea/kea-api-user" PASS_FILE = "/usr/local/etc/kea/kea-api-password" def check(label: str, got, expected) -> None: if got == expected: print(f" ok {label}") else: print(f" FAIL {label}") print(f" attendu : {expected!r}") print(f" obtenu : {got!r}") FAILURES.append(label) def set_conf(**kea) -> None: """Remplace la config globale du module par une config de test.""" cp = configparser.ConfigParser() cp["kea"] = {k: v for k, v in kea.items()} ksfd.conf = cp # ── http_control_socket ─────────────────────────────────────────────────────── print("http_control_socket") check( "localhost est résolu en 127.0.0.1 (Kea attend une IP)", ksfd.http_control_socket("http://localhost:8000/"), {"socket-type": "http", "socket-address": "127.0.0.1", "socket-port": 8000}, ) check( "adresse et port explicites sont repris tels quels", ksfd.http_control_socket("http://192.168.1.50:8080/"), {"socket-type": "http", "socket-address": "192.168.1.50", "socket-port": 8080}, ) check( "URL sans port -> port 8000 (défaut Kea)", ksfd.http_control_socket("http://127.0.0.1/"), {"socket-type": "http", "socket-address": "127.0.0.1", "socket-port": 8000}, ) check( "auth -> user-file ET password-file (Kea >= 3.2 refuse les deux en clair)", ksfd.http_control_socket("http://127.0.0.1:8000/", USER_FILE, PASS_FILE), { "socket-type": "http", "socket-address": "127.0.0.1", "socket-port": 8000, "authentication": { "type": "basic", "realm": "kea-dhcp4-server", "directory": "/usr/local/etc/kea", "clients": [{ "user-file": "kea-api-user", "password-file": "kea-api-password", }], }, }, ) check( "pas d'auth si un seul des deux fichiers est connu", "authentication" in (ksfd.http_control_socket("http://127.0.0.1:8000/", USER_FILE, "") or {}), False, ) check( "https -> socket-type https", ksfd.http_control_socket("https://127.0.0.1:8443/")["socket-type"], "https", ) check("URL vide -> None", ksfd.http_control_socket(""), None) check("schéma inconnu -> None", ksfd.http_control_socket("ftp://127.0.0.1/"), None) # ── build_config ────────────────────────────────────────────────────────────── print("build_config") set_conf(api_url="http://localhost:8000/", api_user="", api_password="") cfg = ksfd.build_config([], [], {}, "/tmp/kea-dhcp4-ctrl.sock", None)["Dhcp4"] check( "control-socket (singulier, format pré-3.1.8) n'est plus émis", "control-socket" in cfg, False, ) check( "control-sockets contient la socket unix puis la socket http", cfg["control-sockets"], [ {"socket-type": "unix", "socket-name": "/tmp/kea-dhcp4-ctrl.sock"}, {"socket-type": "http", "socket-address": "127.0.0.1", "socket-port": 8000}, ], ) set_conf(api_url="http://localhost:8000/", api_user="", api_password="") check( "sans control_socket configurée : seule la socket http est émise", ksfd.build_config([], [], {}, "", None)["Dhcp4"]["control-sockets"], [{"socket-type": "http", "socket-address": "127.0.0.1", "socket-port": 8000}], ) set_conf(api_url="", api_user="", api_password="") check( "api_url vide : la socket unix reste, aucune socket http", ksfd.build_config([], [], {}, "/tmp/s.sock", None)["Dhcp4"]["control-sockets"], [{"socket-type": "unix", "socket-name": "/tmp/s.sock"}], ) # Config complète avec authentification : c'est le cas du lab. set_conf( api_url="http://192.168.1.2:8000/", api_user="kea-api", api_password="mot-de-passe-tres-secret", ) cfg_auth = ksfd.build_config([], [], {}, "/var/run/kea/kea-dhcp4-ctrl.sock", None) dumped = json.dumps(cfg_auth) check( "auth activée : http sur l'IP de api_url, secrets par fichiers", cfg_auth["Dhcp4"]["control-sockets"][1], { "socket-type": "http", "socket-address": "192.168.1.2", "socket-port": 8000, "authentication": { "type": "basic", "realm": "kea-dhcp4-server", "directory": "/usr/local/etc/kea", "clients": [{ "user-file": "kea-api-user", "password-file": "kea-api-password", }], }, }, ) # Le pool short-lease sélectionne sa classe par liste : 'client-class' au singulier # est déprécié depuis Kea 3.2 (DHCPSRV_CLIENT_CLASS_DEPRECATED). set_conf(api_url="http://127.0.0.1:8000/", api_user="", api_password="") pools = ksfd.build_config([{ "kea_subnet_id": 7, "cidr": "192.168.9.0/24", "has_dynamic_pool": 1, "range_start": "192.168.9.100", "range_end": "192.168.9.150", "short_lease_pool_start": "192.168.9.200", "short_lease_pool_end": "192.168.9.210", }], [], {}, "", None)["Dhcp4"]["subnet4"][0]["pools"] check( "pool dynamique inchangé, pool short-lease en client-classes (liste)", pools, [ {"pool": "192.168.9.100 - 192.168.9.150"}, {"pool": "192.168.9.200 - 192.168.9.210", "client-classes": ["short-lease"]}, ], ) check("le mot de passe ne fuite jamais dans le JSON", "mot-de-passe-tres-secret" in dumped, False) check("l'identifiant ne fuite jamais dans le JSON", '"kea-api"' in dumped, False) check("aucune clé 'password' en clair", '"password"' in dumped, False) check("aucune clé 'user' en clair", '"user"' in dumped, False) # ── Résultat ────────────────────────────────────────────────────────────────── print() if FAILURES: print(f"{len(FAILURES)} test(s) en échec : " + ", ".join(FAILURES)) sys.exit(1) print("Tous les tests passent.")