✨ version 0.30.2

permission check

src/nonebot_plugin_alconna/__init__.py

@@ -89,7 +89,7 @@
 from .consts import ALCONNA_EXEC_RESULT as ALCONNA_EXEC_RESULT
 from .extension import add_global_extension as add_global_extension
 
-__version__ = "0.30.1"
+__version__ = "0.30.2"
 
 __plugin_meta__ = PluginMetadata(
     name="Alconna 插件",

src/nonebot_plugin_alconna/extension.py

@@ -42,6 +42,7 @@ def __init_subclass__(cls, **kwargs):
         cls._overrides = {
             "send_wrapper": cls.send_wrapper != Extension.send_wrapper,
             "receive_wrapper": cls.receive_wrapper != Extension.receive_wrapper,
+            "permission_check": cls.permission_check != Extension.permission_check,
             "parse_wrapper": cls.parse_wrapper != Extension.parse_wrapper,
             "catch": cls.catch != Extension.catch,
         }
@@ -88,6 +89,10 @@ async def receive_wrapper(self, bot: Bot, event: Event, command: Alconna, receiv
         """接收消息后的钩子函数。"""
         return receive
 
+    async def permission_check(self, bot: Bot, event: Event, command: Alconna) -> bool:
+        """命令首次解析并确认头部匹配（即确认选择响应）时对发送者的权限判断"""
+        return True
+
     async def parse_wrapper(self, bot: Bot, state: T_State, event: Event, res: Arparma) -> None:
         """解析消息后的钩子函数。"""
         pass
@@ -207,6 +212,14 @@ async def receive_wrapper(self, bot: Bot, event: Event, receive: TM) -> TM:
                 res = await ext.receive_wrapper(bot, event, self._rule.command, res)
         return res
 
+    async def permission_check(self, bot: Bot, event: Event) -> bool:
+        for ext in self.context:
+            if ext._overrides["permission_check"]:
+                if await ext.permission_check(bot, event, self._rule.command) is False:
+                    return False
+                continue
+        return True
+
     async def parse_wrapper(self, bot: Bot, state: T_State, event: Event, res: Arparma) -> None:
         await asyncio.gather(
             *(

src/nonebot_plugin_alconna/rule.py

@@ -1,5 +1,5 @@
 import asyncio
-from typing import List, Type, Union, Optional, cast
+from typing import List, Type, Union, Literal, Optional, cast
 
 from nonebot import get_driver
 from nonebot.typing import T_State
@@ -155,14 +155,16 @@ def __eq__(self, other: object) -> bool:
     def __hash__(self) -> int:
         return hash(self.command.__hash__())
 
-    async def handle(self, bot: Bot, event: Event, msg: Message):
+    async def handle(self, bot: Bot, event: Event, msg: Message) -> Union[Arparma, Literal[False]]:
         if self.comp_config is None:
             return self.command.parse(msg)
         res = None
         with self._interface:
             res = self.command.parse(msg)
         if res:
             return res
+        if not await self.executor.permission_check(bot, event):
+            return False
         self._session = event.get_session_id()
         self._waiter.permission = Permission(User.from_event(event))
         matchers[self._waiter.priority].append(self._waiter)
@@ -226,6 +228,8 @@ async def __call__(self, event: Event, state: T_State, bot: Bot) -> bool:
             output_manager.set_action(lambda x: x, self.command.name)
             try:
                 arp = await self.handle(bot, event, msg)
+                if arp is False:
+                    return False
             except Exception as e:
                 arp = Arparma(self.command.path, msg, False, error_info=e)
             may_help_text: Optional[str] = cap.get("output", None)
@@ -240,6 +244,8 @@ async def __call__(self, event: Event, state: T_State, bot: Bot) -> bool:
         if self.auto_send and may_help_text:
             await self.send(may_help_text, bot, event, arp)
             return False
+        if not await self.executor.permission_check(bot, event):
+            return False
         await self.executor.parse_wrapper(bot, state, event, arp)
         state[ALCONNA_RESULT] = CommandResult(self.command, arp, may_help_text)
         exec_result = self.command.exec_result

src/nonebot_plugin_alconna/uniseg/__init__.py

@@ -24,7 +24,7 @@
 from .params import UniversalMessage as UniversalMessage
 from .params import UniversalSegment as UniversalSegment
 
-__version__ = "0.30.1"
+__version__ = "0.30.2"
 
 __plugin_meta__ = PluginMetadata(
     name="Universal Segment 插件",

tests/test_extension.py

@@ -22,14 +22,20 @@ def priority(self) -> int:
         def id(self) -> str:
             return "demo"
 
+        async def permission_check(self, bot, event, command):
+            if event.get_user_id() != "123":
+                await bot.send(event, "权限不足！")
+                return False
+            return True
+
         async def catch(self, interface: Interface[MessageEvent]):
             if interface.annotation is str:
                 return {
                     "hello": "Hello!",
                     "world": "World!",
                 }.get(interface.name, interface.name)
 
-    add = on_alconna(Alconna("add", Args["a", float]["b", float]), extensions=[DemoExtension])
+    add = on_alconna(Alconna("add", Args["a", float]["b", float]), extensions=[DemoExtension], comp_config={})
 
     @add.handle()
     async def h(a: float, b: float, hello: str, world: str, test: str):
@@ -42,3 +48,7 @@ async def h(a: float, b: float, hello: str, world: str, test: str):
         event = fake_group_message_event_v11(message=Message("add 1.3 2.4"), user_id=123)
         ctx.receive_event(bot, event)
         ctx.should_call_send(event, "1.3 + 2.4 = 3.7\nHello! World! test!")
+
+        event = fake_group_message_event_v11(message=Message("add 1.3 2.4"), user_id=456)
+        ctx.receive_event(bot, event)
+        ctx.should_call_send(event, "权限不足！")