Fixes SSH public key auth, misc document clean-up

changes/293.fixed

@@ -0,0 +1 @@
+Makes fixes to platform detection so that netmiko ssh pubkey auth settings are applied.

docs/user/app_getting_started.md

@@ -64,11 +64,7 @@ The new SSoT based jobs each use their own Nornir inventories.
 
 ### Onboarding a Device
 
-Navigate to the Device Onboarding Job: Jobs > Perform Device Onboarding (original).
-
-or
-
-Navigate to the SSoT dashboard and run `Sync Devices` to get basic device and information onboarding, followed by `Sync Network Data` to add additional details from the network to these devices. E.g. Interfaces, IPs, VRFs, VLANs.
+Navigate to the `Jobs` page from the nautobot navigation bar. Run `Sync Devices From Network` to get basic device and information onboarding, followed by `Sync Network Data From Network` to add additional details from the network to these devices. E.g. Interfaces, IPs, VRFs, VLANs.
 
 ## What are the next steps?
 

docs/user/app_use_cases.md

@@ -107,7 +107,7 @@ PLUGINS_CONFIG = {
             "netmiko": {
                 "extras": {
                     "use_keys": True,
-                    "key_file": "/root/.ssh/id_rsa.pub",
+                    "key_file": "/root/.ssh/id_rsa",
                     "disabled_algorithms": {"pubkeys": ["rsa-sha2-256", "rsa-sha2-512"]},
                 },
             },
@@ -116,7 +116,7 @@ PLUGINS_CONFIG = {
 }
 ```
 
-3. Make a secrets group in Nautobot which still had all the elements (username and password), where the username is accurate, a bogus password can be used as its ignored by the backend processing. For example, set the password to the username secret since its ignore.
+3. Make a secrets group in Nautobot which has the accurate `username` to use along with the key specified in configuration above.
 
 4. Run the jobs and ssh public key authentication will be used.
 

nautobot_device_onboarding/jobs.py

@@ -774,7 +774,7 @@ def run(self, *args, **kwargs):  # pragma: no cover
         ip_addresses = kwargs["ip_addresses"].replace(" ", "").split(",")
         port = kwargs["port"]
         platform = kwargs["platform"]
-        username, password, secret = (  # pylint:disable=unused-variable
+        username, password = (  # pylint:disable=unused-variable
             _parse_credentials(kwargs["secrets_group"])
         )
 

nautobot_device_onboarding/nornir_plays/command_getter.py

@@ -2,13 +2,13 @@
 
 import json
 import os
-from typing import Dict
+from typing import Dict, Tuple, Union
 
 from django.conf import settings
 from nautobot.dcim.models import Platform
 from nautobot.dcim.utils import get_all_network_driver_mappings
 from nautobot.extras.choices import SecretsGroupAccessTypeChoices, SecretsGroupSecretTypeChoices
-from nautobot.extras.models import SecretsGroup
+from nautobot.extras.models import SecretsGroup, SecretsGroupAssociation
 from nautobot_plugin_nornir.constants import NORNIR_SETTINGS
 from nautobot_plugin_nornir.plugins.inventory.nautobot_orm import NautobotORMInventory
 from netutils.ping import tcp_ping
@@ -224,8 +224,10 @@ def netmiko_send_commands(
             task.results[result_idx].failed = False
 
 
-def _parse_credentials(credentials):
-    """Parse and return dictionary of credentials."""
+def _parse_credentials(credentials: Union[SecretsGroup, None], logger: NornirLogger = None) -> Tuple[str, str]:
+    """Parse creds from either secretsgroup or settings, return tuple of username/password."""
+    username, password = None, None
+
     if credentials:
         try:
             username = credentials.get_secret_value(
@@ -236,20 +238,22 @@ def _parse_credentials(credentials):
                 access_type=SecretsGroupAccessTypeChoices.TYPE_GENERIC,
                 secret_type=SecretsGroupSecretTypeChoices.TYPE_PASSWORD,
             )
-            try:
-                secret = credentials.get_secret_value(
-                    access_type=SecretsGroupAccessTypeChoices.TYPE_GENERIC,
-                    secret_type=SecretsGroupSecretTypeChoices.TYPE_SECRET,
-                )
-            except Exception:  # pylint: disable=broad-exception-caught
-                secret = None
-        except Exception:  # pylint: disable=broad-exception-caught
-            return (None, None, None)
+        except SecretsGroupAssociation.DoesNotExist:
+            pass
+        except Exception as e:  # pylint: disable=broad-exception-caught
+            logger.debug(f"Error processing credentials from secrets group {credentials.name}: {e}")
+            pass
     else:
         username = settings.NAPALM_USERNAME
         password = settings.NAPALM_PASSWORD
-        secret = settings.NAPALM_ARGS.get("secret", None)
-    return (username, password, secret)
+
+    missing_creds = []
+    for cred_var in ["username", "password"]:
+        if not locals().get(cred_var, None):
+            missing_creds.append(cred_var)
+    if missing_creds:
+        logger.debug(f"Missing credentials for {missing_creds}")
+    return (username, password)
 
 
 def sync_devices_command_getter(job_result, log_level, kwargs):
@@ -265,7 +269,7 @@ def sync_devices_command_getter(job_result, log_level, kwargs):
         port = kwargs["port"]
         # timeout = kwargs["timeout"]
         platform = kwargs["platform"]
-        username, password, secret = _parse_credentials(kwargs["secrets_group"])
+        username, password = _parse_credentials(kwargs["secrets_group"], logger=logger)
 
     # Initiate Nornir instance with empty inventory
     try:
@@ -297,7 +301,7 @@ def sync_devices_command_getter(job_result, log_level, kwargs):
                         if new_secrets_group != loaded_secrets_group:
                             logger.info(f"Parsing credentials from Secrets Group: {new_secrets_group.name}")
                             loaded_secrets_group = new_secrets_group
-                            username, password, secret = _parse_credentials(loaded_secrets_group)
+                            username, password = _parse_credentials(loaded_secrets_group, logger=logger)
                             if not (username and password):
                                 logger.error(f"Unable to onboard {entered_ip}, failed to parse credentials")
                         single_host_inventory_constructed, exc_info = _set_inventory(

nautobot_device_onboarding/nornir_plays/inventory_creator.py

@@ -1,14 +1,18 @@
 """Inventory Creator and Helpers."""
 
+from typing import Dict, Tuple, Union
+
 from netmiko import SSHDetect
 from nornir.core.inventory import ConnectionOptions, Host
 
 from nautobot_device_onboarding.constants import NETMIKO_EXTRAS
 
 
-def guess_netmiko_device_type(hostname, username, password, port):
+def guess_netmiko_device_type(
+    hostname: str, username: str, password: str, port: str
+) -> Tuple[str, Union[Exception, None]]:
     """Guess the device type of host, based on Netmiko."""
-    netmiko_optional_args = {"port": port}
+    netmiko_optional_args = {"port": port, **NETMIKO_EXTRAS}
     guessed_device_type = None
 
     remote_device = {
@@ -30,7 +34,9 @@ def guess_netmiko_device_type(hostname, username, password, port):
     return guessed_device_type, guessed_exc
 
 
-def _set_inventory(host_ip, platform, port, username, password):
+def _set_inventory(
+    host_ip: str, platform: str, port: str, username: str, password: str
+) -> Tuple[Dict, Union[Exception, None]]:
     """Construct Nornir Inventory."""
     inv = {}
     if platform:

nautobot_device_onboarding/tests/test_command_getter.py

@@ -2,10 +2,15 @@
 
 import os
 import unittest
+from unittest.mock import MagicMock, patch
 
 import yaml
+from nautobot.core.testing import TransactionTestCase
+from nautobot.extras.choices import SecretsGroupAccessTypeChoices, SecretsGroupSecretTypeChoices
+from nautobot.extras.models import Secret, SecretsGroup, SecretsGroupAssociation
 
-from nautobot_device_onboarding.nornir_plays.command_getter import _get_commands_to_run
+from nautobot_device_onboarding.nornir_plays.command_getter import _get_commands_to_run, _parse_credentials
+from nautobot_device_onboarding.nornir_plays.logger import NornirLogger
 
 MOCK_DIR = os.path.join("nautobot_device_onboarding", "tests", "mock")
 
@@ -217,3 +222,60 @@ def test_deduplicate_command_list_sync_data_cables(self):
             },
         ]
         self.assertEqual(get_commands_to_run, expected_commands_to_run)
+
+
+@patch("nautobot_device_onboarding.nornir_plays.command_getter.NornirLogger", MagicMock())
+class TestSSHCredParsing(TransactionTestCase):
+    """Tests against the _parse_credentials helper function."""
+
+    databases = ("default", "job_logs")
+
+    def setUp(self):  # pylint: disable=invalid-name
+        """Initialize test case."""
+        username_secret, _ = Secret.objects.get_or_create(
+            name="username", provider="environment-variable", parameters={"variable": "DEVICE_USER"}
+        )
+        password_secret, _ = Secret.objects.get_or_create(
+            name="password", provider="environment-variable", parameters={"variable": "DEVICE_PASS"}
+        )
+        self.secrets_group, _ = SecretsGroup.objects.get_or_create(name="test secrets group")
+        SecretsGroupAssociation.objects.get_or_create(
+            secrets_group=self.secrets_group,
+            secret=username_secret,
+            access_type=SecretsGroupAccessTypeChoices.TYPE_GENERIC,
+            secret_type=SecretsGroupSecretTypeChoices.TYPE_USERNAME,
+        )
+        SecretsGroupAssociation.objects.get_or_create(
+            secrets_group=self.secrets_group,
+            secret=password_secret,
+            access_type=SecretsGroupAccessTypeChoices.TYPE_GENERIC,
+            secret_type=SecretsGroupSecretTypeChoices.TYPE_PASSWORD,
+        )
+
+    @patch.dict(os.environ, {"DEVICE_USER": "admin", "DEVICE_PASS": "worstP$$w0rd"})
+    def test_parse_user_and_pass(self):
+        """Extract correct user and password from secretgroup env-vars"""
+        assert _parse_credentials(credentials=self.secrets_group, logger=NornirLogger(job_result={}, log_level=1)) == (
+            "admin",
+            "worstP$$w0rd",
+        )
+
+    @patch.dict(os.environ, {"DEVICE_USER": "admin"})
+    def test_parse_user_missing_pass(self):
+        """Extract just the username without bailing out if password is missing"""
+        mock_job_result = MagicMock()
+        assert _parse_credentials(
+            credentials=self.secrets_group, logger=NornirLogger(job_result=mock_job_result, log_level=1)
+        ) == ("admin", None)
+        mock_job_result.log.assert_called_with("Missing credentials for ['password']", level_choice="debug")
+
+    @patch(
+        "nautobot_device_onboarding.nornir_plays.command_getter.settings",
+        MagicMock(NAPALM_USERNAME="napalm_admin", NAPALM_PASSWORD="napalamP$$w0rd"),
+    )
+    def test_parse_napalm_creds(self):
+        """When no secrets group is provided, fallback to napalm creds"""
+        assert _parse_credentials(credentials=None, logger=NornirLogger(job_result=None, log_level=1)) == (
+            "napalm_admin",
+            "napalamP$$w0rd",
+        )

nautobot_device_onboarding/tests/test_inventory_creator.py

@@ -1,7 +1,7 @@
 """Test ability to create an inventory."""
 
 import unittest
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 from nautobot.dcim.models import Platform
 
@@ -45,3 +45,19 @@ def test_set_inventory_specified_platform(self):
         inv, exception = _set_inventory(self.host_ip, self.platform, self.port, self.username, self.password)
         self.assertEqual(inv["198.51.100.1"].platform, self.platform.name)
         self.assertIsNone(exception)
+
+    @patch("nautobot_device_onboarding.nornir_plays.inventory_creator.NETMIKO_EXTRAS", {"custom_setting": "enabled"})
+    @patch("nautobot_device_onboarding.nornir_plays.inventory_creator.SSHDetect")
+    def test_guess_netmiko_pass_netmiko_extras(self, mock_sshdetect: MagicMock):
+        """Ensure that we are passing additional Netmiko extras to the SSHDetect method.
+        These would have been fed in via the user in the nautobot configuration file."""
+        mock_sshdetect.return_value.autodetect.return_value = "cisco_ios"
+        guess_netmiko_device_type(self.hostname, self.username, self.password, self.port)
+        mock_sshdetect.mock_calls[0].assert_called_with(
+            device_type="autodetect",
+            host=self.hostname,
+            username=self.username,
+            password=self.password,
+            port=22,
+            custom_setting="enabled",
+        )