From e2163a9de731ff99cba08e1d58e2e5961dde1c78 Mon Sep 17 00:00:00 2001 From: JulioPDX Date: Mon, 16 Dec 2024 16:11:59 -0800 Subject: [PATCH 1/5] Initial work to add netbox as an inventory source --- anta/cli/get/__init__.py | 1 + anta/cli/get/commands.py | 25 +++++++++++++++++++- anta/cli/get/utils.py | 49 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 74 insertions(+), 1 deletion(-) 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..bf8dc8840 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 @@ -105,6 +105,29 @@ def from_ansible(ctx: click.Context, output: Path, ansible_group: str, ansible_i logger.error(str(e)) ctx.exit(ExitCode.USAGE_ERROR) +@click.command +@click.pass_context +@inventory_output_options +@click.option("--nb-instance", "-nbi", help="Name of NetBox instance", type=str, required=True) +@click.option("--nb-token", "-nbt", help="NetBox Token", type=str, required=True) +@click.option("--nb-platform", "-nbt", help="NetBox device platform", type=str, default="Arista EOS", required=True) +@click.option("--nb-token", "-nbt", help="NetBox Token", type=str, required=True) +@click.option("--nb-verify", "-nbf", help="NetBox Verify", type=bool, default=False, required=False) +def from_netbox(ctx: click.Context, nb_instance: str, output: Path, nb_token: str, nb_platform: str, 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, + verify=nb_verify + ) + except ValueError as e: + logger.error(str(e)) + ctx.exit(ExitCode.USAGE_ERROR) + @click.command @inventory_options diff --git a/anta/cli/get/utils.py b/anta/cli/get/utils.py index 6f5d7d0ff..78ff2fef5 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,54 @@ 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", 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 query of the platform 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) + + 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. From 664bb4b0a2b4e8946338586a5bccd45e14043d57 Mon Sep 17 00:00:00 2001 From: JulioPDX Date: Mon, 16 Dec 2024 16:22:29 -0800 Subject: [PATCH 2/5] update wording --- anta/cli/get/commands.py | 18 ++++++------------ anta/cli/get/utils.py | 4 ++-- test.yml | 12 ++++++++++++ 3 files changed, 20 insertions(+), 14 deletions(-) create mode 100644 test.yml diff --git a/anta/cli/get/commands.py b/anta/cli/get/commands.py index bf8dc8840..fc241e3c4 100644 --- a/anta/cli/get/commands.py +++ b/anta/cli/get/commands.py @@ -105,25 +105,19 @@ def from_ansible(ctx: click.Context, output: Path, ansible_group: str, ansible_i logger.error(str(e)) ctx.exit(ExitCode.USAGE_ERROR) + @click.command @click.pass_context @inventory_output_options -@click.option("--nb-instance", "-nbi", help="Name of NetBox instance", type=str, required=True) -@click.option("--nb-token", "-nbt", help="NetBox Token", type=str, required=True) +@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", "-nbt", help="NetBox device platform", type=str, default="Arista EOS", required=True) -@click.option("--nb-token", "-nbt", help="NetBox Token", type=str, required=True) -@click.option("--nb-verify", "-nbf", help="NetBox Verify", type=bool, default=False, required=False) -def from_netbox(ctx: click.Context, nb_instance: str, output: Path, nb_token: str, nb_platform: str, nb_verify: bool=False) -> None: +@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_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, - verify=nb_verify - ) + create_inventory_from_netbox(nb_instance=nb_instance, output=output, token=nb_token, platform=nb_platform, verify=nb_verify) except ValueError as e: logger.error(str(e)) ctx.exit(ExitCode.USAGE_ERROR) diff --git a/anta/cli/get/utils.py b/anta/cli/get/utils.py index 78ff2fef5..c8be7020c 100644 --- a/anta/cli/get/utils.py +++ b/anta/cli/get/utils.py @@ -215,7 +215,7 @@ 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", verify: bool=False) -> None: +def create_inventory_from_netbox(nb_instance: str, output: Path, token: str, platform: str = "Arista EOS", verify: bool = False) -> None: """Fetch devices from NetBox filtered by a specific platform. Parameters @@ -227,7 +227,7 @@ def create_inventory_from_netbox(nb_instance: str, output: Path, token: str, pla token The token used to authenticate to the NetBox instance. platform - The query of the platform to filter devices by. + The platform to filter devices by. verify Verify the SSL certification of the NetBox instance. diff --git a/test.yml b/test.yml new file mode 100644 index 000000000..9644eac8c --- /dev/null +++ b/test.yml @@ -0,0 +1,12 @@ +anta_inventory: + hosts: + - host: 192.168.100.50 + name: dc1-leaf1 + tags: + - Quebec + - Bravo + - Echo + - host: 192.168.100.51 + name: dc1-leaf2 + tags: + - Bravo From 9867f03e7e1a4082dab7b68d61a404e0b413136c Mon Sep 17 00:00:00 2001 From: JulioPDX Date: Mon, 16 Dec 2024 16:24:52 -0800 Subject: [PATCH 3/5] removing test file --- test.yml | 12 ------------ 1 file changed, 12 deletions(-) delete mode 100644 test.yml diff --git a/test.yml b/test.yml deleted file mode 100644 index 9644eac8c..000000000 --- a/test.yml +++ /dev/null @@ -1,12 +0,0 @@ -anta_inventory: - hosts: - - host: 192.168.100.50 - name: dc1-leaf1 - tags: - - Quebec - - Bravo - - Echo - - host: 192.168.100.51 - name: dc1-leaf2 - tags: - - Bravo From f35a57e0609796a488d8e99dbf29e6c7871fc36b Mon Sep 17 00:00:00 2001 From: JulioPDX Date: Mon, 16 Dec 2024 16:28:03 -0800 Subject: [PATCH 4/5] adding pynetbox to install options --- pyproject.toml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index 212418760..68b93876e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -101,6 +101,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" From 95a589ab403ebade16f7e25bf5f4b9ffbe2b518f Mon Sep 17 00:00:00 2001 From: JulioPDX Date: Tue, 17 Dec 2024 13:29:01 -0800 Subject: [PATCH 5/5] adding optional site option --- anta/cli/get/commands.py | 7 ++++--- anta/cli/get/utils.py | 6 ++++-- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/anta/cli/get/commands.py b/anta/cli/get/commands.py index fc241e3c4..f04ce6296 100644 --- a/anta/cli/get/commands.py +++ b/anta/cli/get/commands.py @@ -111,13 +111,14 @@ def from_ansible(ctx: click.Context, output: Path, ansible_group: str, ansible_i @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", "-nbt", help="NetBox device platform", type=str, default="Arista EOS", 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_verify: bool = False) -> None: +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, verify=nb_verify) + 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) diff --git a/anta/cli/get/utils.py b/anta/cli/get/utils.py index c8be7020c..18089dd96 100644 --- a/anta/cli/get/utils.py +++ b/anta/cli/get/utils.py @@ -215,7 +215,7 @@ 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", verify: bool = False) -> None: +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 @@ -228,6 +228,8 @@ def create_inventory_from_netbox(nb_instance: str, output: Path, token: str, pla 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. @@ -246,7 +248,7 @@ def create_inventory_from_netbox(nb_instance: str, output: Path, token: str, pla # Platform name to filter platform = nb.dcim.platforms.get(q=[platform]) - devices = nb.dcim.devices.filter(platform=platform.slug) + 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: