From b54a307cfc77942d661b00f2a955915cded67e1c Mon Sep 17 00:00:00 2001 From: "mazhe.nerd" Date: Fri, 10 Jul 2026 16:32:20 +0800 Subject: [PATCH 1/3] feat(channel): add bot-at-bot support Support multiple bots collaborating in one chat (@-ing each other to hand off work). All additive and opt-in; default receive/send/normalize behavior is unchanged. - InboundMessage.sender_type / sender_is_bot: pass through the raw sender.sender_type (previously dropped); Identity.is_bot semantics unchanged. - get_bot_identity(): the bot's own identity; raises not_connected before connect. - get_chat_members() / get_chat_bots(): chat roster (users / bots), paginated and cached; resolve_chat_members hook to source the roster externally. - reply(msg, ...): replies to the trigger and follows its thread shape. - @name -> open_id: name-only structured mentions and, with resolve_mentions_in_text, @name tokens in text/markdown are resolved against the chat roster; unknown/ambiguous names are left as plain text. - hardening: escape attacker-controlled display names and validate open_ids at every sink (the @all sentinel is preserved). - policy.bot_loop_guard: opt-in sliding-window guard against bot ping-pong, drop or reject (reason bot_loop). - allow_from / group_allowlist: warn on a misplaced cli_ app id. Version 1.1.0 -> 1.2.0. --- README.md | 2 +- README.zh.md | 2 +- docs/reference.md | 83 +++++ lark_channel/channel/__init__.py | 4 + lark_channel/channel/channel.py | 306 +++++++++++++++++- lark_channel/channel/chat_member_cache.py | 200 ++++++++++++ lark_channel/channel/config.py | 44 ++- lark_channel/channel/normalize/mentions.py | 5 +- lark_channel/channel/normalize/pipeline.py | 25 +- .../outbound/markdown/resolve_mentions.py | 158 +++++++++ .../channel/outbound/markdown/to_post.py | 9 +- lark_channel/channel/outbound/sender.py | 9 +- lark_channel/channel/safety/__init__.py | 2 + lark_channel/channel/safety/loop_guard.py | 97 ++++++ lark_channel/channel/safety/pipeline.py | 16 +- lark_channel/channel/safety/policy_gate.py | 37 ++- lark_channel/channel/safety/types.py | 2 + .../channel/tests/test_channel_low_level.py | 4 +- lark_channel/channel/tests/test_chat_bots.py | 116 +++++++ .../channel/tests/test_chat_member_cache.py | 84 +++++ .../channel/tests/test_chat_members.py | 189 +++++++++++ .../tests/test_compose_mentions_escaping.py | 69 ++++ .../channel/tests/test_empty_at_wake.py | 93 ++++++ .../channel/tests/test_get_bot_identity.py | 38 +++ lark_channel/channel/tests/test_loop_guard.py | 160 +++++++++ lark_channel/channel/tests/test_pipeline.py | 8 +- .../tests/test_policy_allowlist_warn.py | 61 ++++ .../channel/tests/test_reject_reasons.py | 1 + lark_channel/channel/tests/test_reply.py | 58 ++++ .../channel/tests/test_resolve_mentions.py | 101 ++++++ .../channel/tests/test_security_config.py | 9 +- .../tests/test_sender_name_resolution.py | 36 +++ .../channel/tests/test_sender_type.py | 72 +++++ lark_channel/channel/types.py | 42 +++ lark_channel/core/const.py | 2 +- tests/bridge/test_media_cache.py | 6 +- .../package_identity/test_import_identity.py | 4 +- 37 files changed, 2125 insertions(+), 29 deletions(-) create mode 100644 lark_channel/channel/chat_member_cache.py create mode 100644 lark_channel/channel/outbound/markdown/resolve_mentions.py create mode 100644 lark_channel/channel/safety/loop_guard.py create mode 100644 lark_channel/channel/tests/test_chat_bots.py create mode 100644 lark_channel/channel/tests/test_chat_member_cache.py create mode 100644 lark_channel/channel/tests/test_chat_members.py create mode 100644 lark_channel/channel/tests/test_compose_mentions_escaping.py create mode 100644 lark_channel/channel/tests/test_empty_at_wake.py create mode 100644 lark_channel/channel/tests/test_get_bot_identity.py create mode 100644 lark_channel/channel/tests/test_loop_guard.py create mode 100644 lark_channel/channel/tests/test_policy_allowlist_warn.py create mode 100644 lark_channel/channel/tests/test_reply.py create mode 100644 lark_channel/channel/tests/test_resolve_mentions.py create mode 100644 lark_channel/channel/tests/test_sender_name_resolution.py create mode 100644 lark_channel/channel/tests/test_sender_type.py 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..2da576b 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -150,6 +150,8 @@ 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 | @@ -378,6 +380,87 @@ 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` permission enabled — and the failure is silent. There is no API to +self-check this; if bot-to-bot mentions never arrive, verify that permission +first. An `@`-only ping (no body) still wakes the bot: `mentioned_bot` is `True` +while `content_text` is empty. Detect it with +`msg.mentioned_bot and not msg.content_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 +(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/channel.py b/lark_channel/channel/channel.py index 61011e6..1ee1fe8 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,23 @@ UATConfig, ) from .driver import LarkClientDriver -from .errors import FeishuChannelError, FeishuChannelErrorCode, OutboundSendError, SendError +from .errors import ( + FeishuChannelError, + FeishuChannelErrorCode, + OutboundSendError, + SendError, + classify_api_error, +) 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 +111,7 @@ CardActionEvent, CardActionPayload, ChatInfo, + ChatMember, CommentContext, CommentTarget, ConnectionSnapshot, @@ -400,6 +418,10 @@ 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() self._media_cache = MediaResourceCache(cfg.media_cache) self._token_store: TokenStore = token_store or InMemoryTokenStore() @@ -572,6 +594,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 +1492,15 @@ async def _handle_message_event(self, data: Any) -> None: ) if inbound is None: return + # 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. Best-effort; source 'mention' never + # overwrites authoritative 'api' names. + self._collect_mentions_into_roster(inbound) + # Opt-in: fill sender_name from the chat roster (warmed via a cached + # get_chat_members). Off by default → no extra roster API call. + if self._config.resolve_sender_names: + await self._resolve_sender_name_from_roster(inbound) if self._safety is not None: await self._safety.push_message(inbound) else: @@ -1458,6 +1508,36 @@ 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) + 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: await self._invoke("message", inbound) @@ -1835,6 +1915,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 +1938,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 +2445,174 @@ 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 not force: + cached = self._chat_member_cache.get_members(chat_id) + if cached is not None: + return cached + members = await self._fetch_chat_members( + chat_id, page_size=page_size, max_pages=max_pages, id_type=id_type + ) + self._chat_member_cache.set_members(chat_id, members, "api") + return members + + async def _fetch_chat_members( + self, chat_id: str, *, page_size: int, max_pages: int, id_type: str + ) -> List[ChatMember]: + hook = self._config.resolve_chat_members + if hook is not None: + result = hook(chat_id) + if inspect.isawaitable(result): + result = await result + # 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: + return list(result) + + size = min(max(page_size, 1), 100) + out: List[ChatMember] = [] + page_token: Optional[str] = None + for _ 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, + ) + ) + if not data.get("has_more") or not data.get("page_token"): + break + page_token = data.get("page_token") + return out + + 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._fetch_chat_bots(chat_id) + self._chat_member_cache.set_bots(chat_id, bots) + return bots + + 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() + # Raw Transport bypasses the Chain that normally injects the tenant + # token; call verify explicitly (see bot_identity for the same pattern). + _token_auth.verify(self._client.config, req, option) + try: + resp = await Transport.aexecute(self._client.config, req, option) + 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" + ) + try: + body = json.loads(resp.content.decode("utf-8")) + except (ValueError, UnicodeDecodeError) as e: + raise FeishuChannelError( + FeishuChannelErrorCode.UNKNOWN, "malformed roster response", cause=e + ) from e + 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}, + ) + 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..9c7a56c --- /dev/null +++ b/lark_channel/channel/chat_member_cache.py @@ -0,0 +1,200 @@ +"""Per-chat roster cache backing bot-at-bot name resolution. + +Consumers: :meth:`FeishuChannel.get_chat_members` (source ``'api'``, +authoritative users), :meth:`FeishuChannel.get_chat_bots` (source ``'api'``, +authoritative bots), inbound-mention collection (source ``'mention'``, can carry +bots), ``sender_name`` resolution and ``@name → open_id`` normalization. + +Two safety invariants: + +- A display name that maps to more than one open_id is **ambiguous** and + resolves to ``None`` — never last-writer-wins — so a name collision (e.g. an + attacker renaming to a real bot's name) can't misroute an ``@``. An ``'api'`` + name→open_id is never silently replaced by a ``'mention'`` one. +- Entries expire after a TTL and the number of cached chats is capped, so stale + or poisoned mappings don't linger. + +Clock, TTL and capacity are injectable so time and eviction are deterministic in +tests. Modelled on :class:`~lark_channel.channel.chat_mode.ChatModeCache`. +""" + +import time +from collections import OrderedDict +from dataclasses import dataclass, field +from typing import Callable, Dict, List, Optional + +from .types import ChatMember + +MemberSource = str # "api" | "mention" + +# Sentinel for a display name shared by more than one open_id — unresolvable. +_AMBIGUOUS = object() + +DEFAULT_TTL_SECONDS = 300.0 +DEFAULT_MAX_CHATS = 500 +DEFAULT_MAX_ENTRIES_PER_CHAT = 1000 + + +@dataclass +class _Roster: + # open_id → display name. 'api' names win over 'mention' names for the same open_id. + by_open_id: Dict[str, str] = field(default_factory=dict) + # display name → single open_id, or _AMBIGUOUS when the name is shared. + by_name: Dict[str, object] = field(default_factory=dict) + # open_id → which source last set its name, so 'api' isn't overwritten by 'mention'. + name_source: Dict[str, MemberSource] = field(default_factory=dict) + # Last user list from get_chat_members — served back on a cache hit, with its + # own fetch timestamp so 'mention' writes don't keep the API cache alive. + api_members: Optional[List[ChatMember]] = None + api_fetched_at: Optional[float] = None + # Last bot list from get_chat_bots — cached separately so it can't clobber api_members. + api_bots: Optional[List[ChatMember]] = None + api_bots_fetched_at: Optional[float] = None + 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, + ) -> None: + self._now = now + self._ttl = ttl_seconds + self._max_chats = max_chats + self._max_entries = max_entries_per_chat + self._chats: "OrderedDict[str, _Roster]" = OrderedDict() + + # ---- writes -------------------------------------------------------------- + + def set_members( + self, chat_id: str, members: List[ChatMember], source: MemberSource + ) -> None: + roster = self._roster_for_write(chat_id) + self._index_all(roster, members, source) + if source == "api": + roster.api_members = list(members) + roster.api_fetched_at = self._now() + roster.updated_at = self._now() + self._store(chat_id, roster) + + def set_bots(self, chat_id: str, bots: List[ChatMember]) -> None: + """Cache an authoritative bot list (from ``get_chat_bots``) — separate + from the user list so neither clobbers the other. Indexed as ``'api'``.""" + roster = self._roster_for_write(chat_id) + self._index_all(roster, bots, "api") + roster.api_bots = list(bots) + roster.api_bots_fetched_at = self._now() + roster.updated_at = self._now() + self._store(chat_id, roster) + + # ---- reads --------------------------------------------------------------- + + def get_members(self, chat_id: str) -> Optional[List[ChatMember]]: + """The last API user list, or ``None`` when absent or expired.""" + return self._live_list(chat_id, "members") + + def get_bots(self, chat_id: str) -> Optional[List[ChatMember]]: + """The last API bot list, or ``None`` when absent or expired.""" + return self._live_list(chat_id, "bots") + + def resolve_name(self, chat_id: str, open_id: str) -> Optional[str]: + roster = self._live_roster(chat_id) + return roster.by_open_id.get(open_id) if roster else None + + def resolve_open_id(self, chat_id: str, name: str) -> Optional[str]: + roster = self._live_roster(chat_id) + if roster is None: + return None + target = roster.by_name.get(name) + return target if isinstance(target, str) else None + + # ---- internals ----------------------------------------------------------- + + def _index_all( + self, roster: _Roster, members: List[ChatMember], source: MemberSource + ) -> None: + for m in members: + if not m.id or not m.name: + continue + self._index_member(roster, m.id, m.name, source) + + def _index_member( + self, roster: _Roster, open_id: str, name: str, source: MemberSource + ) -> None: + # open_id → name: an 'api' name is authoritative; don't let a later + # 'mention' name overwrite it, but a fresh 'api' name always wins. + prev_source = roster.name_source.get(open_id) + prev_name = roster.by_open_id.get(open_id) + if source == "api" or prev_source != "api": + roster.by_open_id[open_id] = name + roster.name_source[open_id] = source + # On a rename, drop the previous name→open_id entry so the reverse + # index doesn't accumulate every historical display name — but only + # when it still uniquely pointed here (leave a shared/ambiguous name). + if ( + prev_name is not None + and prev_name != name + and roster.by_name.get(prev_name) == open_id + ): + del roster.by_name[prev_name] + + # name → open_id: a second distinct open_id for the same name makes it + # ambiguous (unresolvable) regardless of source. + existing = roster.by_name.get(name) + if existing is None: + roster.by_name[name] = open_id + elif existing != open_id: + roster.by_name[name] = _AMBIGUOUS + + self._cap_entries(roster) + + def _cap_entries(self, roster: _Roster) -> None: + _evict_oldest(roster.by_name, self._max_entries) + _evict_oldest(roster.by_open_id, self._max_entries) + _evict_oldest(roster.name_source, self._max_entries) + + 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 _live_list( + self, chat_id: str, kind: str + ) -> Optional[List[ChatMember]]: + roster = self._chats.get(chat_id) + if roster is None: + return None + if kind == "members": + lst, fetched_at = roster.api_members, roster.api_fetched_at + else: + lst, fetched_at = roster.api_bots, roster.api_bots_fetched_at + if lst is None or fetched_at is None: + return None + if self._now() - fetched_at > self._ttl: + return None + return lst + + def _store(self, chat_id: str, roster: _Roster) -> None: + # LRU touch + evict oldest chats past the cap. + self._chats.pop(chat_id, None) + self._chats[chat_id] = roster + while len(self._chats) > self._max_chats: + self._chats.popitem(last=False) + + +def _evict_oldest(mapping: Dict, max_size: int) -> None: + """Drop oldest-inserted keys until ``mapping`` is within ``max_size``.""" + while len(mapping) > max_size: + oldest = next(iter(mapping)) + del mapping[oldest] diff --git a/lark_channel/channel/config.py b/lark_channel/channel/config.py index 969ff6b..d87bb6d 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,19 @@ 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 provided and it returns a member list, that list is used and + # the Feishu API call is skipped; return ``None`` to fall back to the API. + # The callable takes a ``chat_id`` and may be sync or async; results still + # flow through the internal roster cache. (Typed ``Any`` to avoid importing + # ``ChatMember`` from ``types`` — that would create an import cycle back + # through this module.) + resolve_chat_members: Optional[Callable[[str], Any]] = None 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..d007f63 100644 --- a/lark_channel/channel/normalize/pipeline.py +++ b/lark_channel/channel/normalize/pipeline.py @@ -68,18 +68,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 +194,11 @@ async def process( ext = extract_mentions(raw_mentions) mentions: List[Mention] = list(ext.mention_list) mentioned_all = ext.mentioned_all + # Strip the bot's own @-mention from rendered content (only when this + # message actually mentions the bot, so non-bot messages are untouched): + # a bare "@bot" wake then normalizes to empty content, and the documented + # `mentioned_bot and not content_text.strip()` ping detection works. + strip_bot = bool(self._bot_open_id) and self._bot_open_id in ext.mentions_by_open_id if isinstance(content, TextContent): # Feishu frequently ships ``@all`` messages with # ``mentions = null`` — the only signal is an ``@_all`` @@ -199,11 +208,15 @@ async def process( # silently skipped. if not mentioned_all and text_has_mention_all(content.text): mentioned_all = True - content.text = resolve_mentions(content.text, ext) + content.text = resolve_mentions( + content.text, ext, strip_bot_mentions=strip_bot, bot_open_id=self._bot_open_id + ) elif isinstance(content, PostContent): if not mentioned_all and text_has_mention_all(content.text): mentioned_all = True - content.text = resolve_mentions(content.text, ext) + content.text = resolve_mentions( + content.text, ext, strip_bot_mentions=strip_bot, bot_open_id=self._bot_open_id + ) at_mentions, at_all, stripped = parse_at_tags(content.text) content.text = stripped # Merge -tag mentions in (dedup by open_id/user_id/key). @@ -292,7 +305,9 @@ async def process( # merge_forward child content (parsed but not yet resolved because # flatten walks children sync without a ctx). Idempotent on text that # contains no placeholders. - flat_text = resolve_mentions(flat_text, ext) + flat_text = resolve_mentions( + flat_text, ext, strip_bot_mentions=strip_bot, bot_open_id=self._bot_open_id + ) safe_flat_text = _safe_content_text(flat_text) content_text = safe_flat_text if self._cfg.security.strict_content_text else flat_text 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..1760707 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, @@ -182,11 +183,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)} 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/loop_guard.py b/lark_channel/channel/safety/loop_guard.py new file mode 100644 index 0000000..0263e1e --- /dev/null +++ b/lark_channel/channel/safety/loop_guard.py @@ -0,0 +1,97 @@ +"""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 (and, being event-supplied, only best-effort — a backstop, not a +protocol-level defense). +""" + +from collections import OrderedDict +from typing import Any, Dict, List, Optional, Tuple + +from ..config import BotLoopGuardConfig +from ..types import InboundMessage + +# 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 (which keeps the sliding-window cutoff from advancing) +# can't accumulate distinct message_ids without bound. Only the count relative +# to the threshold matters, so trimming the oldest entries past this cap is safe +# and keeps counting monotonic toward "tripped". +_MAX_ENTRIES_PER_KEY = 1024 + + +class LoopGuard: + def __init__(self, cfg: Optional[BotLoopGuardConfig], logger: Any) -> None: + cfg = cfg or BotLoopGuardConfig() + self.enabled: bool = cfg.enabled + self.on_trip: str = cfg.on_trip + self._window_ms: int = cfg.window_ms + self._threshold: int = cfg.max_bot_mentions + self._scope: str = cfg.scope + self._logger = logger + # key -> {"entries": List[(message_id, create_time)], "warned": bool} + self._states: "OrderedDict[str, Dict[str, Any]]" = OrderedDict() + + def record(self, msg: InboundMessage) -> 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 emits exactly + one warning. + """ + if not self.enabled: + return False + 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} + entries: List[Tuple[str, int]] = state["entries"] + cutoff = msg.create_time - self._window_ms + entries = [(mid, t) for (mid, t) in entries if t >= cutoff] + if not any(mid == msg.id for (mid, _t) in entries): + entries.append((msg.id, msg.create_time)) + # Backstop against an adversarial constant/far-future create_time that + # freezes the cutoff so entries never slide out: bound the list length. + if len(entries) > _MAX_ENTRIES_PER_KEY: + entries = entries[-_MAX_ENTRIES_PER_KEY:] + state["entries"] = entries + + 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: + # Re-arm the one-time warn once the window drains below threshold. + 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: InboundMessage) -> 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..06539d5 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, @@ -166,6 +169,17 @@ 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). Runs after dedup + self_sent + + # policy so a re-delivery can't inflate the count and a + # policy-rejected message never counts; a human message (handled + # inside record) resets the window. + 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_channel_low_level.py b/lark_channel/channel/tests/test_channel_low_level.py index 26a332b..b359edc 100644 --- a/lark_channel/channel/tests/test_channel_low_level.py +++ b/lark_channel/channel/tests/test_channel_low_level.py @@ -137,7 +137,9 @@ async def test_fetch_inbound_message_returns_normalized_message_without_changing assert inbound is not None assert inbound.id == "om_1" - assert inbound.content_text == "hello @Bot" + # The bot's own @-mention is stripped from rendered content (bot-at-bot); + # the body survives and mentioned_bot stays True. + assert inbound.content_text == "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..d47ddad --- /dev/null +++ b/lark_channel/channel/tests/test_chat_member_cache.py @@ -0,0 +1,84 @@ +"""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_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..fd20f26 --- /dev/null +++ b/lark_channel/channel/tests/test_chat_members.py @@ -0,0 +1,189 @@ +"""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 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_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..542cdb5 --- /dev/null +++ b/lark_channel/channel/tests/test_empty_at_wake.py @@ -0,0 +1,93 @@ +"""Empty @-mention wake. + +A message that only @-mentions the bot with no body must still be delivered +(not dropped as empty): ``mentioned_bot`` is True and ``content_text`` is empty, +so downstream code can detect a bare "poke" via +``mentioned_bot and not content_text.strip()``. +""" + +import pytest + +from lark_channel.channel.normalize.pipeline import ( + InboundPipeline, + PipelineConfig, + PipelineDeps, +) + + +async def test_at_only_message_is_delivered_with_empty_body(): + # A REAL bare-@ event carries the mention placeholder in the text (e.g. + # "@_user_1 "); the bot's own mention must be stripped so content_text + # normalizes to empty — not rendered as "@BotName". + pipeline = InboundPipeline(PipelineConfig(), PipelineDeps()) + pipeline.set_bot_open_id("ou_bot") + + inbound = await pipeline.process( + event_id="e1", + message_event={ + "message_id": "om_1", + "create_time": 1, + "chat_id": "oc_c", + "chat_type": "group", + "message_type": "text", + "content": {"text": "@_user_1 "}, + "mentions": [{"key": "@_user_1", "id": {"open_id": "ou_bot"}, "name": "Bot"}], + }, + sender={"sender_id": {"open_id": "ou_human"}, "sender_type": "user"}, + ) + + assert inbound is not None # not dropped + assert inbound.mentioned_bot is True + assert inbound.content_text.strip() == "" + + +async def test_bot_mention_stripped_but_body_preserved(): + # "@bot hello" → the bot mention is removed but the real body survives, so + # the ping heuristic (mentioned_bot and not content_text.strip()) correctly + # treats this as NOT a bare poke. + pipeline = InboundPipeline(PipelineConfig(), PipelineDeps()) + pipeline.set_bot_open_id("ou_bot") + + inbound = await pipeline.process( + event_id="e2", + message_event={ + "message_id": "om_2", + "create_time": 1, + "chat_id": "oc_c", + "chat_type": "group", + "message_type": "text", + "content": {"text": "@_user_1 hello there"}, + "mentions": [{"key": "@_user_1", "id": {"open_id": "ou_bot"}, "name": "Bot"}], + }, + sender={"sender_id": {"open_id": "ou_human"}, "sender_type": "user"}, + ) + + assert inbound is not None + assert inbound.mentioned_bot is True + assert inbound.content_text.strip() == "hello there" + assert "@Bot" not in inbound.content_text + + +async def test_other_user_mention_not_stripped(): + # A message that @-mentions someone who is NOT the bot keeps that mention + # rendered as @Name (only the bot's own mention is stripped). + pipeline = InboundPipeline(PipelineConfig(), PipelineDeps()) + pipeline.set_bot_open_id("ou_bot") + + inbound = await pipeline.process( + event_id="e3", + message_event={ + "message_id": "om_3", + "create_time": 1, + "chat_id": "oc_c", + "chat_type": "group", + "message_type": "text", + "content": {"text": "@_user_1 ping"}, + "mentions": [{"key": "@_user_1", "id": {"open_id": "ou_alice"}, "name": "Alice"}], + }, + sender={"sender_id": {"open_id": "ou_human"}, "sender_type": "user"}, + ) + + assert inbound is not None + assert inbound.mentioned_bot is False + assert "@Alice" in 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..ee0b452 --- /dev/null +++ b/lark_channel/channel/tests/test_loop_guard.py @@ -0,0 +1,160 @@ +"""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, 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" + + +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..b33793c 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_and_strips_bot_mention_from_content(): + # Bot-at-bot: the bot's own @-mention is stripped from the + # rendered content (so a bare "@bot" wake normalizes to empty and the + # `mentioned_bot and not content_text.strip()` ping heuristic works), while + # `mentioned_bot` stays True and the bot remains in `mentions`. msg = _msg( chat_type="group", content={"text": "hey @_user_1"}, @@ -229,7 +233,7 @@ async def test_pipeline_sets_mentioned_bot_without_removing_bot_mention(): assert inbound is not None 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" @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_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("