|
|
@@ -73,27 +73,44 @@ class PhpIpamClient:
|
|
|
# phpIPAM attend le token dans l'en-tête "token"
|
|
|
self.session.headers.update({"token": token})
|
|
|
|
|
|
- def _get(self, endpoint: str) -> list | dict:
|
|
|
+ def _get(self, endpoint: str, fatal: bool = False) -> list | dict:
|
|
|
+ """
|
|
|
+ Effectue un GET sur l'API phpIPAM.
|
|
|
+ Si fatal=True, une erreur réseau/HTTP arrête le script (pour les appels critiques).
|
|
|
+ Sinon, l'erreur est loggée en warning et une liste vide est retournée.
|
|
|
+ """
|
|
|
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é, "
|
|
|
+ msg = (
|
|
|
+ "Erreur SSL sur %s. Si vous utilisez un certificat auto-signé, "
|
|
|
"passez verify_ssl: false dans la config."
|
|
|
)
|
|
|
- sys.exit(1)
|
|
|
+ if fatal:
|
|
|
+ log.error(msg, url)
|
|
|
+ sys.exit(1)
|
|
|
+ log.warning(msg, url)
|
|
|
+ return []
|
|
|
except requests.exceptions.ConnectionError as exc:
|
|
|
- log.error("Impossible de joindre phpIPAM (%s) : %s", url, exc)
|
|
|
- sys.exit(1)
|
|
|
+ msg = "Impossible de joindre phpIPAM (%s) : %s"
|
|
|
+ if fatal:
|
|
|
+ log.error(msg, url, exc)
|
|
|
+ sys.exit(1)
|
|
|
+ log.warning(msg, url, exc)
|
|
|
+ return []
|
|
|
except requests.exceptions.HTTPError as exc:
|
|
|
- log.error("Erreur HTTP %s sur %s : %s", resp.status_code, url, exc)
|
|
|
- sys.exit(1)
|
|
|
+ msg = "Erreur HTTP %s sur %s : %s"
|
|
|
+ if fatal:
|
|
|
+ log.error(msg, resp.status_code, url, exc)
|
|
|
+ sys.exit(1)
|
|
|
+ log.warning(msg, resp.status_code, url, exc)
|
|
|
+ return []
|
|
|
|
|
|
data = resp.json()
|
|
|
if not data.get("success"):
|
|
|
- # 404 "No addresses found" n'est pas une erreur fatale
|
|
|
+ # 404 signifie simplement "aucun résultat"
|
|
|
if data.get("code") == 404:
|
|
|
return []
|
|
|
log.warning("Réponse phpIPAM non-success pour %s : %s", endpoint, data.get("message"))
|
|
|
@@ -101,7 +118,8 @@ class PhpIpamClient:
|
|
|
return data.get("data") or []
|
|
|
|
|
|
def get_sections(self) -> list[dict]:
|
|
|
- return self._get("sections/")
|
|
|
+ # Les sections sont indispensables : échec fatal
|
|
|
+ return self._get("sections/", fatal=True)
|
|
|
|
|
|
def get_subnets_for_section(self, section_id: int) -> list[dict]:
|
|
|
return self._get(f"sections/{section_id}/subnets/")
|