Automatiser NetBox par l'API

Un utilisateur d'automatisation aux droits bornés, pynetbox ou curl pour lire et écrire, un import CSV qui survit aux erreurs, des webhooks sur les changements d'équipements, et un script maison lancé par l'API.

intermédiaire~45 min de manipulation
#netbox#api#pynetbox#python#webhooks#automation

Pas encore validée de bout en bout — sois le premier.Signaler un problème

Brouillon — pas encore exécuté de bout en bout. Cette page est écrite mais son auteur ne l'a pas encore déroulée sur une vraie machine. Des commandes peuvent être fausses : lis avant de lancer, et dis-nous ce qui casse.

En un coup d'œilDe ton script à ton récepteur
Serveur NetBox
HTTPS · tokenwriteseventPOST
Ton scriptpynetbox · curl
API RESThttps:///api
NetBoxpermissions · validation
PostgreSQLobjects · change log
Worker RQnetbox-rq
Récepteur de webhooks

Ton script parle à l'API REST avec le jeton de l'utilisateur d'automatisation. NetBox écrit les objets, met un événement en file, et le worker RQ du serveur livre le webhook à ton récepteur.

Tout ce que tu as cliqué dans les pages précédentes est un appel HTTP, et NetBox ne vaut le coup que quand d'autres systèmes le lisent et l'écrivent. Cette page tourne sur ta machine, pas sur le serveur : un utilisateur fait pour les scripts, pynetbox ou curl pour lire et écrire, un import CSV qui survit à une mauvaise ligne, des webhooks quand un équipement change, et un script qui tourne dans NetBox à la demande.

Avant de commencer

$python3 -m venv ~/.venvs/netbox
$source ~/.venvs/netbox/bin/activate
$pip install pynetbox
Vérification
$pip show pynetbox | head -1
Retour attendu
Name: pynetbox

Un utilisateur d'automatisation et son jeton

$api=https:///api
$auth="Authorization: Token "
$curl -sk -H "$auth" -H 'Content-Type: application/json' "$api/users/users/" \
$ -d "{\"username\": \"\", \"password\": \"$(openssl rand -base64 24)\"}"

Note l'id dans la réponse ; il va dans la liste users des permissions. Trois permissions :

$curl -sk -H "$auth" -H 'Content-Type: application/json' "$api/users/permissions/" -d '{
$ "name": "automation: view", "actions": ["view"], "users": [ID],
$ "object_types": ["dcim.site", "dcim.manufacturer", "dcim.devicerole", "dcim.devicetype",
$ "dcim.device", "dcim.interface", "ipam.ipaddress"]
$}'

Maintenant le jeton, avec une expiration à 90 jours :

$key=$(openssl rand -hex 20)
$exp=$(python3 -c 'import datetime as d; print((d.datetime.now(d.timezone.utc) + d.timedelta(days=90)).strftime("%Y-%m-%dT%H:%M:%SZ"))')
$curl -sk -H "$auth" -H 'Content-Type: application/json' "$api/users/tokens/" \
$ -d "{\"user\": ID, \"key\": \"$key\", \"expires\": \"$exp\", \"write_enabled\": true, \"description\": \"automation\"}"
$echo "$key"

Mets la clé affichée dans tes valeurs comme AUTOMATION_TOKEN. À partir d'ici, les appels sur les données l'utilisent.

Vérification
$curl -sk -o /dev/null -w '%{http_code}' -H 'Authorization: Token ' https:///api/dcim/sites/
Retour attendu
200
Vérification
$curl -sk -o /dev/null -w '%{http_code}' -H 'Authorization: Token ' https:///api/users/users/
Retour attendu
403

Lire et écrire

sites.py
import re
import pynetbox
nb = pynetbox.api("https://", token="")
for site in nb.dcim.sites.all():
print(site.id, site.name, site.status)
slug = re.sub(r"[^a-z0-9]+", "-", "".lower()).strip("-")
site = nb.dcim.sites.get(slug=slug)
if site is None:
site = nb.dcim.sites.create(name="", slug=slug, status="active")
print("created", site.name)
elif site.update({"description": "managed by "}):
print("updated", site.name)
else:
print("unchanged", site.name)
$python sites.py && python sites.py
Vérification
$python -c 'import pynetbox; nb = pynetbox.api("https://", token=""); print(nb.dcim.sites.get(name="").status.value)'
Retour attendu
active

