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.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(hostname='192.168.1.2', port=22, username='cisco', password='cisco')
client = ssh.invoke_shell()

def run_cmd(cmd, endswith): # Parameters: cmd command, end character
    buff = ''
    client.send(cmd)
    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\\
', '#')
print(res)
ssh.close()

# <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.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        ssh.connect(hostname=hostname, port=22, username='USERNAME', password='PASSWORD')
    except (NoValidConnectionsError, TimeoutError, AuthenticationException) as e:
        return {<!-- -->'status': False, 'ip': hostname}
    else:
        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
    ssh.close()

if __name__ == '__main__':
    sshconnect('['show version','show run']', '192.168.1.1')
# !/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)
        #else:
        # 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 = stdout.read().decode()
        reserr = stderr.read().decode()
        return result,reserr

    def exec_com(self, com):
        # Execute the command and return the command result and status code
        self.channel = self.client.get_transport().open_session()
        self.channel.exec_command(com)
        stdout = self.channel.makefile().read()
        stderr = self.channel.makefile_stderr().read()
        exit_code = self.channel.recv_exit_status()
        self.channel.close()
        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:
        :return:
        '''
        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
    #condb(
    # 'UPDATE `project` SET `sendsize`="{}",`percent`="{}" WHERE `name`="{}";'.format(sendsize, percent, self.project))

    def sshclose(self):
        # Close the connection
        self.sftp.close()
        self.client.close()


if __name__ == '__main__':
    ssh_sftp = comupload('172.17.0.5')
    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')
    ssh_sftp.sshclose()
#!/usr/bin/python3.8
# -*- 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)

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

        Returns:
            paramiko object: paramiko client object
        """
        client = SSHClient()
        # client.load_system_host_keys()
        client.set_missing_host_key_policy(AutoAddPolicy())
        client.connect(hostname=ip,
                       port=int(port),
                       username=user,
                       password=passwd,
                       timeout=timeout,
                       allow_agent=False,
                       look_for_keys=False)

        client = client.invoke_shell()
        return client

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

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

        Returns:
            bool: True/False
        """
        if re.search(regex, content):
            return True

        return False

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

        Args:
            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.

        Returns:
            string: execution result
        """
        client.send(f"{str(cmd).strip()}\\
")
        # Waiting for ready
        while not client.recv_ready():
            time.sleep(0.2)

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

        return result

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

        Args:
            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.

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

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

        try:
            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}"

        finally:
            if client:
                client.close()

if __name__ == '__main__':
    sshclient=RemoteCMD()
    print(sshclient.command('172.18.188.5',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
Notice:
Protocols and commands do not include single quotes or double quotes, help(SNMP), help(tcpdump)
[root@localhost ~]# git clone https://github.com/secdev/scapy.git
[root@localhost ~]# cd scapy/
[root@localhost ~]# ./run_scapy

Related Links:
https://github.com/tecstack/forward
https://scapy.net
https://www.cnblogs.com/denise1108/p/10095042.html #python md5 encryption
https://www.cnblogs.com/lanyinhao/p/9634742.html #requestsRequests
https://www.cnblogs.com/jfl-xx/p/8024596.html #Conversion of time, date and timestamp in python
https://www.cnblogs.com/hahaxzy9500/p/7685955.html #python method to solve the problem of converting strings starting with ‘\u’ into Chinese
https://www.cnblogs.com/yezl/p/5803522.html
https://www.cnblogs.com/wztshine/p/11964321.html #paramiko module usage
https://www.cnblogs.com/wagyuze/p/11398484.html #python records database connection module
https://www.cnblogs.com/xiao-apple36/p/13968213.html #python records database connection module