Python SSH and Telnet in 2 minutes with sdncore

Have you ever needed to run some commands on many devices? If you don’t, it is just a matter of time. Instead, if you did, you know how painful it can be. Even if the task is easy, like checking the local users on a device, the implementation isn’t. For example, you will have to detect when the device returned all the output. How do you know that some output isn’t yet to come? All this complexity can translate our easy task into a nightmare. Well, my friend, not anymore! With sdncore, you can implement python SSH and Telnet connections in literally less than a minute.

Getting SDNCore

If you are following our articles about SDN, you’ll know about SDNCore. SDNCore is our network automation project, a Python Library for network management. It includes all you need to implement easy Python SSH connections and Python Telnet connections, as well as SNMP and more. We are currently developing this project and posting updates here on ICTShore.com.

To get yourself running, you just need to install sdncore. This is easy, as we publish even the core development versions on PyPi. Thus, you can simply use pip.

pip install sdncore

Python SSH & Telnet with SDNCore

To understand how easy it is to use SDNCore for Python SSH and Telnet connections, just look at the following snippet.

from sdncore.vty.session import Session


with Session('10.0.0.1', username='admin', password='cisco', driver=Session.SSH) as session:
    session.command('terminal length 0')
    print(session.command('show run'))
    print(session.command('show ip interface brief'))

You just create a Session, and then you can send commands to it. The Session is smart enough to detect when the output finishes because it understands the prompt of the device. For example, if you connect to Router#, it will know that the command finishes when it finds Router# in the output. You can customize the behavior, as you will see below.

Switching between SSH and Telnet

This is extremely useful when you migrate your Telnet devices to SSH. The same script you wrote for Telnet can work on SSH, and vice versa. You simply need to change the driver, which is SSH by default.

Session('10.0.0.1', username='admin', password='cisco', driver=Session.Telnet)

Telnet Caveats

SSH implements authentication in the protocol itself, but Telnet doesn’t. With Telnet, you need to exchange credentials by communicating directly with the device. By default, our session will provide the username when it encounters username: , and the password when it finds password: . Not all devices ask the password in this way, so you may want to change these strings. Easy enough, just provide them as driver parameters.

Session('10.0.0.1', username='admin', password='cisco', driver=Session.Telnet,
             driver_parameters={
                 'username_finder': 'user: ',
                 'password_finder': 'psw: '
             })

A little hint in finding the prompt…

To detect the prompt of the device, the Session uses a special stop string. Session expects to find the string at the end of each prompt. By default, this string is #, as we expect the prompt to end with that character. However, you can change it by providing stop_character.

Session('10.0.0.1', username='admin', password='cisco', driver=Session.Telnet, stop_character='>')

If you want, you can also change it on the fly by modifying the stop_character member of the instance. You can do it any time you want.

Note that our Session is smart enough, and will use the stop character alone only to find the prompt. For example, if the prompt switches from R1# to R1(config)#, here we are going to use the stop_character to identify the new prompt. In all other cases, the whole prompt (e.g. R1#) is used, so we are sure we don’t stop where we shouldn’t.

Sending commands and reading the output

With sdncore, we get rid of all the pain that we used to experience when writing Python SSH or Telnet code. This is especially true when it comes to sending commands and reading the output.

By default, the command() method just wants to know the command. It will send it to the remote device and return the output. In that output, it will remove the first and last line. This is because in the first line we just see the command we entered, and in the last we just see the prompt. It will assume the prompt remains unchanged. Another important item here is that the command will run for all the time needed to find the prompt, and block indefinitely if it is never found.

Tuning our command

Luckily, our command() method comes with several flags and optional parameters we can set to adjust its behavior.

• timeout represents how many seconds to wait before terminating the execution, no matter what. By default is None, which means to block indefinitely. If you specify the timeout, but the prompt is found first, the session understands it has all the output it needs and won’t wait for the remaining time unnecessarily.
• custom_stopper is a string we want to provide if we don’t want to stop to the prompt, but to something else.
• changes_prompt is a boolean flag that defaults to False. If True, it tells the session that this commands only changes the prompt, and nothing else. The session will run the command and identify the new prompt, based on the stop_character.
• show_command is another boolean flag defaulting to False. If True, the first line of the output won’t be removed.

More elaboration on a single output

Besides returning it, the command() method stores the output in a buffer. That buffer always contains the output of the last command, in case we need to perform multiple checks on it. You can access this buffer in two ways.

• session.buffer is the raw buffer, containing the first line and last line
• session.cleaned_buffer() returns the buffer without the first line and last line. You have control on showing or hiding the first line by providing the optional parameter remove_first.

Opening and Closing

If you don’t want to use Session through the with statement, you need to open and close the session. Like any other Python SSH or Telnet implementation, you use session.open() to start the session. After that, you can send commands and, once you finish, you can call session.close().

The code behind SDNCore sessions

If you just care about using our vty session, you can just scroll down to the conclusion. However, if you want to understand what happens inside SDNCore, read on.

Session Structure

To put it short, our Session acts as a middle man with the chosen VTY driver and adds some intel. The driver is responsible for all the nasty stuff, like handling the timing and reading the output as long as there is output to read. Instead, the session coordinates the driver by providing the stop character, consolidating the output, and so on. The following schema should give you an idea about it.

For the nasty part of the code, you should check the driver itself. Don’t worry, we already covered them in previous articles here on ICTShore.com.

• You can find the details of the Telnet driver on this article about SDN and Telnet
• Instead, we covered the SSH driver in this article about Paramiko, and then in another to implement shell-based SSH

The Session Code

Driver Selection

First, we define three subclasses to represent the driver to use. We will use them for reference so that the final user doesn’t need to include vty.drivers.

class Session:
    class Driver:
        def __init__(self):
            ReferenceError('Session.Driver may not be instantiated')

    class SSH(Driver):
        def __init__(self):
            super().__init__()

    class Telnet(Driver):
        def __init__(self):
            super().__init__()

Construction, open and close

The constructor stores the parameters and starts the hook – the instance of the driver. The functions to open and close the session basically relay the command to the hook. However, as soon as you open the connection, the session will detect the prompt.

def __init__(self, target, port=None,
             username='', password='', driver=SSH, driver_parameters={},
             stop_character='#'):
    self.target = target
    self.port = port
    self.username = username
    self.password = password
    self.stop_character = stop_character
    self.buffer = ''
    self.prompt = self.stop_character
    if driver == self.SSH:
        self.driver = ssh.SSHDriver
        if self.port is None:
            self.port = 22
    elif driver == self.Telnet:
        self.driver = telnet.TelnetDriver
        if self.port is None:
            self.port = 23
    self.hook = self.driver(
        target=self.target,
        port=self.port,
        username=self.username,
        password=self.password,
        **driver_parameters
    )

def open(self):
    self.hook.open()
    self.buffer = self.hook.read_until(self.stop_character)
    self.detect_prompt()

def close(self):
    self.hook.close()

And we also implement __enter__ and __exit__ to support the with statement.

def __enter__(self):
    self.open()
    return self

def __exit__(self, type, value, traceback):
    self.close()

Command & Buffer

As you can see, the code behind the execution of a command is relatively simple. It just processes the parameters and constructs the calls to the hook and to the cleaned buffer. Just note that we create a dictionary from the timeout (pass_timeout). This is because the driver may not support simply providing a None timeout, so we just provide it if it isn’t None. The cleaned buffer, instead, simply removes lines and recompose the output.

def cleaned_buffer(self, remove_first=True):
    lines = self.buffer.split('\n')
    if remove_first and len(lines) > 0:
        lines.pop(0)
    last = lines.pop()
    if last != self.prompt:
        lines.append(last)
    return '\n'.join(lines)

def command(self, command, timeout=None, custom_stopper=None, changes_prompt=False, show_command=False):
    self.hook.send_text(command + '\n')
    if changes_prompt:
        self.detect_prompt()
        return
    stopper = self.prompt
    if custom_stopper is not None:
        stopper = custom_stopper
    pass_timeout = {}
    if timeout is not None:
        pass_timeout = {'timeout': timeout}
    self.buffer = self.hook.read_until(stopper, **pass_timeout)
    return self.cleaned_buffer(remove_first=not show_command)

Prompt Detection

The function that detects the prompt is even simpler. It just splits the buffer, fetches the last line (if any), and set it to the prompt.

def detect_prompt(self):
    lines = self.buffer.split('\n')
    if len(lines) > 0:
        last_line = lines.pop()
        if len(last_line) > 0:
            self.prompt = last_line
    return self.prompt

And there is more!

We provided all the code inside vty.session in this article, except docstrings. If you want to see the last version and the complete code, consult the GitHub sdncore project. Or, simply check the vty.session file on GitHub.

Wrapping it up

Hopefully, with sdncore you will automate your network configuration in minutes, being able to execute complex stuff on large infrastructure. Try sdncore, and let me know what you think. Did you find it useful? What was your use case? And if you want to contribute, consider forking on GitHub!