messages: Allow service bots to access private messages.

This commit allows service bots including embedded bots and outgoing
webhook bots to access the private messages they sent or received in
a group or a 1-to-1 chat.

Related: zulip#13658

zerver/lib/message.py

@@ -49,6 +49,7 @@
     UserMessage,
     UserProfile,
     get_display_recipient_by_id,
+    get_huddle_user_ids,
     get_usermessage_by_message_id,
     query_for_ids,
 )
@@ -660,6 +661,7 @@ def access_message(
     (1) You received or have previously accessed via starring
         (aka have a UserMessage row for).
     (2) Was sent to a public stream in your realm.
+    (3) You received the PM or group PM as a service bot.
 
     We produce consistent, boring error messages to avoid leaking any
     information from a security perspective.
@@ -680,6 +682,22 @@ def has_message_access(
     user_profile: UserProfile, message: Message, user_message: Optional[UserMessage]
 ) -> bool:
     if user_message is None:
+        if (
+            user_profile.bot_type == UserProfile.EMBEDDED_BOT
+            or user_profile.bot_type == UserProfile.OUTGOING_WEBHOOK_BOT
+        ):
+            #  For service bots that don't have UserMessage rows, we handle access to private message differently
+            if message.recipient.type == Recipient.PERSONAL:
+                # Bots can access a message that is directly sent to/by the bot
+                return (
+                    message.recipient.id == user_profile.recipient.id
+                    or message.sender.id == user_profile.id
+                )
+
+            if message.recipient.type == Recipient.HUDDLE:
+                # Bots can access a message that is sent in a private group
+                return user_profile.id in get_huddle_user_ids(message.recipient)
+
         if message.recipient.type != Recipient.STREAM:
             # You can't access private messages you didn't receive
             return False

zerver/tests/test_messages.py

@@ -1,10 +1,12 @@
 import datetime
+import responses
 from typing import Dict, List
 
 from django.utils.timezone import now as timezone_now
 
 from zerver.lib.actions import get_active_presence_idle_user_ids, get_client
 from zerver.lib.test_classes import ZulipTestCase
+from zerver.lib.message import access_message
 from zerver.models import (
     Message,
     UserPresence,
@@ -87,3 +89,76 @@ def test_bulk_get_huddle_user_ids(self) -> None:
 
     def test_bulk_get_huddle_user_ids_empty_list(self) -> None:
         self.assertEqual(bulk_get_huddle_user_ids([]), {})
+
+
+class TestServiceBotAccessMessage(ZulipTestCase):
+    def create_outgoing_webhook_bot(self, bot_owner: UserProfile) -> UserProfile:
+        return self.create_test_bot(
+            "outgoing-webhook",
+            bot_owner,
+            "Outgoing Webhook Bot",
+            bot_type=UserProfile.OUTGOING_WEBHOOK_BOT,
+            service_name="foo-service",
+            payload_url='"https://bot.example.com"',
+        )
+
+    def create_embedded_bot(self, bot_owner: UserProfile) -> UserProfile:
+        return self.create_test_bot(
+            "embedded-bot",
+            bot_owner,
+            "Embedded Bot",
+            bot_type=UserProfile.EMBEDDED_BOT,
+            service_name="helloworld",
+        )
+
+    @responses.activate
+    def test_outgoing_webhook_bot_access_private_message(self) -> None:
+        bot_owner = self.example_user("othello")
+        bot = self.create_outgoing_webhook_bot(bot_owner)
+
+        responses.add("POST", "https://bot.example.com", json={"content": "beep boop"})
+
+        self.send_personal_message(bot_owner, bot, content="foo")
+
+        last_message = self.get_last_message()
+        self.assertIsNotNone(access_message(bot, last_message.id))
+        second_to_last_message = self.get_second_to_last_message()
+        self.assertIsNotNone(access_message(bot, second_to_last_message.id))
+
+    @responses.activate
+    def test_outgoing_webhook_bot_access_huddle_message(self) -> None:
+        bot_owner = self.example_user("othello")
+        bot = self.create_outgoing_webhook_bot(bot_owner)
+        other_user = self.example_user("hamlet")
+
+        responses.add("POST", "https://bot.example.com", json={"content": "beep boop"})
+
+        self.send_huddle_message(bot_owner, [bot, other_user], "bar")
+
+        last_message = self.get_last_message()
+        self.assertIsNotNone(access_message(bot, last_message.id))
+        second_to_last_message = self.get_second_to_last_message()
+        self.assertIsNotNone(access_message(bot, second_to_last_message.id))
+
+    def test_embedded_bot_access_private_message(self) -> None:
+        bot_owner = self.example_user("othello")
+        bot = self.create_embedded_bot(bot_owner)
+
+        self.send_personal_message(bot_owner, bot, content="foo")
+
+        last_message = self.get_last_message()
+        self.assertIsNotNone(access_message(bot, last_message.id))
+        second_to_last_message = self.get_second_to_last_message()
+        self.assertIsNotNone(access_message(bot, second_to_last_message.id))
+
+    def test_embedded_bot_access_huddle_message(self) -> None:
+        bot_owner = self.example_user("othello")
+        bot = self.create_embedded_bot(bot_owner)
+        other_user = self.example_user("hamlet")
+
+        self.send_huddle_message(bot_owner, [bot, other_user], "bar")
+
+        last_message = self.get_last_message()
+        self.assertIsNotNone(access_message(bot, last_message.id))
+        second_to_last_message = self.get_second_to_last_message()
+        self.assertIsNotNone(access_message(bot, second_to_last_message.id))