update: 增加命令读更新执行权限配置

maubot_llmplus/aibot.py

@@ -37,6 +37,31 @@ class AiBotPlugin(AbsExtraConfigPlugin):
         self.log.debug(f"{sender} doesn't match allowed_users")
         pass
 
+    def is_allow_command(self, sender: str, command_key: str) -> bool:
+        allow_users = self.config[command_key]
+        # 如果一个都没有配置，都没有权限可以执行更新命令
+        if len(allow_users) <= 0:
+            return False
+
+        # sender是否是配置中的一员, 如果是就允许进行命令修改执行，否则只有可读命令的执行
+        for u in allow_users:
+            if re.match(u, sender):
+                return True
+        self.log.debug(f"{sender} doesn't match {command_key}")
+        pass
+
+    def is_allow_update_read_command(self, sender: str) -> bool:
+        return self.is_allow_command(sender, "allow_update_read_command_users")
+
+    def is_allow_readonly_command(self, sender: str) -> bool:
+        is_update_read = self.is_allow_update_read_command(sender)
+        # 如果读写都有权限，就一定会有读权限，返回True
+        if is_update_read:
+            return True
+
+        # 如果没有读写权限，需要判断是否有只读权限
+        return self.is_allow_command(sender, "allow_readonly_command_users")
+
     """
     判断是否应该让AI进行回应
     回应条件：
@@ -142,6 +167,12 @@ class AiBotPlugin(AbsExtraConfigPlugin):
     """
     @ai_command.subcommand(help="View the configuration information currently in official use")
     async def info(self, event: MessageEvent) -> None:
+        # 判断是否有更新命令权限，如果没有就返回没有权限的提示
+        is_allow = self.is_allow_readonly_command(event.sender)
+        if not is_allow:
+            event.reply(f"{sender} have not read permission")
+            return
+
         show_infos = []
         # 当前机器人名称
         show_infos.append(f"bot name: {self.get_bot_name()}\n\n")
@@ -178,6 +209,12 @@ class AiBotPlugin(AbsExtraConfigPlugin):
     @ai_command.subcommand(help="List platforms or query current platform in use")
     @command.argument("argus")
     async def platform(self, event: MessageEvent, argus: str):
+        # 判断是否有更新命令权限，如果没有就返回没有权限的提示
+        is_allow = self.is_allow_readonly_command(event.sender)
+        if not is_allow:
+            event.reply(f"{sender} have not read permission")
+            return
+
         if argus == 'list':
             p_dict = dict(self.config['platforms'])
             platforms = [f"- {platform}" for platform in set(p_dict.keys())]
@@ -190,6 +227,12 @@ class AiBotPlugin(AbsExtraConfigPlugin):
     @ai_command.subcommand(help="List models or query current model in use")
     @command.argument("argus")
     async def model(self, event: MessageEvent, argus: str):
+        # 判断是否有更新命令权限，如果没有就返回没有权限的提示
+        is_allow = self.is_allow_readonly_command(event.sender)
+        if not is_allow:
+            event.reply(f"{sender} have not read permission")
+            return
+
         # 如果是list表示查看当前可以使用的模型列表
         if argus == 'list':
             platform = self.get_ai_platform()
@@ -204,6 +247,12 @@ class AiBotPlugin(AbsExtraConfigPlugin):
     @ai_command.subcommand(help="switch model in platform")
     @command.argument("argus")
     async def use(self, event: MessageEvent, argus: str):
+        # 判断是否有更新命令权限，如果没有就返回没有权限的提示
+        is_allow = self.is_allow_update_read_command(event.sender)
+        if not is_allow:
+            event.reply(f"{sender} have not update permission")
+            return
+
         platform = self.get_ai_platform()
         # 获取模型列表，判断使用的模型是否存在于列表中
         models = await platform.list_models()
@@ -217,6 +266,12 @@ class AiBotPlugin(AbsExtraConfigPlugin):
     @ai_command.subcommand(help="switch platform")
     @command.argument("argus")
     async def switch(self, event: MessageEvent, argus: str):
+        # 判断是否有更新命令权限，如果没有就返回没有权限的提示
+        is_allow = self.is_allow_update_read_command(event.sender)
+        if not is_allow:
+            event.reply(f"{sender} have not update permission")
+            return
+
         # 判断是否是本地ai模型，如果是还需要解析#后的type
         if argus == 'local_ai':
             await event.reply("local ai platform has ollama and lmstudio. "