diff --git a/anta/cli/get/__init__.py b/anta/cli/get/__init__.py index 8763b35c1..4645871cd 100644 --- a/anta/cli/get/__init__.py +++ b/anta/cli/get/__init__.py @@ -15,6 +15,7 @@ def get() -> None: get.add_command(commands.from_cvp) get.add_command(commands.from_ansible) +get.add_command(commands.from_netbox) get.add_command(commands.inventory) get.add_command(commands.tags) get.add_command(commands.tests) diff --git a/anta/cli/get/commands.py b/anta/cli/get/commands.py index 2bdd9cb9f..f04ce6296 100644 --- a/anta/cli/get/commands.py +++ b/anta/cli/get/commands.py @@ -22,7 +22,7 @@ from anta.cli.get.utils import inventory_output_options from anta.cli.utils import ExitCode, inventory_options -from .utils import create_inventory_from_ansible, create_inventory_from_cvp, explore_package, get_cv_token +from .utils import create_inventory_from_ansible, create_inventory_from_cvp, create_inventory_from_netbox, explore_package, get_cv_token if TYPE_CHECKING: from anta.inventory import AntaInventory @@ -106,6 +106,24 @@ def from_ansible(ctx: click.Context, output: Path, ansible_group: str, ansible_i ctx.exit(ExitCode.USAGE_ERROR) +@click.command +@click.pass_context +@inventory_output_options +@click.option("--nb-instance", "-nbi", help="Name of the NetBox instance", type=str, required=True) +@click.option("--nb-token", "-nbt", help="NetBox token", type=str, required=True) +@click.option("--nb-platform", "-nbp", help="NetBox device platform", type=str, default="Arista EOS", required=False) +@click.option("--nb-site", "-nbs", help="NetBox site (case sensitive)", type=str, default=None, required=False) +@click.option("--nb-verify", "-nbf", help="NetBox verify SSL", type=bool, default=False, required=False) +def from_netbox(ctx: click.Context, nb_instance: str, output: Path, nb_token: str, nb_platform: str, nb_site: str | None = None, nb_verify: bool = False) -> None: + """Build ANTA inventory from a NetBox instance.""" + logger.info("Building inventory from netbox instance file '%s'", nb_instance) + try: + create_inventory_from_netbox(nb_instance=nb_instance, output=output, token=nb_token, platform=nb_platform, site=nb_site, verify=nb_verify) + except ValueError as e: + logger.error(str(e)) + ctx.exit(ExitCode.USAGE_ERROR) + + @click.command @inventory_options @click.option("--connected/--not-connected", help="Display inventory after connection has been created", default=False, required=False) diff --git a/anta/cli/get/utils.py b/anta/cli/get/utils.py index 6f5d7d0ff..18089dd96 100644 --- a/anta/cli/get/utils.py +++ b/anta/cli/get/utils.py @@ -8,6 +8,7 @@ import functools import importlib import inspect +import ipaddress import json import logging import pkgutil @@ -214,6 +215,56 @@ def create_inventory_from_ansible(inventory: Path, output: Path, ansible_group: write_inventory_to_file(ansible_hosts, output) +def create_inventory_from_netbox(nb_instance: str, output: Path, token: str, platform: str = "Arista EOS", site: str | None = None, verify: bool = False) -> None: + """Fetch devices from NetBox filtered by a specific platform. + + Parameters + ---------- + nb_instance + The NetBox API instance. + output + ANTA inventory file to generate. + token + The token used to authenticate to the NetBox instance. + platform + The platform to filter devices by. + site + The site to filter devices by. + verify + Verify the SSL certification of the NetBox instance. + + """ + session = requests.session() + session.verify = verify + try: + import pynetbox + except ImportError as e: + logging.error(e) + + try: + # Initialize NetBox API + nb = pynetbox.api(nb_instance, token=token) + + # Platform name to filter + platform = nb.dcim.platforms.get(q=[platform]) + + devices = nb.dcim.devices.filter(platform=platform.slug, site=site) if site else nb.dcim.devices.filter(platform=platform.slug) + + inventory = [] + for device in devices: + host_entry = { + "host": str(ipaddress.ip_interface(device.primary_ip).ip), + "name": device.name, + "tags": [tag.name for tag in device.tags], + } + inventory.append(host_entry) + + write_inventory_to_file(inventory, output) + + except Exception as e: + raise ValueError(e) from e + + def explore_package(module_name: str, test_name: str | None = None, *, short: bool = False, count: bool = False) -> int: """Parse ANTA test submodules recursively and print AntaTest examples. diff --git a/pyproject.toml b/pyproject.toml index ac2ca3363..cf7b530aa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -100,6 +100,9 @@ doc = [ "black>=24.10.0", "mkdocs-github-admonitions-plugin>=0.0.3" ] +netbox = [ + "pynetbox>=7.4.1" +] [project.urls] Homepage = "https://anta.arista.com"