diff --git a/netbox_inventory.py b/netbox_inventory.py index 330f4ce..ecccdd9 100755 --- a/netbox_inventory.py +++ b/netbox_inventory.py @@ -25,6 +25,16 @@ SOURCE_IP_ADDRESSES = "ip-addresses" SOURCE_VIRTUAL_MACHINES = "virtual-machines" SOURCES = (SOURCE_IP_ADDRESSES, SOURCE_VIRTUAL_MACHINES) +# 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).""" @@ -111,6 +121,18 @@ class NetBoxClient: 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]: """ @@ -162,6 +184,95 @@ def load_config(path: str | None = None) -> dict[str, Any]: 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) -> None: + """ + Validate the source/filter 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. + """ + if source not in SOURCES: + raise InventoryError(f'Unknown source "{source}", valid sources: {", ".join(SOURCES)}.') + 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] @@ -207,12 +318,15 @@ def vm_host_entry(vm: dict[str, Any]) -> tuple[str, dict[str, Any]]: return name, hostvars -def fetch_entries(client: NetBoxClient, source: str) -> list[tuple[str, dict[str, Any]]]: +def fetch_entries( + client: NetBoxClient, source: str, params: dict[str, Any] | None = None +) -> 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") + records = client.get_all("ipam/ip-addresses", params) return [ip_host_entry(record) for record in records if record.get("address")] - records = client.get_all("virtualization/virtual-machines", {"exclude": "config_context"}) + 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")] @@ -221,14 +335,15 @@ def build_inventory(config: dict[str, Any], client: NetBoxClient) -> dict[str, A Build the full --list inventory structure for Ansible. Raises: - InventoryError: On an unknown source or NetBox API failures. + InventoryError: On invalid source/filter options 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) + filter_name = config["filter"] + filter_value = config["filter_value"] + validate_options(source, filter_name, filter_value) + params = server_side_params(client, filter_name, filter_value) if filter_name else {} + entries = fetch_entries(client, source, params) + 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: