#!/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 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: load_config() print(json.dumps({GROUP_NAME: {"hosts": []}, "_meta": {"hostvars": {}}}, indent=2)) except InventoryError as exc: print(f"netbox-inventory: {exc}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(main())