Python Paramiko module

1. Paramiko automatically logs in to the network device to capture configuration information

# !/usr/bin/env python
# -*- coding:utf-8 -*-

import paramiko

ssh = paramiko.SSHClient()
ssh.connect(hostname='', port=22, username='cisco', password='cisco')
client = ssh.invoke_shell()

def run_cmd(cmd, endswith): # Parameters: cmd command, end character
    buff = ''
    while not buff.endswith(endswith):
        resp = str(client.recv(1024), 'utf-8')
        buff + = resp
    return buff

res = ''
res + = run_cmd('enable\\
', 'Password: ')
res + = run_cmd('cisco\\
', '#')
res + = run_cmd('terminal length 0\\
', '#')
res + = run_cmd('show run\\
', '#')

# <H3C> screen-length disable # no screen-length disable (H3C)
# Cisco# terminal length 0 # terminal no length (Cisco)
# MaiPu# more off # more on (Mai Pu)
# <HuaWei> screen-length 0 temporary # undo screen-length temporary (Huawei)
# RuiJie# terminal length 0 # terminal no length (RuiJie)
# ZTE# terminal length 0 # no terminal length (ZTE)
# !/usr/bin/env python
# -*- coding:utf-8 -*-

import paramiko
from paramiko.ssh_exception import NoValidConnectionsError, AuthenticationException

def sshconnect(command, hostname):
    :param command: The format must be list
    :param hostname: single host IP address
    :return: The user does not exist or the connection timed out and returns an error
    ssh = paramiko.SSHClient()
        ssh.connect(hostname=hostname, port=22, username='USERNAME', password='PASSWORD')
    except (NoValidConnectionsError, TimeoutError, AuthenticationException) as e:
        return {<!-- -->'status': False, 'ip': hostname}
        client = ssh.invoke_shell()
        for i in range(0, len(command)):
            buff = ''
            client.send(command[i] + '\\
            while not buff.endswith('#'):
                resp = str(client.recv(1024), 'utf-8')
                buff + = resp

if __name__ == '__main__':
    sshconnect('['show version','show run']', '')
# !/usr/bin/env python
# -*- coding:utf-8 -*-

import paramiko
from .conndb import condb

class compupload(object):
    def __init__(self, hostname, project=None, username='root', port=22, password='123456'):
        # self.private_key = paramiko.RSAKey.from_private_key_file('/root/.ssh/id_rsa')
        self.hostname = hostname
        self.username = username
        self.port = port
        self.project = project
        self.password = password
        self.transport = paramiko.Transport((self.hostname, self.port))
        # self.transport.connect(username=self.username, pkey=self.private_key)
        self.transport.connect(username=self.username, password=self.password)
        self.sftp = paramiko.SFTPClient.from_transport(self.transport)
        self.client = paramiko.SSHClient()
        self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # Allow connections to hosts that do not exist in the know_hosts file
        # self.client.connect(hostname=self.hostname, port=self.port, username=self.username, pkey=self.private_key)
        self.client.connect(hostname=self.hostname, port=self.port, username=self.username, password=self.password)

    def upload(self, local_path, remote_path):
        #Upload the file to the server
        self.sftp.put(local_path, remote_path, callback=self.__callback)
        # if self.project:
        # self.sftp.put(local_path, remote_path, callback=self.__callback)
        # self.sftp.put(local_path, remote_path)

    def download(self, remotepath, localpath):
        # Download the file to local
        self.sftp.get(remotepath, localpath, callback=self.__callback)

    def comand(self, com):
        # Execute the command and return the command result
        stdin, stdout, stderr = self.client.exec_command(com)
        result =
        reserr =
        return result,reserr

    def exec_com(self, com):
        # Execute the command and return the command result and status code = self.client.get_transport().open_session()
        stdout =
        stderr =
        exit_code =
        return stdout, stderr, exit_code

    def __callback(self, send, total):
        Requirements: Make a transfer progress bar
        1. The CallBack method obtains the total number of bytes and the number of transferred bytes
        2. Calculate the percentage of the number of bytes transferred to the total number of bytes
        3. Create a progress bar
        :param send:
        :param total:
        end = '' if send != total else '\\
        # Upload progress bar
        print('\r |{:<50}| {:.2f} M [{:.2f}%]'.format('#' * int(send * 100 / total / 2), send / 1024 / 1024, send * 100 / total), end=end, flush=True)
    # sendize = send / 1024 / 1024
    # percent = send * 100 / total
    # 'UPDATE `project` SET `sendsize`="{}",`percent`="{}" WHERE `name`="{}";'.format(sendsize, percent, self.project))

    def sshclose(self):
        # Close the connection

if __name__ == '__main__':
    ssh_sftp = comupload('')
    ssh_sftp.upload('test.tar.gz', '/root/test.tar.gz')
    r, re = ssh_sftp.comand('echo hello')
    res, reserr, exit_code = ssh_sftp.exec_com('echo helloworld')
# -*- coding:UTF-8 -*-
# pip install func-timeout
# Execute the command and return the command result

import re
import time
from paramiko import SSHClient, AutoAddPolicy
from func_timeout import func_set_timeout, exceptions

class RemoteCMD():
    """Execute remote command

    def _init_connection(self,
                         ip: str,
                         port: int,
                         user: str,
                         passwd: str,
                         timeout: int = 5):
        """Initialize the connection (extracting it separately can avoid func timeout and not closing the session)

            ip (str): IP address
            port (int): SSH port
            user (str): username
            passwd (str): password
            timeout (int, optional): timeout length (seconds). Defaults to 5.

            paramiko object: paramiko client object
        client = SSHClient()
        # client.load_system_host_keys()

        client = client.invoke_shell()
        return client

    def _match_prompt(self, content: str, regex: str):
        """ ends prompt matching

            content (str): the string to be matched
            regex (str): Keyword or regular expression to be matched

            bool: True/False
        if, content):
            return True

        return False

    def _send_cmd(self,
                  client: object,
                  cmd: str,
                  recv_end_prompt: str,
                  recv_size: int = 512):
        """Send commands and pull results

            client (object): paramiko client object
            cmd (str): command to execute
            recv_size (int, optional): The size of the result data of a single pull. Defaults to 512.

            string: execution result
        # Waiting for ready
        while not client.recv_ready():

        result = ""
        while not self._match_prompt(result, recv_end_prompt):
            result + = client.recv(recv_size).decode()

        return result

    def command(self,
                port: int,
                user: str,
                passwd: str,
                cmd: str,
                end_prompt: str = r"#\s$",
                timeout: int = 5):
        """Execute remote command

            ip (str): IP address
            port (int): SSH port
            user (port): username
            passwd (str): password
            cmd (str): command to execute
            timeout (int, optional): timeout length (seconds). Defaults to 5.

            tuple: (True/False, execution result or error)
            client = self._init_connection(ip, port, user, passwd, timeout)

        except Exception as err: # pylint: disable=broad-except
            return False, f"SSH connection exception: {err}"

            return True, self._send_cmd(client, cmd, end_prompt)

        except exceptions.FunctionTimedOut:
            return False, "SSH execution timeout error!"

        except Exception as err: # pylint: disable=broad-except
            return False, f"Other exceptions occurred during SSH execution: {err}"

            if client:

if __name__ == '__main__':
    print(sshclient.command('',22,'root','123456','touch test.txt'))

2. scapy module (network scanning module)
pip install scapy #install
from scapy.all import * #Import
ls(): shows support for all protocols
lsc(): Displays support for all commands (that is, sending and receiving data packet functions)
help (protocol|command): displays detailed help for protocols and commands
Protocols and commands do not include single quotes or double quotes, help(SNMP), help(tcpdump)
[root@localhost ~]# git clone
[root@localhost ~]# cd scapy/
[root@localhost ~]# ./run_scapy

Related Links: #python md5 encryption #requestsRequests #Conversion of time, date and timestamp in python #python method to solve the problem of converting strings starting with ‘\u’ into Chinese #paramiko module usage #python records database connection module #python records database connection module