ProjectHentai/nonebot2
6
0
Fork 0
mirror of https://github.com/nonebot/nonebot2.git synced 2025-07-28 08:41:29 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

🚧 update ding adapter

Browse Source
This commit is contained in:
yanyongyu
2020-12-03 17:08:16 +08:00
parent dc691889e3
commit afd01796aa
12 changed files with 496 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
nonebot/adapters/cqhttp/__init__.py
View File

@ -14,3 +14,4 @@ from .event import Event
from .message import Message, MessageSegment
from .utils import log, escape, unescape, _b2s
from .bot import Bot, _check_at_me, _check_nickname, _check_reply, _handle_api_result
from .exception import CQHTTPAdapterException, ApiNotAvailable, ActionFailed, NetworkError

8
nonebot/adapters/cqhttp/bot.py
View File

@ -247,7 +247,7 @@ class Bot(BaseBot):
"""
x_self_id = headers.get("x-self-id")
x_signature = headers.get("x-signature")
access_token = get_auth_bearer(headers.get("authorization"))
token = get_auth_bearer(headers.get("authorization"))
# 检查连接方式
if connection_type not in ["http", "websocket"]:
@ -272,13 +272,13 @@ class Bot(BaseBot):
raise RequestDenied(403, "Signature is invalid")
access_token = driver.config.access_token
if access_token and access_token != access_token:
if access_token and access_token != token:
log(
"WARNING", "Authorization Header is invalid"
if access_token else "Missing Authorization Header")
if token else "Missing Authorization Header")
raise RequestDenied(
403, "Authorization Header is invalid"
if access_token else "Missing Authorization Header")
if token else "Missing Authorization Header")
return str(x_self_id)
@overrides(BaseBot)

4
nonebot/adapters/ding/__init__.py
View File

@ -9,7 +9,9 @@
"""
from .utils import log
from .bot import Bot
from .event import Event
from .message import Message, MessageSegment
from .exception import ApiError, SessionExpired, DingAdapterException
from .exception import (DingAdapterException, ApiNotAvailable, NetworkError,
ActionFailed, SessionExpired)

76
nonebot/adapters/ding/bot.py
View File

