message_send: Check group mention permission when sending message.

This commit adds backend code to check whether user has permission
to mention a group while sending message as per the can_mention_group
setting of the group.

Fixes a part of #25927.

zerver/actions/message_send.py

@@ -47,6 +47,7 @@ from zerver.lib.mention import MentionBackend, MentionData
 from zerver.lib.message import (
     MessageDict,
     SendMessageRequest,
+    check_user_group_mention_allowed,
     normalize_body,
     render_markdown,
     truncate_topic,
@@ -1486,6 +1487,11 @@ def check_message(
         raise JsonableError(
             _("You do not have permission to use wildcard mentions in this stream.")
         )
+
+    if message_send_dict.rendering_result.mentions_user_group_ids:
+        mentioned_group_ids = list(message_send_dict.rendering_result.mentions_user_group_ids)
+        check_user_group_mention_allowed(sender, mentioned_group_ids)
+
     return message_send_dict
 
 

zerver/lib/message.py

@@ -52,6 +52,7 @@ from zerver.lib.timestamp import datetime_to_timestamp
 from zerver.lib.topic import DB_TOPIC_NAME, MESSAGE__TOPIC, TOPIC_LINKS, TOPIC_NAME
 from zerver.lib.types import DisplayRecipientT, EditHistoryEvent, UserDisplayRecipient
 from zerver.lib.url_preview.types import UrlEmbedData
+from zerver.lib.user_groups import is_user_in_group
 from zerver.lib.user_topics import build_topic_mute_checker, topic_has_visibility_policy
 from zerver.models import (
     MAX_TOPIC_NAME_LENGTH,
@@ -62,6 +63,7 @@ from zerver.models import (
     Stream,
     SubMessage,
     Subscription,
+    UserGroup,
     UserMessage,
     UserProfile,
     UserTopic,
@@ -1633,6 +1635,21 @@ def wildcard_mention_allowed(sender: UserProfile, stream: Stream) -> bool:
     raise AssertionError("Invalid wildcard mention policy")
 
 
+def check_user_group_mention_allowed(sender: UserProfile, user_group_ids: List[int]) -> None:
+    user_groups = UserGroup.objects.filter(id__in=user_group_ids).select_related(
+        "can_mention_group"
+    )
+
+    for group in user_groups:
+        can_mention_group = group.can_mention_group
+        if not is_user_in_group(can_mention_group, sender, direct_member_only=False):
+            raise JsonableError(
+                _(
+                    "You are not allowed to mention user group '{user_group_name}'. You must be a member of '{can_mention_group_name}' to mention this group."
+                ).format(user_group_name=group.name, can_mention_group_name=can_mention_group.name)
+            )
+
+
 def parse_message_time_limit_setting(
     value: Union[int, str],
     special_values_map: Mapping[str, Optional[int]],

zerver/tests/test_message_send.py

@@ -28,6 +28,7 @@ from zerver.actions.message_send import (
 )
 from zerver.actions.realm_settings import do_set_realm_property
 from zerver.actions.streams import do_change_stream_post_policy
+from zerver.actions.user_groups import add_subgroups_to_user_group, check_add_user_group
 from zerver.actions.users import do_change_can_forge_sender, do_deactivate_user
 from zerver.lib.addressee import Addressee
 from zerver.lib.cache import cache_delete, get_stream_cache_key
@@ -52,6 +53,7 @@ from zerver.models import (
     Recipient,
     Stream,
     Subscription,
+    UserGroup,
     UserMessage,
     UserProfile,
     flush_per_request_caches,
@@ -1732,6 +1734,88 @@ class StreamMessagesTest(ZulipTestCase):
             with self.assertRaisesRegex(AssertionError, "Invalid wildcard mention policy"):
                 self.send_stream_message(cordelia, "test_stream", content)
 
+    def test_user_group_mention_restrictions(self) -> None:
+        iago = self.example_user("iago")
+        shiva = self.example_user("shiva")
+        cordelia = self.example_user("cordelia")
+        othello = self.example_user("othello")
+        self.subscribe(iago, "test_stream")
+        self.subscribe(shiva, "test_stream")
+        self.subscribe(othello, "test_stream")
+        self.subscribe(cordelia, "test_stream")
+
+        leadership = check_add_user_group(othello.realm, "leadership", [othello], acting_user=None)
+        support = check_add_user_group(othello.realm, "support", [othello], acting_user=None)
+
+        moderators_system_group = UserGroup.objects.get(
+            realm=iago.realm, name=UserGroup.MODERATORS_GROUP_NAME, is_system_group=True
+        )
+
+        content = "Test mentioning user group @*leadership*"
+        msg_id = self.send_stream_message(cordelia, "test_stream", content)
+        result = self.api_get(cordelia, "/api/v1/messages/" + str(msg_id))
+        self.assert_json_success(result)
+
+        leadership.can_mention_group = moderators_system_group
+        leadership.save()
+        with self.assertRaisesRegex(
+            JsonableError,
+            f"You are not allowed to mention user group '{leadership.name}'. You must be a member of '{moderators_system_group.name}' to mention this group.",
+        ):
+            self.send_stream_message(cordelia, "test_stream", content)
+
+        # The restriction does not apply on silent mention.
+        content = "Test mentioning user group @_*leadership*"
+        msg_id = self.send_stream_message(cordelia, "test_stream", content)
+        result = self.api_get(cordelia, "/api/v1/messages/" + str(msg_id))
+        self.assert_json_success(result)
+
+        content = "Test mentioning user group @*leadership*"
+        msg_id = self.send_stream_message(shiva, "test_stream", content)
+        result = self.api_get(shiva, "/api/v1/messages/" + str(msg_id))
+        self.assert_json_success(result)
+
+        msg_id = self.send_stream_message(iago, "test_stream", content)
+        result = self.api_get(iago, "/api/v1/messages/" + str(msg_id))
+        self.assert_json_success(result)
+
+        test = check_add_user_group(shiva.realm, "test", [shiva], acting_user=None)
+        add_subgroups_to_user_group(leadership, [test], acting_user=None)
+        support.can_mention_group = leadership
+        support.save()
+
+        content = "Test mentioning user group @*support*"
+        with self.assertRaisesRegex(
+            JsonableError,
+            f"You are not allowed to mention user group '{support.name}'. You must be a member of '{leadership.name}' to mention this group.",
+        ):
+            self.send_stream_message(iago, "test_stream", content)
+
+        msg_id = self.send_stream_message(othello, "test_stream", content)
+        result = self.api_get(othello, "/api/v1/messages/" + str(msg_id))
+        self.assert_json_success(result)
+
+        msg_id = self.send_stream_message(shiva, "test_stream", content)
+        result = self.api_get(shiva, "/api/v1/messages/" + str(msg_id))
+        self.assert_json_success(result)
+
+        content = "Test mentioning user group @*support* @*leadership*"
+        with self.assertRaisesRegex(
+            JsonableError,
+            f"You are not allowed to mention user group '{support.name}'. You must be a member of '{leadership.name}' to mention this group.",
+        ):
+            self.send_stream_message(iago, "test_stream", content)
+
+        with self.assertRaisesRegex(
+            JsonableError,
+            f"You are not allowed to mention user group '{leadership.name}'. You must be a member of '{moderators_system_group.name}' to mention this group.",
+        ):
+            self.send_stream_message(othello, "test_stream", content)
+
+        msg_id = self.send_stream_message(shiva, "test_stream", content)
+        result = self.api_get(shiva, "/api/v1/messages/" + str(msg_id))
+        self.assert_json_success(result)
+
     def test_stream_message_mirroring(self) -> None:
         user = self.mit_user("starnine")
         self.subscribe(user, "Verona")