diff --git a/README.md b/README.md index 51988a0..be273c8 100644 --- a/README.md +++ b/README.md @@ -41,7 +41,7 @@ asyncio.run(channel.connect()) - [Quickstart](docs/quickstart.md) - [Migration from `lark_oapi.channel`](docs/migration-from-lark-oapi.md) -- [API reference](docs/reference.md) +- [API reference](docs/reference.md) — including [Bot-at-bot](docs/reference.md#bot-at-bot) (multi-bot collaboration: sender type, roster, reply threading, `@`-by-name, loop guard) - [Security configuration](docs/security.md) - [Markdown messages](docs/markdown.md) - [Webhook server adapter](docs/webhook-server.md) diff --git a/README.zh.md b/README.zh.md index a60a3e4..6c4fdfb 100644 --- a/README.zh.md +++ b/README.zh.md @@ -40,7 +40,7 @@ asyncio.run(channel.connect()) - [快速开始](docs/quickstart.md) - [从 `lark_oapi.channel` 迁移](docs/migration-from-lark-oapi.md) -- [API 参考](docs/reference.md) +- [API 参考](docs/reference.md) —— 含 [Bot-at-bot](docs/reference.md#bot-at-bot)（多 bot 协作：发送方类型、群成员 roster、回复话题跟随、按名字 `@`、死循环守卫） - [安全配置](docs/security.md) - [Markdown 消息](docs/markdown.md) - [Webhook 服务适配](docs/webhook-server.md) diff --git a/docs/reference.md b/docs/reference.md index 19e6ea4..9182fa0 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -150,13 +150,16 @@ Snake-case aliases such as `card_action`, `bot_added`, `bot_leave`, and | `sender` | `Identity` for the sender | | `sender_id` | Shortcut for `sender.open_id` | | `sender_name` | Optional display name | +| `sender_type` | Raw sender kind (`user` / `bot` / `system` / `anonymous` / `app`), or `None` when the event omits it (see [Bot-at-bot](#bot-at-bot)) | +| `sender_is_bot` | `True` when the sender is a bot/app (`sender.is_bot`) | | `mentions` | List of `Mention` objects | | `mentioned_all` | Whether the message mentioned all members | | `mentioned_bot` | Whether the message mentioned the bot | | `reply_to_message_id` | Parent message id when present | | `content` | Typed `MessageContent` dataclass | -| `content_text` | Flattened markdown/XML-style text | +| `content_text` | Flattened markdown/XML-style text (unchanged — keeps rendered mentions, incl. the bot's own) | | `safe_content_text` | Escaped flattened text for security-sensitive rendering | +| `body_text` | `content_text` with the current bot's own `@`-mention removed — for command parsing / bare-`@` wake detection; equals `content_text` when the bot isn't mentioned (see [Bot-at-bot](#bot-at-bot)) | | `resources` | Resource descriptors for download | | `raw_content_type` | Original Feishu message type | | `raw` | Original event payload | @@ -378,6 +381,92 @@ Known `FeishuChannelErrorCode` values: | `not_connected` | Transport is not connected or startup failed | | `unknown` | Uncategorized upstream or SDK error | +## Bot-at-bot + +Support for multiple bots collaborating in one chat — agents `@`-ing each other +to hand off work. Everything here is **opt-in and additive**; default behavior +is unchanged. + +**Know who sent it.** Every inbound message carries `sender_type` +(`user` / `bot` / …) and the convenience `sender_is_bot`, so an agent can tell a +human, itself, and another bot apart. Get the bot's own identity for its system +prompt with `channel.get_bot_identity()` → `BotIdentity(open_id, name, …)` +(raises `FeishuChannelError(code=not_connected)` before `connect()`). Set +`resolve_sender_names=True` to fill `sender_name` from the chat roster. + +**Receiving events from other bots.** Feishu does **not** deliver "another bot +`@`-ed me" events unless the app has the +`im:message.group_at_msg.include_bot:readonly` permission enabled (distinct from +`im:message.group_at_msg:readonly`, which only covers *user* mentions) — and the +failure is silent. There is no API to self-check this; if bot-to-bot mentions +never arrive, verify that permission first (see the Feishu open-platform +"receive message events" documentation). An `@`-only ping (no body) still wakes +the bot: `mentioned_bot` is `True`. +`content_text` still renders the mention; use `body_text` (the bot's own mention +removed) to detect a bare poke: `msg.mentioned_bot and not msg.body_text.strip()`. + +**Roster.** + +| Method | Return | Notes | +|---|---|---| +| `await channel.get_chat_members(chat_id, *, page_size=100, max_pages=10, id_type="open_id", force=False)` | `list[ChatMember]` | A chat's **users** (Feishu filters bots out, so `is_bot` is never `True`); paginated + cached. `force` refetches | +| `await channel.get_chat_bots(chat_id, *, force=False)` | `list[ChatMember]` | The chat's **bots** (`is_bot=True`); cached; seeds the roster so a bot is `@`-able by name without first appearing in a mention | +| `channel.get_bot_identity()` | `BotIdentity` | This bot's own identity; raises `not_connected` before `connect()` | + +`ChatMember` = `id`, `id_type`, `name`, `tenant_key`, `is_bot`. Provide a +`resolve_chat_members` config hook (`(chat_id) -> list[ChatMember] | None`, sync +or async) to source the roster from your own directory instead of the API. +**Both `None` and an empty list `[]` fall back to the API** — the hook only +overrides the roster when it returns a non-empty list +(return `None` to fall back to the API). + +**Replying to the right place.** `await channel.reply(msg, message, opts=None)` +replies to `msg` and follows its shape: `reply_to` defaults to `msg.message_id` +and `reply_in_thread` defaults to whether the trigger was in a topic thread, so +a reply stays in a thread when the message was in one and stays flat when it +wasn't. `opts` overrides either default; `reply()` never promotes a flat message +into a thread on its own. + +**`@`-mentioning by name.** Either pass structured mentions — a name-only +`Identity(open_id="", display_name="Alice")` on `OutboundText` / `OutboundPost` +is resolved to an open_id against the chat roster (dropped if it doesn't +resolve) — or set `SendOpts.resolve_mentions_in_text=True` to rewrite `@name` +tokens in a text / markdown body. A name that is unknown **or shared by more +than one member** is left as plain text and never mis-mentioned; for +security-sensitive handoffs, pass an explicit open_id. Roster names come from +`get_chat_members` (users), `get_chat_bots` (bots), and bots observed in earlier +inbound mentions. + +**Restrict who can trigger the bot — by chat, not by sender.** Sender open_ids +are hard to get up front, so enumerating them in `allow_from` is impractical. +Instead allow only specific chats with `group_allowlist=['oc_…']` and require an +`@` with `require_mention=True`. `allow_from` takes **sender ids** +(`ou_…` / user_id / union_id); `group_allowlist` takes **chat ids** (`oc_…`); an +app id (`cli_…`) belongs in neither and logs a warning. + +**Breaking ping-pong loops.** The opt-in `policy.bot_loop_guard` counts only +"another bot `@`-ed me" messages in a sliding window (a human message resets it) +and trips past a threshold: + +```python +from lark_channel import BotLoopGuardConfig, PolicyConfig + +policy = PolicyConfig( + bot_loop_guard=BotLoopGuardConfig( + enabled=True, + window_ms=60_000, # sliding window W + max_bot_mentions=5, # trip at N bot @-mentions in W + scope="chat", # or "chat+sender" + on_trip="reject", # "drop" (default) silently mutes; "reject" emits reject(reason="bot_loop") + ) +) +``` + +The default `on_trip="drop"` silently stops replying (one warning on the first +trip); prefer `"reject"` (emits a `reject` event with `reason="bot_loop"`) when +the app needs to know. This is a heuristic backstop, not a protocol-level +guarantee. + ## Related Documents - [Channel quickstart](./quickstart.md) diff --git a/lark_channel/channel/__init__.py b/lark_channel/channel/__init__.py index 5af8bea..83aa361 100644 --- a/lark_channel/channel/__init__.py +++ b/lark_channel/channel/__init__.py @@ -30,6 +30,7 @@ async def on_message(msg): from .card import CardBuilder, new_card from .channel import FeishuChannel from .config import ( + BotLoopGuardConfig, ChannelConfig, ChatModeCacheConfig, ChatQueueConfig, @@ -115,6 +116,7 @@ async def on_message(msg): CardPayload, CachedResource, ChatInfo, + ChatMember, CommentContext, CommentTarget, ConnectionSnapshot, @@ -178,6 +180,7 @@ async def on_message(msg): # Entry points "FeishuChannel", # Config + "BotLoopGuardConfig", "ChannelConfig", "ChatModeCacheConfig", "ChatQueueConfig", @@ -275,6 +278,7 @@ async def on_message(msg): "CardPayload", "CachedResource", "ChatInfo", + "ChatMember", "CommentContext", "CommentTarget", "CommentEvent", diff --git a/lark_channel/channel/_coerce.py b/lark_channel/channel/_coerce.py index 94757e5..8972467 100644 --- a/lark_channel/channel/_coerce.py +++ b/lark_channel/channel/_coerce.py @@ -190,6 +190,11 @@ def coerce_send_opts( reply_target_gone=_coerce_reply_target_gone( _dict_get_any(opts, ("replyTargetGone", "reply_target_gone")) ), + resolve_mentions_in_text=bool( + _dict_get_any( + opts, ("resolveMentionsInText", "resolve_mentions_in_text") + ) + ), ) diff --git a/lark_channel/channel/channel.py b/lark_channel/channel/channel.py index 61011e6..2e7765c 100644 --- a/lark_channel/channel/channel.py +++ b/lark_channel/channel/channel.py @@ -38,13 +38,19 @@ import threading import time from collections import OrderedDict -from dataclasses import dataclass +from dataclasses import dataclass, replace from pathlib import Path from typing import Any, Callable, Dict, List, Literal, Mapping, Optional, Set, Union from lark_channel.client import Client -from lark_channel.core.enum import LogLevel +from lark_channel.core.enum import AccessTokenType, HttpMethod, LogLevel +from lark_channel.core.http import Transport from lark_channel.core.log import logger +from lark_channel.core.model import BaseRequest, RequestOption +# Imported as a module (not ``from …auth import verify``) so the tenant-token +# injection is reached via ``_token_auth.verify`` — tests patch it at its source +# (``lark_channel.core.token.auth.verify``), which a local alias would bypass. +from lark_channel.core.token import auth as _token_auth from lark_channel.event.callback.model.p2_card_action_trigger import ( P2CardActionTrigger, P2CardActionTriggerResponse, @@ -60,6 +66,7 @@ from .auth.token_store import InMemoryTokenStore, TokenStore from .auth.uat_runner import require_user_auth from .bot_identity import BotIdentity, fetch_bot_identity +from .chat_member_cache import ChatMemberCache from .chat_mode import ChatModeCache from .comment import CommentPrimitiveClient from .config import ( @@ -73,13 +80,24 @@ UATConfig, ) from .driver import LarkClientDriver -from .errors import FeishuChannelError, FeishuChannelErrorCode, OutboundSendError, SendError +from .errors import ( + FeishuChannelError, + FeishuChannelErrorCode, + OutboundSendError, + SendError, + classify_api_error, + classify_http_status, +) from .identity import IdentityResolver, NameCache from .keepalive import KeepaliveWatchdog from .media_cache import MediaResourceCache from .normalize.comment import normalize_comment from .normalize.dedup import Deduper, InMemoryDedupStore from .normalize.pipeline import InboundPipeline, PipelineConfig, PipelineDeps +from .outbound.markdown.resolve_mentions import ( + resolve_mentions_in_text, + resolve_name_mentions, +) from .outbound.routing import infer_receive_id_type from .outbound.sender import OutboundSender from .outbound.streaming.card_stream import CardStreamController @@ -94,6 +112,7 @@ CardActionEvent, CardActionPayload, ChatInfo, + ChatMember, CommentContext, CommentTarget, ConnectionSnapshot, @@ -221,6 +240,15 @@ def _normalize_fetched_message_item(item: Dict[str, Any]) -> Dict[str, Any]: return msg +def _status_error_code(status: Optional[int]) -> "FeishuChannelErrorCode": + """Map an HTTP status to a channel error code, defaulting to UNKNOWN when + the status is absent or a 2xx (a malformed/unexpected body on a 2xx isn't an + HTTP-level failure).""" + if status is not None and status >= 400: + return classify_http_status(status) + return FeishuChannelErrorCode.UNKNOWN + + def _extract_fetched_sender(item: Dict[str, Any]) -> Any: sender = item.get("sender") if sender is not None: @@ -400,6 +428,14 @@ def __init__( cache=NameCache(cfg.inbound.name_cache), ) self._chat_mode_cache = ChatModeCache(cfg.chat_mode_cache) + # Per-chat roster (users + bots + observed mentions), shared by + # get_chat_members/get_chat_bots, sender_name resolution and the + # outbound "@name → open_id" normalization. + self._chat_member_cache = ChatMemberCache() + # In-flight roster fetches, keyed by (kind, chat_id, id_type), so a burst + # of cold requests for the same roster collapses to one upstream call + # (also limits concurrent hits on the shared token cache). + self._roster_inflight: Dict[str, "asyncio.Future"] = {} self._media_cache = MediaResourceCache(cfg.media_cache) self._token_store: TokenStore = token_store or InMemoryTokenStore() @@ -572,6 +608,25 @@ def bot_identity(self) -> Optional[BotIdentity]: with self._bot_identity_lock: return self._bot_identity + def get_bot_identity(self) -> BotIdentity: + """This bot's own identity (:class:`BotIdentity`) — e.g. to inline into + an agent's system prompt so it can tell itself apart from other bots and + decide whom to reply to. + + Resolved during :meth:`connect`. Raises + :class:`FeishuChannelError` with code ``not_connected`` if called before + the identity is available, rather than returning ``None``, so callers + don't silently build a prompt with a missing identity. Use the + :attr:`bot_identity` property when a ``None`` sentinel is preferred. + """ + identity = self.bot_identity + if identity is None: + raise FeishuChannelError( + FeishuChannelErrorCode.NOT_CONNECTED, + "bot identity not resolved yet — call connect() first", + ) + return identity + @property def config(self) -> ChannelConfig: return self._config @@ -1451,6 +1506,10 @@ async def _handle_message_event(self, data: Any) -> None: ) if inbound is None: return + # NB: roster collection and sender-name resolution happen only AFTER + # the safety pipeline admits the message (in _dispatch_inbound_to_user), + # so a policy/stale/dedup/self-sent-rejected message can neither write + # the roster nor trigger the external members API. if self._safety is not None: await self._safety.push_message(inbound) else: @@ -1458,7 +1517,50 @@ async def _handle_message_event(self, data: Any) -> None: except Exception as e: logger.exception("FeishuChannel.handle_message_event failed: %s", e) + def _collect_mentions_into_roster(self, inbound) -> None: + chat_id = inbound.conversation.chat_id if inbound.conversation else None + if not chat_id: + return + seen = [ + ChatMember(id=m.open_id, name=m.name, is_bot=m.is_bot) + for m in inbound.mentions + if m.open_id and m.name + ] + if seen: + self._chat_member_cache.set_members(chat_id, seen, "mention") + + async def _resolve_sender_name_from_roster(self, inbound) -> None: + """Warm the chat roster and fill ``sender.display_name`` from it when the + pipeline's own name resolution didn't. Best-effort — a roster miss or API + failure degrades to leaving the name unset, never raises into dispatch.""" + chat_id = inbound.conversation.chat_id if inbound.conversation else None + open_id = inbound.sender.open_id if inbound.sender else None + if not chat_id or not open_id: + return + try: + await self.get_chat_members(chat_id) + # A bot sender is NOT in the users list — warm the bots roster too so + # a bot-to-bot handoff resolves the sender's name (the core + # bot-at-bot case). Cached + singleflight; failure degrades below. + if inbound.sender.is_bot: + await self.get_chat_bots(chat_id) + except Exception as e: # pragma: no cover - defensive + logger.debug("channel: sender-name roster warm failed: %s", e) + return + if not inbound.sender.display_name: + name = self._chat_member_cache.resolve_name(chat_id, open_id) + if name: + inbound.sender.display_name = name + async def _dispatch_inbound_to_user(self, inbound) -> None: + # Runs only for admitted messages (the safety pipeline's on_message, or + # the direct path when no safety pipeline). Seed the roster from observed + # mentions (incl. bots, which the members API filters out) so a bot that + # has "shown its face" can later be @'d by name; source 'mention' never + # overwrites authoritative 'api' names. Then optionally fill sender_name. + self._collect_mentions_into_roster(inbound) + if self._config.resolve_sender_names: + await self._resolve_sender_name_from_roster(inbound) await self._invoke("message", inbound) def _emit_reject(self, event: RejectEvent) -> None: @@ -1835,6 +1937,7 @@ async def send(self, to, message, opts=None) -> SendResult: outbound = _coerce.coerce_outbound(message) send_opts = _coerce.coerce_send_opts(opts) rit = send_opts.receive_id_type or infer_receive_id_type(to) + outbound = self._resolve_outbound_mentions(to, outbound, send_opts) result = await self._sender.send( outbound, receive_id=to, @@ -1857,6 +1960,57 @@ async def send(self, to, message, opts=None) -> SendResult: ) return result + def _resolve_outbound_mentions(self, to, outbound, opts): + """Resolve "@name" into real mentions against the target chat's roster + before sending: fill open_id on name-only structured mentions, and — when + ``opts.resolve_mentions_in_text`` — rewrite ``@name`` tokens in a + text / markdown body. Both no-op when there's nothing to resolve (or when + ``to`` isn't a chat with a roster), so the default send path is untouched. + """ + def lookup(name: str) -> Optional[str]: + return self._chat_member_cache.resolve_open_id(to, name) + + if isinstance(outbound, (OutboundText, OutboundPost)) and outbound.mentions: + outbound = replace( + outbound, mentions=resolve_name_mentions(outbound.mentions, lookup) + ) + + if opts.resolve_mentions_in_text: + if isinstance(outbound, OutboundText) and outbound.text: + outbound = replace( + outbound, text=resolve_mentions_in_text(outbound.text, lookup) + ) + elif isinstance(outbound, OutboundPost) and outbound.markdown: + outbound = replace( + outbound, markdown=resolve_mentions_in_text(outbound.markdown, lookup) + ) + + return outbound + + async def reply(self, msg, message, opts=None) -> SendResult: + """Reply to a received message, following the triggering message's shape. + + Defaults ``reply_to`` to ``msg.message_id`` and — when the trigger was + inside a topic thread (``msg.conversation.thread_id`` present) — defaults + ``reply_in_thread`` to ``True`` so the reply lands back in that thread. + A flat message stays flat: ``reply`` only *follows* the trigger, it never + promotes a flat message into a thread. ``opts`` overrides either default. + Semantically a :meth:`send`. + """ + base = _coerce.coerce_send_opts(opts) + conversation = getattr(msg, "conversation", None) + thread_id = getattr(conversation, "thread_id", None) + resolved = replace( + base, + reply_to=base.reply_to or msg.message_id, + reply_in_thread=( + base.reply_in_thread + if base.reply_in_thread is not None + else bool(thread_id) + ), + ) + return await self.send(msg.chat_id, message, resolved) + async def _forward_outbound_error(self, err: Any) -> None: """Fan-out a send/stream failure to any ``on("error", ...)`` handlers. @@ -2313,6 +2467,320 @@ async def get_chat_mode(self, chat_id: str) -> Optional[str]: return mode return self._config.chat_mode_cache.fallback + # ------------------------------------------------------------------ + # Chat roster (bot-at-bot) + # ------------------------------------------------------------------ + async def get_chat_members( + self, + chat_id: str, + *, + page_size: int = 100, + max_pages: int = 10, + id_type: str = "open_id", + force: bool = False, + ) -> List[ChatMember]: + """List a chat's members, following pagination. + + Returns **users only** — Feishu's chat-members API filters bots out, so + ``is_bot`` is never ``True`` here; use :meth:`get_chat_bots` for the + bots. ``page_size`` is clamped to Feishu's max of 100; ``max_pages`` + (default 10) caps paging. Results are cached per chat and reused by + ``sender_name`` resolution and "@name → open_id"; a second call hits the + cache and ``force`` bypasses it. A ``resolve_chat_members`` config hook, + if provided, overrides the API. Raises :class:`FeishuChannelError` on + API failure. + """ + if max_pages < 1: + raise FeishuChannelError( + FeishuChannelErrorCode.UNKNOWN, + f"get_chat_members: max_pages must be >= 1, got {max_pages}", + ) + # Cache is keyed by id_type: a user_id query must not be satisfied by an + # open_id snapshot (their ChatMember.id values differ). + if not force: + cached = self._chat_member_cache.get_members(chat_id, id_type) + if cached is not None: + return cached + members, complete = await self._roster_singleflight( + f"members::{chat_id}::{id_type}", + lambda: self._fetch_chat_members( + chat_id, page_size=page_size, max_pages=max_pages, id_type=id_type + ), + ) + # Only a COMPLETE snapshot becomes authoritative (cached + feeds the + # name index). A truncated result is returned to this caller but never + # poisons the cache or the name→open_id index — otherwise a partial + # roster could make a shared name look unique and mis-@. + if complete: + self._chat_member_cache.set_members( + chat_id, members, "api", id_type=id_type, complete=True + ) + return members + + @staticmethod + def _hook_takes_id_type(hook: Callable) -> bool: + try: + params = [ + p + for p in inspect.signature(hook).parameters.values() + if p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD) + ] + return len(params) >= 2 + except (ValueError, TypeError): # builtins / C callables without a signature + return False + + async def _call_member_hook( + self, hook: Callable, chat_id: str, id_type: str + ) -> Any: + """Invoke a ``resolve_chat_members`` hook without blocking the event loop. + + Supports both ``hook(chat_id)`` and ``hook(chat_id, id_type)``; a sync + hook runs in an executor, an async hook is awaited — both bounded by a + total timeout so a slow directory can't stall message processing. + """ + timeout = self._config.transport.http_timeout_seconds or 30.0 + takes_id_type = self._hook_takes_id_type(hook) + try: + if inspect.iscoroutinefunction(hook): + coro = hook(chat_id, id_type) if takes_id_type else hook(chat_id) + return await asyncio.wait_for(coro, timeout) + loop = asyncio.get_running_loop() + call = ( + (lambda: hook(chat_id, id_type)) + if takes_id_type + else (lambda: hook(chat_id)) + ) + result = await asyncio.wait_for(loop.run_in_executor(None, call), timeout) + # A sync hook may still hand back an awaitable; await it too. + if inspect.isawaitable(result): + result = await asyncio.wait_for(result, timeout) + return result + except asyncio.TimeoutError as e: + raise FeishuChannelError( + FeishuChannelErrorCode.SEND_TIMEOUT, + f"resolve_chat_members hook timed out after {timeout}s", + cause=e, + ) from e + + async def _fetch_chat_members( + self, chat_id: str, *, page_size: int, max_pages: int, id_type: str + ) -> "tuple[List[ChatMember], bool]": + """Returns ``(members, complete)``. ``complete`` is ``False`` when paging + stopped at ``max_pages`` while the API still reported more members (so a + truncated list is never treated as an authoritative full roster).""" + hook = self._config.resolve_chat_members + if hook is not None: + result = await self._call_member_hook(hook, chat_id, id_type) + # A non-empty roster from the hook wins; ``None`` or an empty list + # both mean "no data from the hook" and fall back to the API. + if result: + for m in result: + if getattr(m, "id_type", id_type) != id_type: + raise FeishuChannelError( + FeishuChannelErrorCode.UNKNOWN, + f"resolve_chat_members returned id_type " + f"{getattr(m, 'id_type', None)!r}; expected {id_type!r}", + ) + return list(result), True + + size = min(max(page_size, 1), 100) + out: List[ChatMember] = [] + page_token: Optional[str] = None + seen_tokens: Set[str] = set() + complete = True + for page in range(max_pages): + queries: List[tuple] = [ + ("member_id_type", id_type), + ("page_size", str(size)), + ] + if page_token: + queries.append(("page_token", page_token)) + data = await self._raw_tenant_get( + "/open-apis/im/v1/chats/:chat_id/members", + {"chat_id": chat_id}, + queries, + ) + for it in data.get("items") or []: + member_id = it.get("member_id") or it.get("open_id") + if not member_id: + continue + out.append( + ChatMember( + id=member_id, + id_type=it.get("member_id_type") or id_type, + name=it.get("name"), + tenant_key=it.get("tenant_key"), + is_bot=False, + ) + ) + next_token = data.get("page_token") + if not data.get("has_more") or not next_token: + break + if next_token in seen_tokens: + logger.warning( + "channel: get_chat_members(%s) repeated page_token — stopping", + chat_id, + ) + complete = False + break + seen_tokens.add(next_token) + page_token = next_token + if page == max_pages - 1: + # About to exit on max_pages while more members remain. + complete = False + logger.warning( + "channel: get_chat_members(%s) truncated at max_pages=%d " + "(has_more still true) — result cached as incomplete", + chat_id, + max_pages, + ) + return out, complete + + async def get_chat_bots( + self, chat_id: str, *, force: bool = False + ) -> List[ChatMember]: + """List the **bots** in a chat — the companion to :meth:`get_chat_members`, + which returns users only. Returns :class:`ChatMember`\\ s with + ``is_bot=True`` and seeds them into the roster so another bot can be + ``@``-ed by name without having appeared in an inbound mention first. + Cached per chat like ``get_chat_members`` (``force`` bypasses). Raises + :class:`FeishuChannelError` on API failure. + """ + if not force: + cached = self._chat_member_cache.get_bots(chat_id) + if cached is not None: + return cached + bots = await self._roster_singleflight( + f"bots::{chat_id}", lambda: self._fetch_chat_bots(chat_id) + ) + self._chat_member_cache.set_bots(chat_id, bots) + return bots + + async def _roster_singleflight(self, key: str, factory: Callable) -> Any: + """Collapse concurrent cold roster fetches for the same key into one + upstream call. Same-loop cooperative: the check-and-register below has no + await between it, so racers within a loop share the first future.""" + existing = self._roster_inflight.get(key) + if existing is not None: + return await existing + loop = asyncio.get_running_loop() + fut = loop.create_future() + self._roster_inflight[key] = fut + try: + fut.set_result(await factory()) + except BaseException as e: + fut.set_exception(e) + finally: + self._roster_inflight.pop(key, None) + return await fut + + async def _fetch_chat_bots(self, chat_id: str) -> List[ChatMember]: + data = await self._raw_tenant_get( + "/open-apis/im/v1/chats/:chat_id/members/bots", + {"chat_id": chat_id}, + [], + ) + out: List[ChatMember] = [] + for it in data.get("items") or []: + bot_id = it.get("bot_id") + if not bot_id: + continue + out.append( + ChatMember( + id=bot_id, + id_type="open_id", + name=it.get("bot_name"), + is_bot=True, + ) + ) + return out + + async def _raw_tenant_get( + self, uri: str, paths: Dict[str, str], queries: List[tuple] + ) -> Dict[str, Any]: + """GET a tenant-scoped endpoint that has no generated SDK resource + (mirrors :mod:`.bot_identity`). ``chat_id`` and other caller-supplied + values go through ``req.paths`` so ``Transport._build_url`` URL-encodes + them — never f-stringed raw into the URI (path-traversal guard). + """ + req = BaseRequest() + req.http_method = HttpMethod.GET + req.uri = uri + req.paths = dict(paths) + req.queries = list(queries) + req.token_types = {AccessTokenType.TENANT} + option = RequestOption() + timeout = self._config.transport.http_timeout_seconds or 30.0 + try: + # Token verify may fetch/refresh the tenant token synchronously; run + # it off the event loop so it can't block other message processing, + # and fold token + transport errors into the unified error type. A + # total timeout bounds the whole operation (not just a socket read). + async def _run() -> Any: + loop = asyncio.get_running_loop() + await loop.run_in_executor( + None, _token_auth.verify, self._client.config, req, option + ) + return await Transport.aexecute(self._client.config, req, option) + + resp = await asyncio.wait_for(_run(), timeout=timeout) + except FeishuChannelError: + raise + except asyncio.TimeoutError as e: + raise FeishuChannelError( + FeishuChannelErrorCode.SEND_TIMEOUT, + f"chat roster request timed out after {timeout}s", + cause=e, + ) from e + except Exception as e: + raise FeishuChannelError( + FeishuChannelErrorCode.UNKNOWN, + f"chat roster request failed: {e}", + cause=e, + ) from e + return self._parse_tenant_body(resp) + + @staticmethod + def _parse_tenant_body(resp: Any) -> Dict[str, Any]: + if resp is None or getattr(resp, "content", None) is None: + raise FeishuChannelError( + FeishuChannelErrorCode.UNKNOWN, "empty roster response" + ) + status = getattr(resp, "status_code", None) + try: + body = json.loads(resp.content.decode("utf-8")) + except (ValueError, UnicodeDecodeError) as e: + raise FeishuChannelError( + _status_error_code(status), + "malformed roster response", + cause=e, + context={"status": status}, + ) from e + # A non-object top-level JSON (e.g. a bare array) must not reach `.get`. + if not isinstance(body, dict): + raise FeishuChannelError( + _status_error_code(status), + "unexpected roster response shape (expected a JSON object)", + context={"status": status}, + ) + code = body.get("code") + if code: + raise FeishuChannelError( + classify_api_error(code, body.get("msg") or ""), + body.get("msg") or f"roster request returned code {code}", + context={"code": code, "status": status}, + ) + # An HTTP error with no (or zero) business code must not be treated as an + # empty success — surface it as a typed error instead of caching []. + if status is not None and not (200 <= status < 300): + raise FeishuChannelError( + classify_http_status(status), + body.get("msg") or f"roster request failed with HTTP {status}", + context={"status": status}, + ) + data = body.get("data") + return data if isinstance(data, dict) else {} + # ------------------------------------------------------------------ # CardKit preallocation API # ------------------------------------------------------------------ diff --git a/lark_channel/channel/chat_member_cache.py b/lark_channel/channel/chat_member_cache.py new file mode 100644 index 0000000..ea31d22 --- /dev/null +++ b/lark_channel/channel/chat_member_cache.py @@ -0,0 +1,249 @@ +"""Per-chat roster cache backing bot-at-bot name resolution. + +Consumers: `get_chat_members` (source ``'api'`` users, per ``id_type``), +`get_chat_bots` (source ``'api'`` bots), inbound-mention collection (source +``'mention'``, short-lived, can carry bots), ``sender_name`` resolution and +``@name → open_id`` normalization. + +Design (rebuilt-from-sources, thread-safe): + +- Each chat keeps authoritative **users** snapshots **keyed by id_type**, one + **bots** snapshot, plus short-lived **mention** observations. The name↔open_id + indices are rebuilt from these live sources on every read, so a full API + refresh that drops a member (or resolves a name collision) is reflected + immediately. +- Only a **complete**, ``open_id``-typed users snapshot (and the bots snapshot) + feeds the ``name → open_id`` index — a truncated roster is never treated as an + authoritative full membership, and a ``user_id``/``union_id`` snapshot never + contributes (its ids aren't usable in an ````). ``get_members`` is keyed + by ``id_type`` so a ``user_id`` query never returns an ``open_id`` snapshot. +- On a complete API refresh the matching-category mention observations that the + snapshot doesn't contain are dropped (a departed/renamed member can't be + resurrected by a stale observation), and any observation for an open_id the + API already knows is ignored (the API name/mapping is authoritative). + +Two safety invariants: a name mapping to more than one open_id is **ambiguous** +→ ``None`` (never mis-@); snapshots/observations expire after a TTL and the +number of chats is capped. Reads return defensive copies so callers can't mutate +the cached fact source. Clock/TTL/capacity are injectable for deterministic tests. +""" + +import threading +import time +from collections import OrderedDict +from dataclasses import dataclass, field, replace +from typing import Callable, Dict, List, Optional, Tuple + +from .types import ChatMember + +MemberSource = str # "api" | "mention" + +_AMBIGUOUS = object() + +DEFAULT_TTL_SECONDS = 300.0 +DEFAULT_MAX_CHATS = 500 +DEFAULT_MAX_ENTRIES_PER_CHAT = 1000 + + +@dataclass +class _Snapshot: + members: List[ChatMember] + id_type: str + complete: bool + fetched_at: float + + +@dataclass +class _Roster: + api_users: Dict[str, _Snapshot] = field(default_factory=dict) # keyed by id_type + api_bots: Optional[_Snapshot] = None + # open_id -> (name, observed_at, is_bot); short-lived, never overrides api. + mentions: "OrderedDict[str, Tuple[str, float, bool]]" = field(default_factory=OrderedDict) + updated_at: float = 0.0 + + +class ChatMemberCache: + def __init__( + self, + *, + now: Callable[[], float] = time.time, + ttl_seconds: float = DEFAULT_TTL_SECONDS, + max_chats: int = DEFAULT_MAX_CHATS, + max_entries_per_chat: int = DEFAULT_MAX_ENTRIES_PER_CHAT, + mention_ttl_seconds: Optional[float] = None, + ) -> None: + self._now = now + self._ttl = ttl_seconds + self._mention_ttl = mention_ttl_seconds if mention_ttl_seconds is not None else ttl_seconds + self._max_chats = max_chats + self._max_entries = max_entries_per_chat + self._chats: "OrderedDict[str, _Roster]" = OrderedDict() + # Guards the whole read-modify-write of a roster so concurrent inbound + # (background loop) and public API / send calls (possibly other threads) + # can't lose updates or read a half-rebuilt index. + self._lock = threading.Lock() + + # ---- writes -------------------------------------------------------------- + + def set_members( + self, + chat_id: str, + members: List[ChatMember], + source: MemberSource, + *, + id_type: str = "open_id", + complete: bool = True, + ) -> None: + with self._lock: + roster = self._roster_for_write(chat_id) + now = self._now() + if source == "api": + prev = roster.api_users.get(id_type) + # Don't let a truncated refresh clobber a complete snapshot. + if complete or prev is None or not prev.complete: + roster.api_users[id_type] = _Snapshot(list(members), id_type, complete, now) + if complete and id_type == "open_id": + self._reconcile_mentions(roster, {m.id for m in members}, want_bot=False) + else: + for m in members: + if not m.id or not m.name: + continue + roster.mentions.pop(m.id, None) + roster.mentions[m.id] = (m.name, now, bool(m.is_bot)) + self._cap(roster.mentions) + roster.updated_at = now + self._store(chat_id, roster) + + def set_bots(self, chat_id: str, bots: List[ChatMember], *, complete: bool = True) -> None: + with self._lock: + roster = self._roster_for_write(chat_id) + now = self._now() + roster.api_bots = _Snapshot(list(bots), "open_id", complete, now) + if complete: + self._reconcile_mentions(roster, {b.id for b in bots}, want_bot=True) + roster.updated_at = now + self._store(chat_id, roster) + + def _reconcile_mentions(self, roster: _Roster, snapshot_ids: set, want_bot: bool) -> None: + """Drop same-category mention observations the authoritative snapshot + doesn't contain (they left / were renamed), so a complete refresh can + negate stale observations instead of them being resurrected.""" + for open_id in list(roster.mentions): + _name, _ts, is_bot = roster.mentions[open_id] + if is_bot == want_bot and open_id not in snapshot_ids: + del roster.mentions[open_id] + + # ---- reads --------------------------------------------------------------- + + def get_members(self, chat_id: str, id_type: str = "open_id") -> Optional[List[ChatMember]]: + with self._lock: + snap = self._live_users(chat_id, id_type) + return [replace(m) for m in snap.members] if snap is not None else None + + def get_bots(self, chat_id: str) -> Optional[List[ChatMember]]: + with self._lock: + snap = self._live_bots(chat_id) + return [replace(m) for m in snap.members] if snap is not None else None + + def resolve_name(self, chat_id: str, open_id: str) -> Optional[str]: + with self._lock: + by_open_id, _ = self._index(chat_id) + return by_open_id.get(open_id) + + def resolve_open_id(self, chat_id: str, name: str) -> Optional[str]: + with self._lock: + _, by_name = self._index(chat_id) + target = by_name.get(name) + return target if isinstance(target, str) else None + + # ---- internals (call under self._lock) ----------------------------------- + + def _index(self, chat_id: str) -> Tuple[Dict[str, str], Dict[str, object]]: + roster = self._live_roster(chat_id) + by_open_id: Dict[str, str] = {} + by_name: Dict[str, object] = {} + api_ids: set = set() + if roster is None: + return by_open_id, by_name + + def register(open_id: str, name: str, is_api: bool) -> None: + if not open_id or not name: + return + prev_name = by_open_id.get(open_id) + if is_api or open_id not in by_open_id: + if ( + prev_name is not None + and prev_name != name + and by_name.get(prev_name) == open_id + ): + del by_name[prev_name] + by_open_id[open_id] = name + if is_api: + api_ids.add(open_id) + existing = by_name.get(name) + if existing is None: + by_name[name] = open_id + elif existing != open_id: + by_name[name] = _AMBIGUOUS + + now = self._now() + # Only a complete, open_id-typed users snapshot feeds the name index. + su = roster.api_users.get("open_id") + if su and su.complete and now - su.fetched_at <= self._ttl: + for m in su.members: + if m.id and m.name: + register(m.id, m.name, True) + sb = roster.api_bots + if sb and sb.complete and now - sb.fetched_at <= self._ttl: + for m in sb.members: + if m.id and m.name: + register(m.id, m.name, True) + # Short-lived mention observations: ignore any open_id the API already + # knows (api authoritative), and expire stale ones. + for open_id, (name, ts, _is_bot) in list(roster.mentions.items()): + if now - ts > self._mention_ttl: + roster.mentions.pop(open_id, None) + continue + if open_id in api_ids: + continue + register(open_id, name, False) + return by_open_id, by_name + + def _live_users(self, chat_id: str, id_type: str) -> Optional[_Snapshot]: + roster = self._live_roster(chat_id) + if roster is None: + return None + snap = roster.api_users.get(id_type) + if snap is None or self._now() - snap.fetched_at > self._ttl: + return None + return snap + + def _live_bots(self, chat_id: str) -> Optional[_Snapshot]: + roster = self._live_roster(chat_id) + if roster is None or roster.api_bots is None: + return None + if self._now() - roster.api_bots.fetched_at > self._ttl: + return None + return roster.api_bots + + def _roster_for_write(self, chat_id: str) -> _Roster: + return self._live_roster(chat_id) or _Roster(updated_at=self._now()) + + def _live_roster(self, chat_id: str) -> Optional[_Roster]: + roster = self._chats.get(chat_id) + if roster is None: + return None + if self._now() - roster.updated_at > self._ttl: + self._chats.pop(chat_id, None) + return None + return roster + + def _store(self, chat_id: str, roster: _Roster) -> None: + self._chats.pop(chat_id, None) + self._chats[chat_id] = roster + while len(self._chats) > self._max_chats: + self._chats.popitem(last=False) + + def _cap(self, mapping: "OrderedDict") -> None: + while len(mapping) > self._max_entries: + mapping.popitem(last=False) diff --git a/lark_channel/channel/config.py b/lark_channel/channel/config.py index 969ff6b..ce94dda 100644 --- a/lark_channel/channel/config.py +++ b/lark_channel/channel/config.py @@ -107,6 +107,26 @@ class GroupOverride: enabled: Optional[bool] = None +@dataclass +class BotLoopGuardConfig: + """Opt-in heuristic guard against two bots ``@``-ing each other in an + endless ping-pong. Off by default. + + Only "another bot @'d me" messages count; a human message resets the count. + When the count reaches ``max_bot_mentions`` inside the ``window_ms`` sliding + window the guard trips. The default ``on_trip='drop'`` **silently** stops + replying (one warn on the first trip); prefer ``'reject'`` (emits a + ``reject`` event with ``reason='bot_loop'``) when the app needs to know. + A heuristic backstop, not a protocol-level guarantee. + """ + + enabled: bool = False + window_ms: int = 60000 + max_bot_mentions: int = 5 + scope: Literal["chat", "chat+sender"] = "chat" + on_trip: Literal["drop", "reject"] = "drop" + + @dataclass class PolicyConfig: """Admission / routing policy for inbound messages. @@ -122,7 +142,12 @@ class PolicyConfig: under DM ``allowlist``/``blocklist`` modes, matched by ``sender_identity_fields`` - ``group_allowlist`` / ``group_blocklist`` — chat_ids allowed/denied - under group ``allowlist``/``blocklist`` modes + under group ``allowlist``/``blocklist`` modes. ``group_allowlist`` takes + **chat ids** (``oc_…``); paired with ``require_mention`` it scopes the bot + to specific rooms without enumerating every sender — a lightweight + "allowFrom" for bot-at-bot. ``allow_from`` takes **sender ids** (``ou_…`` + / user_id / union_id). An app id (``cli_…``) belongs in neither list and + matches nothing (the SDK logs a warning if it sees one) - ``admins`` — sender identities that bypass every gate (always allowed; required sender list for ``admin_only`` group policy), matched by ``sender_identity_fields`` @@ -144,6 +169,7 @@ class PolicyConfig: default_factory=lambda: ["open_id"] ) group_overrides: Dict[str, GroupOverride] = field(default_factory=dict) + bot_loop_guard: Optional[BotLoopGuardConfig] = None # --------------------------------------------------------------------------- @@ -495,3 +521,20 @@ class ChannelConfig: http_executor: Optional[Callable] = None media_cache: MediaCacheConfig = field(default_factory=MediaCacheConfig) security: SecurityConfig = field(default_factory=SecurityConfig) + + # Bot-at-bot knobs. Appended at the end so existing positional field order + # (guarded by test_security_config) stays stable for positional callers. + # + # ``resolve_sender_names``: populate ``InboundMessage.sender_name`` by + # resolving the sender's display name from the chat's member roster (warmed + # via a cached ``get_chat_members``). Off by default → no extra roster API. + resolve_sender_names: bool = False + # ``resolve_chat_members``: override how ``get_chat_members`` sources a chat's + # roster. When it returns a **non-empty** list, that list is used and the + # Feishu API call is skipped; **both ``None`` and an empty list ``[]`` fall + # back to the API**. The callable takes ``(chat_id)`` or ``(chat_id, id_type)`` + # and may be sync or async (a sync hook runs off the event loop; both are + # bounded by a total timeout); every returned ``ChatMember.id_type`` must + # match the requested ``id_type``. Results flow through the internal roster + # cache. (Typed ``Any`` to avoid an import cycle with ``types``.) + resolve_chat_members: Optional[Callable[..., Any]] = None diff --git a/lark_channel/channel/normalize/converters/post.py b/lark_channel/channel/normalize/converters/post.py index 3479568..863217e 100644 --- a/lark_channel/channel/normalize/converters/post.py +++ b/lark_channel/channel/normalize/converters/post.py @@ -16,6 +16,16 @@ def convert(content: PostContent) -> Tuple[str, List[ResourceDescriptor]]: return md, resources +def convert_body(content: PostContent, drop_open_id: str) -> str: + """Flatten a post to Markdown with the given open_id's ```` mention + removed (both structured ``tag:at`` nodes and inline ````), + preserving the title, formatting, links and everyone else's mentions. + Used to build :attr:`InboundMessage.body_text` for post content.""" + if content.post: + return _post_to_markdown(content.post, drop_open_id=drop_open_id)[0] + return content.text or "" + + def _iter_documents(post: Dict[str, Any]) -> List[Dict[str, Any]]: if not isinstance(post, dict) or not post: return [] @@ -24,7 +34,9 @@ def _iter_documents(post: Dict[str, Any]) -> List[Dict[str, Any]]: return [doc for doc in post.values() if isinstance(doc, dict)] -def _post_to_markdown(post: Dict[str, Any]) -> Tuple[str, List[ResourceDescriptor]]: +def _post_to_markdown( + post: Dict[str, Any], drop_open_id: str = "" +) -> Tuple[str, List[ResourceDescriptor]]: docs = _iter_documents(post) if not docs: return "", [] @@ -63,6 +75,9 @@ def _post_to_markdown(post: Dict[str, Any]) -> Tuple[str, List[ResourceDescripto elif tag == "a": chunks.append(f"[{el.get('text') or ''}]({el.get('href') or ''})") elif tag == "at": + # Drop the current bot's own mention when building body_text. + if drop_open_id and el.get("user_id") == drop_open_id: + continue nm = el.get("user_name") or el.get("user_id") or "" chunks.append(f"@{nm}") elif tag == "emotion": @@ -78,7 +93,7 @@ def _post_to_markdown(post: Dict[str, Any]) -> Tuple[str, List[ResourceDescripto elif tag == "hr": chunks.append("---") elif tag == "md": - text, res = _process_md_text(el.get("text") or "") + text, res = _process_md_text(el.get("text") or "", drop_open_id=drop_open_id) chunks.append(text) resources.extend(res) line = "".join(chunks) @@ -123,7 +138,9 @@ def add(kind: str, key: Any, *, file_name: Any = None) -> None: return resources -def _process_md_text(text: str) -> Tuple[str, List[ResourceDescriptor]]: +def _process_md_text( + text: str, drop_open_id: str = "" +) -> Tuple[str, List[ResourceDescriptor]]: """Post-process raw markdown text from an "md" element. Splits by fenced code block delimiters (```) and only applies @@ -145,6 +162,8 @@ def _process_md_text(text: str) -> Tuple[str, List[ResourceDescriptor]]: def _replace_at(m: re.Match) -> str: user_id = m.group(4) name = m.group(5) + if drop_open_id and user_id == drop_open_id: + return "" # remove the current bot's own inline mention if user_id in ("all", "all_members"): return "@all" return f"@{name}" if name else f"@{user_id}" diff --git a/lark_channel/channel/normalize/mentions.py b/lark_channel/channel/normalize/mentions.py index 73dde1a..6f6cd2c 100644 --- a/lark_channel/channel/normalize/mentions.py +++ b/lark_channel/channel/normalize/mentions.py @@ -183,7 +183,10 @@ def _replace(match: "re.Match[str]") -> str: # user-visible content carries the raw token. result = _AT_ALL_RE.sub(MENTION_ALL_DISPLAY, result) if strip_bot_mentions: - result = re.sub(r"\s{2,}", " ", result).strip() + # Collapse the whitespace left where a bot mention was removed, then + # trim ends. Only horizontal whitespace is collapsed — newlines are + # preserved so a multi-line body that @-ed the bot keeps its structure. + result = re.sub(r"[^\S\n]{2,}", " ", result).strip() return result diff --git a/lark_channel/channel/normalize/pipeline.py b/lark_channel/channel/normalize/pipeline.py index a5949fe..8ca7f60 100644 --- a/lark_channel/channel/normalize/pipeline.py +++ b/lark_channel/channel/normalize/pipeline.py @@ -24,6 +24,7 @@ ReplyRef, TextContent, ) +from .converters.post import convert_body as post_convert_body from .dedup import Deduper from .flatten import flatten from .interactive import fetch_interactive @@ -68,18 +69,22 @@ def _sender_to_identity(sender: Any) -> Identity: """Normalize an event.sender payload (dict or EventSender) to Identity.""" if isinstance(sender, dict): sid = sender.get("sender_id") or {} + sender_type = sender.get("sender_type") return Identity( open_id=sid.get("open_id") or "", union_id=sid.get("union_id"), user_id=sid.get("user_id"), - is_bot=_is_bot_sender_type(sender.get("sender_type")), + is_bot=_is_bot_sender_type(sender_type), + sender_type=sender_type, ) sid = getattr(sender, "sender_id", None) + sender_type = getattr(sender, "sender_type", None) return Identity( open_id=getattr(sid, "open_id", "") or "", union_id=getattr(sid, "union_id", None), user_id=getattr(sid, "user_id", None), - is_bot=_is_bot_sender_type(getattr(sender, "sender_type", None)), + is_bot=_is_bot_sender_type(sender_type), + sender_type=sender_type, ) @@ -190,6 +195,14 @@ async def process( ext = extract_mentions(raw_mentions) mentions: List[Mention] = list(ext.mention_list) mentioned_all = ext.mentioned_all + # Whether this message @-mentions the current bot. Used only to derive + # the bot-mention-stripped `body_text` view below; `content_text` itself + # keeps the rendered mention so default behavior is unchanged. + strip_bot = bool(self._bot_open_id) and self._bot_open_id in ext.mentions_by_open_id + # Placeholder-form primary text captured before resolution, so `body_text` + # can strip the bot's own mention precisely (by placeholder, not by + # rendered-name string matching). + primary_raw_text: Optional[str] = None if isinstance(content, TextContent): # Feishu frequently ships ``@all`` messages with # ``mentions = null`` — the only signal is an ``@_all`` @@ -199,10 +212,12 @@ async def process( # silently skipped. if not mentioned_all and text_has_mention_all(content.text): mentioned_all = True + primary_raw_text = content.text content.text = resolve_mentions(content.text, ext) elif isinstance(content, PostContent): if not mentioned_all and text_has_mention_all(content.text): mentioned_all = True + primary_raw_text = content.text content.text = resolve_mentions(content.text, ext) at_mentions, at_all, stripped = parse_at_tags(content.text) content.text = stripped @@ -296,6 +311,27 @@ async def process( safe_flat_text = _safe_content_text(flat_text) content_text = safe_flat_text if self._cfg.security.strict_content_text else flat_text + # `body_text`: content_text with the CURRENT bot's own @-mention removed. + # content_text above keeps the rendered mention, so callers that don't + # read body_text see unchanged default behavior. Text uses a + # placeholder-precise strip; post drops the bot's node at the AST + # level (preserving title / formatting / other mentions). + body_raw = None + if strip_bot: + if isinstance(content, PostContent): + body_raw = resolve_mentions( + post_convert_body(content, self._bot_open_id), ext + ) + elif isinstance(content, TextContent) and primary_raw_text is not None: + body_raw = resolve_mentions( + primary_raw_text, ext, strip_bot_mentions=True, bot_open_id=self._bot_open_id + ) + if body_raw is not None: + body_safe = _safe_content_text(body_raw) + body_text = body_safe if self._cfg.security.strict_content_text else body_raw + else: + body_text = content_text + return InboundMessage( id=message_id, create_time=_int_or_zero(msg.get("create_time")), @@ -308,6 +344,7 @@ async def process( raw=msg if self._cfg.inbound.include_raw and isinstance(msg, dict) else {}, content_text=content_text, safe_content_text=safe_flat_text, + body_text=body_text, resources=resources, mentioned_bot=mentioned_bot, chat_mode=chat_mode, diff --git a/lark_channel/channel/outbound/markdown/resolve_mentions.py b/lark_channel/channel/outbound/markdown/resolve_mentions.py new file mode 100644 index 0000000..a42c164 --- /dev/null +++ b/lark_channel/channel/outbound/markdown/resolve_mentions.py @@ -0,0 +1,158 @@ +"""``@name → open_id`` normalization + ```` sink hardening. + +Two pure resolvers back the outbound path: + +- :func:`resolve_name_mentions` — fill ``open_id`` on name-only structured + mentions via a roster lookup; drop the ones that don't resolve; keep entries + that already carry an open_id. +- :func:`resolve_mentions_in_text` — rewrite plaintext ``@`` tokens into + real ```` tags when the name resolves; leave unknown / ambiguous ``@xxx`` + untouched (syntax fallback). Matching is Map-lookup based and never compiles a + roster name into a regex, so it stays linear on pathological input (no ReDoS). + +Plus the shared ```` hardening primitives :func:`is_valid_open_id` and +:func:`escape_at_name`, applied wherever a display name reaches an ```` sink. + +``lookup(name) -> Optional[str]`` returns an open_id, or ``None`` for an unknown +or ambiguous name. +""" + +import re +from typing import Callable, List, Optional + +from ...types import Identity + +MentionLookup = Callable[[str], Optional[str]] + +# A well-formed Feishu open_id / union_id for an ````. +_OPEN_ID_RE = re.compile(r"^(ou_|on_)[A-Za-z0-9_-]+$") + + +def is_valid_open_id(open_id: Optional[str]) -> bool: + """Whether ``open_id`` is safe to drop into ````. + + Accepts ``ou_…`` / ``on_…`` ids, plus the everyone-sentinel ``"all"`` — the + SDK renders ``@all`` as ````, so the hardening must not + drop it. Rejects app ids (``cli_…``), markup, and empty values. + """ + if not open_id: + return False + if open_id == "all": + return True + return bool(_OPEN_ID_RE.match(open_id)) + + +def escape_at_name(name: str) -> str: + """Neutralize a display name before it lands in an ```` tag body. + + Display names are attacker-influenced (any group member sets their own) and + Feishu renders the ```` sink without escaping — a name containing ``<``, + ``>`` or ``"`` could inject a second, forged mention in the bot's own voice. + Strip those characters (rather than HTML-encode, which Feishu would render + literally in the plain-text tag body). + """ + return re.sub(r'[<>"]', "", name or "") + + +def resolve_name_mentions( + mentions: List[Identity], lookup: MentionLookup +) -> List[Identity]: + """Fill ``open_id`` on name-only structured mentions from the roster. + + Entries that already carry an open_id pass through untouched; name-only + entries that don't resolve (unknown / ambiguous) are dropped rather than + sent with a wrong or missing id. + """ + out: List[Identity] = [] + for m in mentions: + if m.open_id: + out.append(m) + continue + if not m.display_name: + continue + open_id = lookup(m.display_name) + if open_id: + out.append( + Identity( + open_id=open_id, + union_id=m.union_id, + user_id=m.user_id, + display_name=m.display_name, + is_bot=m.is_bot, + sender_type=m.sender_type, + ) + ) + return out + + +# Longest candidate name to try after an ``@``: bounded words and characters so +# the scan stays linear on hostile input — a 50k-char token yields one bounded +# window, not a quadratic prefix walk. +_MAX_NAME_WORDS = 5 +_MAX_NAME_CHARS = 64 +_TRAILING_PUNCTUATION = re.compile(r"[.,!?;:)\]}]+$") + + +def resolve_mentions_in_text(text: str, lookup: MentionLookup) -> str: + """Rewrite plaintext ``@`` tokens into real ```` tags when the + name resolves against the roster. Unknown or ambiguous names are left + verbatim (syntax fallback). Longest-match-first (so ``@John Smith`` resolves + ahead of ``@John``), using per-candidate Map lookups rather than a + roster-name-derived regex, so it stays linear even on pathological input. + """ + if "@" not in text: + return text + out: List[str] = [] + i = 0 + n = len(text) + while i < n: + at = text.find("@", i) + if at == -1: + out.append(text[i:]) + break + out.append(text[i:at]) + # An ``@`` mid-word (e.g. inside an email ``a@b``) is not a mention. + starts_token = at == 0 or text[at - 1].isspace() + match = _match_name_at(text, at + 1, lookup) if starts_token else None + if match is not None: + name, open_id, length = match + out.append(f'{escape_at_name(name)}') + i = at + 1 + length + else: + out.append("@") + i = at + 1 + return "".join(out) + + +def _match_name_at(text: str, start: int, lookup: MentionLookup): + """Longest resolvable name starting at ``start``, or ``None``.""" + window = text[start : start + _MAX_NAME_CHARS] + for candidate in _candidate_prefixes(window): + direct = lookup(candidate) + if is_valid_open_id(direct): + return candidate, direct, len(candidate) + trimmed = _TRAILING_PUNCTUATION.sub("", candidate) + if trimmed != candidate: + t = lookup(trimmed) + if is_valid_open_id(t): + return trimmed, t, len(trimmed) + return None + + +def _candidate_prefixes(window: str) -> List[str]: + """Word-boundary prefixes of ``window``, longest first (up to MAX_NAME_WORDS).""" + if not window or window[0].isspace(): + return [] + word_ends: List[int] = [] + in_word = False + for k, ch in enumerate(window): + if not ch.isspace(): + in_word = True + elif in_word: + word_ends.append(k) + in_word = False + if len(word_ends) >= _MAX_NAME_WORDS: + break + if in_word and len(word_ends) < _MAX_NAME_WORDS: + word_ends.append(len(window)) + return [window[:end] for end in reversed(word_ends)] diff --git a/lark_channel/channel/outbound/markdown/to_post.py b/lark_channel/channel/outbound/markdown/to_post.py index 6fb8738..5c94761 100644 --- a/lark_channel/channel/outbound/markdown/to_post.py +++ b/lark_channel/channel/outbound/markdown/to_post.py @@ -14,6 +14,7 @@ from typing import Any, Dict, List, Tuple from ...types import Identity +from .resolve_mentions import escape_at_name, is_valid_open_id # Inline marker patterns (order matters: most specific first). # Italic is tricky because `**bold**` also matches `*...*`. We require that @@ -230,13 +231,13 @@ def _flush_paragraph(buf: List[str]) -> None: if mentions and paragraphs: at_runs: List[Dict[str, Any]] = [] for ident in mentions: - if not ident or not ident.open_id: + if not ident or not is_valid_open_id(ident.open_id): continue at_runs.append( { "tag": "at", "user_id": ident.open_id, - "user_name": ident.display_name or "", + "user_name": escape_at_name(ident.display_name or ""), } ) at_runs.append({"tag": "text", "text": " "}) @@ -316,12 +317,12 @@ def _build_native_md_ast( if mentions and rows: at_runs: List[Dict[str, Any]] = [] for ident in mentions: - if not ident or not ident.open_id: + if not ident or not is_valid_open_id(ident.open_id): continue at_runs.append({ "tag": "at", "user_id": ident.open_id, - "user_name": ident.display_name or "", + "user_name": escape_at_name(ident.display_name or ""), }) at_runs.append({"tag": "text", "text": " "}) rows[0] = at_runs + rows[0] diff --git a/lark_channel/channel/outbound/sender.py b/lark_channel/channel/outbound/sender.py index f4ef5ea..e40ff68 100644 --- a/lark_channel/channel/outbound/sender.py +++ b/lark_channel/channel/outbound/sender.py @@ -26,6 +26,7 @@ is_reply_target_gone, ) from .retry import with_retry +from .markdown.resolve_mentions import escape_at_name, is_valid_open_id from .media.uploader import resolve_media_key from ..types import ( OutboundAudio, @@ -52,12 +53,66 @@ _ChunkMode = str # "newline" | "paragraph" | "none" +# A structured `...` mention is atomic — the chunker must never +# split one across a message boundary, or the mention breaks (and both halves +# render broken text). +_AT_CLOSE = "" + + +def _at_spans(text: str) -> list: + """Absolute (start, end) spans of complete ``…`` tags. + + Single-pass / linear-time by design: each character is visited at most once, + so adversarial input with many unclosed ``" / "/"), else it's not + # an opener (e.g. ""); advance past it. + if after and not (after.isspace() or after in (">", "/")): + i = start + 3 + continue + gt = text.find(">", start + 3) + if gt == -1: + break # no more complete tags + close = lowered.find(_AT_CLOSE, gt + 1) + if close == -1: + i = start + 3 # unclosed opener — skip it, keep scanning forward + continue + end = close + len(_AT_CLOSE) + spans.append((start, end)) + i = end + return spans + + +def _protect_cut(cut: int, start: int, n: int, spans: list) -> int: + """Adjust an absolute cut index so it never falls *inside* an ```` span. + If the cut is inside a span, move it to the span start (push the whole tag + to the next chunk); if the span already starts at/before this chunk's start, + emit the whole tag instead (the chunk then exceeds ``limit`` — unavoidable + for a single oversized tag).""" + for s, e in spans: + if s < cut < e: + return s if s > start else min(e, n) + return cut + + def chunk_text(text: str, limit: int = 3500, mode: _ChunkMode = "newline") -> list: """Split ``text`` into ordered chunks of <= ``limit`` chars. - ``newline``: prefers breaking at the last ``\\n`` within the window. - ``paragraph``: prefers breaking at blank-line boundaries. - ``none``: hard slice at ``limit`` chars. + + A ``...`` mention is never split across chunks. """ if not text: return [] @@ -65,30 +120,35 @@ def chunk_text(text: str, limit: int = 3500, mode: _ChunkMode = "newline") -> li return [text] if len(text) <= limit: return [text] - if mode == "none": - return [text[i : i + limit] for i in range(0, len(text), limit)] - if mode == "paragraph": - return _chunk_by_delim(text, limit, delim="\n\n") - return _chunk_by_delim(text, limit, delim="\n") + delim = {"paragraph": "\n\n", "newline": "\n"}.get(mode) # None for "none" + return _chunk_by_delim(text, limit, delim, _at_spans(text)) -def _chunk_by_delim(text: str, limit: int, delim: str) -> list: +def _chunk_by_delim(text: str, limit: int, delim: "Optional[str]", spans: list) -> list: chunks = [] i = 0 n = len(text) - delim_len = len(delim) while i < n: if n - i <= limit: chunks.append(text[i:]) break - window = text[i : i + limit] - idx = window.rfind(delim) - if idx <= 0: - chunks.append(window) - i += limit + idx = text.rfind(delim, i, i + limit) if delim else -1 + if idx <= i: + # No usable delimiter in the window: hard cut at the limit, nudged + # off any tag boundary. + cut = _protect_cut(i + limit, i, n, spans) + chunks.append(text[i:cut]) + i = cut else: - chunks.append(window[:idx]) - i += idx + delim_len + cut = _protect_cut(idx, i, n, spans) + if cut != idx: + # The delimiter break landed inside an tag — use the + # protected boundary and don't consume a delimiter there. + chunks.append(text[i:cut]) + i = cut + else: + chunks.append(text[i:idx]) + i = idx + len(delim) return [c for c in chunks if c] @@ -182,11 +242,13 @@ def _extract_message_id(resp: Dict[str, Any]) -> Optional[str]: def _build_text(msg: OutboundText) -> Dict[str, str]: - # Inject tags for mentioned identities so they get notified. + # Inject tags for mentioned identities so they get notified. A display + # name is attacker-influenced, so it flows through escape_at_name, and only + # well-formed open_ids reach the sink (see resolve_mentions). at_prefix = "" for ident in msg.mentions or []: - if ident and ident.open_id: - name = ident.display_name or "" + if ident and is_valid_open_id(ident.open_id): + name = escape_at_name(ident.display_name or "") at_prefix += f'{name} ' content = at_prefix + (msg.text or "") return {"msg_type": "text", "content": json.dumps({"text": content}, ensure_ascii=False)} @@ -385,9 +447,15 @@ async def send( last_result: SendResult = SendResult.fail(SendError(code=FeishuChannelErrorCode.UNKNOWN, retryable=False)) for idx, body in enumerate(body_list): req_uuid = uuid_ if (idx == 0 and uuid_) else str(uuid.uuid4()) - # Only apply `reply_to` to the first chunk; subsequent chunks are - # fresh messages so they all render in the original chat. - effective_reply_to = reply_to if idx == 0 else None + # For a thread reply, EVERY chunk must reply (with reply_in_thread) + # so the whole message stays in the topic thread — otherwise only the + # first chunk lands in-thread and the rest fall to the main timeline. + # For a flat reply, keep the legacy behavior: only the first chunk + # quote-replies; subsequent chunks are fresh messages. + if reply_in_thread is True: + effective_reply_to = reply_to + else: + effective_reply_to = reply_to if idx == 0 else None result = await self._send_one_with_fallback( body=body, receive_id=receive_id, diff --git a/lark_channel/channel/safety/__init__.py b/lark_channel/channel/safety/__init__.py index 8882647..bcf8307 100644 --- a/lark_channel/channel/safety/__init__.py +++ b/lark_channel/channel/safety/__init__.py @@ -1,6 +1,7 @@ """Safety pipeline — dedup, stale detection, policy, lock, batch+queue.""" from .chat_pipeline import ChatPipeline, ChatPipelineManager, merge_batch +from .loop_guard import LoopGuard from .pipeline import SafetyPipeline from .policy_gate import PolicyDecision, PolicyGate from .processing_lock import ProcessingLock @@ -23,6 +24,7 @@ "ChatQueueConfig", "DEFAULT_STALE_MS", "DedupConfig", + "LoopGuard", "MediaBatchConfig", "PolicyDecision", "PolicyGate", diff --git a/lark_channel/channel/safety/chat_pipeline.py b/lark_channel/channel/safety/chat_pipeline.py index 8c62a95..34b36e7 100644 --- a/lark_channel/channel/safety/chat_pipeline.py +++ b/lark_channel/channel/safety/chat_pipeline.py @@ -64,6 +64,15 @@ def merge_batch(batch: List[InboundMessage]) -> InboundMessage: merged_content = last.content if isinstance(merged_content, TextContent) and texts: merged_content = TextContent(text="\n\n".join(texts), raw=merged_content.raw) + + # Recombine the derived text views from the source messages (in order) — + # otherwise a batched message reverts them to empty and callers relying on + # content_text / body_text (e.g. command parsing, bare-@ wake detection) + # break on the default aggregation path. Each source already carries its own + # correctly-computed views; join the non-empty ones. + def _join(field: str) -> str: + return "\n\n".join(v for v in (getattr(m, field, "") for m in batch) if v) + merged = _IM( id=last.id, create_time=last.create_time, @@ -76,6 +85,12 @@ def merge_batch(batch: List[InboundMessage]) -> InboundMessage: content=merged_content, raw=last.raw, chat_mode=chat_mode, + content_text=_join("content_text"), + safe_content_text=_join("safe_content_text"), + body_text=_join("body_text"), + raw_content_type=last.raw_content_type, + resources=[r for m in batch for r in (m.resources or [])], + batched_sources=list(batch), ) return merged diff --git a/lark_channel/channel/safety/loop_guard.py b/lark_channel/channel/safety/loop_guard.py new file mode 100644 index 0000000..0f1cb26 --- /dev/null +++ b/lark_channel/channel/safety/loop_guard.py @@ -0,0 +1,165 @@ +"""Heuristic guard against two bots ``@``-ing each other forever. + +Opt-in, default off. Only "another bot @'d me" messages count +(``sender.is_bot and mentioned_bot``); a human message (``sender_type == 'user'``) +resets the count; when the count reaches the threshold inside a sliding window +the key is tripped. ``msg.create_time`` is the clock, so counting is +deterministic — but it is event-supplied, hence only a best-effort backstop +(a bot supplying adversarial timestamps can degrade the heuristic). Out-of-order +events use a per-key monotonic clock so a stale event can't reopen a passed +window, and config values are validated so a misconfiguration can't silently +disable the guard (e.g. a threshold above the per-key cap that could never trip). +""" + +import threading +from collections import OrderedDict +from typing import Any, Dict, List, Optional, Tuple + +from ..config import BotLoopGuardConfig + +# Hard cap on tracked keys so the state map can't grow unbounded. +_MAX_KEYS = 5000 +# Hard cap on the per-key window so a malicious bot pinning ``create_time`` to a +# fixed far-future value can't accumulate distinct message_ids without bound. +_MAX_ENTRIES_PER_KEY = 1024 +_DEFAULT_WINDOW_MS = 60000 +_SCOPES = ("chat", "chat+sender") +_ON_TRIP = ("drop", "reject") + + +class LoopGuard: + def __init__(self, cfg: Optional[BotLoopGuardConfig], logger: Any) -> None: + self._logger = logger + # key -> {"entries": List[(message_id, time)], "warned": bool, "clock": int} + self._states: "OrderedDict[str, Dict[str, Any]]" = OrderedDict() + # Guards record / reset / reconfigure so a runtime update_policy can't + # race the hot record() path — reconfigure atomically swaps config + state. + self._lock = threading.Lock() + self._apply(cfg) + + # ---- configuration ------------------------------------------------------- + + def _apply(self, cfg: Optional[BotLoopGuardConfig]) -> None: + cfg = cfg or BotLoopGuardConfig() + self.enabled = bool(cfg.enabled) + self._window_ms = cfg.window_ms if cfg.window_ms and cfg.window_ms > 0 else _DEFAULT_WINDOW_MS + if cfg.window_ms is not None and cfg.window_ms <= 0: + self._warn("botLoopGuard.window_ms must be > 0; using %d", _DEFAULT_WINDOW_MS) + threshold = cfg.max_bot_mentions + if threshold < 1: + self._warn("botLoopGuard.max_bot_mentions must be >= 1; clamping to 1") + threshold = 1 + elif threshold > _MAX_ENTRIES_PER_KEY: + # A threshold above the per-key entry cap could never be reached. + self._warn( + "botLoopGuard.max_bot_mentions %d exceeds the per-key cap %d and " + "could never trip; clamping", + threshold, + _MAX_ENTRIES_PER_KEY, + ) + threshold = _MAX_ENTRIES_PER_KEY + self._threshold = threshold + self._scope = cfg.scope if cfg.scope in _SCOPES else "chat" + self.on_trip = cfg.on_trip if cfg.on_trip in _ON_TRIP else "drop" + + def reconfigure(self, cfg: Optional[BotLoopGuardConfig]) -> None: + """Atomically apply a new config (e.g. from ``update_policy``). Counting + state is preserved when only ``on_trip`` changed, and cleared otherwise + (a new window / threshold / scope, or enable/disable, starts clean).""" + with self._lock: + prev = (self.enabled, self._window_ms, self._threshold, self._scope) + self._apply(cfg) + if prev != (self.enabled, self._window_ms, self._threshold, self._scope): + self._states.clear() + + def reset_on_human(self, msg: Any) -> None: + """Reset the loop counters for a chat when a human speaks — called + BEFORE the policy gate so a plain (possibly no-mention, hence + policy-rejected) human message still breaks a bot ping-pong. Clears the + chat's state in ``chat`` scope, and every ``chat::bot`` key in + ``chat+sender`` scope.""" + if not self.enabled or getattr(msg.sender, "sender_type", None) != "user": + return + chat_id = msg.conversation.chat_id + with self._lock: + if self._scope == "chat+sender": + prefix = f"{chat_id}::" + for key in [k for k in self._states if k == chat_id or k.startswith(prefix)]: + self._states.pop(key, None) + else: + self._states.pop(chat_id, None) + + def _warn(self, fmt: str, *args: Any) -> None: + warn = getattr(self._logger, "warning", None) + if callable(warn): + warn("channel: " + fmt, *args) + + # ---- counting ------------------------------------------------------------ + + def record(self, msg: Any) -> bool: + """Record a message; return whether its key is now tripped. + + A human message resets the key and never trips; messages that aren't + "another bot @'d me" don't count. A re-delivered ``message_id`` already + inside the window is counted once. The first trip of a key warns once. + """ + if not self.enabled: + return False + with self._lock: + return self._record_locked(msg) + + def _record_locked(self, msg: Any) -> bool: + key = self._key_for(msg) + + if msg.sender.sender_type == "user": + self._states.pop(key, None) + return False + if not (msg.sender.is_bot and msg.mentioned_bot): + return False + + state = self._states.get(key) or {"entries": [], "warned": False, "clock": None} + entries: List[Tuple[str, int]] = state["entries"] + # Monotonic per-key clock: an out-of-order (older) event can't reopen an + # already-passed window. + prev_clock = state["clock"] + clock = msg.create_time if prev_clock is None else max(prev_clock, msg.create_time) + cutoff = clock - self._window_ms + entries = [(mid, t) for (mid, t) in entries if t >= cutoff] + # Only count the current message if it falls inside the current window + # (a stale, out-of-order event older than the window does not count and + # must not be grouped with newer/future entries). + if msg.create_time >= cutoff and not any(mid == msg.id for (mid, _t) in entries): + entries.append((msg.id, msg.create_time)) + if len(entries) > _MAX_ENTRIES_PER_KEY: + entries = entries[-_MAX_ENTRIES_PER_KEY:] + state["entries"] = entries + state["clock"] = clock + + tripped = len(entries) >= self._threshold + if tripped and not state["warned"]: + self._logger.warning( + "channel: botLoopGuard tripped for %s — >=%d bot @-mentions " + "within %dms (on_trip=%s)", + key, + self._threshold, + self._window_ms, + self.on_trip, + ) + state["warned"] = True + elif not tripped: + state["warned"] = False + + self._remember(key, state) + return tripped + + def _remember(self, key: str, state: Dict[str, Any]) -> None: + self._states.pop(key, None) + self._states[key] = state + while len(self._states) > _MAX_KEYS: + self._states.popitem(last=False) + + def _key_for(self, msg: Any) -> str: + chat_id = msg.conversation.chat_id + if self._scope == "chat+sender": + return f"{chat_id}::{msg.sender.open_id}" + return chat_id diff --git a/lark_channel/channel/safety/pipeline.py b/lark_channel/channel/safety/pipeline.py index d1909b1..b46b8bb 100644 --- a/lark_channel/channel/safety/pipeline.py +++ b/lark_channel/channel/safety/pipeline.py @@ -21,6 +21,7 @@ from ..config import PolicyConfig from ..types import InboundMessage from .chat_pipeline import ChatPipelineManager +from .loop_guard import LoopGuard from .media_pipeline import MediaPipelineManager from .policy_gate import PolicyGate from .processing_lock import ProcessingLock @@ -73,7 +74,9 @@ def __init__( sweep_seconds=dedup_config.sweep_seconds, ) self._lock = ProcessingLock(ttl_ms=processing_lock_ttl_ms) - self._policy = PolicyGate(policy or PolicyConfig()) + policy_obj = policy or PolicyConfig() + self._policy = PolicyGate(policy_obj) + self._loop_guard = LoopGuard(policy_obj.bot_loop_guard, logger) self._manager = ChatPipelineManager( config=batch_config or TextBatchConfig(), loop=loop, @@ -95,6 +98,11 @@ def set_bot_open_id(self, open_id: Optional[str]) -> None: def update_policy(self, **changes) -> None: self._policy.update_policy(**changes) + # Keep the runtime loop guard in sync — otherwise a runtime + # enable/disable or threshold change would be reflected in get_policy() + # but never actually take effect. + if "bot_loop_guard" in changes: + self._loop_guard.reconfigure(changes["bot_loop_guard"]) def get_policy(self) -> PolicyConfig: return self._policy.get_policy() @@ -157,6 +165,12 @@ async def push_message(self, msg: InboundMessage) -> None: self._emit_reject(msg, "self_sent") return + # 2.9 Human-activity reset for the bot loop guard — runs BEFORE the + # policy gate so a plain human message (which require_mention would + # otherwise reject) still breaks an ongoing bot ping-pong. + if self._loop_guard.enabled: + self._loop_guard.reset_on_human(msg) + # 3. Policy gate decision = self._policy.evaluate(msg) if not decision.allowed: @@ -166,6 +180,16 @@ async def push_message(self, msg: InboundMessage) -> None: logger.debug("safety: policy drop message_id=%s reason=None", msg.id) return + # 3.5 Bot ping-pong guard (opt-in). Counting runs after dedup + self_sent + # + policy so a re-delivery can't inflate the count and a + # policy-rejected message never counts. + if self._loop_guard.enabled and self._loop_guard.record(msg): + if self._loop_guard.on_trip == "reject": + self._emit_reject(msg, "bot_loop") + else: + logger.debug("safety: drop bot-loop message message_id=%s", msg.id) + return + # 4. Processing lock if not self._lock.acquire(msg.id): logger.debug("safety: lock contention drop message_id=%s", msg.id) diff --git a/lark_channel/channel/safety/policy_gate.py b/lark_channel/channel/safety/policy_gate.py index 8245667..a9b0b5c 100644 --- a/lark_channel/channel/safety/policy_gate.py +++ b/lark_channel/channel/safety/policy_gate.py @@ -16,7 +16,9 @@ import threading from dataclasses import dataclass -from typing import List, Optional, Set +from typing import List, Optional, Sequence, Set + +from lark_channel.core.log import logger from ..config import PolicyConfig from ..types import Identity, InboundMessage @@ -52,6 +54,7 @@ def __init__(self, policy: Optional[PolicyConfig] = None) -> None: self._policy = policy or PolicyConfig() self._bot_open_id: Optional[str] = None self._lock = threading.Lock() + self._warn_on_misconfigured_allowlists() def set_bot_open_id(self, open_id: Optional[str]) -> None: with self._lock: @@ -67,6 +70,38 @@ def update_policy(self, **changes) -> None: for k, v in changes.items(): if hasattr(self._policy, k): setattr(self._policy, k, v) + self._warn_on_misconfigured_allowlists() + + def _warn_on_misconfigured_allowlists(self) -> None: + """Flag the most common allowlist misconfiguration: an app id (``cli_…``) + in a list that expects sender ids / chat ids, which silently matches + nothing. Logs only the field name and the single offending value — never + the whole list (no PII / full-table dumps).""" + self._warn_on_cli_entry( + "allow_from", "sender ids (ou_/user_id/union_id)", self._policy.allow_from + ) + self._warn_on_cli_entry( + "group_allowlist", "chat ids (oc_)", self._policy.group_allowlist + ) + + @staticmethod + def _warn_on_cli_entry( + field: str, accepts: str, values: Optional[Sequence[str]] + ) -> None: + if not values: + return + offending = next( + (v for v in values if isinstance(v, str) and v.startswith("cli_")), None + ) + if offending is None: + return + logger.warning( + "channel: PolicyConfig.%s contains an app id (%r) — it accepts %s, " + "not cli_; this entry matches nothing", + field, + offending, + accepts, + ) def evaluate(self, msg: InboundMessage) -> PolicyDecision: with self._lock: diff --git a/lark_channel/channel/safety/types.py b/lark_channel/channel/safety/types.py index fae568c..c66c7da 100644 --- a/lark_channel/channel/safety/types.py +++ b/lark_channel/channel/safety/types.py @@ -39,6 +39,8 @@ "duplicate", "lock_contention", "self_sent", + # Opt-in bot ping-pong guard (PolicyConfig.bot_loop_guard, on_trip='reject'). + "bot_loop", # Policy reasons (unified policy_ prefix) "policy_dm_disabled", "policy_group_disabled", diff --git a/lark_channel/channel/tests/test_body_text_batching_post.py b/lark_channel/channel/tests/test_body_text_batching_post.py new file mode 100644 index 0000000..80a1a38 --- /dev/null +++ b/lark_channel/channel/tests/test_body_text_batching_post.py @@ -0,0 +1,85 @@ +"""body_text survives default batching and is correct for post content.""" + +import asyncio +import time + +from lark_channel.channel.config import PolicyConfig, TextBatchConfig +from lark_channel.channel.normalize.converters.post import convert, convert_body +from lark_channel.channel.safety import SafetyPipeline +from lark_channel.channel.safety.chat_pipeline import merge_batch +from lark_channel.channel.types import ( + Conversation, + Identity, + InboundMessage, + PostContent, + TextContent, +) + + +def _im(mid, text, content_text, body_text): + return InboundMessage( + id=mid, + create_time=int(time.time() * 1000), + conversation=Conversation(chat_id="oc_1", chat_type="p2p"), + sender=Identity(open_id="ou_h"), + content=TextContent(text=text), + content_text=content_text, + safe_content_text=content_text, + body_text=body_text, + mentioned_bot=True, + ) + + +# ── 2.1a: batching recombines derived text views ───────────────────────────── + +def test_merge_batch_recombines_content_and_body_text(): + m1 = _im("1", "@Bot first", "@Bot first", "first") + m2 = _im("2", "@Bot second", "@Bot second", "second") + merged = merge_batch([m1, m2]) + assert merged.content_text == "@Bot first\n\n@Bot second" + assert merged.body_text == "first\n\nsecond" + assert merged.safe_content_text == "@Bot first\n\n@Bot second" + assert merged.batched_sources == [m1, m2] + + +async def test_safety_pipeline_batch_preserves_body_text(): + loop = asyncio.get_running_loop() + got = [] + sp = SafetyPipeline( + loop=loop, + on_message=lambda m: got.append(m), + policy=PolicyConfig(dm_policy="open"), + batch_config=TextBatchConfig(delay_ms=40, max_messages=10, max_chars=100000), + ) + await sp.push_message(_im("1", "@Bot first", "@Bot first", "first")) + await sp.push_message(_im("2", "@Bot second", "@Bot second", "second")) + await asyncio.sleep(0.15) + assert len(got) == 1 + assert got[0].body_text == "first\n\nsecond" + assert got[0].content_text == "@Bot first\n\n@Bot second" + + +# ── 2.1b: post body_text drops only the bot's , keeps title/format ─────── + +def test_post_convert_body_drops_bot_at_node_keeps_rest(): + ast = { + "zh_cn": { + "title": "Deploy", + "content": [[ + {"tag": "at", "user_id": "ou_bot", "user_name": "Bot"}, + {"tag": "text", "text": " run ", "style": ["bold"]}, + {"tag": "at", "user_id": "ou_alice", "user_name": "Alice"}, + ]], + } + } + c = PostContent(post=ast) + full, _ = convert(c) + body = convert_body(c, "ou_bot") + + # content_text (default) keeps everything, incl. the bot mention. + assert "# Deploy" in full and "@Bot" in full and "@Alice" in full + # body drops only the bot mention; title, bold and other mentions survive. + assert "# Deploy" in body + assert "@Bot" not in body + assert "@Alice" in body + assert "**" in body diff --git a/lark_channel/channel/tests/test_channel_low_level.py b/lark_channel/channel/tests/test_channel_low_level.py index 26a332b..8d7d78c 100644 --- a/lark_channel/channel/tests/test_channel_low_level.py +++ b/lark_channel/channel/tests/test_channel_low_level.py @@ -137,7 +137,10 @@ async def test_fetch_inbound_message_returns_normalized_message_without_changing assert inbound is not None assert inbound.id == "om_1" + # content_text keeps the rendered bot mention (default behavior unchanged); + # the bot-mention-stripped view lives on body_text. assert inbound.content_text == "hello @Bot" + assert inbound.body_text.strip() == "hello" assert inbound.mentioned_bot is True assert inbound.raw["message_id"] == "om_1" diff --git a/lark_channel/channel/tests/test_chat_bots.py b/lark_channel/channel/tests/test_chat_bots.py new file mode 100644 index 0000000..05d09a5 --- /dev/null +++ b/lark_channel/channel/tests/test_chat_bots.py @@ -0,0 +1,116 @@ +"""FeishuChannel.get_chat_bots(). + +The members API filters out bots, so in-group bots are listed via a separate +``/members/bots`` endpoint returning ``{items:[{bot_id, bot_name}]}``. Bots are +mapped to ``ChatMember(is_bot=True)`` and seeded into the roster so they can be +@-mentioned by name without first appearing in an inbound mention. +""" + +import json +from unittest.mock import patch + +import pytest + +from lark_channel.channel import FeishuChannel +from lark_channel.channel.errors import FeishuChannelError +from lark_channel.core.model import RawResponse + +_TRANSPORT = "lark_channel.core.http.Transport.aexecute" +_VERIFY = "lark_channel.core.token.auth.verify" + + +def _raw(body, status=200): + r = RawResponse() + r.status_code = status + r.content = json.dumps(body).encode("utf-8") + return r + + +def _fake_verify(cfg, request, option): + option.tenant_access_token = "t-test" + + +def _bots_page(items): + return _raw({"code": 0, "msg": "ok", "data": {"items": items}}) + + +def _members_page(items): + return _raw({"code": 0, "msg": "ok", "data": {"items": items, "has_more": False}}) + + +def _Spy(responses): + """A coroutine-function spy for ``Transport.aexecute`` (see test_chat_members + for why an ``async def`` closure is required over a callable instance).""" + responses = list(responses) + + async def aexecute(conf, req, option=None): + aexecute.requests.append(req) + return responses[0] if len(responses) == 1 else responses.pop(0) + + aexecute.requests = [] + return aexecute + + +def _channel(): + return FeishuChannel(app_id="cli_x", app_secret="s") + + +async def test_lists_bots_mapped_is_bot_true_and_hits_bots_endpoint(): + spy = _Spy([_bots_page([{"bot_id": "ou_bot1", "bot_name": "HelperBot"}])]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + bots = await ch.get_chat_bots("oc_chat") + + assert len(bots) == 1 + assert bots[0].id == "ou_bot1" + assert bots[0].name == "HelperBot" + assert bots[0].is_bot is True + assert (spy.requests[0].uri or "").endswith("/members/bots") + + +async def test_second_call_hits_cache_and_force_bypasses(): + spy = _Spy([_bots_page([{"bot_id": "ou_bot1", "bot_name": "HelperBot"}])]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + await ch.get_chat_bots("oc_chat") + await ch.get_chat_bots("oc_chat") + assert len(spy.requests) == 1 + await ch.get_chat_bots("oc_chat", force=True) + assert len(spy.requests) == 2 + + +async def test_seeds_roster_for_name_resolution(): + spy = _Spy([_bots_page([{"bot_id": "ou_bot1", "bot_name": "HelperBot"}])]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + await ch.get_chat_bots("oc_chat") + + # The per-chat roster cache lives alongside the chat-mode cache; the bot is + # resolvable by name after the fetch without an inbound mention. + cache = ch._chat_member_cache + assert cache.resolve_open_id("oc_chat", "HelperBot") == "ou_bot1" + assert cache.resolve_name("oc_chat", "ou_bot1") == "HelperBot" + + +async def test_does_not_overwrite_member_users_cache(): + spy = _Spy([ + _members_page([{"member_id": "ou_user", "open_id": "ou_user", "name": "Alice"}]), + _bots_page([{"bot_id": "ou_bot1", "bot_name": "HelperBot"}]), + ]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + await ch.get_chat_members("oc_chat") + await ch.get_chat_bots("oc_chat") + # The users list must still be cached (no third request). + users = await ch.get_chat_members("oc_chat") + + assert [m.id for m in users] == ["ou_user"] + assert len(spy.requests) == 2 + + +async def test_api_failure_raises_channel_error(): + spy = _Spy([_raw({"code": 99991672, "msg": "no permission"})]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + with pytest.raises(FeishuChannelError): + await ch.get_chat_bots("oc_chat") diff --git a/lark_channel/channel/tests/test_chat_member_cache.py b/lark_channel/channel/tests/test_chat_member_cache.py new file mode 100644 index 0000000..1e1dc37 --- /dev/null +++ b/lark_channel/channel/tests/test_chat_member_cache.py @@ -0,0 +1,175 @@ +"""ChatMemberCache. + +Per-chat roster cache (OrderedDict + LRU + TTL, like ChatModeCache). The +security-critical invariant is fail-safe name→open_id resolution: a name that +maps to more than one open_id is ambiguous and resolves to None (never +last-writer-wins), so nobody can hijack an @-mention by renaming. + +Clock and capacity are injected so time / eviction are deterministic. +""" + +from lark_channel.channel.chat_member_cache import ChatMemberCache +from lark_channel.channel.types import ChatMember + + +def _member(open_id, name, is_bot=False): + return ChatMember(id=open_id, name=name, is_bot=is_bot) + + +def test_bidirectional_resolution_and_unknown_is_none(): + cache = ChatMemberCache() + cache.set_members("oc_c", [_member("ou_a", "Alice")], source="api") + + assert cache.resolve_name("oc_c", "ou_a") == "Alice" + assert cache.resolve_open_id("oc_c", "Alice") == "ou_a" + assert cache.resolve_name("oc_c", "ou_missing") is None + assert cache.resolve_open_id("oc_c", "Ghost") is None + + +def test_duplicate_name_is_ambiguous_but_reverse_still_resolves(): + cache = ChatMemberCache() + cache.set_members( + "oc_c", + [_member("ou_a", "Sam"), _member("ou_b", "Sam")], + source="api", + ) + + # Two open_ids share the name → refuse to guess. + assert cache.resolve_open_id("oc_c", "Sam") is None + # But each open_id still maps back to its name. + assert cache.resolve_name("oc_c", "ou_a") == "Sam" + assert cache.resolve_name("oc_c", "ou_b") == "Sam" + + +def test_same_open_id_reobserved_is_not_a_conflict(): + cache = ChatMemberCache() + cache.set_members("oc_c", [_member("ou_a", "Alice")], source="api") + # A later inbound mention re-observes the same (open_id, name) pair — this + # must not be treated as an ambiguity. + cache.set_members("oc_c", [_member("ou_a", "Alice")], source="mention") + + assert cache.resolve_open_id("oc_c", "Alice") == "ou_a" + + +def test_mention_does_not_overwrite_api_and_creates_ambiguity(): + cache = ChatMemberCache() + cache.set_members("oc_c", [_member("ou_api", "Sam")], source="api") + # Attacker-controlled mention claims the name "Sam" for a different id. + cache.set_members("oc_c", [_member("ou_mention", "Sam")], source="mention") + + # Conflicting ids for one name → ambiguous (fail-safe), never the mention id. + assert cache.resolve_open_id("oc_c", "Sam") is None + # The authoritative api open_id→name mapping is untouched. + assert cache.resolve_name("oc_c", "ou_api") == "Sam" + + +def test_ttl_expiry_with_injected_clock(): + now = [1000.0] + cache = ChatMemberCache(now=lambda: now[0], ttl_seconds=300) + cache.set_members("oc_c", [_member("ou_a", "Alice")], source="api") + assert cache.resolve_name("oc_c", "ou_a") == "Alice" + + now[0] += 301 # past the TTL + assert cache.resolve_name("oc_c", "ou_a") is None + assert not cache.get_members("oc_c") + + +def test_api_refresh_drops_departed_member(): + # A full API snapshot replaces the previous one, so a member who left the + # chat is no longer resolvable (index rebuilt from the fresh snapshot). + cache = ChatMemberCache() + cache.set_members("c", [_member("ou_a", "Alice"), _member("ou_b", "Bob")], source="api") + assert cache.resolve_open_id("c", "Alice") == "ou_a" + + cache.set_members("c", [_member("ou_b", "Bob")], source="api") # Alice left + assert cache.resolve_open_id("c", "Alice") is None + assert cache.resolve_open_id("c", "Bob") == "ou_b" + + +def test_api_refresh_reevaluates_resolved_ambiguity(): + # Two "Sam"s → ambiguous; after one leaves, the name resolves again. + cache = ChatMemberCache() + cache.set_members("c", [_member("ou_a", "Sam"), _member("ou_b", "Sam")], source="api") + assert cache.resolve_open_id("c", "Sam") is None # ambiguous + + cache.set_members("c", [_member("ou_a", "Sam")], source="api") # one Sam left + assert cache.resolve_open_id("c", "Sam") == "ou_a" + + +def test_non_open_id_snapshot_does_not_feed_name_index(): + # A user_id-typed snapshot must not populate the name→open_id index (a + # user_id can't be used in an ), and must not satisfy an open_id query. + cache = ChatMemberCache() + cache.set_members("c", [_member("u_a", "Alice")], source="api", id_type="user_id") + assert cache.resolve_open_id("c", "Alice") is None + assert cache.get_members("c", "user_id") is not None + assert cache.get_members("c", "open_id") is None + + +def test_incomplete_snapshot_does_not_feed_name_index(): + cache = ChatMemberCache() + cache.set_members("c", [_member("ou_a", "Alice")], source="api", complete=False) + assert cache.resolve_open_id("c", "Alice") is None # truncated ≠ authoritative + + +def test_incomplete_refresh_does_not_clobber_complete_snapshot(): + cache = ChatMemberCache() + cache.set_members("c", [_member("ou_a", "Alice")], source="api", complete=True) + cache.set_members("c", [_member("ou_a", "Alice")], source="api", complete=False) + assert cache.resolve_open_id("c", "Alice") == "ou_a" + + +def test_per_id_type_slots_do_not_overwrite_each_other(): + cache = ChatMemberCache() + cache.set_members("c", [_member("ou_a", "Alice")], source="api", id_type="open_id") + cache.set_members("c", [ChatMember(id="u_a", name="Alice")], source="api", id_type="user_id") + # The user_id snapshot must not erase the open_id name index. + assert cache.resolve_open_id("c", "Alice") == "ou_a" + assert cache.get_members("c", "open_id") is not None + assert cache.get_members("c", "user_id") is not None + + +def test_full_refresh_drops_stale_mention_observation(): + cache = ChatMemberCache() + cache.set_members("c", [ChatMember(id="ou_a", name="Alice", is_bot=False)], source="mention") + assert cache.resolve_open_id("c", "Alice") == "ou_a" + cache.set_members("c", [_member("ou_b", "Bob")], source="api", complete=True) # Alice gone + assert cache.resolve_open_id("c", "Alice") is None + + +def test_api_rename_supersedes_stale_mention_alias(): + cache = ChatMemberCache() + cache.set_members("c", [ChatMember(id="ou_a", name="OldName")], source="mention") + cache.set_members("c", [_member("ou_a", "NewName")], source="api", complete=True) + assert cache.resolve_open_id("c", "OldName") is None + assert cache.resolve_open_id("c", "NewName") == "ou_a" + + +def test_bots_refresh_reconciles_bot_mention(): + cache = ChatMemberCache() + cache.set_members("c", [ChatMember(id="ou_x", name="GhostBot", is_bot=True)], source="mention") + assert cache.resolve_open_id("c", "GhostBot") == "ou_x" + cache.set_bots("c", [ChatMember(id="ou_y", name="RealBot", is_bot=True)], complete=True) + assert cache.resolve_open_id("c", "GhostBot") is None + + +def test_reads_return_defensive_copies(): + cache = ChatMemberCache() + cache.set_members("c", [_member("ou_a", "Alice")], source="api") + got = cache.get_members("c", "open_id") + got.clear() + got[:] = [] + again = cache.get_members("c", "open_id") + assert [m.id for m in again] == ["ou_a"] # list mutation didn't affect cache + again[0].name = "Hacked" + assert cache.resolve_open_id("c", "Alice") == "ou_a" # member mutation didn't corrupt index + + +def test_lru_evicts_oldest_chat_over_max_chats(): + cache = ChatMemberCache(max_chats=2) + cache.set_members("c1", [_member("ou_1", "One")], source="api") + cache.set_members("c2", [_member("ou_2", "Two")], source="api") + cache.set_members("c3", [_member("ou_3", "Three")], source="api") + + assert not cache.get_members("c1") # oldest evicted + assert cache.get_members("c3") diff --git a/lark_channel/channel/tests/test_chat_members.py b/lark_channel/channel/tests/test_chat_members.py new file mode 100644 index 0000000..383aa2f --- /dev/null +++ b/lark_channel/channel/tests/test_chat_members.py @@ -0,0 +1,289 @@ +"""FeishuChannel.get_chat_members(). + +Raw-Transport pattern (mirrors ``bot_identity``): the ``im.v1.chat`` resource +has no members method, so the port builds a ``BaseRequest`` and calls +``Transport.aexecute`` directly. These tests patch that seam + the tenant-token +verify step so nothing hits the network. + +Patch targets mirror ``test_bot_identity.py``: ``Transport.aexecute`` is patched +on the shared class object (``lark_channel.core.http.Transport``), and the +tenant-token injection is patched at its source +(``lark_channel.core.token.auth.verify``). +""" + +import asyncio +import json +from unittest.mock import patch + +import pytest + +from lark_channel.channel import FeishuChannel +from lark_channel.channel.errors import FeishuChannelError +from lark_channel.channel.types import ChatMember +from lark_channel.channel.config import ChannelConfig +from lark_channel.core.model import RawResponse + +_TRANSPORT = "lark_channel.core.http.Transport.aexecute" +_VERIFY = "lark_channel.core.token.auth.verify" + + +def _raw(body, status=200): + r = RawResponse() + r.status_code = status + r.content = json.dumps(body).encode("utf-8") + return r + + +def _fake_verify(cfg, request, option): + option.tenant_access_token = "t-test" + + +def _member_page(items, *, has_more=False, page_token=""): + data = {"items": items, "has_more": has_more} + if page_token: + data["page_token"] = page_token + return _raw({"code": 0, "msg": "ok", "data": data}) + + +def _member_item(open_id, name): + # Hedge on the wire field name for the member id (member_id is the Feishu + # field; open_id is included so either mapping resolves the same value). + return {"member_id_type": "open_id", "member_id": open_id, "open_id": open_id, "name": name} + + +def _channel(**cfg_kwargs): + if cfg_kwargs: + return FeishuChannel(app_id="cli_x", app_secret="s", config=ChannelConfig(**cfg_kwargs)) + return FeishuChannel(app_id="cli_x", app_secret="s") + + +def _Spy(responses): + """A coroutine-function spy for ``Transport.aexecute``. + + ``patch`` auto-detects ``aexecute`` (an ``async def``) and installs an + AsyncMock, which awaits its ``side_effect`` only when the side_effect is + itself a coroutine *function* (``inspect.iscoroutinefunction``) — a callable + class instance is not one. So the spy is a real ``async def`` closure that + records requests on its ``.requests`` attribute. + """ + responses = list(responses) + + async def aexecute(conf, req, option=None): + aexecute.requests.append(req) + return responses[0] if len(responses) == 1 else responses.pop(0) + + aexecute.requests = [] + return aexecute + + +async def test_follows_pagination_and_maps_members(): + spy = _Spy([ + _member_page([_member_item("ou_a", "Alice")], has_more=True, page_token="tok2"), + _member_page([_member_item("ou_b", "Bob")], has_more=False), + ]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + members = await ch.get_chat_members("oc_chat") + + ids = {m.id for m in members} + assert ids == {"ou_a", "ou_b"} + by_id = {m.id: m for m in members} + assert by_id["ou_a"].name == "Alice" + # Second request must carry the page_token from the first page. + assert ("page_token", "tok2") in spy.requests[1].queries + + +async def test_page_size_clamped_to_100(): + spy = _Spy([_member_page([_member_item("ou_a", "Alice")])]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + await ch.get_chat_members("oc_chat", page_size=500) + + assert ("page_size", "100") in spy.requests[0].queries + + +async def test_max_pages_caps_requests(): + # has_more is always True; max_pages=1 must stop after a single request. + always_more = _member_page([_member_item("ou_a", "Alice")], has_more=True, page_token="next") + spy = _Spy([always_more]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + await ch.get_chat_members("oc_chat", max_pages=1) + + assert len(spy.requests) == 1 + + +async def test_second_call_hits_cache_and_force_bypasses(): + spy = _Spy([_member_page([_member_item("ou_a", "Alice")])]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + await ch.get_chat_members("oc_chat") + await ch.get_chat_members("oc_chat") + assert len(spy.requests) == 1 # cache hit — no new request + await ch.get_chat_members("oc_chat", force=True) + assert len(spy.requests) == 2 # force bypasses the cache + + +async def test_resolve_chat_members_hook_short_circuits_api(): + hook_members = [ChatMember(id="ou_hook", name="Hooked")] + spy = _Spy([_member_page([_member_item("ou_a", "Alice")])]) + ch = _channel(resolve_chat_members=lambda chat_id: hook_members) + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + members = await ch.get_chat_members("oc_chat") + + assert [m.id for m in members] == ["ou_hook"] + assert spy.requests == [] # hook returned a list — API not called + + +async def test_resolve_chat_members_hook_none_falls_back_to_api(): + spy = _Spy([_member_page([_member_item("ou_a", "Alice")])]) + ch = _channel(resolve_chat_members=lambda chat_id: None) + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + members = await ch.get_chat_members("oc_chat") + + assert [m.id for m in members] == ["ou_a"] + assert len(spy.requests) == 1 + + +async def test_resolve_chat_members_hook_empty_list_falls_back_to_api(): + # An empty list means "no data from the hook" — same as None, it falls back + # to the API rather than yielding an empty roster. + spy = _Spy([_member_page([_member_item("ou_a", "Alice")])]) + ch = _channel(resolve_chat_members=lambda chat_id: []) + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + members = await ch.get_chat_members("oc_chat") + + assert [m.id for m in members] == ["ou_a"] + assert len(spy.requests) == 1 + + +async def test_members_are_never_bots(): + spy = _Spy([_member_page([_member_item("ou_a", "Alice")])]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + members = await ch.get_chat_members("oc_chat") + + assert all(m.is_bot is False for m in members) + + +async def test_api_failure_raises_channel_error(): + spy = _Spy([_raw({"code": 99991672, "msg": "no permission"})]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + with pytest.raises(FeishuChannelError): + await ch.get_chat_members("oc_chat") + + +async def test_hook_receives_id_type_when_two_arg(): + seen = {} + + def hook(chat_id, id_type): + seen["id_type"] = id_type + return [ChatMember(id="u_h", name="H", id_type="user_id")] + + ch = _channel(resolve_chat_members=hook) + members = await ch.get_chat_members("oc_chat", id_type="user_id") + assert seen["id_type"] == "user_id" + assert members[0].id == "u_h" + + +async def test_hook_id_type_mismatch_raises(): + # hook returns open_id members for a user_id request → typed error, not a + # silently mistyped id. + ch = _channel(resolve_chat_members=lambda cid: [ChatMember(id="ou_a", name="A")]) + with pytest.raises(FeishuChannelError): + await ch.get_chat_members("oc_chat", id_type="user_id") + + +async def test_async_hook_supported(): + async def hook(chat_id): + return [ChatMember(id="ou_h", name="Hooked")] + + ch = _channel(resolve_chat_members=hook) + members = await ch.get_chat_members("oc_chat") + assert [m.id for m in members] == ["ou_h"] + + +async def test_concurrent_cold_requests_collapse_to_one_fetch(): + # A burst of cold requests for the same (chat, id_type) shares one upstream + # call (singleflight) rather than stampeding the members API / token cache. + spy = _Spy([_member_page([_member_item("ou_a", "Alice")])]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + results = await asyncio.gather(*[ch.get_chat_members("oc_chat") for _ in range(10)]) + assert all(r[0].id == "ou_a" for r in results) + assert len(spy.requests) == 1 + + +async def test_max_pages_zero_raises_and_does_not_cache(): + ch = _channel() + with pytest.raises(FeishuChannelError): + await ch.get_chat_members("oc_chat", max_pages=0) + assert ch._chat_member_cache.get_members("oc_chat", "open_id") is None + + +async def test_truncated_result_is_not_cached(): + # max_pages=1 with has_more always true → truncated → must not be cached, so + # a second default call re-fetches rather than serving the partial roster. + always_more = _member_page([_member_item("ou_a", "Alice")], has_more=True, page_token="t") + spy = _Spy([always_more]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + await ch.get_chat_members("oc_chat", max_pages=1) + await ch.get_chat_members("oc_chat", max_pages=1) + assert len(spy.requests) == 2 # not cached + # And a truncated roster never feeds the name index. + assert ch._chat_member_cache.resolve_open_id("oc_chat", "Alice") is None + + +async def test_http_error_without_business_code_is_not_empty_success(): + # HTTP 503 with a JSON body that has no non-zero `code` must raise, not be + # cached as an empty (zero-member) roster. + spy = _Spy([_raw({"message": "unavailable"}, status=503)]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + with pytest.raises(FeishuChannelError): + await ch.get_chat_members("oc_chat") + + +async def test_non_object_json_raises_channel_error_not_attributeerror(): + # A bare top-level array must surface as FeishuChannelError, not leak an + # AttributeError from calling .get() on a list. + spy = _Spy([_raw([{"member_id": "ou_a"}], status=200)]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + with pytest.raises(FeishuChannelError): + await ch.get_chat_members("oc_chat") + + +async def test_cache_is_keyed_by_id_type(): + # A user_id query must not be served from an open_id snapshot (their .id + # values differ), so a differing id_type is a cache miss → a fresh request. + spy = _Spy([ + _member_page([_member_item("ou_a", "Alice")]), + _member_page([{"member_id_type": "user_id", "member_id": "u_a", "name": "Alice"}]), + ]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + first = await ch.get_chat_members("oc_chat", id_type="open_id") + second = await ch.get_chat_members("oc_chat", id_type="user_id") + + assert first[0].id == "ou_a" + assert second[0].id == "u_a" + assert len(spy.requests) == 2 # different id_type did not hit the cache + + +async def test_chat_id_is_url_encoded_via_paths_not_bare_fstring(): + """Security: caller-supplied chat_id must go through ``req.paths`` (encoded + by ``_build_url``), never interpolated raw into ``req.uri`` — otherwise a + ``/../`` chat_id walks the path.""" + evil = "oc_x/../y" + spy = _Spy([_member_page([_member_item("ou_a", "Alice")])]) + ch = _channel() + with patch(_VERIFY, side_effect=_fake_verify), patch(_TRANSPORT, side_effect=spy): + await ch.get_chat_members(evil) + + req = spy.requests[0] + assert req.paths.get("chat_id") == evil + assert ":chat_id" in (req.uri or "") + assert evil not in (req.uri or "") diff --git a/lark_channel/channel/tests/test_compose_mentions_escaping.py b/lark_channel/channel/tests/test_compose_mentions_escaping.py new file mode 100644 index 0000000..2eab7b1 --- /dev/null +++ b/lark_channel/channel/tests/test_compose_mentions_escaping.py @@ -0,0 +1,69 @@ +""" sink hardening on outbound composition. + +``display_name`` is attacker-controllable and the Feishu ```` sink does not +escape it. The text builder and the post AST builder must both run every +mention through ``escape_at_name`` + ``is_valid_open_id`` so a malicious name +can't inject a second ```` / spoof a user_id, and an invalid open_id +(``cli_`` / markup) is skipped. The @all sentinel (``open_id == "all"``) must +survive the hardening. +""" + +import json + +from lark_channel.channel.types import Identity, OutboundText +from lark_channel.channel.outbound.sender import _build_text +from lark_channel.channel.outbound.markdown.to_post import markdown_to_post_ast + + +def _text_of(built): + return json.loads(built["content"])["text"] + + +def _at_nodes(ast): + nodes = [] + for locale in ast.values(): + for row in locale.get("content", []): + for el in row: + if isinstance(el, dict) and el.get("tag") == "at": + nodes.append(el) + return nodes + + +def test_malicious_display_name_cannot_inject_second_at_tag(): + evil = Identity(open_id="ou_a", display_name='') + content = _text_of(_build_text(OutboundText(text="hi", mentions=[evil]))) + + # The security property is that no SECOND parseable tag and no forged + # user_id can be injected — the only is the legitimate wrapper, and the + # attacker's `user_id="ou_evil"` never survives as markup. (After stripping + # `<>"` the escaped name may still contain the inert substring "ou_evil" as + # plain text inside the legit tag body, which is harmless.) + assert content.count(""): + content = _text_of( + _build_text(OutboundText(text="hi", mentions=[Identity(open_id=bad, display_name="X")])) + ) + assert "Alice' in content + + +def test_at_all_sentinel_survives_hardening_in_text(): + content = _text_of( + _build_text(OutboundText(text="hi", mentions=[Identity(open_id="all", display_name="所有人")])) + ) + assert '' in content + + +def test_invalid_open_id_mention_is_skipped_in_post_ast(): + ast = markdown_to_post_ast("hi", mentions=[Identity(open_id="cli_x", display_name="X")]) + assert all(node.get("user_id") != "cli_x" for node in _at_nodes(ast)) diff --git a/lark_channel/channel/tests/test_empty_at_wake.py b/lark_channel/channel/tests/test_empty_at_wake.py new file mode 100644 index 0000000..b9fa243 --- /dev/null +++ b/lark_channel/channel/tests/test_empty_at_wake.py @@ -0,0 +1,74 @@ +"""Empty @-mention wake + body_text. + +A message that only @-mentions the bot with no body must still be delivered +(not dropped as empty): ``mentioned_bot`` is True. ``content_text`` keeps the +rendered mention (default behavior unchanged), while ``body_text`` has the +bot's own mention removed, so downstream code detects a bare "poke" via +``mentioned_bot and not body_text.strip()``. +""" + +import pytest + +from lark_channel.channel.normalize.pipeline import ( + InboundPipeline, + PipelineConfig, + PipelineDeps, +) + + +async def _process(content_text, mentions, *, bot="ou_bot"): + pipeline = InboundPipeline(PipelineConfig(), PipelineDeps()) + pipeline.set_bot_open_id(bot) + return await pipeline.process( + event_id="e", + message_event={ + "message_id": "om_1", + "create_time": 1, + "chat_id": "oc_c", + "chat_type": "group", + "message_type": "text", + "content": {"text": content_text}, + "mentions": mentions, + }, + sender={"sender_id": {"open_id": "ou_human"}, "sender_type": "user"}, + ) + + +async def test_at_only_message_delivers_with_empty_body_text(): + # A real bare-@ event carries the mention placeholder in the text ("@_user_1 "). + inbound = await _process( + "@_user_1 ", + [{"key": "@_user_1", "id": {"open_id": "ou_bot"}, "name": "Bot"}], + ) + assert inbound is not None # not dropped + assert inbound.mentioned_bot is True + # content_text keeps the rendered mention (default behavior unchanged)... + assert "@Bot" in inbound.content_text + # ...body_text is empty, so `mentioned_bot and not body_text.strip()` fires. + assert inbound.body_text.strip() == "" + + +async def test_content_text_default_preserved_body_text_stripped(): + # "@bot hello there": content_text unchanged (still renders the mention); + # only body_text drops the bot's own mention. + inbound = await _process( + "@_user_1 hello there", + [{"key": "@_user_1", "id": {"open_id": "ou_bot"}, "name": "Bot"}], + ) + assert inbound.mentioned_bot is True + assert "@Bot" in inbound.content_text + assert "hello there" in inbound.content_text + assert inbound.body_text.strip() == "hello there" + assert "@Bot" not in inbound.body_text + + +async def test_body_text_equals_content_text_when_bot_not_mentioned(): + # A message @-mentioning someone who is NOT the bot: nothing is stripped, + # body_text == content_text, and the other mention is preserved in both. + inbound = await _process( + "@_user_1 ping", + [{"key": "@_user_1", "id": {"open_id": "ou_alice"}, "name": "Alice"}], + ) + assert inbound.mentioned_bot is False + assert "@Alice" in inbound.content_text + assert inbound.body_text == inbound.content_text diff --git a/lark_channel/channel/tests/test_get_bot_identity.py b/lark_channel/channel/tests/test_get_bot_identity.py new file mode 100644 index 0000000..677b056 --- /dev/null +++ b/lark_channel/channel/tests/test_get_bot_identity.py @@ -0,0 +1,38 @@ +"""FeishuChannel.get_bot_identity(). + +Unlike the existing ``bot_identity`` property (which returns Optional), the new +method raises ``FeishuChannelError(code=not_connected)`` when identity is not +yet resolved, so callers can't silently write a missing identity into a system +prompt. +""" + +import pytest + +from lark_channel.channel import FeishuChannel +from lark_channel.channel.bot_identity import BotIdentity +from lark_channel.channel.errors import FeishuChannelError, FeishuChannelErrorCode + + +def _channel(): + return FeishuChannel(app_id="cli_x", app_secret="secret") + + +def test_returns_resolved_identity(): + ch = _channel() + ident = BotIdentity(open_id="ou_bot", name="Helper") + ch._bot_identity = ident + assert ch.get_bot_identity() is ident + + +def test_raises_not_connected_when_unresolved(): + ch = _channel() + with pytest.raises(FeishuChannelError) as excinfo: + ch.get_bot_identity() + assert excinfo.value.code == FeishuChannelErrorCode.NOT_CONNECTED + + +def test_property_still_returns_none_when_unresolved(): + # Regression: the new method must not change the Optional-returning + # ``bot_identity`` property behaviour. + ch = _channel() + assert ch.bot_identity is None diff --git a/lark_channel/channel/tests/test_loop_guard.py b/lark_channel/channel/tests/test_loop_guard.py new file mode 100644 index 0000000..0e0925e --- /dev/null +++ b/lark_channel/channel/tests/test_loop_guard.py @@ -0,0 +1,286 @@ +"""botLoopGuard. + +Heuristic guard against two bots @-mentioning each other forever. It only +counts "another bot @-mentioned me" (``sender.is_bot and mentioned_bot``), +uses ``msg.create_time`` as its clock (deterministic), dedups by message_id +within the window, and a ``user`` message clears the key. Opt-in, default off. + +Pipeline wiring: it sits after dedup/self_sent/policy and before the lock; +``drop`` silently discards, ``reject`` emits ``RejectEvent(reason="bot_loop")``. +""" + +import asyncio +import time +from unittest.mock import Mock + +import pytest + +from lark_channel.channel.config import BotLoopGuardConfig, PolicyConfig, TextBatchConfig +from lark_channel.channel.safety import SafetyPipeline +from lark_channel.channel.safety.loop_guard import LoopGuard +from lark_channel.channel.types import ( + Conversation, + Identity, + InboundMessage, + Mention, + TextContent, +) + + +def _msg(mid, t, *, chat="c1", sender="ou_bot", sender_type="bot", mentioned_bot=True): + is_bot = sender_type in {"bot", "app"} + return InboundMessage( + id=mid, + create_time=t, + conversation=Conversation(chat_id=chat, chat_type="group"), + sender=Identity(open_id=sender, sender_type=sender_type, is_bot=is_bot), + content=TextContent(text="hi"), + mentioned_bot=mentioned_bot, + ) + + +def _guard(**cfg): + cfg.setdefault("enabled", True) + return LoopGuard(BotLoopGuardConfig(**cfg), Mock()) + + +# ---- record() counting ----------------------------------------------------- + + +def test_trips_when_window_reaches_threshold(): + guard = _guard(window_ms=60000, max_bot_mentions=3) + assert guard.record(_msg("m1", 0)) is False + assert guard.record(_msg("m2", 10)) is False + assert guard.record(_msg("m3", 20)) is True + + +def test_old_entries_slide_out_of_window(): + guard = _guard(window_ms=1000, max_bot_mentions=2) + assert guard.record(_msg("m1", 0)) is False + # m1 is 5000ms before m2 → outside the 1000ms window, so it doesn't count. + assert guard.record(_msg("m2", 5000)) is False + assert guard.record(_msg("m3", 5500)) is True + + +def test_ineligible_messages_are_not_counted(): + guard = _guard(window_ms=60000, max_bot_mentions=2) + # A human sender never counts. + assert guard.record(_msg("m1", 0, sender_type="user", mentioned_bot=True)) is False + # A bot that didn't @ me never counts. + assert guard.record(_msg("m2", 10, mentioned_bot=False)) is False + # ...so a single genuine bot@me is still below threshold. + assert guard.record(_msg("m3", 20)) is False + + +def test_user_message_clears_the_key(): + guard = _guard(window_ms=60000, max_bot_mentions=2) + assert guard.record(_msg("m1", 0)) is False + guard.record(_msg("u1", 10, sender_type="user", mentioned_bot=False)) # clears + # Count restarts, so the next bot@me is only #1 — not a trip. + assert guard.record(_msg("m2", 20)) is False + + +def test_same_message_id_counts_once_within_window(): + guard = _guard(window_ms=60000, max_bot_mentions=2) + assert guard.record(_msg("m1", 0)) is False + assert guard.record(_msg("m1", 5)) is False # redelivery — not double-counted + assert guard.record(_msg("m2", 10)) is True + + +def test_first_trip_warns_exactly_once(): + logger = Mock() + guard = LoopGuard(BotLoopGuardConfig(enabled=True, window_ms=60000, max_bot_mentions=2), logger) + guard.record(_msg("m1", 0)) + guard.record(_msg("m2", 10)) # trip 1 → warn + guard.record(_msg("m3", 20)) # still tripping → no second warn + assert logger.warning.call_count == 1 + + +def test_scope_chat_plus_sender_isolates_senders(): + guard = _guard(window_ms=60000, max_bot_mentions=2, scope="chat+sender") + assert guard.record(_msg("m1", 0, sender="ou_botA")) is False + assert guard.record(_msg("m2", 10, sender="ou_botB")) is False # different key + assert guard.record(_msg("m3", 20, sender="ou_botA")) is True # botA now at 2 + + +def test_scope_chat_merges_senders(): + guard = _guard(window_ms=60000, max_bot_mentions=2, scope="chat") + assert guard.record(_msg("m1", 0, sender="ou_botA")) is False + assert guard.record(_msg("m2", 10, sender="ou_botB")) is True # merged count = 2 + + +# ---- pipeline wiring ------------------------------------------------------- + + +def _recent_bot_msg(): + return _msg("mp1", int(time.time() * 1000), chat="c_pipe", sender="ou_peer") + + +async def test_pipeline_drop_suppresses_delivery(): + loop = asyncio.get_running_loop() + delivered = [] + pipe = SafetyPipeline( + loop=loop, + on_message=lambda m: delivered.append(m), + policy=PolicyConfig( + require_mention=False, + bot_loop_guard=BotLoopGuardConfig(enabled=True, max_bot_mentions=1, on_trip="drop"), + ), + batch_config=TextBatchConfig(delay_ms=0), + ) + await pipe.push_message(_recent_bot_msg()) + await asyncio.sleep(0.05) + assert delivered == [] + + +async def test_pipeline_reject_emits_bot_loop_reason(): + loop = asyncio.get_running_loop() + rejects = [] + pipe = SafetyPipeline( + loop=loop, + on_message=lambda m: None, + on_reject=lambda r: rejects.append(r), + policy=PolicyConfig( + require_mention=False, + bot_loop_guard=BotLoopGuardConfig(enabled=True, max_bot_mentions=1, on_trip="reject"), + ), + batch_config=TextBatchConfig(delay_ms=0), + ) + await pipe.push_message(_recent_bot_msg()) + await asyncio.sleep(0.05) + assert len(rejects) == 1 + assert rejects[0].reason == "bot_loop" + + +def test_reconfigure_enable_disable_takes_effect(): + # disabled -> enabled makes the next eligible message trip immediately; + # enabled -> disabled stops tripping (runtime config actually takes effect). + guard = LoopGuard(BotLoopGuardConfig(enabled=False), Mock()) + assert guard.record(_msg("m1", 0)) is False # disabled + + guard.reconfigure(BotLoopGuardConfig(enabled=True, max_bot_mentions=1, window_ms=60000)) + assert guard.record(_msg("m2", 10)) is True # now enabled, threshold 1 + + guard.reconfigure(BotLoopGuardConfig(enabled=False)) + assert guard.record(_msg("m3", 20)) is False # disabled again + + +def test_reconfigure_threshold_change_clears_state(): + guard = _guard(window_ms=60000, max_bot_mentions=3) + guard.record(_msg("m1", 0)) + guard.record(_msg("m2", 10)) # count 2 of 3 + # Changing the threshold clears counting state, so we start fresh at 1. + guard.reconfigure(BotLoopGuardConfig(enabled=True, window_ms=60000, max_bot_mentions=2)) + assert guard.record(_msg("m3", 20)) is False # count 1 of 2 + assert guard.record(_msg("m4", 30)) is True # count 2 of 2 + + +def test_threshold_below_one_is_clamped_and_can_trip(): + # max_bot_mentions=0 would otherwise be nonsensical; it clamps to 1 so a + # single eligible message trips (never silently disables the guard). + guard = _guard(window_ms=60000, max_bot_mentions=0) + assert guard.record(_msg("m1", 0)) is True + + +def test_out_of_order_stale_event_does_not_join_future_window(): + # A future-timestamped event followed by a stale (older) event must not be + # grouped into the same window — the stale event falls outside it. + guard = _guard(window_ms=1000, max_bot_mentions=2) + assert guard.record(_msg("m_future", 100000)) is False # count 1 + assert guard.record(_msg("m_stale", 0)) is False # outside window, not counted + + +def test_reset_on_human_chat_scope(): + guard = _guard(window_ms=60000, max_bot_mentions=2, scope="chat") + assert guard.record(_msg("m1", 0)) is False # count 1 + guard.reset_on_human(_msg("h", 10, sender_type="user", mentioned_bot=False)) + assert guard.record(_msg("m2", 20)) is False # restarted at 1, not tripped + + +def test_reset_on_human_clears_all_bot_keys_in_chat_plus_sender(): + guard = _guard(window_ms=60000, max_bot_mentions=2, scope="chat+sender") + assert guard.record(_msg("m1", 0, sender="ou_botA")) is False + # A human (different sender key) must still clear the bot's counter. + guard.reset_on_human( + _msg("h", 10, sender="ou_human", sender_type="user", mentioned_bot=False) + ) + assert guard.record(_msg("m2", 20, sender="ou_botA")) is False + + +def _bot_at_me(mid): + m = _msg(mid, int(time.time() * 1000), chat="c_reset", sender="ou_peer") + m.mentions = [Mention(key="@_1", open_id="ou_bot")] + return m + + +def _human_plain(mid): + return _msg( + mid, int(time.time() * 1000), chat="c_reset", + sender="ou_human", sender_type="user", mentioned_bot=False, + ) + + +async def test_pipeline_human_message_resets_guard_before_policy(): + # require_mention=True → a plain human message is policy-rejected, yet it + # must still reset the guard (reset runs before the policy gate), so a + # subsequent bot @-mention doesn't trip. + loop = asyncio.get_running_loop() + delivered, rejects = [], [] + pipe = SafetyPipeline( + loop=loop, + on_message=lambda m: delivered.append(m.id), + on_reject=lambda r: rejects.append(r), + policy=PolicyConfig( + require_mention=True, + bot_loop_guard=BotLoopGuardConfig( + enabled=True, max_bot_mentions=2, window_ms=60_000, on_trip="reject" + ), + ), + batch_config=TextBatchConfig(delay_ms=0), + ) + pipe.set_bot_open_id("ou_bot") + + await pipe.push_message(_bot_at_me("b1")) + await asyncio.sleep(0.02) + await pipe.push_message(_human_plain("h1")) # policy-rejected, but resets guard + await asyncio.sleep(0.02) + await pipe.push_message(_bot_at_me("b2")) + await asyncio.sleep(0.02) + + assert not any(r.reason == "bot_loop" for r in rejects) # reset prevented the trip + assert "b2" in delivered + + +async def test_pipeline_update_policy_reconfigures_loop_guard(): + loop = asyncio.get_running_loop() + rejects = [] + pipe = SafetyPipeline( + loop=loop, + on_message=lambda m: None, + on_reject=lambda r: rejects.append(r), + policy=PolicyConfig(require_mention=False), # no guard initially + batch_config=TextBatchConfig(delay_ms=0), + ) + # Enable the guard at runtime; it must actually take effect on the pipeline. + pipe.update_policy( + bot_loop_guard=BotLoopGuardConfig( + enabled=True, max_bot_mentions=1, window_ms=60000, on_trip="reject" + ) + ) + await pipe.push_message(_recent_bot_msg()) + await asyncio.sleep(0.05) + assert len(rejects) == 1 and rejects[0].reason == "bot_loop" + + +async def test_pipeline_without_guard_delivers_bot_mention(): + loop = asyncio.get_running_loop() + delivered = [] + pipe = SafetyPipeline( + loop=loop, + on_message=lambda m: delivered.append(m), + policy=PolicyConfig(require_mention=False), # no bot_loop_guard + batch_config=TextBatchConfig(delay_ms=0), + ) + await pipe.push_message(_recent_bot_msg()) + await asyncio.sleep(0.05) + assert len(delivered) == 1 diff --git a/lark_channel/channel/tests/test_pipeline.py b/lark_channel/channel/tests/test_pipeline.py index ce77cd0..6d7b2bc 100644 --- a/lark_channel/channel/tests/test_pipeline.py +++ b/lark_channel/channel/tests/test_pipeline.py @@ -213,7 +213,11 @@ async def test_mentions_resolved_and_stripped(): @pytest.mark.asyncio -async def test_pipeline_sets_mentioned_bot_without_removing_bot_mention(): +async def test_pipeline_sets_mentioned_bot_keeps_content_strips_only_body_text(): + # Bot-at-bot: `content_text` and `content.text` keep the rendered bot + # mention (default behavior unchanged); only the new `body_text` view drops + # the bot's own @-mention. `mentioned_bot` stays True; the bot remains in + # `mentions`. msg = _msg( chat_type="group", content={"text": "hey @_user_1"}, @@ -230,6 +234,8 @@ async def test_pipeline_sets_mentioned_bot_without_removing_bot_mention(): assert inbound.mentioned_bot is True assert [m.open_id for m in inbound.mentions] == ["ou_bot"] assert inbound.content.text == "hey @Bot" + assert inbound.content_text == "hey @Bot" + assert inbound.body_text.strip() == "hey" @pytest.mark.asyncio diff --git a/lark_channel/channel/tests/test_policy_allowlist_warn.py b/lark_channel/channel/tests/test_policy_allowlist_warn.py new file mode 100644 index 0000000..484e5fc --- /dev/null +++ b/lark_channel/channel/tests/test_policy_allowlist_warn.py @@ -0,0 +1,61 @@ +"""allowlist / group_allowlist misconfiguration warn. + +``allow_from`` expects sender ids (ou_ / user_id / union_id) and +``group_allowlist`` expects chat ids (oc_). A ``cli_`` app id in either is +almost certainly a mistake that silently matches nobody. PolicyGate warns once +per offending field — naming the field and a single offending value only (no +full-table dump / PII) — without changing matching behaviour. +""" + +import logging + +from lark_channel.channel import Conversation, Identity, InboundMessage +from lark_channel.channel.config import PolicyConfig +from lark_channel.channel.safety.policy_gate import PolicyGate +from lark_channel.channel.types import TextContent + + +def _warnings(caplog): + return [r.getMessage() for r in caplog.records if r.levelno >= logging.WARNING] + + +def test_cli_in_allow_from_warns_with_field_and_value(caplog): + with caplog.at_level(logging.WARNING, logger="Lark"): + PolicyGate(PolicyConfig(allow_from=["cli_abc", "ou_valid"])) + + hits = [m for m in _warnings(caplog) if "allow_from" in m and "cli_abc" in m] + assert len(hits) == 1 + assert "ou_valid" not in hits[0] + + +def test_cli_in_group_allowlist_warns_with_field_and_value(caplog): + with caplog.at_level(logging.WARNING, logger="Lark"): + PolicyGate(PolicyConfig(group_allowlist=["cli_x", "oc_valid"])) + + hits = [m for m in _warnings(caplog) if "group_allowlist" in m and "cli_x" in m] + assert len(hits) == 1 + assert "oc_valid" not in hits[0] + + +def test_valid_lists_do_not_warn(caplog): + with caplog.at_level(logging.WARNING, logger="Lark"): + PolicyGate(PolicyConfig(allow_from=["ou_a"], group_allowlist=["oc_c"])) + + assert _warnings(caplog) == [] + + +def test_warn_does_not_change_matching(caplog): + # The cli_ entry still matches nobody: a real sender is rejected as usual. + with caplog.at_level(logging.WARNING, logger="Lark"): + gate = PolicyGate(PolicyConfig(dm_policy="allowlist", allow_from=["cli_x"])) + + msg = InboundMessage( + id="om_1", + create_time=1, + conversation=Conversation(chat_id="oc_c", chat_type="p2p"), + sender=Identity(open_id="ou_alice"), + content=TextContent(text="hi"), + ) + decision = gate.evaluate(msg) + assert decision.allowed is False + assert decision.reason == "policy_dm_not_in_allowlist" diff --git a/lark_channel/channel/tests/test_reject_reasons.py b/lark_channel/channel/tests/test_reject_reasons.py index 2e834f0..512178a 100644 --- a/lark_channel/channel/tests/test_reject_reasons.py +++ b/lark_channel/channel/tests/test_reject_reasons.py @@ -15,6 +15,7 @@ "duplicate", "lock_contention", "self_sent", + "bot_loop", # opt-in PolicyConfig.bot_loop_guard, on_trip='reject' # Policy "policy_dm_disabled", "policy_group_disabled", diff --git a/lark_channel/channel/tests/test_reply.py b/lark_channel/channel/tests/test_reply.py new file mode 100644 index 0000000..2b4706d --- /dev/null +++ b/lark_channel/channel/tests/test_reply.py @@ -0,0 +1,58 @@ +"""FeishuChannel.reply(). + +``reply(msg, message, opts)`` is send() that *follows* the triggering message: +it defaults ``reply_to`` to ``msg.message_id`` and ``reply_in_thread`` to +whether the trigger was in a thread. It never upgrades a flat message into a +thread, and ``opts`` can override both defaults. + +The outbound ``OutboundSender.send`` seam is mocked so the derived reply_to / +reply_in_thread are observable regardless of how ``reply`` routes internally. +""" + +from unittest.mock import AsyncMock + +from lark_channel.channel import Conversation, Identity, InboundMessage +from lark_channel.channel import FeishuChannel +from lark_channel.channel.types import SendOpts, SendResult, TextContent + + +def _channel(): + ch = FeishuChannel(app_id="cli_x", app_secret="s") + ch._sender.send = AsyncMock(return_value=SendResult.ok(message_id="om_out")) + return ch + + +def _inbound(*, thread_id=None): + return InboundMessage( + id="om_trigger", + create_time=1, + conversation=Conversation(chat_id="oc_c", chat_type="group", thread_id=thread_id), + sender=Identity(open_id="ou_a"), + content=TextContent(text="ping"), + ) + + +async def test_reply_in_thread_follows_trigger_thread(): + ch = _channel() + await ch.reply(_inbound(thread_id="omt_root"), "pong") + + kwargs = ch._sender.send.await_args.kwargs + assert kwargs["reply_to"] == "om_trigger" + assert kwargs["reply_in_thread"] is True + + +async def test_flat_trigger_does_not_force_thread(): + ch = _channel() + await ch.reply(_inbound(thread_id=None), "pong") + + kwargs = ch._sender.send.await_args.kwargs + assert kwargs["reply_to"] == "om_trigger" + assert kwargs.get("reply_in_thread") in (None, False) + + +async def test_opts_can_override_reply_in_thread_off(): + ch = _channel() + await ch.reply(_inbound(thread_id="omt_root"), "pong", SendOpts(reply_in_thread=False)) + + kwargs = ch._sender.send.await_args.kwargs + assert kwargs["reply_in_thread"] is False diff --git a/lark_channel/channel/tests/test_reply_thread_and_chunk.py b/lark_channel/channel/tests/test_reply_thread_and_chunk.py new file mode 100644 index 0000000..1b5e6d0 --- /dev/null +++ b/lark_channel/channel/tests/test_reply_thread_and_chunk.py @@ -0,0 +1,95 @@ +"""Long thread replies stay in-thread, and chunking never splits an tag.""" + +from typing import Any, Dict, List + +from lark_channel.channel.config import OutboundConfig +from lark_channel.channel.outbound.sender import OutboundSender, SendDriver, chunk_text +from lark_channel.channel.types import OutboundText + + +def _driver(): + calls: List[Dict[str, Any]] = [] + + async def create_message(**kwargs): + calls.append({"op": "create", **kwargs}) + return {"code": 0, "msg": "ok", "data": {"message_id": "om_new"}} + + async def reply_message(**kwargs): + calls.append({"op": "reply", **kwargs}) + return {"code": 0, "msg": "ok", "data": {"message_id": "om_reply"}} + + return SendDriver(create_message=create_message, reply_message=reply_message), calls + + +def _sender(calls_cfg=None): + driver, calls = _driver() + cfg = OutboundConfig(text_chunk_limit=10, chunk_mode="none") + return OutboundSender(driver, cfg), calls + + +# ── #2: long thread reply must keep every chunk in the thread ──────────────── + +async def test_thread_reply_keeps_all_chunks_in_thread(): + s, calls = _sender() + await s.send( + OutboundText(text="abcdefghij" * 3), # 30 chars → 3 chunks at limit 10 + receive_id="oc_1", + reply_to="om_1", + reply_in_thread=True, + ) + assert [c["op"] for c in calls] == ["reply", "reply", "reply"] + assert all(c.get("reply_in_thread") is True for c in calls) + + +async def test_flat_reply_only_first_chunk_replies(): + s, calls = _sender() + await s.send( + OutboundText(text="abcdefghij" * 3), + receive_id="oc_1", + reply_to="om_1", + # reply_in_thread unset → flat reply: legacy behavior preserved. + ) + ops = [c["op"] for c in calls] + assert ops[0] == "reply" + assert ops[1:] == ["create", "create"] + + +# ── #10: the chunker never splits an ... tag ──────────────────────── + +def _balanced(chunks): + return all(c.count("") for c in chunks) + + +def test_oversized_at_tag_emitted_whole(): + tag = 'Alice' # longer than the limit + text = ("x" * 20) + tag + ("y" * 20) + for mode in ("none", "newline", "paragraph"): + chunks = chunk_text(text, limit=25, mode=mode) + assert _balanced(chunks) + assert any(tag in c for c in chunks) # intact in exactly one chunk + + +def test_at_tag_pushed_whole_to_next_chunk(): + tag = 'A' + text = ("x" * 40) + tag + ("y" * 10) + chunks = chunk_text(text, limit=50, mode="none") + assert _balanced(chunks) + assert any(tag in c for c in chunks) + + +def test_many_unclosed_at_openers_bounded_time(): + # Adversarial: many unclosed " tags → identical to the delimiter chunker (regression guard). + text = "line1\nline2\nline3\nline4" + assert chunk_text(text, limit=12, mode="newline") == ["line1\nline2", "line3\nline4"] diff --git a/lark_channel/channel/tests/test_resolve_mentions.py b/lark_channel/channel/tests/test_resolve_mentions.py new file mode 100644 index 0000000..76d9b5f --- /dev/null +++ b/lark_channel/channel/tests/test_resolve_mentions.py @@ -0,0 +1,101 @@ +"""@name → open_id normalization pure functions. + +New module ``outbound/markdown/resolve_mentions.py``: + - resolve_name_mentions(mentions, lookup): fill open_id for name-only + Identities, drop unresolvable ones, pass valid ones through. + - resolve_mentions_in_text(text, lookup): rewrite ``@name`` to ```` only + for names the lookup resolves; unknown / ambiguous stay literal; @ mid-word + is not a mention; pathological input returns in bounded time (no ReDoS). + - is_valid_open_id / escape_at_name: the shared hardening primitives. + +``lookup(name) -> Optional[str]`` is Map-backed on purpose — roster names are +never compiled into a regex. +""" + +import time + +from lark_channel.channel.types import Identity +from lark_channel.channel.outbound.markdown.resolve_mentions import ( + escape_at_name, + is_valid_open_id, + resolve_mentions_in_text, + resolve_name_mentions, +) + +_LOOKUP = {"Alice": "ou_a"}.get + + +def test_resolve_name_mentions_fills_open_id_for_name_only(): + out = resolve_name_mentions([Identity(open_id="", display_name="Alice")], _LOOKUP) + assert len(out) == 1 + assert out[0].open_id == "ou_a" + + +def test_resolve_name_mentions_drops_unresolvable(): + out = resolve_name_mentions([Identity(open_id="", display_name="Ghost")], _LOOKUP) + assert out == [] + + +def test_resolve_name_mentions_passes_valid_open_id_through(): + out = resolve_name_mentions([Identity(open_id="ou_x", display_name="X")], _LOOKUP) + assert len(out) == 1 + assert out[0].open_id == "ou_x" + + +def test_resolve_text_rewrites_known_name_and_keeps_surrounding_text(): + out = resolve_mentions_in_text("你好 @Alice 请看", _LOOKUP) + assert '' in out + assert "你好" in out and "请看" in out + + +def test_resolve_text_leaves_unknown_name_literal(): + out = resolve_mentions_in_text("hi @Ghost there", _LOOKUP) + assert ". + assert is_valid_open_id("all") is True + + +def test_is_valid_open_id_rejects_cli_and_markup(): + assert is_valid_open_id("cli_bad") is False + assert is_valid_open_id("