实际开发中,有时候经常需要查看日志,有时候使用ssh工具打开就为了看一下错误日志又比较麻烦,所以今天带来一个简单的基于python的小工具.

首先需要先安装一个库 paramiko

使用命令直接安装

pip install paramiko 

paramiko库是一个开源的、基于SSH2协议的库，可以实现SSH连接以及数据的传输。

paramiko是用纯Python实现的SSH2客户端，支持身份验证、SFTP客户端以及远程执行命令等功能。paramiko库提供了丰富的类和方法，帮助用户快速实现SSH通信功能。

在实际应用中，paramiko库常用于构建自动化运维系统、远程部署、多机协作等场景.

定义MySshClient 类

class MySshClient:      def __init__(self, ssh_client):          self.ssh_client = ssh_client               def exec_command(self, cmd):          try:              stdin, stdout, stderr = self.ssh_client.exec_command(cmd)              return stdin, stdout, stderr          except Exception as e:              print(f"Error executing command {cmd}: {e}")                   def __enter__(self):          return self               def __exit__(self, exc_type, exc_val, exc_tb):          self.ssh_client.close()  在此代码中，我们定义了一个 MySshClient 类，用于连接 SSH 服务器并执行命令。您可以使用该类创建一个实例，并通过 connect() 方法连接到 SSH 服务器。  def connect(host, port, username, password):      ssh = paramiko.SSHClient()      ssh.set_missing_host_key_policy(paramiko.WarningPolicy())               try:          ssh.load_system_host_keys()          ssh.connect(host, port, username, password,timeout=3)      except paramiko.AuthenticationException:          raise Exception(f"在主机 {host}连接失败,请检查你的参数")      except paramiko.SSHException as e:          raise Exception(f"在 {host}连接出错: {e}")      except paramiko.BadHostKeyException as e:          raise Exception(f" {host} 无法验证通过: {e}")      except Exception as e:          raise Exception(f" 连接到{host}:{port}: {e}超时")               return ssh 

这个是给函数加上异常处理,方便代码的调试

执行命令

在连接到 SSH 服务器后，您可以使用 exec_command() 方法执行命令。该方法返回一个元组，包含服务器的输出、错误和输入。您可以将此元组提供给 with 关键字，以便在需要时进行处理。

2. 连接到主机

在连接到 SSH 服务器之前，您需要指定主机名、端口、用户名和密码。例如：

host = input("Host: 请输入主机名") port = int(input("Port: 默认为22")or 22)  username = input("Username: 请输入用户名") password = getpass.getpass("Password: 请输入密码") 

port = int(input(“Port: 默认为22”)or 22) 这段代码的意思是

默认端口22,如果你的ssh端口不是22,可以自行修改

密码的话,如果不想被别人看见,使用getpass函数进行加密

3. 创建并连接到 MySshClient 实例

ssh = connect(host, port, username, password)

4. 连接成功后，执行命令并处理输出

连接成功后，您可以使用 exec_command() 方法执行命令，并使用 with 关键字处理服务器的输出：

with MySshClient(ssh) as client:     stdin, stdout, stderr = client.exec_command("ls -l")     with stdout:         print(stdout.read().decode())     with stderr:         print(stderr.read().decode()) 

这将在连接到 SSH 服务器后，执行 “ls -l” 命令，并输出结果。请注意，此示例仅显示输出，如果有错误，错误信息将显示在控制台。

完整代码如下:

import paramiko  import os  import getpass  class MySshClient:      def __init__(self, ssh_client):          self.ssh_client = ssh_client               def exec_command(self, cmd):          try:              stdin, stdout, stderr = self.ssh_client.exec_command(cmd)              return stdin, stdout, stderr          except Exception as e:              print(f"Error executing command {cmd}: {e}")                   def __enter__(self):          return self               def __exit__(self, exc_type, exc_val, exc_tb):          self.ssh_client.close()           def connect(host, port, username, password):      ssh = paramiko.SSHClient()      ssh.set_missing_host_key_policy(paramiko.WarningPolicy())               try:          ssh.load_system_host_keys()          ssh.connect(host, port, username, password,timeout=3)      except paramiko.AuthenticationException:          raise Exception(f"在主机 {host}连接失败,请检查你的参数")      except paramiko.SSHException as e:          raise Exception(f"Error connecting to {host}: {e}")      except paramiko.BadHostKeyException as e:          raise Exception(f"Host key for {host} could not be verified: {e}")      except Exception as e:          raise Exception(f" 连接到{host}:{port}: {e}超时")               return ssh     if __name__ == '__main__':        host = input("Host: 请输入主机名")      port = int(input("Port: 默认为22")or 22)      username = input("Username: 请输入用户名")        password = getpass.getpass("Password: 请输入密码")      print('连接中...........')      ssh = connect(host, port, username, password,)           with MySshClient(ssh) as client:          stdin, stdout, stderr = client.exec_command("ls -l")          with stdout:              print(stdout.read().decode())          with stderr:              print(stderr.read().decode()) 

当然你也可以使用sftp进行传输

# 上传 current_path = os.getcwd()   #获取当前路径 print(current_path) filename = '\\main.py'     		# 文件名 file = current_path + f'{filename}'			# 当前文件的路径 sftp = client.open_sftp() 						# 使用sftp连接 sftp.put(file, '/root/main.py')			#  上传文件 sftp.get('/root/server.sh', current_path+'/code.sh')     #下载文件 

如果你觉得文章对你有用,多多支持一下