feat(telegram): auto-approve groups via HSM when admin adds the bot

.env.example

@@ -361,6 +361,8 @@ BROWSER_INACTIVITY_TIMEOUT=120
 # TELEGRAM_HOME_CHANNEL=                   # Default chat for cron delivery
 # TELEGRAM_HOME_CHANNEL_NAME=              # Display name for home channel
 # TELEGRAM_CRON_THREAD_ID=                 # Forum topic ID for cron deliveries; overrides TELEGRAM_HOME_CHANNEL_THREAD_ID for cron so replies work in topic mode
+# TELEGRAM_ALLOW_BOTS=none                 # Accept messages from other bots: none (default) | mentions | all
+# TELEGRAM_GROUP_AUTOAPPROVE=false         # When true, gate "bot added to group" through HSM auto-approval (requires HSM_URL + HERMES_AGENT_NAME); adder must be a platform admin or the bot declines and leaves. Default false = log only.
 
 # Webhook mode (optional — for cloud deployments like Fly.io/Railway)
 # Default is long polling. Setting TELEGRAM_WEBHOOK_URL switches to webhook mode.

plugins/platforms/telegram/adapter.py

@@ -215,6 +215,7 @@ async def _shutdown_abandoned_app(app) -> None:
         Application,
         CommandHandler,
         CallbackQueryHandler,
+        ChatMemberHandler,
         MessageHandler as TelegramMessageHandler,
         ContextTypes,
         filters,
@@ -233,6 +234,7 @@ async def _shutdown_abandoned_app(app) -> None:
     Application = Any
     CommandHandler = Any
     CallbackQueryHandler = Any
+    ChatMemberHandler = Any
     TelegramMessageHandler = Any
     HTTPXRequest = Any
     filters = None
@@ -373,7 +375,8 @@ def check_telegram_requirements() -> bool:
     """
     global TELEGRAM_AVAILABLE, Update, Bot, Message, InlineKeyboardButton
     global InlineKeyboardMarkup, LinkPreviewOptions, Application
-    global CommandHandler, CallbackQueryHandler, TelegramMessageHandler
+    global CommandHandler, CallbackQueryHandler, ChatMemberHandler
+    global TelegramMessageHandler
     global ContextTypes, filters, ParseMode, ChatType, HTTPXRequest
     if TELEGRAM_AVAILABLE:
         return True
@@ -392,6 +395,7 @@ def check_telegram_requirements() -> bool:
         from telegram.ext import (
             Application as _App, CommandHandler as _CH,
             CallbackQueryHandler as _CQH,
+            ChatMemberHandler as _CMH,
             MessageHandler as _MH,
             ContextTypes as _CT, filters as _filters,
         )
@@ -408,6 +412,7 @@ def check_telegram_requirements() -> bool:
     Application = _App
     CommandHandler = _CH
     CallbackQueryHandler = _CQH
+    ChatMemberHandler = _CMH
     TelegramMessageHandler = _MH
     ContextTypes = _CT
     filters = _filters
@@ -3589,6 +3594,12 @@ def _with_limits(httpx_kwargs: Optional[dict] = None) -> dict:
             ))
             # Handle inline keyboard button callbacks (update prompts)
             self._app.add_handler(CallbackQueryHandler(self._handle_callback_query))
+            # Bot's own membership changes (added to / removed from chats) —
+            # drives the HSM group auto-approval gate (TELEGRAM_GROUP_AUTOAPPROVE).
+            self._app.add_handler(ChatMemberHandler(
+                self._handle_my_chat_member,
+                ChatMemberHandler.MY_CHAT_MEMBER,
+            ))
 
             # Start polling — retry initialize() for transient TLS resets.
             # Each attempt is capped by _init_timeout so a single unreachable
@@ -8127,6 +8138,115 @@ def _effective_update_message(self, update: Update) -> Optional[Message]:
         """
         return getattr(update, "effective_message", None) or getattr(update, "message", None)
 
