From c1f0b4f724ddaad693c9a9fe4d64b6ab56826d30 Mon Sep 17 00:00:00 2001 From: Bryan Joshua Pedini Date: Tue, 25 Aug 2026 14:53:24 +0200 Subject: [PATCH] feat: ip_version option to restrict inventories to ipv4 or ipv6 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. --- config.yaml.example | 2 ++ netbox_inventory.py | 51 +++++++++++++++++++++++++++++++---------- tests/test_inventory.py | 46 +++++++++++++++++++++++++++++++++++++ 3 files changed, 87 insertions(+), 12 deletions(-) diff --git a/config.yaml.example b/config.yaml.example index b8e4cb4..f091760 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -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" diff --git a/netbox_inventory.py b/netbox_inventory.py index bcf8da5..d83ada4 100755 --- a/netbox_inventory.py +++ b/netbox_inventory.py @@ -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: diff --git a/tests/test_inventory.py b/tests/test_inventory.py index f3a9837..b432210 100644 --- a/tests/test_inventory.py +++ b/tests/test_inventory.py @@ -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(