bugfix:修复命令前缀判断问题、修复超级用户无法豁免问题 (#2141)

* bugfix:修复命令前缀判断问题、修复超级用户无法豁免问题

* bugfix:修复格式问题

Author: Copaan

zhenxun/builtin_plugins/hooks/auth/auth_limit.py

@@ -346,7 +346,7 @@ async def __reserve(
         limit = limit_model.limit
         limiter = limit_model.limiter
         is_limit = (
-            LimitWatchType.ALL
+            limit.watch_type == LimitWatchType.ALL
             or (group_id and limit.watch_type == LimitWatchType.GROUP)
             or (not group_id and limit.watch_type == LimitWatchType.USER)
         )

zhenxun/builtin_plugins/hooks/auth_activation.py

@@ -71,6 +71,7 @@ class HandlerDescriptor:
 class AlconnaShortcutDescriptor:
     pattern: str
     fuzzy: bool = False
+    prefix: bool = False
     flags: int = 0
 
 
@@ -1059,6 +1060,7 @@ def _extract_alconna_shortcut_descriptors(
                     AlconnaShortcutDescriptor(
                         pattern=pattern,
                         fuzzy=bool(getattr(args, "fuzzy", False)),
+                        prefix=bool(getattr(args, "prefix", False)),
                         flags=int(getattr(args, "flags", 0) or 0),
                     )
                 )
@@ -1140,7 +1142,7 @@ def matcher_alconna_head_matches(
             if decision == "unknown":
                 saw_unknown = True
         for shortcut in alconna.shortcuts:
-            decision = _alconna_shortcut_matches(text, shortcut)
+            decision = _alconna_shortcut_matches(text, shortcut, alconna.prefixes)
             if decision == "match":
                 return "match"
             if decision == "unknown":
@@ -1194,11 +1196,43 @@ def _alconna_literal_head_matches(text: str, head: str, *, compact: bool) -> boo
 def _alconna_shortcut_matches(
     text: str,
     shortcut: AlconnaShortcutDescriptor,
+    prefixes: tuple[str, ...] = (),
 ) -> ActivationDecision:
     pattern = shortcut.pattern.strip()
     if not pattern:
         return "unknown"
+    saw_unknown = False
+    for normalized in _alconna_shortcut_patterns(pattern, shortcut, prefixes):
+        decision = _alconna_shortcut_pattern_matches(text, normalized, shortcut)
+        if decision == "match":
+            return "match"
+        if decision == "unknown":
+            saw_unknown = True
+    return "unknown" if saw_unknown else "miss"
+
+
+def _alconna_shortcut_patterns(
+    pattern: str,
+    shortcut: AlconnaShortcutDescriptor,
+    prefixes: tuple[str, ...],
+) -> tuple[str, ...]:
     normalized = normalize_shortcut_pattern(pattern)
+    if not normalized:
+        return ()
+    patterns = [normalized]
+    if shortcut.prefix:
+        for prefix in prefixes or ("",):
+            candidate = normalize_shortcut_pattern(f"{prefix}{normalized}")
+            if candidate and candidate not in patterns:
+                patterns.append(candidate)
+    return tuple(patterns)
+
+
+def _alconna_shortcut_pattern_matches(
+    text: str,
+    normalized: str,
+    shortcut: AlconnaShortcutDescriptor,
+) -> ActivationDecision:
     placeholder_match = _placeholder_shortcut_decision(text, normalized)
     if placeholder_match == "match":
         return "match"
@@ -1211,8 +1245,10 @@ def _alconna_shortcut_matches(
         return "match"
     try:
         if shortcut.fuzzy:
-            return "match" if re.match(f"^{pattern}", text, shortcut.flags) else "miss"
-        return "match" if re.fullmatch(pattern, text, shortcut.flags) else "miss"
+            return (
+                "match" if re.match(f"^{normalized}", text, shortcut.flags) else "miss"
+            )
+        return "match" if re.fullmatch(normalized, text, shortcut.flags) else "miss"
     except re.error:
         return "unknown"
 

zhenxun/builtin_plugins/hooks/auth_checker.py

@@ -496,10 +496,11 @@ def add(text: object) -> None:
             candidates.append(normalized)
 
     add(_trie_command_text_from_state(state))
-    add(_trie_raw_command_from_state(state))
+    trie_raw = _trie_raw_command_from_state(state)
+    add(trie_raw)
     trie_arg = _trie_command_arg_text_from_state(state)
-    if _trie_raw_command_from_state(state) and trie_arg:
-        add(f"{_trie_raw_command_from_state(state)} {trie_arg}")
+    if trie_raw and trie_arg:
+        add(f"{trie_raw} {trie_arg}")
     add(plain_text)
     if event is not None:
         with contextlib.suppress(Exception):
@@ -794,9 +795,10 @@ def _prepare_handle_event_state(event: Event, state: dict) -> None:
 
 
 def _build_matcher_state(base_state: dict) -> dict:
+    # 第一次调用即在 base_state 写入副作用缓存对象;copy() 后 matcher_state
+    # 与之共享同一引用,无需二次 get(B7)。
     get_permission_side_effect_cache(state=base_state)
     matcher_state = base_state.copy()
-    get_permission_side_effect_cache(state=matcher_state)
     return matcher_state
 
 
@@ -965,25 +967,28 @@ async def _db_section():
         DB_ACTIVE_COUNT = max(DB_ACTIVE_COUNT - 1, 0)
 
 
+_POLICY_SKIP_MESSAGES = {
+    "user_or_group_banned": "user or group banned (cached)",
+    "superuser_required": "超级管理员权限不足...",
+    "admin_required": "管理员权限不足...",
+    "bot_not_found": "Bot不存在，阻断权限检测...",
+    "bot_sleeping": "Bot休眠中阻断权限检测...",
+    "bot_plugin_blocked": "Bot插件权限检查结果为关闭...",
+    "group_not_found": "群组信息不存在...",
+    "group_blacklisted": "群组黑名单, 目标群组群权限权限-1...",
+    "group_sleeping": "群组休眠状态...",
+    "group_level_low": "群等级限制...",
+    "admin_level_low": "管理员权限不足...",
+    "plugin_disabled_in_group": "该插件在群组中已被禁用...",
+    "plugin_superuser_blocked_in_group": "超级管理员禁用了该群此功能...",
+    "plugin_blocked_in_group": "该群未开启此功能...",
+    "plugin_disabled_in_private": "该插件在私聊中已被禁用...",
+    "plugin_global_disabled": "全局未开启此功能...",
+}
+
+
 def _policy_skip_message(reason: str) -> str:
-    return {
-        "user_or_group_banned": "user or group banned (cached)",
-        "superuser_required": "超级管理员权限不足...",
-        "admin_required": "管理员权限不足...",
-        "bot_not_found": "Bot不存在，阻断权限检测...",
-        "bot_sleeping": "Bot休眠中阻断权限检测...",
-        "bot_plugin_blocked": "Bot插件权限检查结果为关闭...",
-        "group_not_found": "群组信息不存在...",
-        "group_blacklisted": "群组黑名单, 目标群组群权限权限-1...",
-        "group_sleeping": "群组休眠状态...",
-        "group_level_low": "群等级限制...",
-        "admin_level_low": "管理员权限不足...",
-        "plugin_disabled_in_group": "该插件在群组中已被禁用...",
-        "plugin_superuser_blocked_in_group": "超级管理员禁用了该群此功能...",
-        "plugin_blocked_in_group": "该群未开启此功能...",
-        "plugin_disabled_in_private": "该插件在私聊中已被禁用...",
-        "plugin_global_disabled": "全局未开启此功能...",
-    }.get(reason, reason or "permission denied")
+    return _POLICY_SKIP_MESSAGES.get(reason, reason or "permission denied")
 
 
 # 超时装饰器
@@ -1342,18 +1347,24 @@ async def _check_ban_from_snapshot(
     hook_recorder: HookTraceRecorder,
     session: Uninfo,
 ) -> None:
+    # skip_ban 上移到 cached 判断之前(A7),否则豁免参数对 cached 命中形同虚设。
+    if skip_ban:
+        hook_recorder.set("auth_ban", "skipped")
+        return
+    is_superuser = bool(getattr(prep.permission_context, "is_superuser", False))
     ban_cache_state = prep.snapshot.ban_state
     if event_cache is not None:
         ban_cache_state = event_cache.get("ban_state")
     if ban_cache_state is True:
+        # 超级用户豁免(A7):与 PDP / 旧轨权威路径保持一致,避免被 ban 后无法自救。
+        if is_superuser:
+            hook_recorder.set("auth_ban", "cached_superuser_exempt")
+            return
         hook_recorder.set("auth_ban", "cached")
         raise SkipPluginException("user or group banned (cached)")
     if ban_cache_state is False:
         hook_recorder.set("auth_ban", "cached")
         return
-    if skip_ban:
-        hook_recorder.set("auth_ban", "skipped")
-        return
 
     ban_start = time.time()
     try:

zhenxun/builtin_plugins/hooks/auth_pipeline.py

@@ -246,7 +246,11 @@ async def route_gate_stage(
     if deps.is_hidden_plugin(ctx.matcher):
         ctx.stop(allowed=True, effect="allow", reason="hidden_plugin")
         return
-    if ctx.event_cache is not None and ctx.event_cache.get("ban_state") is True:
+    if (
+        ctx.event_cache is not None
+        and ctx.event_cache.get("ban_state") is True
+        and not ctx.event_context.is_superuser
+    ):
         ctx.decision_effect = "skip"
         ctx.decision_reason = "ban_cached"
         raise SkipPluginException("user or group banned (cached)")

zhenxun/builtin_plugins/hooks/auth_policy.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import contextlib
 from dataclasses import dataclass, field
 from typing import Any, Literal
 
@@ -116,8 +117,14 @@ def decide_bot(self, context: PolicyContext) -> PolicyDecision:
         if not bot_data.status and not context.allow_sleep_bypass:
             return PolicyDecision("deny", "bot_sleeping")
         module = snapshot.profile.module
-        if module and self._module_in_block_string(module, bot_data.block_plugins):
-            return PolicyDecision("deny", "bot_plugin_blocked")
+        if module:
+            value = bot_data.block_plugins or ""
+            # 缓存解析后的 frozenset,避免每次 bot 检查重复 split(B8-3);
+            # 仍保留原子串判定以保持行为等价。
+            if CommonUtils.format(module) in value or module in self._bot_block_set(
+                bot_data
+            ):
+                return PolicyDecision("deny", "bot_plugin_blocked")
         return PolicyDecision("allow", "bot_allowed")
 
     def decide_group(self, context: PolicyContext) -> PolicyDecision:
@@ -208,6 +215,17 @@ def _group_block_sets(group: object) -> tuple[frozenset[str], frozenset[str]]:
             setattr(group, "superuser_block_plugin_set", super_block_set)
         return block_set, super_block_set
 
+    @staticmethod
+    def _bot_block_set(bot_data: object) -> frozenset[str]:
+        block_set = getattr(bot_data, "block_plugin_set", None)
+        if block_set is None:
+            block_set = _parse_block_modules(
+                getattr(bot_data, "block_plugins", "") or ""
+            )
+            with contextlib.suppress(Exception):
+                setattr(bot_data, "block_plugin_set", block_set)
+        return block_set
+
     @staticmethod
     def _module_in_block_string(module: str, value: str | None) -> bool:
         if not value:

zhenxun/builtin_plugins/hooks/auth_runtime_config.py

@@ -23,7 +23,7 @@ class AuthDispatchRuntimeConfig:
     circuit_reset_time: int = 300
     matcher_route_prefilter_ttl: int = 2
     prefilter_stats_log_interval: float = 10.0
-    cache_sweep_interval: float = 1.0
+    cache_sweep_interval: float = 45.0
     dispatch_stats_log_interval: float = 10.0
 
 

zhenxun/builtin_plugins/hooks/chkdsk_hook.py

@@ -82,6 +82,11 @@ async def _(
     if matcher.type == "notice":
         return
 
+    # AI 重路由注入的合成事件不计入恶意检测(A6):AI 链路有自己的预算/审批,
+    # 不应被人类反垃圾逻辑封禁(此前批量转发误封超级用户的事故根因之一)。
+    if getattr(event, "_ai_triggered", False):
+        return
+
     # 提前判断插件类型，跳过不需要检测的插件
     if plugin := matcher.plugin:
         if metadata := plugin.metadata:
@@ -99,6 +104,13 @@ async def _(
 
     user_id = resolve_actor_user_id(event, session.id1)
     group_id = resolve_event_group_id(event, session.id3 or session.id2)
+    # 超级用户豁免恶意检测(A6):与权威权限路径保持一致,避免误封管理者。
+    if user_id:
+        is_superuser = state.get("_zx_is_superuser")
+        if not isinstance(is_superuser, bool):
+            is_superuser = user_id in bot.config.superusers
+        if is_superuser:
+            return
     malicious_check_time = float(_get_positive_config("MALICIOUS_CHECK_TIME", float))
     malicious_ban_count = int(_get_positive_config("MALICIOUS_BAN_COUNT", int))
     malicious_ban_time = int(_get_positive_config("MALICIOUS_BAN_TIME", int))

zhenxun/builtin_plugins/statistics/statistics_hook.py

@@ -53,8 +53,12 @@ async def _flush_statistics_buffer(reason: str) -> int:
 
 
 async def _append_statistics(record: Statistics) -> None:
-    TEMP_LIST.append(record)
-    if len(TEMP_LIST) >= STATS_BUFFER_FLUSH_SIZE and not _STATS_FLUSH_LOCK.locked():
+    # 在锁内追加(B8),与 flush 的 copy+clear 串行,消除逻辑窗口;
+    # flush 自身再次获取同一把锁,故在锁外触发避免重入。
+    async with _STATS_FLUSH_LOCK:
+        TEMP_LIST.append(record)
+        should_flush = len(TEMP_LIST) >= STATS_BUFFER_FLUSH_SIZE
+    if should_flush:
         await _flush_statistics_buffer("缓冲区触发")
 
 

zhenxun/models/statistics.py

@@ -15,7 +15,7 @@ class Statistics(Model):
     """群聊id"""
     plugin_name = fields.CharField(255)
     """插件名称"""
-    create_time = fields.DatetimeField(auto_now=True)
+    create_time = fields.DatetimeField(auto_now_add=True)
     """添加日期"""
     bot_id = fields.CharField(255, null=True)
     """Bot Id"""

zhenxun/models/user_console.py

@@ -151,9 +151,11 @@ async def add_gold(
             source: 来源
             platform: 平台.
         """
-        user = await cls._get_user_for_write(user_id=user_id, platform=platform)
-        user.gold += gold
-        await user.save(update_fields=["gold"])
+        await cls._get_user_for_write(user_id=user_id, platform=platform)
+        # 原子自增,避免并发 read-modify-write 丢币(A2);filter().update()
+        # 不触发基类 save() 的缓存失效,需手动失效。
+        await cls.filter(user_id=user_id).update(gold=F("gold") + gold)
+        await cls.invalidate_user_cache(user_id)
         await append_user_gold_log(
             user_id=user_id, gold=gold, handle=GoldHandle.GET, source=source
         )
@@ -182,8 +184,14 @@ async def reduce_gold(
         user = await cls._get_user_for_write(user_id=user_id, platform=platform)
         if user.gold < gold:
             raise InsufficientGold()
-        user.gold -= gold
-        await user.save(update_fields=["gold"])
+        # 原子扣减 + gold__gte 守卫,防并发超扣(A2);未命中说明余额已被
+        # 其他协程扣走,按金币不足处理。
+        updated = await cls.filter(user_id=user_id, gold__gte=gold).update(
+            gold=F("gold") - gold
+        )
+        if not updated:
+            raise InsufficientGold()
+        await cls.invalidate_user_cache(user_id)
         await append_user_gold_log(
             user_id=user_id, gold=gold, handle=handle, source=plugin_module
         )