streams: Make sure an anonymous group is used for only one setting.

When creating multiple streams using "POST /users/me/subscriptions"
endpoint, same anonymous group was being used for settings of all
the streams, like if can_subscribe_group was being set to an
anonymous group, all streams created using that endpoint would
use the same anonymous group for can_subscribe_group setting.

Using the same anonymous groups for multiple streams caused
unintended behavior  -

- Changing a setting for one stream also changed it for all
other streams sharing the the same anonymous group, because
eventually the members and subgroups of the group were updated.

- Changing the setting to a named user group failed because
`do_change_stream_permission_group` attempts to delete the
anonymous group during the update. However, deletion of
anonymous group is restricted if they are still referenced
by setting fields.

This commit fixes this so that one anonymous group is used
only for one setting for a single stream.

zerver/lib/streams.py

@@ -1488,48 +1488,55 @@ def filter_stream_authorization_for_adding_subscribers(
     )
 
 
-def access_requested_group_permissions(
+def access_requested_group_permissions_for_streams(
+    stream_names: list[str],
     user_profile: UserProfile,
     realm: Realm,
     request_settings_dict: dict[str, Any],
-) -> tuple[dict[str, UserGroup], dict[int, UserGroupMembersData]]:
+) -> tuple[dict[str, dict[str, UserGroup]], dict[int, UserGroupMembersData]]:
     anonymous_group_membership = {}
-    group_settings_map = {}
+    stream_group_settings_map = {}
     system_groups_name_dict = get_role_based_system_groups_dict(realm)
