Download running-config form network device over ssh

.gitignore

__pycache__
*.cfg
.idea

README.md

GET SSH

Download data from network device over SSH protocol

ssh_get.py

#!/usr/bin/python3

__version__ = "0.3"
__copyright__ = "Vladimir Kushnir aka Kvantum i(c)2019"

import argparse
import re
from ssh_utils import get_terminal_data


# TODO: add enable support

def get_options(arguments=None):
    """Load commandline arguments"""
    parser = argparse.ArgumentParser(usage='%(prog)s [options] host1 [host2 ... [hocstn]]',
                                     description='Connect to SSH terminal and grab running-config output')
    parser.add_argument('--version', action='version', version='%(prog)s ' + __version__)
    parser.add_argument('-c', '--commands', dest="commands", nargs="+", default=["show running-config"])
    parser.add_argument('-a', '--write-all', dest="write_all", action='store_true', default=False)

    auth_group = parser.add_argument_group('Authentification')
    auth_group.add_argument("-u", "--username", dest="username", help="User name", required=True)
    auth_group.add_argument("-p", "--password", dest="password", help="User password", required=True)
    parser.add_argument(dest="hosts", nargs="+", help="List of devices")
    parser.add_argument("-f", "--format", dest="file_name_fmt", default="config_{}.cfg",
                        help="Filename format (use {} to place host IP)")
    parser.add_argument("--re", dest="re", default="^([\[\]\w.@-]+[#:>]\s?)$",
                        help="Regexp to match device \"waiting to enter command\" mode")
    parser.add_argument("-v", "--verbose", dest="verbose", action="store_true", default=False,
                        help="Log output to stdout")
    parser.add_argument("-b", "--buffer", dest="buffer_size", default=1024,
                        help="Reseive buffer size")

    return parser.parse_args(arguments)


def main(options):
    """Processes a list of hosts"""
    match_re = re.compile(options.re, flags=re.MULTILINE)
    # TODO: add threading support
    for host in options.hosts:
        get_terminal_data(host,
                          options.username,
                          options.password,
                          options.commands,
                          options.file_name_fmt,
                          match_re,
                          write_all=options.write_all,
                          buff_size=int(options.buffer_size),
                          verbose=options.verbose)


if __name__ == '__main__':
    main(get_options())

ssh_utils.py

#!/usr/bin/python3

import paramiko
import time
import re


fix_re = re.compile("\r(?!\n)")


def get_terminal_data(host_port: str,
                      username: str,
                      password: str,
                      commands: list,
                      file_name_fmt: str,
                      match_re: object,
                      buff_size: int,
                      write_all: bool = True,
                      strip: bool = True,
                      verbose: bool = False):
    """Connect to host and download running-config to file
    :param host_port: address of ssh host in format ip:port
    :param username: ssh host username
    :param password: ssh host password
    :param commands: list of commands to execute
    :param file_name_fmt: output file name format
    :param match_re: compiled regexp to search waiting to enter command mode
    :param buff_size: receive buffer size
    :param write_all: write all commands output else write only last
    :param strip: strip match_re from output
    :param verbose: print device output
    :return: None
    """
    host = host_port.split(':')
    ip = host[0]
    port = len(host) == 1 and 22 or host[1]
    file_name = file_name_fmt.format(ip, port)
    print("Connecting to {}:{}, output file:{}".format(ip, port, file_name))
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(
        hostname=ip,
        port=port,
        username=username,
        password=password,
        look_for_keys=False,
        allow_agent=False)
    with client.invoke_shell() as ssh:
        with open(file_name, 'w') as fn:
            last_command = commands.pop()
            for command in commands:
                execute_command(ssh, command,
                                write_all and fn or None,
                                match_re, buff_size, strip=strip, verbose=verbose)
            execute_command(ssh, last_command, fn, match_re, buff_size, strip=strip, verbose=verbose)
        return


def execute_command(client: object,
                    command: str,
                    fn: object,
                    match_re: object,
                    buff_size: int,
                    strip: bool = True,
                    verbose: bool = False) -> None:
    """Send a command to the device and intercept the output until the device goes into standby mode
    :param client: terminal shell of device
    :param command: command to execute on device
    :param fn: file object to write output
    :param match_re: compiled regexp to match 'waiting to command' mode
    :param buff_size: recieive buffer size
    :param strip: strip match_re from output
    :param verbose: print device output
    :return:
    """
    print("  Execute command \"{}\"!".format(command))
    client.send('{}\n'.format(command))
    time.sleep(1)
    # Loop 1
    while not client.exit_status_ready():
        time.sleep(1)
        if client.recv_ready():
            bchunk = client.recv(buff_size)
            # fix Eltex
            chunk = fix_re.sub("", bchunk.decode('ascii'))
            finish = match_re.search(chunk)
            if finish and strip:
                chunk = match_re.sub("", chunk)
            if verbose:
                print(chunk)
            if fn is not None:
                fn.write(chunk)
            if finish:
                break

    # Loop2
    while client.recv_ready():
        bchunk = client.recv(buff_size)
        # fix Eltex
        chunk = fix_re.sub("", bchunk.decode('ascii'))
        finish = match_re.search(chunk)
        if finish and strip:
            chunk = match_re.sub("", chunk)
        if verbose:
            print(chunk)
        if fn is not None:
            fn.write(chunk)
        if finish:
            break
    return