You've already forked ansible-netbox-inventory
Bryan Joshua Pedini
c1f0b4f724
inventory.ip_version (or NETBOX_INVENTORY_IP_VERSION): both (default), v4 or v6; the ip-addresses source filters by family server-side, the virtual-machines source picks primary_ip4/primary_ip6 for ansible_host.
422 lines
16 KiB
Python
Executable File
422 lines
16 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)
|
|
|
|
IP_VERSION_BOTH = "both"
|
|
IP_VERSIONS = (IP_VERSION_BOTH, "v4", "v6")
|
|
# ip_version -> family query parameter (ip-addresses source)
|
|
IP_FAMILIES = {"v4": "4", "v6": "6"}
|
|
# ip_version -> primary IP field (virtual-machines source)
|
|
PRIMARY_IP_FIELDS = {IP_VERSION_BOTH: "primary_ip", "v4": "primary_ip4", "v6": "primary_ip6"}
|
|
|
|
# filter option -> sources it applies to
|
|
FILTERS = {
|
|
"tenant": SOURCES,
|
|
"domain": SOURCES,
|
|
"named": (SOURCE_IP_ADDRESSES,),
|
|
"vrf": (SOURCE_IP_ADDRESSES,),
|
|
"cluster": (SOURCE_VIRTUAL_MACHINES,),
|
|
"site": (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}"
|
|
|
|
|
|
def auth_header(token: str) -> str:
|
|
"""
|
|
Build the Authorization header value for a NetBox API token.
|
|
|
|
Version 2 tokens (NetBox 4.5+, "nbt_<key>.<secret>") use the Bearer
|
|
scheme, legacy tokens the Token scheme.
|
|
"""
|
|
if token.startswith("nbt_"):
|
|
return f"Bearer {token}"
|
|
return f"Token {token}"
|
|
|
|
|
|
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": auth_header(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 resolve_id(self, path: str, name: str, label: str) -> int:
|
|
"""
|
|
Look up an object by name and return its numeric id.
|
|
|
|
Raises:
|
|
NetBoxAPIError: If no object with that name exists.
|
|
"""
|
|
matches = self.get_all(path, params={"name": name})
|
|
if not matches:
|
|
raise NetBoxAPIError(f'No {label} named "{name}" found in NetBox.')
|
|
return matches[0]["id"]
|
|
|
|
|
|
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 ""
|
|
),
|
|
"ip_version": os.environ.get(
|
|
"NETBOX_INVENTORY_IP_VERSION", inventory.get("ip_version") or IP_VERSION_BOTH
|
|
)
|
|
.strip()
|
|
.lower(),
|
|
}
|
|
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 parse_bool(value: str) -> bool:
|
|
"""
|
|
Parse a yes/no filter value, an empty string counts as yes.
|
|
|
|
Raises:
|
|
InventoryError: If the value is not recognisable as a boolean.
|
|
"""
|
|
lowered = value.strip().lower()
|
|
if lowered in ("", "1", "true", "yes", "on"):
|
|
return True
|
|
if lowered in ("0", "false", "no", "off"):
|
|
return False
|
|
raise InventoryError(f'Filter value "{value}" is not a valid boolean.')
|
|
|
|
|
|
def domain_matches(hostname: str, domain: str) -> bool:
|
|
"""
|
|
Check whether a hostname belongs to a domain.
|
|
|
|
The hostname must end with the domain on a label boundary, so
|
|
"web.example.com" matches "example.com" but "notexample.com" does not.
|
|
"""
|
|
if not hostname:
|
|
return False
|
|
domain = domain.lstrip(".").rstrip(".").lower()
|
|
hostname = hostname.rstrip(".").lower()
|
|
return hostname == domain or hostname.endswith("." + domain)
|
|
|
|
|
|
def validate_options(
|
|
source: str, filter_name: str, filter_value: str, ip_version: str = IP_VERSION_BOTH
|
|
) -> None:
|
|
"""
|
|
Validate the source/filter/ip_version combination before touching the API.
|
|
|
|
Raises:
|
|
InventoryError: On unknown sources, filters or IP versions, a filter
|
|
that does not apply to the source, or a missing filter value.
|
|
"""
|
|
if source not in SOURCES:
|
|
raise InventoryError(f'Unknown source "{source}", valid sources: {", ".join(SOURCES)}.')
|
|
if ip_version not in IP_VERSIONS:
|
|
raise InventoryError(
|
|
f'Unknown ip_version "{ip_version}", valid values: {", ".join(IP_VERSIONS)}.'
|
|
)
|
|
if not filter_name:
|
|
return
|
|
if filter_name not in FILTERS:
|
|
raise InventoryError(
|
|
f'Unknown filter "{filter_name}", valid filters: {", ".join(FILTERS)}.'
|
|
)
|
|
if source not in FILTERS[filter_name]:
|
|
raise InventoryError(f'Filter "{filter_name}" does not apply to source "{source}".')
|
|
if filter_name != "named" and not filter_value:
|
|
raise InventoryError(f'Filter "{filter_name}" requires a value.')
|
|
|
|
|
|
def server_side_params(client: NetBoxClient, filter_name: str, filter_value: str) -> dict[str, Any]:
|
|
"""Translate name-based filters into NetBox API query parameters."""
|
|
if filter_name == "tenant":
|
|
return {"tenant_id": client.resolve_id("tenancy/tenants", filter_value, "tenant")}
|
|
if filter_name == "vrf":
|
|
return {"vrf_id": client.resolve_id("ipam/vrfs", filter_value, "VRF")}
|
|
if filter_name == "cluster":
|
|
return {"cluster_id": client.resolve_id("virtualization/clusters", filter_value, "cluster")}
|
|
if filter_name == "site":
|
|
return {"site_id": client.resolve_id("dcim/sites", filter_value, "site")}
|
|
return {}
|
|
|
|
|
|
def apply_client_side_filters(
|
|
entries: list[tuple[str, dict[str, Any]]], filter_name: str, filter_value: str
|
|
) -> list[tuple[str, dict[str, Any]]]:
|
|
"""
|
|
Apply the filters that cannot be expressed as API query parameters.
|
|
|
|
"domain" keeps hostnames belonging to the given domain, "named" keeps only
|
|
IP addresses with a DNS name and deduplicates them by hostname.
|
|
"""
|
|
if filter_name == "domain":
|
|
return [(name, host) for name, host in entries if domain_matches(name, filter_value)]
|
|
if filter_name == "named" and parse_bool(filter_value):
|
|
seen: set[str] = set()
|
|
named = []
|
|
for name, host in entries:
|
|
if not host.get("netbox_dns_name") or name in seen:
|
|
continue
|
|
seen.add(name)
|
|
named.append((name, host))
|
|
return named
|
|
return entries
|
|
|
|
|
|
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], ip_version: str = IP_VERSION_BOTH
|
|
) -> 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 of the
|
|
requested version, so that Ansible falls back to resolving the name
|
|
otherwise. With ip_version "both" NetBox picks the preferred family.
|
|
"""
|
|
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_FIELDS[ip_version]) or {}
|
|
if primary.get("address"):
|
|
hostvars["ansible_host"] = strip_prefix(primary["address"])
|
|
return name, hostvars
|
|
|
|
|
|
def fetch_entries(
|
|
client: NetBoxClient,
|
|
source: str,
|
|
params: dict[str, Any] | None = None,
|
|
ip_version: str = IP_VERSION_BOTH,
|
|
) -> list[tuple[str, dict[str, Any]]]:
|
|
"""Fetch all hosts of the configured source from NetBox."""
|
|
if source == SOURCE_IP_ADDRESSES:
|
|
if ip_version != IP_VERSION_BOTH:
|
|
params = {**(params or {}), "family": IP_FAMILIES[ip_version]}
|
|
records = client.get_all("ipam/ip-addresses", params)
|
|
return [ip_host_entry(record) for record in records if record.get("address")]
|
|
params = {"exclude": "config_context", **(params or {})}
|
|
records = client.get_all("virtualization/virtual-machines", params)
|
|
return [vm_host_entry(record, ip_version) 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 invalid source/filter options or NetBox API failures.
|
|
"""
|
|
source = config["source"]
|
|
filter_name = config["filter"]
|
|
filter_value = config["filter_value"]
|
|
ip_version = config.get("ip_version") or IP_VERSION_BOTH
|
|
validate_options(source, filter_name, filter_value, ip_version)
|
|
params = server_side_params(client, filter_name, filter_value) if filter_name else {}
|
|
entries = fetch_entries(client, source, params, ip_version)
|
|
entries = apply_client_side_filters(entries, filter_name, filter_value)
|
|
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())
|