Un équipement de bout en bout

device.py
import re
import pynetbox
nb = pynetbox.api("https://", token="")
slug = lambda s: re.sub(r"[^a-z0-9]+", "-", s.lower()).strip("-")
def ensure(endpoint, lookup, **fields):
"""Cherche par champs uniques, sinon crée. Relançable sans risque."""
obj = endpoint.get(**lookup)
return obj if obj is not None else endpoint.create(**fields)
site = ensure(nb.dcim.sites, {"slug": slug("")},
name="", slug=slug(""), status="active")
mfr = ensure(nb.dcim.manufacturers, {"slug": "generic"}, name="Generic", slug="generic")
dtype = ensure(nb.dcim.device_types, {"slug": "generic-48p"},
manufacturer=mfr.id, model="Generic 48-port switch", slug="generic-48p")
role = ensure(nb.dcim.device_roles, {"slug": "access-switch"},
name="Access switch", slug="access-switch", color="2196f3")
dev = ensure(nb.dcim.devices, {"name": "sw-01", "site_id": site.id},
name="sw-01", site=site.id, role=role.id, device_type=dtype.id, status="active")
iface = ensure(nb.dcim.interfaces, {"device_id": dev.id, "name": "Ethernet1"},
device=dev.id, name="Ethernet1", type="1000base-t")
ip = ensure(nb.ipam.ip_addresses, {"address": "192.0.2.11/24"},
address="192.0.2.11/24", status="active",
assigned_object_type="dcim.interface", assigned_object_id=iface.id)
if dev.primary_ip4 is None or dev.primary_ip4.id != ip.id:
dev.update({"primary_ip4": ip.id})
print("ok:", dev.name, "at", site.name, "primary", ip.address)
$python device.py
Vérification
$python -c 'import pynetbox; nb = pynetbox.api("https://", token=""); print(nb.dcim.devices.get(name="sw-01").primary_ip4.address)'
Retour attendu
192.0.2.11/24
Si une création répond 400 ou 403

Un 400 vient avec un corps JSON qui nomme le champ : lis-le, il est précis ("slug": ["This field is required."], "status": ["\"Active\" is not a valid choice."]). Un 403 sur une création que tu attendais valide, c'est la contrainte de la permission : l'équipement a été enregistré, vérifié contre site__name, et annulé. Compare le nom du site dans la permission avec , orthographe et casse.

Import en masse depuis un CSV

devices.csv
name,site,role,device_type,status
sw-02,paris-dc1,access-switch,generic-48p,planned
sw-03,paris-dc1,access-switch,generic-48p,planned
sw-04,paris-dc1,no-such-role,generic-48p,planned
import_devices.py
import csv
import pynetbox
nb = pynetbox.api("https://", token="")
created = skipped = failed = 0
with open("devices.csv", newline="") as f:
for row in csv.DictReader(f):
if nb.dcim.devices.get(name=row["name"], site=row["site"]) is not None:
skipped += 1
continue
try:
nb.dcim.devices.create(
name=row["name"], status=row["status"],
site={"slug": row["site"]}, role={"slug": row["role"]},
device_type={"slug": row["device_type"]},
)
created += 1
except pynetbox.RequestError as e:
failed += 1
print(f"FAILED {row['name']}: {e.error}")
print(f"created {created}, skipped {skipped}, failed {failed}")
$python import_devices.py
Vérification
$python import_devices.py | tail -1
Retour attendu
created 0, skipped 2, failed 1

Règles d'événements et webhooks

$api=https:///api
$auth="Authorization: Token "
$curl -sk -H "$auth" -H 'Content-Type: application/json' "$api/extras/webhooks/" -d '{
$ "name": "device-changes", "payload_url": "",
$ "http_method": "POST", "http_content_type": "application/json"
$}'

Avec l'id du webhook comme action_object_id :

$curl -sk -H "$auth" -H 'Content-Type: application/json' "$api/extras/event-rules/" -d '{
$ "name": "device created or updated",
$ "object_types": ["dcim.device"], "event_types": ["object_created", "object_updated"],
$ "action_type": "webhook", "action_object_type": "extras.webhook", "action_object_id": ID,
$ "enabled": true
$}'

Un récepteur pour voir les charges utiles. python -m http.server ne convient pas, il n'affiche jamais le corps des requêtes ; dix lignes de Flask le font :

