directory

• The resources

• Paramiko

  • The installation

• Connect the Linux

• File upload/download

• The file enclosed

• other

The resources

www.liujiangblog.com/blog/15/

Blog.csdn.net/leorx01/art…

Docs.paramiko.org/en/stable/a…

www.bilibili.com/video/BV1cQ…

Paramiko

SSH2 iS a connection tool, written by Python, as a test to dry ni, in order to achieve a demand, connect to the database server, backup/restore SQL, (data backup before the test – data recovery after the test, to achieve a non-pollution data environment, thought provider QQ: 1301559180)

The installation

pip install paramiko

Connect the Linux

import paramiko

SSH connection object
ssh_client = paramiko.SSHClient()
# Automatically accept the key sent by the server
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# username + password connection
ssh_client.connect(hostname="192.168.60.222", port=22, username="root", password="123456")
User name + private key connection; Path of the private key file
# private = paramiko.RSAKey.from_private_key_file(r'C:\Users\zy7y\.ssh\id_rsa')
# ssh_client.connect(hostname="192.168.60.222", port=22, username="root", pkey=private)

The first standard input - for interactive commands, the second standard output - saves the normal execution of the command, and the third standard error output
stdin, stdout, stderr = ssh_client.exec_command("pwd")
# Return type byte type(stdout.read()) 
      
# to a string
print(stdout.read().decode())

# close the connection
ssh_client.close()
Copy the code

File upload/download

import paramiko

# SSH connection
ssh_client = paramiko.Transport(("192.168.60.222".22))
ssh_client.connect(username="root", password="123456")

Configure the private key file location
# private = paramiko.RSAKey.from_private_key_file('/Users/root/.ssh/id_rsa')
Use pkey to specify the private key
# ssh_client.connect(username="root", pkey=private)

Create an FTP client
ftp_client = paramiko.SFTPClient.from_transport(ssh_client)

Upload test. SQL from current directory to server
ftp_client.put(localpath="test.sql", remotepath="/root/test3/hello.sql")


Download the file locally
ftp_client.get(localpath="test1.sql", remotepath="/root/test3/hello.sql")

# Close the SSH connection
ssh_client.close()
Copy the code

The file enclosed

#! /usr/bin/env python
# _*_ coding: utf-8 _*_
""" @project: apiAutoTest @file: ssh_demo.py @author: zy7y @time: 2021/1/18 @site: https://cnblogs.com/zy7y @github: https://github.com/zy7y @gitee: https://gitee.com/zy7y @desc: Paramiko Encapsulation """
import paramiko
import os


class SSHTools:
    def __init__(self, host: str, port: int = 22, username: str = "root",
                 password: str = None, private_key_file: str = None) :
        "" You can use the password or the private key file private_key_file method to connect to the server using SSH :param host: indicates the host ADDRESS STR :param port: indicates the host address. Host port default (int) 22 :param username: the account used for login Default (STR) root :param password: the password of the account default (STR) None :param private_key_file: Private key file path (STR) Default None and password Only one ""
        ssh_client = paramiko.SSHClient()
        # Automatically accept the key sent by the server
        ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        try:
            if password is None:
                # username + private key file connection
                ssh_client.connect(hostname=host, port=port, username=username,
                                   pkey=paramiko.RSAKey.from_private_key_file(private_key_file))
            else:
                # username + password connection
                ssh_client.connect(hostname=host, port=port, username=username, password=password)
            print(F "SSH connection successful,address:{host}:{port}!")
            self.ssh_client = ssh_client
        except Exception as e:
            print(F "SSH connection failure, error content:{e}")

    def execute_cmd(self, cmd: str) :
        """ :param CMD: the corresponding command on the server, can be list, or STR """
        try:
            if isinstance(cmd, list) :for c in cmd:
                    stdin, stdout, stderr = self.ssh_client.exec_command(c)
                    print(F "Enter the command:{c}-> Output result:{stdout.read().decode()},\n Exception message:{stderr.read().decode()}")
            else:
                stdin, stdout, stderr = self.ssh_client.exec_command(cmd)
                print(F "Enter the command:{cmd}-> Output result:{stdout.read().decode()}\n Exception message:{stderr.read().decode()}")
        except Exception as e:
            print(F "error is as follows,{e}")

    def ssh_close(self) :
        """ Close the connection """
        self.ssh_client.close()
        print(SSH connection closed...)


