diff --git a/README.md b/README.md index c07ed87..c015dfb 100644 --- a/README.md +++ b/README.md @@ -26,8 +26,10 @@ pip install ntc-netbox-plugin-onboarding systemctl restart netbox netbox-rq ``` -> The plugin is compatible with NetBox 2.8.1 and higher - +> The ntc-netbox-plugin-onboarding v1.3 is compatible with NetBox 2.8 + +> The ntc-netbox-plugin-onboarding v2 is compatible with NetBox 2.8 and NetBox 2.9 + To ensure NetBox Onboarding plugin is automatically re-installed during future upgrades, create a file named `local_requirements.txt` (if not already existing) in the NetBox root directory (alongside `requirements.txt`) and list the `ntc-netbox-plugin-onboarding` package: ```no-highlight @@ -64,26 +66,38 @@ The plugin behavior can be controlled with the following list of settings - `default_device_role_color` string (default FF0000), color assigned to the device role if it needs to be created. - `default_management_interface` string (default "PLACEHOLDER"), name of the management interface that will be created, if one can't be identified on the device. - `default_management_prefix_length` integer ( default 0), length of the prefix that will be used for the management IP address, if the IP can't be found. +- `skip_device_type_on_update` boolean (default False), If True, an existing NetBox device will not get its device type updated. If False, device type will be updated with one discovered on a device. +- `skip_manufacturer_on_update` boolean (default False), If True, an existing NetBox device will not get its manufacturer updated. If False, manufacturer will be updated with one discovered on a device. - `platform_map` (dictionary), mapping of an **auto-detected** Netmiko platform to the **NetBox slug** name of your Platform. The dictionary should be in the format: ```python { - : + : + } + ``` +- `onboarding_extensions_map` (dictionary), mapping of a NAPALM driver name to the loadable Python module used as an onboarding extension. The dictionary should be in the format: + ```python + { + : } ``` +- `object_match_strategy` (string), defines the method for searching models. There are +currently two strategies, strict and loose. Strict has to be a direct match, normally +using a slug. Loose allows a range of search criteria to match a single object. If multiple +objects are returned an error is raised. ## Usage ### Preparation -To work properly the plugin needs to know the Site, Platform, Device Type, Device Role of each -device as well as its primary IP address or DNS Name. It's recommended to create these objects in -NetBox ahead of time and to provide them when you want to start the onboarding process. +To properly onboard a device, the plugin needs to only know the Site as well as device's primary IP address or DNS Name. > For DNS Name Resolution to work, the instance of NetBox must be able to resolve the name of the > device to IP address. +Providing other attributes (`Platform`, `Device Type`, `Device Role`) is optional - if any of these attributes is provided, plugin will use provided value for the onboarded device. If `Platform`, `Device Type` and/or `Device Role` are not provided, the plugin will try to identify these information automatically and, based on the settings, it can create them in NetBox as needed. -> If the Platform is provided, it must contains a valid Napalm driver available to the worker in Python +> If the Platform is provided, it must point to an existing NetBox Platform. NAPALM driver of this platform will be used only if it is defined for the platform in NetBox. +> To use a preferred NAPALM driver, either define it in NetBox per platform or in the plugins settings under `platform_map` ### Onboard a new device diff --git a/development/base_configuration.py b/development/base_configuration.py index 5bbe5a6..12aa5a1 100644 --- a/development/base_configuration.py +++ b/development/base_configuration.py @@ -114,7 +114,20 @@ # Enable custom logging. Please see the Django documentation for detailed guidance on configuring custom logs: # https://docs.djangoproject.com/en/1.11/topics/logging/ -LOGGING = {} +LOGGING = { + "version": 1, + "disable_existing_loggers": False, + "formatters": {"rq_console": {"format": "%(asctime)s %(message)s", "datefmt": "%H:%M:%S",},}, + "handlers": { + "rq_console": { + "level": "DEBUG", + "class": "rq.utils.ColorizingStreamHandler", + "formatter": "rq_console", + "exclude": ["%(asctime)s"], + }, + }, + "loggers": {"rq.worker": {"handlers": ["rq_console"], "level": "DEBUG"},}, +} # Setting this to True will permit only authenticated users to access any part of NetBox. By default, anonymous users # are permitted to access most data in NetBox (excluding secrets) but not make any changes. diff --git a/docs/examples/example_ios_set_device_role.py b/docs/examples/example_ios_set_device_role.py new file mode 100644 index 0000000..e0743d8 --- /dev/null +++ b/docs/examples/example_ios_set_device_role.py @@ -0,0 +1,96 @@ +"""Example of custom onboarding class. + +(c) 2020 Network To Code +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +from netbox_onboarding.netbox_keeper import NetboxKeeper +from netbox_onboarding.onboarding.onboarding import Onboarding + + +class MyOnboardingClass(Onboarding): + """Custom onboarding class example. + + Main purpose of this class is to access and modify the onboarding_kwargs. + By accessing the onboarding kwargs, user gains ability to modify + onboarding parameters before the objects are created in NetBox. + + This class adds the get_device_role method that does the static + string comparison and returns the device role. + """ + + def run(self, onboarding_kwargs): + """Ensures network device.""" + # Access hostname from onboarding_kwargs and get device role automatically + device_new_role = self.get_device_role(hostname=onboarding_kwargs["netdev_hostname"]) + + # Update the device role in onboarding kwargs dictionary + onboarding_kwargs["netdev_nb_role_slug"] = device_new_role + + nb_k = NetboxKeeper(**onboarding_kwargs) + nb_k.ensure_device() + + self.created_device = nb_k.device + + @staticmethod + def get_device_role(hostname): + """Returns the device role based on hostname data. + + This is a static analysis of hostname string content only + """ + hostname_lower = hostname.lower() + if ("rtr" in hostname_lower) or ("router" in hostname_lower): + role = "router" + elif ("sw" in hostname_lower) or ("switch" in hostname_lower): + role = "switch" + elif ("fw" in hostname_lower) or ("firewall" in hostname_lower): + role = "firewall" + elif "dc" in hostname_lower: + role = "datacenter" + else: + role = "generic" + + return role + + +class OnboardingDriverExtensions: + """This is an example of a custom onboarding driver extension. + + This extension sets the onboarding_class to MyOnboardingClass, + which is an example class of how to access and modify the device + role automatically through the onboarding process. + """ + + def __init__(self, napalm_device): + """Inits the class.""" + self.napalm_device = napalm_device + self.onboarding_class = MyOnboardingClass + self.ext_result = None + + def get_onboarding_class(self): + """Return onboarding class for IOS driver. + + Currently supported is Standalone Onboarding Process + + Result of this method is used by the OnboardingManager to + initiate the instance of the onboarding class. + """ + return self.onboarding_class + + def get_ext_result(self): + """This method is used to store any object as a return value. + + Result of this method is passed to the onboarding class as + driver_addon_result argument. + + :return: Any() + """ + return self.ext_result diff --git a/docs/release-notes/version-2.0.md b/docs/release-notes/version-2.0.md new file mode 100644 index 0000000..d43a64d --- /dev/null +++ b/docs/release-notes/version-2.0.md @@ -0,0 +1,24 @@ +# ntc-netbox-plugin-onboarding v2.0 Release Notes + +## v2.0 + +### Enhancements + +* NetBox 2.9 support - Supported releases 2.8 and 2.9 +* Onboarding extensions - Customizable onboarding process through Python modules. +* Onboarding details exposed in a device view - Date, Status, Last success and Latest task id related to the onboarded device are presented under the device view. +* Onboarding task view - Onboarding details exposed in a dedicated view, including NetBox's ChangeLog. +* Onboarding Changelog - Onboarding uses NetBox's ChangeLog to display user and changes made to the Onboarding Task object. +* Skip onboarding feature - New attribute in the OnboardingDevice model allows to skip the onboarding request on devices with disabled onboarding setting. + +### Bug Fixes + +* Fixed race condition in `worker.py` +* Improved logging + +### Additional Changes + +* Platform map now includes NAPALM drivers as defined in NetBox +* Tests have been refactored to inherit NetBox's tests +* Onboarding process will update the Device found by the IP-address lookup. In case of no existing device with onboarded IP-address is found in NetBox, onboarding might update the existing NetBox' looking up by network device's hostname. +* Onboarding will raise Exception when `create_device_type_if_missing` is set to `False` for existing Device with DeviceType mismatch (behaviour pre https://github.com/networktocode/ntc-netbox-plugin-onboarding/issues/74) diff --git a/netbox_onboarding/__init__.py b/netbox_onboarding/__init__.py index 964f4a9..d1bc886 100644 --- a/netbox_onboarding/__init__.py +++ b/netbox_onboarding/__init__.py @@ -12,7 +12,7 @@ limitations under the License. """ -__version__ = "1.3.0" +__version__ = "2.0.0" from extras.plugins import PluginConfig @@ -39,7 +39,11 @@ class OnboardingConfig(PluginConfig): "default_management_prefix_length": 0, "default_device_status": "active", "create_management_interface_if_missing": True, + "skip_device_type_on_update": False, + "skip_manufacturer_on_update": False, "platform_map": {}, + "onboarding_extensions_map": {"ios": "netbox_onboarding.onboarding_extensions.ios",}, + "object_match_strategy": "loose", } caching_config = {} diff --git a/netbox_onboarding/admin.py b/netbox_onboarding/admin.py index 9b2df8c..4287565 100644 --- a/netbox_onboarding/admin.py +++ b/netbox_onboarding/admin.py @@ -32,5 +32,5 @@ class OnboardingTaskAdmin(admin.ModelAdmin): "failed_reason", "port", "timeout", - "created_on", + "created", ) diff --git a/netbox_onboarding/choices.py b/netbox_onboarding/choices.py index ebcd9f4..f79bcaf 100644 --- a/netbox_onboarding/choices.py +++ b/netbox_onboarding/choices.py @@ -22,12 +22,14 @@ class OnboardingStatusChoices(ChoiceSet): STATUS_PENDING = "pending" STATUS_RUNNING = "running" STATUS_SUCCEEDED = "succeeded" + STATUS_SKIPPED = "skipped" CHOICES = ( (STATUS_FAILED, "failed"), (STATUS_PENDING, "pending"), (STATUS_RUNNING, "running"), (STATUS_SUCCEEDED, "succeeded"), + (STATUS_SKIPPED, "skipped"), ) diff --git a/netbox_onboarding/constants.py b/netbox_onboarding/constants.py index c6f148f..01523ef 100644 --- a/netbox_onboarding/constants.py +++ b/netbox_onboarding/constants.py @@ -1,5 +1,6 @@ """Constants for netbox_onboarding plugin.""" -NETMIKO_TO_NAPALM = { + +NETMIKO_TO_NAPALM_STATIC = { "cisco_ios": "ios", "cisco_nxos": "nxos_ssh", "arista_eos": "eos", diff --git a/netbox_onboarding/exceptions.py b/netbox_onboarding/exceptions.py new file mode 100644 index 0000000..1ed4db1 --- /dev/null +++ b/netbox_onboarding/exceptions.py @@ -0,0 +1,39 @@ +"""Exceptions. + +(c) 2020 Network To Code +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + + +class OnboardException(Exception): + """A failure occurred during the onboarding process. + + The exception includes a reason "slug" as defined below as well as a humanized message. + """ + + REASONS = ( + "fail-config", # config provided is not valid + "fail-connect", # device is unreachable at IP:PORT + "fail-execute", # unable to execute device/API command + "fail-login", # bad username/password + "fail-dns", # failed to get IP address from name resolution + "fail-general", # other error + ) + + def __init__(self, reason, message, **kwargs): + """Exception Init.""" + super(OnboardException, self).__init__(kwargs) + self.reason = reason + self.message = message + + def __str__(self): + """Exception __str__.""" + return f"{self.__class__.__name__}: {self.reason}: {self.message}" diff --git a/netbox_onboarding/filters.py b/netbox_onboarding/filters.py index 33dd7e3..f9ae743 100644 --- a/netbox_onboarding/filters.py +++ b/netbox_onboarding/filters.py @@ -26,8 +26,6 @@ class OnboardingTaskFilter(NameSlugSearchFilterSet): q = django_filters.CharFilter(method="search", label="Search",) - site_id = django_filters.ModelMultipleChoiceFilter(queryset=Site.objects.all(), label="Site (ID)",) - site = django_filters.ModelMultipleChoiceFilter( field_name="site__slug", queryset=Site.objects.all(), to_field_name="slug", label="Site (slug)", ) @@ -46,7 +44,7 @@ class Meta: # noqa: D106 "Missing docstring in public nested class" model = OnboardingTask fields = ["id", "site", "site_id", "platform", "role", "status", "failed_reason"] - def search(self, queryset, name, value): + def search(self, queryset, name, value): # pylint: disable=unused-argument, no-self-use """Perform the filtered search.""" if not value.strip(): return queryset @@ -55,7 +53,7 @@ def search(self, queryset, name, value): | Q(ip_address__icontains=value) | Q(site__name__icontains=value) | Q(platform__name__icontains=value) - | Q(device__icontains=value) + | Q(created_device__name__icontains=value) | Q(status__icontains=value) | Q(failed_reason__icontains=value) | Q(message__icontains=value) diff --git a/netbox_onboarding/forms.py b/netbox_onboarding/forms.py index 49ca9ac..2c3cc0f 100644 --- a/netbox_onboarding/forms.py +++ b/netbox_onboarding/forms.py @@ -13,11 +13,11 @@ """ from django import forms +from django.db import transaction from django_rq import get_queue -from utilities.forms import BootstrapMixin +from utilities.forms import BootstrapMixin, CSVModelForm from dcim.models import Site, Platform, DeviceRole, DeviceType -from extras.forms import CustomFieldModelCSVForm from .models import OnboardingTask from .choices import OnboardingStatusChoices, OnboardingFailChoices @@ -33,7 +33,7 @@ class OnboardingTaskForm(BootstrapMixin, forms.ModelForm): required=True, label="IP address", help_text="IP Address/DNS Name of the device to onboard" ) - site = forms.ModelChoiceField(required=True, queryset=Site.objects.all(), to_field_name="slug") + site = forms.ModelChoiceField(required=True, queryset=Site.objects.all()) username = forms.CharField(required=False, help_text="Device username (will not be stored in database)") password = forms.CharField( @@ -106,7 +106,7 @@ class Meta: # noqa: D106 "Missing docstring in public nested class" fields = ["q", "site", "platform", "status", "failed_reason"] -class OnboardingTaskFeedCSVForm(CustomFieldModelCSVForm): +class OnboardingTaskFeedCSVForm(CSVModelForm): """Form for entering CSV to bulk-import OnboardingTask entries.""" site = forms.ModelChoiceField( @@ -149,12 +149,21 @@ class OnboardingTaskFeedCSVForm(CustomFieldModelCSVForm): class Meta: # noqa: D106 "Missing docstring in public nested class" model = OnboardingTask - fields = OnboardingTask.csv_headers + fields = [ + "site", + "ip_address", + "port", + "timeout", + "platform", + "role", + ] def save(self, commit=True, **kwargs): """Save the model, and add it and the associated credentials to the onboarding worker queue.""" model = super().save(commit=commit, **kwargs) if commit: credentials = Credentials(self.data.get("username"), self.data.get("password"), self.data.get("secret")) - get_queue("default").enqueue("netbox_onboarding.worker.onboard_device", model.pk, credentials) + transaction.on_commit( + lambda: get_queue("default").enqueue("netbox_onboarding.worker.onboard_device", model.pk, credentials) + ) return model diff --git a/netbox_onboarding/metrics.py b/netbox_onboarding/metrics.py new file mode 100644 index 0000000..9bde6fe --- /dev/null +++ b/netbox_onboarding/metrics.py @@ -0,0 +1,18 @@ +"""Plugin additions to the NetBox navigation menu. + +(c) 2020 Network To Code +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" +from prometheus_client import Counter + +onboardingtask_results_counter = Counter( + name="onboardingtask_results_total", documentation="Count of results for Onboarding Task", labelnames=("status",) +) diff --git a/netbox_onboarding/migrations/0002_onboardingdevice.py b/netbox_onboarding/migrations/0002_onboardingdevice.py new file mode 100644 index 0000000..45b8503 --- /dev/null +++ b/netbox_onboarding/migrations/0002_onboardingdevice.py @@ -0,0 +1,22 @@ +# Generated by Django 2.2.10 on 2020-08-21 11:05 + +from django.db import migrations, models +import django.db.models.deletion + + +class Migration(migrations.Migration): + + dependencies = [ + ("netbox_onboarding", "0001_initial"), + ] + + operations = [ + migrations.CreateModel( + name="OnboardingDevice", + fields=[ + ("id", models.AutoField(auto_created=True, primary_key=True, serialize=False)), + ("enabled", models.BooleanField(default=True)), + ("device", models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, to="dcim.Device")), + ], + ), + ] diff --git a/netbox_onboarding/migrations/0003_onboardingtask_change_logging_model.py b/netbox_onboarding/migrations/0003_onboardingtask_change_logging_model.py new file mode 100644 index 0000000..8fd22a0 --- /dev/null +++ b/netbox_onboarding/migrations/0003_onboardingtask_change_logging_model.py @@ -0,0 +1,19 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("netbox_onboarding", "0002_onboardingdevice"), + ] + + operations = [ + migrations.AddField( + model_name="onboardingtask", name="created", field=models.DateField(auto_now_add=True, null=True), + ), + migrations.AddField( + model_name="onboardingtask", name="last_updated", field=models.DateTimeField(auto_now=True, null=True), + ), + migrations.AlterModelOptions(name="onboardingtask", options={},), + migrations.RemoveField(model_name="onboardingtask", name="created_on",), + ] diff --git a/netbox_onboarding/models.py b/netbox_onboarding/models.py index ac2bd6f..5179034 100644 --- a/netbox_onboarding/models.py +++ b/netbox_onboarding/models.py @@ -11,11 +11,23 @@ See the License for the specific language governing permissions and limitations under the License. """ +from django.db.models.signals import post_save +from django.dispatch import receiver from django.db import models +from django.urls import reverse +from dcim.models import Device from .choices import OnboardingStatusChoices, OnboardingFailChoices +from .release import NETBOX_RELEASE_CURRENT, NETBOX_RELEASE_29 +# Support NetBox 2.8 +if NETBOX_RELEASE_CURRENT < NETBOX_RELEASE_29: + from utilities.models import ChangeLoggedModel # pylint: disable=no-name-in-module, import-error +# Support NetBox 2.9 +else: + from extras.models import ChangeLoggedModel # pylint: disable=no-name-in-module, import-error -class OnboardingTask(models.Model): + +class OnboardingTask(ChangeLoggedModel): """The status of each onboarding Task is tracked in the OnboardingTask table.""" created_device = models.ForeignKey(to="dcim.Device", on_delete=models.SET_NULL, blank=True, null=True) @@ -45,20 +57,97 @@ class OnboardingTask(models.Model): help_text="Timeout period in sec to wait while connecting to the device", default=30 ) - created_on = models.DateTimeField(auto_now_add=True) - - csv_headers = [ - "site", - "ip_address", - "port", - "timeout", - "platform", - "role", - ] - - class Meta: # noqa: D106 "missing docstring in public nested class" - ordering = ["created_on"] - def __str__(self): """String representation of an OnboardingTask.""" return f"{self.site} : {self.ip_address}" + + def get_absolute_url(self): + """Provide absolute URL to an OnboardingTask.""" + return reverse("plugins:netbox_onboarding:onboardingtask", kwargs={"pk": self.pk}) + + if NETBOX_RELEASE_CURRENT >= NETBOX_RELEASE_29: + from utilities.querysets import RestrictedQuerySet # pylint: disable=no-name-in-module, import-outside-toplevel + + objects = RestrictedQuerySet.as_manager() + + +class OnboardingDevice(models.Model): + """The status of each Onboarded Device is tracked in the OnboardingDevice table.""" + + device = models.OneToOneField(to="dcim.Device", on_delete=models.CASCADE) + enabled = models.BooleanField(default=True, help_text="Whether (re)onboarding of this device is permitted") + + @property + def last_check_attempt_date(self): + """Date of last onboarding attempt for a device.""" + if self.device.primary_ip4: + try: + return ( + OnboardingTask.objects.filter( + ip_address=self.device.primary_ip4.address.ip.format() # pylint: disable=no-member + ) + .latest("last_updated") + .created + ) + except OnboardingTask.DoesNotExist: + return "unknown" + else: + return "unknown" + + @property + def last_check_successful_date(self): + """Date of last successful onboarding for a device.""" + if self.device.primary_ip4: + try: + return ( + OnboardingTask.objects.filter( + ip_address=self.device.primary_ip4.address.ip.format(), # pylint: disable=no-member + status=OnboardingStatusChoices.STATUS_SUCCEEDED, + ) + .latest("last_updated") + .created + ) + except OnboardingTask.DoesNotExist: + return "unknown" + else: + return "unknown" + + @property + def status(self): + """Last onboarding status.""" + if self.device.primary_ip4: + try: + return ( + OnboardingTask.objects.filter( + ip_address=self.device.primary_ip4.address.ip.format() # pylint: disable=no-member + ) + .latest("last_updated") + .status + ) + except OnboardingTask.DoesNotExist: + return "unknown" + else: + return "unknown" + + @property + def last_ot(self): + """Last onboarding task.""" + if self.device.primary_ip4: + try: + return OnboardingTask.objects.filter( + ip_address=self.device.primary_ip4.address.ip.format() # pylint: disable=no-member + ).latest("last_updated") + except OnboardingTask.DoesNotExist: + return "unknown" + else: + return "unknown" + + +@receiver(post_save, sender=Device) +def init_onboarding_for_new_device(sender, instance, created, **kwargs): # pylint: disable=unused-argument + """Register to create a OnboardingDevice object for each new Device Object using Django Signal. + + https://docs.djangoproject.com/en/3.0/ref/signals/#post-save + """ + if created: + OnboardingDevice.objects.create(device=instance) diff --git a/netbox_onboarding/navigation.py b/netbox_onboarding/navigation.py index efb445f..3d1fb33 100644 --- a/netbox_onboarding/navigation.py +++ b/netbox_onboarding/navigation.py @@ -17,19 +17,19 @@ menu_items = ( PluginMenuItem( - link="plugins:netbox_onboarding:onboarding_task_list", + link="plugins:netbox_onboarding:onboardingtask_list", link_text="Onboarding Tasks", permissions=["netbox_onboarding.view_onboardingtask"], buttons=( PluginMenuButton( - link="plugins:netbox_onboarding:onboarding_task_add", + link="plugins:netbox_onboarding:onboardingtask_add", title="Onboard", icon_class="fa fa-plus", color=ButtonColorChoices.GREEN, permissions=["netbox_onboarding.add_onboardingtask"], ), PluginMenuButton( - link="plugins:netbox_onboarding:onboarding_task_import", + link="plugins:netbox_onboarding:onboardingtask_import", title="Bulk Onboard", icon_class="fa fa-download", color=ButtonColorChoices.BLUE, diff --git a/netbox_onboarding/netbox_keeper.py b/netbox_onboarding/netbox_keeper.py new file mode 100644 index 0000000..2ebe9c4 --- /dev/null +++ b/netbox_onboarding/netbox_keeper.py @@ -0,0 +1,434 @@ +"""NetBox Keeper. + +(c) 2020 Network To Code +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +import logging +import re + +from django.conf import settings +from django.utils.text import slugify +from dcim.models import Manufacturer, Device, Interface, DeviceType, DeviceRole +from dcim.models import Platform +from dcim.models import Site +from ipam.models import IPAddress + +from .constants import NETMIKO_TO_NAPALM_STATIC +from .exceptions import OnboardException + +logger = logging.getLogger("rq.worker") + +PLUGIN_SETTINGS = settings.PLUGINS_CONFIG["netbox_onboarding"] + + +def object_match(obj, search_array): + """Used to search models for multiple criteria. + + Inputs: + obj: The model used for searching. + search_array: Nested dictionaries used to search models. First criteria will be used + for strict searching. Loose searching will loop through the search_array + until it finds a match. Example below. + [ + {"slug__iexact": 'switch1'}, + {"model__iexact": 'Cisco'} + ] + """ + try: + result = obj.objects.get(**search_array[0]) + return result + except obj.DoesNotExist: + if PLUGIN_SETTINGS["object_match_strategy"] == "loose": + for search_array_element in search_array[1:]: + try: + result = obj.objects.get(**search_array_element) + return result + except obj.DoesNotExist: + pass + except obj.MultipleObjectsReturned: + raise OnboardException( + reason="fail-general", + message=f"ERROR multiple objects found in {str(obj)} searching on {str(search_array_element)})", + ) + raise + except obj.MultipleObjectsReturned: + raise OnboardException( + reason="fail-general", + message=f"ERROR multiple objects found in {str(obj)} searching on {str(search_array_element)})", + ) + + +class NetboxKeeper: + """Used to manage the information relating to the network device within the NetBox server.""" + + def __init__( # pylint: disable=R0913,R0914 + self, + netdev_hostname, + netdev_nb_role_slug, + netdev_vendor, + netdev_nb_site_slug, + netdev_nb_device_type_slug=None, + netdev_model=None, + netdev_nb_role_color=None, + netdev_mgmt_ip_address=None, + netdev_nb_platform_slug=None, + netdev_serial_number=None, + netdev_mgmt_ifname=None, + netdev_mgmt_pflen=None, + netdev_netmiko_device_type=None, + onboarding_class=None, + driver_addon_result=None, + ): + """Create an instance and initialize the managed attributes that are used throughout the onboard processing. + + Args: + netdev_hostname (str): NetBox's device name + netdev_nb_role_slug (str): NetBox's device role slug + netdev_vendor (str): Device's vendor name + netdev_nb_site_slug (str): Device site's slug + netdev_nb_device_type_slug (str): Device type's slug + netdev_model (str): Device's model + netdev_nb_role_color (str): NetBox device's role color + netdev_mgmt_ip_address (str): IPv4 Address of a device + netdev_nb_platform_slug (str): NetBox device's platform slug + netdev_serial_number (str): Device's serial number + netdev_mgmt_ifname (str): Device's management interface name + netdev_mgmt_pflen (str): Device's management IP prefix-len + netdev_netmiko_device_type (str): Device's Netmiko device type + onboarding_class (Object): Onboarding Class (future use) + driver_addon_result (Any): Attached extended result (future use) + """ + self.netdev_mgmt_ip_address = netdev_mgmt_ip_address + self.netdev_nb_site_slug = netdev_nb_site_slug + self.netdev_nb_device_type_slug = netdev_nb_device_type_slug + self.netdev_nb_role_slug = netdev_nb_role_slug + self.netdev_nb_role_color = netdev_nb_role_color + self.netdev_nb_platform_slug = netdev_nb_platform_slug + + self.netdev_hostname = netdev_hostname + self.netdev_vendor = netdev_vendor + self.netdev_model = netdev_model + self.netdev_serial_number = netdev_serial_number + self.netdev_mgmt_ifname = netdev_mgmt_ifname + self.netdev_mgmt_pflen = netdev_mgmt_pflen + self.netdev_netmiko_device_type = netdev_netmiko_device_type + + self.onboarding_class = onboarding_class + self.driver_addon_result = driver_addon_result + + # these attributes are netbox model instances as discovered/created + # through the course of processing. + self.nb_site = None + self.nb_manufacturer = None + self.nb_device_type = None + self.nb_device_role = None + self.nb_platform = None + + self.device = None + self.onboarded_device = None + self.nb_mgmt_ifname = None + self.nb_primary_ip = None + + def ensure_onboarded_device(self): + """Lookup if the device already exists in the NetBox. + + Lookup is performed by querying for the IP address of the onboarded device. + If the device with a given IP is already in NetBox, its attributes including name could be updated + """ + try: + if self.netdev_mgmt_ip_address: + self.onboarded_device = Device.objects.get(primary_ip4__address__net_host=self.netdev_mgmt_ip_address) + except Device.DoesNotExist: + logger.info( + "Could not find existing NetBox device for requested primary IP address (%s)", + self.netdev_mgmt_ip_address, + ) + except Device.MultipleObjectsReturned: + raise OnboardException( + reason="fail-general", + message=f"ERROR multiple devices using same IP in NetBox: {self.netdev_mgmt_ip_address}", + ) + + def ensure_device_site(self): + """Ensure device's site.""" + try: + self.nb_site = Site.objects.get(slug=self.netdev_nb_site_slug) + except Site.DoesNotExist: + raise OnboardException(reason="fail-config", message=f"Site not found: {self.netdev_nb_site_slug}") + + def ensure_device_manufacturer( + self, + create_manufacturer=PLUGIN_SETTINGS["create_manufacturer_if_missing"], + skip_manufacturer_on_update=PLUGIN_SETTINGS["skip_manufacturer_on_update"], + ): + """Ensure device's manufacturer.""" + # Support to skip manufacturer updates for existing devices + if self.onboarded_device and skip_manufacturer_on_update: + self.nb_manufacturer = self.onboarded_device.device_type.manufacturer + + return + + # First ensure that the vendor, as extracted from the network device exists + # in NetBox. We need the ID for this vendor when ensuring the DeviceType + # instance. + + nb_manufacturer_slug = slugify(self.netdev_vendor) + + try: + search_array = [{"slug__iexact": nb_manufacturer_slug}] + self.nb_manufacturer = object_match(Manufacturer, search_array) + except Manufacturer.DoesNotExist: + if create_manufacturer: + self.nb_manufacturer = Manufacturer.objects.create(name=self.netdev_vendor, slug=nb_manufacturer_slug) + else: + raise OnboardException( + reason="fail-config", message=f"ERROR manufacturer not found: {self.netdev_vendor}" + ) + + def ensure_device_type( + self, + create_device_type=PLUGIN_SETTINGS["create_device_type_if_missing"], + skip_device_type_on_update=PLUGIN_SETTINGS["skip_device_type_on_update"], + ): + """Ensure the Device Type (slug) exists in NetBox associated to the netdev "model" and "vendor" (manufacturer). + + Args: + create_device_type (bool): Flag to indicate if we need to create the device_type, if not already present + skip_device_type_on_update (bool): Flag to indicate if we skip device type updates for existing devices + Raises: + OnboardException('fail-config'): + When the device vendor value does not exist as a Manufacturer in + NetBox. + + OnboardException('fail-config'): + When the device-type exists by slug, but is assigned to a different + manufacturer. This should *not* happen, but guard-rail checking + regardless in case two vendors have the same model name. + """ + # Support to skip device type updates for existing devices + if self.onboarded_device and skip_device_type_on_update: + self.nb_device_type = self.onboarded_device.device_type + + return + + # Now see if the device type (slug) already exists, + # if so check to make sure that it is not assigned as a different manufacturer + # if it doesn't exist, create it if the flag 'create_device_type_if_missing' is defined + + slug = self.netdev_model + if self.netdev_model and re.search(r"[^a-zA-Z0-9\-_]+", slug): + logger.warning("device model is not sluggable: %s", slug) + self.netdev_model = slug.replace(" ", "-") + logger.warning("device model is now: %s", self.netdev_model) + + # Use declared device type or auto-discovered model + nb_device_type_text = self.netdev_nb_device_type_slug or self.netdev_model + + if not nb_device_type_text: + raise OnboardException(reason="fail-config", message="ERROR device type not found") + + nb_device_type_slug = slugify(nb_device_type_text) + + try: + search_array = [ + {"slug__iexact": nb_device_type_slug}, + {"model__iexact": self.netdev_model}, + {"part_number__iexact": self.netdev_model}, + ] + + self.nb_device_type = object_match(DeviceType, search_array) + + if self.nb_device_type.manufacturer.id != self.nb_manufacturer.id: + raise OnboardException( + reason="fail-config", + message=f"ERROR device type {self.netdev_model} " f"already exists for vendor {self.netdev_vendor}", + ) + + except DeviceType.DoesNotExist: + if create_device_type: + logger.info("CREATE: device-type: %s", self.netdev_model) + self.nb_device_type = DeviceType.objects.create( + slug=nb_device_type_slug, model=nb_device_type_slug.upper(), manufacturer=self.nb_manufacturer, + ) + else: + raise OnboardException( + reason="fail-config", message=f"ERROR device type not found: {self.netdev_model}" + ) + + def ensure_device_role( + self, create_device_role=PLUGIN_SETTINGS["create_device_role_if_missing"], + ): + """Ensure that the device role is defined / exist in NetBox or create it if it doesn't exist. + + Args: + create_device_role (bool) :Flag to indicate if we need to create the device_role, if not already present + Raises: + OnboardException('fail-config'): + When the device role value does not exist + NetBox. + """ + try: + self.nb_device_role = DeviceRole.objects.get(slug=self.netdev_nb_role_slug) + except DeviceRole.DoesNotExist: + if create_device_role: + self.nb_device_role = DeviceRole.objects.create( + name=self.netdev_nb_role_slug, + slug=self.netdev_nb_role_slug, + color=self.netdev_nb_role_color, + vm_role=False, + ) + else: + raise OnboardException( + reason="fail-config", message=f"ERROR device role not found: {self.netdev_nb_role_slug}" + ) + + def ensure_device_platform(self, create_platform_if_missing=PLUGIN_SETTINGS["create_platform_if_missing"]): + """Get platform object from NetBox filtered by platform_slug. + + Args: + platform_slug (string): slug of a platform object present in NetBox, object will be created if not present + and create_platform_if_missing is enabled + + Return: + dcim.models.Platform object + + Raises: + OnboardException + + Lookup is performed based on the object's slug field (not the name field) + """ + try: + self.netdev_nb_platform_slug = ( + self.netdev_nb_platform_slug + or PLUGIN_SETTINGS["platform_map"].get(self.netdev_netmiko_device_type) + or self.netdev_netmiko_device_type + ) + + if not self.netdev_nb_platform_slug: + raise OnboardException( + reason="fail-config", message=f"ERROR device platform not found: {self.netdev_hostname}" + ) + + self.nb_platform = Platform.objects.get(slug=self.netdev_nb_platform_slug) + + logger.info("PLATFORM: found in NetBox %s", self.netdev_nb_platform_slug) + + except Platform.DoesNotExist: + if create_platform_if_missing: + platform_to_napalm_netbox = { + platform.slug: platform.napalm_driver + for platform in Platform.objects.all() + if platform.napalm_driver + } + + # Update Constants if Napalm driver is defined for NetBox Platform + netmiko_to_napalm = {**NETMIKO_TO_NAPALM_STATIC, **platform_to_napalm_netbox} + + self.nb_platform = Platform.objects.create( + name=self.netdev_nb_platform_slug, + slug=self.netdev_nb_platform_slug, + napalm_driver=netmiko_to_napalm[self.netdev_netmiko_device_type], + ) + else: + raise OnboardException( + reason="fail-general", message=f"ERROR platform not found in NetBox: {self.netdev_nb_platform_slug}" + ) + + def ensure_device_instance(self, default_status=PLUGIN_SETTINGS["default_device_status"]): + """Ensure that the device instance exists in NetBox and is assigned the provided device role or DEFAULT_ROLE. + + Args: + default_status (str) : status assigned to a new device by default. + """ + if self.onboarded_device: + # Construct lookup arguments if onboarded device already exists in NetBox + + logger.info( + "Found existing NetBox device (%s) for requested primary IP address (%s)", + self.onboarded_device.name, + self.netdev_mgmt_ip_address, + ) + lookup_args = { + "pk": self.onboarded_device.pk, + "defaults": dict( + name=self.netdev_hostname, + device_type=self.nb_device_type, + device_role=self.nb_device_role, + platform=self.nb_platform, + site=self.nb_site, + serial=self.netdev_serial_number, + # status= field is not updated in case of already existing devices to prevent changes + ), + } + else: + # Construct lookup arguments if onboarded device does not exist in NetBox + + lookup_args = { + "name": self.netdev_hostname, + "defaults": dict( + device_type=self.nb_device_type, + device_role=self.nb_device_role, + platform=self.nb_platform, + site=self.nb_site, + serial=self.netdev_serial_number, + # status= defined only for new devices, no update for existing should occur + status=default_status, + ), + } + + try: + self.device, created = Device.objects.update_or_create(**lookup_args) + + if created: + logger.info("CREATED device: %s", self.netdev_hostname) + else: + logger.info("GOT/UPDATED device: %s", self.netdev_hostname) + + except Device.MultipleObjectsReturned: + raise OnboardException( + reason="fail-general", + message=f"ERROR multiple devices using same name in NetBox: {self.netdev_hostname}", + ) + + def ensure_interface(self): + """Ensures that the interface associated with the mgmt_ipaddr exists and is assigned to the device.""" + self.nb_mgmt_ifname, _ = Interface.objects.get_or_create(name=self.netdev_mgmt_ifname, device=self.device) + + def ensure_primary_ip(self): + """Ensure mgmt_ipaddr exists in IPAM, has the device interface, and is assigned as the primary IP address.""" + # see if the primary IP address exists in IPAM + self.nb_primary_ip, created = IPAddress.objects.get_or_create( + address=f"{self.netdev_mgmt_ip_address}/{self.netdev_mgmt_pflen}" + ) + + if created or not self.nb_primary_ip in self.nb_mgmt_ifname.ip_addresses.all(): + logger.info("ASSIGN: IP address %s to %s", self.nb_primary_ip.address, self.nb_mgmt_ifname.name) + self.nb_mgmt_ifname.ip_addresses.add(self.nb_primary_ip) + self.nb_mgmt_ifname.save() + + # Ensure the primary IP is assigned to the device + self.device.primary_ip4 = self.nb_primary_ip + self.device.save() + + def ensure_device(self): + """Ensure that the device represented by the DevNetKeeper exists in the NetBox system.""" + self.ensure_onboarded_device() + self.ensure_device_site() + self.ensure_device_manufacturer() + self.ensure_device_type() + self.ensure_device_role() + self.ensure_device_platform() + self.ensure_device_instance() + + if PLUGIN_SETTINGS["create_management_interface_if_missing"]: + self.ensure_interface() + self.ensure_primary_ip() diff --git a/netbox_onboarding/netdev_keeper.py b/netbox_onboarding/netdev_keeper.py new file mode 100644 index 0000000..07fb1b8 --- /dev/null +++ b/netbox_onboarding/netdev_keeper.py @@ -0,0 +1,312 @@ +"""NetDev Keeper. + +(c) 2020 Network To Code +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +import importlib +import logging +import socket + +import netaddr +from django.conf import settings +from napalm import get_network_driver +from napalm.base.exceptions import ConnectionException, CommandErrorException +from netaddr.core import AddrFormatError +from netmiko.ssh_autodetect import SSHDetect +from netmiko.ssh_exception import NetMikoAuthenticationException +from netmiko.ssh_exception import NetMikoTimeoutException +from paramiko.ssh_exception import SSHException + +from dcim.models import Platform + +from netbox_onboarding.onboarding.onboarding import StandaloneOnboarding +from .constants import NETMIKO_TO_NAPALM_STATIC +from .exceptions import OnboardException + +logger = logging.getLogger("rq.worker") + +PLUGIN_SETTINGS = settings.PLUGINS_CONFIG["netbox_onboarding"] + + +def get_mgmt_info( + hostname, + ip_ifs, + default_mgmt_if=PLUGIN_SETTINGS["default_management_interface"], + default_mgmt_pfxlen=PLUGIN_SETTINGS["default_management_prefix_length"], +): + """Get the interface name and prefix length for the management interface. + + Locate the interface assigned with the hostname value and retain + the interface name and IP prefix-length so that we can use it + when creating the IPAM IP-Address instance. + + Note that in some cases (e.g., NAT) the hostname may differ than + the interface addresses present on the device. We need to handle this. + """ + for if_name, if_data in ip_ifs.items(): + for if_addr, if_addr_data in if_data["ipv4"].items(): + if if_addr == hostname: + return if_name, if_addr_data["prefix_length"] + + return default_mgmt_if, default_mgmt_pfxlen + + +class NetdevKeeper: + """Used to maintain information about the network device during the onboarding process.""" + + def __init__( # pylint: disable=R0913 + self, hostname, port=None, timeout=None, username=None, password=None, secret=None, napalm_driver=None + ): + """Initialize the network device keeper instance and ensure the required configuration parameters are provided. + + Args: + hostname (str): IP Address or FQDN of an onboarded device + port (int): Port used to connect to an onboarded device + timeout (int): Connection timeout of an onboarded device + username (str): Device username (if unspecified, NAPALM_USERNAME settings variable will be used) + password (str): Device password (if unspecified, NAPALM_PASSWORD settings variable will be used) + secret (str): Device secret password (if unspecified, NAPALM_ARGS["secret"] settings variable will be used) + napalm_driver (str): Napalm driver name to use to onboard network device + + Raises: + OnboardException('fail-config'): + When any required config options are missing. + """ + # Attributes + self.hostname = hostname + self.port = port + self.timeout = timeout + self.username = username or settings.NAPALM_USERNAME + self.password = password or settings.NAPALM_PASSWORD + self.secret = secret or settings.NAPALM_ARGS.get("secret", None) + self.napalm_driver = napalm_driver + + self.facts = None + self.ip_ifs = None + self.netmiko_device_type = None + self.onboarding_class = StandaloneOnboarding + self.driver_addon_result = None + + def check_ip(self): + """Method to check if the IP address form field was an IP address. + + If it is a DNS name, attempt to resolve the DNS address and assign the IP address to the + name. + + Returns: + (bool): True if the IP address is an IP address, or a DNS entry was found and + reassignment of the ot.ip_address was done. + False if unable to find a device IP (error) + + Raises: + OnboardException("fail-general"): + When a prefix was entered for an IP address + OnboardException("fail-dns"): + When a Name lookup via DNS fails to resolve an IP address + """ + try: + # Assign checked_ip to None for error handling + # If successful, this is an IP address and can pass + checked_ip = netaddr.IPAddress(self.hostname) + return True + # Catch when someone has put in a prefix address, raise an exception + except ValueError: + raise OnboardException( + reason="fail-general", message=f"ERROR appears a prefix was entered: {self.hostname}" + ) + # An AddrFormatError exception means that there is not an IP address in the field, and should continue on + except AddrFormatError: + try: + # Do a lookup of name to get the IP address to connect to + checked_ip = socket.gethostbyname(self.hostname) + self.hostname = checked_ip + return True + except socket.gaierror: + # DNS Lookup has failed, Raise an exception for unable to complete DNS lookup + raise OnboardException( + reason="fail-dns", message=f"ERROR failed to complete DNS lookup: {self.hostname}" + ) + + def check_reachability(self): + """Ensure that the device at the mgmt-ipaddr provided is reachable. + + We do this check before attempting other "show" commands so that we know we've got a + device that can be reached. + + Raises: + OnboardException('fail-connect'): + When device unreachable + """ + logger.info("CHECK: IP %s:%s", self.hostname, self.port) + + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(self.timeout) + sock.connect((self.hostname, self.port)) + + except (socket.error, socket.timeout, ConnectionError): + raise OnboardException( + reason="fail-connect", message=f"ERROR device unreachable: {self.hostname}:{self.port}" + ) + + def guess_netmiko_device_type(self): + """Guess the device type of host, based on Netmiko.""" + guessed_device_type = None + + remote_device = { + "device_type": "autodetect", + "host": self.hostname, + "username": self.username, + "password": self.password, + "secret": self.secret, + } + + try: + logger.info("INFO guessing device type: %s", self.hostname) + guesser = SSHDetect(**remote_device) + guessed_device_type = guesser.autodetect() + logger.info("INFO guessed device type: %s", guessed_device_type) + + except NetMikoAuthenticationException as err: + logger.error("ERROR %s", err) + raise OnboardException(reason="fail-login", message=f"ERROR: {str(err)}") + + except (NetMikoTimeoutException, SSHException) as err: + logger.error("ERROR: %s", str(err)) + raise OnboardException(reason="fail-connect", message=f"ERROR: {str(err)}") + + except Exception as err: + logger.error("ERROR: %s", str(err)) + raise OnboardException(reason="fail-general", message=f"ERROR: {str(err)}") + + logger.info("INFO device type is: %s", guessed_device_type) + + return guessed_device_type + + def set_napalm_driver_name(self): + """Sets napalm driver name.""" + if not self.napalm_driver: + netmiko_device_type = self.guess_netmiko_device_type() + logger.info("Guessed Netmiko Device Type: %s", netmiko_device_type) + + self.netmiko_device_type = netmiko_device_type + + platform_to_napalm_netbox = { + platform.slug: platform.napalm_driver for platform in Platform.objects.all() if platform.napalm_driver + } + + # Update Constants if Napalm driver is defined for NetBox Platform + netmiko_to_napalm = {**NETMIKO_TO_NAPALM_STATIC, **platform_to_napalm_netbox} + + self.napalm_driver = netmiko_to_napalm.get(netmiko_device_type) + + def check_napalm_driver_name(self): + """Checks for napalm driver name.""" + if not self.napalm_driver: + raise OnboardException( + reason="fail-general", + message=f"Onboarding for Platform {self.netmiko_device_type} not " + f"supported, as it has no specified NAPALM driver", + ) + + def get_onboarding_facts(self): + """Gather information from the network device that is needed to onboard the device into the NetBox system. + + Raises: + OnboardException('fail-login'): + When unable to login to device + + OnboardException('fail-execute'): + When unable to run commands to collect device information + + OnboardException('fail-general'): + Any other unexpected device comms failure. + """ + # Check to see if the IP address entered was an IP address or a DNS entry, get the IP address + self.check_ip() + + self.check_reachability() + + logger.info("COLLECT: device information %s", self.hostname) + + try: + # Get Napalm Driver with Netmiko if needed + self.set_napalm_driver_name() + + # Raise if no Napalm Driver not selected + self.check_napalm_driver_name() + + driver = get_network_driver(self.napalm_driver) + optional_args = settings.NAPALM_ARGS.copy() + optional_args["secret"] = self.secret + + napalm_device = driver( + hostname=self.hostname, + username=self.username, + password=self.password, + timeout=self.timeout, + optional_args=optional_args, + ) + + napalm_device.open() + + logger.info("COLLECT: device facts") + self.facts = napalm_device.get_facts() + + logger.info("COLLECT: device interface IPs") + self.ip_ifs = napalm_device.get_interfaces_ip() + + module_name = PLUGIN_SETTINGS["onboarding_extensions_map"].get(self.napalm_driver) + + if module_name: + try: + module = importlib.import_module(module_name) + driver_addon_class = module.OnboardingDriverExtensions(napalm_device=napalm_device) + self.onboarding_class = driver_addon_class.onboarding_class + self.driver_addon_result = driver_addon_class.ext_result + except ModuleNotFoundError as exc: + raise OnboardException( + reason="fail-general", + message=f"ERROR: ModuleNotFoundError: Onboarding extension for napalm driver {self.napalm_driver} configured but can not be imported per configuration", + ) + except ImportError as exc: + raise OnboardException(reason="fail-general", message="ERROR: ImportError: %s" % exc.args[0]) + else: + logger.info( + "INFO: No onboarding extension defined for napalm driver %s, using default napalm driver", + self.napalm_driver, + ) + + except ConnectionException as exc: + raise OnboardException(reason="fail-login", message=exc.args[0]) + + except CommandErrorException as exc: + raise OnboardException(reason="fail-execute", message=exc.args[0]) + + except Exception as exc: + raise OnboardException(reason="fail-general", message=str(exc)) + + def get_netdev_dict(self): + """Construct network device dict.""" + netdev_dict = { + "netdev_hostname": self.facts["hostname"], + "netdev_vendor": self.facts["vendor"].title(), + "netdev_model": self.facts["model"].lower(), + "netdev_serial_number": self.facts["serial_number"], + "netdev_mgmt_ifname": get_mgmt_info(hostname=self.hostname, ip_ifs=self.ip_ifs)[0], + "netdev_mgmt_pflen": get_mgmt_info(hostname=self.hostname, ip_ifs=self.ip_ifs)[1], + "netdev_netmiko_device_type": self.netmiko_device_type, + "onboarding_class": self.onboarding_class, + "driver_addon_result": self.driver_addon_result, + } + + return netdev_dict diff --git a/netbox_onboarding/onboard.py b/netbox_onboarding/onboard.py index fe63c2a..cbe68c0 100644 --- a/netbox_onboarding/onboard.py +++ b/netbox_onboarding/onboard.py @@ -1,4 +1,4 @@ -"""Worker code for processing inbound OnboardingTasks. +"""Onboard. (c) 2020 Network To Code Licensed under the Apache License, Version 2.0 (the "License"); @@ -12,562 +12,110 @@ limitations under the License. """ -import logging -import re -import socket - -from napalm import get_network_driver -from napalm.base.exceptions import ConnectionException, CommandErrorException -import netaddr -from netaddr.core import AddrFormatError - from django.conf import settings -from django.utils.text import slugify - -from netmiko.ssh_autodetect import SSHDetect -from netmiko.ssh_exception import NetMikoAuthenticationException -from netmiko.ssh_exception import NetMikoTimeoutException -from paramiko.ssh_exception import SSHException - -from dcim.models import Manufacturer, Device, Interface, DeviceType, Platform, DeviceRole -from ipam.models import IPAddress -from .constants import NETMIKO_TO_NAPALM - -__all__ = [] +from .netdev_keeper import NetdevKeeper PLUGIN_SETTINGS = settings.PLUGINS_CONFIG["netbox_onboarding"] -class OnboardException(Exception): - """A failure occurred during the onboarding process. - - The exception includes a reason "slug" as defined below as well as a humanized message. - """ - - REASONS = ( - "fail-config", # config provided is not valid - "fail-connect", # device is unreachable at IP:PORT - "fail-execute", # unable to execute device/API command - "fail-login", # bad username/password - "fail-dns", # failed to get IP address from name resolution - "fail-general", # other error - ) - - def __init__(self, reason, message, **kwargs): - super(OnboardException, self).__init__(kwargs) - self.reason = reason - self.message = message - - def __str__(self): - return f"{self.__class__.__name__}: {self.reason}: {self.message}" - - -# ----------------------------------------------------------------------------- -# -# Network Device Keeper -# -# ----------------------------------------------------------------------------- - - -class NetdevKeeper: - """Used to maintain information about the network device during the onboarding process.""" - - def __init__(self, onboarding_task, username=None, password=None, secret=None): - """Initialize the network device keeper instance and ensure the required configuration parameters are provided. - - Args: - onboarding_task (OnboardingTask): Task being processed - username (str): Device username (if unspecified, NAPALM_USERNAME settings variable will be used) - password (str): Device password (if unspecified, NAPALM_PASSWORD settings variable will be used) - secret (str): Device secret password (if unspecified, NAPALM_ARGS["secret"] settings variable will be used) - - Raises: - OnboardException('fail-config'): - When any required config options are missing. - """ - self.ot = onboarding_task - - # Attributes that are set when reading info from device - - self.hostname = None - self.vendor = None - self.model = None - self.serial_number = None - self.mgmt_ifname = None - self.mgmt_pflen = None - self.username = username or settings.NAPALM_USERNAME - self.password = password or settings.NAPALM_PASSWORD - self.secret = secret or settings.NAPALM_ARGS.get("secret", None) - - def check_reachability(self): - """Ensure that the device at the mgmt-ipaddr provided is reachable. - - We do this check before attempting other "show" commands so that we know we've got a - device that can be reached. - - Raises: - OnboardException('fail-connect'): - When device unreachable - """ - ip_addr = self.ot.ip_address - port = self.ot.port - timeout = self.ot.timeout - - logging.info("CHECK: IP %s:%s", ip_addr, port) - - try: - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.settimeout(timeout) - sock.connect((ip_addr, port)) - - except (socket.error, socket.timeout, ConnectionError): - raise OnboardException(reason="fail-connect", message=f"ERROR device unreachable: {ip_addr}:{port}") - - @staticmethod - def check_netmiko_conversion(guessed_device_type, platform_map=None): - """Method to convert Netmiko device type into the mapped type if defined in the settings file. - - Args: - guessed_device_type (string): Netmiko device type guessed platform - test_platform_map (dict): Platform Map for use in testing - - Returns: - string: Platform name - """ - # If this is defined, process the mapping - if platform_map: - # Attempt to get a mapped slug. If there is no slug, return the guessed_device_type as the slug - return platform_map.get(guessed_device_type, guessed_device_type) - - # There is no mapping configured, return what was brought in - return guessed_device_type - - def guess_netmiko_device_type(self, **kwargs): - """Guess the device type of host, based on Netmiko.""" - guessed_device_type = None - - remote_device = { - "device_type": "autodetect", - "host": kwargs.get("host"), - "username": kwargs.get("username"), - "password": kwargs.get("password"), - "secret": kwargs.get("secret"), +class OnboardingTaskManager: + """Onboarding Task Manager.""" + + def __init__(self, ot): + """Inits class.""" + self.ot = ot + + @property + def napalm_driver(self): + """Return napalm driver name.""" + if self.ot.platform and self.ot.platform.napalm_driver: + return self.ot.platform.napalm_driver + + return None + + @property + def ip_address(self): + """Return ot's ip address.""" + return self.ot.ip_address + + @property + def port(self): + """Return ot's port.""" + return self.ot.port + + @property + def timeout(self): + """Return ot's timeout.""" + return self.ot.timeout + + @property + def site(self): + """Return ot's site.""" + return self.ot.site + + @property + def device_type(self): + """Return ot's device type.""" + return self.ot.device_type + + @property + def role(self): + """Return it's device role.""" + return self.ot.role + + @property + def platform(self): + """Return ot's device platform.""" + return self.ot.platform + + +class OnboardingManager: + """Onboarding Manager.""" + + def __init__(self, ot, username, password, secret): + """Inits class.""" + self.username = username + self.password = password + self.secret = secret + + # Create instance of Onboarding Task Manager class: + otm = OnboardingTaskManager(ot) + + netdev = NetdevKeeper( + hostname=otm.ip_address, + port=otm.port, + timeout=otm.timeout, + username=self.username, + password=self.password, + secret=self.secret, + napalm_driver=otm.napalm_driver, + ) + + netdev.get_onboarding_facts() + netdev_dict = netdev.get_netdev_dict() + + onboarding_kwargs = { + # Kwargs extracted from OnboardingTask: + "netdev_mgmt_ip_address": otm.ip_address, + "netdev_nb_site_slug": otm.site.slug if otm.site else None, + "netdev_nb_device_type_slug": otm.device_type, + "netdev_nb_role_slug": otm.role.slug if otm.role else PLUGIN_SETTINGS["default_device_role"], + "netdev_nb_role_color": PLUGIN_SETTINGS["default_device_role_color"], + "netdev_nb_platform_slug": otm.platform.slug if otm.platform else None, + # Kwargs discovered on the Onboarded Device: + "netdev_hostname": netdev_dict["netdev_hostname"], + "netdev_vendor": netdev_dict["netdev_vendor"], + "netdev_model": netdev_dict["netdev_model"], + "netdev_serial_number": netdev_dict["netdev_serial_number"], + "netdev_mgmt_ifname": netdev_dict["netdev_mgmt_ifname"], + "netdev_mgmt_pflen": netdev_dict["netdev_mgmt_pflen"], + "netdev_netmiko_device_type": netdev_dict["netdev_netmiko_device_type"], + "onboarding_class": netdev_dict["onboarding_class"], + "driver_addon_result": netdev_dict["driver_addon_result"], } - try: - logging.info("INFO guessing device type: %s", kwargs.get("host")) - guesser = SSHDetect(**remote_device) - guessed_device_type = guesser.autodetect() - logging.info("INFO guessed device type: %s", guessed_device_type) - - except NetMikoAuthenticationException as err: - logging.error("ERROR %s", err) - raise OnboardException(reason="fail-login", message="ERROR {}".format(str(err))) - - except (NetMikoTimeoutException, SSHException) as err: - logging.error("ERROR %s", err) - raise OnboardException(reason="fail-connect", message="ERROR {}".format(str(err))) - - except Exception as err: - logging.error("ERROR %s", err) - raise OnboardException(reason="fail-general", message="ERROR {}".format(str(err))) - - logging.info("INFO device type is %s", guessed_device_type) - - # Get the platform map from the PLUGIN SETTINGS, Return the result of doing a check_netmiko_conversion - return self.check_netmiko_conversion(guessed_device_type, platform_map=PLUGIN_SETTINGS.get("platform_map", {})) - - def get_platform_slug(self): - """Get platform slug in netmiko format (ie cisco_ios, cisco_xr etc).""" - if self.ot.platform: - platform_slug = self.ot.platform.slug - else: - platform_slug = self.guess_netmiko_device_type( - host=self.ot.ip_address, username=self.username, password=self.password, secret=self.secret, - ) - - logging.info("PLATFORM NAME is %s", platform_slug) - - return platform_slug - - @staticmethod - def get_platform_object_from_netbox( - platform_slug, create_platform_if_missing=PLUGIN_SETTINGS["create_platform_if_missing"] - ): - """Get platform object from NetBox filtered by platform_slug. - - Args: - platform_slug (string): slug of a platform object present in NetBox, object will be created if not present - and create_platform_if_missing is enabled - - Return: - dcim.models.Platform object - - Raises: - OnboardException - - Lookup is performed based on the object's slug field (not the name field) - """ - try: - # Get the platform from the NetBox DB - platform = Platform.objects.get(slug=platform_slug) - logging.info("PLATFORM: found in NetBox %s", platform_slug) - except Platform.DoesNotExist: - - if not create_platform_if_missing: - raise OnboardException( - reason="fail-general", message=f"ERROR platform not found in NetBox: {platform_slug}" - ) - - if platform_slug not in NETMIKO_TO_NAPALM.keys(): - raise OnboardException( - reason="fail-general", - message=f"ERROR platform not found in NetBox and it's eligible for auto-creation: {platform_slug}", - ) - - platform = Platform.objects.create( - name=platform_slug, slug=platform_slug, napalm_driver=NETMIKO_TO_NAPALM[platform_slug] - ) - platform.save() - - else: - if not platform.napalm_driver: - raise OnboardException( - reason="fail-general", message=f"ERROR platform is missing the NAPALM Driver: {platform_slug}", - ) - - return platform - - def check_ip(self): - """Method to check if the IP address form field was an IP address. - - If it is a DNS name, attempt to resolve the DNS address and assign the IP address to the - name. - - Returns: - (bool): True if the IP address is an IP address, or a DNS entry was found and - reassignment of the ot.ip_address was done. - False if unable to find a device IP (error) - - Raises: - OnboardException("fail-general"): - When a prefix was entered for an IP address - OnboardException("fail-dns"): - When a Name lookup via DNS fails to resolve an IP address - """ - try: - # Assign checked_ip to None for error handling - # If successful, this is an IP address and can pass - checked_ip = netaddr.IPAddress(self.ot.ip_address) - return True - # Catch when someone has put in a prefix address, raise an exception - except ValueError: - raise OnboardException( - reason="fail-general", message=f"ERROR appears a prefix was entered: {self.ot.ip_address}" - ) - # An AddrFormatError exception means that there is not an IP address in the field, and should continue on - except AddrFormatError: - try: - # Do a lookup of name to get the IP address to connect to - checked_ip = socket.gethostbyname(self.ot.ip_address) - self.ot.ip_address = checked_ip - return True - except socket.gaierror: - # DNS Lookup has failed, Raise an exception for unable to complete DNS lookup - raise OnboardException( - reason="fail-dns", message=f"ERROR failed to complete DNS lookup: {self.ot.ip_address}" - ) - - def get_required_info( - self, - default_mgmt_if=PLUGIN_SETTINGS["default_management_interface"], - default_mgmt_pfxlen=PLUGIN_SETTINGS["default_management_prefix_length"], - ): - """Gather information from the network device that is needed to onboard the device into the NetBox system. - - Raises: - OnboardException('fail-login'): - When unable to login to device - - OnboardException('fail-execute'): - When unable to run commands to collect device information - - OnboardException('fail-general'): - Any other unexpected device comms failure. - """ - # Check to see if the IP address entered was an IP address or a DNS entry, get the IP address - self.check_ip() - self.check_reachability() - mgmt_ipaddr = self.ot.ip_address - - logging.info("COLLECT: device information %s", mgmt_ipaddr) - - try: - platform_slug = self.get_platform_slug() - platform_object = self.get_platform_object_from_netbox(platform_slug=platform_slug) - if self.ot.platform != platform_object: - self.ot.platform = platform_object - self.ot.save() - - driver_name = platform_object.napalm_driver - - if not driver_name: - raise OnboardException( - reason="fail-general", - message=f"Onboarding for Platform {platform_slug} not supported, as it has no specified NAPALM driver", - ) - - driver = get_network_driver(driver_name) - optional_args = settings.NAPALM_ARGS.copy() - optional_args["secret"] = self.secret - dev = driver( - hostname=mgmt_ipaddr, - username=self.username, - password=self.password, - timeout=self.ot.timeout, - optional_args=optional_args, - ) - - dev.open() - logging.info("COLLECT: device facts") - facts = dev.get_facts() - - logging.info("COLLECT: device interface IPs") - ip_ifs = dev.get_interfaces_ip() - - except ConnectionException as exc: - raise OnboardException(reason="fail-login", message=exc.args[0]) - - except CommandErrorException as exc: - raise OnboardException(reason="fail-execute", message=exc.args[0]) - - except Exception as exc: - raise OnboardException(reason="fail-general", message=str(exc)) - - # locate the interface assigned with the mgmt_ipaddr value and retain - # the interface name and IP prefix-length so that we can use it later - # when creating the IPAM IP-Address instance. - # Note that in some cases (e.g., NAT) the mgmt_ipaddr may differ than - # the interface addresses present on the device. We need to handle this. - - def get_mgmt_info(): - """Get the interface name and prefix length for the management interface.""" - for if_name, if_data in ip_ifs.items(): - for if_addr, if_addr_data in if_data["ipv4"].items(): - if if_addr == mgmt_ipaddr: - return (if_name, if_addr_data["prefix_length"]) - return (default_mgmt_if, default_mgmt_pfxlen) - - # retain the attributes that will be later used by NetBox processing. - - self.hostname = facts["hostname"] - self.vendor = facts["vendor"].title() - self.model = facts["model"].lower() - self.serial_number = facts["serial_number"] - self.mgmt_ifname, self.mgmt_pflen = get_mgmt_info() - - -# ----------------------------------------------------------------------------- -# -# NetBox Device Keeper -# -# ----------------------------------------------------------------------------- - - -class NetboxKeeper: - """Used to manage the information relating to the network device within the NetBox server.""" - - def __init__(self, netdev): - """Create an instance and initialize the managed attributes that are used throughout the onboard processing. - - Args: - netdev (NetdevKeeper): instance - """ - self.netdev = netdev - - # these attributes are netbox model instances as discovered/created - # through the course of processing. - - self.manufacturer = None - self.device_type = None - self.device = None - self.interface = None - self.primary_ip = None - - def ensure_device_type( - self, - create_manufacturer=PLUGIN_SETTINGS["create_manufacturer_if_missing"], - create_device_type=PLUGIN_SETTINGS["create_device_type_if_missing"], - ): - """Ensure the Device Type (slug) exists in NetBox associated to the netdev "model" and "vendor" (manufacturer). - - Args: - create_manufacturer (bool) :Flag to indicate if we need to create the manufacturer, if not already present - create_device_type (bool): Flag to indicate if we need to create the device_type, if not already present - Raises: - OnboardException('fail-config'): - When the device vendor value does not exist as a Manufacturer in - NetBox. - - OnboardException('fail-config'): - When the device-type exists by slug, but is assigned to a different - manufacturer. This should *not* happen, but guard-rail checking - regardless in case two vendors have the same model name. - """ - # First ensure that the vendor, as extracted from the network device exists - # in NetBox. We need the ID for this vendor when ensuring the DeviceType - # instance. - - try: - self.manufacturer = Manufacturer.objects.get(slug=slugify(self.netdev.vendor)) - except Manufacturer.DoesNotExist: - if not create_manufacturer: - raise OnboardException( - reason="fail-config", message=f"ERROR manufacturer not found: {self.netdev.vendor}" - ) - - self.manufacturer = Manufacturer.objects.create(name=self.netdev.vendor, slug=slugify(self.netdev.vendor)) - self.manufacturer.save() - - # Now see if the device type (slug) already exists, - # if so check to make sure that it is not assigned as a different manufacturer - # if it doesn't exist, create it if the flag 'create_device_type_if_missing' is defined - - slug = self.netdev.model - if re.search(r"[^a-zA-Z0-9\-_]+", slug): - logging.warning("device model is not sluggable: %s", slug) - self.netdev.model = slug.replace(" ", "-") - logging.warning("device model is now: %s", self.netdev.model) - - try: - self.device_type = DeviceType.objects.get(slug=slugify(self.netdev.model)) - self.netdev.ot.device_type = self.device_type.slug - self.netdev.ot.save() - except DeviceType.DoesNotExist: - if not create_device_type: - raise OnboardException( - reason="fail-config", message=f"ERROR device type not found: {self.netdev.model}" - ) - - logging.info("CREATE: device-type: %s", self.netdev.model) - self.device_type = DeviceType.objects.create( - slug=slugify(self.netdev.model), model=self.netdev.model.upper(), manufacturer=self.manufacturer - ) - self.device_type.save() - self.netdev.ot.device_type = self.device_type.slug - self.netdev.ot.save() - return - - if self.device_type.manufacturer.id != self.manufacturer.id: - raise OnboardException( - reason="fail-config", - message=f"ERROR device type {self.netdev.model} already exists for vendor {self.netdev.vendor}", - ) - - def ensure_device_role( - self, - create_device_role=PLUGIN_SETTINGS["create_device_role_if_missing"], - default_device_role=PLUGIN_SETTINGS["default_device_role"], - default_device_role_color=PLUGIN_SETTINGS["default_device_role_color"], - ): - """Ensure that the device role is defined / exist in NetBox or create it if it doesn't exist. - - Args: - create_device_role (bool) :Flag to indicate if we need to create the device_role, if not already present - default_device_role (str): Default value for the device_role, if we need to create it - default_device_role_color (str): Default color to assign to the device_role, if we need to create it - Raises: - OnboardException('fail-config'): - When the device role value does not exist - NetBox. - """ - if self.netdev.ot.role: - return - - try: - device_role = DeviceRole.objects.get(slug=slugify(default_device_role)) - except DeviceRole.DoesNotExist: - if not create_device_role: - raise OnboardException( - reason="fail-config", message=f"ERROR device role not found: {default_device_role}" - ) - - device_role = DeviceRole.objects.create( - name=default_device_role, - slug=slugify(default_device_role), - color=default_device_role_color, - vm_role=False, - ) - device_role.save() - - self.netdev.ot.role = device_role - self.netdev.ot.save() - return - - def ensure_device_instance(self, default_status=PLUGIN_SETTINGS["default_device_status"]): - """Ensure that the device instance exists in NetBox and is assigned the provided device role or DEFAULT_ROLE. - - Args: - default_status (str) : status assigned to a new device by default. - """ - try: - device = Device.objects.get(name=self.netdev.hostname, site=self.netdev.ot.site) - except Device.DoesNotExist: - device = Device.objects.create( - name=self.netdev.hostname, - site=self.netdev.ot.site, - device_type=self.device_type, - device_role=self.netdev.ot.role, - status=default_status, - ) - - device.platform = self.netdev.ot.platform - device.serial = self.netdev.serial_number - device.save() - - self.netdev.ot.created_device = device - self.netdev.ot.save() - - self.device = device - - def ensure_interface(self): - """Ensure that the interface associated with the mgmt_ipaddr exists and is assigned to the device.""" - self.interface, _ = Interface.objects.get_or_create(name=self.netdev.mgmt_ifname, device=self.device) - - def ensure_primary_ip(self): - """Ensure mgmt_ipaddr exists in IPAM, has the device interface, and is assigned as the primary IP address.""" - mgmt_ipaddr = self.netdev.ot.ip_address - - # see if the primary IP address exists in IPAM - self.primary_ip, created = IPAddress.objects.get_or_create(address=f"{mgmt_ipaddr}/{self.netdev.mgmt_pflen}") - - if created or not self.primary_ip.interface: - logging.info("ASSIGN: IP address %s to %s", self.primary_ip.address, self.interface.name) - self.primary_ip.interface = self.interface - - self.primary_ip.save() - - # Ensure the primary IP is assigned to the device - self.device.primary_ip4 = self.primary_ip - self.device.save() - - def check_if_device_already_exist(self): - """Check if a device with the same name / site already exist in the database.""" - try: - Device.objects.get(name=self.netdev.hostname, site=self.netdev.ot.site) - return True - except Device.DoesNotExist: - return False - - def ensure_device(self): - """Ensure that the device represented by the DevNetKeeper exists in the NetBox system.""" - # Only check the device role and device type if the device do not exist already - if not self.check_if_device_already_exist(): - self.ensure_device_type() - self.ensure_device_role() + onboarding_cls = netdev_dict["onboarding_class"]() + onboarding_cls.run(onboarding_kwargs=onboarding_kwargs) - self.ensure_device_instance() - if PLUGIN_SETTINGS["create_management_interface_if_missing"]: - self.ensure_interface() - self.ensure_primary_ip() + self.created_device = onboarding_cls.created_device diff --git a/netbox_onboarding/onboarding/__init__.py b/netbox_onboarding/onboarding/__init__.py new file mode 100644 index 0000000..b8dce31 --- /dev/null +++ b/netbox_onboarding/onboarding/__init__.py @@ -0,0 +1,13 @@ +"""Onboarding. + +(c) 2020 Network To Code +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" diff --git a/netbox_onboarding/onboarding/onboarding.py b/netbox_onboarding/onboarding/onboarding.py new file mode 100644 index 0000000..763e21b --- /dev/null +++ b/netbox_onboarding/onboarding/onboarding.py @@ -0,0 +1,38 @@ +"""Onboarding module. + +(c) 2020 Network To Code +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +from netbox_onboarding.netbox_keeper import NetboxKeeper + + +class Onboarding: + """Generic onboarding class.""" + + def __init__(self): + """Init the class.""" + self.created_device = None + + def run(self, onboarding_kwargs): + """Implement run method.""" + raise NotImplementedError + + +class StandaloneOnboarding(Onboarding): + """Standalone onboarding class.""" + + def run(self, onboarding_kwargs): + """Ensure device is created with NetBox Keeper.""" + nb_k = NetboxKeeper(**onboarding_kwargs) + nb_k.ensure_device() + + self.created_device = nb_k.device diff --git a/netbox_onboarding/onboarding_extensions/__init__.py b/netbox_onboarding/onboarding_extensions/__init__.py new file mode 100644 index 0000000..8df2253 --- /dev/null +++ b/netbox_onboarding/onboarding_extensions/__init__.py @@ -0,0 +1,13 @@ +"""Onboarding Extensions. + +(c) 2020 Network To Code +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" diff --git a/netbox_onboarding/onboarding_extensions/ios.py b/netbox_onboarding/onboarding_extensions/ios.py new file mode 100644 index 0000000..f3dfedc --- /dev/null +++ b/netbox_onboarding/onboarding_extensions/ios.py @@ -0,0 +1,45 @@ +"""Onboarding Extension for IOS. + +(c) 2020 Network To Code +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +from netbox_onboarding.onboarding.onboarding import StandaloneOnboarding + + +class OnboardingDriverExtensions: + """Onboarding Driver's Extensions.""" + + def __init__(self, napalm_device): + """Initialize class.""" + self.napalm_device = napalm_device + + @property + def onboarding_class(self): + """Return onboarding class for IOS driver. + + Currently supported is Standalone Onboarding Process. + + Result of this method is used by the OnboardingManager to + initiate the instance of the onboarding class. + """ + return StandaloneOnboarding + + @property + def ext_result(self): + """This method is used to store any object as a return value. + + Result of this method is passed to the onboarding class as + driver_addon_result argument. + + :return: Any() + """ + return None diff --git a/netbox_onboarding/release.py b/netbox_onboarding/release.py new file mode 100644 index 0000000..648f8f3 --- /dev/null +++ b/netbox_onboarding/release.py @@ -0,0 +1,20 @@ +"""Release variables of the NetBox. + +(c) 2020 Network To Code +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +from packaging import version +from django.conf import settings + +NETBOX_RELEASE_CURRENT = version.parse(settings.VERSION) +NETBOX_RELEASE_28 = version.parse("2.8") +NETBOX_RELEASE_29 = version.parse("2.9") diff --git a/netbox_onboarding/tables.py b/netbox_onboarding/tables.py index a70b056..ce69326 100644 --- a/netbox_onboarding/tables.py +++ b/netbox_onboarding/tables.py @@ -20,6 +20,7 @@ class OnboardingTaskTable(BaseTable): """Table for displaying OnboardingTask instances.""" pk = ToggleColumn() + id = tables.LinkColumn() site = tables.LinkColumn() platform = tables.LinkColumn() created_device = tables.LinkColumn() @@ -28,7 +29,8 @@ class Meta(BaseTable.Meta): # noqa: D106 "Missing docstring in public nested cl model = OnboardingTask fields = ( "pk", - "created_on", + "id", + "created", "ip_address", "site", "platform", @@ -48,7 +50,7 @@ class Meta(BaseTable.Meta): # noqa: D106 "Missing docstring in public nested cl model = OnboardingTask fields = ( "id", - "created_on", + "created", "site", "platform", "ip_address", diff --git a/netbox_onboarding/template_content.py b/netbox_onboarding/template_content.py new file mode 100644 index 0000000..4b3189a --- /dev/null +++ b/netbox_onboarding/template_content.py @@ -0,0 +1,47 @@ +"""Onboarding template content. + +(c) 2020 Network To Code +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +from extras.plugins import PluginTemplateExtension +from .models import OnboardingDevice + + +class DeviceContent(PluginTemplateExtension): # pylint: disable=abstract-method + """Table to show onboarding details on Device objects.""" + + model = "dcim.device" + + def right_page(self): + """Show table on right side of view.""" + onboarding = OnboardingDevice.objects.filter(device=self.context["object"]).first() + + if not onboarding or not onboarding.enabled: + return "" + + status = onboarding.status + last_check_attempt_date = onboarding.last_check_attempt_date + last_check_successful_date = onboarding.last_check_successful_date + last_ot = onboarding.last_ot + + return self.render( + "netbox_onboarding/device_onboarding_table.html", + extra_context={ + "status": status, + "last_check_attempt_date": last_check_attempt_date, + "last_check_successful_date": last_check_successful_date, + "last_ot": last_ot, + }, + ) + + +template_extensions = [DeviceContent] diff --git a/netbox_onboarding/templates/netbox_onboarding/device_onboarding_table.html b/netbox_onboarding/templates/netbox_onboarding/device_onboarding_table.html new file mode 100644 index 0000000..2a20488 --- /dev/null +++ b/netbox_onboarding/templates/netbox_onboarding/device_onboarding_table.html @@ -0,0 +1,31 @@ +{% block content %} +
+
+ Device Onboarding +
+ + + + + + + + + + + + + + + +
DateStatusDate of last successLatest Task
+ {{ last_check_attempt_date }} + + {{ status }} + + {{ last_check_successful_date }} + + {{ last_ot.pk }} +
+
+{% endblock %} diff --git a/netbox_onboarding/templates/netbox_onboarding/onboarding_tasks_list.html b/netbox_onboarding/templates/netbox_onboarding/onboarding_tasks_list.html index 6d1ab9c..3850fbb 100644 --- a/netbox_onboarding/templates/netbox_onboarding/onboarding_tasks_list.html +++ b/netbox_onboarding/templates/netbox_onboarding/onboarding_tasks_list.html @@ -4,14 +4,14 @@ {% block content %}
{% if permissions.add %} - {% add_button 'plugins:netbox_onboarding:onboarding_task_add' %} - {% import_button 'plugins:netbox_onboarding:onboarding_task_import' %} + {% add_button 'plugins:netbox_onboarding:onboardingtask_add' %} + {% import_button 'plugins:netbox_onboarding:onboardingtask_import' %} {% endif %}

