upload: Explicitly return a bool and the Attachment object.

zerver/actions/uploads.py

@@ -48,7 +48,7 @@ def do_claim_attachments(
             is_message_realm_public = stream.is_public()
             is_message_web_public = stream.is_web_public
 
-        if not validate_attachment_request(user_profile, path_id):
+        if not validate_attachment_request(user_profile, path_id)[0]:
             # Technically, there are 2 cases here:
             # * The user put something in their message that has the form
             # of an upload URL, but does not actually correspond to a previously

zerver/lib/attachments.py

@@ -90,15 +90,15 @@ def validate_attachment_request(
     maybe_user_profile: UserProfile | AnonymousUser,
     path_id: str,
     realm: Realm | None = None,
-) -> bool | None:
+) -> tuple[bool, Attachment | None]:
     try:
         attachment = Attachment.objects.get(path_id=path_id)
     except Attachment.DoesNotExist:
-        return None
+        return False, None
 
     if isinstance(maybe_user_profile, AnonymousUser):
         assert realm is not None
-        return validate_attachment_request_for_spectator_access(realm, attachment)
+        return validate_attachment_request_for_spectator_access(realm, attachment), attachment
 
     user_profile = maybe_user_profile
     assert isinstance(user_profile, UserProfile)
@@ -122,20 +122,20 @@ def validate_attachment_request(
 
     if user_profile == attachment.owner:
         # If you own the file, you can access it.
-        return True
+        return True, attachment
     if (
         attachment.is_realm_public
         and attachment.realm == user_profile.realm
         and user_profile.can_access_public_streams()
     ):
         # Any user in the realm can access realm-public files
-        return True
+        return True, attachment
 
     messages = attachment.messages.all()
     if UserMessage.objects.filter(user_profile=user_profile, message__in=messages).exists():
         # If it was sent in a direct message or private stream
         # message, then anyone who received that message can access it.
-        return True
+        return True, attachment
 
     # The user didn't receive any of the messages that included this
     # attachment.  But they might still have access to it, if it was
@@ -150,11 +150,11 @@ def validate_attachment_request(
         recipient__in=[m.recipient_id for m in messages],
     ).values_list("recipient__type_id", flat=True)
     if len(relevant_stream_ids) == 0:
-        return False
+        return False, attachment
 
     return Stream.objects.filter(
         id__in=relevant_stream_ids, history_public_to_subscribers=True
-    ).exists()
+    ).exists(), attachment
 
 
 def get_old_unclaimed_attachments(

zerver/tests/test_subs.py

@@ -1233,8 +1233,8 @@ class StreamAdminTest(ZulipTestCase):
         self.assertIsNone(attachment.is_realm_public)
 
         cordelia = self.example_user("cordelia")
-        self.assertFalse(validate_attachment_request(cordelia, attachment.path_id))
-        self.assertTrue(validate_attachment_request(owner, attachment.path_id))
+        self.assertFalse(validate_attachment_request(cordelia, attachment.path_id)[0])
+        self.assertTrue(validate_attachment_request(owner, attachment.path_id)[0])
         attachment.refresh_from_db()
         self.assertFalse(attachment.is_realm_public)
         self.assertFalse(validate_attachment_request_for_spectator_access(realm, attachment))
@@ -1259,7 +1259,7 @@ class StreamAdminTest(ZulipTestCase):
         self.assertTrue(attachment.is_web_public)
         self.assertIsNone(attachment.is_realm_public)
 
-        self.assertTrue(validate_attachment_request(cordelia, attachment.path_id))
+        self.assertTrue(validate_attachment_request(cordelia, attachment.path_id)[0])
         attachment.refresh_from_db()
         self.assertTrue(attachment.is_realm_public)
 
@@ -1312,8 +1312,8 @@ class StreamAdminTest(ZulipTestCase):
         self.assertFalse(attachment.is_web_public)
         self.assertIsNone(attachment.is_realm_public)
 
-        self.assertFalse(validate_attachment_request(cordelia, attachment.path_id))
-        self.assertTrue(validate_attachment_request(owner, attachment.path_id))
+        self.assertFalse(validate_attachment_request(cordelia, attachment.path_id)[0])
+        self.assertTrue(validate_attachment_request(owner, attachment.path_id)[0])
         attachment.refresh_from_db()
         self.assertFalse(attachment.is_realm_public)
 

zerver/tests/test_upload.py

@@ -875,7 +875,7 @@ class FileUploadTest(UploadSerializeMixin, ZulipTestCase):
             self.assertEqual(response.getvalue(), b"zulip!")
 
         with self.assert_database_query_count(6):
-            self.assertTrue(validate_attachment_request(user, fp_path_id))
+            self.assertTrue(validate_attachment_request(user, fp_path_id)[0])
 
         self.logout()
 

zerver/views/upload.py

@@ -250,7 +250,7 @@ def serve_file(
 ) -> HttpResponseBase:
     path_id = f"{realm_id_str}/{filename}"
     realm = get_valid_realm_from_request(request)
-    is_authorized = validate_attachment_request(maybe_user_profile, path_id, realm)
+    is_authorized, attachment = validate_attachment_request(maybe_user_profile, path_id, realm)
 
     def serve_image_error(status: int, image_path: str) -> HttpResponseBase:
         # We cannot use X-Accel-Redirect to offload the serving of
@@ -258,7 +258,7 @@ def serve_file(
         # code of this response, nor the Vary: header.
         return FileResponse(open(static_path(image_path), "rb"), status=status)  # noqa: SIM115
 
-    if is_authorized is None:
+    if attachment is None:
         if preferred_accept(request, ["text/html", "image/png"]) == "image/png":
             response = serve_image_error(404, "images/errors/image-not-exist.png")
         else: