首页实践项目 Python 简单无脑的阿里钉钉信息推送脚本

简单无脑的阿里钉钉信息推送脚本

2019年4月1日

2078

0x01 前言

我公司并没有使用钉钉，但我觉得这是一款很难评价的软件。就我用来做测试的这几天收到了超级无敌多的垃圾提醒和超级无敌多的垃圾短信，虽然软件里面的功能比一般的OA系统要好，但是垃圾信息提醒却是真真实实的败笔。

但用阿里钉钉的企业还是挺多的，受一位网友的建议，写了一个方便zabbix推送信息的简简单、无脑的信息推送脚本。为什么无脑？因为它会将zabbix发送的所有告警都通过钉钉API直接推送到用户终端。

0x02 配置文件生成器

在使用脚本前需要准备钉钉的一些信息：

钉钉应用的appKey
钉钉应用的appSecret
钉钉告警群id

0x02.1 告警群id

我这脚本只支持将告警信息推送到群组，而钉钉有个问题：无法轻易地获取群组的id，但是推送API却需要用到这个id。

为了尽可能地减少工作量，我还编写了配置文件生成助手，通过简陋的交互即可完成群组的创建，并将群组id保存到配置文件中以备后用。这个配置过程需要输入群主的用户id，而这个用户id在钉钉的通讯录里即可找到：

除此之外还需要一些自定义信息，比如群名称等，这些都可以在后期进行修改，只需要跟随配置文件生成器的提示输入即可。

0x02.2 钉钉应用

钉钉没有全局的api，只能通过创建钉钉应用，并对该应用下的获得授权的资源做出调用API的动作。需要创建钉钉应用则需要通过浏览器打开以下地址：

https://oa.dingtalk.com/

并定位到“工作台”，在页面的最底部创建“自建应用”：

在新弹出的钉钉开发者平台中打开“服务管理”，并单击“创建应用”：

然后填写基本信息并单击“下一步”：

因为钉钉应用需要指定一个服务器出口IP，只有这个值里面的所有IP才有权限调用钉钉API，所以请准备一个固定IP的服务器。

其实我觉得这是一个钉钉自我防护的一个手段，建议大家使用云服务并借此机会推广阿里云，虽然并没有指明需要使用阿里云，但这个限制对非固定IP的用户及其不友好。

推送所需的“消息通知”功能是一个基础功能，默认处于开通的状态：

接下来需要查看“应用信息”中的“AppKey”和“AppSecret”，并记下来备用：

0x02.3 生成配置文件

将代码clone到本地后，通过python3运行脚本并跟随引导输入内容，回车确认即可：

完成后，在程序根目录下可以找到以下文件：

alarm_sender.conf：json格式的配置信息
alarm_sender.tmp：存储token的临时文件

用“AppKey”和“AppSecret”可以向钉钉的API服务器获取一个有几小时有效期的token，为了尽可能减少API的调用，程序会将这个token保存到临时文件中：

[root@web alarm_sender]# cat alarm_sender.tmp 
{"ali": {"ding0dqrnjhlgb5l8aut": {"timestamp": 1554131608412, "token": "c0050eaf7b583290bd68aeab851c6fb4"}}}

而配置文件的内容如下：

[root@web alarm_sender]# cat alarm_sender.conf 
{'dingding': [{'app_key': 'ding0dqrnjhlgb5l8aut', 'app_secret': 'ud9cVcVcJe1lMv_CEuR0i2xUmLGAgW3Dd7yi1zL4SdXdbeyccYXcVFNmprYIjUXm', 'chat_room_id': 'chat20e64face5d4f10a50cc65a89d7d2de0'}]}

配置文件生成器可以添加多个告警目标，即多个钉钉应用，只需要交互最后选择“继续添加”即可。

完成后，钉钉应用内会多一个群聊：

而建立群聊时需要用到的代码如下：

            post_content = {
                "name": chat_room_name,
                "owner": chat_room_owner,
                "useridlist": [chat_room_owner],
                "showHistoryType": show_history_type,
                "searchable": searchable,
                "validationType": validation_type,
                "mentionAllAuthority": mention_all_authority,
                "chatBannedType": chat_banned_type,
                "managementType": 1
            }
            msg_content = json.dumps(post_content, ensure_ascii=False)
            r = requests.post(url, msg_content.encode('utf-8'), timeout=5)
            r_content = r.content.decode()
            content_dict = json.loads(r_content)

            if content_dict["errcode"] == 0:
                chat_room_id = content_dict["chatid"]
            else:
                output_content = "# 您输入的信息有误，阿里钉钉返回了错误信息，请留意告警信息内容：" + r_content
                return output_content

内容比较简单，需要重点关注的是post的载荷，这部分内容可以在钉钉的开发文档中找到。