{% block title %}Onboarding Tasks{% endblock %}

- {% include 'utilities/obj_table.html' with bulk_delete_url="plugins:netbox_onboarding:onboarding_task_bulk_delete" %} + {% include 'utilities/obj_table.html' with bulk_delete_url="plugins:netbox_onboarding:onboardingtask_bulk_delete" %}
{% include 'inc/search_panel.html' %} diff --git a/netbox_onboarding/templates/netbox_onboarding/onboardingtask.html b/netbox_onboarding/templates/netbox_onboarding/onboardingtask.html new file mode 100644 index 0000000..d638540 --- /dev/null +++ b/netbox_onboarding/templates/netbox_onboarding/onboardingtask.html @@ -0,0 +1,93 @@ +{% extends 'base.html' %} +{% load helpers %} +{% load static %} + +{% block header %} +
+
+
    +
  1. Onboarding tasks
    2. +
  2. {{ onboardingtask.pk }}
    3. +
+
+
+ +

{% block title %}Device: {{ onboardingtask.ip_address }}{% endblock %}

+ + +{% endblock %} + +{% block content %} +
+
+
+
+ Onboarding Task +
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
Created Device{{ onboardingtask.created_device|placeholder }}
IP Address{{ onboardingtask.ip_address|placeholder }}
Port{{ onboardingtask.port|placeholder }}
Timeout{{ onboardingtask.timeout|placeholder }}
Site{{ onboardingtask.site|placeholder }}
Role{{ onboardingtask.role|placeholder }}
Device Type{{ onboardingtask.device_type|placeholder }}
Platform{{ onboardingtask.platform|placeholder }}
Status{{ onboardingtask.status|placeholder }}
Failed Reason{{ onboardingtask.failed_reason|placeholder }}
Message{{ onboardingtask.message|placeholder }}
Created{{ onboardingtask.created|placeholder }}
+
+
+
+{% endblock %} + +{% block javascript %} + +{% endblock %} diff --git a/netbox_onboarding/tests/test_models.py b/netbox_onboarding/tests/test_models.py new file mode 100644 index 0000000..900e445 --- /dev/null +++ b/netbox_onboarding/tests/test_models.py @@ -0,0 +1,97 @@ +"""Unit tests for netbox_onboarding OnboardingDevice model. + +(c) 2020 Network To Code +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" +from django.test import TestCase + +from dcim.models import Site, DeviceRole, DeviceType, Manufacturer, Device, Interface +from ipam.models import IPAddress + +from netbox_onboarding.models import OnboardingTask +from netbox_onboarding.models import OnboardingDevice +from netbox_onboarding.choices import OnboardingStatusChoices + + +class OnboardingDeviceModelTestCase(TestCase): + """Test the Onboarding models.""" + + def setUp(self): + """Setup objects for Onboarding Model tests.""" + self.site = Site.objects.create(name="USWEST", slug="uswest") + manufacturer = Manufacturer.objects.create(name="Juniper", slug="juniper") + device_role = DeviceRole.objects.create(name="Firewall", slug="firewall") + device_type = DeviceType.objects.create(slug="srx3600", model="SRX3600", manufacturer=manufacturer) + + self.device = Device.objects.create( + device_type=device_type, name="device1", device_role=device_role, site=self.site, + ) + + intf = Interface.objects.create(name="test_intf", device=self.device) + + primary_ip = IPAddress.objects.create(address="10.10.10.10/32") + intf.ip_addresses.add(primary_ip) + + self.device.primary_ip4 = primary_ip + self.device.save() + + self.succeeded_task1 = OnboardingTask.objects.create( + ip_address="10.10.10.10", + site=self.site, + status=OnboardingStatusChoices.STATUS_SUCCEEDED, + created_device=self.device, + ) + + self.succeeded_task2 = OnboardingTask.objects.create( + ip_address="10.10.10.10", + site=self.site, + status=OnboardingStatusChoices.STATUS_SUCCEEDED, + created_device=self.device, + ) + + self.failed_task1 = OnboardingTask.objects.create( + ip_address="10.10.10.10", + site=self.site, + status=OnboardingStatusChoices.STATUS_FAILED, + created_device=self.device, + ) + + self.failed_task2 = OnboardingTask.objects.create( + ip_address="10.10.10.10", + site=self.site, + status=OnboardingStatusChoices.STATUS_FAILED, + created_device=self.device, + ) + + def test_onboardingdevice_autocreated(self): + """Verify that OnboardingDevice is auto-created.""" + onboarding_device = OnboardingDevice.objects.get(device=self.device) + self.assertEqual(self.device, onboarding_device.device) + + def test_last_check_attempt_date(self): + """Verify OnboardingDevice last attempt.""" + onboarding_device = OnboardingDevice.objects.get(device=self.device) + self.assertEqual(onboarding_device.last_check_attempt_date, self.failed_task2.created) + + def test_last_check_successful_date(self): + """Verify OnboardingDevice last success.""" + onboarding_device = OnboardingDevice.objects.get(device=self.device) + self.assertEqual(onboarding_device.last_check_successful_date, self.succeeded_task2.created) + + def test_status(self): + """Verify OnboardingDevice status.""" + onboarding_device = OnboardingDevice.objects.get(device=self.device) + self.assertEqual(onboarding_device.status, self.failed_task2.status) + + def test_last_ot(self): + """Verify OnboardingDevice last ot.""" + onboarding_device = OnboardingDevice.objects.get(device=self.device) + self.assertEqual(onboarding_device.last_ot, self.failed_task2) diff --git a/netbox_onboarding/tests/test_netbox_keeper.py b/netbox_onboarding/tests/test_netbox_keeper.py index 5e7d5fb..f42d96d 100644 --- a/netbox_onboarding/tests/test_netbox_keeper.py +++ b/netbox_onboarding/tests/test_netbox_keeper.py @@ -11,16 +11,17 @@ See the License for the specific language governing permissions and limitations under the License. """ -from socket import gaierror -from unittest import mock +from django.conf import settings from django.test import TestCase from django.utils.text import slugify - -from dcim.models import Site, Device, Interface, Manufacturer, DeviceType, DeviceRole, Platform +from dcim.models import Site, Manufacturer, DeviceType, DeviceRole, Device, Interface, Platform from ipam.models import IPAddress -from netbox_onboarding.models import OnboardingTask -from netbox_onboarding.onboard import NetboxKeeper, NetdevKeeper, OnboardException +# from netbox_onboarding.netbox_keeper import NetdevKeeper +from netbox_onboarding.exceptions import OnboardException +from netbox_onboarding.netbox_keeper import NetboxKeeper + +PLUGIN_SETTINGS = settings.PLUGINS_CONFIG["netbox_onboarding"] class NetboxKeeperTestCase(TestCase): @@ -30,234 +31,427 @@ def setUp(self): """Create a superuser and token for API calls.""" self.site1 = Site.objects.create(name="USWEST", slug="uswest") - self.manufacturer1 = Manufacturer.objects.create(name="Juniper", slug="juniper") - self.platform1 = Platform.objects.create(name="JunOS", slug="junos") - self.platform2 = Platform.objects.create(name="Cisco NX-OS", slug="cisco-nx-os") - self.device_type1 = DeviceType.objects.create(slug="srx3600", model="SRX3600", manufacturer=self.manufacturer1) - self.device_role1 = DeviceRole.objects.create(name="Firewall", slug="firewall") + def test_ensure_device_manufacturer_strict_missing(self): + """Verify ensure_device_manufacturer function when Manufacturer object is not present.""" + PLUGIN_SETTINGS["object_match_strategy"] = "strict" + onboarding_kwargs = { + "netdev_hostname": "device1", + "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"], + "netdev_vendor": "Cisco", + "netdev_model": "CSR1000v", + "netdev_nb_site_slug": self.site1.slug, + } - self.onboarding_task1 = OnboardingTask.objects.create(ip_address="10.10.10.10", site=self.site1) - self.onboarding_task2 = OnboardingTask.objects.create( - ip_address="192.168.1.1", site=self.site1, role=self.device_role1 - ) - self.onboarding_task3 = OnboardingTask.objects.create( - ip_address="192.168.1.2", site=self.site1, role=self.device_role1, platform=self.platform1 - ) - self.onboarding_task4 = OnboardingTask.objects.create( - ip_address="ntc123.local", site=self.site1, role=self.device_role1, platform=self.platform1 - ) - self.onboarding_task5 = OnboardingTask.objects.create( - ip_address="bad.local", site=self.site1, role=self.device_role1, platform=self.platform1 - ) - self.onboarding_task6 = OnboardingTask.objects.create( - ip_address="192.0.2.2", site=self.site1, role=self.device_role1, platform=self.platform2 - ) - self.onboarding_task7 = OnboardingTask.objects.create( - ip_address="192.0.2.1/32", site=self.site1, role=self.device_role1, platform=self.platform1 - ) + nbk = NetboxKeeper(**onboarding_kwargs) - self.ndk1 = NetdevKeeper(self.onboarding_task1) - self.ndk1.hostname = "device1" - self.ndk1.vendor = "Cisco" - self.ndk1.model = "CSR1000v" - self.ndk1.serial_number = "123456" - self.ndk1.mgmt_ifname = "GigaEthernet0" - self.ndk1.mgmt_pflen = 24 - - self.ndk2 = NetdevKeeper(self.onboarding_task2) - self.ndk2.hostname = "device2" - self.ndk2.vendor = "juniper" - self.ndk2.model = "srx3600" - self.ndk2.serial_number = "123456" - self.ndk2.mgmt_ifname = "ge-0/0/0" - self.ndk2.mgmt_pflen = 24 - - def test_ensure_device_type_missing(self): - """Verify ensure_device_type function when Manufacturer and DeviceType object are not present.""" - nbk = NetboxKeeper(self.ndk1) + with self.assertRaises(OnboardException) as exc_info: + nbk.ensure_device_manufacturer(create_manufacturer=False) + self.assertEqual(exc_info.exception.message, "ERROR manufacturer not found: Cisco") + self.assertEqual(exc_info.exception.reason, "fail-config") + + nbk.ensure_device_manufacturer(create_manufacturer=True) + self.assertIsInstance(nbk.nb_manufacturer, Manufacturer) + self.assertEqual(nbk.nb_manufacturer.slug, slugify(onboarding_kwargs["netdev_vendor"])) + + def test_ensure_device_manufacturer_loose_missing(self): + """Verify ensure_device_manufacturer function when Manufacturer object is not present.""" + PLUGIN_SETTINGS["object_match_strategy"] = "loose" + onboarding_kwargs = { + "netdev_hostname": "device1", + "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"], + "netdev_vendor": "Cisco", + "netdev_model": "CSR1000v", + "netdev_nb_site_slug": self.site1.slug, + } + + nbk = NetboxKeeper(**onboarding_kwargs) with self.assertRaises(OnboardException) as exc_info: - nbk.ensure_device_type(create_manufacturer=False, create_device_type=False) + nbk.ensure_device_manufacturer(create_manufacturer=False) self.assertEqual(exc_info.exception.message, "ERROR manufacturer not found: Cisco") self.assertEqual(exc_info.exception.reason, "fail-config") + nbk.ensure_device_manufacturer(create_manufacturer=True) + self.assertIsInstance(nbk.nb_manufacturer, Manufacturer) + self.assertEqual(nbk.nb_manufacturer.slug, slugify(onboarding_kwargs["netdev_vendor"])) + + def test_ensure_device_type_strict_missing(self): + """Verify ensure_device_type function when DeviceType object is not present.""" + PLUGIN_SETTINGS["object_match_strategy"] = "strict" + onboarding_kwargs = { + "netdev_hostname": "device1", + "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"], + "netdev_vendor": "Cisco", + "netdev_model": "CSR1000v", + "netdev_nb_site_slug": self.site1.slug, + } + + nbk = NetboxKeeper(**onboarding_kwargs) + nbk.nb_manufacturer = Manufacturer.objects.create(name="Cisco", slug="cisco") + with self.assertRaises(OnboardException) as exc_info: - nbk.ensure_device_type(create_manufacturer=True, create_device_type=False) + nbk.ensure_device_type(create_device_type=False) self.assertEqual(exc_info.exception.message, "ERROR device type not found: CSR1000v") self.assertEqual(exc_info.exception.reason, "fail-config") - nbk.ensure_device_type(create_manufacturer=True, create_device_type=True) - self.assertIsInstance(nbk.manufacturer, Manufacturer) - self.assertIsInstance(nbk.device_type, DeviceType) - self.assertEqual(nbk.manufacturer.slug, slugify(self.ndk1.vendor)) - self.assertEqual(nbk.device_type.slug, slugify(self.ndk1.model)) + nbk.ensure_device_type(create_device_type=True) + self.assertIsInstance(nbk.nb_device_type, DeviceType) + self.assertEqual(nbk.nb_device_type.slug, slugify(onboarding_kwargs["netdev_model"])) - def test_ensure_device_type_present(self): - """Verify ensure_device_type function when Manufacturer and DeviceType object are already present.""" - nbk = NetboxKeeper(self.ndk2) + def test_ensure_device_type_loose_missing(self): + """Verify ensure_device_type function when DeviceType object is not present.""" + PLUGIN_SETTINGS["object_match_strategy"] = "loose" + onboarding_kwargs = { + "netdev_hostname": "device1", + "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"], + "netdev_vendor": "Cisco", + "netdev_model": "CSR1000v", + "netdev_nb_site_slug": self.site1.slug, + } - nbk.ensure_device_type(create_manufacturer=False, create_device_type=False) - self.assertEqual(nbk.manufacturer, self.manufacturer1) - self.assertEqual(nbk.device_type, self.device_type1) + nbk = NetboxKeeper(**onboarding_kwargs) + nbk.nb_manufacturer = Manufacturer.objects.create(name="Cisco", slug="cisco") + + with self.assertRaises(OnboardException) as exc_info: + nbk.ensure_device_type(create_device_type=False) + self.assertEqual(exc_info.exception.message, "ERROR device type not found: CSR1000v") + self.assertEqual(exc_info.exception.reason, "fail-config") + + nbk.ensure_device_type(create_device_type=True) + self.assertIsInstance(nbk.nb_device_type, DeviceType) + self.assertEqual(nbk.nb_device_type.slug, slugify(onboarding_kwargs["netdev_model"])) + + def test_ensure_device_type_strict_present(self): + """Verify ensure_device_type function when DeviceType object is already present.""" + PLUGIN_SETTINGS["object_match_strategy"] = "strict" + manufacturer = Manufacturer.objects.create(name="Juniper", slug="juniper") + + device_type = DeviceType.objects.create(slug="srx3600", model="SRX3600", manufacturer=manufacturer) + + onboarding_kwargs = { + "netdev_hostname": "device2", + "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"], + "netdev_vendor": "Juniper", + "netdev_nb_device_type_slug": device_type.slug, + "netdev_nb_site_slug": self.site1.slug, + } + + nbk = NetboxKeeper(**onboarding_kwargs) + nbk.nb_manufacturer = manufacturer + + nbk.ensure_device_type(create_device_type=False) + self.assertEqual(nbk.nb_device_type, device_type) + + def test_ensure_device_type_loose_present(self): + """Verify ensure_device_type function when DeviceType object is already present.""" + PLUGIN_SETTINGS["object_match_strategy"] = "loose" + manufacturer = Manufacturer.objects.create(name="Juniper", slug="juniper") + + device_type = DeviceType.objects.create(slug="srx3600", model="SRX3600", manufacturer=manufacturer) + + onboarding_kwargs = { + "netdev_hostname": "device2", + "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"], + "netdev_vendor": "Juniper", + "netdev_nb_device_type_slug": device_type.slug, + "netdev_nb_site_slug": self.site1.slug, + } + + nbk = NetboxKeeper(**onboarding_kwargs) + nbk.nb_manufacturer = manufacturer + + nbk.ensure_device_type(create_device_type=False) + self.assertEqual(nbk.nb_device_type, device_type) def test_ensure_device_role_not_exist(self): - """Verify ensure_device_role function when DeviceRole do not already exist.""" - nbk = NetboxKeeper(self.ndk1) + """Verify ensure_device_role function when DeviceRole does not already exist.""" + test_role_name = "mytestrole" + + onboarding_kwargs = { + "netdev_hostname": "device1", + "netdev_nb_role_slug": test_role_name, + "netdev_nb_role_color": PLUGIN_SETTINGS["default_device_role_color"], + "netdev_vendor": "Cisco", + "netdev_nb_site_slug": self.site1.slug, + } + + nbk = NetboxKeeper(**onboarding_kwargs) with self.assertRaises(OnboardException) as exc_info: - nbk.ensure_device_role(create_device_role=False, default_device_role="mytestrole") - self.assertEqual(exc_info.exception.message, "ERROR device role not found: mytestrole") + nbk.ensure_device_role(create_device_role=False) + self.assertEqual(exc_info.exception.message, f"ERROR device role not found: {test_role_name}") self.assertEqual(exc_info.exception.reason, "fail-config") - role = "My-Test-Role" - nbk.ensure_device_role(create_device_role=True, default_device_role=role) - self.assertIsInstance(nbk.netdev.ot.role, DeviceRole) - self.assertEqual(nbk.netdev.ot.role.slug, slugify(role)) + nbk.ensure_device_role(create_device_role=True) + self.assertIsInstance(nbk.nb_device_role, DeviceRole) + self.assertEqual(nbk.nb_device_role.slug, slugify(test_role_name)) def test_ensure_device_role_exist(self): """Verify ensure_device_role function when DeviceRole exist but is not assigned to the OT.""" - nbk = NetboxKeeper(self.ndk1) + device_role = DeviceRole.objects.create(name="Firewall", slug="firewall") + + onboarding_kwargs = { + "netdev_hostname": "device1", + "netdev_nb_role_slug": device_role.slug, + "netdev_nb_role_color": PLUGIN_SETTINGS["default_device_role_color"], + "netdev_vendor": "Cisco", + "netdev_nb_site_slug": self.site1.slug, + } - nbk.ensure_device_role(create_device_role=True, default_device_role="firewall") - self.assertEqual(nbk.netdev.ot.role, self.device_role1) + nbk = NetboxKeeper(**onboarding_kwargs) + nbk.ensure_device_role(create_device_role=False) + self.assertEqual(nbk.nb_device_role, device_role) + + # def test_ensure_device_role_assigned(self): """Verify ensure_device_role function when DeviceRole exist and is already assigned.""" - nbk = NetboxKeeper(self.ndk2) + device_role = DeviceRole.objects.create(name="Firewall", slug="firewall") + + onboarding_kwargs = { + "netdev_hostname": "device1", + "netdev_nb_role_slug": device_role.slug, + "netdev_nb_role_color": PLUGIN_SETTINGS["default_device_role_color"], + "netdev_vendor": "Cisco", + "netdev_nb_site_slug": self.site1.slug, + } - nbk.ensure_device_role(create_device_role=True, default_device_role="firewall") - self.assertEqual(nbk.netdev.ot.role, self.device_role1) + nbk = NetboxKeeper(**onboarding_kwargs) + nbk.ensure_device_role(create_device_role=True) + + self.assertEqual(nbk.nb_device_role, device_role) def test_ensure_device_instance_not_exist(self): """Verify ensure_device_instance function.""" - nbk = NetboxKeeper(self.ndk2) - nbk.device_type = self.device_type1 - nbk.netdev.ot = self.onboarding_task3 + serial_number = "123456" + platform_slug = "cisco_ios" + hostname = "device1" + + onboarding_kwargs = { + "netdev_hostname": hostname, + "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"], + "netdev_nb_role_color": PLUGIN_SETTINGS["default_device_role_color"], + "netdev_vendor": "Cisco", + "netdev_model": "CSR1000v", + "netdev_nb_site_slug": self.site1.slug, + "netdev_netmiko_device_type": platform_slug, + "netdev_serial_number": serial_number, + "netdev_mgmt_ip_address": "192.0.2.10", + "netdev_mgmt_ifname": "GigaEthernet0", + "netdev_mgmt_pflen": 24, + } + + nbk = NetboxKeeper(**onboarding_kwargs) + + nbk.ensure_device() - nbk.ensure_device_instance(default_status="planned") self.assertIsInstance(nbk.device, Device) - self.assertEqual(nbk.device.status, "planned") - self.assertEqual(nbk.device.platform, self.platform1) - self.assertEqual(nbk.device, nbk.netdev.ot.created_device) - self.assertEqual(nbk.device.serial, "123456") + self.assertEqual(nbk.device.name, hostname) + self.assertEqual(nbk.device.status, PLUGIN_SETTINGS["default_device_status"]) + self.assertEqual(nbk.device.platform.slug, platform_slug) + self.assertEqual(nbk.device.serial, serial_number) def test_ensure_device_instance_exist(self): """Verify ensure_device_instance function.""" + manufacturer = Manufacturer.objects.create(name="Cisco", slug="cisco") + + device_role = DeviceRole.objects.create(name="Switch", slug="switch") + + device_type = DeviceType.objects.create(slug="c2960", model="c2960", manufacturer=manufacturer) + + device_name = "test_name" + device = Device.objects.create( - name=self.ndk2.hostname, + name=device_name, site=self.site1, - device_type=self.device_type1, - device_role=self.device_role1, + device_type=device_type, + device_role=device_role, status="planned", serial="987654", ) - nbk = NetboxKeeper(self.ndk2) - nbk.netdev.ot = self.onboarding_task3 - self.assertEqual(nbk.device, None) - nbk.ensure_device_instance(default_status="active") + onboarding_kwargs = { + "netdev_hostname": device_name, + "netdev_nb_role_slug": "switch", + "netdev_vendor": "Cisco", + "netdev_model": "c2960", + "netdev_nb_site_slug": self.site1.slug, + "netdev_netmiko_device_type": "cisco_ios", + "netdev_serial_number": "123456", + "netdev_mgmt_ip_address": "192.0.2.10", + "netdev_mgmt_ifname": "GigaEthernet0", + "netdev_mgmt_pflen": 24, + } + + nbk = NetboxKeeper(**onboarding_kwargs) + + nbk.ensure_device() + self.assertIsInstance(nbk.device, Device) - self.assertEqual(nbk.device.status, "planned") - self.assertEqual(nbk.device.platform, self.platform1) - self.assertEqual(nbk.device, device) + self.assertEqual(nbk.device.pk, device.pk) + + self.assertEqual(nbk.device.name, device_name) + self.assertEqual(nbk.device.platform.slug, "cisco_ios") self.assertEqual(nbk.device.serial, "123456") def test_ensure_interface_not_exist(self): """Verify ensure_interface function when the interface do not exist.""" - nbk = NetboxKeeper(self.ndk2) - nbk.device_type = self.device_type1 - nbk.netdev.ot = self.onboarding_task3 - - nbk.ensure_device_instance() - - nbk.ensure_interface() - self.assertIsInstance(nbk.interface, Interface) - self.assertEqual(nbk.interface.name, "ge-0/0/0") + onboarding_kwargs = { + "netdev_hostname": "device1", + "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"], + "netdev_nb_role_color": PLUGIN_SETTINGS["default_device_role_color"], + "netdev_vendor": "Cisco", + "netdev_model": "CSR1000v", + "netdev_nb_site_slug": self.site1.slug, + "netdev_netmiko_device_type": "cisco_ios", + "netdev_serial_number": "123456", + "netdev_mgmt_ip_address": "192.0.2.10", + "netdev_mgmt_ifname": "ge-0/0/0", + "netdev_mgmt_pflen": 24, + } + + nbk = NetboxKeeper(**onboarding_kwargs) + nbk.ensure_device() + + self.assertIsInstance(nbk.nb_mgmt_ifname, Interface) + self.assertEqual(nbk.nb_mgmt_ifname.name, "ge-0/0/0") def test_ensure_interface_exist(self): """Verify ensure_interface function when the interface already exist.""" - nbk = NetboxKeeper(self.ndk2) - nbk.device_type = self.device_type1 - nbk.netdev.ot = self.onboarding_task3 + manufacturer = Manufacturer.objects.create(name="Cisco", slug="cisco") + + device_role = DeviceRole.objects.create(name="Switch", slug="switch") + + device_type = DeviceType.objects.create(slug="c2960", model="c2960", manufacturer=manufacturer) + + device_name = "test_name" + netdev_mgmt_ifname = "GigaEthernet0" + + device = Device.objects.create( + name=device_name, + site=self.site1, + device_type=device_type, + device_role=device_role, + status="planned", + serial="987654", + ) + + intf = Interface.objects.create(name=netdev_mgmt_ifname, device=device) + + onboarding_kwargs = { + "netdev_hostname": device_name, + "netdev_nb_role_slug": "switch", + "netdev_vendor": "Cisco", + "netdev_model": "c2960", + "netdev_nb_site_slug": self.site1.slug, + "netdev_netmiko_device_type": "cisco_ios", + "netdev_serial_number": "123456", + "netdev_mgmt_ip_address": "192.0.2.10", + "netdev_mgmt_ifname": netdev_mgmt_ifname, + "netdev_mgmt_pflen": 24, + } - nbk.ensure_device_instance() - intf = Interface.objects.create(name=nbk.netdev.mgmt_ifname, device=nbk.device) + nbk = NetboxKeeper(**onboarding_kwargs) - nbk.ensure_interface() - self.assertEqual(nbk.interface, intf) + nbk.ensure_device() + + self.assertEqual(nbk.nb_mgmt_ifname, intf) def test_ensure_primary_ip_not_exist(self): """Verify ensure_primary_ip function when the IP address do not already exist.""" - nbk = NetboxKeeper(self.ndk2) - nbk.device_type = self.device_type1 - nbk.netdev.ot = self.onboarding_task3 - - nbk.ensure_device_instance() - nbk.ensure_interface() - nbk.ensure_primary_ip() - self.assertIsInstance(nbk.primary_ip, IPAddress) - self.assertEqual(nbk.primary_ip.interface, nbk.interface) - - @mock.patch("netbox_onboarding.onboard.socket.gethostbyname") - def test_check_ip(self, mock_get_hostbyname): - """Check DNS to IP address.""" - # Look up response value - mock_get_hostbyname.return_value = "192.0.2.1" - - # Create a Device Keeper object of the device - ndk4 = NetdevKeeper(self.onboarding_task4) - - # Check that the IP address is returned - self.assertTrue(ndk4.check_ip()) - - # Run the check to change the IP address - self.assertEqual(ndk4.ot.ip_address, "192.0.2.1") - - @mock.patch("netbox_onboarding.onboard.socket.gethostbyname") - def test_failed_check_ip(self, mock_get_hostbyname): - """Check DNS to IP address failing.""" - # Look up a failed response - mock_get_hostbyname.side_effect = gaierror(8) - ndk5 = NetdevKeeper(self.onboarding_task5) - ndk7 = NetdevKeeper(self.onboarding_task7) - - # Check for bad.local raising an exception - with self.assertRaises(OnboardException) as exc_info: - ndk5.check_ip() - self.assertEqual(exc_info.exception.message, "ERROR failed to complete DNS lookup: bad.local") - self.assertEqual(exc_info.exception.reason, "fail-dns") + onboarding_kwargs = { + "netdev_hostname": "device1", + "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"], + "netdev_nb_role_color": PLUGIN_SETTINGS["default_device_role_color"], + "netdev_vendor": "Cisco", + "netdev_model": "CSR1000v", + "netdev_nb_site_slug": self.site1.slug, + "netdev_netmiko_device_type": "cisco_ios", + "netdev_serial_number": "123456", + "netdev_mgmt_ip_address": "192.0.2.10", + "netdev_mgmt_ifname": "ge-0/0/0", + "netdev_mgmt_pflen": 24, + } + + nbk = NetboxKeeper(**onboarding_kwargs) + nbk.ensure_device() + + self.assertIsInstance(nbk.nb_primary_ip, IPAddress) + self.assertIn(nbk.nb_primary_ip, Interface.objects.get(device=nbk.device, name="ge-0/0/0").ip_addresses.all()) + self.assertEqual(nbk.device.primary_ip, nbk.nb_primary_ip) + + def test_ensure_device_platform_missing(self): + """Verify ensure_device_platform function when Platform object is not present.""" + platform_name = "cisco_ios" + + onboarding_kwargs = { + "netdev_hostname": "device1", + "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"], + "netdev_vendor": "Cisco", + "netdev_model": "CSR1000v", + "netdev_nb_site_slug": self.site1.slug, + "netdev_nb_platform_slug": platform_name, + "netdev_netmiko_device_type": platform_name, + } + + nbk = NetboxKeeper(**onboarding_kwargs) - # Check for exception with prefix address entered with self.assertRaises(OnboardException) as exc_info: - ndk7.check_ip() - self.assertEqual(exc_info.exception.reason, "fail-prefix") - self.assertEqual(exc_info.exception.message, "ERROR appears a prefix was entered: 192.0.2.1/32") + nbk.ensure_device_platform(create_platform_if_missing=False) + self.assertEqual(exc_info.exception.message, f"ERROR device platform not found: {platform_name}") + self.assertEqual(exc_info.exception.reason, "fail-config") - def test_platform_map(self): - """Verify platform mapping of netmiko to slug functionality.""" - # Create static mapping - platform_map = {"cisco_ios": "ios", "arista_eos": "eos", "cisco_nxos": "cisco-nxos"} + nbk.ensure_device_platform(create_platform_if_missing=True) + self.assertIsInstance(nbk.nb_platform, Platform) + self.assertEqual(nbk.nb_platform.slug, slugify(platform_name)) + + def test_ensure_platform_present(self): + """Verify ensure_device_platform function when Platform object is present.""" + platform_name = "juniper_junos" + + manufacturer = Manufacturer.objects.create(name="Juniper", slug="juniper") + + device_type = DeviceType.objects.create(slug="srx3600", model="SRX3600", manufacturer=manufacturer) - # Generate an instance of a Cisco IOS device with the mapping defined - self.ndk1 = NetdevKeeper(self.onboarding_task1) + platform = Platform.objects.create(slug=platform_name, name=platform_name,) - # - # Test positive assertions - # + onboarding_kwargs = { + "netdev_hostname": "device2", + "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"], + "netdev_vendor": "Juniper", + "netdev_nb_device_type_slug": device_type.slug, + "netdev_nb_site_slug": self.site1.slug, + "netdev_nb_platform_slug": platform_name, + } - # Test Cisco_ios - self.assertEqual(self.ndk1.check_netmiko_conversion("cisco_ios", platform_map=platform_map), "ios") - # Test Arista EOS - self.assertEqual(self.ndk1.check_netmiko_conversion("arista_eos", platform_map=platform_map), "eos") - # Test cisco_nxos - self.assertEqual(self.ndk1.check_netmiko_conversion("cisco_nxos", platform_map=platform_map), "cisco-nxos") + nbk = NetboxKeeper(**onboarding_kwargs) - # - # Test Negative assertion - # + nbk.ensure_device_platform(create_platform_if_missing=False) - # Test a non-converting item + self.assertIsInstance(nbk.nb_platform, Platform) + self.assertEqual(nbk.nb_platform, platform) + self.assertEqual(nbk.nb_platform.slug, slugify(platform_name)) + + def test_platform_map(self): + """Verify platform mapping of netmiko to slug functionality.""" + # Create static mapping + PLUGIN_SETTINGS["platform_map"] = {"cisco_ios": "ios", "arista_eos": "eos", "cisco_nxos": "cisco-nxos"} + + onboarding_kwargs = { + "netdev_hostname": "device1", + "netdev_nb_role_slug": PLUGIN_SETTINGS["default_device_role"], + "netdev_vendor": "Cisco", + "netdev_model": "CSR1000v", + "netdev_nb_site_slug": self.site1.slug, + "netdev_netmiko_device_type": "cisco_ios", + } + + nbk = NetboxKeeper(**onboarding_kwargs) + + nbk.ensure_device_platform(create_platform_if_missing=True) + self.assertIsInstance(nbk.nb_platform, Platform) + self.assertEqual(nbk.nb_platform.slug, slugify(PLUGIN_SETTINGS["platform_map"]["cisco_ios"])) self.assertEqual( - self.ndk1.check_netmiko_conversion("cisco-device-platform", platform_map=platform_map), - "cisco-device-platform", + Platform.objects.get(name=PLUGIN_SETTINGS["platform_map"]["cisco_ios"]).name, + slugify(PLUGIN_SETTINGS["platform_map"]["cisco_ios"]), ) diff --git a/netbox_onboarding/tests/test_netdev_keeper.py b/netbox_onboarding/tests/test_netdev_keeper.py index 5d43472..36623da 100644 --- a/netbox_onboarding/tests/test_netdev_keeper.py +++ b/netbox_onboarding/tests/test_netdev_keeper.py @@ -1,4 +1,4 @@ -"""Unit tests for netbox_onboarding.onboard module and its classes. +"""Unit tests for netbox_onboarding.netdev_keeper module and its classes. (c) 2020 Network To Code Licensed under the Apache License, Version 2.0 (the "License"); @@ -11,10 +11,16 @@ See the License for the specific language governing permissions and limitations under the License. """ + +from socket import gaierror +from unittest import mock + from django.test import TestCase +from dcim.models import Site, DeviceRole, Platform -from dcim.models import Platform -from netbox_onboarding.onboard import NetdevKeeper, OnboardException +from netbox_onboarding.exceptions import OnboardException +from netbox_onboarding.models import OnboardingTask +from netbox_onboarding.netdev_keeper import NetdevKeeper class NetdevKeeperTestCase(TestCase): @@ -22,37 +28,55 @@ class NetdevKeeperTestCase(TestCase): def setUp(self): """Create a superuser and token for API calls.""" + self.site1 = Site.objects.create(name="USWEST", slug="uswest") + self.device_role1 = DeviceRole.objects.create(name="Firewall", slug="firewall") + self.platform1 = Platform.objects.create(name="JunOS", slug="junos", napalm_driver="junos") - self.platform2 = Platform.objects.create(name="Cisco NX-OS", slug="cisco-nx-os") + # self.platform2 = Platform.objects.create(name="Cisco NX-OS", slug="cisco-nx-os") - def test_get_platform_object_from_netbox(self): - """Test of platform object from netbox.""" - # Test assigning platform - platform = NetdevKeeper.get_platform_object_from_netbox("junos", create_platform_if_missing=False) - self.assertIsInstance(platform, Platform) + self.onboarding_task4 = OnboardingTask.objects.create( + ip_address="ntc123.local", site=self.site1, role=self.device_role1, platform=self.platform1 + ) - # Test creation of missing platform object - platform = NetdevKeeper.get_platform_object_from_netbox("arista_eos", create_platform_if_missing=True) - self.assertIsInstance(platform, Platform) - self.assertEqual(platform.napalm_driver, "eos") + self.onboarding_task5 = OnboardingTask.objects.create( + ip_address="bad.local", site=self.site1, role=self.device_role1, platform=self.platform1 + ) - # Test failed unable to find the device and not part of the NETMIKO TO NAPALM keys - with self.assertRaises(OnboardException) as exc_info: - platform = NetdevKeeper.get_platform_object_from_netbox("notthere", create_platform_if_missing=True) - self.assertEqual( - exc_info.exception.message, - "ERROR platform not found in NetBox and it's eligible for auto-creation: notthere", - ) - self.assertEqual(exc_info.exception.reason, "fail-general") - - # Test searching for an object, does not exist, but create_platform is false + self.onboarding_task7 = OnboardingTask.objects.create( + ip_address="192.0.2.1/32", site=self.site1, role=self.device_role1, platform=self.platform1 + ) + + @mock.patch("netbox_onboarding.netdev_keeper.socket.gethostbyname") + def test_check_ip(self, mock_get_hostbyname): + """Check DNS to IP address.""" + # Look up response value + mock_get_hostbyname.return_value = "192.0.2.1" + + # Create a Device Keeper object of the device + ndk4 = NetdevKeeper(hostname=self.onboarding_task4.ip_address) + + # Check that the IP address is returned + self.assertTrue(ndk4.check_ip()) + + # Run the check to change the IP address + self.assertEqual(ndk4.hostname, "192.0.2.1") + + @mock.patch("netbox_onboarding.netdev_keeper.socket.gethostbyname") + def test_failed_check_ip(self, mock_get_hostbyname): + """Check DNS to IP address failing.""" + # Look up a failed response + mock_get_hostbyname.side_effect = gaierror(8) + ndk5 = NetdevKeeper(hostname=self.onboarding_task5.ip_address) + ndk7 = NetdevKeeper(hostname=self.onboarding_task7.ip_address) + + # Check for bad.local raising an exception with self.assertRaises(OnboardException) as exc_info: - platform = NetdevKeeper.get_platform_object_from_netbox("cisco_ios", create_platform_if_missing=False) - self.assertEqual(exc_info.exception.message, "ERROR platform not found in NetBox: cisco_ios") - self.assertEqual(exc_info.exception.reason, "fail-general") + ndk5.check_ip() + self.assertEqual(exc_info.exception.message, "ERROR failed to complete DNS lookup: bad.local") + self.assertEqual(exc_info.exception.reason, "fail-dns") - # Test NAPALM Driver not defined in NetBox + # Check for exception with prefix address entered with self.assertRaises(OnboardException) as exc_info: - platform = NetdevKeeper.get_platform_object_from_netbox("cisco-nx-os", create_platform_if_missing=False) - self.assertEqual(exc_info.exception.message, "ERROR platform is missing the NAPALM Driver: cisco-nx-os") - self.assertEqual(exc_info.exception.reason, "fail-general") + ndk7.check_ip() + self.assertEqual(exc_info.exception.reason, "fail-prefix") + self.assertEqual(exc_info.exception.message, "ERROR appears a prefix was entered: 192.0.2.1/32") diff --git a/netbox_onboarding/tests/test_views.py b/netbox_onboarding/tests/test_views.py deleted file mode 100644 index c40f60e..0000000 --- a/netbox_onboarding/tests/test_views.py +++ /dev/null @@ -1,229 +0,0 @@ -"""Unit tests for netbox_onboarding views. - -(c) 2020 Network To Code -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - http://www.apache.org/licenses/LICENSE-2.0 -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -""" -from django.contrib.auth.models import User, Permission -from django.test import Client, TestCase, override_settings -from django.urls import reverse - -from dcim.models import Site - -from netbox_onboarding.models import OnboardingTask - - -class OnboardingTaskListViewTestCase(TestCase): - """Test the OnboardingTaskListView view.""" - - def setUp(self): - """Create a user and baseline data for testing.""" - self.user = User.objects.create(username="testuser") - self.client = Client() - self.client.force_login(self.user) - - self.url = reverse("plugins:netbox_onboarding:onboarding_task_list") - - self.site1 = Site.objects.create(name="USWEST", slug="uswest") - self.onboarding_task1 = OnboardingTask.objects.create(ip_address="10.10.10.10", site=self.site1) - self.onboarding_task2 = OnboardingTask.objects.create(ip_address="192.168.1.1", site=self.site1) - - @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"]) - def test_list_onboarding_tasks_anonymous(self): - """Verify that OnboardingTasks can be listed without logging in if permissions are exempted.""" - self.client.logout() - response = self.client.get(self.url) - self.assertEqual(response.status_code, 200) - self.assertTemplateUsed(response, "netbox_onboarding/onboarding_tasks_list.html") - - @override_settings(EXEMPT_VIEW_PERMISSIONS=[]) - def test_list_onboarding_tasks(self): - """Verify that OnboardingTasks can be listed by a user with appropriate permissions.""" - # Attempt to access without permissions - response = self.client.get(self.url) - self.assertEqual(response.status_code, 403) - - # Add permission - self.user.user_permissions.add( - Permission.objects.get(content_type__app_label="netbox_onboarding", codename="view_onboardingtask") - ) - - response = self.client.get(self.url) - self.assertEqual(response.status_code, 200) - self.assertTemplateUsed(response, "netbox_onboarding/onboarding_tasks_list.html") - - -class OnboardingTaskCreateViewTestCase(TestCase): - """Test the OnboardingTaskCreateView view.""" - - def setUp(self): - """Create a user and baseline data for testing.""" - self.user = User.objects.create(username="testuser") - self.client = Client() - self.client.force_login(self.user) - - self.url = reverse("plugins:netbox_onboarding:onboarding_task_add") - - self.site1 = Site.objects.create(name="USWEST", slug="uswest") - - @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"]) - def test_get_anonymous(self): - """Verify that the view cannot be accessed by anonymous users even if permissions are exempted.""" - self.client.logout() - response = self.client.get(self.url) - # Redirected to the login page - self.assertEqual(response.status_code, 302) - - @override_settings(EXEMPT_VIEW_PERMISSIONS=[]) - def test_get(self): - """Verify that the view can be seen by a user with appropriate permissions.""" - # Attempt to access without permissions - response = self.client.get(self.url) - self.assertEqual(response.status_code, 403) - - # Add permission - self.user.user_permissions.add( - Permission.objects.get(content_type__app_label="netbox_onboarding", codename="add_onboardingtask") - ) - - response = self.client.get(self.url) - self.assertEqual(response.status_code, 200) - self.assertTemplateUsed(response, "netbox_onboarding/onboarding_task_edit.html") - - @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"]) - def test_post_anonymous(self): - """Verify that the view cannot be accessed by anonymous users even if permissions are exempted.""" - self.client.logout() - response = self.client.get(self.url) - # Redirected to the login page - self.assertEqual(response.status_code, 302) - - @override_settings(EXEMPT_VIEW_PERMISSIONS=[]) - def test_post(self): - """Verify that the view can be used by a user with appropriate permissions.""" - # Attempt to access without permissions - response = self.client.post(self.url) - self.assertEqual(response.status_code, 403) - - # Add permission - self.user.user_permissions.add( - Permission.objects.get(content_type__app_label="netbox_onboarding", codename="add_onboardingtask") - ) - - response = self.client.post( - self.url, data={"ip_address": "10.10.10.10", "site": "uswest", "port": "22", "timeout": "30"} - ) - self.assertEqual(response.status_code, 302) - self.assertEqual(OnboardingTask.objects.count(), 1) - - -class OnboardingTaskBulkDeleteViewTestCase(TestCase): - """Test the OnboardingTaskBulkDeleteView view.""" - - def setUp(self): - """Create a user and baseline data for testing.""" - self.user = User.objects.create(username="testuser") - self.client = Client() - self.client.force_login(self.user) - - self.url = reverse("plugins:netbox_onboarding:onboarding_task_bulk_delete") - - self.site1 = Site.objects.create(name="USWEST", slug="uswest") - self.onboarding_task1 = OnboardingTask.objects.create(ip_address="10.10.10.10", site=self.site1) - self.onboarding_task2 = OnboardingTask.objects.create(ip_address="192.168.1.1", site=self.site1) - - @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"]) - def test_post_anonymous(self): - """Verify that the view cannot be accessed by anonymous users even if permissions are exempted.""" - self.client.logout() - response = self.client.post(self.url) - # Redirected to the login page - self.assertEqual(response.status_code, 302) - - @override_settings(EXEMPT_VIEW_PERMISSIONS=[]) - def test_post(self): - """Verify that the view can be seen by a user with appropriate permissions.""" - # Attempt to access without permissions - response = self.client.post( - self.url, data={"pk": [self.onboarding_task1.pk], "confirm": True, "_confirm": True} - ) - self.assertEqual(response.status_code, 403) - - # Add permission - self.user.user_permissions.add( - Permission.objects.get(content_type__app_label="netbox_onboarding", codename="delete_onboardingtask") - ) - - response = self.client.post( - self.url, data={"pk": [self.onboarding_task1.pk], "confirm": True, "_confirm": True} - ) - self.assertEqual(response.status_code, 302) - self.assertEqual(OnboardingTask.objects.count(), 1) - - -class OnboardingTaskFeedBulkImportViewTestCase(TestCase): - """Test the OnboardingTaskFeedBulkImportView view.""" - - def setUp(self): - """Create a superuser and baseline data for testing.""" - self.user = User.objects.create(username="testuser") - self.client = Client() - self.client.force_login(self.user) - - self.url = reverse("plugins:netbox_onboarding:onboarding_task_import") - - self.site1 = Site.objects.create(name="USWEST", slug="uswest") - - @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"]) - def test_get_anonymous(self): - """Verify that the import view cannot be seen by an anonymous user even if permissions are exempted.""" - self.client.logout() - response = self.client.get(self.url) - # Redirected to the login page - self.assertEqual(response.status_code, 302) - - @override_settings(EXEMPT_VIEW_PERMISSIONS=[]) - def test_get(self): - """Verify that the import view can be seen by a user with appropriate permissions.""" - # Attempt to access without permissions - response = self.client.get(self.url) - self.assertEqual(response.status_code, 403) - - # Add permission - self.user.user_permissions.add( - Permission.objects.get(content_type__app_label="netbox_onboarding", codename="add_onboardingtask") - ) - - response = self.client.get(self.url) - self.assertEqual(response.status_code, 200) - self.assertTemplateUsed(response, "utilities/obj_bulk_import.html") - - @override_settings(EXEMPT_VIEW_PERMISSIONS=[]) - def test_post(self): - """Verify that tasks can be bulk-imported.""" - csv_data = [ - "site,ip_address", - "uswest,10.10.10.10", - "uswest,10.10.10.20", - "uswest,10.10.10.30", - ] - - # Attempt to access without permissions - response = self.client.post(self.url, data={"csv": "\n".join(csv_data)}) - self.assertEqual(response.status_code, 403) - - # Add permission - self.user.user_permissions.add( - Permission.objects.get(content_type__app_label="netbox_onboarding", codename="add_onboardingtask") - ) - - response = self.client.post(self.url, data={"csv": "\n".join(csv_data)}) - self.assertEqual(response.status_code, 200) - self.assertEqual(OnboardingTask.objects.count(), len(csv_data) - 1) diff --git a/netbox_onboarding/tests/test_views_28.py b/netbox_onboarding/tests/test_views_28.py new file mode 100644 index 0000000..b84e16d --- /dev/null +++ b/netbox_onboarding/tests/test_views_28.py @@ -0,0 +1,56 @@ +"""Unit tests for netbox_onboarding views. + +(c) 2020 Network To Code +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" +from dcim.models import Site +from utilities.testing import ViewTestCases + +from netbox_onboarding.models import OnboardingTask +from netbox_onboarding.release import NETBOX_RELEASE_CURRENT, NETBOX_RELEASE_29 + + +if NETBOX_RELEASE_CURRENT < NETBOX_RELEASE_29: + + class OnboardingTestCase( + ViewTestCases.GetObjectViewTestCase, + ViewTestCases.ListObjectsViewTestCase, + ViewTestCases.CreateObjectViewTestCase, + ViewTestCases.BulkDeleteObjectsViewTestCase, + ViewTestCases.ImportObjectsViewTestCase, # pylint: disable=no-member + ): + """Test the OnboardingTask views.""" + + def _get_base_url(self): + return "plugins:{}:{}_{{}}".format(self.model._meta.app_label, self.model._meta.model_name) + + model = OnboardingTask + + @classmethod + def setUpTestData(cls): # pylint: disable=invalid-name, missing-function-docstring + """Setup test data.""" + site = Site.objects.create(name="USWEST", slug="uswest") + OnboardingTask.objects.create(ip_address="10.10.10.10", site=site) + OnboardingTask.objects.create(ip_address="192.168.1.1", site=site) + + cls.form_data = { + "site": site.pk, + "ip_address": "192.0.2.99", + "port": 22, + "timeout": 30, + } + + cls.csv_data = ( + "site,ip_address", + "uswest,10.10.10.10", + "uswest,10.10.10.20", + "uswest,10.10.10.30", + ) diff --git a/netbox_onboarding/tests/test_views_29.py b/netbox_onboarding/tests/test_views_29.py new file mode 100644 index 0000000..1986801 --- /dev/null +++ b/netbox_onboarding/tests/test_views_29.py @@ -0,0 +1,56 @@ +"""Unit tests for netbox_onboarding views. + +(c) 2020 Network To Code +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" +from dcim.models import Site +from utilities.testing import ViewTestCases + +from netbox_onboarding.models import OnboardingTask +from netbox_onboarding.release import NETBOX_RELEASE_CURRENT, NETBOX_RELEASE_29 + + +if NETBOX_RELEASE_CURRENT >= NETBOX_RELEASE_29: + + class OnboardingTestCase( + ViewTestCases.GetObjectViewTestCase, + ViewTestCases.ListObjectsViewTestCase, + ViewTestCases.CreateObjectViewTestCase, + ViewTestCases.BulkDeleteObjectsViewTestCase, + ViewTestCases.BulkImportObjectsViewTestCase, # pylint: disable=no-member + ): + """Test the OnboardingTask views.""" + + def _get_base_url(self): + return "plugins:{}:{}_{{}}".format(self.model._meta.app_label, self.model._meta.model_name) + + model = OnboardingTask + + @classmethod + def setUpTestData(cls): # pylint: disable=invalid-name, missing-function-docstring + """Setup test data.""" + site = Site.objects.create(name="USWEST", slug="uswest") + OnboardingTask.objects.create(ip_address="10.10.10.10", site=site) + OnboardingTask.objects.create(ip_address="192.168.1.1", site=site) + + cls.form_data = { + "site": site.pk, + "ip_address": "192.0.2.99", + "port": 22, + "timeout": 30, + } + + cls.csv_data = ( + "site,ip_address", + "uswest,10.10.10.10", + "uswest,10.10.10.20", + "uswest,10.10.10.30", + ) diff --git a/netbox_onboarding/urls.py b/netbox_onboarding/urls.py index 009c836..12353e1 100644 --- a/netbox_onboarding/urls.py +++ b/netbox_onboarding/urls.py @@ -12,8 +12,11 @@ limitations under the License. """ from django.urls import path +from extras.views import ObjectChangeLogView +from .models import OnboardingTask from .views import ( + OnboardingTaskView, OnboardingTaskListView, OnboardingTaskCreateView, OnboardingTaskBulkDeleteView, @@ -21,8 +24,15 @@ ) urlpatterns = [ - path("", OnboardingTaskListView.as_view(), name="onboarding_task_list"), - path("add/", OnboardingTaskCreateView.as_view(), name="onboarding_task_add"), - path("delete/", OnboardingTaskBulkDeleteView.as_view(), name="onboarding_task_bulk_delete"), - path("import/", OnboardingTaskFeedBulkImportView.as_view(), name="onboarding_task_import"), + path("", OnboardingTaskListView.as_view(), name="onboardingtask_list"), + path("/", OnboardingTaskView.as_view(), name="onboardingtask"), + path("add/", OnboardingTaskCreateView.as_view(), name="onboardingtask_add"), + path("delete/", OnboardingTaskBulkDeleteView.as_view(), name="onboardingtask_bulk_delete"), + path("import/", OnboardingTaskFeedBulkImportView.as_view(), name="onboardingtask_import"), + path( + "/changelog/", + ObjectChangeLogView.as_view(), + name="onboardingtask_changelog", + kwargs={"model": OnboardingTask}, + ), ] diff --git a/netbox_onboarding/views.py b/netbox_onboarding/views.py index 7fcd612..79ab623 100644 --- a/netbox_onboarding/views.py +++ b/netbox_onboarding/views.py @@ -12,22 +12,85 @@ limitations under the License. """ import logging -from django.contrib.auth.mixins import PermissionRequiredMixin + +from django.shortcuts import get_object_or_404, render +from django.views.generic import View + from utilities.views import BulkDeleteView, BulkImportView, ObjectEditView, ObjectListView +from .release import NETBOX_RELEASE_CURRENT, NETBOX_RELEASE_29 + from .filters import OnboardingTaskFilter from .forms import OnboardingTaskForm, OnboardingTaskFilterForm, OnboardingTaskFeedCSVForm from .models import OnboardingTask from .tables import OnboardingTaskTable, OnboardingTaskFeedBulkTable -log = logging.getLogger("rq.worker") -log.setLevel(logging.DEBUG) +logger = logging.getLogger("rq.worker") + + +if NETBOX_RELEASE_CURRENT < NETBOX_RELEASE_29: + from django.contrib.auth.mixins import PermissionRequiredMixin # pylint: disable=ungrouped-imports + + class ReleaseMixinOnboardingTaskView(PermissionRequiredMixin, View): + """Release Mixin View for presenting a single OnboardingTask.""" + + permission_required = "netbox_onboarding.view_onboardingtask" + + class ReleaseMixinOnboardingTaskListView(PermissionRequiredMixin): + """Release Mixin View for listing all extant OnboardingTasks.""" + + permission_required = "netbox_onboarding.view_onboardingtask" + + class ReleaseMixinOnboardingTaskCreateView(PermissionRequiredMixin): + """Release Mixin View for creating a new OnboardingTask.""" + + permission_required = "netbox_onboarding.add_onboardingtask" + + class ReleaseMixinOnboardingTaskBulkDeleteView(PermissionRequiredMixin): + """Release Mixin View for deleting one or more OnboardingTasks.""" + + permission_required = "netbox_onboarding.delete_onboardingtask" + + class ReleaseMixinOnboardingTaskFeedBulkImportView(PermissionRequiredMixin): + """Release Mixin View for bulk-importing a CSV file to create OnboardingTasks.""" + + permission_required = "netbox_onboarding.add_onboardingtask" -class OnboardingTaskListView(PermissionRequiredMixin, ObjectListView): +else: + from utilities.views import ObjectView # pylint: disable=ungrouped-imports, no-name-in-module + + class ReleaseMixinOnboardingTaskView(ObjectView): + """Release Mixin View for presenting a single OnboardingTask.""" + + class ReleaseMixinOnboardingTaskListView: + """Release Mixin View for listing all extant OnboardingTasks.""" + + class ReleaseMixinOnboardingTaskCreateView: + """Release Mixin View for creating a new OnboardingTask.""" + + class ReleaseMixinOnboardingTaskBulkDeleteView: + """Release Mixin View for deleting one or more OnboardingTasks.""" + + class ReleaseMixinOnboardingTaskFeedBulkImportView: + """Release Mixin View for bulk-importing a CSV file to create OnboardingTasks.""" + + +class OnboardingTaskView(ReleaseMixinOnboardingTaskView): + """View for presenting a single OnboardingTask.""" + + queryset = OnboardingTask.objects.all() + + def get(self, request, pk): # pylint: disable=invalid-name, missing-function-docstring + """Get request.""" + onboardingtask = get_object_or_404(self.queryset, pk=pk) + + return render(request, "netbox_onboarding/onboardingtask.html", {"onboardingtask": onboardingtask,}) + + +class OnboardingTaskListView(ReleaseMixinOnboardingTaskListView, ObjectListView): """View for listing all extant OnboardingTasks.""" - permission_required = "netbox_onboarding.view_onboardingtask" queryset = OnboardingTask.objects.all().order_by("-id") filterset = OnboardingTaskFilter filterset_form = OnboardingTaskFilterForm @@ -35,30 +98,28 @@ class OnboardingTaskListView(PermissionRequiredMixin, ObjectListView): template_name = "netbox_onboarding/onboarding_tasks_list.html" -class OnboardingTaskCreateView(PermissionRequiredMixin, ObjectEditView): +class OnboardingTaskCreateView(ReleaseMixinOnboardingTaskCreateView, ObjectEditView): """View for creating a new OnboardingTask.""" - permission_required = "netbox_onboarding.add_onboardingtask" model = OnboardingTask queryset = OnboardingTask.objects.all() model_form = OnboardingTaskForm template_name = "netbox_onboarding/onboarding_task_edit.html" - default_return_url = "plugins:netbox_onboarding:onboarding_task_list" + default_return_url = "plugins:netbox_onboarding:onboardingtask_list" -class OnboardingTaskBulkDeleteView(PermissionRequiredMixin, BulkDeleteView): +class OnboardingTaskBulkDeleteView(ReleaseMixinOnboardingTaskBulkDeleteView, BulkDeleteView): """View for deleting one or more OnboardingTasks.""" - permission_required = "netbox_onboarding.delete_onboardingtask" queryset = OnboardingTask.objects.filter() # TODO: can we exclude currently-running tasks? table = OnboardingTaskTable - default_return_url = "plugins:netbox_onboarding:onboarding_task_list" + default_return_url = "plugins:netbox_onboarding:onboardingtask_list" -class OnboardingTaskFeedBulkImportView(PermissionRequiredMixin, BulkImportView): +class OnboardingTaskFeedBulkImportView(ReleaseMixinOnboardingTaskFeedBulkImportView, BulkImportView): """View for bulk-importing a CSV file to create OnboardingTasks.""" - permission_required = "netbox_onboarding.add_onboardingtask" + queryset = OnboardingTask.objects.all() model_form = OnboardingTaskFeedCSVForm table = OnboardingTaskFeedBulkTable - default_return_url = "plugins:netbox_onboarding:onboarding_task_list" + default_return_url = "plugins:netbox_onboarding:onboardingtask_list" diff --git a/netbox_onboarding/worker.py b/netbox_onboarding/worker.py index c26cfa8..e4202a3 100644 --- a/netbox_onboarding/worker.py +++ b/netbox_onboarding/worker.py @@ -12,61 +12,98 @@ limitations under the License. """ import logging -import time +from django.core.exceptions import ValidationError from django_rq import job +from prometheus_client import Summary +from dcim.models import Device + +from .choices import OnboardingFailChoices +from .choices import OnboardingStatusChoices +from .exceptions import OnboardException +from .metrics import onboardingtask_results_counter +from .models import OnboardingDevice from .models import OnboardingTask -from .onboard import NetboxKeeper, NetdevKeeper, OnboardException -from .choices import OnboardingStatusChoices, OnboardingFailChoices +from .onboard import OnboardingManager logger = logging.getLogger("rq.worker") -logger.setLevel(logging.DEBUG) +REQUEST_TIME = Summary("onboardingtask_processing_seconds", "Time spent processing onboarding request") + + +@REQUEST_TIME.time() @job("default") -def onboard_device(task_id, credentials): +def onboard_device(task_id, credentials): # pylint: disable=too-many-statements """Process a single OnboardingTask instance.""" username = credentials.username password = credentials.password secret = credentials.secret - try: - ot = OnboardingTask.objects.get(id=task_id) - except OnboardingTask.DoesNotExist: - # TODO: maybe we started before the DB was done writing it, or maybe it was deleted out from under us? - time.sleep(1) - ot = OnboardingTask.objects.get(id=task_id) + ot = OnboardingTask.objects.get(id=task_id) - logging.info("START: onboard device") + logger.info("START: onboard device") + onboarded_device = None try: + try: + if ot.ip_address: + onboarded_device = Device.objects.get(primary_ip4__address__net_host=ot.ip_address) + + if OnboardingDevice.objects.filter(device=onboarded_device, enabled=False): + ot.status = OnboardingStatusChoices.STATUS_SKIPPED + + return dict(ok=True) + + except Device.DoesNotExist as exc: + logger.info("Getting device with IP lookup failed: %s", str(exc)) + except Device.MultipleObjectsReturned as exc: + logger.info("Getting device with IP lookup failed: %s", str(exc)) + raise OnboardException( + reason="fail-general", message=f"ERROR Multiple devices exist for IP {ot.ip_address}" + ) + except ValueError as exc: + logger.info("Getting device with IP lookup failed: %s", str(exc)) + except ValidationError as exc: + logger.info("Getting device with IP lookup failed: %s", str(exc)) + ot.status = OnboardingStatusChoices.STATUS_RUNNING ot.save() - netdev = NetdevKeeper(ot, username, password, secret) - nbk = NetboxKeeper(netdev=netdev) + onboarding_manager = OnboardingManager(ot=ot, username=username, password=password, secret=secret) + + if onboarding_manager.created_device: + ot.created_device = onboarding_manager.created_device - netdev.get_required_info() - nbk.ensure_device() + ot.status = OnboardingStatusChoices.STATUS_SUCCEEDED + ot.save() + logger.info("FINISH: onboard device") + onboarding_status = True except OnboardException as exc: + if onboarded_device: + ot.created_device = onboarded_device + + logger.error("%s", exc) ot.status = OnboardingStatusChoices.STATUS_FAILED ot.failed_reason = exc.reason ot.message = exc.message ot.save() - # return dict(ok=False) - raise + onboarding_status = False + + except Exception as exc: # pylint: disable=broad-except + if onboarded_device: + ot.created_device = onboarded_device - except Exception as exc: + logger.error("Onboarding Error - Exception") + logger.error(str(exc)) ot.status = OnboardingStatusChoices.STATUS_FAILED ot.failed_reason = OnboardingFailChoices.FAIL_GENERAL ot.message = str(exc) ot.save() - raise + onboarding_status = False - logging.info("FINISH: onboard device") - ot.status = OnboardingStatusChoices.STATUS_SUCCEEDED - ot.save() + onboardingtask_results_counter.labels(status=ot.status).inc() - return dict(ok=True) + return dict(ok=onboarding_status) diff --git a/pyproject.toml b/pyproject.toml index 2c81e31..7029357 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "ntc-netbox-plugin-onboarding" -version = "1.3.0" +version = "2.0.0" description = "A plugin for NetBox to easily onboard new devices." authors = ["Network to Code, LLC "] license = "Apache-2.0"