From b2c5b6abbd916db1eed82b8ce5d4089cf2abb837 Mon Sep 17 00:00:00 2001 From: BegoniaHe Date: Tue, 15 Sep 2026 04:56:04 +0200 Subject: [PATCH 1/3] feat(session): add match and except filters Allow operators to restrict forwarded human messages per directed edge with /session filter after a watch or connect exists. Fixes #195 AI-Generated: true Generated-At: 2026-09-15T02:55:53Z --- .../.astrbot-plugin/i18n/en-US.json | 9 + .../.astrbot-plugin/i18n/zh-CN.json | 9 + .../builtin_commands/commands/session.py | 132 +++++++++- .../builtin_stars/builtin_commands/main.py | 8 + astrbot/core/platform/message_projection.py | 5 +- astrbot/core/platform/session_bridge.py | 183 ++++++++++++- .../core/platform/session_bridge_filter.py | 242 +++++++++++++++++ astrbot/core/platform/session_bridge_state.py | 27 +- docs/en/dev/star/guides/send-message.md | 2 +- docs/en/use/authorization.md | 2 +- docs/en/use/command.md | 5 +- docs/zh/dev/star/guides/send-message.md | 2 +- docs/zh/use/authorization.md | 2 +- docs/zh/use/command.md | 5 +- tests/unit/test_builtin_command_extensions.py | 3 + tests/unit/test_session_bridge.py | 249 ++++++++++++++++++ tests/unit/test_session_bridge_filter.py | 151 +++++++++++ tests/unit/test_session_commands.py | 78 ++++++ 18 files changed, 1101 insertions(+), 13 deletions(-) create mode 100644 astrbot/core/platform/session_bridge_filter.py create mode 100644 tests/unit/test_session_bridge_filter.py diff --git a/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/en-US.json b/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/en-US.json index 7c3e41b5a6..660a553074 100644 --- a/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/en-US.json +++ b/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/en-US.json @@ -37,6 +37,15 @@ "session.unpair.ok": "Removed both pair edges.", "session.unpair.missing": "No pair in this session.", "session.unpair.ambiguous": "This session has more than one pair. Specify the other UMO.", + "session.filter.usage": "Usage: /session filter [match|except] [subject|role|text] , or /session filter clear [match|except|all]. Omit extra tokens to show the current filter. clear without a side equals all. text uses the rest of the line as a Python re.search pattern.", + "session.filter.missing": "No matching session-bridge rule.", + "session.filter.invalid_role": "Role must be guest, member, session_admin, session_owner, instance_operator, operator, or root.", + "session.filter.invalid_pattern": "The text pattern must be a valid Python re expression of at most 256 characters.", + "session.filter.limit": "Each filter dimension accepts at most 16 values.", + "session.filter.empty": "(all)", + "session.filter.body": "Filter for {rule_id}:\nmatch:\n{match}\nexcept:\n{except_}", + "session.filter.updated": "Updated filter for {rule_id}:\nmatch:\n{match}\nexcept:\n{except_}", + "session.filter.cleared": "Cleared filter for {rule_id}:\nmatch:\n{match}\nexcept:\n{except_}", "session.disconnect.ok": "Session disconnected.", "session.disconnect.missing": "No unbounded link in this session.", "session.send.invalid": "Usage: /send , or /send after /session connect. Keep the command header in the message text.", diff --git a/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/zh-CN.json b/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/zh-CN.json index 609562ba98..43ad75ef70 100644 --- a/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/zh-CN.json +++ b/astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/zh-CN.json @@ -37,6 +37,15 @@ "session.unpair.ok": "已删除该 pair 的两条边。", "session.unpair.missing": "当前会话没有 pair。", "session.unpair.ambiguous": "当前会话有多对 pair，请补上对端 UMO。", + "session.filter.usage": "用法：/session filter [match|except] [subject|role|text] <值>，或 /session filter clear [match|except|all]。省略其余参数可查看当前过滤。clear 省略范围等于 all。text 把行内剩余内容当作 Python re.search 模式。", + "session.filter.missing": "没有找到对应的会话桥接规则。", + "session.filter.invalid_role": "角色必须是 guest、member、session_admin、session_owner、instance_operator、operator 或 root。", + "session.filter.invalid_pattern": "正文模式必须是不超过 256 个字符的合法 Python re 表达式。", + "session.filter.limit": "每个过滤维度最多 16 条。", + "session.filter.empty": "（全部）", + "session.filter.body": "{rule_id} 的过滤：\nmatch：\n{match}\nexcept：\n{except_}", + "session.filter.updated": "已更新 {rule_id} 的过滤：\nmatch：\n{match}\nexcept：\n{except_}", + "session.filter.cleared": "已清空 {rule_id} 的过滤：\nmatch：\n{match}\nexcept：\n{except_}", "session.disconnect.ok": "已断开会话连接。", "session.disconnect.missing": "当前会话没有无期限连接。", "session.send.invalid": "用法：/send <文字或附件>，或先 /session connect 后再 /send。请保留消息文字中的指令头。", diff --git a/astrbot/builtin_stars/builtin_commands/commands/session.py b/astrbot/builtin_stars/builtin_commands/commands/session.py index c28118fe3f..5bb27b7fa8 100644 --- a/astrbot/builtin_stars/builtin_commands/commands/session.py +++ b/astrbot/builtin_stars/builtin_commands/commands/session.py @@ -1,3 +1,5 @@ +from dataclasses import dataclass + from astrbot.api import Subject, star from astrbot.api.event import AstrMessageEvent from astrbot.api.platform import ( @@ -9,6 +11,20 @@ from .reply import reply_i18n from .target import resolve_target_umo +_RULE_ID_CHARS = frozenset("0123456789abcdef") +_FILTER_SIDES = frozenset({"match", "except"}) +_COMMAND_DIMENSIONS = { + "subject": "subjects", + "role": "roles", + "text": "text", +} + + +def _parse_rule_id(token: str) -> str: + if len(token) != 12 or any(char not in _RULE_ID_CHARS for char in token): + raise ValueError("Invalid filter arguments") + return token + def _resolve_listener(token: str, current_umo: str) -> str: if token.lower() == "this": @@ -55,10 +71,55 @@ def parse_unlink_spec(spec: str) -> str: parts = spec.split() if len(parts) != 1: raise ValueError("Invalid unlink arguments") - rule_id = parts[0] - if len(rule_id) != 12 or any(char not in "0123456789abcdef" for char in rule_id): - raise ValueError("Invalid unlink arguments") - return rule_id + try: + return _parse_rule_id(parts[0]) + except ValueError: + raise ValueError("Invalid unlink arguments") from None + + +@dataclass(frozen=True, slots=True) +class FilterSpec: + """Parsed `/session filter` arguments.""" + + rule_id: str + action: str + side: str = "" + dimension: str = "" + value: str = "" + + +def parse_filter_spec(spec: str) -> FilterSpec: + """Parse `/session filter [match|except|clear ...]`. + + ``text`` consumes the remainder as ``GreedyStr``. ``clear`` without a + side equals ``all``. + """ + stripped = spec.strip() + if not stripped: + raise ValueError("Invalid filter arguments") + parts = stripped.split() + rule_id = _parse_rule_id(parts[0]) + if len(parts) == 1: + return FilterSpec(rule_id, "show") + if parts[1] == "clear": + if len(parts) == 2: + return FilterSpec(rule_id, "clear", "all") + if len(parts) == 3 and parts[2] in {"match", "except", "all"}: + return FilterSpec(rule_id, "clear", parts[2]) + raise ValueError("Invalid filter arguments") + if parts[1] not in _FILTER_SIDES or len(parts) < 4: + raise ValueError("Invalid filter arguments") + command_dimension = parts[2] + dimension = _COMMAND_DIMENSIONS.get(command_dimension) + if dimension is None: + raise ValueError("Invalid filter arguments") + if command_dimension == "text": + value = stripped.split(None, 3)[3] + elif len(parts) == 4: + value = parts[3] + else: + raise ValueError("Invalid filter arguments") + return FilterSpec(rule_id, "append", parts[1], dimension, value) def parse_pair_spec(spec: str) -> str: @@ -476,6 +537,69 @@ async def unpair(self, event: AstrMessageEvent, spec: str = "") -> None: "session.unpair.ok" if removed else "session.unpair.missing", ) + async def _format_filter_side(self, event: AstrMessageEvent, payload: dict) -> str: + lines = [] + for key in ("subjects", "roles", "text"): + values = payload.get(key) or [] + if values: + lines.append(f" {key}: {', '.join(values)}") + if lines: + return "\n".join(lines) + return await self.context.i18n.t(event, "session.filter.empty") + + async def filter_rule(self, event: AstrMessageEvent, spec: str) -> None: + """Show or change match/except filters on one directed edge.""" + try: + parsed = parse_filter_spec(spec) + manager = self.context.bridges._manager + if parsed.action == "show": + result = await manager.get_filter(event, parsed.rule_id) + elif parsed.action == "clear": + result = await manager.clear_filter(event, parsed.rule_id, parsed.side) + else: + result = await manager.append_filter( + event, + parsed.rule_id, + parsed.side, + parsed.dimension, + parsed.value, + ) + except PermissionError: + await reply_i18n(self.context, event, "session.bridge.denied") + return + except ValueError as exc: + message = str(exc) + if message == "Invalid filter role": + await reply_i18n(self.context, event, "session.filter.invalid_role") + return + if message == "Invalid filter pattern": + await reply_i18n(self.context, event, "session.filter.invalid_pattern") + return + if message == "Filter limit exceeded": + await reply_i18n(self.context, event, "session.filter.limit") + return + await reply_i18n(self.context, event, "session.filter.usage") + return + if result is None: + await reply_i18n(self.context, event, "session.filter.missing") + return + match, except_ = result + match_text = await self._format_filter_side(event, match) + except_text = await self._format_filter_side(event, except_) + key = "session.filter.body" + if parsed.action == "clear": + key = "session.filter.cleared" + elif parsed.action == "append": + key = "session.filter.updated" + await reply_i18n( + self.context, + event, + key, + rule_id=parsed.rule_id, + match=match_text, + except_=except_text, + ) + async def disconnect(self, event: AstrMessageEvent) -> None: """Drop the unbounded link owned by the current actor.""" try: diff --git a/astrbot/builtin_stars/builtin_commands/main.py b/astrbot/builtin_stars/builtin_commands/main.py index b5172fe5fe..22e72bc5fb 100644 --- a/astrbot/builtin_stars/builtin_commands/main.py +++ b/astrbot/builtin_stars/builtin_commands/main.py @@ -178,6 +178,14 @@ async def session_unlink( """Remove a session-bridge rule by id.""" await self.session_c.unlink(event, spec) + @filter.permission("session.read") + @session.command("filter") + async def session_filter( + self, event: AstrMessageEvent, spec: GreedyStr = GreedyStr("") + ) -> None: + """Show or change match/except filters on one session-bridge edge.""" + await self.session_c.filter_rule(event, spec) + @filter.permission("session.block") @session.command("block") async def session_block( diff --git a/astrbot/core/platform/message_projection.py b/astrbot/core/platform/message_projection.py index 751fc8ccdb..0dd6c97950 100644 --- a/astrbot/core/platform/message_projection.py +++ b/astrbot/core/platform/message_projection.py @@ -228,7 +228,10 @@ def envelope_from_event(event: AstrMessageEvent) -> MessageEnvelope: source_message_id=str(source_message_id) if source_message_id else None, quote=quote, created_at=event.created_at, - metadata={"platform_name": event.get_platform_name()}, + metadata={ + "platform_name": event.get_platform_name(), + "bot_account_id": str(event.get_self_id() or ""), + }, ) diff --git a/astrbot/core/platform/session_bridge.py b/astrbot/core/platform/session_bridge.py index dadba2fb02..948857062b 100644 --- a/astrbot/core/platform/session_bridge.py +++ b/astrbot/core/platform/session_bridge.py @@ -11,7 +11,13 @@ from typing import TYPE_CHECKING from astrbot import logger -from astrbot.core.auth.models import AuthContext, Resource, Role, Subject +from astrbot.core.auth.models import ( + AuthContext, + AuthorizationValueError, + Resource, + Role, + Subject, +) from astrbot.core.message.message_event_result import MessageChain from .message_delivery import plan_message_delivery @@ -27,6 +33,15 @@ from .message_renderers import render_source_header from .message_session import MessageSession from .send_result import DeliveryAttempt, DeliveryReceipt, PlatformSendResult +from .session_bridge_filter import ( + append_filter_value, + coerce_filter_side, + compact_filter_side, + evaluate_filter, + has_filter_constraints, + portable_text, + validate_filter_value, +) from .session_bridge_state import ( PAIR_AMBIGUOUS, PAIR_OCCUPIED, @@ -68,6 +83,7 @@ def __init__( get_callback_base: Callable[[str], str] | None = None, get_locale: Callable[[str], Awaitable[str]] | None = None, get_platform_family: Callable[[str], str] | None = None, + get_self_id: Callable[[str], str | None] | None = None, store: SessionBridgeStore, default_ttl_seconds: int = DEFAULT_WATCH_TTL_SECONDS, max_watches_per_subject: int = 16, @@ -81,6 +97,7 @@ def __init__( self._get_callback_base = get_callback_base self._get_locale = get_locale self._get_platform_family = get_platform_family + self._get_self_id = get_self_id self._default_ttl_seconds = min( MAX_WATCH_TTL_SECONDS, max(MIN_WATCH_TTL_SECONDS, default_ttl_seconds) ) @@ -368,6 +385,92 @@ async def unlink(self, event: AstrMessageEvent, rule_id: str) -> bool: await self._state.delete_rule(rule_id) return True + async def get_filter( + self, event: AstrMessageEvent, rule_id: str + ) -> tuple[dict, dict] | None: + """Return match/except for a rule visible like ``list_links``.""" + row = await self._accessible_rule(event, rule_id) + if row is None: + return None + return coerce_filter_side(row.match), coerce_filter_side(row.except_) + + async def append_filter( + self, + event: AstrMessageEvent, + rule_id: str, + side: str, + dimension: str, + value: str, + ) -> tuple[dict, dict] | None: + """Append one match or except value onto a directed edge.""" + stored = validate_filter_value(dimension, value) + row = await self._accessible_rule(event, rule_id) + if row is None: + return None + match = coerce_filter_side(row.match) + except_ = coerce_filter_side(row.except_) + if side == "match": + match = append_filter_value(match, dimension, stored) + elif side == "except": + except_ = append_filter_value(except_, dimension, stored) + else: + raise ValueError("Invalid filter arguments") + return await self._persist_filters(row, match, except_) + + async def clear_filter( + self, event: AstrMessageEvent, rule_id: str, side: str + ) -> tuple[dict, dict] | None: + """Clear match, except, or both sides of one edge.""" + row = await self._accessible_rule(event, rule_id) + if row is None: + return None + match = coerce_filter_side(row.match) + except_ = coerce_filter_side(row.except_) + if side in {"match", "all"}: + match = {} + if side in {"except", "all"}: + except_ = {} + if side not in {"match", "except", "all"}: + raise ValueError("Invalid filter arguments") + return await self._persist_filters(row, match, except_) + + async def _accessible_rule( + self, event: AstrMessageEvent, rule_id: str + ) -> SessionBridgeRule | None: + stored = await self._state.stored_rule(rule_id) + if stored is None: + return None + subject, _ = self._actor(event) + if stored.subject_id == subject.id: + return stored + role = await self._operator_role(event) + current_config = self._config_id(event.unified_msg_origin) + allowed = role in {Role.ROOT, Role.OPERATOR} or ( + role is Role.INSTANCE_OPERATOR + and current_config in {stored.source_config_id, stored.target_config_id} + ) + if not allowed: + raise PermissionError("Session operation is not authorized") + return stored + + async def _persist_filters( + self, + row: SessionBridgeRule, + match: dict, + except_: dict, + ) -> tuple[dict, dict]: + match = compact_filter_side(match) + except_ = compact_filter_side(except_) + async with self._lock: + updated = await self._state.persist_filters(row.rule_id, match, except_) + stored = updated or row + grant = self._state.refresh_grant_filters(stored, match, except_) + if grant is not None: + self._state.arm_expiry( + self._state.store_key(grant.watch), grant, self._expire_watch + ) + return match, except_ + def _watch_from_row(self, row: SessionBridgeRule) -> SessionWatch: expires_at = None if row.expires_at is None else float(row.expires_at) return SessionWatch( @@ -466,6 +569,8 @@ async def observe(self, envelope: MessageEnvelope) -> None: await self._authorize( grant.subject, grant.context, watch.target_umo, "session.watch" ) + if not await self._passes_filter(grant, envelope): + continue async with self._lock: if not self._grant_active(key, grant, time()): continue @@ -510,6 +615,82 @@ async def observe(self, envelope: MessageEnvelope) -> None: # Transport exceptions may contain credentials or private URLs. logger.warning("Session bridge delivery failed") + async def _passes_filter( + self, grant: WatchGrant, envelope: MessageEnvelope + ) -> bool: + match = coerce_filter_side(grant.match) + except_ = coerce_filter_side(grant.except_) + if not has_filter_constraints(match) and not has_filter_constraints(except_): + return True + needs_subjects = bool(match.get("subjects") or except_.get("subjects")) + needs_roles = bool(match.get("roles") or except_.get("roles")) + subject_id = self._sender_subject_id(envelope) if needs_subjects else None + role = Role.GUEST.value + if needs_roles: + role = (await self._sender_role(envelope)).value + return evaluate_filter( + match, + except_, + subject_id=subject_id, + has_sender=envelope.sender is not None, + role=role, + text=portable_text(envelope.content), + ) + + def _bot_account_id(self, envelope: MessageEnvelope) -> str: + metadata_id = envelope.metadata.get("bot_account_id") + if isinstance(metadata_id, str) and metadata_id: + return metadata_id + if self._get_self_id is None: + return "" + return self._get_self_id(envelope.source_route.platform_id) or "" + + def _sender_subject_id(self, envelope: MessageEnvelope) -> str | None: + if envelope.sender is None: + return None + bot_account_id = self._bot_account_id(envelope) + if not bot_account_id: + return None + try: + return Subject.im( + platform_instance=envelope.source_route.platform_id, + bot_account_id=bot_account_id, + sender_id=envelope.sender.id, + ).id + except AuthorizationValueError, ValueError: + return None + + async def _sender_role(self, envelope: MessageEnvelope) -> Role: + subject_id = self._sender_subject_id(envelope) + if ( + subject_id is None + or self._authorization is None + or self._get_config_id is None + ): + return Role.GUEST + try: + subject = Subject.from_id(subject_id) + origin = envelope.source_umo + config_id = self._config_id(origin) + resource = Resource.session(config_id, origin) + context = AuthContext( + subject=subject, + source="im", + config_id=config_id, + authenticated=True, + origin_session_resource_id=resource.id, + ) + decision = await self._authorization.authorize( + subject, "session.read", resource, context + ) + except AuthorizationValueError, PermissionError, ValueError: + return Role.GUEST + except asyncio.CancelledError: + raise + except Exception: + return Role.GUEST + return decision.effective_role or Role.GUEST + def _platform_family(self, umo: str) -> str: if self._get_platform_family is None: return umo.split(":", 1)[0] diff --git a/astrbot/core/platform/session_bridge_filter.py b/astrbot/core/platform/session_bridge_filter.py new file mode 100644 index 0000000000..d90e3bb780 --- /dev/null +++ b/astrbot/core/platform/session_bridge_filter.py @@ -0,0 +1,242 @@ +"""Match/except evaluation for one session-bridge directed edge.""" + +from __future__ import annotations + +import re +from collections.abc import Mapping, Sequence +from typing import Any + +from astrbot.core.auth.models import Role + +from .message_protocol import NativeContent, PortablePart + +FILTER_DIMENSIONS = ("subjects", "roles", "text") +FILTER_SIDES = ("match", "except") +MAX_FILTER_ITEMS = 16 +MAX_PATTERN_LENGTH = 256 +FILTER_ROLES = frozenset(role.value for role in Role) +COMMAND_DIMENSIONS = { + "subject": "subjects", + "role": "roles", + "text": "text", +} + +_RULE_ID_CHARS = frozenset("0123456789abcdef") + + +def coerce_filter_side(payload: object) -> dict[str, list[str]]: + """Return a copy of one match or except document with known list keys.""" + if not isinstance(payload, Mapping): + return {} + out: dict[str, list[str]] = {} + for key in FILTER_DIMENSIONS: + raw = payload.get(key) + if not isinstance(raw, list): + continue + values = [item for item in raw if isinstance(item, str) and item] + if values: + out[key] = list(values) + return out + + +def compact_filter_side(payload: Mapping[str, Sequence[str]]) -> dict[str, list[str]]: + """Drop empty dimensions so an empty document stays ``{}``.""" + out: dict[str, list[str]] = {} + for key in FILTER_DIMENSIONS: + values = [item for item in payload.get(key, ()) if item] + if values: + out[key] = list(values) + return out + + +def has_filter_constraints(payload: Mapping[str, Sequence[str]]) -> bool: + """Return True when at least one dimension has a value.""" + return any(payload.get(key) for key in FILTER_DIMENSIONS) + + +def portable_text(content: Sequence[object]) -> str: + """Concatenate PortablePart string values; ignore media and NativeContent.""" + parts: list[str] = [] + for item in content: + if isinstance(item, NativeContent): + continue + if not isinstance(item, PortablePart): + continue + if isinstance(item.value, str): + parts.append(item.value) + return "".join(parts) + + +def compile_text_pattern(pattern: str) -> re.Pattern[str]: + """Compile one stdlib ``re`` pattern at save time. + + Args: + pattern: Caller-supplied search pattern. + + Returns: + The compiled pattern. Matching is Unicode and case-sensitive unless + the pattern itself enables ``(?i)``. + + Raises: + ValueError: Pattern is empty, longer than 256 characters, or invalid. + """ + if not pattern or len(pattern) > MAX_PATTERN_LENGTH: + raise ValueError("Invalid filter pattern") + try: + return re.compile(pattern) + except re.error as exc: + raise ValueError("Invalid filter pattern") from exc + + +def validate_filter_value(dimension: str, value: str) -> str: + """Normalize one appended filter value. + + Args: + dimension: ``subjects``, ``roles``, or ``text``. + value: Raw command value. + + Returns: + The stored value. + + Raises: + ValueError: The value is empty, an unknown role, or an invalid pattern. + """ + text = value.strip() if dimension != "text" else value + if dimension == "text": + compile_text_pattern(text) + return text + if not text: + raise ValueError("Invalid filter arguments") + if dimension == "roles": + if text not in FILTER_ROLES: + raise ValueError("Invalid filter role") + return text + if dimension != "subjects": + raise ValueError("Invalid filter arguments") + return text + + +def append_filter_value( + payload: Mapping[str, Sequence[str]], + dimension: str, + value: str, +) -> dict[str, list[str]]: + """Append one value onto a dimension, rejecting a 17th distinct item. + + Args: + payload: Current match or except document. + dimension: ``subjects``, ``roles``, or ``text``. + value: Already-validated value. + + Returns: + Compacted document including the new value. + + Raises: + ValueError: The dimension already has 16 values and ``value`` is new. + """ + current = coerce_filter_side(payload) + items = list(current.get(dimension, [])) + if value in items: + return compact_filter_side(current) + if len(items) >= MAX_FILTER_ITEMS: + raise ValueError("Filter limit exceeded") + items.append(value) + current[dimension] = items + return compact_filter_side(current) + + +def parse_rule_id(token: str) -> str: + """Return a 12-character lowercase hex rule id.""" + if len(token) != 12 or any(char not in _RULE_ID_CHARS for char in token): + raise ValueError("Invalid filter arguments") + return token + + +def _patterns_match(patterns: Sequence[str], text: str) -> bool: + for pattern in patterns: + try: + compiled = re.compile(pattern) + except re.error: + continue + if compiled.search(text) is not None: + return True + return False + + +def _dimension_hits( + payload: Mapping[str, Sequence[str]], + dimension: str, + *, + subject_id: str | None, + has_sender: bool, + role: str, + text: str, +) -> bool: + values = list(payload.get(dimension, ())) + if not values: + return True + if dimension == "subjects": + if not has_sender or subject_id is None: + return False + return subject_id in values + if dimension == "roles": + return role in values + if not text: + return False + return _patterns_match(values, text) + + +def side_matches( + payload: Mapping[str, Sequence[str]], + *, + subject_id: str | None, + has_sender: bool, + role: str, + text: str, +) -> bool: + """Return whether every set dimension on one side hits. + + Empty documents hit. Dimensions AND; values in one dimension OR. + Missing sender fails ``subjects``. Empty body fails ``text``. + """ + if not has_filter_constraints(payload): + return True + return all( + _dimension_hits( + payload, + dimension, + subject_id=subject_id, + has_sender=has_sender, + role=role, + text=text, + ) + for dimension in FILTER_DIMENSIONS + ) + + +def evaluate_filter( + match: Mapping[str, Sequence[str]] | None, + except_: Mapping[str, Sequence[str]] | None, + *, + subject_id: str | None, + has_sender: bool, + role: str, + text: str, +) -> bool: + """Return True when the snapshot should be forwarded. + + Empty match and except forward. Match dimensions AND; except hit drops. + """ + match_doc = coerce_filter_side(match) + except_doc = coerce_filter_side(except_) + kwargs: dict[str, Any] = { + "subject_id": subject_id, + "has_sender": has_sender, + "role": role, + "text": text, + } + if not side_matches(match_doc, **kwargs): + return False + if not has_filter_constraints(except_doc): + return True + return not side_matches(except_doc, **kwargs) diff --git a/astrbot/core/platform/session_bridge_state.py b/astrbot/core/platform/session_bridge_state.py index 7911fd1dc5..96ea6420c8 100644 --- a/astrbot/core/platform/session_bridge_state.py +++ b/astrbot/core/platform/session_bridge_state.py @@ -5,12 +5,13 @@ import asyncio import secrets from collections.abc import Callable, Coroutine -from dataclasses import dataclass +from dataclasses import dataclass, field, replace from time import time from typing import TYPE_CHECKING from astrbot import logger from astrbot.core.auth.models import AuthContext, Subject +from astrbot.core.platform.session_bridge_filter import coerce_filter_side if TYPE_CHECKING: from astrbot.core.db.po.session_bridge import SessionBridgeRule @@ -79,6 +80,8 @@ class WatchGrant: target_config_id: str header: bool = True pair_id: str | None = None + match: dict = field(default_factory=dict) + except_: dict = field(default_factory=dict) ExpireWatch = Callable[[GrantKey, WatchGrant], Coroutine[object, object, None]] @@ -444,6 +447,14 @@ async def stored_rules_touching_config( async def discard_stored_rule(self, rule_id: str) -> None: await self._store.delete_session_bridge_rule(rule_id) + async def persist_filters( + self, rule_id: str, match: dict, except_: dict + ) -> SessionBridgeRule | None: + """Replace match and except documents on one stored edge.""" + return await self._store.update_session_bridge_rule( + rule_id, match=match, except_=except_ + ) + async def persist_config_ids( self, row: SessionBridgeRule, @@ -497,8 +508,22 @@ def _grant_from_row( row.target_config_id, row.header, row.pair_id, + coerce_filter_side(row.match), + coerce_filter_side(row.except_), ) + def refresh_grant_filters( + self, row: SessionBridgeRule, match: dict, except_: dict + ) -> WatchGrant | None: + """Replace live match/except without dropping the grant identity map.""" + key = (row.subject_id, row.source_umo, row.target_umo) + grant = self._grants.get(key) + if grant is None or grant.watch.rule_id != row.rule_id: + return None + updated = replace(grant, match=match, except_=except_) + self._index(updated) + return updated + def _index(self, grant: WatchGrant) -> None: key = self.store_key(grant.watch) self._grants[key] = grant diff --git a/docs/en/dev/star/guides/send-message.md b/docs/en/dev/star/guides/send-message.md index 810b0fa6a9..05d37e9111 100644 --- a/docs/en/dev/star/guides/send-message.md +++ b/docs/en/dev/star/guides/send-message.md @@ -89,7 +89,7 @@ links, and pairs persist in SQLite, so unexpired rules survive process restart. `remaining_seconds` uses the wall clock; unbounded links return `0`. Do not construct `SessionBridgeManager` yourself. Import `SessionWatch` and the duration constants from `astrbot.api.platform`. `links` / `unlink` / -`pair` / `unpair` are IM commands only in this stage; +`pair` / `unpair` / `filter` are IM commands only in this stage; `SessionBridgeCapability` does not expose them. A pair is two headerless directed edges that share a `pair_id`: the far side sees the destination Bot account, the source platform identity is not forged, `/send` is not bound, diff --git a/docs/en/use/authorization.md b/docs/en/use/authorization.md index 0c2d9348bc..d85c1a6ca3 100644 --- a/docs/en/use/authorization.md +++ b/docs/en/use/authorization.md @@ -54,7 +54,7 @@ Profiles can bind separately to platforms, groups, or DMs. Editing `default` may `session.watch` and `session.send` are separate cross-session permissions requiring `instance_operator` or a higher role. Both sessions must belong to the same configuration. A group admin or private-session owner in the source session cannot use that status to watch another session. Existing cross-session restrictions on `session.manage` and `session.assign` remain in effect. -`/session watch` creates an expiring watch from the trusted event identity; command arguments cannot choose an actor. `/session connect` creates an unbounded 1:1 link with the same authorization rules. `/session pair` also requires `session.watch` and writes two headerless directed edges; `/session unpair` uses the same permission as `disconnect` (`session.read`). Rules persist across restart. On restore, AstrBot rebuilds the subject with `Subject.from_id` and authorizes again; lost permission deletes the edge instead of silently continuing. Authorization is checked again before submission. Stopping a watch, link, or pair, or revoking access, prevents subsequent queued submissions. Forwarded content is visible to everyone in the receiving session. A pair does not forge the source sender's platform identity and does not inject forwards into the target inbound pipeline. This stage has no Dashboard management surface; see [Built-in commands](./command#cross-session-watches-and-sending). +`/session watch` creates an expiring watch from the trusted event identity; command arguments cannot choose an actor. `/session connect` creates an unbounded 1:1 link with the same authorization rules. `/session pair` also requires `session.watch` and writes two headerless directed edges; `/session unpair` uses the same permission as `disconnect` (`session.read`). Rules persist across restart. On restore, AstrBot rebuilds the subject with `Subject.from_id` and authorizes again; lost permission deletes the edge instead of silently continuing. Authorization is checked again before submission. Stopping a watch, link, or pair, or revoking access, prevents subsequent queued submissions. Forwarded content is visible to everyone in the receiving session. A pair does not forge the source sender's platform identity and does not inject forwards into the target inbound pipeline. After a rule exists, `/session filter` can restrict which human messages are forwarded; create commands do not take filter flags. This stage has no Dashboard management surface; see [Built-in commands](./command#cross-session-watches-and-sending). ## Step-up diff --git a/docs/en/use/command.md b/docs/en/use/command.md index 5759a617a6..a8ff9ebe4e 100644 --- a/docs/en/use/command.md +++ b/docs/en/use/command.md @@ -77,10 +77,13 @@ The user ID from `/session info` can be granted current-session `session_admin` - `/session unpair [UMO]`: Delete both edges that share a `pair_id`; requires `session.read`. Omit the UMO only when this actor has exactly one pair that includes the current session; zero pairs reports none; more than one pair requires the other UMO. - `/session links`: List visible watches, connects, and pairs with `rule_id`, kind, and remaining time or no expiry; pairs also show `pair_id`. Permission matches `watches` (`session.read`). Creators see only their own rules. An `instance_operator` on the current session configuration also sees rules whose source or target config id equals that configuration. `operator` / `root` see all rules. - `/session unlink `: Remove a watch or connect by its 12-character lowercase hex id. Creators can remove their own rules after their role is revoked. An `instance_operator` on the current configuration can remove a rule whose either config id belongs to that configuration; `operator` / `root` can remove any rule. If the target is `kind=pair`, the command is refused and both edges remain; use `unpair`. +- `/session filter `: Show the match/except filter on one directed edge. Permission matches `links` (`session.read` plus the operator extra scope). +- `/session filter match|except subject|role|text `: Append one filter value to that edge. The same dimension is OR; dimensions AND; an except hit drops the message. `text` takes the rest of the line as a Python `re.search` pattern (at most 256 characters, Unicode and case-sensitive unless the pattern uses `(?i)`). Invalid patterns are rejected at save time. `role` is an AstrBot authorization role on the **source session** (`guest` / `member` / `session_admin` / `session_owner` / `instance_operator` / `operator` / `root`), not a platform group role. A lookup miss is `guest`. Each dimension accepts at most 16 values. Creators can change their own edges after their role is revoked, with the same extra operator scope as `unlink`. Pair edges can be filtered independently; unpair still removes both. +- `/session filter clear [match|except|all]`: Clear one side or both. Omitting the side equals `all`. Clearing removes the whole side; to change one subject, clear and add again. - `/send [content]`: Send text and attachments from the same message through the target Bot account; requires `session.send`. An attachment-only body is allowed. This does not register a `reply` command. - `/send [content]`: After `/session connect`, send to the linked target without repeating the UMO. Attachment-only bodies are allowed. `pair` is not a default `/send` target. -Watching, connecting, pairing, and sending require the current identity to hold `instance_operator` permission in the configuration shared by both sessions. Group admin or private-session ownership does not grant this access. Forwarded content is visible to everyone in the receiving session. Watches, links, and pairs forward new incoming messages only, without reading history. Rules are stored in SQLite, so unexpired watches, all connects, and all pairs survive process restart; expired watches are cleared at startup or when they elapse, and the listener is notified. Limits remain 16 watches, 16 links, and 8 pairs per actor. Authorization is checked again for each forwarded message, so revocation stops delivery. Running `/session watch` again on the same endpoints keeps the `rule_id` and resets the duration. A watch and a connect on the same direction replace each other. If either target direction is already a pair, `watch` / `connect` is refused and asks for `unpair` first; the pair is not split. +Watching, connecting, pairing, and sending require the current identity to hold `instance_operator` permission in the configuration shared by both sessions. Group admin or private-session ownership does not grant this access. Forwarded content is visible to everyone in the receiving session. Watches, links, and pairs forward new incoming messages only, without reading history. Empty match and except filters forward every human message. `watch` / `connect` / `pair` do not accept filter flags; set filters afterwards with `/session filter`. Subject filters rebuild `Subject.im` from the source route platform id, the source adapter `self_id`, and `SenderSnapshot.id`. Text filters concatenate `PortablePart` string values and ignore media and `NativeContent`; a media-only body fails `match.text` and does not trip `except.text`. Rules are stored in SQLite, so unexpired watches, all connects, and all pairs survive process restart; expired watches are cleared at startup or when they elapse, and the listener is notified. Limits remain 16 watches, 16 links, and 8 pairs per actor. Authorization is checked again for each forwarded message, so revocation stops delivery. Running `/session watch` again on the same endpoints keeps the `rule_id` and resets the duration. A watch and a connect on the same direction replace each other. If either target direction is already a pair, `watch` / `connect` is refused and asks for `unpair` first; the pair is not split. Text and media retain their message-chain order; targets without mixed-content delivery receive separate messages. Cross-platform mentions become text. Quotes use accepted-message ID mappings when available and otherwise become a quote summary. Unavailable attachments and unsupported native content leave text placeholders. Platform cards, private syntax, and mini apps cannot be guaranteed to reproduce on another platform. diff --git a/docs/zh/dev/star/guides/send-message.md b/docs/zh/dev/star/guides/send-message.md index 2a65b1f8a9..f4ab2dadb2 100644 --- a/docs/zh/dev/star/guides/send-message.md +++ b/docs/zh/dev/star/guides/send-message.md @@ -75,7 +75,7 @@ async def watch_room(self, event: AstrMessageEvent, target_umo: str): - `connection(event)`：返回当前无期限连接，没有则是 `None`。 - `send(event, target_umo, *, target_in_header=True)`：把当前消息去掉指令头后的正文和附件投递到目标会话，返回 `DeliveryReceipt`。已连接且不带 UMO 的 `/send` 传 `target_in_header=False`。 -这些方法会再次调用 `authorize()`，要求 `session.watch` 或 `session.send`，且两个会话属于同一配置。监听、连接和 pair 会持久化，重启后未过期的规则仍在。`remaining_seconds` 按墙钟计算，无期限返回 `0`。不要自己构造 `SessionBridgeManager`。`SessionWatch` 和时长常量可从 `astrbot.api.platform` 导入。`links` / `unlink` / `pair` / `unpair` 本阶段只做 IM 指令，没有对应的 `SessionBridgeCapability` 方法。`pair` 是两条共享 `pair_id` 的无头有向边：对岸看到的是目标侧 Bot 账号，不伪造源平台身份，不绑定 `/send`，拆对只能 `unpair`。当前没有 Dashboard 管理面，插件也不应假设存在对应 HTTP API。 +这些方法会再次调用 `authorize()`，要求 `session.watch` 或 `session.send`，且两个会话属于同一配置。监听、连接和 pair 会持久化，重启后未过期的规则仍在。`remaining_seconds` 按墙钟计算，无期限返回 `0`。不要自己构造 `SessionBridgeManager`。`SessionWatch` 和时长常量可从 `astrbot.api.platform` 导入。`links` / `unlink` / `pair` / `unpair` / `filter` 本阶段只做 IM 指令，没有对应的 `SessionBridgeCapability` 方法。`pair` 是两条共享 `pair_id` 的无头有向边：对岸看到的是目标侧 Bot 账号，不伪造源平台身份，不绑定 `/send`，拆对只能 `unpair`。当前没有 Dashboard 管理面，插件也不应假设存在对应 HTTP API。 ## 富媒体消息链 diff --git a/docs/zh/use/authorization.md b/docs/zh/use/authorization.md index 1002b0aee1..2fe5d8c821 100644 --- a/docs/zh/use/authorization.md +++ b/docs/zh/use/authorization.md @@ -54,7 +54,7 @@ AstrBot 把 Dashboard 登录、IM 会话管理和高风险操作拆开。把群 `session.watch` 和 `session.send` 是独立的跨会话权限，要求 `instance_operator` 或更高角色，且两个会话必须属于同一配置。源会话的群管理员或私聊所有者身份不能用于监听其他会话。已有的 `session.manage`、`session.assign` 跨会话限制保持不变。 -`/session watch` 使用可信事件身份创建有期限的监听，不接受命令参数指定身份。`/session connect` 用同一套授权规则创建无期限的一对一连接。`/session pair` 同样需要 `session.watch`，写出两条无头有向边；`/session unpair` 的权限与 `disconnect` 相同（`session.read`）。规则会持久化；启动时用 `Subject.from_id` 重建主体并再次鉴权，权限已无则删除该边，不会静默续转。每次投递前重新检查权限，停止监听、连接或 pair、或撤销权限会阻止后续排队的投递。转发内容对接收会话的所有成员可见。`pair` 不伪造源发送者的平台身份，也不把转发灌进目标入站管道。本阶段没有 Dashboard 管理面，见 [内置指令](./command#跨会话监听与发送)。 +`/session watch` 使用可信事件身份创建有期限的监听，不接受命令参数指定身份。`/session connect` 用同一套授权规则创建无期限的一对一连接。`/session pair` 同样需要 `session.watch`，写出两条无头有向边；`/session unpair` 的权限与 `disconnect` 相同（`session.read`）。规则会持久化；启动时用 `Subject.from_id` 重建主体并再次鉴权，权限已无则删除该边，不会静默续转。每次投递前重新检查权限，停止监听、连接或 pair、或撤销权限会阻止后续排队的投递。转发内容对接收会话的所有成员可见。`pair` 不伪造源发送者的平台身份，也不把转发灌进目标入站管道。规则建好后可用 `/session filter` 按主体、角色或正文限制转发；创建指令不带过滤 flag。本阶段没有 Dashboard 管理面，见 [内置指令](./command#跨会话监听与发送)。 ## 二次验证（step-up） diff --git a/docs/zh/use/command.md b/docs/zh/use/command.md index 4ddd290788..40208c6da5 100644 --- a/docs/zh/use/command.md +++ b/docs/zh/use/command.md @@ -77,10 +77,13 @@ Orbit 不执行变量、命令、算术或波浪号展开，也不执行 glob、 - `/session unpair [UMO]`：删除共享 `pair_id` 的两条边，需要 `session.read`。恰好一对且当前会话是其中一端时可省略 UMO；0 对提示没有；多于一对必须带 UMO。 - `/session links`：列出可见的监听、连接和 pair，包含 `rule_id`、类型和剩余时间或无期限；pair 额外显示 `pair_id`。权限与 `watches` 相同（`session.read`）。创建者只看到自己的规则；当前会话配置上的 `instance_operator` 额外看到两端任一配置 id 属于当前会话配置的规则；`operator` / `root` 看到全部。 - `/session unlink `：按 12 位小写十六进制 id 撤销一条监听或连接。创建者可撤自己的规则（角色被撤后仍可）。本配置 `instance_operator` 可撤两端任一配置属于当前会话配置的规则；`operator` / `root` 可撤全部。目标是 `kind=pair` 时拒绝并提示使用 `unpair`，两条边都还在。 +- `/session filter `：查看一条有向边的 match/except 过滤。权限与 `links` 相同（`session.read` 加运营附加范围）。 +- `/session filter match|except subject|role|text <值>`：向该边追加一条过滤。同一维度为 OR，维度之间为 AND，命中 except 则丢弃。`text` 把行内剩余内容当作 Python `re.search` 模式（最长 256 字符，默认 Unicode 且大小写敏感，忽略大小写用 `(?i)`）。非法模式在保存时拒绝。`role` 是发送者在**源会话**上的 AstrBot 授权角色（`guest` / `member` / `session_admin` / `session_owner` / `instance_operator` / `operator` / `root`），不是平台群身份。查不到当 `guest`。每一维最多 16 条。创建者在撤权后仍可改自己的边，运营附加范围与 `unlink` 相同。pair 的两条边可分别过滤，拆对仍走 `unpair`。 +- `/session filter clear [match|except|all]`：清空一侧或两侧。省略范围等于 `all`。`clear` 清的是整个 match 或 except 侧，不能单删一条；要改就先 `clear` 再加。 - `/send [内容]`：借助目标平台的 Bot 账号发送文字和同一条消息中的附件，需要 `session.send`。可以只附图片而不填写正文；不会占用 `reply` 指令。 - `/send [内容]`：在 `/session connect` 之后，不写 UMO 也会发往已连接的目标会话。可以只附图片。`pair` 不会成为 `/send` 的默认目标。 -监听、连接、配对和发送要求当前身份拥有同一配置下的 `instance_operator` 权限。群管理员、私聊会话所有者身份不能替代它。监听内容对接收会话的所有成员可见；仅转发开始监听之后收到的消息，不读取历史。规则写入 SQLite，进程重启后未过期的监听、全部连接和全部 pair 仍在；过期监听会在启动或到期时清除并通知监听端。每人最多 16 条监听、16 条连接和 8 对 pair。每次转发都会重新检查权限，撤权后停止投递。对同一对会话再次 `/session watch` 会保留 `rule_id` 并重置时长。同一方向的 watch 与 connect 会互相替换。任一条目标方向已是 pair 时，`watch` / `connect` 会拒绝并提示先 `unpair`，不会拆成半对。 +监听、连接、配对和发送要求当前身份拥有同一配置下的 `instance_operator` 权限。群管理员、私聊会话所有者身份不能替代它。监听内容对接收会话的所有成员可见；仅转发开始监听之后收到的消息，不读取历史。空的 match 与 except 会转发全部真人消息。`watch` / `connect` / `pair` 创建指令不解析过滤 flag，事后用 `/session filter` 设置。主体过滤用 `Subject.im` 重建：平台实例取源路由 `platform_id`，`bot_account_id` 取源适配器 `self_id`，`sender_id` 取 `SenderSnapshot.id`。正文只拼接 `PortablePart` 文本，忽略媒体和 `NativeContent`；纯媒体在写了 `match.text` 时不转，只写 `except.text` 则放行。规则写入 SQLite，进程重启后未过期的监听、全部连接和全部 pair 仍在；过期监听会在启动或到期时清除并通知监听端。每人最多 16 条监听、16 条连接和 8 对 pair。每次转发都会重新检查权限，撤权后停止投递。对同一对会话再次 `/session watch` 会保留 `rule_id` 并重置时长。同一方向的 watch 与 connect 会互相替换。任一条目标方向已是 pair 时，`watch` / `connect` 会拒绝并提示先 `unpair`，不会拆成半对。 正文保留消息链中的图文先后顺序；不能混排的目标拆成多条消息。跨平台提及转成文字，引用优先通过已接受消息的 ID 映射还原；映射不存在时附引用摘要。无法解析的附件和不支持的原生内容会保留文字占位。平台自己的卡片、私有语法和小程序不能保证在别的平台重现。 diff --git a/tests/unit/test_builtin_command_extensions.py b/tests/unit/test_builtin_command_extensions.py index f3dce32a63..caaf967867 100644 --- a/tests/unit/test_builtin_command_extensions.py +++ b/tests/unit/test_builtin_command_extensions.py @@ -267,6 +267,7 @@ def test_all_builtin_extension_commands_use_native_command_schemas(): "session_unpair", "session_links", "session_unlink", + "session_filter", "session_block", "session_unblock", "send_to_session", @@ -1109,6 +1110,7 @@ def command_names(group: CommandGroupFilter) -> set[str]: "unpair", "links", "unlink", + "filter", }, "conversation": { "create", @@ -1179,6 +1181,7 @@ def test_non_public_builtin_commands_declare_the_planned_actions(): "session_unpair": "session.read", "session_links": "session.read", "session_unlink": "session.read", + "session_filter": "session.read", "session_block": "session.block", "session_unblock": "session.block", "send_to_session": "session.send", diff --git a/tests/unit/test_session_bridge.py b/tests/unit/test_session_bridge.py index cb7b7fa261..5928fbca4f 100644 --- a/tests/unit/test_session_bridge.py +++ b/tests/unit/test_session_bridge.py @@ -1210,3 +1210,252 @@ async def test_pair_insert_failure_keeps_replaced_watch(): assert [item.rule_id for item in listed] == [watch.rule_id] assert await store.get_session_bridge_rule(watch.rule_id) is not None await manager.terminate() + + +_OBSERVE_SEQ = 0 + + +def _observed_envelope(*, sender="1", content=None, metadata=None, message_id=""): + global _OBSERVE_SEQ + _OBSERVE_SEQ += 1 + return MessageEnvelope( + PlatformRouteIdentity("target", MessageType.GROUP_MESSAGE, "room"), + source_message_id=message_id or f"msg-{_OBSERVE_SEQ}", + sender=None if sender is None else SenderSnapshot(sender, "Alice", "napcat"), + content=content or (PortablePart(ContentKind.TEXT, "hello from room"),), + metadata=metadata or {}, + ) + + +@pytest.mark.asyncio +async def test_empty_filter_forwards_human_messages(): + sent = [] + + async def send(session, chain): + sent.append(chain.get_plain_text()) + return PlatformSendResult(session.platform_id, True, str(session)) + + manager, _, _ = _manager(send) + event = _event() + await manager.watch(event, "target:GroupMessage:room", ttl_seconds=60) + await manager.observe(_observed_envelope()) + assert any("hello from room" in item for item in sent) + await manager.terminate() + + +@pytest.mark.asyncio +async def test_observe_match_except_subjects_roles_text_and_media(): + from astrbot.core.platform.message_protocol import MediaReference + + sent = [] + + async def send(session, chain): + sent.append(chain.get_plain_text()) + return PlatformSendResult(session.platform_id, True, str(session)) + + async def authorize(subject, action, resource, context): + if action == "session.read": + role = Role.MEMBER if subject.id.endswith(":1") else Role.GUEST + return SimpleNamespace(allowed=True, effective_role=role) + return SimpleNamespace(allowed=True, effective_role=Role.INSTANCE_OPERATOR) + + manager, authorization, _ = _manager(send) + authorization.authorize = AsyncMock(side_effect=authorize) + event = _event() + watch = await manager.watch(event, "target:GroupMessage:room", ttl_seconds=60) + subject_id = Subject.im( + platform_instance="target", bot_account_id="bot", sender_id="1" + ).id + assert await manager.append_filter( + event, watch.rule_id, "match", "subjects", subject_id + ) + sent.clear() + await manager.observe(_observed_envelope(sender="2")) + assert sent == [] + await manager.observe(_observed_envelope(sender="1")) + assert any("hello from room" in item for item in sent) + await manager.append_filter(event, watch.rule_id, "except", "text", "hello") + sent.clear() + await manager.observe(_observed_envelope(sender="1")) + assert sent == [] + await manager.clear_filter(event, watch.rule_id, "all") + await manager.append_filter(event, watch.rule_id, "match", "roles", "member") + sent.clear() + await manager.observe(_observed_envelope(sender="1")) + assert any("hello from room" in item for item in sent) + await manager.observe(_observed_envelope(sender="9")) + assert len([item for item in sent if "hello from room" in item]) == 1 + read_calls = [ + call + for call in authorization.authorize.await_args_list + if call.args[1] == "session.read" + ] + assert read_calls + assert read_calls[0].args[3].platform_member_role == "unknown" + await manager.clear_filter(event, watch.rule_id, "all") + await manager.append_filter(event, watch.rule_id, "match", "text", "hello") + sent.clear() + await manager.observe( + _observed_envelope( + content=(PortablePart(ContentKind.IMAGE, MediaReference("file:///x.jpg")),) + ) + ) + assert sent == [] + await manager.clear_filter(event, watch.rule_id, "match") + await manager.append_filter(event, watch.rule_id, "except", "text", "hello") + sent.clear() + await manager.observe( + _observed_envelope( + content=(PortablePart(ContentKind.IMAGE, MediaReference("file:///x.jpg")),) + ) + ) + assert sent + await manager.terminate() + + +@pytest.mark.asyncio +async def test_filter_uses_adapter_self_id_not_default(): + sent = [] + + async def send(session, chain): + sent.append(chain.get_plain_text()) + return PlatformSendResult(session.platform_id, True, str(session)) + + manager, _, _ = _manager(send) + manager._get_self_id = lambda _: "real-bot" + event = _event() + watch = await manager.watch(event, "target:GroupMessage:room", ttl_seconds=60) + default_id = Subject.im( + platform_instance="target", bot_account_id="default", sender_id="1" + ).id + real_id = Subject.im( + platform_instance="target", bot_account_id="real-bot", sender_id="1" + ).id + await manager.append_filter(event, watch.rule_id, "match", "subjects", default_id) + await manager.observe(_observed_envelope()) + assert sent == [] + await manager.clear_filter(event, watch.rule_id, "match") + await manager.append_filter(event, watch.rule_id, "match", "subjects", real_id) + await manager.observe(_observed_envelope(metadata={"bot_account_id": "real-bot"})) + assert any("hello from room" in item for item in sent) + await manager.terminate() + + +@pytest.mark.asyncio +async def test_filter_missing_sender_and_pair_edges_independent(): + sent = [] + + async def send(session, chain): + sent.append(1) + return PlatformSendResult(session.platform_id, True, str(session)) + + store = FakeSessionBridgeStore() + manager, _, _ = _manager(send, store=store) + event = _event() + watch = await manager.watch(event, "target:GroupMessage:room", ttl_seconds=60) + subject_id = Subject.im( + platform_instance="target", bot_account_id="bot", sender_id="1" + ).id + await manager.append_filter(event, watch.rule_id, "match", "subjects", subject_id) + await manager.observe(_observed_envelope(sender=None)) + assert sent == [] + await manager.clear_filter(event, watch.rule_id, "match") + await manager.append_filter(event, watch.rule_id, "except", "subjects", subject_id) + await manager.observe(_observed_envelope(sender=None)) + assert sent + left = store.seed( + rule_id="pairleft00001", + subject_id=event.subject.id, + source_umo="pair:FriendMessage:left", + target_umo="pair:FriendMessage:right", + source_config_id="default", + target_config_id="default", + kind="pair", + expires_at=None, + pair_id="pairid000001", + ) + right = store.seed( + rule_id="pairright0001", + subject_id=event.subject.id, + source_umo="pair:FriendMessage:right", + target_umo="pair:FriendMessage:left", + source_config_id="default", + target_config_id="default", + kind="pair", + expires_at=None, + pair_id="pairid000001", + ) + await manager.append_filter(event, left.rule_id, "match", "text", "alpha") + await manager.append_filter(event, right.rule_id, "except", "text", "beta") + left_match, left_except = await manager.get_filter(event, left.rule_id) + right_match, right_except = await manager.get_filter(event, right.rule_id) + assert left_match == {"text": ["alpha"]} + assert left_except == {} + assert right_match == {} + assert right_except == {"text": ["beta"]} + with pytest.raises(ValueError, match="Invalid filter pattern"): + await manager.append_filter(event, watch.rule_id, "match", "text", "(") + await manager.clear_filter(event, watch.rule_id, "all") + for index in range(16): + await manager.append_filter( + event, watch.rule_id, "match", "subjects", f"im:target:bot:{index}" + ) + with pytest.raises(ValueError, match="Filter limit exceeded"): + await manager.append_filter( + event, watch.rule_id, "match", "subjects", "im:target:bot:16" + ) + await manager.terminate() + + +@pytest.mark.asyncio +async def test_filter_allows_creator_after_revoke_and_scopes_instance_operator(): + owner = _event() + other = _event(sender="other") + store = FakeSessionBridgeStore() + store.seed( + rule_id="ownwatch00001", + subject_id=owner.subject.id, + source_umo=owner.unified_msg_origin, + target_umo="target:GroupMessage:room", + source_config_id="default", + target_config_id="default", + kind="watch", + expires_at=int(time()) + 60, + ) + store.seed( + rule_id="foreignwatch01", + subject_id=other.subject.id, + source_umo="alt:FriendMessage:x", + target_umo="alt:GroupMessage:y", + source_config_id="othercfg", + target_config_id="othercfg", + kind="watch", + expires_at=int(time()) + 60, + ) + manager, authorization, _ = _manager(store=store) + authorization.authorize = AsyncMock( + return_value=SimpleNamespace(allowed=False, effective_role=None) + ) + assert await manager.append_filter(owner, "ownwatch00001", "match", "text", "keep") + assert await manager.get_filter(owner, "missingrule01") is None + store.seed( + rule_id="samecfgwatch01", + subject_id=other.subject.id, + source_umo="ops:FriendMessage:box", + target_umo="target:GroupMessage:room", + source_config_id="default", + target_config_id="ops", + kind="connect", + expires_at=None, + ) + authorization.authorize = AsyncMock( + return_value=SimpleNamespace( + allowed=True, effective_role=Role.INSTANCE_OPERATOR + ) + ) + assert await manager.append_filter( + owner, "samecfgwatch01", "except", "roles", "guest" + ) + with pytest.raises(PermissionError): + await manager.append_filter(owner, "foreignwatch01", "match", "text", "no") + await manager.terminate() diff --git a/tests/unit/test_session_bridge_filter.py b/tests/unit/test_session_bridge_filter.py new file mode 100644 index 0000000000..7371800e3f --- /dev/null +++ b/tests/unit/test_session_bridge_filter.py @@ -0,0 +1,151 @@ +import re + +import pytest + +from astrbot.core.auth.models import Role +from astrbot.core.platform.message_protocol import ( + ContentKind, + MediaReference, + NativeContent, + PortablePart, +) +from astrbot.core.platform.session_bridge_filter import ( + append_filter_value, + compile_text_pattern, + evaluate_filter, + portable_text, + validate_filter_value, +) + + +def test_empty_filter_forwards(): + assert evaluate_filter( + {}, + {}, + subject_id="im:target:bot:1", + has_sender=True, + role=Role.MEMBER.value, + text="hello", + ) + + +def test_match_and_except_subjects_roles_and_text(): + assert evaluate_filter( + {"subjects": ["im:target:bot:1"], "roles": ["member"], "text": ["hello"]}, + {}, + subject_id="im:target:bot:1", + has_sender=True, + role=Role.MEMBER.value, + text="hello world", + ) + assert not evaluate_filter( + {"subjects": ["im:target:bot:1"], "roles": ["member"], "text": ["hello"]}, + {}, + subject_id="im:target:bot:1", + has_sender=True, + role=Role.GUEST.value, + text="hello world", + ) + assert not evaluate_filter( + {"subjects": ["im:target:bot:1"]}, + {"subjects": ["im:target:bot:1"]}, + subject_id="im:target:bot:1", + has_sender=True, + role=Role.MEMBER.value, + text="hello", + ) + + +def test_same_dimension_is_or(): + assert evaluate_filter( + {"roles": ["member", "session_admin"]}, + {}, + subject_id="im:target:bot:1", + has_sender=True, + role=Role.SESSION_ADMIN.value, + text="hello", + ) + + +def test_missing_sender_subject_rules(): + assert not evaluate_filter( + {"subjects": ["im:target:bot:1"]}, + {}, + subject_id=None, + has_sender=False, + role=Role.GUEST.value, + text="hello", + ) + assert evaluate_filter( + {}, + {"subjects": ["im:target:bot:1"]}, + subject_id=None, + has_sender=False, + role=Role.GUEST.value, + text="hello", + ) + + +def test_pure_media_text_rules(): + assert ( + portable_text( + ( + PortablePart(ContentKind.IMAGE, MediaReference("file:///x.jpg")), + NativeContent("telegram", "sticker", "{}"), + ) + ) + == "" + ) + assert not evaluate_filter( + {"text": ["hello"]}, + {}, + subject_id="im:target:bot:1", + has_sender=True, + role=Role.MEMBER.value, + text="", + ) + assert evaluate_filter( + {}, + {"text": ["hello"]}, + subject_id="im:target:bot:1", + has_sender=True, + role=Role.MEMBER.value, + text="", + ) + + +def test_portable_text_concatenates_string_parts_only(): + assert ( + portable_text( + ( + PortablePart(ContentKind.TEXT, "a"), + PortablePart(ContentKind.IMAGE, MediaReference("file:///x.jpg")), + PortablePart(ContentKind.TEXT, "b"), + NativeContent("telegram", "sticker", "{}", fallback="no"), + ) + ) + == "ab" + ) + + +def test_compile_text_pattern_uses_stdlib_re(): + assert compile_text_pattern("hello").search("say hello") + assert compile_text_pattern("(?i)hello").search("HELLO") + assert compile_text_pattern("hello").search("HELLO") is None + with pytest.raises(ValueError, match="Invalid filter pattern"): + compile_text_pattern("(") + with pytest.raises(ValueError, match="Invalid filter pattern"): + compile_text_pattern("x" * 257) + assert isinstance(compile_text_pattern("a"), re.Pattern) + + +def test_validate_and_append_limits(): + assert validate_filter_value("roles", "session_admin") == "session_admin" + with pytest.raises(ValueError, match="Invalid filter role"): + validate_filter_value("roles", "admin") + current = {} + for index in range(16): + current = append_filter_value(current, "subjects", f"im:target:bot:{index}") + current = append_filter_value(current, "subjects", "im:target:bot:0") + with pytest.raises(ValueError, match="Filter limit exceeded"): + append_filter_value(current, "subjects", "im:target:bot:16") diff --git a/tests/unit/test_session_commands.py b/tests/unit/test_session_commands.py index a4255f7772..ab8951fe87 100644 --- a/tests/unit/test_session_commands.py +++ b/tests/unit/test_session_commands.py @@ -4,10 +4,13 @@ import pytest from astrbot.builtin_stars.builtin_commands.commands.session import ( + FilterSpec, SessionCommands, + parse_filter_spec, parse_pair_spec, parse_unlink_spec, parse_unpair_spec, + parse_watch_spec, ) from astrbot.core.auth.models import AuthContext, Resource, Subject from astrbot.core.platform.route_identity import PlatformRouteIdentity @@ -211,3 +214,78 @@ async def translate(_event, key, **_kwargs): await commands.watch(event, "target:GroupMessage:room") await commands.connect(event, "target:GroupMessage:room") assert replies == ["session.watch.occupied", "session.connect.occupied"] + + +def test_parse_watch_spec_rejects_filter_flags(): + current = "source:FriendMessage:sender" + with pytest.raises(ValueError, match="Invalid watch arguments"): + parse_watch_spec("target:GroupMessage:room match text hello", current) + + +def test_parse_filter_spec_show_append_and_clear(): + assert parse_filter_spec("abcdef123456") == FilterSpec("abcdef123456", "show") + assert parse_filter_spec("abcdef123456 clear") == FilterSpec( + "abcdef123456", "clear", "all" + ) + assert parse_filter_spec("abcdef123456 clear match") == FilterSpec( + "abcdef123456", "clear", "match" + ) + assert parse_filter_spec("abcdef123456 match text hello world") == FilterSpec( + "abcdef123456", "append", "match", "text", "hello world" + ) + assert parse_filter_spec("abcdef123456 except role member") == FilterSpec( + "abcdef123456", "append", "except", "roles", "member" + ) + with pytest.raises(ValueError): + parse_filter_spec("abcdef123456 match role member extra") + with pytest.raises(ValueError): + parse_filter_spec("ABCDEF123456") + + +@pytest.mark.asyncio +async def test_session_commands_filter_show_append_and_clear(): + replies: list[str] = [] + + async def append_filter(_event, _rule_id, _side, _dimension, value): + if value == "admin": + raise ValueError("Invalid filter role") + return ({"text": ["hello world"]}, {}) + + manager = SimpleNamespace( + get_filter=AsyncMock(return_value=({}, {})), + append_filter=AsyncMock(side_effect=append_filter), + clear_filter=AsyncMock(return_value=({}, {})), + ) + + async def translate(_event, key, **_kwargs): + replies.append(key) + return key + + context = SimpleNamespace( + bridges=SimpleNamespace(_manager=manager), + i18n=SimpleNamespace(t=translate), + ) + commands = SessionCommands(context) + event = _event() + await commands.filter_rule(event, "abcdef123456") + await commands.filter_rule(event, "abcdef123456 match text hello world") + await commands.filter_rule(event, "abcdef123456 clear") + await commands.filter_rule(event, "not-an-id") + await commands.filter_rule(event, "abcdef123456 match role admin") + assert replies == [ + "session.filter.empty", + "session.filter.empty", + "session.filter.body", + "session.filter.empty", + "session.filter.updated", + "session.filter.empty", + "session.filter.empty", + "session.filter.cleared", + "session.filter.usage", + "session.filter.invalid_role", + ] + manager.get_filter.assert_awaited_once_with(event, "abcdef123456") + manager.append_filter.assert_any_await( + event, "abcdef123456", "match", "text", "hello world" + ) + manager.clear_filter.assert_awaited_once_with(event, "abcdef123456", "all") From 7e324c573914e2a8536a0d841c8e28b42c81353e Mon Sep 17 00:00:00 2001 From: BegoniaHe Date: Tue, 15 Sep 2026 05:55:19 +0200 Subject: [PATCH 2/3] fix(session): keep filter writes atomic Re-read the stored edge under the lock before committing match/except, and treat a missing store update as a miss. AI-Generated: true Generated-At: 2026-09-15T03:55:12Z --- astrbot/core/platform/session_bridge.py | 72 ++++++++++++++----------- tests/unit/test_message_protocol.py | 7 +++ tests/unit/test_session_bridge.py | 33 ++++++++++++ tests/unit/test_session_commands.py | 16 +++++- 4 files changed, 94 insertions(+), 34 deletions(-) diff --git a/astrbot/core/platform/session_bridge.py b/astrbot/core/platform/session_bridge.py index 948857062b..c91a0bcea5 100644 --- a/astrbot/core/platform/session_bridge.py +++ b/astrbot/core/platform/session_bridge.py @@ -34,6 +34,7 @@ from .message_session import MessageSession from .send_result import DeliveryAttempt, DeliveryReceipt, PlatformSendResult from .session_bridge_filter import ( + FILTER_SIDES, append_filter_value, coerce_filter_side, compact_filter_side, @@ -404,35 +405,41 @@ async def append_filter( ) -> tuple[dict, dict] | None: """Append one match or except value onto a directed edge.""" stored = validate_filter_value(dimension, value) - row = await self._accessible_rule(event, rule_id) - if row is None: - return None - match = coerce_filter_side(row.match) - except_ = coerce_filter_side(row.except_) - if side == "match": - match = append_filter_value(match, dimension, stored) - elif side == "except": - except_ = append_filter_value(except_, dimension, stored) - else: + if side not in FILTER_SIDES: raise ValueError("Invalid filter arguments") - return await self._persist_filters(row, match, except_) + if await self._accessible_rule(event, rule_id) is None: + return None + async with self._lock: + fresh = await self._state.stored_rule(rule_id) + if fresh is None: + return None + match = coerce_filter_side(fresh.match) + except_ = coerce_filter_side(fresh.except_) + if side == "match": + match = append_filter_value(match, dimension, stored) + else: + except_ = append_filter_value(except_, dimension, stored) + return await self._commit_filters(fresh, match, except_) async def clear_filter( self, event: AstrMessageEvent, rule_id: str, side: str ) -> tuple[dict, dict] | None: """Clear match, except, or both sides of one edge.""" - row = await self._accessible_rule(event, rule_id) - if row is None: - return None - match = coerce_filter_side(row.match) - except_ = coerce_filter_side(row.except_) - if side in {"match", "all"}: - match = {} - if side in {"except", "all"}: - except_ = {} - if side not in {"match", "except", "all"}: + if side not in {*FILTER_SIDES, "all"}: raise ValueError("Invalid filter arguments") - return await self._persist_filters(row, match, except_) + if await self._accessible_rule(event, rule_id) is None: + return None + async with self._lock: + fresh = await self._state.stored_rule(rule_id) + if fresh is None: + return None + match = coerce_filter_side(fresh.match) + except_ = coerce_filter_side(fresh.except_) + if side in {"match", "all"}: + match = {} + if side in {"except", "all"}: + except_ = {} + return await self._commit_filters(fresh, match, except_) async def _accessible_rule( self, event: AstrMessageEvent, rule_id: str @@ -453,22 +460,23 @@ async def _accessible_rule( raise PermissionError("Session operation is not authorized") return stored - async def _persist_filters( + async def _commit_filters( self, row: SessionBridgeRule, match: dict, except_: dict, - ) -> tuple[dict, dict]: + ) -> tuple[dict, dict] | None: + """Write compacted filters. Caller must hold ``self._lock``.""" match = compact_filter_side(match) except_ = compact_filter_side(except_) - async with self._lock: - updated = await self._state.persist_filters(row.rule_id, match, except_) - stored = updated or row - grant = self._state.refresh_grant_filters(stored, match, except_) - if grant is not None: - self._state.arm_expiry( - self._state.store_key(grant.watch), grant, self._expire_watch - ) + updated = await self._state.persist_filters(row.rule_id, match, except_) + if updated is None: + return None + grant = self._state.refresh_grant_filters(updated, match, except_) + if grant is not None: + self._state.arm_expiry( + self._state.store_key(grant.watch), grant, self._expire_watch + ) return match, except_ def _watch_from_row(self, row: SessionBridgeRule) -> SessionWatch: diff --git a/tests/unit/test_message_protocol.py b/tests/unit/test_message_protocol.py index 480698a120..bd068ce382 100644 --- a/tests/unit/test_message_protocol.py +++ b/tests/unit/test_message_protocol.py @@ -751,6 +751,13 @@ def test_projection_bounds_forward_nesting_and_component_count(): ) +def test_envelope_from_event_records_bot_account_id(): + from astrbot.core.platform.message_projection import envelope_from_event + + envelope = envelope_from_event(_event()) + assert envelope.metadata["bot_account_id"] == "bot" + + def test_projection_unknown_component_keeps_native_payload_and_safe_fallback(): from astrbot.core.message.components import Poke from astrbot.core.platform.message_projection import envelope_from_event diff --git a/tests/unit/test_session_bridge.py b/tests/unit/test_session_bridge.py index 5928fbca4f..45fe15dca7 100644 --- a/tests/unit/test_session_bridge.py +++ b/tests/unit/test_session_bridge.py @@ -1459,3 +1459,36 @@ async def test_filter_allows_creator_after_revoke_and_scopes_instance_operator() with pytest.raises(PermissionError): await manager.append_filter(owner, "foreignwatch01", "match", "text", "no") await manager.terminate() + + +@pytest.mark.asyncio +async def test_append_filter_rereads_and_returns_none_on_persist_miss(): + store = FakeSessionBridgeStore() + manager, _, _ = _manager(store=store) + event = _event() + watch = await manager.watch(event, "target:GroupMessage:room", ttl_seconds=60) + original_get = store.get_session_bridge_rule + reads = {"n": 0} + + async def get_after_concurrent_write(rule_id: str): + reads["n"] += 1 + row = await original_get(rule_id) + if reads["n"] == 2 and row is not None: + row.match = {"text": ["alpha"]} + return row + + store.get_session_bridge_rule = get_after_concurrent_write + result = await manager.append_filter(event, watch.rule_id, "match", "text", "beta") + assert result == ({"text": ["alpha", "beta"]}, {}) + + async def miss(_rule_id: str, **_kwargs): + return None + + store.update_session_bridge_rule = miss + assert ( + await manager.append_filter(event, watch.rule_id, "match", "text", "gamma") + is None + ) + live = await manager.get_filter(event, watch.rule_id) + assert live == ({"text": ["alpha", "beta"]}, {}) + await manager.terminate() diff --git a/tests/unit/test_session_commands.py b/tests/unit/test_session_commands.py index ab8951fe87..05cd5ffe87 100644 --- a/tests/unit/test_session_commands.py +++ b/tests/unit/test_session_commands.py @@ -246,13 +246,20 @@ def test_parse_filter_spec_show_append_and_clear(): async def test_session_commands_filter_show_append_and_clear(): replies: list[str] = [] + async def get_filter(_event, rule_id): + if rule_id == "deadbeef0001": + return None + return ({}, {}) + async def append_filter(_event, _rule_id, _side, _dimension, value): if value == "admin": raise ValueError("Invalid filter role") + if value == "(": + raise ValueError("Invalid filter pattern") return ({"text": ["hello world"]}, {}) manager = SimpleNamespace( - get_filter=AsyncMock(return_value=({}, {})), + get_filter=AsyncMock(side_effect=get_filter), append_filter=AsyncMock(side_effect=append_filter), clear_filter=AsyncMock(return_value=({}, {})), ) @@ -272,6 +279,8 @@ async def translate(_event, key, **_kwargs): await commands.filter_rule(event, "abcdef123456 clear") await commands.filter_rule(event, "not-an-id") await commands.filter_rule(event, "abcdef123456 match role admin") + await commands.filter_rule(event, "abcdef123456 match text (") + await commands.filter_rule(event, "deadbeef0001") assert replies == [ "session.filter.empty", "session.filter.empty", @@ -283,8 +292,11 @@ async def translate(_event, key, **_kwargs): "session.filter.cleared", "session.filter.usage", "session.filter.invalid_role", + "session.filter.invalid_pattern", + "session.filter.missing", ] - manager.get_filter.assert_awaited_once_with(event, "abcdef123456") + manager.get_filter.assert_any_await(event, "abcdef123456") + manager.get_filter.assert_any_await(event, "deadbeef0001") manager.append_filter.assert_any_await( event, "abcdef123456", "match", "text", "hello world" ) From e5948ac2526fb2a26f93f34b4808cb8ff0b1c95f Mon Sep 17 00:00:00 2001 From: BegoniaHe Date: Tue, 15 Sep 2026 06:04:36 +0200 Subject: [PATCH 3/3] test(session): restore adapter self_id in filter tests Pass get_self_id through the session-bridge test manager so subject filters still rebuild after rebasing onto pair. AI-Generated: true Generated-At: 2026-09-15T04:04:31Z --- tests/unit/test_session_bridge.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/unit/test_session_bridge.py b/tests/unit/test_session_bridge.py index 45fe15dca7..09efa4706e 100644 --- a/tests/unit/test_session_bridge.py +++ b/tests/unit/test_session_bridge.py @@ -217,6 +217,7 @@ def _manager( lambda _: MessageDeliveryCapabilities(quote=True, media=frozenset({"image"})), authorization=authorization, get_config_id=get_config_id or (lambda _: "default"), + get_self_id=lambda _: "bot", store=store or FakeSessionBridgeStore(), max_watches_per_subject=max_watches_per_subject, max_pairs_per_subject=max_pairs_per_subject,