Python Parallel SSH - Netmiko/Napalm - Cisco SMB switches stuck at sending command

Question

I am trying to determine vendor + version (using python NAPALM and parallel-ssh) of network switches (Huawei VRP5/8, Cisco Catalyst and Cisco SMB (SF/SG):

admin@server:~$ python3
Python 3.8.10 (default, Nov 26 2021, 20:14:08)
[GCC 9.3.0] on linux
Type "help", "copyright", "credits" or "license" for more information.

>>> from napalm import get_network_driver
>>> driver = get_network_driver('ios')
>>> device = driver('ip', 'username', 'password')
>>> device.open()
>>> print(device.get_facts())

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/altepro/.local/lib/python3.8/site-packages/napalm/ios/ios.py", line 811, in get_facts
    show_ver = self._send_command('show version')
  File "/home/altepro/.local/lib/python3.8/site-packages/napalm/ios/ios.py", line 165, in _send_command
    output = self.device.send_command(command)
  File "/home/altepro/.local/lib/python3.8/site-packages/netmiko/utilities.py", line 600, in wrapper_decorator
    return func(self, *args, **kwargs)
  File "/home/altepro/.local/lib/python3.8/site-packages/netmiko/base_connection.py", line 1694, in send_command
    raise ReadTimeout(msg)
netmiko.exceptions.ReadTimeout:
Pattern not detected: '\x1b\\[Ksg300\\-ab\\-1\\#' in output.

Things you might try to fix this:
1. Explicitly set your pattern using the expect_string argument.
2. Increase the read_timeout to a larger value.

Where sg300-ab-1 is sysname of the switch (Cisco SMB - sg300 in this case, but i have tested this on several versions and types of the SMB lineup)

Things that i have tried: Tried several version of netmiko, napalm (And its drivers including ios-350) and parallel-ssh. Tried several fresh linux servers with fresh install of napalm and parallel-ssh.

SSH is tested using the same server and credentials and it works without any problems.

When i use parallel-ssh the device doesnt even raise exception or timeout - it just goes stuck in the command:

output = client.run_command(cmd)

hosts = ['192.168.1.50']
client = ParallelSSHClient(hosts, user='my_user', password='my_pass')
cmd = 'show version'

output = client.run_command(cmd)
for host_out in output:
    for line in host_out.stdout:
        print(line)

Thanks for any kind of help !

Answer 1

It looks like the prompt isn't getting recognized properly. I'm not very familiar with either ParallelSSHClient or napalm, but I have worked with netmiko and that looks like where the error is. Here's some steps that can possible get you closer to figuring out what's happening. I suspect it's the prompt not being read correctly from the device.

Set up debugging and a netmiko session and run a simple command

import logging
import netmiko
logging.basicConfig(level=logging.DEBUG)

session = netmiko.ConnectHandler(
    host='192.168.1.50', 
    username='my_user', 
    password='my_pass', 
    device_type='cisco_ios')

results = session.send_command('show version')

If this fails with the same error, then it's the prompt (possibly the \x1b escape character). Try again but with a simpler expect_string, like what's expected at the end of the prompt:

session.send_command('show version', expect_string="#")

If this gets you a result, then it's something about the how the prompt is being set for this device.

To see what's being found for the prompt:

session.find_prompt()

Edit:

Based on what you're reporting, the issue seems to be with the control code \x1b\[ being included in the prompt. It's possible this can be disabled on the device itself, but I'm unfamiliar with that platform. The napalm API doesn't expose netmiko's send_command method. It should still be fixable. This solution would be a hack to make things work, nothing that I'd recommend relying on.

Establish a class that will act as your fix. This will be instantiated with the netmiko session (device.device) and will be used to replace the send_command method.

class HackyFix:
    def __init__(self, session):
        self.session = session
        self.original_send_command = session.send_command

    def send_command(self, command):
        original_prompt = self.session.find_prompt()
        fixed_prompt = original_prompt.replace(r"\x1b[", "")
        print(
            f"send_command intercepted. {original_prompt} replaced with {fixed_prompt}"
        )
        return self.original_send_command(command, expect_string=fixed_prompt)

Then in your existing napalm code, add this right after device.open():

hackyfix = HackyFix(device.device)
device.device.send_command = hackyfix.send_command

Now all of napalm's calls to send_command will go through your custom fix that will find the prompt and modify it before passing it to expect_string.

Last edit.

It's an ANSI Escape Code that's being thrown in by the SG300. Specifically it's the one that clears from cursor to end of line. It's also a known issue with the SG300. The good news is that someone made a napalm driver to support it. One big difference between the SG300 driver and the IOS driver is the netmiko device_type is cisco_s300. When this device_type is used, strip_ansi_escape_codes is ran against the output.

Behavior of that escape code tested in bash:

$ printf "This gets cleared\r"; code="\x1b[K"; printf "${code}This is what you see\n"
This is what you see

You can validate that setting cisco_s300 as the device_type fixes the issue:

session = netmiko.ConnectHandler(
    host='192.168.1.50', 
    username='my_user', 
    password='my_pass', 
    device_type='cisco_s300')

results = session.send_command('show version')

This should give a result with no modification to the expect_string value. If that works and you're looking to get results sooner or later, the following is a better fix than the hacky fix above.

from napalm.ios import IOSDriver

class QuickCiscoSG300Driver(IOSDriver):
    def __init__(self, hostname, username, password, timeout=60, optional_args=None):
        super().__init__(hostname, username, password, timeout, optional_args)

    def open(self):
        device_type = "cisco_s300"
        self.device = self._netmiko_open(
            device_type, netmiko_optional_args=self.netmiko_optional_args
        )


device = QuickCiscoSG300Driver("192.168.1.50", "my_user", "my_pass")
device.open()
device.get_facts()

Or you can get the driver (better option, unless this happens to be the driver you already tried)