Merge pull request #108 from networktocode/develop-2.0-napalm-support

Napalm and Netmiko optional arguments support

Author: mzbroch

netbox_onboarding/netdev_keeper.py

@@ -19,6 +19,7 @@
 from django.conf import settings
 from napalm import get_network_driver
 from napalm.base.exceptions import ConnectionException, CommandErrorException
+from napalm.base.netmiko_helpers import netmiko_args
 from netmiko.ssh_autodetect import SSHDetect
 from netmiko.ssh_exception import NetMikoAuthenticationException
 from netmiko.ssh_exception import NetMikoTimeoutException
@@ -62,7 +63,15 @@ 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
+        self,
+        hostname,
+        port=None,
+        timeout=None,
+        username=None,
+        password=None,
+        secret=None,
+        napalm_driver=None,
+        optional_args=None,
     ):
         """Initialize the network device keeper instance and ensure the required configuration parameters are provided.
 
@@ -83,17 +92,21 @@ def __init__(  # pylint: disable=R0913
         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.username = username
+        self.password = password
+        self.secret = secret
         self.napalm_driver = napalm_driver
+        self.optional_args = optional_args
 
         self.facts = None
         self.ip_ifs = None
         self.netmiko_device_type = None
         self.onboarding_class = StandaloneOnboarding
         self.driver_addon_result = None
 
+        # Enable loading driver extensions
+        self.load_driver_extension = True
+
     def check_reachability(self):
         """Ensure that the device at the mgmt-ipaddr provided is reachable.
 
@@ -120,14 +133,25 @@ def guess_netmiko_device_type(self):
         """Guess the device type of host, based on Netmiko."""
         guessed_device_type = None
 
+        netmiko_optional_args = netmiko_args(self.optional_args)
+
         remote_device = {
             "device_type": "autodetect",
             "host": self.hostname,
             "username": self.username,
             "password": self.password,
-            "secret": self.secret,
+            **netmiko_optional_args,
         }
 
+        if self.secret:
+            remote_device["secret"] = self.secret
+
+        if self.port:
+            remote_device["port"] = self.port
+
+        if self.timeout:
+            remote_device["timeout"] = self.timeout
+
         try:
             logger.info("INFO guessing device type: %s", self.hostname)
             guesser = SSHDetect(**remote_device)
@@ -201,8 +225,13 @@ def get_onboarding_facts(self):
             self.check_napalm_driver_name()
 
             driver = get_network_driver(self.napalm_driver)
-            optional_args = settings.NAPALM_ARGS.copy()
-            optional_args["secret"] = self.secret
+
+            optional_args = self.optional_args.copy()
+            if self.port:
+                optional_args["port"] = self.port
+
+            if self.secret:
+                optional_args["secret"] = self.secret
 
             napalm_device = driver(
                 hostname=self.hostname,
@@ -222,7 +251,7 @@ def get_onboarding_facts(self):
 
             module_name = PLUGIN_SETTINGS["onboarding_extensions_map"].get(self.napalm_driver)
 
-            if module_name:
+            if module_name and self.load_driver_extension:
                 try:
                     module = importlib.import_module(module_name)
                     driver_addon_class = module.OnboardingDriverExtensions(napalm_device=napalm_device)

netbox_onboarding/onboard.py

@@ -12,6 +12,8 @@
 limitations under the License.
 """
 
+import json
+
 from django.conf import settings
 
 from .netdev_keeper import NetdevKeeper
@@ -34,6 +36,16 @@ def napalm_driver(self):
 
         return None
 
+    @property
+    def optional_args(self):
+        """Return platform optional args."""
+        if self.ot.platform and self.ot.platform.napalm_args:
+            napalm_args = json.loads(self.ot.platform.napalm_args)
+
+            return napalm_args
+
+        return {}
+
     @property
     def ip_address(self):
         """Return ot's ip address."""
@@ -75,13 +87,13 @@ class OnboardingManager:
 
     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)
 
+        self.username = username or settings.NAPALM_USERNAME
+        self.password = password or settings.NAPALM_PASSWORD
+        self.secret = secret or otm.optional_args.get("secret", None) or settings.NAPALM_ARGS.get("secret", None)
+
         netdev = NetdevKeeper(
             hostname=otm.ip_address,
             port=otm.port,
@@ -90,6 +102,7 @@ def __init__(self, ot, username, password, secret):
             password=self.password,
             secret=self.secret,
             napalm_driver=otm.napalm_driver,
+            optional_args=otm.optional_args or settings.NAPALM_ARGS,
         )
 
         netdev.get_onboarding_facts()
@@ -116,6 +129,7 @@ def __init__(self, ot, username, password, secret):
         }
 
         onboarding_cls = netdev_dict["onboarding_class"]()
+        onboarding_cls.credentials = {"username": self.username, "password": self.password, "secret": self.secret}
         onboarding_cls.run(onboarding_kwargs=onboarding_kwargs)
 
         self.created_device = onboarding_cls.created_device

netbox_onboarding/onboarding/onboarding.py

@@ -21,6 +21,7 @@ class Onboarding:
     def __init__(self):
         """Init the class."""
         self.created_device = None
+        self.credentials = None
 
     def run(self, onboarding_kwargs):
         """Implement run method."""