-    for setting_name, permission_configuration in Stream.stream_permission_group_settings.items():
-        assert setting_name in request_settings_dict
-        if request_settings_dict[setting_name] is not None:
-            setting_request_value = request_settings_dict[setting_name]
-            setting_value = parse_group_setting_value(
-                setting_request_value, system_groups_name_dict[SystemGroups.NOBODY]
-            )
-            group_settings_map[setting_name] = access_user_group_for_setting(
-                setting_value,
-                user_profile,
-                setting_name=setting_name,
-                permission_configuration=permission_configuration,
-            )
-            if (
-                setting_name in ["can_delete_any_message_group", "can_delete_own_message_group"]
-                and group_settings_map[setting_name].id
-                != system_groups_name_dict[SystemGroups.NOBODY].id
-                and not user_profile.can_set_delete_message_policy()
-            ):
-                raise JsonableError(_("Insufficient permission"))
-
-            if not isinstance(setting_value, int):
-                anonymous_group_membership[group_settings_map[setting_name].id] = setting_value
-        else:
-            group_settings_map[setting_name] = get_stream_permission_default_group(
-                setting_name, system_groups_name_dict, creator=user_profile
-            )
-            if permission_configuration.default_group_name == "channel_creator":
-                # Default for some settings like "can_administer_channel_group"
-                # is anonymous group with stream creator.
-                anonymous_group_membership[group_settings_map[setting_name].id] = (
-                    UserGroupMembersData(direct_subgroups=[], direct_members=[user_profile.id])
+    for stream_name in stream_names:
+        group_settings_map = {}
+        for (
+            setting_name,
+            permission_configuration,
+        ) in Stream.stream_permission_group_settings.items():
+            assert setting_name in request_settings_dict
+            if request_settings_dict[setting_name] is not None:
+                setting_request_value = request_settings_dict[setting_name]
+                setting_value = parse_group_setting_value(
+                    setting_request_value, system_groups_name_dict[SystemGroups.NOBODY]
                 )
-    return group_settings_map, anonymous_group_membership
+                group_settings_map[setting_name] = access_user_group_for_setting(
+                    setting_value,
+                    user_profile,
+                    setting_name=setting_name,
+                    permission_configuration=permission_configuration,
+                )
+                if (
+                    setting_name in ["can_delete_any_message_group", "can_delete_own_message_group"]
+                    and group_settings_map[setting_name].id
+                    != system_groups_name_dict[SystemGroups.NOBODY].id
+                    and not user_profile.can_set_delete_message_policy()
+                ):
+                    raise JsonableError(_("Insufficient permission"))
+
+                if not isinstance(setting_value, int):
+                    anonymous_group_membership[group_settings_map[setting_name].id] = setting_value
+            else:
+                group_settings_map[setting_name] = get_stream_permission_default_group(
+                    setting_name, system_groups_name_dict, creator=user_profile
+                )
+                if permission_configuration.default_group_name == "channel_creator":
+                    # Default for some settings like "can_administer_channel_group"
+                    # is anonymous group with stream creator.
+                    anonymous_group_membership[group_settings_map[setting_name].id] = (
+                        UserGroupMembersData(direct_subgroups=[], direct_members=[user_profile.id])
+                    )
+        stream_group_settings_map[stream_name] = group_settings_map
+    return stream_group_settings_map, anonymous_group_membership
 
 
 def list_to_streams(
@@ -1595,14 +1602,19 @@ def list_to_streams(
             )
 
         assert request_settings_dict is not None
-        group_settings_map, anonymous_group_membership = access_requested_group_permissions(
-            user_profile,
-            user_profile.realm,
-            request_settings_dict,
+        stream_names = [stream_dict["name"] for stream_dict in missing_stream_dicts]
+        stream_group_settings_map, anonymous_group_membership = (
+            access_requested_group_permissions_for_streams(
+                stream_names,
+                user_profile,
+                user_profile.realm,
+                request_settings_dict,
+            )
         )
 
         # autocreate=True path starts here
         for stream_dict in missing_stream_dicts:
+            group_settings_map = stream_group_settings_map[stream_dict["name"]]
             check_channel_creation_permissions(
                 user_profile,
                 is_default_stream=is_default_stream,

zerver/tests/test_channel_creation.py

@@ -1135,3 +1135,69 @@ class TestCreateStreams(ZulipTestCase):
         self.assertEqual(result[1][1].message_retention_days, -1)
         self.assertEqual(result[1][2].name, "new_stream3")
         self.assertEqual(result[1][2].message_retention_days, None)
+
+    def test_permission_settings_when_creating_multiple_streams(self) -> None:
+        """
+        Check that different anonymous group is used for each setting when creating
+        multiple streams in a single request.
+        """
+        realm = get_realm("zulip")
+        hamlet = self.example_user("hamlet")
+        cordelia = self.example_user("cordelia")
+
+        moderators_group = NamedUserGroup.objects.get(
+            name=SystemGroups.MODERATORS, realm_for_sharding=realm, is_system_group=True
+        )
+
+        subscriptions = [
+            {"name": "new_stream", "description": "New stream"},
+            {"name": "new_stream_2", "description": "New stream 2"},
+        ]
+        extra_post_data = {
+            "can_add_subscribers_group": orjson.dumps(
+                {
+                    "direct_members": [cordelia.id],
+                    "direct_subgroups": [moderators_group.id],
+                }
+            ).decode(),
+        }
+
+        result = self.subscribe_via_post(
+            hamlet,
+            subscriptions,
+            extra_post_data,
+        )
+        self.assert_json_success(result)
+
+        stream_1 = get_stream("new_stream", realm)
+        stream_2 = get_stream("new_stream_2", realm)
+
+        # Check value of can_administer_channel_group setting which is set to its default
+        # of an anonymous group with creator as the only member.
+        self.assertFalse(hasattr(stream_1.can_administer_channel_group, "named_user_group"))
+        self.assertFalse(hasattr(stream_2.can_administer_channel_group, "named_user_group"))
+        self.assertEqual(list(stream_1.can_administer_channel_group.direct_members.all()), [hamlet])
+        self.assertEqual(list(stream_2.can_administer_channel_group.direct_members.all()), [hamlet])
+        self.assertEqual(list(stream_1.can_administer_channel_group.direct_subgroups.all()), [])
+        self.assertEqual(list(stream_2.can_administer_channel_group.direct_subgroups.all()), [])
+
+        # Check value of can_add_subscribers_group setting which is set to an anonymous
+        # group as request.
+        self.assertFalse(hasattr(stream_1.can_add_subscribers_group, "named_user_group"))
+        self.assertFalse(hasattr(stream_2.can_add_subscribers_group, "named_user_group"))
+        self.assertEqual(list(stream_1.can_add_subscribers_group.direct_members.all()), [cordelia])
+        self.assertEqual(list(stream_2.can_add_subscribers_group.direct_members.all()), [cordelia])
+        self.assertEqual(
+            list(stream_1.can_add_subscribers_group.direct_subgroups.all()), [moderators_group]
+        )
+        self.assertEqual(
+            list(stream_2.can_add_subscribers_group.direct_subgroups.all()), [moderators_group]
+        )
+
+        # Check that for each stream, different anonymous group is used.
+        self.assertNotEqual(
+            stream_1.can_administer_channel_group_id, stream_2.can_administer_channel_group_id
+        )
+        self.assertNotEqual(
+            stream_1.can_add_subscribers_group_id, stream_2.can_add_subscribers_group_id
+        )

zerver/views/streams.py

@@ -69,7 +69,7 @@ from zerver.lib.stream_traffic import get_streams_traffic
 from zerver.lib.streams import (
     StreamDict,
     access_default_stream_group_by_id,
-    access_requested_group_permissions,
+    access_requested_group_permissions_for_streams,
     access_stream_by_id,
     access_stream_by_name,
     access_stream_for_delete_or_update_requiring_metadata_access,
@@ -711,12 +711,16 @@ def create_channel(
         message_retention_days=parsed_message_retention_days,
     )
 
-    group_settings_map, anonymous_group_membership = access_requested_group_permissions(
-        user_profile,
-        realm,
-        request_settings_dict,
+    stream_group_settings_map, anonymous_group_membership = (
+        access_requested_group_permissions_for_streams(
+            [name],
+            user_profile,
+            realm,
+            request_settings_dict,
+        )
     )
 
+    group_settings_map = stream_group_settings_map[name]
     new_channel, created = create_stream_if_needed(
         realm,
         name,