+    def _telegram_group_autoapprove_enabled(self) -> bool:
+        """Return whether bot-added-to-group events are gated through HSM."""
+        return os.getenv("TELEGRAM_GROUP_AUTOAPPROVE", "false").lower() in {"true", "1", "yes", "on"}
+
+    async def _handle_my_chat_member(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+        """Handle changes to the bot's OWN chat membership (my_chat_member).
+
+        When the bot is newly added to a group/supergroup and
+        ``TELEGRAM_GROUP_AUTOAPPROVE`` is enabled, ask HSM to auto-approve the
+        group: HSM checks whether the adder is a platform admin and, if so,
+        adds the group to the allowlist. On approval the bot greets the group;
+        otherwise it politely declines and leaves (fail-closed — any error on
+        the enforcement path counts as not approved).
+
+        With the flag unset (default) the event is only logged, preserving
+        prior behavior for fleets that deploy this adapter before the HSM
+        endpoint is live.
+        """
+        change = getattr(update, "my_chat_member", None)
+        if change is None:
+            return
+
+        chat = getattr(change, "chat", None)
+        chat_type = str(getattr(chat, "type", "") or "").lower()
+        if chat_type not in ("group", "supergroup"):
+            return
+
+        old_status = str(
+            getattr(getattr(change, "old_chat_member", None), "status", "") or ""
+        ).lower()
+        new_status = str(
+            getattr(getattr(change, "new_chat_member", None), "status", "") or ""
+        ).lower()
+
+        # Only act on transitions INTO the chat. Promotions/demotions
+        # (member <-> administrator) re-fire my_chat_member and must not
+        # re-trigger approval; the bot leaving/being kicked is ignored too.
+        if new_status not in ("member", "administrator"):
+            return
+        if old_status not in ("", "left", "kicked"):
+            return
+
+        chat_id = getattr(chat, "id", None)
+        if chat_id is None:
+            return
+        group_id = str(chat_id)
+        adder = str(getattr(getattr(change, "from_user", None), "id", "") or "")
+
+        if not self._telegram_group_autoapprove_enabled():
+            logger.info(
+                "[%s] Bot added to Telegram group %s by user %s — "
+                "auto-approval disabled (TELEGRAM_GROUP_AUTOAPPROVE unset), no action",
+                self.name, group_id, adder or "unknown",
+            )
+            return
+
+        approved = False
+        try:
+            from plugins.swarm_map_policy import approve_group_add
+            approved = approve_group_add(group_id, adder, platform="telegram")
+        except Exception as e:
+            logger.warning(
+                "[%s] Group auto-approval check failed for %s (fail-closed): %s",
+                self.name, group_id, e,
+            )
+            approved = False
+
+        if approved:
+            logger.info(
+                "[%s] Telegram group %s approved via HSM (added by %s) — staying",
+                self.name, group_id, adder or "unknown",
+            )
+            try:
+                await self._bot.send_message(
+                    chat_id=chat_id,
+                    text="Hi! This group has been approved for me — I'm ready to help.",
+                )
+            except Exception as e:
+                logger.warning(
+                    "[%s] Failed to send group greeting to %s: %s",
+                    self.name, group_id, _redact_telegram_error_text(e),
+                )
+            return
+
+        logger.info(
+            "[%s] Telegram group %s not approved via HSM (added by %s) — leaving",
+            self.name, group_id, adder or "unknown",
+        )
+        try:
+            await self._bot.send_message(
+                chat_id=chat_id,
+                text=(
+                    "Sorry — this group isn't approved for me yet. "
+                    "Ask my admin to approve it, then add me again."
+                ),
+            )
+        except Exception as e:
+            logger.warning(
+                "[%s] Failed to send decline notice to %s: %s",
+                self.name, group_id, _redact_telegram_error_text(e),
+            )
+        try:
+            await self._bot.leave_chat(chat_id=chat_id)
+        except Exception as e:
+            logger.warning(
+                "[%s] Failed to leave unapproved group %s: %s",
+                self.name, group_id, _redact_telegram_error_text(e),
+            )
+
     async def _handle_text_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
         """Handle incoming text messages.
 

plugins/swarm_map_policy/__init__.py

@@ -83,6 +83,57 @@ def is_group_allowed(group_id: str, platform: str) -> bool:
         return False
 
 
+def approve_group_add(
+    group_id: str, added_by_user_id: str, platform: str = "telegram"
+) -> bool:
+    """Request HSM auto-approval for a group the bot was just added to.
+
+    Called when someone adds the bot to a new group. HSM verifies that the
+    adder is a platform admin and, if so, adds the group to the allowlist.
+    Fail-closed: returns True only on HTTP 200 with ``approved: true`` —
+    network errors, non-200 responses, and missing fields all deny.
+    """
+    url = _hsm_url()
+    harness = _harness_id()
+    if not url or not harness:
+        logger.warning("swarm-map-policy: HSM not configured, denying group add")
+        return False
+    try:
+        resp = requests.post(
+            f"{url}/api/harnesses/{harness}/surfaces/{platform}/groups/{group_id}",
+            json={"addedByUserId": added_by_user_id},
+            timeout=5,
+        )
+        if resp.status_code != 200:
+            reason = ""
+            try:
+                reason = resp.json().get("error", "")
+            except Exception:
+                pass
+            logger.info(
+                "swarm-map-policy: group add denied for %s (HTTP %s%s)",
+                group_id, resp.status_code, f": {reason}" if reason else "",
+            )
+            return False
+        data = resp.json()
+        if data.get("approved") is True:
+            logger.info(
+                "swarm-map-policy: group add approved for %s (already_allowed=%s restarted=%s)",
+                group_id, data.get("already_allowed", False), data.get("restarted"),
+            )
+            return True
+        logger.info(
+            "swarm-map-policy: group add not approved for %s: %s",
+            group_id, data.get("reason", "no reason given"),
+        )
+        return False
+    except Exception as e:
+        logger.warning(
+            "swarm-map-policy: group add approval failed (fail-closed): %s", e
+        )
+        return False
+
+
 def is_tool_allowed(tool_name: str, group_id: str) -> bool:
     """Check if a tool is allowed for a group. Fail-open."""
     url = _hsm_url()