Add handling of device secrets

Author: glennmatthews

netbox_onboarding/api/serializers.py

@@ -32,6 +32,8 @@ class OnboardingTaskSerializer(serializers.ModelSerializer):
 
     password = serializers.CharField(required=False, write_only=True, help_text="Device password",)
 
+    secret = serializers.CharField(required=False, write_only=True, help_text="Device secret password",)
+
     site = serializers.SlugRelatedField(
         many=False,
         read_only=False,
@@ -63,7 +65,7 @@ class OnboardingTaskSerializer(serializers.ModelSerializer):
 
     status = serializers.CharField(required=False, help_text="Onboarding Status")
 
-    failed_raison = serializers.CharField(required=False, help_text="Failure reason")
+    failed_reason = serializers.CharField(required=False, help_text="Failure reason")
 
     message = serializers.CharField(required=False, help_text="Status message")
 
@@ -79,12 +81,13 @@ class Meta:  # noqa: D106 "Missing docstring in public nested class"
             "ip_address",
             "username",
             "password",
+            "secret",
             "site",
             "role",
             "device_type",
             "platform",
             "status",
-            "failed_raison",
+            "failed_reason",
             "message",
             "port",
             "timeout",
@@ -95,8 +98,9 @@ def create(self, validated_data):
         # Fields are string-type so default to empty (instead of None)
         username = validated_data.pop("username", "")
         password = validated_data.pop("password", "")
+        secret = validated_data.pop("secret", "")
 
-        credentials = Credentials(username=username, password=password)
+        credentials = Credentials(username=username, password=password, secret=secret,)
 
         ot = OnboardingTask.objects.create(**validated_data)
 

netbox_onboarding/forms.py

@@ -37,6 +37,9 @@ class OnboardingTaskForm(BootstrapMixin, forms.ModelForm):
     password = forms.CharField(
         required=False, widget=forms.PasswordInput, help_text="Device password (will not be stored in database)"
     )
+    secret = forms.CharField(
+        required=False, widget=forms.PasswordInput, help_text="Device secret (will not be stored in database)"
+    )
 
     platform = forms.ModelChoiceField(
         queryset=Platform.objects.all(),
@@ -66,6 +69,7 @@ class Meta:  # noqa: D106 "Missing docstring in public nested class"
             "timeout",
             "username",
             "password",
+            "secret",
             "platform",
             "role",
             "device_type",
@@ -75,7 +79,7 @@ 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"))
+            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)
         return model
 
@@ -113,6 +117,7 @@ class OnboardingTaskFeedCSVForm(CustomFieldModelCSVForm):
     ip_address = forms.CharField(required=True, help_text="IP Address of the onboarded device")
     username = forms.CharField(required=False, help_text="Username, will not be stored in database")
     password = forms.CharField(required=False, help_text="Password, will not be stored in database")
+    secret = forms.CharField(required=False, help_text="Secret password, will not be stored in database")
     platform = forms.ModelChoiceField(
         queryset=Platform.objects.all(),
         required=False,
@@ -140,6 +145,6 @@ 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"))
+            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)
         return model

netbox_onboarding/models.py

@@ -58,3 +58,7 @@ class OnboardingTask(models.Model):
 
     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}"

netbox_onboarding/onboard.py

@@ -13,7 +13,6 @@
 """
 
 import logging
-import os
 import re
 import socket
 
@@ -70,13 +69,14 @@ def __str__(self):
 class NetdevKeeper:
     """Used to maintain information about the network device during the onboarding process."""
 
-    def __init__(self, onboarding_task, username=None, password=None):
+    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 environment variable will be used)
-          password (str): Device password (if unspecified, NAPALM_PASSWORD environment variable will be used)
+          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'):
@@ -92,8 +92,9 @@ def __init__(self, onboarding_task, username=None, password=None):
         self.serial_number = None
         self.mgmt_ifname = None
         self.mgmt_pflen = None
-        self.username = username or os.environ.get("NAPALM_USERNAME", None)
-        self.password = password or os.environ.get("NAPALM_PASSWORD", 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.
@@ -129,6 +130,7 @@ def guess_netmiko_device_type(**kwargs):
             "host": kwargs.get("host"),
             "username": kwargs.get("username"),
             "password": kwargs.get("password"),
+            "secret": kwargs.get("secret"),
         }
 
         try:
@@ -159,7 +161,7 @@ def get_platform_slug(self):
             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
+                host=self.ot.ip_address, username=self.username, password=self.password, secret=self.secret,
             )
 
         logging.info("PLATFORM NAME is %s", platform_slug)
@@ -233,7 +235,15 @@ def get_required_info(
                 )
 
             driver = get_network_driver(driver_name)
-            dev = driver(hostname=mgmt_ipaddr, username=self.username, password=self.password, timeout=self.ot.timeout)
+            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")
@@ -265,16 +275,13 @@ def get_mgmt_info():
                         return (if_name, if_addr_data["prefix_length"])
             return (default_mgmt_if, default_mgmt_pfxlen)
 
-        mgmt_ifname, mgmt_pflen = get_mgmt_info()
-
         # 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 = mgmt_ifname
-        self.mgmt_pflen = mgmt_pflen
+        self.mgmt_ifname, self.mgmt_pflen = get_mgmt_info()
 
 
 # -----------------------------------------------------------------------------

netbox_onboarding/utils/credentials.py

@@ -16,10 +16,11 @@
 class Credentials:
     """Class used to hide user's credentials in RQ worker and Django."""
 
-    def __init__(self, username=None, password=None):
+    def __init__(self, username=None, password=None, secret=None):
         """Create a Credentials instance."""
         self.username = username
         self.password = password
+        self.secret = secret
 
     def __repr__(self):
         """Return string representation of a Credentials object."""