Содержание
Python SSH Tutorial
SSH (secure shell) is good for remotely managing machines using a secure connection. Typically you will log in to a server using the command-line ssh tool, or something like PuTTy or MobaXTerm. This guide will show you how to use Python to connect and run commands over SSH using the Paramiko package.
Installation
The easiest way to install paramiko is using pip .
python -m pip install paramiko To install from source, clone from GitHub and then install using setup.py .
git clone https://github.com/paramiko/paramiko cd paramiko python setup.py install Connect to SSH
Connect to an SSH server using paramiko.client.SSHClient.connect(). The hostname is the only required parameter.
connect(hostname, port=22, username=None, password=None, pkey=None, key_filename=None, timeout=None, allow_agent=True, look_for_keys=True, compress=False, sock=None, gss_auth=False, gss_kex=False, gss_deleg_creds=True, gss_host=None, banner_timeout=None, auth_timeout=None, gss_trust_dns=True, passphrase=None, disabled_algorithms=None) One thing to consider is what trusted known host keys you have. You can use paramiko.client.SSHClient.load_system_host_keys() . You can also explicitly load a specific known hosts file with load_host_keys() and set the client to automatically accept and add unknown hosts with set_missing_host_key_policy() to paramiko.AutoAddPolicy . Use these options as needed. The AutoAddPolicy is not very secure since it will trust any remote host.
Connect using password
from paramiko import SSHClient client = SSHClient() #client.load_system_host_keys() #client.load_host_keys('~/.ssh/known_hosts') #client.set_missing_host_key_policy(AutoAddPolicy()) client.connect('example.com', username='user', password='secret') client.close() Connect using SSH key
Using an SSH key is more secure than using a password. Call connect() just like using the password, but instead of providing the user password, we will provide a key_filename and maybe a passphrase for the key. The passphrase may not be needed if your private key file does not have a passphrase.
You can also omit the key file and let Paramiko try to find the right key automatically in your ~/.ssh/ directory. You can turn that on by calling client.look_for_keys(True) . The first example will show how to explicitly provide the key and passphrase and the second one will show how to look for keys.
# Explicitly provide key and passphrase from paramiko import SSHClient, AutoAddPolicy client = SSHClient() #client.load_system_host_keys() #client.load_host_keys('~/.ssh/known_hosts') #client.set_missing_host_key_policy(AutoAddPolicy()) client.connect('example.com', username='user', key_filename='mykey.pem', passphrase='mysshkeypassphrase') client.close() This example show hows to look for keys in ~/.ssh/ automatically.
from paramiko import SSHClient client = SSHClient() #client.load_system_host_keys() #client.load_host_keys('~/.ssh/known_hosts') #client.set_missing_host_key_policy(AutoAddPolicy()) client.look_for_keys(True) client.connect('example.com', username='user') client.close() Run commands over SSH
Once you have a connection open, you can execute any command just like you would if you were in a regular interactive SSH session.
This example shows how to:
- Run a command on the server
- Provide standard input data to command
- Read standard output and error from command
- Get the return code of the command
The command run in this example is the PHP interpreter with the code provided via standard input. The output is then printed out and the return code is checked.
from paramiko import SSHClient # Connect client = SSHClient() client.load_system_host_keys() client.connect('example.com', username='user', password='secret') # Run a command (execute PHP interpreter) stdin, stdout, stderr = client.exec_command('php') print(type(stdin)) # print(type(stdout)) # print(type(stderr)) # # Optionally, send data via STDIN, and shutdown when done stdin.write('') stdin.channel.shutdown_write() # Print output of command. Will wait for command to finish. print(f'STDOUT: ') print(f'STDERR: ') # Get return code from command (0 is default for success) print(f'Return code: ') # Because they are file objects, they need to be closed stdin.close() stdout.close() stderr.close() # Close the client itself client.close() Conclusion
After reading this guide you should know how to connect to an SSH server using a password or SSH key. You should also know how to run a command, pass data to standard input, and read the output from standard output and error.
References
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
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 codeFile 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 codeThe 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: : !") self.ssh_client = ssh_client except Exception as e: print(F "SSH connection failure, error content: ") 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: - Output result: ,\n Exception message: ") else: stdin, stdout, stderr = self.ssh_client.exec_command(cmd) print(F "Enter the command: - Output result: \n Exception message: ") except Exception as e: print(F "error is as follows, ") 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" ]>") print(F "File uploaded successfully: - : ]>") else: # Download file if local_path[-1] != "\ \": local_path += "\ \" self.ftp_client.get(localpath=f" ]>", remotepath=remote_path) print(F "File downloaded successfully: : ]> - ") 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: - Output result: ,\n Exception message: ") else: stdin, stdout, stderr = self.ssh.exec_command(cmd) print(F "Enter the command: - Output result: \n Exception message: ") except Exception as e: print(F "error is as follows, ") 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" ]>") print(F "File uploaded successfully: - : ]>") else: # Download file if local_path[-1] != "\ \": local_path += "\ \" self.ftp_client.get(localpath=f" ]>", remotepath=remote_path) print(F "File downloaded successfully: : ]> - ") 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