email_mirror: Create attachments as the message sender.

When the email mirror gateway is sending messages "as" a user (as
triggered by having access to the missed-message email address),
attachments were still created as the Email Gateway bot.  Since the
sender (the end-user) was not the owner of those attachments (the
gateway bot), nor were they referenced yet anywhere, this resulted in
the attachments being "orphaned" and not allowed to be accessed by
anyone -- despite the attachment links being embedded in the message.
This was accompanied by the error:

```
WARN [] User 12345 tried to share upload 123/3LkSA4OcoG6OpAknS2I0SFAQ/example.jpf in message 123456, but lacks permission
INFO [zerver.lib.email_mirror] Successfully processed email from user 12345 to example-stream
```

We solve this by creating attachment objects as the users the message
will be sent from.

zerver/lib/email_mirror.py

@@ -155,6 +155,8 @@ def create_missed_message_address(user_profile: UserProfile, message: Message) -
 def construct_zulip_body(
     message: EmailMessage,
     realm: Realm,
+    *,
+    sender: UserProfile,
     show_sender: bool = False,
     include_quotes: bool = False,
     include_footer: bool = False,
@@ -168,13 +170,13 @@ def construct_zulip_body(
 
     if not body.endswith("\n"):
         body += "\n"
-    body += extract_and_upload_attachments(message, realm)
+    body += extract_and_upload_attachments(message, realm, sender)
     if not body.rstrip():
         body = "(No email body)"
 
     if show_sender:
-        sender = str(message.get("From", ""))
-        body = f"From: {sender}\n{body}"
+        from_address = str(message.get("From", ""))
+        body = f"From: {from_address}\n{body}"
 
     return body
 
@@ -314,9 +316,7 @@ def filter_footer(text: str) -> str:
     return re.split(r"^\s*--\s*$", text, 1, flags=re.MULTILINE)[0].strip()
 
 
-def extract_and_upload_attachments(message: EmailMessage, realm: Realm) -> str:
-    user_profile = get_system_bot(settings.EMAIL_GATEWAY_BOT, realm.id)
-
+def extract_and_upload_attachments(message: EmailMessage, realm: Realm, sender: UserProfile) -> str:
     attachment_links = []
     for part in message.walk():
         content_type = part.get_content_type()
@@ -329,7 +329,7 @@ def extract_and_upload_attachments(message: EmailMessage, realm: Realm) -> str:
                     len(attachment),
                     content_type,
                     attachment,
-                    user_profile,
+                    sender,
                     target_realm=realm,
                 )
                 formatted_link = f"[{filename}]({s3_url})"
@@ -414,8 +414,9 @@ def process_stream_message(to: str, message: EmailMessage) -> None:
     if "include_quotes" not in options:
         options["include_quotes"] = is_forwarded(subject_header)
 
-    body = construct_zulip_body(message, stream.realm, **options)
-    send_zulip(get_system_bot(settings.EMAIL_GATEWAY_BOT, stream.realm_id), stream, subject, body)
+    user_profile = get_system_bot(settings.EMAIL_GATEWAY_BOT, stream.realm_id)
+    body = construct_zulip_body(message, stream.realm, sender=user_profile, **options)
+    send_zulip(user_profile, stream, subject, body)
     logger.info(
         "Successfully processed email to %s (%s)",
         stream.name,
@@ -440,7 +441,7 @@ def process_missed_message(to: str, message: EmailMessage) -> None:
         logger.warning("Sending user is not active. Ignoring this message notification email.")
         return
 
-    body = construct_zulip_body(message, user_profile.realm)
+    body = construct_zulip_body(message, user_profile.realm, sender=user_profile)
 
     assert recipient is not None
     if recipient.type == Recipient.STREAM:

zerver/tests/test_email_mirror.py

@@ -834,6 +834,43 @@ class TestEmailMirrorMessagesWithAttachments(ZulipTestCase):
         # HTML body is empty, so the plaintext content should be picked, despite prefer-html option.
         self.assertEqual(message.content, "Test message")
 
+    def test_message_with_valid_attachment_missed_message(self) -> None:
+        user_profile = self.example_user("othello")
+        usermessage = most_recent_usermessage(user_profile)
+        mm_address = create_missed_message_address(user_profile, usermessage.message)
+
+        incoming_valid_message = EmailMessage()
+        incoming_valid_message.set_content("Test body")
+        with open(
+            os.path.join(settings.DEPLOY_ROOT, "static/images/default-avatar.png"), "rb"
+        ) as f:
+            image_bytes = f.read()
+
+        incoming_valid_message.add_attachment(
+            image_bytes,
+            maintype="image",
+            subtype="png",
+            filename="image.png",
+        )
+
+        incoming_valid_message["Subject"] = "TestStreamEmailMessages subject"
+        incoming_valid_message["From"] = self.example_email("othello")
+        incoming_valid_message["To"] = mm_address
+        incoming_valid_message["Reply-to"] = self.example_email("othello")
+
+        process_message(incoming_valid_message)
+
+        message = most_recent_message(user_profile)
+        self.assertEqual(message.sender, user_profile)
+        self.assertTrue(message.has_attachment)
+
+        attachment = Attachment.objects.last()
+        assert attachment is not None
+        self.assertEqual(attachment.realm, user_profile.realm)
+        self.assertEqual(attachment.owner, user_profile)
+        self.assertEqual(attachment.is_realm_public, True)
+        self.assertEqual(list(attachment.messages.values_list("id", flat=True)), [message.id])
+
 
 class TestStreamEmailMessagesEmptyBody(ZulipTestCase):
     def test_receive_stream_email_messages_empty_body(self) -> None: