Merge 94838b3296 into 4fade39c90

gateway/platforms/email.py

@@ -410,6 +410,17 @@ class EmailAdapter(BasePlatformAdapter):
         if sender_addr == self._address.lower():
             return
 
+        # Drop senders explicitly blocklisted via EMAIL_IGNORED_SENDERS.
+        # Defense in depth against intra-fleet email loops where two
+        # gateways configured with each other as unauthorized senders
+        # ping-pong busy-ack notifications forever.
+        ignored_raw = os.getenv('EMAIL_IGNORED_SENDERS', '').strip()
+        if ignored_raw and sender_addr:
+            ignored_set = {a.strip().lower() for a in ignored_raw.split(',') if a.strip()}
+            if sender_addr.lower() in ignored_set:
+                logger.info('[Email] Ignoring blocklisted sender: %s', sender_addr)
+                return
+
         # Never reply to automated senders
         if _is_automated_sender(sender_addr, {}):
             logger.debug("[Email] Dropping automated sender at dispatch: %s", sender_addr)

gateway/run.py

@@ -1553,6 +1553,21 @@ class GatewayRunner:
         merge_pending_message_event(adapter._pending_messages, session_key, event)
 
     async def _handle_active_session_busy_message(self, event: MessageEvent, session_key: str) -> bool:
+        # Auth check FIRST: unauthorized senders must never trigger an
+        # outbound reply, including the busy-ack. Without this, two
+        # gateways configured with each other's address as unauthorized
+        # can ping-pong busy-acks forever (Toryx 2026-04-19 incident:
+        # 487 emails between two MD profiles in 20 min). Internal events
+        # bypass auth as elsewhere in the codebase.
+        if not getattr(event, 'internal', False) and event.source.user_id is not None:
+            if not self._is_user_authorized(event.source):
+                logger.info(
+                    'Suppressed busy-ack to unauthorized sender %s on %s',
+                    event.source.user_id,
+                    event.source.platform.value if event.source.platform else 'unknown',
+                )
+                return True  # consume event silently
+
         # --- Draining case (gateway restarting/stopping) ---
         if self._draining:
             adapter = self.adapters.get(event.source.platform)