Credential check on devices

[sidhuharj]

Hello,

I'm wondering if someone can help me understand what I am doing wrong.

I have a number of devices and a list of credentials. I would like to use nornir to test the credentials against a device and once a credential is successful stop any further attempts.

So far, I am able to go through a list and receive a login authentication error for all the failed attempts, but when it finds a successful match, it continues to run the task again and any further attempts report a success even though it is using a previous login cred that had failed. In this example, test fails and good works, by third run of the task test credentials report as successful. I figured the "on_good=True" would prevent any subsequent runs of the task.

Thank you.

`from nornir import InitNornir
from nornir_utils.plugins.functions import print_result
from nornir.plugins.inventory.simple import SimpleInventory
from nornir.core.plugins.inventory import InventoryPluginRegister
from nornir_netmiko.tasks import netmiko_send_command,netmiko_send_config
from nornir_napalm.plugins.tasks import napalm_get
from nornir.core.exceptions import NornirSubTaskError
from netmiko.ssh_exception import NetMikoAuthenticationException, NetMikoTimeoutException


def get_version(task):

    try:
        # result = task.run(task=netmiko_send_config, config_file=config)
        result = task.run(task=napalm_get, getters=["config"])
        print (f"{task.host.name} - Login OK - {task.host.username}")
        quit()

    except NornirSubTaskError as e:
        if isinstance(e.result.exception, NetMikoTimeoutException):
            print(f"{task.host.name} : Error! Timeout")
        elif isinstance(e.result.exception, NetMikoAuthenticationException):
            print(f"{task.host.name} : Error! Failed authentication")
        else:
            raise e

def main():

    InventoryPluginRegister.register("SimpleInventory", SimpleInventory)

    nr = InitNornir(
        runner={
            "plugin": "threaded",
        },
        inventory={
            "plugin": "SimpleInventory",
            "options": {
                    "host_file": "./inventory/hosts.yaml",
                    "group_file": "./inventory/groups.yaml",
                    "defaults_file": "./inventory/defaults.yaml"
            },
        }
    )
    # with open('./cred_check/cred.txt', 'r') as reader:
    #     cred_list=reader.read().splitlines()
    cred_list = ["test:test", "good:good", "test:test"]

    for c in cred_list:
        cred = c.split(":")

        userid = cred[0]
        passwd = cred[1]
        for hostname, host_obj in nr.inventory.hosts.items():
            host_obj.username=userid
            host_obj.password=passwd
        results = nr.run(task=get_version,on_failed=True,on_good=True)



if __name__ == "__main__":
    main()
`

[dgi-cglavan]

Not sure if you ever got this figured out, but I came across your code while looking for a similar solution. I made some changes to the get_version function that resulted in a functional script:

rhosts = []

def get_version(task):
    result = task.run(task=napalm_get, getters=["get_facts"])
    version = result[0].result["get_facts"]["os_version"]
    found_flag = False

    try:
        if not version:
            print(f"{task.host.name}: authentication failed!")
        elif task.host.name in rhosts:
            found_flag = True
        else:
            print(f"{task.host.name}: login OK - {task.host.username}")
            rhosts.append(task.host.name)

    except NornirSubTaskError as e:
        if isinstance(e.result.exception, NetMikoTimeoutException):
            print(f"{task.host.name} : Error! Timeout")
        elif isinstance(e.result.exception, NetMikoAuthenticationException):
            print(f"{task.host.name} : Error! Failed authentication")
        else:
            raise e
    return()

The function now checks to see if the host has already been tried for each credential set in our list, and once we find a good set we append it to the list and bypass it on future tries. We also return() the output of the function instead of quit() to make sure we iterate through the entire set of hosts and all credential sets in the list, and only raise an error if no credentials in the list return a result.

I'm still a novice so this method likely isn't the most efficient. Feedback welcome.

Hope this helps!

[dbarrosop]

I wouldn't do it like this. It is subverting a bit how the connections work. I think a better approach would be to test connecting to the device manually:

https://nornir.readthedocs.io/en/latest/howto/handling_connections.html#Manually

