feat(safety): add group access control with blacklist/whitelist support#15
Merged
Conversation
for more information, see https://pre-commit.ci
Contributor
Reviewer's Guide实现了可配置的安全访问控制，为 setu 和 fortune 功能分别提供黑名单/白名单模式；将安全列表与 AstrBot 的 WebUI 主配置同步；重构指令处理器以使用按功能划分的用户/群组列表和模式；为图片 URL 拉取添加超时；改进限流锁的处理方式；并为 fortune 渲染器引入 WOFF2 字体子集。 Setu 指令在按功能模式下进行访问控制的时序图sequenceDiagram
actor User
participant AstrBot
participant SetuPlugin
participant CommandHandler
participant SetuCore
participant SetuConfig
participant AccessControlManager
participant ConfigManager
participant provider
User->>AstrBot: send "/setu" message
AstrBot->>SetuPlugin: dispatch command event
SetuPlugin->>CommandHandler: handle_setu_command_internal(event)
CommandHandler->>SetuCore: is_group_blocked(event, feature=setu)
SetuCore->>SetuConfig: get setu_user_access_control_mode
SetuCore->>SetuConfig: get setu_group_access_control_mode
SetuConfig-->>SetuCore: user_mode, group_mode
SetuCore->>AccessControlManager: check_setu_access(user_id, group_id, user_mode, group_mode)
alt user_mode is blacklist
AccessControlManager->>ConfigManager: get_list(KEY_SETU_BLOCKED_USERS)
ConfigManager-->>AccessControlManager: blocked_users
AccessControlManager-->>SetuCore: (is_blocked, reason)
else user_mode is whitelist
AccessControlManager->>ConfigManager: get_list(KEY_SETU_WHITELIST_USERS)
ConfigManager-->>AccessControlManager: whitelist_users
AccessControlManager-->>SetuCore: (is_blocked, reason)
else user_mode is none
AccessControlManager-->>SetuCore: (False, "")
end
rect rgb(245,245,245)
Note over AccessControlManager,ConfigManager: Group level check (group_mode) is performed similarly using setu_blocked_groups and setu_whitelist_groups
end
SetuCore-->>CommandHandler: True or False
alt blocked
CommandHandler-->>AstrBot: stop, no images sent
AstrBot-->>User: (no response or error message)
else allowed
CommandHandler->>SetuCore: _get_provider()
SetuCore-->>CommandHandler: provider
CommandHandler->>provider: fetch_image_urls(...)
provider-->>CommandHandler: urls
CommandHandler->>SetuCore: send_images_by_url(...)
SetuCore-->>User: images
end
安全访问控制与配置管理的类图classDiagram
class ConfigManager {
-Path _data_dir
-Path _config_file
-dict _cache
-AstrBotConfig _astrbot_config
+SAFETY_LIST_KEYS
+SAFETY_MODE_KEYS
+LEGACY_MODE_TO_NEW
+initialize()
-_load_config()
-_save_config() bool
-_iter_main_config_candidates() list_Path
-_get_safety_value_from_file(key) Any
-_sync_to_astrbot_config() void
-_sync_from_astrbot_config() bool
+get(key, default) Any
-_get_access_control_mode_from_file(key, default) Any
+set(key, value) bool
+get_list(key) list_str
+add_to_list(key, item) bool
+remove_from_list(key, item) bool
+is_in_list(key, item) bool
}
class AccessControlManager {
-ConfigManager _cfg
+KEY_SETU_ACCESS_CONTROL_MODE
+KEY_SETU_BLOCKED_USERS
+KEY_SETU_WHITELIST_USERS
+KEY_SETU_BLOCKED_GROUPS
+KEY_SETU_WHITELIST_GROUPS
+KEY_FORTUNE_ACCESS_CONTROL_MODE
+KEY_FORTUNE_BLOCKED_USERS
+KEY_FORTUNE_WHITELIST_USERS
+KEY_FORTUNE_BLOCKED_GROUPS
+KEY_FORTUNE_WHITELIST_GROUPS
+add_setu_blocked_user(user_id) bool
+remove_setu_blocked_user(user_id) bool
+is_setu_user_blocked(user_id) bool
+get_setu_blocked_users() list_str
+add_setu_whitelist_user(user_id) bool
+remove_setu_whitelist_user(user_id) bool
+is_setu_user_whitelisted(user_id) bool
+get_setu_whitelist_users() list_str
+add_setu_blocked_group(group_id) bool
+remove_setu_blocked_group(group_id) bool
+is_setu_group_blocked(group_id) bool
+get_setu_blocked_groups() list_str
+add_setu_whitelist_group(group_id) bool
+remove_setu_whitelist_group(group_id) bool
+is_setu_group_whitelisted(group_id) bool
+get_setu_whitelist_groups() list_str
+add_fortune_blocked_user(user_id) bool
+remove_fortune_blocked_user(user_id) bool
+is_fortune_user_blocked(user_id) bool
+get_fortune_blocked_users() list_str
+add_fortune_whitelist_user(user_id) bool
+remove_fortune_whitelist_user(user_id) bool
+is_fortune_user_whitelisted(user_id) bool
+get_fortune_whitelist_users() list_str
+add_fortune_blocked_group(group_id) bool
+remove_fortune_blocked_group(group_id) bool
+is_fortune_group_blocked(group_id) bool
+get_fortune_blocked_groups() list_str
+add_fortune_whitelist_group(group_id) bool
+remove_fortune_whitelist_group(group_id) bool
+is_fortune_group_whitelisted(group_id) bool
+get_fortune_whitelist_groups() list_str
+check_setu_access(user_id, group_id, user_access_control_mode, group_access_control_mode) tuple_bool_str
+check_fortune_access(user_id, group_id, user_access_control_mode, group_access_control_mode) tuple_bool_str
+get_all_lists() dict_str_list_str
}
class SafetyConfigMixin {
+_read
+_normalize_access_mode(value, default) str
+setu_access_control_mode str
+fortune_access_control_mode str
+setu_user_access_control_mode str
+setu_group_access_control_mode str
+fortune_user_access_control_mode str
+fortune_group_access_control_mode str
+cache_enabled bool
}
class SetuConfig {
}
class SetuCore {
-ConfigManager _config_manager
-AccessControlManager _access_control
-SetuConfig _config
+is_group_blocked(event, feature) bool
+_get_provider()
+_get_fortune_provider()
}
class CommandHandler {
-SetuCore _core
-SetuConfig _config
+handle_setu_block_user(event, args)
+handle_setu_unblock_user(event, args)
+handle_setu_trust_user(event, args)
+handle_setu_untrust_user(event, args)
+handle_fortune_block_user(event, args)
+handle_fortune_unblock_user(event, args)
+handle_fortune_trust_user(event, args)
+handle_fortune_untrust_user(event, args)
+handle_enable_setu_group(event, args)
+handle_disable_setu_group(event, args)
+handle_enable_fortune_group(event, args)
+handle_disable_fortune_group(event, args)
-_extract_target_id(event, args) str
}
class AstrBotConfig {
+save_config()
+__getitem__(key)
+get(key, default)
}
SetuConfig ..|> SafetyConfigMixin
SetuCore --> ConfigManager : uses
SetuCore --> AccessControlManager : uses
SetuCore --> SetuConfig : reads_access_modes
CommandHandler --> SetuCore : calls_is_group_blocked
AccessControlManager --> ConfigManager : reads_writes_safety_lists
ConfigManager --> AstrBotConfig : syncs_safety_section
文件级改动
针对关联 issue 的评估
Tips and commandsInteracting with Sourcery
自定义你的体验访问你的 dashboard 以：
获取帮助Original review guide in EnglishReviewer's GuideImplements configurable safety access control with separate blacklist/whitelist modes for setu and fortune features, synchronizes safety lists with AstrBot’s main config for WebUI, refactors command handlers to use feature-specific user/group lists and modes, adds timeouts to image URL fetches, improves rate limiter lock handling, and introduces font subsetting to WOFF2 for the fortune renderer. Sequence diagram for Setu command access control with feature-specific modessequenceDiagram
actor User
participant AstrBot
participant SetuPlugin
participant CommandHandler
participant SetuCore
participant SetuConfig
participant AccessControlManager
participant ConfigManager
participant provider
User->>AstrBot: send "/setu" message
AstrBot->>SetuPlugin: dispatch command event
SetuPlugin->>CommandHandler: handle_setu_command_internal(event)
CommandHandler->>SetuCore: is_group_blocked(event, feature=setu)
SetuCore->>SetuConfig: get setu_user_access_control_mode
SetuCore->>SetuConfig: get setu_group_access_control_mode
SetuConfig-->>SetuCore: user_mode, group_mode
SetuCore->>AccessControlManager: check_setu_access(user_id, group_id, user_mode, group_mode)
alt user_mode is blacklist
AccessControlManager->>ConfigManager: get_list(KEY_SETU_BLOCKED_USERS)
ConfigManager-->>AccessControlManager: blocked_users
AccessControlManager-->>SetuCore: (is_blocked, reason)
else user_mode is whitelist
AccessControlManager->>ConfigManager: get_list(KEY_SETU_WHITELIST_USERS)
ConfigManager-->>AccessControlManager: whitelist_users
AccessControlManager-->>SetuCore: (is_blocked, reason)
else user_mode is none
AccessControlManager-->>SetuCore: (False, "")
end
rect rgb(245,245,245)
Note over AccessControlManager,ConfigManager: Group level check (group_mode) is performed similarly using setu_blocked_groups and setu_whitelist_groups
end
SetuCore-->>CommandHandler: True or False
alt blocked
CommandHandler-->>AstrBot: stop, no images sent
AstrBot-->>User: (no response or error message)
else allowed
CommandHandler->>SetuCore: _get_provider()
SetuCore-->>CommandHandler: provider
CommandHandler->>provider: fetch_image_urls(...)
provider-->>CommandHandler: urls
CommandHandler->>SetuCore: send_images_by_url(...)
SetuCore-->>User: images
end
Class diagram for safety access control and config managementclassDiagram
class ConfigManager {
-Path _data_dir
-Path _config_file
-dict _cache
-AstrBotConfig _astrbot_config
+SAFETY_LIST_KEYS
+SAFETY_MODE_KEYS
+LEGACY_MODE_TO_NEW
+initialize()
-_load_config()
-_save_config() bool
-_iter_main_config_candidates() list_Path
-_get_safety_value_from_file(key) Any
-_sync_to_astrbot_config() void
-_sync_from_astrbot_config() bool
+get(key, default) Any
-_get_access_control_mode_from_file(key, default) Any
+set(key, value) bool
+get_list(key) list_str
+add_to_list(key, item) bool
+remove_from_list(key, item) bool
+is_in_list(key, item) bool
}
class AccessControlManager {
-ConfigManager _cfg
+KEY_SETU_ACCESS_CONTROL_MODE
+KEY_SETU_BLOCKED_USERS
+KEY_SETU_WHITELIST_USERS
+KEY_SETU_BLOCKED_GROUPS
+KEY_SETU_WHITELIST_GROUPS
+KEY_FORTUNE_ACCESS_CONTROL_MODE
+KEY_FORTUNE_BLOCKED_USERS
+KEY_FORTUNE_WHITELIST_USERS
+KEY_FORTUNE_BLOCKED_GROUPS
+KEY_FORTUNE_WHITELIST_GROUPS
+add_setu_blocked_user(user_id) bool
+remove_setu_blocked_user(user_id) bool
+is_setu_user_blocked(user_id) bool
+get_setu_blocked_users() list_str
+add_setu_whitelist_user(user_id) bool
+remove_setu_whitelist_user(user_id) bool
+is_setu_user_whitelisted(user_id) bool
+get_setu_whitelist_users() list_str
+add_setu_blocked_group(group_id) bool
+remove_setu_blocked_group(group_id) bool
+is_setu_group_blocked(group_id) bool
+get_setu_blocked_groups() list_str
+add_setu_whitelist_group(group_id) bool
+remove_setu_whitelist_group(group_id) bool
+is_setu_group_whitelisted(group_id) bool
+get_setu_whitelist_groups() list_str
+add_fortune_blocked_user(user_id) bool
+remove_fortune_blocked_user(user_id) bool
+is_fortune_user_blocked(user_id) bool
+get_fortune_blocked_users() list_str
+add_fortune_whitelist_user(user_id) bool
+remove_fortune_whitelist_user(user_id) bool
+is_fortune_user_whitelisted(user_id) bool
+get_fortune_whitelist_users() list_str
+add_fortune_blocked_group(group_id) bool
+remove_fortune_blocked_group(group_id) bool
+is_fortune_group_blocked(group_id) bool
+get_fortune_blocked_groups() list_str
+add_fortune_whitelist_group(group_id) bool
+remove_fortune_whitelist_group(group_id) bool
+is_fortune_group_whitelisted(group_id) bool
+get_fortune_whitelist_groups() list_str
+check_setu_access(user_id, group_id, user_access_control_mode, group_access_control_mode) tuple_bool_str
+check_fortune_access(user_id, group_id, user_access_control_mode, group_access_control_mode) tuple_bool_str
+get_all_lists() dict_str_list_str
}
class SafetyConfigMixin {
+_read
+_normalize_access_mode(value, default) str
+setu_access_control_mode str
+fortune_access_control_mode str
+setu_user_access_control_mode str
+setu_group_access_control_mode str
+fortune_user_access_control_mode str
+fortune_group_access_control_mode str
+cache_enabled bool
}
class SetuConfig {
}
class SetuCore {
-ConfigManager _config_manager
-AccessControlManager _access_control
-SetuConfig _config
+is_group_blocked(event, feature) bool
+_get_provider()
+_get_fortune_provider()
}
class CommandHandler {
-SetuCore _core
-SetuConfig _config
+handle_setu_block_user(event, args)
+handle_setu_unblock_user(event, args)
+handle_setu_trust_user(event, args)
+handle_setu_untrust_user(event, args)
+handle_fortune_block_user(event, args)
+handle_fortune_unblock_user(event, args)
+handle_fortune_trust_user(event, args)
+handle_fortune_untrust_user(event, args)
+handle_enable_setu_group(event, args)
+handle_disable_setu_group(event, args)
+handle_enable_fortune_group(event, args)
+handle_disable_fortune_group(event, args)
-_extract_target_id(event, args) str
}
class AstrBotConfig {
+save_config()
+__getitem__(key)
+get(key, default)
}
SetuConfig ..|> SafetyConfigMixin
SetuCore --> ConfigManager : uses
SetuCore --> AccessControlManager : uses
SetuCore --> SetuConfig : reads_access_modes
CommandHandler --> SetuCore : calls_is_group_blocked
AccessControlManager --> ConfigManager : reads_writes_safety_lists
ConfigManager --> AstrBotConfig : syncs_safety_section
File-Level Changes
Assessment against linked issues
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - 我发现了两个问题，并给出了一些整体性的反馈：
- 在
access_control_mode中，非法模式静默回退到blacklist可能会掩盖配置错误；建议记录一条警告日志或抛出显式错误，让运维人员能看到错误配置。 - README 中的 JSON 配置示例使用了全角/中文引号（
“/”），这会导致示例并非合法 JSON；请改用标准 ASCII 引号，方便用户直接复制粘贴使用。
给 AI Agent 的提示
Please address the comments from this code review:
## Overall Comments
- 在 `access_control_mode` 中，非法模式静默回退到 `blacklist` 可能会掩盖配置错误；建议记录一条警告日志或抛出显式错误，让运维人员能看到错误配置。
- README 中的 JSON 配置示例使用了全角/中文引号（`“`/`”`），这会导致示例并非合法 JSON；请改用标准 ASCII 引号，方便用户直接复制粘贴使用。
## Individual Comments
### Comment 1
<location path="config/safety.py" line_range="28-32" />
<code_context>
+ 返回:
+ none=不启用黑白名单，blacklist=仅黑名单模式，whitelist=仅白名单模式
+ """
+ mode = self._read(
+ ("safety", "access_control_mode"), "access_control_mode", default="blacklist"
+ )
+ valid_modes = ("none", "blacklist", "whitelist")
+ if mode not in valid_modes:
+ return "blacklist"
+ return mode
</code_context>
<issue_to_address>
**suggestion (bug_risk):** 在回退到 `"blacklist"` 之前，考虑对 `access_control_mode` 做更健壮的归一化和校验。
目前，只要不是 `("none", "blacklist", "whitelist")` 中的值（包括大小写不同或多余空格等轻微差异），都会被静默地当作 `"blacklist"` 处理，这可能会掩盖配置错误。建议先对值进行归一化（例如 `mode = str(mode).strip().lower()`），并在值非法时记录日志或抛出错误，而不是默默改变行为。使用更严格的返回类型，例如 `Literal["none", "blacklist", "whitelist"]` 或一个小型 `Enum`，也能让调用端更安全。
Suggested implementation:
```python
@property
def access_control_mode(self) -> "Literal['none', 'blacklist', 'whitelist']":
"""访问控制模式。
返回:
none=不启用黑白名单，blacklist=仅黑名单模式，whitelist=仅白名单模式
"""
raw_mode = self._read(
("safety", "access_control_mode"), "access_control_mode", default="blacklist"
)
mode = str(raw_mode).strip().lower()
valid_modes: tuple[str, ...] = ("none", "blacklist", "whitelist")
if mode not in valid_modes:
# 配置错误时记录警告并回退到安全的默认值
import logging
logging.warning(
"Invalid access_control_mode %r (normalized: %r), falling back to 'blacklist'",
raw_mode,
mode,
)
return "blacklist"
# 这里 mode 一定是 valid_modes 之一
return mode # type: ignore[return-value]
```
1. 为了让类型注解工作得更好，建议在文件顶部添加：
```python
from typing import Literal
```
并移除返回类型上的字符串引号（如果你启用了 `from __future__ import annotations` 则可以保留字符串形式），例如：
```python
def access_control_mode(self) -> Literal["none", "blacklist", "whitelist"]:
```
2. 如果你不希望在方法体中局部 `import logging`，可改为：
- 在文件顶部添加 `import logging`
- 删除方法中的 `import logging` 行。
</issue_to_address>
### Comment 2
<location path="config/safety.py" line_range="75-79" />
<code_context>
"""
if not group_id:
return False
- return str(group_id) in self.blocked_groups
+
+ mode = self.access_control_mode
+ gid = str(group_id)
+
+ if mode == "none":
</code_context>
<issue_to_address>
**suggestion (bug_risk):** 明确对假值 `group_id` 的处理逻辑，避免无意中跳过访问控制。
由于这里使用的是 `if not group_id`，空字符串或其他假值（例如 `""`、类似 `0` 的值）会直接跳过访问控制检查并被视为允许。如果只有 `None` 才表示“没有群组”，建议使用 `if group_id is None:`（或在边界位置对 `group_id` 做校验/归一化），以避免在屏蔽逻辑中被悄悄绕过。
```suggestion
"""
# 仅当 group_id 为 None 时视为“无群组”，直接返回不屏蔽
# 其他 falsy 值（例如 "", 0 等）仍会参与访问控制检查
if group_id is None:
return False
mode = self.access_control_mode
```
</issue_to_address>帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据你的反馈改进 review 质量。
Original comment in English
Hey - I've found 2 issues, and left some high level feedback:
- In
access_control_mode, invalid modes silently falling back toblacklistmay hide misconfigurations; consider logging a warning or surfacing an explicit error to make bad values visible to operators. - The JSON config example in the README uses full-width/Chinese quotation marks (
“/”), which makes the snippet invalid JSON; please switch to standard ASCII quotes so users can copy-paste it directly.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `access_control_mode`, invalid modes silently falling back to `blacklist` may hide misconfigurations; consider logging a warning or surfacing an explicit error to make bad values visible to operators.
- The JSON config example in the README uses full-width/Chinese quotation marks (`“`/`”`), which makes the snippet invalid JSON; please switch to standard ASCII quotes so users can copy-paste it directly.
## Individual Comments
### Comment 1
<location path="config/safety.py" line_range="28-32" />
<code_context>
+ 返回:
+ none=不启用黑白名单，blacklist=仅黑名单模式，whitelist=仅白名单模式
+ """
+ mode = self._read(
+ ("safety", "access_control_mode"), "access_control_mode", default="blacklist"
+ )
+ valid_modes = ("none", "blacklist", "whitelist")
+ if mode not in valid_modes:
+ return "blacklist"
+ return mode
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Consider normalizing and validating `access_control_mode` more robustly before falling back to `"blacklist"`.
Currently, any value outside `("none", "blacklist", "whitelist")` (including trivial variants like different casing or extra spaces) is silently treated as `"blacklist"`, which can hide config mistakes. Consider normalizing the value first (e.g. `mode = str(mode).strip().lower()`) and logging or raising on invalid values instead of changing behavior silently. A stricter return type such as `Literal["none", "blacklist", "whitelist"]` or a small `Enum` would also make call sites safer.
Suggested implementation:
```python
@property
def access_control_mode(self) -> "Literal['none', 'blacklist', 'whitelist']":
"""访问控制模式。
返回:
none=不启用黑白名单，blacklist=仅黑名单模式，whitelist=仅白名单模式
"""
raw_mode = self._read(
("safety", "access_control_mode"), "access_control_mode", default="blacklist"
)
mode = str(raw_mode).strip().lower()
valid_modes: tuple[str, ...] = ("none", "blacklist", "whitelist")
if mode not in valid_modes:
# 配置错误时记录警告并回退到安全的默认值
import logging
logging.warning(
"Invalid access_control_mode %r (normalized: %r), falling back to 'blacklist'",
raw_mode,
mode,
)
return "blacklist"
# 这里 mode 一定是 valid_modes 之一
return mode # type: ignore[return-value]
```
1. 为了让类型注解工作得更好，建议在文件顶部添加：
```python
from typing import Literal
```
并移除返回类型上的字符串引号（如果你启用了 `from __future__ import annotations` 则可以保留字符串形式），例如：
```python
def access_control_mode(self) -> Literal["none", "blacklist", "whitelist"]:
```
2. 如果你不希望在方法体中局部 `import logging`，可改为：
- 在文件顶部添加 `import logging`
- 删除方法中的 `import logging` 行。
</issue_to_address>
### Comment 2
<location path="config/safety.py" line_range="75-79" />
<code_context>
"""
if not group_id:
return False
- return str(group_id) in self.blocked_groups
+
+ mode = self.access_control_mode
+ gid = str(group_id)
+
+ if mode == "none":
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Clarify handling of falsy `group_id` values to avoid unintentionally skipping access control.
Because this uses `if not group_id`, empty strings or other falsy values (e.g. `""`, `0`-like) will skip the access-control check and be treated as allowed. If `None` is the only sentinel meaning “no group”, use `if group_id is None:` (or validate/normalize `group_id` at the boundary) to avoid silently bypassing blocking logic.
```suggestion
"""
# 仅当 group_id 为 None 时视为“无群组”，直接返回不屏蔽
# 其他 falsy 值（例如 "", 0 等）仍会参与访问控制检查
if group_id is None:
return False
mode = self.access_control_mode
```
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
for more information, see https://pre-commit.ci
# Conflicts: # config/safety.py # fortune/handlers.py # handlers/command_handlers.py # services/config_manager.py
for more information, see https://pre-commit.ci
Owner
Author
|
@sourcery-ai review |
Contributor
There was a problem hiding this comment.
Hey - 我发现了 3 个问题，并给出了一些整体性的反馈：
- 在
RateLimiter.acquire中，新的lock变量只在键不在_locks中时才会初始化，但在离开_global_lock之后会无条件地await它，这会在锁已存在时抛出NameError。建议在with块之后统一使用lock = self._locks[key]，让两条分支共享同一个变量。 - 在
ConfigManager.get中，访问控制逻辑现在被拆分到了_get_safety_value_from_file和_get_access_control_mode_from_file之间，导致对 mode key 的逻辑重复。你可以通过为SAFETY_MODE_KEYS保留一条统一的逻辑路径来简化（例如：让_get_safety_value_from_file既处理 list 又处理 mode，并删除额外的 mode 专用查询）。
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `RateLimiter.acquire`, the new `lock` variable is only initialized when the key is absent in `_locks`, but it is awaited unconditionally after leaving `_global_lock`, which will raise a `NameError` when a lock already exists; consider always using `lock = self._locks[key]` after the `with` block so both paths share the same variable.
- The access control handling is now split between `_get_safety_value_from_file` and `_get_access_control_mode_from_file` in `ConfigManager.get`, leading to duplicated logic for mode keys; you could simplify by having a single code path for SAFETY_MODE_KEYS (e.g., reuse `_get_safety_value_from_file` for both lists and modes, and remove the extra mode-specific lookup).
## Individual Comments
### Comment 1
<location path="core/__init__.py" line_range="261-270" />
<code_context>
return False
- def is_group_blocked(self, event: AstrMessageEvent) -> bool:
+ def is_group_blocked(
+ self, event: AstrMessageEvent, feature: str | None = None
+ ) -> bool:
</code_context>
<issue_to_address>
**🚨 issue (security):** 当 `feature` 为 None 时，目前的逻辑会完全跳过访问控制，这可能在无提示的情况下改变既有行为。
之前，`is_group_blocked` 始终会执行 `check_global_access`；而现在的 `else` 分支不会做任何检查，只是记录一条日志 “skipping access control”。任何现有的、未传入 `feature` 的调用方现在都会绕过访问控制。
我建议：
- 要么把 `feature` 设为必填参数，强制调用方明确选择模式；
- 要么让 `else` 分支保留旧行为（例如调用全局/默认访问控制模式），而不是直接跳过检查。
</issue_to_address>
### Comment 2
<location path="services/config_manager.py" line_range="57-66" />
<code_context>
+ def get(self, key: str, default: Any = None) -> Any:
</code_context>
<issue_to_address>
**suggestion (performance):** 在高消息量场景下，`get()` 对安全相关 key 的反复读文件会比较昂贵。
对于 `SAFETY_LIST_KEYS`/`SAFETY_MODE_KEYS`，`get()` 会调用 `_get_safety_value_from_file`，该函数会遍历 `_iter_main_config_candidates()`，并在每次访问时读取/解析 JSON。由于这是按消息调用的，这可能带来可观的磁盘 I/O 和解析开销。
建议将解析后的主配置（或仅其 `safety` 段）缓存在内存中，并配合失效机制（例如基于时间戳/手动重载）；或者让所有磁盘读取都通过 `_sync_from_astrbot_config()` 进行，而 `get()` 从该缓存中提供数据，以在保持原有行为的同时，避免在热点路径中频繁访问文件系统。
Suggested implementation:
```python
def get(self, key: str, default: Any = None) -> Any:
```
要完整实现缓存并将其接入 `get()`，请在 `services/config_manager.py` 的其他位置做如下修改：
1. **在配置管理类中添加（延迟初始化的）缓存属性**
在你的配置管理类中（定义了 `get()`, `_iter_main_config_candidates()`, `_get_safety_value_from_file()` 等方法的那个类），添加一些延迟初始化的属性，用于存储缓存的主配置及其最后修改时间。这样如果你不想修改 `__init__`，也可以通过延迟初始化来实现：
```python
# At top of the class (or in __init__), define:
self._main_config_cache: dict[str, Any] | None = None
self._main_config_cache_mtime: float | None = None
self._main_config_cache_path: Path | None = None
self._main_config_cache_lock = threading.Lock()
```
如果不能修改 `__init__`，可以在后面要用到的 helper 中使用 `setattr` / `getattr` 做延迟创建（见下一步）。
2. **添加一个带基于 mtime 失效机制的主配置加载 helper**
在 `_iter_main_config_candidates` 附近实现类似如下的方法：
```python
def _load_main_config_with_cache(self) -> dict[str, Any] | None:
"""
加载 AstrBot 主配置文件，并使用内存缓存避免频繁读取磁盘。
返回完整配置字典；如果找不到主配置文件或解析失败则返回 None。
"""
# Lazy init to avoid touching __init__
if not hasattr(self, "_main_config_cache_lock"):
import threading
self._main_config_cache_lock = threading.Lock()
self._main_config_cache = None
self._main_config_cache_mtime = None
self._main_config_cache_path = None
candidates = list(self._iter_main_config_candidates())
if not candidates:
return None
# 选择第一个存在的配置文件作为主配置
config_path: Path | None = None
for candidate in candidates:
if candidate.is_file():
config_path = candidate
break
if config_path is None:
return None
try:
mtime = config_path.stat().st_mtime
except OSError:
return None
with self._main_config_cache_lock:
# 如果缓存命中且未过期，直接返回
if (
getattr(self, "_main_config_cache_path", None) == config_path
and getattr(self, "_main_config_cache_mtime", None) == mtime
and getattr(self, "_main_config_cache", None) is not None
):
return self._main_config_cache # type: ignore[return-value]
# 否则从磁盘重新加载并更新缓存
try:
with config_path.open("r", encoding="utf-8") as f:
data = json.load(f)
except (OSError, json.JSONDecodeError) as e:
logger.warning("Failed to load main config %s: %s", config_path, e)
return None
self._main_config_cache = data
self._main_config_cache_mtime = mtime
self._main_config_cache_path = config_path
return data
```
这样可以集中管理文件系统访问，并保证对同一文件的重复读取都复用内存中的对象，直到磁盘上的文件发生变化。
3. **为 safety 段添加 helper，并让 `_get_safety_value_from_file` 使用它**
实现一个面向 safety 段的辅助方法：
```python
def _get_safety_section(self) -> dict[str, Any] | None:
"""从主配置中获取 safety 配置段，带缓存。"""
main_cfg = self._load_main_config_with_cache()
if not main_cfg:
return None
safety = main_cfg.get("safety")
if not isinstance(safety, dict):
return None
return safety
```
然后重构 `_get_safety_value_from_file`（如果还不存在该方法，则新增一个）——让它 **只** 从这个带缓存的 safety 段读数据，而不是每次调用都遍历 `_iter_main_config_candidates()` 并重新打开/解析 JSON。例如：
```python
def _get_safety_value_from_file(self, key: str, default: Any = None) -> Any:
"""
从 AstrBot 主配置的 safety 段获取指定 key 的值。
使用内存缓存避免在高频调用时频繁访问磁盘。
"""
safety = self._get_safety_section()
if safety is None:
return default
return safety.get(key, default)
```
4. **更新 `get()`，让安全相关 key 走缓存后的 safety 配置**
在 `get()` 中，调整现在调用 `_get_safety_value_from_file` 并触发重读文件的那条分支。大致上你可能有类似代码：
```python
def get(self, key: str, default: Any = None) -> Any:
# ...
if key in self.SAFETY_LIST_KEYS or key in self.SAFETY_MODE_KEYS:
value = self._get_safety_value_from_file(key, default)
# possibly merge with self._cache overrides
return value
# ...
```
确保：
- 所有 safety key 的解析都通过 `_get_safety_value_from_file`，而后者使用的是带缓存的配置；
- 如果你允许在 `self._cache` 中做每插件覆盖，那么可以在 `get()` 中先查 `self._cache`，再回退到 `_get_safety_value_from_file(key, default)`。
5. **可选：在 `_sync_to_astrbot_config()` 或其他写入逻辑中失效缓存**
如果这个类同时也负责写 AstrBot 主配置（例如通过 `_sync_to_astrbot_config()` 或其他写入方法），那么在写入之后立即失效缓存：
```python
with self._main_config_cache_lock:
self._main_config_cache = None
self._main_config_cache_mtime = None
self._main_config_cache_path = None
```
这样可以确保后续的 `get()` 调用即使在文件 mtime 尚未变化（或存在多进程写入）的情况下，也能看到最新数据。
这些修改会让所有针对安全 key 的文件读取都走主配置的缓存版本，从而在保持行为不变的前提下，消除在每条消息的热点路径上反复的磁盘 I/O 和 JSON 解析开销。
</issue_to_address>
### Comment 3
<location path="handlers/command_handlers.py" line_range="106-113" />
<code_context>
- tags=tags,
- r18=is_r18,
- exclude_ai=self._config.exclude_ai,
+ img_urls = await asyncio.wait_for(
+ provider.fetch_image_urls(
+ num=num,
+ tags=tags,
+ r18=is_r18,
+ exclude_ai=self._config.exclude_ai,
+ ),
+ timeout=60.0,
)
async for result in self._core.send_images_by_url(
</code_context>
<issue_to_address>
**suggestion (bug_risk):** 将 `fetch_image_urls` 包在 `asyncio.wait_for` 中而不处理 `TimeoutError`，可能会导致超时异常未被调用方正确捕获。
当达到 60 秒的时间限制时，`asyncio.wait_for` 会抛出 `asyncio.TimeoutError`，目前该异常会向上传播，被视为一个通用失败。请在这次调用周围捕获 `asyncio.TimeoutError`，将其转换为用户可见的超时提示信息，并可选地记录日志，以便清晰地区分超时与其他错误。
Suggested implementation:
```python
try:
img_urls = await asyncio.wait_for(
provider.fetch_image_urls(
num=num,
tags=tags,
r18=is_r18,
exclude_ai=self._config.exclude_ai,
),
timeout=60.0,
)
except asyncio.TimeoutError:
logger.warning(
"Timed out fetching image URLs from provider=%r after %ss (tags=%r, r18=%r, exclude_ai=%r)",
provider,
60.0,
tags,
is_r18,
self._config.exclude_ai,
)
await event.reply("图片获取超时，请稍后重试。")
return
async for result in self._core.send_images_by_url(
event, img_urls, is_r18, tags
```
为了保证代码能够编译并符合现有约定，你可能还需要：
1. 确认在 `handlers/command_handlers.py` 中已经导入了 `asyncio`（如果尚未导入）：
- `import asyncio`
2. 确认在合适的位置定义并使用模块级 logger，例如：
- 在文件顶部添加 `import logging`；
- 在顶层定义附近添加 `logger = logging.getLogger(__name__)`。
3. 如果在你的框架中 `event.reply` 不是正确的发送消息方式，请将 `await event.reply("图片获取超时，请稍后重试。")` 替换为该 handler 中其他地方所使用的、面向用户的消息发送 API。
</issue_to_address>帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据你的反馈持续改进评审质量。
Original comment in English
Hey - I've found 3 issues, and left some high level feedback:
- In
RateLimiter.acquire, the newlockvariable is only initialized when the key is absent in_locks, but it is awaited unconditionally after leaving_global_lock, which will raise aNameErrorwhen a lock already exists; consider always usinglock = self._locks[key]after thewithblock so both paths share the same variable. - The access control handling is now split between
_get_safety_value_from_fileand_get_access_control_mode_from_fileinConfigManager.get, leading to duplicated logic for mode keys; you could simplify by having a single code path for SAFETY_MODE_KEYS (e.g., reuse_get_safety_value_from_filefor both lists and modes, and remove the extra mode-specific lookup).
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `RateLimiter.acquire`, the new `lock` variable is only initialized when the key is absent in `_locks`, but it is awaited unconditionally after leaving `_global_lock`, which will raise a `NameError` when a lock already exists; consider always using `lock = self._locks[key]` after the `with` block so both paths share the same variable.
- The access control handling is now split between `_get_safety_value_from_file` and `_get_access_control_mode_from_file` in `ConfigManager.get`, leading to duplicated logic for mode keys; you could simplify by having a single code path for SAFETY_MODE_KEYS (e.g., reuse `_get_safety_value_from_file` for both lists and modes, and remove the extra mode-specific lookup).
## Individual Comments
### Comment 1
<location path="core/__init__.py" line_range="261-270" />
<code_context>
return False
- def is_group_blocked(self, event: AstrMessageEvent) -> bool:
+ def is_group_blocked(
+ self, event: AstrMessageEvent, feature: str | None = None
+ ) -> bool:
</code_context>
<issue_to_address>
**🚨 issue (security):** When `feature` is None, access control is now completely skipped, which may silently change previous behavior.
Previously `is_group_blocked` always enforced `check_global_access`; now the `else` branch does no checks and only logs "skipping access control". Any existing callers that don’t pass `feature` will now bypass access control.
I’d recommend either:
- Making `feature` required so callers must choose the correct mode, or
- Having the `else` branch preserve the old behavior (e.g., call a global/default access mode) rather than skipping checks.
</issue_to_address>
### Comment 2
<location path="services/config_manager.py" line_range="57-66" />
<code_context>
+ def get(self, key: str, default: Any = None) -> Any:
</code_context>
<issue_to_address>
**suggestion (performance):** Repeated file reads for safety keys in `get()` can be expensive under high message volume.
For `SAFETY_LIST_KEYS`/`SAFETY_MODE_KEYS`, `get()` calls `_get_safety_value_from_file`, which walks `_iter_main_config_candidates()` and reads/parses JSON on every access. Since this runs per message, it can introduce substantial disk I/O and parsing overhead.
Consider caching the parsed main config (or just the `safety` section) in memory with an invalidation mechanism (e.g., timestamp/manual reload), or routing all disk reads through `_sync_from_astrbot_config()` and having `get()` serve from that cache to keep behavior while avoiding repeated filesystem work in the hot path.
Suggested implementation:
```python
def get(self, key: str, default: Any = None) -> Any:
```
To fully implement the caching and wire it into `get()`, make the following changes elsewhere in `services/config_manager.py`:
1. **Add cache attributes (lazy-initialized) to the config manager class**
Inside your config manager class (the one that defines `get()`, `_iter_main_config_candidates()`, `_get_safety_value_from_file()`, etc.), add lazy-initialized attributes to store the cached main config and its last modification time. This avoids needing to edit `__init__` if you prefer:
```python
# At top of the class (or in __init__), define:
self._main_config_cache: dict[str, Any] | None = None
self._main_config_cache_mtime: float | None = None
self._main_config_cache_path: Path | None = None
self._main_config_cache_lock = threading.Lock()
```
If you cannot modify `__init__`, you can lazy-create them in the helper (see next step) using `setattr` and `getattr`.
2. **Add a helper to get the parsed main config with mtime-based invalidation**
Near `_iter_main_config_candidates`, implement a method like:
```python
def _load_main_config_with_cache(self) -> dict[str, Any] | None:
"""
加载 AstrBot 主配置文件，并使用内存缓存避免频繁读取磁盘。
返回完整配置字典；如果找不到主配置文件或解析失败则返回 None。
"""
# Lazy init to avoid touching __init__
if not hasattr(self, "_main_config_cache_lock"):
import threading
self._main_config_cache_lock = threading.Lock()
self._main_config_cache = None
self._main_config_cache_mtime = None
self._main_config_cache_path = None
candidates = list(self._iter_main_config_candidates())
if not candidates:
return None
# 选择第一个存在的配置文件作为主配置
config_path: Path | None = None
for candidate in candidates:
if candidate.is_file():
config_path = candidate
break
if config_path is None:
return None
try:
mtime = config_path.stat().st_mtime
except OSError:
return None
with self._main_config_cache_lock:
# 如果缓存命中且未过期，直接返回
if (
getattr(self, "_main_config_cache_path", None) == config_path
and getattr(self, "_main_config_cache_mtime", None) == mtime
and getattr(self, "_main_config_cache", None) is not None
):
return self._main_config_cache # type: ignore[return-value]
# 否则从磁盘重新加载并更新缓存
try:
with config_path.open("r", encoding="utf-8") as f:
data = json.load(f)
except (OSError, json.JSONDecodeError) as e:
logger.warning("Failed to load main config %s: %s", config_path, e)
return None
self._main_config_cache = data
self._main_config_cache_mtime = mtime
self._main_config_cache_path = config_path
return data
```
This centralizes filesystem access and ensures repeated reads of the same file re-use the in-memory object until the file changes on disk.
3. **Add a helper for the safety section and wire `_get_safety_value_from_file` to it**
Implement a safety-specific helper:
```python
def _get_safety_section(self) -> dict[str, Any] | None:
"""从主配置中获取 safety 配置段，带缓存。"""
main_cfg = self._load_main_config_with_cache()
if not main_cfg:
return None
safety = main_cfg.get("safety")
if not isinstance(safety, dict):
return None
return safety
```
Then refactor `_get_safety_value_from_file` (or introduce it if it doesn't exist yet) so that it **only** reads from this cached safety section instead of walking `_iter_main_config_candidates()` and re-opening/parsing JSON each time. For example:
```python
def _get_safety_value_from_file(self, key: str, default: Any = None) -> Any:
"""
从 AstrBot 主配置的 safety 段获取指定 key 的值。
使用内存缓存避免在高频调用时频繁访问磁盘。
"""
safety = self._get_safety_section()
if safety is None:
return default
return safety.get(key, default)
```
4. **Update `get()` to use the cached safety config for safety keys**
Inside `get()`, adjust the branch that currently calls `_get_safety_value_from_file` and triggers the heavy path. Conceptually, you likely have something like:
```python
def get(self, key: str, default: Any = None) -> Any:
# ...
if key in self.SAFETY_LIST_KEYS or key in self.SAFETY_MODE_KEYS:
value = self._get_safety_value_from_file(key, default)
# possibly merge with self._cache overrides
return value
# ...
```
Ensure that:
- All safety-key resolution goes through `_get_safety_value_from_file`, which now uses the cached config.
- If you allow per-plugin overrides in `self._cache`, you may want `get()` to check `self._cache` first, then fall back to `_get_safety_value_from_file(key, default)`.
5. **Optionally, invalidate the cache when `_sync_to_astrbot_config()` or related writers run**
If this class is also responsible for writing the AstrBot main config (e.g. in `_sync_to_astrbot_config()` or other methods), invalidate the cache right after writing:
```python
with self._main_config_cache_lock:
self._main_config_cache = None
self._main_config_cache_mtime = None
self._main_config_cache_path = None
```
This makes sure subsequent `get()` calls see fresh data even if the file’s mtime hasn't changed yet (or if multiple processes may be writing).
These changes will route all filesystem reads for safety keys through the cached main config, keeping behavior intact while eliminating repeated disk I/O and JSON parsing from the per-message hot path.
</issue_to_address>
### Comment 3
<location path="handlers/command_handlers.py" line_range="106-113" />
<code_context>
- tags=tags,
- r18=is_r18,
- exclude_ai=self._config.exclude_ai,
+ img_urls = await asyncio.wait_for(
+ provider.fetch_image_urls(
+ num=num,
+ tags=tags,
+ r18=is_r18,
+ exclude_ai=self._config.exclude_ai,
+ ),
+ timeout=60.0,
)
async for result in self._core.send_images_by_url(
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Wrapping `fetch_image_urls` in `asyncio.wait_for` without handling `TimeoutError` may surface as unhandled exceptions to callers.
`asyncio.wait_for` will raise `asyncio.TimeoutError` when the 60s limit is hit, which will currently propagate as a generic failure. Please catch `asyncio.TimeoutError` around this call, translate it into a user-facing timeout message, and optionally log it so timeouts are clearly distinguished from other errors.
Suggested implementation:
```python
try:
img_urls = await asyncio.wait_for(
provider.fetch_image_urls(
num=num,
tags=tags,
r18=is_r18,
exclude_ai=self._config.exclude_ai,
),
timeout=60.0,
)
except asyncio.TimeoutError:
logger.warning(
"Timed out fetching image URLs from provider=%r after %ss (tags=%r, r18=%r, exclude_ai=%r)",
provider,
60.0,
tags,
is_r18,
self._config.exclude_ai,
)
await event.reply("图片获取超时，请稍后重试。")
return
async for result in self._core.send_images_by_url(
event, img_urls, is_r18, tags
```
To make this compile and follow existing conventions, you may also need to:
1. Ensure `asyncio` is imported in `handlers/command_handlers.py` (if not already present):
- `import asyncio`
2. Ensure a module-level logger is defined and imported where appropriate, for example:
- Add `import logging` at the top of the file.
- Add `logger = logging.getLogger(__name__)` near the top-level definitions.
3. If `event.reply` is not the correct way to send a message in your framework, replace `await event.reply("图片获取超时，请稍后重试。")` with the appropriate user-facing message API used elsewhere in this handler.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Owner
Author
|
@sourcery-ai review |
Contributor
|
Sorry @FlanChanXwO, you have reached your weekly rate limit of 500000 diff characters. Please try again later or upgrade to continue using Sourcery |
Owner
Author
|
@copilot review |
I reviewed the latest 9 commits on this PR and didn’t find any additional issues requiring changes at this time. Current head: |
Owner
Author
我重新逐项复查了： |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Close #16
access_control_modeconfig to switch between none/blacklist/whitelist modeswhitelist_groupsfor explicit allowlist, keepblocked_groupsfor backwards compatibilitySummary by Sourcery
为安全功能添加可配置的用户组访问控制模式，并相应更新文档和元数据。
New Features:
none、blacklist和whitelist三种用户组权限策略。Enhancements:
Documentation:
Chores:
1.3.0，并在更新日志中记录新的访问控制功能。Original summary in English
Summary by Sourcery
Add configurable group access control modes for safety features and update documentation and metadata accordingly.
New Features:
Enhancements:
Documentation:
Chores:
Sourcery 总结
为 setu 和 fortune 功能引入特性级安全访问控制模式，分别配置黑名单/白名单，并将其集成到 AstrBot 的主配置和指令处理器中。
新功能：
增强：
文档：
Original summary in English
Summary by Sourcery
Introduce feature-level safety access control modes with distinct blacklist/whitelist configurations for setu and fortune features, and integrate these with AstrBot's main config and command handlers.
New Features:
Enhancements:
Documentation: