source/ansible-netbox-inventory
1
0
Fork 0
Code Issues Pull Requests Releases Activity

feat: ip_version option to restrict inventories to ipv4 or ipv6

Browse Source 
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.
This commit is contained in:
Bryan Joshua Pedini
2026-08-25 14:53:24 +02:00
parent 9dbcdfad66
commit c1f0b4f724
3 changed files with 87 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
config.yaml.example
View File

@@ -13,3 +13,5 @@ inventory:
# optional filter, see README.md for the full list
filter: ""
filter_value: ""
# restrict to one IP family: both | v4 | v6
ip_version: "both"

51
netbox_inventory.py
View File

@@ -25,6 +25,13 @@ 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,
@@ -181,6 +188,11 @@ def load_config(path: str | None = None) -> dict[str, Any]:
"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"])
@@ -223,16 +235,22 @@ def domain_matches(hostname: str, domain: str) -> bool:
return hostname == domain or hostname.endswith("." + domain)
def validate_options(source: str, filter_name: str, filter_value: str) -> None:
def validate_options(
source: str, filter_name: str, filter_value: str, ip_version: str = IP_VERSION_BOTH
) -> None:
"""
Validate the source/filter combination before touching the API.
Validate the source/filter/ip_version combination before touching the API.
Raises:
InventoryError: On unknown sources or filters, a filter that does not
apply to the source, or a missing filter value.
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:
@@ -306,12 +324,15 @@ def ip_host_entry(ip: dict[str, Any]) -> tuple[str, dict[str, Any]]:
return name, hostvars
def vm_host_entry(vm: dict[str, Any]) -> tuple[str, dict[str, Any]]:
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, so that
Ansible falls back to resolving the name otherwise.
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 = {
@@ -320,22 +341,27 @@ def vm_host_entry(vm: dict[str, Any]) -> tuple[str, dict[str, Any]]:
"netbox_site": (vm.get("site") or {}).get("name") or "",
"netbox_tenant": (vm.get("tenant") or {}).get("name") or "",
}
primary = vm.get("primary_ip") 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
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) for record in records if record.get("name")]
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]:
@@ -348,9 +374,10 @@ def build_inventory(config: dict[str, Any], client: NetBoxClient) -> dict[str, A
source = config["source"]
filter_name = config["filter"]
filter_value = config["filter_value"]
validate_options(source, filter_name, 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)
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:

46
tests/test_inventory.py
View File

@@ -91,6 +91,14 @@ class TestValidateOptions:
def test_named_without_value_is_fine(self):
inv.validate_options("ip-addresses", "named", "")
def test_valid_ip_versions_pass(self):
for ip_version in ("both", "v4", "v6"):
inv.validate_options("ip-addresses", "", "", ip_version)
def test_unknown_ip_version(self):
with pytest.raises(inv.InventoryError, match="Unknown ip_version"):
inv.validate_options("ip-addresses", "", "", "ipv4")
class TestHostEntries:
def test_ip_with_dns_name(self):
@@ -133,6 +141,21 @@ class TestHostEntries:
_, host = inv.vm_host_entry({"name": "vm02"})
assert "ansible_host" not in host
def test_vm_ip_version_selects_primary_ip_field(self):
vm = {
"name": "vm01",
"primary_ip": {"address": "2001:db8::20/64"},
"primary_ip4": {"address": "192.0.2.20/24"},
"primary_ip6": {"address": "2001:db8::20/64"},
}
assert inv.vm_host_entry(vm, "both")[1]["ansible_host"] == "2001:db8::20"
assert inv.vm_host_entry(vm, "v4")[1]["ansible_host"] == "192.0.2.20"
assert inv.vm_host_entry(vm, "v6")[1]["ansible_host"] == "2001:db8::20"
def test_vm_without_requested_family_has_no_ansible_host(self):
vm = {"name": "vm01", "primary_ip4": {"address": "192.0.2.20/24"}}
assert "ansible_host" not in inv.vm_host_entry(vm, "v6")[1]
class TestClientSideFilters:
def entries(self):
@@ -202,6 +225,15 @@ class TestLoadConfig:
with pytest.raises(inv.InventoryError, match="url and token"):
inv.load_config(path)
def test_ip_version_defaults_to_both(self, tmp_path):
path = self.write_config(tmp_path, "netbox:\n url: https://nb.example.com\n token: abc\n")
assert inv.load_config(path)["ip_version"] == "both"
def test_ip_version_env_override_is_normalised(self, tmp_path, monkeypatch):
path = self.write_config(tmp_path, "netbox:\n url: https://nb.example.com\n token: abc\n")
monkeypatch.setenv("NETBOX_INVENTORY_IP_VERSION", " V4 ")
assert inv.load_config(path)["ip_version"] == "v4"
def test_invalid_timeout_raises(self, tmp_path):
path = self.write_config(
tmp_path,
@@ -291,6 +323,20 @@ class TestBuildInventory:
inventory = inv.build_inventory(config, make_client())
assert inventory["netbox"]["hosts"] == ["192.0.2.11", "web.example.com"]
@responses.activate
def test_ip_version_becomes_family_parameter(self):
responses.get(
f"{NETBOX_URL}/api/ipam/ip-addresses/",
json={
"next": None,
"results": [{"address": "2001:db8::10/64", "dns_name": "v6.example.com"}],
},
)
config = {"source": "ip-addresses", "filter": "", "filter_value": "", "ip_version": "v6"}
inventory = inv.build_inventory(config, make_client())
assert inventory["netbox"]["hosts"] == ["v6.example.com"]
assert "family=6" in responses.calls[0].request.url
@responses.activate
def test_vm_inventory_with_cluster_filter(self):
responses.get(