钉钉机器人（python实现）具体流程分析&学习

白速龙王的回眸

1685人浏览 · 2021-07-19 10:40:58

白速龙王的回眸 · 2021-07-19 10:40:58 发布

开发背景：
为了丰富自己的项目经历，同时熟悉网络编程模式以及python语言

基本知识：

1.hmac_code = hmac.new(app_secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()

这里对app_secret_enc, string_to_sign_enc使用hashlib中的sha256哈希算法，其中采用的digest摘要模式。需要注意的是app_secret_enc, string_to_sign_enc都表示为b’xxxxx’，也就是utf8的编码模式。

2.sign = base64.b64encode(hmac_code).decode(‘utf-8’)

用base64对hmac_code进行编码，同时通过decode(utf8)去掉b’xxxx’的形式，得到纯字符串。

3.timestamp = str(round(time.time() * 1000))

获取毫秒级别的当前时间戳。

4.string_to_sign = ‘{}\n{}’.format(post_timestamp, app_secret)

将post_timestamp, app_secret塞进前面的格式中。

5.items.append(item.split(’:’))

以：为分隔符，转化成dict的形式，方便get操作

6.post_sign = post_useful.get(‘sign’).strip()

获取dict中相应的元素，strip()去除首尾空格

7.json.loads()和json.dumps()

json.loads()是将字符串转成字典
json.dumps()是将字典转成字符串

主函数分析：

if __name__ == "__main__":
    # 启动服务,端口30421
    # 实例化一个socket，socket.AF_INET表示默认的ipv4，socket.SOCK_STREAM表示TCP协议
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # 绑定本机+端口号
    server_socket.bind(("", 30421))
    # listen中的函数表示排队的客户个数
    server_socket.listen(120)
    while True:
        # 接受客户端的连接请求,并返回一个套接字以及客服地址
        client_socket, client_address = server_socket.accept()
        #print("[%s, %s]用户连接上了" % client_address)

        # 创建进程，处理客户信息，将client_socket传给handle_client函数
        handle_client_process = Process(target=handle_client, args=(client_socket,))
        handle_client_process.start()
        #关闭客户socket，方便下次调用
        client_socket.close()

首先实例化一个主服务socket，同时用bind绑定端口号和用listen绑定监听的最大排队人数。注意，我们这里用到了内网渗透的技术（参见钉钉开放平台-机器人开发指南），也就是将访问的外网映射到本机的端口上，方便抓取post的信息。如下图：

在这里插入图片描述

接下来进入while循环，首先使用socket中的accpet()函数，另起一个新的client_socket并获取客户机的访问端口，随后，创建进程Process对client_socket使用handle_client的函数，后面利用start()拉起进程，最后结束客户的socket。注意，这里我们是多个客户socket和客户进程并发执行，所以要引入多进程库。

handle_clinet()函数

def handle_client(client_socket):
    # 获取socket
    # 2w表示缓冲区的大小
    request_data = client_socket.recv(20000)

    post_userid, post_sign, post_timestamp, post_mes = getPost(request_data)
    # 回应socket
    initKey(post_userid, post_sign, post_timestamp, post_mes)
    # 关闭socket
    client_socket.close()

逻辑较为简单，通过socket的recv抓取request的数据，同时利用getPost将数据格式化。利用initKey函数验证客户的权限以及实现回应的逻辑，最后关闭socket。

getPost()函数

格式化信息的函数。

def getPost(request_data):
    # 格式化socket数据
    request_data = str(request_data, encoding="utf8").split('\r\n')
    #print(request_data)
    items = []
    for item in request_data[1:-2]:
        items.append(item.split(':'))
    post_useful = dict(items)
    #print(post_useful)
    #print(request_data[-1])
    #将post_mes转为字典类型
    post_mes = json.loads(request_data[-1])
    #print(post_mes)
    # 得到钉钉请求的sign,timestamp,发送者加密id,消息内容,strip()去除首尾空格
    post_sign = post_useful.get('sign').strip()
    #print(post_sign)
    post_timestamp = post_useful.get('timestamp').strip()
    #print(post_timestamp)
    post_userid = post_mes.get('senderId').strip()
    #print(post_userid)
    post_mes = post_mes.get('text').get('content').strip()
    #print(post_mes)

    return post_userid, post_sign, post_timestamp, post_mes

initKey()函数

校验钉钉用户&处理回复信息

def initKey(post_userid, post_sign, post_timestamp, post_mes):
    # 配置token
    whtoken = "2c1ebd7abd938a41d6ad0d9a120bf4ed6572325209d02953ad87c68cc898cfdb"
    # 得到当前时间戳
    timestamp = str(round(time.time() * 1000))
    print(timestamp)
    # 计算签名
    app_secret = 'd8Ovq-OUKZRaPzzKLBCMN_E05RA8zDPbOs861zxb7y5b8jpLw67zHbWMQ4yKXo5y'
    app_secret_enc = app_secret.encode('utf-8')
    print(app_secret_enc)
    print('{}\n{}')
    string_to_sign = '{}\n{}'.format(post_timestamp, app_secret)
    print(string_to_sign)
    string_to_sign_enc = string_to_sign.encode('utf-8')
    print(string_to_sign_enc)
    hmac_code = hmac.new(app_secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    print(hmac_code)
    sign = base64.b64encode(hmac_code).decode('utf-8')
    print(sign)

    # 验证是否来自钉钉的合法请求
    if (abs(int(post_timestamp) - int(timestamp)) < 3600000 and post_sign == sign):
        webhook = "https://oapi.dingtalk.com/robot/send?access_token=" + whtoken + "&timestamp=" + timestamp + "&sign=" + sign
        header = {
            "Content-Type": "application/json",
            "Charset": "UTF-8"
        }
        # 发送消息
        message_json = json.dumps(selectMes(post_userid, post_mes))
        print(message_json)
        # 返回发送状态
        info = requests.post(url=webhook, data=message_json, headers=header)
        print(info)
    else:
        print("Warning:Not DingDing's post")

其中，whtoken和app_secret都是机器人创建的时候给出的，只有开发者才有权限获取。首先利用post_timestamp, app_secret生成一个utf8格式的string_to_sign_enc，然后利用app_secret_enc, string_to_sign_enc利用sha256算法取摘要得到一个hmac_code，最后再通过base64编码得到sign。

通过比对当前时间戳和post时间戳的差距，以及确认post中的签名和解密的签名一致来判定用户的合法性。

通过验证后，我们会对指定的webhook post一段信息，这个就会通过钉钉服务器转成消息的形式出现在手机信息上。作为post信息，我们需要设置其header，message_json和url。

message_json通过json.dumps变成字符串格式，selectMes(post_userid, post_mes))负责由post_mes生成回复信息。

selectMes（）函数：

def selectMes(post_userid, post_mes):
    # 判断指令选择对应回复
    if (post_mes == '肥马去天津玩了啥'):
        send_mes = '玩了个寂寞'
        return sendText(post_userid, send_mes)
    elif (post_mes == '天津哪里最好玩'):
        send_mes = '垃圾加工厂'
        return sendText(post_userid, send_mes)
    elif (post_mes == '为什么要去天津'):
        send_mes = '闲的蛋疼'
        return sendMarkdown('莫名奇妙', send_mes)
    elif (post_mes == '真棒！'):
        send_mes = '鼓掌！'
        return sendText(post_userid, send_mes)
    elif (post_mes == '天气'):
        send_mes = getWeather()
        return sendMarkdown('天气预报', send_mes)
    else:
        return sendText(post_userid, '我懒得理你')

如上图，简单if回复，分两种方式，sendText和sendMarkdown

sendText（）函数：

def sendText(post_userid, send_mes):
    # 发送文本形式
    message = {
        "msgtype": "text",
        "text": {
            "content": send_mes
        },
        "at": {
            "atDingtalkIds": [post_userid],
            "isAtAll": False
        }
    }
    return message

简单json的文本回复，需要指定消息类型，内容以及at的对象

sendMarkdown（）函数：

def sendMarkdown(title, send_mes):
    # 发送Markdown形式
    message = {
        "msgtype": "markdown",
        "markdown": {
            "title": title,
            "text": send_mes
        },
        "at": {
            "atDingtalkIds": [],
            "isAtAll": False
        }
    }
    return message

markdown的形式

整体流程：

在这里插入图片描述
完整代码：

# -*- coding: GBK -*-
import requests
import json
import time
import hmac
import hashlib
import base64
import socket
from multiprocessing import Process


def handle_client(client_socket):
    # 获取socket
    # 2w表示缓冲区的大小
    request_data = client_socket.recv(20000)

    post_userid, post_sign, post_timestamp, post_mes = getPost(request_data)
    # 回应socket
    initKey(post_userid, post_sign, post_timestamp, post_mes)
    # 关闭socket
    client_socket.close()


def getPost(request_data):
    # 格式化socket数据
    request_data = str(request_data, encoding="utf8").split('\r\n')
    print(request_data)
    items = []
    for item in request_data[1:-2]:
        items.append(item.split(':'))
    post_useful = dict(items)
    print(post_useful)
    print(request_data[-1])
    post_mes = json.loads(request_data[-1])
    print(post_mes)
    # 得到钉钉请求的sign,timestamp,发送者加密id,消息内容,strip()去除首尾空格
    post_sign = post_useful.get('sign').strip()
    print(post_sign)
    post_timestamp = post_useful.get('timestamp').strip()
    print(post_timestamp)
    post_userid = post_mes.get('senderId').strip()
    print(post_userid)
    post_mes = post_mes.get('text').get('content').strip()
    print(post_mes)

    return post_userid, post_sign, post_timestamp, post_mes


def initKey(post_userid, post_sign, post_timestamp, post_mes):
    # 配置token
    whtoken = "2c1ebd7abd938a41d6ad0d9a120bf4ed6572325209d02953ad87c68cc898cfdb"
    # 得到当前时间戳
    timestamp = str(round(time.time() * 1000))
    print(timestamp)
    # 计算签名
    app_secret = 'd8Ovq-OUKZRaPzzKLBCMN_E05RA8zDPbOs861zxb7y5b8jpLw67zHbWMQ4yKXo5y'
    app_secret_enc = app_secret.encode('utf-8')
    print(app_secret_enc)
    print('{}\n{}')
    string_to_sign = '{}\n{}'.format(post_timestamp, app_secret)
    print(string_to_sign)
    string_to_sign_enc = string_to_sign.encode('utf-8')
    print(string_to_sign_enc)
    hmac_code = hmac.new(app_secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    print(hmac_code)
    sign = base64.b64encode(hmac_code).decode('utf-8')
    print(sign)

    # 验证是否来自钉钉的合法请求
    if (abs(int(post_timestamp) - int(timestamp)) < 3600000 and post_sign == sign):
        webhook = "https://oapi.dingtalk.com/robot/send?access_token=" + whtoken + "&timestamp=" + timestamp + "&sign=" + sign
        header = {
            "Content-Type": "application/json",
            "Charset": "UTF-8"
        }
        # 发送消息
        message_json = json.dumps(selectMes(post_userid, post_mes))
        print(message_json)
        # 返回发送状态
        info = requests.post(url=webhook, data=message_json, headers=header)
        print(info)
    else:
        print("Warning:Not DingDing's post")


def selectMes(post_userid, post_mes):
    # 判断指令选择对应回复
    if (post_mes == '肥马去天津玩了啥'):
        send_mes = '玩了个寂寞'
        return sendText(post_userid, send_mes)
    elif (post_mes == '天津哪里最好玩'):
        send_mes = '垃圾加工厂'
        return sendText(post_userid, send_mes)
    elif (post_mes == '为什么要去天津'):
        send_mes = '闲的蛋疼'
        return sendMarkdown('莫名奇妙', send_mes)
    elif (post_mes == '真棒！'):
        send_mes = '鼓掌！'
        return sendText(post_userid, send_mes)
    elif (post_mes == '天气'):
        send_mes = getWeather()
        return sendMarkdown('天气预报', send_mes)
    else:
        return sendText(post_userid, '我懒得理你')


def sendText(post_userid, send_mes):
    # 发送文本形式
    message = {
        "msgtype": "text",
        "text": {
            "content": send_mes
        },
        "at": {
            "atDingtalkIds": [post_userid],
            "isAtAll": False
        }
    }
    return message


def sendMarkdown(title, send_mes):
    # 发送Markdown形式
    message = {
        "msgtype": "markdown",
        "markdown": {
            "title": title,
            "text": send_mes
        },
        "at": {
            "atDingtalkIds": [],
            "isAtAll": False
        }
    }
    return message


def getWeather():
    # 爬取天气数据返回的方法，这里就不多编写了
    send_mes = '#### 今日天气 \n' \
               '> 9度，西北风1级，空气良89，相对温度73%\n' \
               '> ![screenshot](https://img.alicdn.com/tfs/TB1NwmBEL9TBuNjy1zbXXXpepXa-2400-1218.png)\n' \
               '> ###### 10点20分发布 [天气](https://www.dingalk.com) \n'
    return send_mes


if __name__ == "__main__":
    # 启动服务,端口30421
    # 实例化一个socket，socket.AF_INET表示默认的ipv4，socket.SOCK_STREAM表示TCP协议
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # 绑定本机+端口号
    server_socket.bind(("", 30421))
    # listen中的函数表示排队的客户个数
    server_socket.listen(120)
    while True:
        # 接受客户端的连接请求,并返回一个套接字以及客服地址
        client_socket, client_address = server_socket.accept()
        #print("[%s, %s]用户连接上了" % client_address)

        # 创建进程，处理客户信息，将client_socket传给handle_client函数
        handle_client_process = Process(target=handle_client, args=(client_socket,))
        handle_client_process.start()
        #关闭客户socket，方便下次调用
        client_socket.close()

总结：
1.熟悉了python的一些加密、格式化、提取信息函数
2.熟悉了python的socket编程模板
3.体验了内网渗透的便利
4.体验了签名校验的安全手段

如有想法，欢迎交流！

DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

MiniMax M3 新手快速上手指南

在开发智能应用时，很多开发者往往卡在“如何快速让代码跑通”这一步。面对复杂的文档和繁多的配置项，容易陷入细节而忽略了核心逻辑的验证。其实，接入一个大语言模型并没有想象中那么困难，关键在于理清从密钥获取到首次成功调用的最小路径。一旦打通了这个闭环，后续的上下文管理、流式输出等高级功能自然就能顺势展开。对于正在构建客服机器人、智能助手或数据分析工具的技术人员来说，掌握标准的 API 调用流程是必备技能

DAMO开发者矩阵

【人工智能】AI基础知识（第四节：工具）

文章摘要：本文探讨了AI聊天机器人升级为Agent的关键步骤——工具集成。传统聊天机器人受限于"输入-思考-输出"模式，无法完成需要外部交互的复杂任务（如文件操作）。升级过程分为两个阶段：首先让大模型掌握工具元信息（功能、参数等），然后通过React技术实现"思考-执行-观察"循环。实例演示了Agent如何调用读写工具协同解决代码文件问题。最终指出，工具赋予是大模型从对话系统进化为实用Agent的

DAMO开发者矩阵

【机器人技术】基础搬运离线仿真完整实操

《机器人基础搬运离线仿真实训总结》摘要：本文记录了大专机器人专业学生通过离线仿真实训的学习体会。实训使用RobotStudio软件搭建工作站，完成工业机器人基础搬运任务仿真。相比真机调试，离线仿真具有零风险、高效试错等优势。文章详细介绍了工作站搭建、TCP标定、点位示教、程序编写等关键步骤，特别总结了新手常见的指令混淆、碰撞检测等实操问题。通过实训，作者认识到离线仿真是从理论到实操的重要过渡，能有