email_mirror: Check that email gateway bot can send to the channel.

Otherwise this leads to an uncaught exception when the check happens
later in `check_message`.

zerver/lib/email_mirror.py

@@ -28,6 +28,7 @@ from zerver.lib.message import normalize_body, truncate_content, truncate_topic
 from zerver.lib.queue import queue_json_publish
 from zerver.lib.rate_limiter import RateLimitedObject
 from zerver.lib.send_email import FromAddress
+from zerver.lib.streams import access_stream_for_send_message
 from zerver.lib.string_validation import is_character_printable
 from zerver.lib.upload import upload_message_attachment
 from zerver.models import Message, MissedMessageEmailAddress, Realm, Recipient, Stream, UserProfile
@@ -431,6 +432,15 @@ def process_stream_message(to: str, message: EmailMessage) -> None:
         options["include_quotes"] = is_forwarded(subject_header)
 
     user_profile = get_system_bot(settings.EMAIL_GATEWAY_BOT, stream.realm_id)
+
+    try:
+        access_stream_for_send_message(user_profile, stream, forwarder_user_profile=None)
+    except JsonableError as e:
+        logger.info(
+            "Failed to process email to %s (%s): %s", stream.name, stream.realm.string_id, e
+        )
+        return
+
     body = construct_zulip_body(message, stream.realm, sender=user_profile, **options)
     send_zulip(user_profile, stream, subject, body)
     logger.info(

zerver/tests/test_email_mirror.py

@@ -13,7 +13,7 @@ import orjson
 from django.conf import settings
 
 from zerver.actions.realm_settings import do_deactivate_realm
-from zerver.actions.streams import do_change_stream_post_policy
+from zerver.actions.streams import do_change_stream_post_policy, do_deactivate_stream
 from zerver.actions.users import do_deactivate_user
 from zerver.lib.email_mirror import (
     create_missed_message_address,
@@ -39,6 +39,7 @@ from zerver.lib.streams import ensure_stream
 from zerver.lib.test_classes import ZulipTestCase
 from zerver.lib.test_helpers import mock_queue_publish, most_recent_message, most_recent_usermessage
 from zerver.models import Attachment, Recipient, Stream, UserProfile
+from zerver.models.messages import Message
 from zerver.models.realms import get_realm
 from zerver.models.streams import get_stream
 from zerver.models.users import get_system_bot
@@ -228,7 +229,7 @@ class TestFilterFooter(ZulipTestCase):
         self.assertEqual(result, text)
 
 
-class TestStreamEmailMessagesSuccess(ZulipTestCase):
+class TestStreamEmailMessages(ZulipTestCase):
     def create_incoming_valid_message(
         self, msgtext: str, stream: Stream, include_quotes: bool
     ) -> EmailMessage:
@@ -415,6 +416,31 @@ class TestStreamEmailMessagesSuccess(ZulipTestCase):
         self.assert_message_stream_name(message, stream.name)
         self.assertEqual(message.topic_name(), incoming_valid_message["Subject"])
 
+    def test_receive_stream_email_deactivated_stream(self) -> None:
+        user_profile = self.example_user("hamlet")
+        self.login_user(user_profile)
+        self.subscribe(user_profile, "Denmark")
+        stream = get_stream("Denmark", user_profile.realm)
+
+        msgtext = "TestStreamEmailMessages Body"
+        incoming_valid_message = self.create_incoming_valid_message(
+            msgtext, stream, include_quotes=False
+        )
+
+        do_deactivate_stream(stream, acting_user=None)
+        last_message_id = Message.objects.latest("id").id
+
+        with self.assertLogs(logger_name, level="INFO") as m:
+            process_message(incoming_valid_message)
+        self.assertEqual(
+            m.output,
+            [
+                f"INFO:{logger_name}:Failed to process email to {stream.name} ({stream.realm.string_id}): "
+                f"Not authorized to send to channel '{stream.name}'",
+            ],
+        )
+        self.assertEqual(Message.objects.latest("id").id, last_message_id)
+
     def test_receive_stream_email_show_sender_success(self) -> None:
         user_profile = self.example_user("hamlet")
         self.login_user(user_profile)