diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index ce04883cd38ba..c113da303806b 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -221,21 +221,16 @@ VALID_HOOKS: Set[str] = { "kanban_task_completed", "kanban_task_blocked", # Gateway platform-boundary observer hooks (#64176). Observer-only; each - # callback isolated by invoke_hook. Normalized envelopes only — NO raw - # platform SDK objects in the payload (per #64176 / #64182 ground rule); - # raw access is a separate capability-gated action API, not a hook. + # callback isolated by invoke_hook. Payloads are normalized envelopes only, + # never raw platform SDK objects (per #64176 / #64182 ground rule); raw + # access is a separate capability-gated action API, not a hook. # - # gateway_platform_event — inbound platform event (reactions, forwards, - # edits, chat-member) as a normalized envelope. Kwargs: platform, - # event_type, payload (event_type-specific dict). Fired for Telegram - # reactions today; other event types + fire-sites land with #64176's - # taxonomy (#64231). - # gateway_session_titled / gateway_message_delivered / gateway_thread_created - # — reserved (fire-sites pending #64176); names registered so plugins - # can subscribe ahead of implementation. - "gateway_session_titled", - "gateway_message_delivered", - "gateway_thread_created", + # gateway_platform_event: inbound platform event as a normalized envelope. + # Kwargs: platform, event_type, payload (event_type-specific dict). + # Telegram reactions fire today. Other event types and their hook + # names land here together with their real fire-sites and payload + # contracts when #64176's taxonomy is finalized (#64231); no inert + # VALID_HOOKS surface is registered ahead of implementation. "gateway_platform_event", } diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 91cc773134a93..a4fd8bed17e7f 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -1056,6 +1056,62 @@ class TelegramAdapter(BasePlatformAdapter): thread_id=thread_id, ) + def _source_from_reaction_for_auth(self, update): + """Build the SessionSource for a ``message_reaction`` update's actor. + + Mirrors ``_source_from_message_for_auth`` but for reactions, which + carry the reactor (``user``, or ``actor_chat`` for an anonymous admin) + and ``chat`` but no ``Message``. Chat type is resolved as far as the + reaction shape allows so the shared authorization decision + (``_is_source_authorized``) matches the message intake outcome (the + runner treats group and forum identically, so the lack of a thread id + on reactions does not change the decision). Reactions expose no thread + id, so ``thread_id`` is None. + + Raises ``ValueError`` for a non-reaction update (no ``message_reaction``) + so the post-auth gate in ``_on_platform_update`` fails closed (drops the + event via its try/except) rather than resolving an empty identity and + authorizing it. This keeps a future event type from silently bypassing + auth before its own source extraction is wired. + """ + from gateway.session import SessionSource + + mr = getattr(update, "message_reaction", None) + if mr is None: + raise ValueError( + "gateway_platform_event source extraction requires a " + "message_reaction update" + ) + user = getattr(mr, "user", None) or getattr(mr, "actor_chat", None) + chat = getattr(mr, "chat", None) + user_id = str(getattr(user, "id", "")).strip() or None + user_name = ( + str( + getattr(user, "username", "") + or getattr(user, "full_name", "") + or getattr(user, "title", "") + ).strip() + or None + ) + + chat_id = str(getattr(chat, "id", "")).strip() or user_id + chat_type = str(getattr(chat, "type", "dm")).strip().lower() or "dm" + if chat_type == "private": + chat_type = "dm" + elif chat_type == "supergroup": + # Reactions carry no message_thread_id; a forum supergroup is the + # only forum signal available without the underlying message. + chat_type = "forum" if getattr(chat, "is_forum", False) is True else "group" + + return SessionSource( + platform=Platform.TELEGRAM, + chat_id=chat_id or "", + chat_type=chat_type, + user_id=user_id, + user_name=user_name, + thread_id=None, + ) + def _telegram_auth_env_configured(self) -> bool: """Return True when Telegram auth env vars make an early decision safe.""" keys = ( @@ -1114,7 +1170,19 @@ class TelegramAdapter(BasePlatformAdapter): Unknown DMs with an allowlist still pass through when pairing is the effective unauthorized-DM behavior (explicit platform override). """ - source = self._source_from_message_for_auth(message) + return self._is_source_authorized(self._source_from_message_for_auth(message)) + + def _is_source_authorized(self, source) -> bool: + """Authorization decision shared by message intake and the + ``gateway_platform_event`` observer (#64176 post-auth requirement). + + Same logic the message intake prefilter applies once a SessionSource is + resolved: adapter ``allow_from`` is the sole authority when set, then a + test-only callback override, then the runner's context-aware auth (only + when an allowlist is configured), then the env allowlist. Returns True + for an empty identity or an unconfigured allowlist so the cold path and + pairing flow still run, matching intake exactly. + """ user_id = source.user_id # No identity at all → genuine group service message (pin, delete, # new_chat_members, etc.). Defer to the cold path. Channel posts @@ -1139,7 +1207,7 @@ class TelegramAdapter(BasePlatformAdapter): # Test/custom injection only. The class method named # _is_callback_user_authorized is for inline button callbacks and must - # not be treated as a user-id-only shortcut for real messages — only + # not be treated as a user-id-only shortcut for real messages; only # honor an instance-level override (set in tests). if authorized is None: callback_auth = self.__dict__.get("_is_callback_user_authorized") @@ -3684,11 +3752,13 @@ class TelegramAdapter(BasePlatformAdapter): async def _on_platform_update(self, update, context) -> None: """Catch-all PTB handler firing ``gateway_platform_event`` per inbound update. - Normalizes the update into a stable envelope (no raw SDK objects — see - #64176) and fires the observer hook. Registered in a dedicated high - group so it observes alongside — never displaces — the core handlers. - Normalization is wrapped so a malformed update can't raise into PTB - dispatch — the observer can't break the adapter. + Normalizes the update into a stable envelope (no raw SDK objects; see + #64176), authorizes the actor on the same decision as inbound gateway + traffic, then fires the observer hook. Registered in a dedicated high + group so it observes alongside, never displaces, the core handlers. + Normalization and authorization are each wrapped so a malformed update + or a missing auth context can't raise into PTB dispatch: the observer + can't break the adapter. """ try: event = self._normalize_platform_event(update) @@ -3697,6 +3767,19 @@ class TelegramAdapter(BasePlatformAdapter): return if event is None: return + # Post-auth gate (#64176): the catch-all sees every inbound update, so a + # reaction from a sender the message intake would reject must not reach + # plugins. Reuse the intake's authorization decision verbatim. Fail + # closed (drop the event) if the decision itself raises. + try: + authorized = self._is_source_authorized( + self._source_from_reaction_for_auth(update) + ) + except Exception as exc: + logger.debug("[%s] gateway_platform_event auth error: %s", self.name, exc) + return + if not authorized: + return self._fire_gateway_hook("gateway_platform_event", **event) def _normalize_platform_event(self, update) -> Optional[Dict[str, Any]]: @@ -3739,6 +3822,40 @@ class TelegramAdapter(BasePlatformAdapter): }, } + def _register_handlers(self, app) -> None: + """Register every PTB handler on ``app``. + + Single source of truth for handler registration. ``connect`` calls this, + and any future rebuild path that re-creates the Application would call + it too, keeping the ``gateway_platform_event`` observer (group 99) in + lockstep with the core handlers. Today reconnect reuses the existing + Application, so handlers already persist; this method exists so there is + one place to add a handler and one re-registration point if a rebuild is + ever introduced. This is the #64176 review's "share registration between + initial and rebuild paths" ask. + """ + app.add_handler(TelegramMessageHandler( + filters.TEXT & ~filters.COMMAND, + self._handle_text_message + )) + app.add_handler(TelegramMessageHandler( + filters.COMMAND, + self._handle_command + )) + app.add_handler(TelegramMessageHandler( + filters.LOCATION | getattr(filters, "VENUE", filters.LOCATION), + self._handle_location_message + )) + app.add_handler(TelegramMessageHandler( + filters.PHOTO | filters.VIDEO | filters.AUDIO | filters.VOICE | filters.Document.ALL | filters.Sticker.ALL, + self._handle_media_message + )) + # Handle inline keyboard button callbacks (update prompts) + app.add_handler(CallbackQueryHandler(self._handle_callback_query)) + # gateway_platform_event observer (see _on_platform_update); group 99 so + # it observes alongside, never displaces, the core handlers. + app.add_handler(TypeHandler(Update, self._on_platform_update), group=99) + async def connect(self, *, is_reconnect: bool = False) -> bool: """Connect to Telegram via polling or webhook. @@ -3951,28 +4068,8 @@ class TelegramAdapter(BasePlatformAdapter): self._app = builder.build() self._bot = self._app.bot - # Register handlers - self._app.add_handler(TelegramMessageHandler( - filters.TEXT & ~filters.COMMAND, - self._handle_text_message - )) - self._app.add_handler(TelegramMessageHandler( - filters.COMMAND, - self._handle_command - )) - self._app.add_handler(TelegramMessageHandler( - filters.LOCATION | getattr(filters, "VENUE", filters.LOCATION), - self._handle_location_message - )) - self._app.add_handler(TelegramMessageHandler( - filters.PHOTO | filters.VIDEO | filters.AUDIO | filters.VOICE | filters.Document.ALL | filters.Sticker.ALL, - self._handle_media_message - )) - # Handle inline keyboard button callbacks (update prompts) - self._app.add_handler(CallbackQueryHandler(self._handle_callback_query)) - # gateway_platform_event observer (see _on_platform_update); group 99 - # so it observes alongside — never displaces — the core handlers. - self._app.add_handler(TypeHandler(Update, self._on_platform_update), group=99) + # Register handlers via the single registration site (#64176). + self._register_handlers(self._app) # Start polling — retry initialize() for transient TLS resets. # Each attempt is capped by _init_timeout so a single unreachable diff --git a/tests/gateway/test_gateway_platform_event_hook.py b/tests/gateway/test_gateway_platform_event_hook.py index 6f46cc0f03f08..03f85152fda6c 100644 --- a/tests/gateway/test_gateway_platform_event_hook.py +++ b/tests/gateway/test_gateway_platform_event_hook.py @@ -1,14 +1,19 @@ """Tests for the ``gateway_platform_event`` observer hook (#64176's observer half). Covers the normalized-envelope pattern that replaces raw-SDK handler args: -* the four ``gateway_*`` hooks are registered in ``VALID_HOOKS`` +* only ``gateway_platform_event`` is registered in ``VALID_HOOKS`` (no inert + hook surface pending #64231) * ``BasePlatformAdapter._fire_gateway_hook`` routes to ``invoke_hook`` with a ``has_hook`` no-subscriber fast-path and per-call error isolation * ``TelegramAdapter._normalize_platform_event`` maps an inbound PTB update to a stable ``{platform, event_type, payload}`` envelope (no raw SDK objects), including custom-emoji reactions -* ``_on_platform_update`` fires ``gateway_platform_event`` with that envelope - and swallows normalization errors so the observer can't break the adapter +* ``_on_platform_update`` fires ``gateway_platform_event`` with that envelope, + gated on the same authorization decision as inbound gateway traffic + (unauthorized reactions never fire), and swallows errors so the observer + can't break the adapter +* ``_register_handlers`` is the single PTB handler registration site, so a + rebuild re-registers the observer alongside the core handlers """ from __future__ import annotations @@ -51,14 +56,17 @@ from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 from hermes_cli.plugins import VALID_HOOKS # noqa: E402 -def _adapter() -> TelegramAdapter: +def _adapter(extra=None) -> TelegramAdapter: """Build a TelegramAdapter without the heavy __init__. - _fire_gateway_hook / _normalize_platform_event only need self.name (a - read-only property over self.platform), so set a stand-in platform. + _fire_gateway_hook / _normalize_platform_event / the post-auth gate only + need self.name (a read-only property over self.platform) and self.config, + so set stand-ins. The default config opens auth (allow_from=["*"]) so a + normal reaction fires; pass a restrictive ``extra`` to exercise the gate. """ a = object.__new__(TelegramAdapter) a.platform = SimpleNamespace(value="telegram") # name -> "Telegram" + a.config = SimpleNamespace(extra=extra if extra is not None else {"allow_from": ["*"]}) return a @@ -85,18 +93,33 @@ def _reaction_update(reactions, chat_id=123, message_id=456): return update +def _auth_reaction_update(user_id, chat_type="private", chat_id=123, message_id=456): + """A PTB Update stand-in carrying a message_reaction with an actor identity. + + Wraps ``_reaction_update`` and pins the reactor's user id + chat type so the + post-auth gate has an identity to authorize against. + """ + update = _reaction_update( + [_reaction(emoji="\U0001F44D")], chat_id=chat_id, message_id=message_id, + ) + update.message_reaction.user.id = str(user_id) + update.message_reaction.chat.type = chat_type + return update + + # --------------------------------------------------------------------------- # Hook registration # --------------------------------------------------------------------------- class TestHookRegistration: - def test_gateway_hooks_are_valid(self): - """register_hook rejects names not in VALID_HOOKS, so the four new - platform-boundary hooks must be present there.""" + def test_gateway_platform_event_registered_reserved_absent(self): + """register_hook rejects names not in VALID_HOOKS, so the implemented + hook must be present. The reserved gateway_* names are deliberately + absent (no inert surface pending #64231); lock that in.""" assert "gateway_platform_event" in VALID_HOOKS - assert "gateway_session_titled" in VALID_HOOKS - assert "gateway_message_delivered" in VALID_HOOKS - assert "gateway_thread_created" in VALID_HOOKS + assert "gateway_session_titled" not in VALID_HOOKS + assert "gateway_message_delivered" not in VALID_HOOKS + assert "gateway_thread_created" not in VALID_HOOKS # --------------------------------------------------------------------------- @@ -249,3 +272,127 @@ class TestOnPlatformUpdate: a._normalize_platform_event = boom # type: ignore[assignment] asyncio.run(a._on_platform_update(MagicMock(), context=MagicMock())) # must not raise + + +# --------------------------------------------------------------------------- +# TelegramAdapter._on_platform_update post-auth gate (#64176) +# --------------------------------------------------------------------------- + +class TestOnPlatformUpdateAuthGate: + """The catch-all sees every inbound update. A reaction from a sender the + message intake would reject must NOT reach plugins, using the same + authorization decision as inbound gateway traffic.""" + + def test_unauthorized_reaction_does_not_fire(self): + a = _adapter(extra={"allow_from": ["999"]}) # reactor 777 not allowed + seen: list = [] + a._fire_gateway_hook = lambda name, **kw: seen.append((name, kw)) # type: ignore[assignment] + + asyncio.run(a._on_platform_update( + _auth_reaction_update(user_id=777), context=MagicMock(), + )) + + assert seen == [] + + def test_authorized_reaction_fires(self): + a = _adapter(extra={"allow_from": ["777"]}) # reactor 777 allowed + seen: list = [] + a._fire_gateway_hook = lambda name, **kw: seen.append((name, kw)) # type: ignore[assignment] + + asyncio.run(a._on_platform_update( + _auth_reaction_update(user_id=777), context=MagicMock(), + )) + + assert len(seen) == 1 + assert seen[0][0] == "gateway_platform_event" + + def test_open_config_defers_to_pairing_flow(self): + """With no allow_from and no env allowlist, intake defers (open) and the + event fires, matching the message intake pairing flow.""" + a = _adapter(extra={}) + seen: list = [] + a._fire_gateway_hook = lambda name, **kw: seen.append((name, kw)) # type: ignore[assignment] + cleared = {k: "" for k in ( + "TELEGRAM_ALLOWED_USERS", "TELEGRAM_GROUP_ALLOWED_USERS", + "TELEGRAM_ALLOW_ALL_USERS", "GATEWAY_ALLOWED_USERS", + "GATEWAY_ALLOW_ALL_USERS", + )} + + with patch.dict("os.environ", cleared, clear=False): + asyncio.run(a._on_platform_update( + _auth_reaction_update(user_id=777), context=MagicMock(), + )) + + assert len(seen) == 1 + + def test_non_reaction_update_fails_closed(self): + """A future event type whose update carries no message_reaction must + NOT fire. _source_from_reaction_for_auth raises and the gate drops it + (fail closed); without the guard the no-identity path would authorize + and fire it despite the restrictive allow_from.""" + a = _adapter(extra={"allow_from": ["999"]}) + seen: list = [] + a._fire_gateway_hook = lambda name, **kw: seen.append((name, kw)) # type: ignore[assignment] + + update = MagicMock() + update.message_reaction = None # a future, not-yet-wired event type + # Simulate that future normalization produced an event for it. + a._normalize_platform_event = lambda u: { # type: ignore[assignment] + "platform": "telegram", "event_type": "future", "payload": {}, + } + + asyncio.run(a._on_platform_update(update, context=MagicMock())) + + assert seen == [] # fail closed: never fires without a real auth decision + + +# --------------------------------------------------------------------------- +# TelegramAdapter._register_handlers single registration site (#64176) +# --------------------------------------------------------------------------- + +class TestRegisterHandlers: + """_register_handlers is the sole PTB handler registration site, so a + handler added there is registered on every (re)build that calls it. The + #64176 review asked to share registration between the initial path and any + rebuild; these tests pin that the observer (group 99) is included alongside + the core handlers.""" + + _HANDLER_ATTRS = ( + "_handle_text_message", "_handle_command", "_handle_location_message", + "_handle_media_message", "_handle_callback_query", "_on_platform_update", + ) + + def _adapter_with_handlers(self) -> TelegramAdapter: + a = _adapter() + # Stand-ins for the bound handler methods. _register_handlers only + # passes them to add_handler, it never calls them. + for name in self._HANDLER_ATTRS: + setattr(a, name, object()) + return a + + @staticmethod + def _observer_calls(app): + return [c for c in app.add_handler.call_args_list if c.kwargs.get("group") == 99] + + def test_registers_core_handlers_plus_observer(self): + a = self._adapter_with_handlers() + app = MagicMock() + a._register_handlers(app) + + # Five core handlers (default group) plus the gateway_platform_event + # observer in group 99. + assert app.add_handler.call_count == 6 + assert len(self._observer_calls(app)) == 1 + + def test_rebuild_re_registers_observer(self): + """A second call on a fresh app (e.g. a future rebuild) re-registers + every handler, observer included.""" + a = self._adapter_with_handlers() + first_app = MagicMock() + rebuilt_app = MagicMock() + + a._register_handlers(first_app) + a._register_handlers(rebuilt_app) # the rebuild path + + assert rebuilt_app.add_handler.call_count == 6 + assert len(self._observer_calls(rebuilt_app)) == 1