0x03 ali_dingtalk.py

信息的推送主要分为3步，第一步是校验token的有效期、第二步是推送，第三步是写日志。

0x03.1 校验

在存储着token的临时文件的内容如下：

{
  "ali": {
    "ding0dqrnjhlgb5l8aut": {
      "timestamp": 1554131608412,
      "token": "c0050eaf7b583290bd68aeab851c6fb4"
    }
  }
}

其中有一个时间戳的key：timestamp，值为获取到该token的时间。如果它与调用脚本时的时间戳的差值大于6600000（约1.9小时）则返还None，否则返还token的值：

    def chk_token(self):

        # 尝试读取文件
        try:
            r = open(tmp_file_path, 'r')
        except FileNotFoundError:
            return None
        else:
            r_content = r.readline()
            r_dict = json.loads(r_content)
            r.close()

        # 尝试获取旧token
        try:
            wc_token_last = r_dict["wechat"][self.appkey]["token"]
        except KeyError:
            return None
        else:
            wc_token_timestamp_last = r_dict["wechat"][self.appkey]["timestamp"]

        # 判断有效期
        if int(timestamp_now) - int(wc_token_timestamp_last) > 6600000:
            return None
        else:
            return wc_token_last

当然，在此之前还需要检查配置文件是否存在，如果不存在，同样返还None。

0x03.2 获取token

如果旧token已过期，但配置文件中存在阿里钉钉的配置内容，则会自动获取新token；如果找不到token的同时也没有在配置文件中找到阿里钉钉的内容，则会返还错误并写入日志：

    def get_token(self):
        url = "https://oapi.dingtalk.com/gettoken?appkey=" + self.appkey + "&appsecret=" + self.appsecret
        r = requests.get(url)
        r_content = r.content.decode()
        r_dict = json.loads(r_content)

        if r_dict["errcode"] == 0:
            write_tmp(self.appkey, r_dict["access_token"])
            return True, r_dict["access_token"]
        else:
            return False, r_dict

0x03.3 推送

如果token通过了检查，那么就可以调用推送脚本了，内容如下：

    @staticmethod
    def send_msg(access_token, chatid, msg_content):
        post_content = {
            "msgtype": "text",
            "text": {
                "content": msg_content
            },
            "chatid": chatid
        }
        msg_content = json.dumps(post_content, ensure_ascii=False)
        url = 'https://oapi.dingtalk.com/chat/send?access_token=' + str(access_token)
        r = requests.post(url, msg_content.encode('utf-8'), timeout=5)
        r_content = r.content.decode()

        output_content = time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime()) + ' dingding ' + r_content
        return output_content

post的载荷中只需要3项内容，其中chatid就是使用配置文件生成器中生成的聊天室的id。

将json格式的载荷使用json dumps格式化并使用requests post到API地址即可完成推送工作。

0x03.4 日志

日志是DEBUG的一个重要途径，也是后期检查推送情况的一个途径。所以脚本内还有一个写日志的函数：

def write_tmp(app_key, new_token):
    try:
        r_tmp = open(tmp_file_path, 'r')
    except FileNotFoundError:
        tmp_content = ''
    else:
        tmp_content = r_tmp.read()
        r_tmp.close()

    try:
        tmp_dict = json.loads(tmp_content)
    except Exception:
        tmp_dict = dict()

    if "ali" in tmp_dict.keys():
        try:
            del tmp_dict["ali"][app_key]
        except KeyError:
            pass
        else:
            tmp_dict["ali"][app_key] = {"timestamp": timestamp_now, "token": new_token}
    else:
        tmp_dict["ali"] = {app_key: {"timestamp": timestamp_now, "token": new_token}}

    r_tmp = open(tmp_file_path, 'w')
    r_tmp.write(json.dumps(tmp_dict))
    r_tmp.close()

0x04 调用

其实调用很简单，只需要依照上述的逻辑进行调用即可，具体的函数如下：

def push_to_dingding(data_content):
    appkey = data_content['appkey']
    appsecret = data_content['appsecret']
    chatid = data_content['chatid']
    msg = data_content['msg']

    ali_main = AliDingTalk(appkey, appsecret)
    last_ali_token = ali_main.chk_token()

    # 若旧token无效则获取新token
    if last_ali_token is None:
        new_ali_token = ali_main.get_token()

        if new_ali_token[0]:
            wc_token = new_ali_token[1]
        else:
            return None
    # 若旧token有效则使用旧token
    else:
        wc_token = last_ali_token

    send_msg = ali_main.send_msg(wc_token, chatid, msg)
    write_log(send_msg)
    return send_msg

首先将传入的json的内容分别传值给变量以便后面使用；然后判断token，如果token失效或不存在，则获取新的；取到token后即可调用推送函数，最终写入日志。