@ -1,19 +1,20 @@
import httpx
import hmac
import base64
from datetime import datetime
import httpx
from nonebot.log import logger
from nonebot.config import Config
from nonebot.adapters import BaseBot
from nonebot.message import handle_event
from nonebot.typing import Driver, NoReturn
from nonebot.typing import Any, Union, Optional
from nonebot.exception import NetworkError, RequestDenied, ApiNotAvailable
from nonebot.exception import RequestDenied
from nonebot.typing import Any, Union, Driver, Optional, NoReturn
from .utils import log
from .event import Event
from .model import MessageModel
from .utils import check_legal, log
from .message import Message, MessageSegment
from .exception import ApiError, SessionExpired
from .exception import NetworkError, ApiNotAvailable, ActionFailed, SessionExpired
class Bot(BaseBot):
@ -35,8 +36,7 @@ class Bot(BaseBot):
@classmethod
async def check_permission(cls, driver: Driver, connection_type: str,
headers: dict,
body: Optional[dict]) -> Union[str, NoReturn]:
headers: dict, body: Optional[dict]) -> str:
"""
:说明:
@ -45,25 +45,29 @@ class Bot(BaseBot):
timestamp = headers.get("timestamp")
sign = headers.get("sign")
# 检查 timestamp
if not timestamp:
raise RequestDenied(400, "Missing `timestamp` Header")
# 检查 sign
if not sign:
raise RequestDenied(400, "Missing `sign` Header")
# 校验 sign 和 timestamp，判断是否是来自钉钉的合法请求
if not check_legal(timestamp, sign, driver):
raise RequestDenied(403, "Signature is invalid")
# 检查连接方式
if connection_type not in ["http"]:
raise RequestDenied(405, "Unsupported connection type")
access_token = driver.config.access_token
if access_token and access_token != access_token:
raise RequestDenied(
403, "Authorization Header is invalid"
if access_token else "Missing Authorization Header")
return body.get("chatbotUserId")
# 检查 timestamp
if not timestamp:
raise RequestDenied(400, "Missing `timestamp` Header")
# 检查 sign
secret = driver.config.secret
if secret:
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"):
log("WARNING", "Signature Header is invalid")
raise RequestDenied(403, "Signature is invalid")
else:
log("WARNING", "Ding signature check ignored!")
return body["chatbotUserId"]
async def handle_message(self, body: dict):
message = MessageModel.parse_obj(body)
@ -79,7 +83,10 @@ class Bot(BaseBot):
)
return
async def call_api(self, api: str, **data) -> Union[Any, NoReturn]:
async def call_api(self,
api: str,
event: Optional[Event] = None,
**data) -> Union[Any, NoReturn]:
"""
:说明:
@ -111,13 +118,15 @@ class Bot(BaseBot):
log("DEBUG", f"Calling API <y>{api}</y>")
if api == "send_message":
raw_event: MessageModel = data["raw_event"]
# 确保 sessionWebhook 没有过期
if int(datetime.now().timestamp()) > int(
raw_event.sessionWebhookExpiredTime / 1000):
raise SessionExpired
if event:
# 确保 sessionWebhook 没有过期
if int(datetime.now().timestamp()) > int(
event.raw_event.sessionWebhookExpiredTime / 1000):
raise SessionExpired
target = raw_event.sessionWebhook
target = event.raw_event.sessionWebhook
else:
target = None
if not target:
raise ApiNotAvailable
@ -136,8 +145,8 @@ class Bot(BaseBot):
result = response.json()
if isinstance(result, dict):
if result.get("errcode") != 0:
raise ApiError(errcode=result.get("errcode"),
errmsg=result.get("errmsg"))
raise ActionFailed(errcode=result.get("errcode"),
errmsg=result.get("errmsg"))
return result
raise NetworkError(f"HTTP request received unexpected "
f"status code: {response.status_code}")
@ -176,7 +185,8 @@ class Bot(BaseBot):
msg = message if isinstance(message, Message) else Message(message)
at_sender = at_sender and bool(event.user_id)
params = {"raw_event": event.raw_event}
params = {}
params["event"] = event
params.update(kwargs)
if at_sender and event.detail_type != "private":

9
nonebot/adapters/ding/event.py
View File

@ -1,6 +1,5 @@
from typing import Literal, Union, Optional
from nonebot.adapters import BaseEvent
from nonebot.typing import Union, Optional
from .message import Message
from .model import MessageModel, ConversationType, TextMessage
@ -67,7 +66,7 @@ class Event(BaseEvent):
pass
@property
def detail_type(self) -> Literal["private", "group"]:
def detail_type(self) -> str:
"""
- 类型: ``str``
- 说明: 事件详细类型
@ -125,10 +124,6 @@ class Event(BaseEvent):
"""
return self.detail_type == "private" or self.raw_event.isInAtList
@to_me.setter
def to_me(self, value) -> None:
pass
@property
def message(self) -> Optional["Message"]:
"""

46
nonebot/adapters/ding/exception.py
View File

@ -1,4 +1,8 @@
from nonebot.exception import AdapterException, ActionFailed, ApiNotAvailable
from nonebot.typing import Optional
from nonebot.exception import (AdapterException, ActionFailed as
BaseActionFailed, ApiNotAvailable as
BaseApiNotAvailable, NetworkError as
BaseNetworkError)
class DingAdapterException(AdapterException):
@ -6,22 +10,27 @@ class DingAdapterException(AdapterException):
:说明:
钉钉 Adapter 错误基类
"""
def __init__(self) -> None:
super().__init__("ding")
class ApiError(DingAdapterException, ActionFailed):
class ActionFailed(BaseActionFailed, DingAdapterException):
"""
:说明:
API 请求返回错误信息。
:参数:
* ``errcode: Optional[int]``: 错误码
* ``errmsg: Optional[str]``: 错误信息
"""
def __init__(self, errcode: int, errmsg: str):
def __init__(self,
errcode: Optional[int] = None,
errmsg: Optional[str] = None):
super().__init__()
self.errcode = errcode
self.errmsg = errmsg
@ -30,12 +39,37 @@ class ApiError(DingAdapterException, ActionFailed):
return f"<ApiError errcode={self.errcode} errmsg={self.errmsg}>"
class SessionExpired(DingAdapterException, ApiNotAvailable):
class ApiNotAvailable(BaseApiNotAvailable, DingAdapterException):
pass
class NetworkError(BaseNetworkError, DingAdapterException):
"""
:说明:
网络错误。
:参数:
* ``retcode: Optional[int]``: 错误码
"""
def __init__(self, msg: Optional[str] = None):
super().__init__()
self.msg = msg
def __repr__(self):
return f"<NetWorkError message={self.msg}>"
def __str__(self):
return self.__repr__()
class SessionExpired(BaseApiNotAvailable, DingAdapterException):
"""
:说明:
发消息的 session 已经过期。
"""
def __repr__(self) -> str:

