python使用ssh命令执行脚本

SanXiaoXing2023-11-062024-07-26

前提需要有python和pip环境，一下只介绍python与ssh命令执行

下载paramiko库

命令框安装

1	pip install paramiko

paramiko使用代码

在一个无限循环中，依次执行commands列表中的命令。使用exec_command()方法执行命令，并获取标准输出和标准错误输出。然后打印标准错误输出和标准输出的内容。接下来，使用range()函数实现倒计时功能，打印剩余秒数并进行延迟。

# 导入必要的库
import paramiko
import time

# ssh连接的相关信息
hostname = '192.168.1.51' # 主机的IP地址
port = 22                 # 端口号
username = 'root'         # 用户名
password = 'root'         # 密码

# 定义执行的命令
commands = [
    'us -m \'{"ctrl":{"mac":"DC8E95FFFE0C77FF","endpoint":1,"OnOff":1}}\'',
    'us -m \'{"ctrl":{"mac":"DC8E95FFFE0C77FF","endpoint":2,"OnOff":1}}\''
]

# 建立ssh连接
# 创建一个SSHClient对象，并设置缺失主机密钥的策略为自动添加。然后使用connect()方法连接到指定的主机。
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(hostname, port, username, password)

time.sleep(2)
try:
    while True:
        for command in commands:
            # 执行命令
            print(command)
            stdin, stdout, stderr = ssh.exec_command(command)
            # print(stdin)
            # print(stdout)
            print('stderr', stderr.readlines())
            for line in stdout.readlines():
                print(line.strip())
                
            # 进入倒计时
            for i in range(3, 0, -1):
                print("\r倒计时{}秒！".format(i), end="", flush=True)
                time.sleep(1)

except KeyboardInterrupt:
    pass

# 关闭ssh连接
ssh.close()

这是一个简单的示例。

paramiko结合多线程的使用使用代码示例如下

import threading
import paramiko
class paramikoThreading(threading.Thread):
    def __init__(self,command,host,username,password,port=22):
        self.command = command
        self.host = host
        self.username = username
        self.password = password
        self.port = port
 
        super(paramikoThreading,self).__init__()
 
 
    def run(self):
        ssh = paramiko.SSHClient()
        # 创建一个ssh的白名单
        know_host = paramiko.AutoAddPolicy()
        # 加载创建的白名单
        ssh.set_missing_host_key_policy(know_host)
 
        # 连接服务器
        ssh.connect(
            hostname=self.host,
            port=self.port,
            username=self.username,
            password=self.password,
        )
 
        stdin, stdout, stderr = ssh.exec_command(self.command)
        print("*"*60)
        print("ip:%s,\ncommand:%s,\n"%(self.host,self.command))
        print(stdout.read().decode())
        print("*"*60)
        ssh.close()
 
if __name__ == '__main__':
    from settings import pool  #调用配置文件配置文件为settings.py
 
    command = "ls"
    t_pool = []
    for host in pool:
        t = paramikoThreading(
            host=host.get("host","localhost"),
            username=host.get("username","root"),
            password=host.get("password","123"),
            command=command
            )
        t_pool.append(t)
    for t in t_pool:
        t.start()
    for t in t_pool:
        t.join()

上文使用的`settings.py`文件如下

pool = [
    dict(host="10.10.21.177", username="root", password="123"),
    dict(host="10.10.21.177", username="root", password="123"),
    dict(host="10.10.21.177", username="root", password="123"),
    dict(host="10.10.21.177", username="root", password="123"),
    dict(host="10.10.21.177", username="root", password="123"),
    dict(host="10.10.21.177", username="root", password="123"),
    dict(host="10.10.21.177", username="root", password="123"),
    dict(host="10.10.21.177", username="root", password="123"),
]

文件的上传与下载

import paramiko
 
trans = paramiko.Transport(
    sock=("10.10.21.177",22) # 设置IP以及端口号
)
 
trans.connect(
    username="root",		# 账号
    password="12345"		#密码
)
sftp = paramiko.SFTPClient.from_transport(trans)
 
#上传
#把本地的文件settings.py，上传到远端为/root/Desktop/settings.py
sftp.put("settings.py","/root/Desktop/settings.py")
 
 
#下载
#从远程/root/Desktop/hh.py获取文件下载到本地名称为hh.py
# sftp.get("/root/Desktop/hh.py","hh.py")
 
sftp.close()