**包含两个主要功能**
1. SshTest类，用于探测ssh是否可以连接以及执行ssh命令。
2. PushDing，用于推送钉钉消息

**本文包含以下三个部分**
1. python脚本正文
2. 只探测ssh是否可连接样例模板。
3. 探测ssh并执行操作系统命令以及推送钉钉消息样例模板。

# 1. python脚本正文
> 下面脚本为通用模板，可直接使用

```python
#!/bin/python3
# 本模块两个包含两个通用功能：
"""
1.SshTest类，用于探测ssh是否可以连接以及执行ssh命令。
2.PushDing，用于推送钉钉消息
"""

# python内置模块
import re
import datetime
import time
import os
import hmac
import hashlib
import urllib
import base64
import requests
import json
# paramiko需要下载模块
import paramiko

"""
一、SshTest类：
1.探测ssh是否可达
2.check_ssh_status方法，用于探测ssh是否可以连接，无法连接将报错，返回三个标志值：
    1.send_msg，显示主机、探测时间、以及ssh无法连接信息
    2.warn_type，告警类型，显示ssh无法连接
    3.sshstat，ssh是否可连接标志位，为boolean值，true为可连接，false为不可连接
3.exec_ssh_cmd用于ssh探测成功之后，执行ssh命令。
"""
class SshTest:
    def __init__(self, host, port, username, pwd):
        self.host = host
        self.port = port
        self.username = username
        self.pwd = pwd
        self.ssh = paramiko.SSHClient()
        self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        self.timenow = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + ' '
        try:
            self.ssh.connect(self.host, self.port, self.username, self.pwd, timeout=5)
            self.sshinvalid = False
            # print(self.sshinvalid)
        except Exception as errmsg:
            self.sshinvalid = True
            self.errormsg = errmsg
            # print(self.sshinvalid)

def check_ssh_status(self):
        if self.sshinvalid:
            send_msg = ('**告警时间：** ' + self.timenow  + ' \\\n **告警主机：** ' + str(self.host) + ' \\\n **告警内容：** ' + str(self.errormsg) + ' ')
            warn_type='主机SSH无法连接'
            sshstat = False
            # 如遇到无法ssh连接，报错详细信息
            # print(self.errormsg)
            print('ssh连接报错为: ' + str(self.errormsg))
        else:
            send_msg = ('**告警时间：** ' + self.timenow  + ' \\\n **告警主机：** ' + str(self.host) + ' \\\n **告警内容：** ')
            warn_type='主机SSH连接正常'
            sshstat = True
            print(warn_type)
        return send_msg,warn_type,sshstat

def exec_ssh_cmd(self,cmd):
        if self.sshinvalid:
            i = 'SSH无法连接，请检查主机'
            # print(i)
            cmd_out = []
            cmd_out.append(i)
        else:
            stdin, stdout, stderr = self.ssh.exec_command(cmd)
            # 定义一个空列表，后续用于存储命令输出
            cmd_out = []
            for i in stderr.readlines():
                print('执行命令错误为： ' + i)
                cmd_out.append(i)
            # print(stdout.readlines)
            for i in stdout.readlines():
                print('命令执行成功，输出为： ' + i)
                cmd_out.append(i)
        # 如果有多行，则下面i返回的是最后一行数据，一定注意
        return cmd_out

"""
推送钉钉类：
1.初始化实例传入钉钉机器人webhook以及secret即可。
2.发送消息，传入需要发送的消息以及告警类型字符串即可，传入消息为markdown格式。
"""
class PushDing:
    def __init__(self,webhook,secret):
        self.webhook = webhook
        self.secret = secret

# 钉钉发送函数，输入markdown格式消息文本即可发送
    def send_message(self,send_msg,warn_type):
        ### 钉钉配置 begin ###                                                                                                                              
        # 如果是内网，需要有互联网代理配置                                                                                                                  
        # proxies = {'http':'http://IP:PORT','https':'http://IP:PORT'}                                                                                      
        # os.environ["https_proxy"] = "http://IP:PORT"                                                                                            
        webhook = self.webhook                        
        timestamp = str(round(time.time() * 1000))                                                                                                          
        secret = self.secret                                                                     
        secret_enc = secret.encode('utf-8')                                                                                                                 
        string_to_sign = '{}\n{}'.format(timestamp, secret)                                                                                                 
        string_to_sign_enc = string_to_sign.encode('utf-8')                                                                                                 
        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()                                                             
        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))                                                                                         
        webhook = webhook + '×tamp=' + timestamp + '&sign=' + sign
        # timestamp 与系统当前时间戳如果相差1小时以上，则认为是非法的请求，所以需要注意发送消息服务器时间不能与标准时间相差超过一小时                                                                         
        headers = {'Content-Type': 'application/json'}                                                                                                      
        user = "all"                                                                                                                                        
        ### 钉钉配置 end ### 
        data={                                                                                                                                          
            "msgtype": "markdown",                                                                                                                      
            "markdown": {                                                                                                                               
                "title": f"{warn_type}告警",                                                                                                                  
                "text": f"# <font face=\"华云彩绘\" color=\"#FF0000\" size=\"10\">{warn_type}</font> \n " + send_msg                                      
            },                                                                                                                                          
            "at": {                                                                                                                                     
                "atMobiles": [                                                                                                                          
                    user                                                                                                                                
                ],                                                                                                                                      
                "isAtAll": False                                                                                                                        
            }                                                                                                                                           
        }                                                                                                                                               
        #print(data)                                                                                                                                    
        x=requests.post(url=webhook,data=json.dumps(data),headers=headers)                                                                              
        # 打印发送钉钉结果                                                                                                                              
        print('钉钉推送结果为： ' + str(x.json()) + '\n')
```