receiver.py
from urllib.parse import urlparse
from flask import Flask, request
target = urlparse("")
app = Flask(__name__)
@app.post(target.path or "/")
def hook():
body = request.get_json(force=True)
print(body.get("event"), body.get("model"), body.get("data", {}).get("name"), flush=True)
return "", 204
app.run(host="0.0.0.0", port=target.port or 80)
$pip install flask && python receiver.py

Depuis le serveur NetBox, prouve qu'il joint le récepteur, puis modifie un équipement et regarde le terminal :

Vérification
$curl -s -o /dev/null -w '%{http_code}' -X POST -H 'Content-Type: application/json' -d '{"event": "test", "model": "none", "data": {}}' 
Retour attendu
204
$python -c 'import pynetbox; nb = pynetbox.api("https://", token=""); nb.dcim.devices.get(name="sw-01").update({"description": "hello webhook"})'
Vérification
$curl -sk -H 'Authorization: Token ' "https:///api/extras/event-rules/?name=device%20created%20or%20updated" | python3 -c 'import sys,json; print(json.load(sys.stdin)["count"])'
Retour attendu
1
Si rien n'arrive

Dans l'ordre : systemctl is-active netbox-rq sur le serveur ; la vérification curl ci-dessus depuis le serveur, pas depuis ta machine (un pare-feu entre les deux est la cause habituelle) ; la règle est activée et liste dcim.device ; Opérations → Tâches de fond montre la tâche et son erreur. Avec un récepteur en https:// et une CA privée, mets ssl_verification à false sur le webhook pour le test, puis corrige la CA.

Scripts personnalisés

create_devices.py
from dcim.choices import DeviceStatusChoices
from dcim.models import Device, DeviceRole, DeviceType, Site
from extras.scripts import IntegerVar, ObjectVar, Script, StringVar
class CreateDevices(Script):
class Meta:
name = "Create N devices"
description = "Numbered devices at one site, planned status"
commit_default = False
site = ObjectVar(model=Site)
role = ObjectVar(model=DeviceRole)
device_type = ObjectVar(model=DeviceType)
prefix = StringVar(default="sw-", description="Name prefix")
count = IntegerVar(default=2, min_value=1, max_value=50)
def run(self, data, commit):
for i in range(1, data["count"] + 1):
name = f"{data['prefix']}{i:02d}"
device, created = Device.objects.get_or_create(
name=name, site=data["site"],
defaults={"role": data["role"], "device_type": data["device_type"],
"status": DeviceStatusChoices.STATUS_PLANNED},
)
self.log_success(f"created {name}" if created else f"exists {name}")
return f"{data['count']} devices checked at {data['site']}"

Téléverse-le dans l'interface : Personnalisation → Scripts → Ajouter, choisis le fichier. Lance-le une fois depuis le formulaire avec Commit décoché pour voir le journal, puis depuis l'API :

Commande sensible — exécute un script distant. Vérifie avant d'exécuter.
$api=https:///api
$auth="Authorization: Token "
$curl -sk -H "$auth" "$api/extras/scripts/" | python3 -m json.tool | grep -E '"(id|name)"'
$curl -sk -H "$auth" -H 'Content-Type: application/json' "$api/extras/scripts/ID/" -d '{
$ "data": {"site": SITE_ID, "role": ROLE_ID, "device_type": TYPE_ID, "prefix": "sw-", "count": 3},
$ "commit": true
$}'
Vérification
$curl -sk -H 'Authorization: Token ' https:///api/extras/scripts/ | grep -q CreateDevices && echo found
Retour attendu
found

Terminé

NetBox a maintenant un utilisateur fait pour les scripts, aux droits bornés à , tu sais créer un équipement avec son interface et son IP en un passage relançable, importer un CSV sans craindre une mauvaise ligne, d'autres systèmes apprennent les changements d'équipements, et un script dans NetBox fait le travail répétitif depuis un formulaire ou depuis l'API. La série s'arrête ici ; les suivantes mettent tout ça à profit dans un lab NetDevOps.

Tout a fonctionné ?

Si tu as suivi cette page jusqu'au bout sur une vraie machine, dis-le. Ta validation est datée et enregistre ta stack : le prochain lecteur sur le même chemin sait que ça marche toujours.

Cette copie est en lecture seule. Pour dire que ça marche, ou que ça ne marche pas, ouvre une issue

Seuls tes choix de stack sont enregistrés, jamais tes valeurs. Le pseudo reste sur ce navigateur.