class SFTPTools:

    def __init__(self, host: str, port: int = 22, username: str = "root", password: str = None, private_key_file: str = None) :
        # SSH connection
        ssh_client = paramiko.Transport((host, port))
        self.host = host
        if password is None:
            ssh_client.connect(username="root", pkey=paramiko.RSAKey.from_private_key_file(private_key_file))
        else:
            ssh_client.connect(username=username, password=password)
        self.ssh_client = ssh_client
        Create an FTP client
        self.ftp_client = paramiko.SFTPClient.from_transport(ssh_client)

    def files_action(self, post: bool, local_path: str = os.getcwd(), remote_path: str = "/root") :
        """ :param post: True: upload False: download :param local_path: local file path, default working directory of the current script :param remote_path: The path to the file on the server, in the /root directory by default.
        if post:     # upload file
            if remote_path[-1] != "/":
                remote_path += "/"
            self.ftp_client.put(localpath=local_path, remotepath=f"{remote_path}{os.path.split(local_path)[1]}")
            print(F "File uploaded successfully:{local_path} -> {host}:{remote_path}{os.path.split(local_path)[1]}")
        else:   # Download file
            if local_path[-1] != "\ \":
                local_path += "\ \"
            self.ftp_client.get(localpath=f"{local_path}{os.path.split(remote_path)[1]}", remotepath=remote_path)
            print(F "File downloaded successfully:{host}:{remote_path}{os.path.split(local_path)[1]} -> {local_path}")

    def ssh_close(self) :
        """ Close the connection """
        self.ssh_client.close()
        print(SSH connection closed...)


if __name__ == '__main__':
    # own virtual machine
    host = "192.168.60.222"
    username = "root"
    password = "123456"

    ssh = SSHTools(host=host, username=username, password=password)
    shell = "uname"
    ssh.execute_cmd(shell)
    ssh.ssh_close()

    # SFTP
    sftp = SFTPTools(host=host, username=username, password=password)
    sftp.files_action(True.r"C:\Users\zy7y\Desktop\FastAPI.xmind"."/root")
    sftp.files_action(0, remote_path="/root/FastAPI.xmind")
    sftp.ssh_close()
Copy the code

Another one that’s a little bit simpler

#! /usr/bin/env python
# _*_ coding: utf-8 _*_
""" @project: apiAutoTest @file: ssh_demo.py @author: zy7y @time: 2021/1/18 @site: https://cnblogs.com/zy7y @github: https://github.com/zy7y @gitee: https://gitee.com/zy7y @desc: Paramiko Encapsulation """
https://www.liujiangblog.com/blog/15/ # SSH + SFTP reference linking
class LinuxTools:
    def __init__(self, host: str, port: int = 22, username: str = "root", password: str = None, private_key_file: str = None) :
        # SSH connection
        self.trans = paramiko.Transport((host, port))
        self.host = host
        if password is None:
            self.trans.connect(username="root", pkey=paramiko.RSAKey.from_private_key_file(private_key_file))
        else:
            self.trans.connect(username=username, password=password)
        Transport of the sshClient object is trans
        self.ssh = paramiko.SSHClient()
        self.ssh._transport = self.trans
        Create an SFTP client
        self.ftp_client = paramiko.SFTPClient.from_transport(self.trans)

    def execute_cmd(self, cmd: str) :
        """ :param CMD: the corresponding command on the server, can be list, or STR """
        try:
            if isinstance(cmd, list) :for c in cmd:
                    stdin, stdout, stderr = self.ssh.exec_command(c)
                    print(F "Enter the command:{c}-> Output result:{stdout.read().decode()},\n Exception message:{stderr.read().decode()}")
            else:
                stdin, stdout, stderr = self.ssh.exec_command(cmd)
                print(F "Enter the command:{cmd}-> Output result:{stdout.read().decode()}\n Exception message:{stderr.read().decode()}")
        except Exception as e:
            print(F "error is as follows,{e}")

    def files_action(self, post: bool, local_path: str = os.getcwd(), remote_path: str = "/root") :
        """ :param post: True: upload False: download :param local_path: local file path, default working directory of the current script :param remote_path: The path to the file on the server, in the /root directory by default.
        if post:  # upload file
            if remote_path[-1] != "/":
                remote_path += "/"
            self.ftp_client.put(localpath=local_path, remotepath=f"{remote_path}{os.path.split(local_path)[1]}")
            print(F "File uploaded successfully:{local_path} -> {self.host}:{remote_path}{os.path.split(local_path)[1]}")
        else:  # Download file
            if local_path[-1] != "\ \":
                local_path += "\ \"
            self.ftp_client.get(localpath=f"{local_path}{os.path.split(remote_path)[1]}", remotepath=remote_path)
            print(F "File downloaded successfully:{self.host}:{remote_path}{os.path.split(local_path)[1]} -> {local_path}")

    def ssh_close(self) :
        """ Close the connection """
        self.trans.close()
        print(SSH connection closed...)



if __name__ == '__main__':
    # own virtual machine
    host = "192.168.60.222"
    username = "root"
    password = "123456"
    
    ssh_sftp = LinuxTools(host=host, username=username, password=password)
    ssh_sftp.execute_cmd("docker images")
    ssh_sftp.files_action(True.r"C:\Users\zy7y\Desktop\FastAPI.xmind"."/root")
    ssh_sftp.files_action(0, remote_path="/root/FastAPI.xmind")
    ssh_sftp.ssh_close()
Copy the code

other

Integrate it with apiAutoTest sometime