# 2. 样例1，只ssh探测主机
```python
#!/bin/python3
# 本脚本用于批量探测主机ssh是否可以连接，无法连接反馈结果以及推送钉钉机器人消息。

# 导入内置模块
import datetime
# 导入自定义模块
import sshtest

"""
变量赋值区域
1. hostlist填写自己的主机IP登录用户名密码即可。
"""
hostlist=[
{'host': '192.168.xxx.xxx', 'port': '22', 'user': 'oracle', 'pwd': 'xxxxxx'},
{'host': '192.168.xxx.xxx', 'port': '22', 'user': 'oracle', 'pwd': 'xxxxxx'}
]

"""
执行命令并推送钉钉机器人
"""
for con_info in hostlist:
    host = con_info['host']
    port = con_info['port']
    username = con_info['user']
    pwd = con_info['pwd']
    # 打印当前时间以及探测IP
    print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + ' Host IP : ' + str(host))
    # 初始化实例
    ssh = sshtest.SshTest(host, port, username, pwd)
    """
    # 检查ssh状态,返回的数据为tuple，
    第一个元素为ssh探测后推送消息正文
    第二个为ssh探测状态
    第三个为ssh探测标志位，true为ssh正常，false为无法连接
    check_ssh_status()方法会自动打印ssh连接报错信息，比如密码不正确，22端口无法连接等
    """
    ssh_return = ssh.check_ssh_status()
    print(str(host) + '主机SSH状态为： ' + str(ssh_return[2]) + '\n')
    ssh.ssh.close()
```

# 3 样例2，探测主机以及执行命令并推送钉钉机器人消息
```python
#!/bin/python3
# 本脚本用于批量探测主机ssh是否可以连接，无法连接反馈结果以及推送钉钉机器人消息。

# 导入内置模块
import datetime
# 导入自定义模块
import sshtest

"""
变量赋值区域
1. hostlist填写自己的主机IP登录用户名密码即可。
2. cmd填写需要ssh执行的命令
3. webhook，钉钉机器人地址。
4. secret，钉钉机器人密钥
"""
hostlist=[
{'host': '192.168.xxx.xxx', 'port': '22', 'user': 'oracle', 'pwd': 'xxxxxx'},
{'host': '192.168.xxx.xxx', 'port': '22', 'user': 'oracle', 'pwd': 'xxxxxx'}
]
cmd = '你需要执行的SSH命令'
webhook = '你的钉钉机器人webhook地址'
secret = '你的钉钉机器人密钥'

"""
执行命令并推送钉钉机器人
"""
for con_info in hostlist:
    host = con_info['host']
    port = con_info['port']
    username = con_info['user']
    pwd = con_info['pwd']
    print(str(datetime.datetime.now()) + ' Host IP : ' + str(host))
    print('执行ssh探测以及ssh命令')
    # 初始化实例
    ssh = sshtest.SshTest(host, port, username, pwd)
    """
    # 检查ssh状态,返回的数据为tuple，
    第一个元素为ssh探测后推送消息正文
    第二个为ssh探测状态
    第三个为ssh探测标志位，true为ssh正常，false为无法连接
    """
    ssh_return = ssh.check_ssh_status()
    # print(ssh_return[2])
    # 当ssh可连接，则执行命令，否则不执行
    if ssh_return[2]:
        cmd_result = ssh.exec_ssh_cmd(cmd)
        print(cmd_result)
        s = "\\\n" .join(cmd_result)
        # 执行完毕，关闭ssh连接
    else:
        s = '主机无法连接，命令未执行.'
    ssh.ssh.close()
    # print(cmd_result)
    pushding = sshtest.PushDing(webhook,secret)
    pushding.send_message(ssh_return[0] + s ,ssh_return[1])

```

ITPub博客

Python执行SSH命令通用模板SSH连接状态探测以及钉钉推送

最新文章