ProjectHentai/nonebot2
6
0
Fork 0
mirror of https://github.com/nonebot/nonebot2.git synced 2024-11-24 09:05:04 +08:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

✨ ding: send by webhook

Browse Source 
close: #189

Feature: 钉钉添加 `send_by_webhook` 方法 #189
This commit is contained in:
Artin 2021-03-11 13:21:18 +08:00 committed by lengthmin
parent 46c65b919f
commit 0fec9915de
No known key found for this signature in database
GPG Key ID: 398474379D617D02
7 changed files with 173 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
docs/guide/ding-guide.md
View File

@ -11,6 +11,10 @@
- [群机器人概述](https://developers.dingtalk.com/document/app/overview-of-group-robots)
- [开发企业内部机器人](https://developers.dingtalk.com/document/app/develop-enterprise-internal-robots)
钉钉官方机器人教程（Java）：
- [开发一个钉钉机器人](https://developers.dingtalk.com/document/tutorial/create-a-robot)
## 安装 NoneBot 钉钉 适配器
```bash
@ -93,6 +97,58 @@ async def raw_handler(bot: DingBot, event: MessageEvent):
其他消息格式请查看 [钉钉适配器的 MessageSegment](https://github.com/nonebot/nonebot2/blob/dev/nonebot/adapters/ding/message.py#L8)，里面封装了很多有关消息的方法，比如 `code`、`image`、`feedCard` 等。
## 发送到特定群聊
钉钉也支持通过 Webhook 的方式直接将消息推送到某个群聊([参考链接](https://developers.dingtalk.com/document/app/custom-robot-access/title-zob-eyu-qse))，你可以在机器人的设置中看到当前群的 Webhook 地址。
![机器人所在群的 Webhook 地址](./images/ding/webhook.png)
获取到Webhook地址后，用户可以向这个地址发起HTTP POST 请求，即可实现给该钉钉群发送消息。
对于这种通过 Webhook 推送的消息，钉钉需要开发者进行安全方面的设置（目前有3种安全设置方式，请根据需要选择一种），如下：
1. **自定义关键词：** 最多可以设置10个关键词，消息中至少包含其中1个关键词才可以发送成功。
例如添加了一个自定义关键词：监控报警，则这个机器人所发送的消息，必须包含监控报警这个词，才能发送成功。
2. **加签：** 发送请求时带上验签的值，可以在机器人设置里看到密钥。
![加签密钥](./images/ding/jiaqian.png)
3. **IP地址（段）：** 设定后，只有来自IP地址范围内的请求才会被正常处理。支持两种设置方式：IP地址和IP地址段，暂不支持IPv6地址白名单。
如果你选择 1/3 两种安全设置，你需要自己确认当前网络和发送的消息能被钉钉接受，然后使用 `bot.send` 的时候将 webhook 地址传入 webhook 参数即可。
如我设置了 `打卡` 为关键词：
```python
message = MessageSegment.text("打卡成功：XXXXXX")
await hello.send(
message,
webhook=
"https://oapi.dingtalk.com/robot/send?access_token=XXXXXXXXXXXXXX",
)
```
对于第二种加签方式，你可以在 `bot.send` 的时候把 `secret` 参数传进去，Nonebot 内部会自动帮你计算发送该消息的签名并发送，如：
这里的 `secret` 参数就是加签选项给出的那个密钥。
```python
message = MessageSegment.raw({
"msgtype": "text",
"text": {
"content": 'hello from webhook,一定要注意安全方式的鉴权哦，否则可能发送失败的'
},
})
message += MessageSegment.atDingtalkIds(event.senderId)
await hello.send(
message,
webhook="https://oapi.dingtalk.com/robot/send?access_token=XXXXXXXXXXXXXX",
secret="SECXXXXXXXXXXXXXXXXXXXXXXXXX",
)
```
然后就可以发送成功了。
![测试 Webhook 发送](images/ding/test_webhook.png)
## 创建机器人并连接
在钉钉官方文档 [「开发企业内部机器人 -> 步骤一：创建机器人应用」](https://developers.dingtalk.com/document/app/develop-enterprise-internal-robots/title-ufs-4gh-poh) 中有详细介绍，这里就省去创建的步骤，介绍一下如何连接上程序。

BIN
docs/guide/images/ding/jiaqian.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 100 KiB

BIN
docs/guide/images/ding/test_webhook.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 75 KiB

BIN
docs/guide/images/ding/webhook.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 100 KiB

44
packages/nonebot-adapter-ding/nonebot/adapters/ding/bot.py
View File

@ -1,6 +1,9 @@
import hmac
import base64
import urllib.parse
from datetime import datetime
import time
from typing import Any, Union, Optional, TYPE_CHECKING
import httpx
@ -10,7 +13,7 @@ from nonebot.message import handle_event
from nonebot.adapters import Bot as BaseBot
from nonebot.exception import RequestDenied
from .utils import log
from .utils import calc_hmac_base64, log
from .config import Config as DingConfig
from .message import Message, MessageSegment
from .exception import NetworkError, ApiNotAvailable, ActionFailed, SessionExpired
@ -20,7 +23,7 @@ if TYPE_CHECKING:
from nonebot.config import Config
from nonebot.drivers import Driver
SEND_BY_SESSION_WEBHOOK = "send_by_sessionWebhook"
SEND = "send"
class Bot(BaseBot):
@ -72,10 +75,8 @@ class Bot(BaseBot):
if not sign:
log("WARNING", "Missing Signature Header")
raise RequestDenied(400, "Missing `sign` Header")
string_to_sign = f"{timestamp}\n{secret}"
sig = hmac.new(secret.encode("utf-8"),
string_to_sign.encode("utf-8"), "sha256").digest()
if sign != base64.b64encode(sig).decode("utf-8"):
sign_base64 = calc_hmac_base64(str(timestamp), secret)
if sign != sign_base64.decode('utf-8'):
log("WARNING", "Signature Header is invalid")
raise RequestDenied(403, "Signature is invalid")
else:
@ -142,15 +143,28 @@ class Bot(BaseBot):
return await bot.call_api(api, **data)
log("DEBUG", f"Calling API <y>{api}</y>")
params = {}
# 传入参数有 webhook，则使用传入的 webhook
webhook = data.get("webhook")
if api == SEND_BY_SESSION_WEBHOOK:
if webhook:
secret = data.get("secret")
if secret:
# 有这个参数的时候再计算加签的值
timestamp = str(round(time.time() * 1000))
params["timestamp"] = timestamp
hmac_code_base64 = calc_hmac_base64(timestamp, secret)
sign = urllib.parse.quote_plus(hmac_code_base64)
params["sign"] = sign
else:
# webhook 不存在则使用 event 中的 sessionWebhook
if event:
# 确保 sessionWebhook 没有过期
if int(datetime.now().timestamp()) > int(
event.sessionWebhookExpiredTime / 1000):
raise SessionExpired
target = event.sessionWebhook
webhook = event.sessionWebhook
else:
raise ApiNotAvailable
@ -160,9 +174,8 @@ class Bot(BaseBot):
raise ValueError("Message not found")
try:
async with httpx.AsyncClient(headers=headers) as client:
response = await client.post(
target,
params={"access_token": self.ding_config.access_token},
response = await client.post(webhook,
params=params,
json=message._produce(),
timeout=self.config.api_timeout)
@ -185,6 +198,8 @@ class Bot(BaseBot):
event: MessageEvent,
message: Union[str, "Message", "MessageSegment"],
at_sender: bool = False,
webhook: Optional[str] = None,
secret: Optional[str] = None,
**kwargs) -> Any:
"""
:说明:
@ -196,6 +211,8 @@ class Bot(BaseBot):
* ``event: Event``: Event 对象
* ``message: Union[str, Message, MessageSegment]``: 要发送的消息
* ``at_sender: bool``: 是否 @ 事件主体
* ``webhook: Optional[str]``: 该条消息将调用的 webhook 地址。不传则将使用 sessionWebhook，若其也不存在，该条消息不发送，使用自定义 webhook 时注意你设置的安全方式，如加关键词，IP地址，加签等等。
* ``secret: Optional[str]``: 如果你使用自定义的 webhook 地址，推荐使用加签方式对消息进行验证，将 `机器人安全设置页面，加签一栏下面显示的SEC开头的字符串` 传入这个参数即可。
* ``**kwargs``: 覆盖默认参数
:返回:
@ -213,6 +230,9 @@ class Bot(BaseBot):
at_sender = at_sender and bool(event.senderId)
params = {}
params["event"] = event
if webhook:
params["webhook"] = webhook
params["secret"] = secret
params.update(kwargs)
if at_sender and event.conversationType != ConversationType.private:
@ -222,4 +242,4 @@ class Bot(BaseBot):
else:
params["message"] = msg
return await self.call_api(SEND_BY_SESSION_WEBHOOK, **params)
return await self.call_api(SEND, **params)

14
packages/nonebot-adapter-ding/nonebot/adapters/ding/utils.py
View File

@ -1,3 +1,15 @@
import hmac
from nonebot.utils import logger_wrapper
import hashlib
import base64
log = logger_wrapper("DING")
def calc_hmac_base64(timestamp: str, secret: str):
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()
return base64.b64encode(hmac_code)

51
tests/test_plugins/test_ding.py
View File

@ -3,6 +3,30 @@ from nonebot.rule import to_me
from nonebot.plugin import on_command
from nonebot.adapters.ding import Bot as DingBot, MessageSegment, MessageEvent
helper = on_command("ding_helper", to_me())
@helper.handle()
async def ding_helper(bot: DingBot, event: MessageEvent):
message = MessageSegment.markdown(
"Hello, This is NoneBot",
"""帮助信息如下：\n
[ding_helper](dtmd://dingtalkclient/sendMessage?content=ding_helper) 查看帮助\n
[markdown](dtmd://dingtalkclient/sendMessage?content=markdown) 发送 markdown\n
[actionCardSingleBtn](dtmd://dingtalkclient/sendMessage?content=actionCardSingleBtn)\n
[actionCard](dtmd://dingtalkclient/sendMessage?content=actionCard)\n
[feedCard](dtmd://dingtalkclient/sendMessage?content=feedCard)\n
[atme](dtmd://dingtalkclient/sendMessage?content=atme)\n
[image](dtmd://dingtalkclient/sendMessage?content=image)\n
[t](dtmd://dingtalkclient/sendMessage?content=t)\n
[code](dtmd://dingtalkclient/sendMessage?content=code) 发送代码\n
[test_message](dtmd://dingtalkclient/sendMessage?content=test_message)\n
[hello](dtmd://dingtalkclient/sendMessage?content=hello)\n
[webhook](dtmd://dingtalkclient/sendMessage?content=webhook)""",
)
await markdown.finish(message)
markdown = on_command("markdown", to_me())
@ -184,3 +208,30 @@ async def hello_handler(bot: DingBot, event: MessageEvent):
message = MessageSegment.text(f"@{event.senderId}，你好")
message += MessageSegment.atDingtalkIds(event.senderId)
await hello.finish(message)
hello = on_command("webhook", to_me())
@hello.handle()
async def webhook_handler(bot: DingBot, event: MessageEvent):
print(event)
message = MessageSegment.raw({
"msgtype": "text",
"text": {
"content": 'hello from webhook,一定要注意安全方式的鉴权哦，否则可能发送失败的'
},
})
message += MessageSegment.atDingtalkIds(event.senderId)
await hello.send(
message,
webhook=
"https://oapi.dingtalk.com/robot/send?access_token=XXXXXXXXXXXXXX",
secret="SECXXXXXXXXXXXXXXXXXXXXXXXXX")
message = MessageSegment.text("TEST 123123 S")
await hello.send(
message,
webhook=
"https://oapi.dingtalk.com/robot/send?access_token=XXXXXXXXXXXXXX",
)