Also, careful with on_good and on_failed. Based on the initial comment I don't think it was understood (of course it is going to run after a succesful run, you are saying so with on_good=True). Check the last example: https://nornir.readthedocs.io/en/latest/tutorial/failed_tasks.html#Skipped-hosts

[dgi-cglavan]

But if the idea is to try more than one set of credentials for a given host, don't you want to force so it will exhaust the entire set of credentials before failing? Is there a better way to do this utilizing on_good/on_fail, or do you suggest a different approach?

Thanks for linking the manual connection, I hadn't seen that before (still relatively new to Nornir and admittedly haven't read all the docs yet).

EDIT: I'm testing in an environment where hosts have one of two known working credential sets. I get a ton of tracebacks when napalm fails to connect to a device using the credential set that doesn't match the device.

[dbarrosop]

your custom task to check credentials could do somehting like (pseudo-code-ish):

def check_creds(task: Task, creds: List[...], ...):
       for cred in credentials:
            try:
                   # try to open connections with cred
                   print("success")
                   return Result(...)
            except WhateveExceptionIsThrownIfCredsFailed:
                   print("failure")
       print("we exhausted credentials and found no valid ones")
       return Result(...)

[tkdebnath]

I am using multiple credentials for login from many years but just saw this question. Here is the code

# import creentials
from secret.credentials import credentials

# progress bar
from tqdm import tqdm

# create directory
from pathlib import Path

# importing nornir and related plugins
from nornir import InitNornir
from nornir_netmiko.tasks import netmiko_send_command
from nornir_utils.plugins.functions import print_result
from nornir_utils.plugins.tasks.files import write_file
from nornir_utils.plugins.inventory.yaml_inventory import ConnectionOptions

#import exceptions
from paramiko.ssh_exception import AuthenticationException
from netmiko.exceptions import NetmikoAuthenticationException
from nornir.core.exceptions import NornirSubTaskError

import ipdb
from datetime import datetime, timezone
import subprocess
import os

from prettytable import PrettyTable
import requests
import json
import pandas as pd

# preparing login to the hosts
def backup(task, progress_bar):
    for cred in credentials:        
        try:
            task.host.username = cred['user']
            task.host.password = cred['pass']
            task.host.connection_options['netmiko'] = ConnectionOptions(extras={'secret': cred['secret']})
            site = task.host.data['pynautobot_dictionary']['site']['name']

            # directory check if exist else create
            Path(f"{dir_path}/{site}").mkdir(parents=True, exist_ok=True)


            # Task 1. Run the Netmiko config getter to collect the config
            device_config = task.run(name="Fetching Running Config", task=netmiko_send_command, command_string="show running-config", enable=True, read_timeout=900)

            # Task 2. Write the device config to a file using the Nornir, write_file task
            task.run(
                task=write_file,
                content=device_config.result,
                filename=f"{dir_path}/{site}/{task.host}.txt",)
            
            # Task 3. Update success
            task.host['fact_status'] = "Backup Complete"
            
            progress_bar.update()
            
            # login successfull
            break

        except AuthenticationException or NetmikoAuthenticationException:
            continue
        except NornirSubTaskError:
            continue
        except:
            continue

if __name__=="__main__":
    dir_path = "/root/plugin/configuration-backup/running-config"
    nr = InitNornir(
    inventory={
        "plugin": "NautobotInventory",
        "options": {
            "nautobot_url": "http://",
            "nautobot_token": "",
            "ssl_verify": False,
        },
    },
    runner={
       "plugin": "threaded",
       "options": {'num_workers': 30} 
    },
    logging={
       "enabled": False
    },
)

    with tqdm(total=len(nr.inventory.hosts)) as progress_bar:
        results = nr.run(task=backup, progress_bar=progress_bar)

credentials = [ {'sequence': 1, 'user': 'first_user', 'pass': 'first_user_pass', 'secret': 'enable_pass'}, {'sequence': 2, 'user': 'second_user', 'pass': 'second_user_pass', 'secret': 'enable_pass'}, {'sequence': 3, 'user': 'xxxxx', 'pass': 'xxxxxx', 'secret': 'xxxxxxx'}, {'sequence': 999, 'user': 'last_user', 'pass': 'last_user_pass', 'secret': 'enable_pass'} ]