You've already forked ansible-netbox-inventory
Bryan Joshua Pedini
43cab4e7d8
hosts land in a single "netbox" group with hostvars under _meta; IP addresses use the DNS name as inventory hostname when set, virtual machines expose their primary IP as ansible_host when they have one.
272 lines
10 KiB
Python
Executable File
272 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-2.0-or-later
|
|
# Copyright (C) 2026 Bryan Joshua Pedini
|
|
"""
|
|
Ansible dynamic inventory script that sources hosts from NetBox.
|
|
|
|
Hosts are read from either the NetBox IP address list or the virtual machine
|
|
list, optionally narrowed down by a single filter option; everything is set in
|
|
the configuration file and can be overridden through environment variables.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from typing import Any
|
|
|
|
import requests
|
|
import yaml
|
|
|
|
DEFAULT_CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.yaml")
|
|
GROUP_NAME = "netbox"
|
|
|
|
SOURCE_IP_ADDRESSES = "ip-addresses"
|
|
SOURCE_VIRTUAL_MACHINES = "virtual-machines"
|
|
SOURCES = (SOURCE_IP_ADDRESSES, SOURCE_VIRTUAL_MACHINES)
|
|
|
|
|
|
class InventoryError(Exception):
|
|
"""Raised when the inventory cannot be built (configuration or API problems)."""
|
|
|
|
|
|
class NetBoxAPIError(InventoryError):
|
|
"""
|
|
Raised when the NetBox API cannot be reached or returns an error.
|
|
|
|
Attributes:
|
|
message: Human readable description of the failure.
|
|
status_code: HTTP status code of the response, if one was received.
|
|
"""
|
|
|
|
def __init__(self, message: str, status_code: int | None = None):
|
|
super().__init__(message)
|
|
self.message = message
|
|
self.status_code = status_code
|
|
|
|
|
|
def _extract_error_message(response: requests.Response) -> str:
|
|
"""Pull a readable error message out of a NetBox error response."""
|
|
try:
|
|
data = response.json()
|
|
if isinstance(data, dict) and "detail" in data:
|
|
return f"NetBox API error: {data['detail']}"
|
|
except ValueError:
|
|
pass
|
|
return f"NetBox API error: HTTP {response.status_code}"
|
|
|
|
|
|
class NetBoxClient:
|
|
"""Thin wrapper around the NetBox REST API."""
|
|
|
|
def __init__(self, url: str, token: str, verify_ssl: bool = True, timeout: int = 30):
|
|
self._url = url.rstrip("/")
|
|
self._timeout = timeout
|
|
self._session = requests.Session()
|
|
self._session.verify = verify_ssl
|
|
self._session.headers.update(
|
|
{
|
|
"Authorization": f"Token {token}",
|
|
"Accept": "application/json",
|
|
}
|
|
)
|
|
|
|
def _request(self, url: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
"""
|
|
Perform a GET request and return the decoded JSON body.
|
|
|
|
Raises:
|
|
NetBoxAPIError: On connection problems, non-200 responses or
|
|
unparsable bodies.
|
|
"""
|
|
try:
|
|
response = self._session.get(url, params=params, timeout=self._timeout)
|
|
except requests.exceptions.Timeout as exc:
|
|
raise NetBoxAPIError("NetBox API request timed out.") from exc
|
|
except requests.exceptions.SSLError as exc:
|
|
raise NetBoxAPIError("SSL error while connecting to the NetBox API.") from exc
|
|
except requests.exceptions.ConnectionError as exc:
|
|
raise NetBoxAPIError("Unable to connect to the NetBox API.") from exc
|
|
except requests.exceptions.RequestException as exc:
|
|
raise NetBoxAPIError(f"NetBox API request failed: {exc}") from exc
|
|
if response.status_code != 200:
|
|
raise NetBoxAPIError(
|
|
_extract_error_message(response), status_code=response.status_code
|
|
)
|
|
try:
|
|
return response.json()
|
|
except ValueError as exc:
|
|
raise NetBoxAPIError(
|
|
"NetBox API returned invalid JSON.", status_code=response.status_code
|
|
) from exc
|
|
|
|
def get_all(self, path: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]:
|
|
"""Return every result of a paginated list endpoint."""
|
|
results: list[dict[str, Any]] = []
|
|
url: str | None = f"{self._url}/api/{path.strip('/')}/"
|
|
while url:
|
|
data = self._request(url, params=params)
|
|
results.extend(data.get("results", []))
|
|
url = data.get("next")
|
|
params = None # the "next" URL already carries the query string
|
|
return results
|
|
|
|
|
|
def load_config(path: str | None = None) -> dict[str, Any]:
|
|
"""
|
|
Load the YAML configuration file and apply environment overrides.
|
|
|
|
Args:
|
|
path: Configuration file path; falls back to $NETBOX_INVENTORY_CONFIG,
|
|
then to config.yaml next to this script.
|
|
|
|
Returns:
|
|
A flat configuration dict with url, token, verify_ssl, timeout,
|
|
source, filter and filter_value keys.
|
|
|
|
Raises:
|
|
InventoryError: If the file is malformed or url/token are missing.
|
|
"""
|
|
path = path or os.environ.get("NETBOX_INVENTORY_CONFIG") or DEFAULT_CONFIG_FILE
|
|
config: dict[str, Any] = {}
|
|
if os.path.exists(path):
|
|
with open(path, encoding="utf-8") as handle:
|
|
loaded = yaml.safe_load(handle) or {}
|
|
if not isinstance(loaded, dict):
|
|
raise InventoryError(f"Invalid configuration file: {path}")
|
|
config = loaded
|
|
netbox = config.get("netbox") or {}
|
|
inventory = config.get("inventory") or {}
|
|
merged = {
|
|
"url": os.environ.get("NETBOX_URL", netbox.get("url") or ""),
|
|
"token": os.environ.get("NETBOX_TOKEN", netbox.get("token") or ""),
|
|
"verify_ssl": bool(netbox.get("verify_ssl", True)),
|
|
"timeout": netbox.get("timeout", 30),
|
|
"source": os.environ.get(
|
|
"NETBOX_INVENTORY_SOURCE", inventory.get("source") or SOURCE_IP_ADDRESSES
|
|
),
|
|
"filter": os.environ.get("NETBOX_INVENTORY_FILTER", inventory.get("filter") or ""),
|
|
"filter_value": os.environ.get(
|
|
"NETBOX_INVENTORY_FILTER_VALUE", inventory.get("filter_value") or ""
|
|
),
|
|
}
|
|
try:
|
|
merged["timeout"] = int(merged["timeout"])
|
|
except (TypeError, ValueError) as exc:
|
|
raise InventoryError("Configuration value netbox.timeout must be a number.") from exc
|
|
if not merged["url"] or not merged["token"]:
|
|
raise InventoryError(
|
|
"NetBox url and token are required, set them in the configuration file "
|
|
"or through the NETBOX_URL / NETBOX_TOKEN environment variables."
|
|
)
|
|
return merged
|
|
|
|
|
|
def strip_prefix(address: str) -> str:
|
|
"""Return a bare IP address without the CIDR prefix length."""
|
|
return address.split("/", 1)[0]
|
|
|
|
|
|
def ip_host_entry(ip: dict[str, Any]) -> tuple[str, dict[str, Any]]:
|
|
"""
|
|
Turn a NetBox IP address record into an inventory (name, hostvars) pair.
|
|
|
|
The DNS name is used as the inventory hostname when set, the bare address
|
|
otherwise; the bare address always ends up in ansible_host.
|
|
"""
|
|
address = strip_prefix(ip.get("address") or "")
|
|
name = ip.get("dns_name") or address
|
|
hostvars = {
|
|
"ansible_host": address,
|
|
"netbox_ip_address": ip.get("address") or "",
|
|
"netbox_dns_name": ip.get("dns_name") or "",
|
|
"netbox_tenant": (ip.get("tenant") or {}).get("name") or "",
|
|
"netbox_vrf": (ip.get("vrf") or {}).get("name") or "",
|
|
"netbox_description": ip.get("description") or "",
|
|
}
|
|
return name, hostvars
|
|
|
|
|
|
def vm_host_entry(vm: dict[str, Any]) -> tuple[str, dict[str, Any]]:
|
|
"""
|
|
Turn a NetBox virtual machine record into an inventory (name, hostvars) pair.
|
|
|
|
ansible_host is only set when the virtual machine has a primary IP, so that
|
|
Ansible falls back to resolving the name otherwise.
|
|
"""
|
|
name = vm.get("name") or ""
|
|
hostvars = {
|
|
"netbox_status": (vm.get("status") or {}).get("value") or "",
|
|
"netbox_cluster": (vm.get("cluster") or {}).get("name") or "",
|
|
"netbox_site": (vm.get("site") or {}).get("name") or "",
|
|
"netbox_tenant": (vm.get("tenant") or {}).get("name") or "",
|
|
}
|
|
primary = vm.get("primary_ip") or {}
|
|
if primary.get("address"):
|
|
hostvars["ansible_host"] = strip_prefix(primary["address"])
|
|
return name, hostvars
|
|
|
|
|
|
def fetch_entries(client: NetBoxClient, source: str) -> list[tuple[str, dict[str, Any]]]:
|
|
"""Fetch all hosts of the configured source from NetBox."""
|
|
if source == SOURCE_IP_ADDRESSES:
|
|
records = client.get_all("ipam/ip-addresses")
|
|
return [ip_host_entry(record) for record in records if record.get("address")]
|
|
records = client.get_all("virtualization/virtual-machines", {"exclude": "config_context"})
|
|
return [vm_host_entry(record) for record in records if record.get("name")]
|
|
|
|
|
|
def build_inventory(config: dict[str, Any], client: NetBoxClient) -> dict[str, Any]:
|
|
"""
|
|
Build the full --list inventory structure for Ansible.
|
|
|
|
Raises:
|
|
InventoryError: On an unknown source or NetBox API failures.
|
|
"""
|
|
source = config["source"]
|
|
if source not in SOURCES:
|
|
raise InventoryError(
|
|
f'Unknown source "{source}", valid sources: {", ".join(SOURCES)}.'
|
|
)
|
|
entries = fetch_entries(client, source)
|
|
hostvars: dict[str, dict[str, Any]] = {}
|
|
for name, host in entries:
|
|
if name in hostvars:
|
|
# duplicate inventory names keep both hosts by falling back to the
|
|
# bare address as the name; identical duplicates are dropped
|
|
name = host.get("ansible_host") or name
|
|
if name in hostvars:
|
|
continue
|
|
hostvars[name] = host
|
|
return {
|
|
GROUP_NAME: {"hosts": sorted(hostvars)},
|
|
"_meta": {"hostvars": hostvars},
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
"""Entry point for the Ansible dynamic inventory protocol."""
|
|
parser = argparse.ArgumentParser(description="Ansible dynamic inventory backed by NetBox.")
|
|
group = parser.add_mutually_exclusive_group(required=True)
|
|
group.add_argument("--list", action="store_true", help="output the full inventory as JSON")
|
|
group.add_argument("--host", metavar="HOSTNAME", help="output variables for a single host")
|
|
args = parser.parse_args()
|
|
if args.host:
|
|
# all host variables are returned through _meta in --list
|
|
print(json.dumps({}))
|
|
return 0
|
|
try:
|
|
config = load_config()
|
|
client = NetBoxClient(
|
|
config["url"], config["token"], config["verify_ssl"], config["timeout"]
|
|
)
|
|
print(json.dumps(build_inventory(config, client), indent=2, sort_keys=True))
|
|
except InventoryError as exc:
|
|
print(f"netbox-inventory: {exc}", file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|