32
nonebot/adapters/ding/utils.py
View File

@ -1,35 +1,3 @@
import base64
import hashlib
import hmac
from typing import TYPE_CHECKING
from nonebot.utils import logger_wrapper
if TYPE_CHECKING:
from nonebot.drivers import BaseDriver
log = logger_wrapper("DING")
def check_legal(timestamp, remote_sign, driver: "BaseDriver"):
"""
1. timestamp 与系统当前时间戳如果相差1小时以上，则认为是非法的请求。
2. sign 与开发者自己计算的结果不一致，则认为是非法的请求。
必须当timestamp和sign同时验证通过，才能认为是来自钉钉的合法请求。
"""
# 目前先设置成 secret
# TODO 后面可能可以从 secret[adapter_name] 获取
app_secret = driver.config.secret # 机器人的 appSecret
if not app_secret:
# TODO warning
log("WARNING", "No ding secrets set, won't check sign")
return True
app_secret_enc = app_secret.encode('utf-8')
string_to_sign = '{}\n{}'.format(timestamp, app_secret)
string_to_sign_enc = string_to_sign.encode('utf-8')
hmac_code = hmac.new(app_secret_enc,
string_to_sign_enc,
digestmod=hashlib.sha256).digest()
sign = base64.b64encode(hmac_code).decode('utf-8')
return remote_sign == sign

23
nonebot/utils.py
View File

@ -5,7 +5,7 @@ import dataclasses
from functools import wraps, partial
from nonebot.log import logger
from nonebot.typing import Any, Callable, Awaitable, overrides
from nonebot.typing import Any, Optional, Callable, Awaitable, overrides
def escape_tag(s: str) -> str:
@ -65,19 +65,20 @@ class DataclassEncoder(json.JSONEncoder):
def logger_wrapper(logger_name: str):
"""
:说明:
def log(level: str, message: str):
"""
:说明:
用于打印 adapter 的日志。
用于打印 adapter 的日志。
:log 参数:
:参数:
* ``level: Literal['WARNING', 'DEBUG', 'INFO']``: 日志等级
* ``message: str``: 日志信息
* ``exception: Optional[Exception]``: 异常信息
"""
* ``level: Literal['WARNING', 'DEBUG', 'INFO']``: 日志等级
* ``message: str``: 日志信息
"""
return logger.opt(colors=True).log(level,
f"<m>{logger_name}</m> | " + message)
def log(level: str, message: str, exception: Optional[Exception] = None):
return logger.opt(colors=True, exception=exception).log(
level, f"<m>{logger_name}</m> | " + message)
return log