realm: Add new group setting for who can create public streams.

api_docs/changelog.md

@@ -20,6 +20,13 @@ format used by the Zulip server that they are interacting with.
 
 ## Changes in Zulip 9.0
 
+**Feature level 264**
+
+* `PATCH /realm`, [`POST /register`](/api/register-queue),
+  [`GET /events`](/api/get-events): Added `can_create_public_channel_group`
+  realm setting, which is a [group-setting value](/api/group-setting-values)
+  describing the set of users with permission to create channels.
+
 **Feature level 263**:
 
 * [`POST /users/me/presence`](/api/update-presence):

zerver/actions/realm_settings.py

@@ -20,6 +20,11 @@ from zerver.lib.sessions import delete_realm_user_sessions
 from zerver.lib.timestamp import datetime_to_timestamp, timestamp_to_datetime
 from zerver.lib.upload import delete_message_attachments
 from zerver.lib.user_counts import realm_user_count_by_role
+from zerver.lib.user_groups import (
+    AnonymousSettingGroupDict,
+    get_group_setting_value_for_api,
+    get_group_setting_value_for_audit_log_data,
+)
 from zerver.models import (
     ArchivedAttachment,
     Attachment,
@@ -151,22 +156,43 @@ def do_set_push_notifications_enabled_end_timestamp(
 
 @transaction.atomic(savepoint=False)
 def do_change_realm_permission_group_setting(
-    realm: Realm, setting_name: str, user_group: UserGroup, *, acting_user: Optional[UserProfile]
+    realm: Realm,
+    setting_name: str,
+    user_group: UserGroup,
+    old_setting_api_value: Optional[Union[int, AnonymousSettingGroupDict]] = None,
+    *,
+    acting_user: Optional[UserProfile],
 ) -> None:
     """Takes in a realm object, the name of an attribute to update, the
     user_group to update and and the user who initiated the update.
     """
     assert setting_name in Realm.REALM_PERMISSION_GROUP_SETTINGS
-    old_user_group_id = getattr(realm, setting_name).id
+    old_value = getattr(realm, setting_name)
 
     setattr(realm, setting_name, user_group)
     realm.save(update_fields=[setting_name])
 
+    if old_setting_api_value is None:
+        # Most production callers will have computed this as part of
+        # verifying whether there's an actual change to make, but it
+        # feels quite clumsy to have to pass it from unit tests, so we
+        # compute it here if not provided by the caller.
+        old_setting_api_value = get_group_setting_value_for_api(old_value)
+    new_setting_api_value = get_group_setting_value_for_api(user_group)
+
+    if not hasattr(old_value, "named_user_group") and hasattr(user_group, "named_user_group"):
+        # We delete the UserGroup which the setting was set to
+        # previously if it does not have any linked NamedUserGroup
+        # object, as it is not used anywhere else. A new UserGroup
+        # object would be created if the setting is later set to
+        # a combination of users and groups.
+        old_value.delete()
+
     event = dict(
         type="realm",
         op="update_dict",
         property="default",
-        data={setting_name: user_group.id},
+        data={setting_name: new_setting_api_value},
     )
 
     send_event_on_commit(realm, event, active_user_ids(realm.id))
@@ -178,8 +204,12 @@ def do_change_realm_permission_group_setting(
         event_time=event_time,
         acting_user=acting_user,
         extra_data={
-            RealmAuditLog.OLD_VALUE: old_user_group_id,
+            RealmAuditLog.OLD_VALUE: get_group_setting_value_for_audit_log_data(
-            RealmAuditLog.NEW_VALUE: user_group.id,
+                old_setting_api_value
+            ),
+            RealmAuditLog.NEW_VALUE: get_group_setting_value_for_audit_log_data(
+                new_setting_api_value
+            ),
             "property": setting_name,
         },
     )

zerver/lib/event_schema.py

@@ -1030,9 +1030,25 @@ night_logo_data = DictType(
     ]
 )
 
+group_setting_type = UnionType(
+    [
+        int,
+        DictType(
+            required_keys=[
+                ("direct_members", ListType(int)),
+                ("direct_subgroups", ListType(int)),
+            ]
+        ),
+    ]
+)
+
 group_setting_update_data_type = DictType(
     required_keys=[],
-    optional_keys=[("create_multiuse_invite_group", int), ("can_access_all_users_group", int)],
+    optional_keys=[
+        ("create_multiuse_invite_group", int),
+        ("can_access_all_users_group", int),
+        ("can_create_public_channel_group", group_setting_type),
+    ],
 )
 
 update_dict_data = UnionType(
@@ -1787,19 +1803,6 @@ update_message_flags_remove_event = event_dict_type(
 )
 check_update_message_flags_remove = make_checker(update_message_flags_remove_event)
 
-
-group_setting_type = UnionType(
-    [
-        int,
-        DictType(
-            required_keys=[
-                ("direct_members", ListType(int)),
-                ("direct_subgroups", ListType(int)),
-            ]
-        ),
-    ]
-)
-
 group_type = DictType(
     required_keys=[
         ("id", int),

zerver/lib/events.py

@@ -57,6 +57,7 @@ from zerver.lib.timestamp import datetime_to_timestamp
 from zerver.lib.timezone import canonicalize_timezone
 from zerver.lib.topic import TOPIC_NAME
 from zerver.lib.user_groups import (
+    get_group_setting_value_for_api,
     get_server_supported_permission_settings,
     user_groups_in_realm_serialized,
 )
@@ -279,6 +280,11 @@ def fetch_initial_state_data(
             setting_name,
             permission_configuration,
         ) in Realm.REALM_PERMISSION_GROUP_SETTINGS.items():
+            if setting_name in Realm.REALM_PERMISSION_GROUP_SETTINGS_WITH_NEW_API_FORMAT:
+                setting_value = getattr(realm, setting_name)
+                state["realm_" + setting_name] = get_group_setting_value_for_api(setting_value)
+                continue
+
             state["realm_" + setting_name] = getattr(realm, permission_configuration.id_field_name)
 
         # Most state is handled via the property_types framework;

zerver/migrations/0532_realm_can_create_public_channel_group.py

@@ -0,0 +1,23 @@
+# Generated by Django 5.0.6 on 2024-05-23 14:02
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("zerver", "0531_convert_most_ids_to_bigints"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="realm",
+            name="can_create_public_channel_group",
+            field=models.ForeignKey(
+                null=True,
+                on_delete=django.db.models.deletion.RESTRICT,
+                related_name="+",
+                to="zerver.usergroup",
+            ),
+        ),
+    ]

zerver/migrations/0533_set_can_create_public_channel_group.py

@@ -0,0 +1,61 @@
+# Generated by Django 5.0.6 on 2024-05-23 14:02
+
+from django.db import migrations
+from django.db.backends.base.schema import BaseDatabaseSchemaEditor
+from django.db.migrations.state import StateApps
+from django.db.models import OuterRef
+
+
+def set_can_create_public_channel_group_for_existing_realms(
+    apps: StateApps, schema_editor: BaseDatabaseSchemaEditor
+) -> None:
+    Realm = apps.get_model("zerver", "Realm")
+    NamedUserGroup = apps.get_model("zerver", "NamedUserGroup")
+
+    MEMBERS_ONLY = 1
+    ADMINS_ONLY = 2
+    FULL_MEMBERS_ONLY = 3
+    MODERATORS_ONLY = 4
+
+    Realm.objects.filter(
+        can_create_public_channel_group=None, create_public_stream_policy=MEMBERS_ONLY
+    ).update(
+        can_create_public_channel_group=NamedUserGroup.objects.filter(
+            name="role:members", realm=OuterRef("id"), is_system_group=True
+        ).values("pk")
+    )
+    Realm.objects.filter(
+        can_create_public_channel_group=None, create_public_stream_policy=ADMINS_ONLY
+    ).update(
+        can_create_public_channel_group=NamedUserGroup.objects.filter(
+            name="role:administrators", realm=OuterRef("id"), is_system_group=True
+        ).values("pk")
+    )
+    Realm.objects.filter(
+        can_create_public_channel_group=None, create_public_stream_policy=FULL_MEMBERS_ONLY
+    ).update(
+        can_create_public_channel_group=NamedUserGroup.objects.filter(
+            name="role:fullmembers", realm=OuterRef("id"), is_system_group=True
+        ).values("pk")
+    )
+    Realm.objects.filter(
+        can_create_public_channel_group=None, create_public_stream_policy=MODERATORS_ONLY
+    ).update(
+        can_create_public_channel_group=NamedUserGroup.objects.filter(
+            name="role:moderators", realm=OuterRef("id"), is_system_group=True
+        ).values("pk")
+    )
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("zerver", "0532_realm_can_create_public_channel_group"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            set_can_create_public_channel_group_for_existing_realms,
+            elidable=True,
+            reverse_code=migrations.RunPython.noop,
+        ),
+    ]

zerver/migrations/0534_alter_realm_can_create_public_channel_group.py

@@ -0,0 +1,22 @@
+# Generated by Django 5.0.6 on 2024-05-23 14:10
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("zerver", "0533_set_can_create_public_channel_group"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="realm",
+            name="can_create_public_channel_group",
+            field=models.ForeignKey(
+                on_delete=django.db.models.deletion.RESTRICT,
+                related_name="+",
+                to="zerver.usergroup",
+            ),
+        ),
+    ]

zerver/models/realms.py

@@ -303,6 +303,10 @@ class Realm(models.Model):  # type: ignore[django-manager-missing] # django-stub
         default=CreateWebPublicStreamPolicyEnum.OWNERS_ONLY
     )
 
+    can_create_public_channel_group = models.ForeignKey(
+        "UserGroup", on_delete=models.RESTRICT, related_name="+"
+    )
+
     # Who in the organization is allowed to delete messages they themselves sent.
     delete_own_message_policy = models.PositiveSmallIntegerField(
         default=CommonMessagePolicyEnum.EVERYONE
@@ -697,8 +701,21 @@ class Realm(models.Model):  # type: ignore[django-manager-missing] # django-stub
             id_field_name="can_access_all_users_group_id",
             allowed_system_groups=[SystemGroups.EVERYONE, SystemGroups.MEMBERS],
         ),
+        can_create_public_channel_group=GroupPermissionSetting(
+            require_system_group=False,
+            allow_internet_group=False,
+            allow_owners_group=False,
+            allow_nobody_group=False,
+            allow_everyone_group=False,
+            default_group_name=SystemGroups.MEMBERS,
+            id_field_name="can_create_public_channel_group_id",
+        ),
     )
 
+    REALM_PERMISSION_GROUP_SETTINGS_WITH_NEW_API_FORMAT = [
+        "can_create_public_channel_group",
+    ]
+
     DIGEST_WEEKDAY_VALUES = [0, 1, 2, 3, 4, 5, 6]
 
     # Icon is the square mobile icon.
@@ -1041,7 +1058,10 @@ post_delete.connect(realm_pre_and_post_delete_handler, sender=Realm)
 
 
 def get_realm(string_id: str) -> Realm:
-    return Realm.objects.get(string_id=string_id)
+    return Realm.objects.select_related(
+        "can_create_public_channel_group",
+        "can_create_public_channel_group__named_user_group",
+    ).get(string_id=string_id)
 
 
 def get_realm_by_id(realm_id: int) -> Realm:

zerver/models/users.py

@@ -882,6 +882,8 @@ def get_user_profile_by_id(user_profile_id: int) -> UserProfile:
         "realm",
         "realm__can_access_all_users_group",
         "realm__can_access_all_users_group__named_user_group",
+        "realm__can_create_public_channel_group",
+        "realm__can_create_public_channel_group__named_user_group",
         "bot_owner",
     ).get(id=user_profile_id)
 
@@ -903,6 +905,8 @@ def maybe_get_user_profile_by_api_key(api_key: str) -> Optional[UserProfile]:
             "realm",
             "realm__can_access_all_users_group",
             "realm__can_access_all_users_group__named_user_group",
+            "realm__can_create_public_channel_group",
+            "realm__can_create_public_channel_group__named_user_group",
             "bot_owner",
         ).get(api_key=api_key)
     except UserProfile.DoesNotExist:
@@ -932,6 +936,8 @@ def get_user_by_delivery_email(email: str, realm: "Realm") -> UserProfile:
         "realm",
         "realm__can_access_all_users_group",
         "realm__can_access_all_users_group__named_user_group",
+        "realm__can_create_public_channel_group",
+        "realm__can_create_public_channel_group__named_user_group",
         "bot_owner",
     ).get(delivery_email__iexact=email.strip(), realm=realm)
 

zerver/openapi/zulip.yaml

@@ -4226,6 +4226,20 @@ paths:
                                         `"role:internet"` and `"role:owners"`.
 
                                         **Changes**: New in Zulip 8.0 (feature level 209).
+                                    can_create_public_channel_group:
+                                      allOf:
+                                        - $ref: "#/components/schemas/GroupSettingValue"
+                                        - description: |
+                                            A [group-setting value](/api/group-setting-values) defining
+                                            the set of users who have permission to create public
+                                            channels in this organization.
+
+                                            This setting cannot be set to `"role:internet"`, `"role:everyone"`,
+                                            `"role:owners"` and `"role:nobody"` system groups.
+
+                                            **Changes**: New in Zulip 9.0 (feature level 264). Previously
+                                            `realm_create_public_stream_policy` field used to control the
+                                            permission to create public channels.
                                     create_public_stream_policy:
                                       type: integer
                                       description: |
@@ -15222,6 +15236,20 @@ paths:
 
                           The [policy](/api/roles-and-permissions#permission-levels)
                           for which users can create bot users in this organization.
+                      realm_can_create_public_channel_group:
+                        allOf:
+                          - $ref: "#/components/schemas/GroupSettingValue"
+                          - description: |
+                              A [group-setting value](/api/group-setting-values) defining
+                              the set of users who have permission to create public
+                              channels in this organization.
+
+                              This setting cannot be set to `"role:internet"`, `"role:everyone"`,
+                              `"role:owners"` and `"role:nobody"` system groups.
+
+                              **Changes**: New in Zulip 9.0 (feature level 264). Previously
+                              `realm_create_public_stream_policy` field used to control the
+                              permission to create public channels.
                       realm_create_public_stream_policy:
                         type: integer
                         description: |

zerver/tests/test_events.py

@@ -216,7 +216,11 @@ from zerver.lib.test_helpers import (
 from zerver.lib.timestamp import convert_to_UTC, datetime_to_timestamp
 from zerver.lib.topic import TOPIC_NAME
 from zerver.lib.types import ProfileDataElementUpdateDict
-from zerver.lib.user_groups import AnonymousSettingGroupDict
+from zerver.lib.user_groups import (
+    AnonymousSettingGroupDict,
+    get_group_setting_value_for_api,
+    get_role_based_system_groups_dict,
+)
 from zerver.models import (
     Attachment,
     CustomProfileField,
@@ -3573,6 +3577,151 @@ class RealmPropertyActionTest(BaseAction):
 
             old_group_id = new_group_id
 
+    def do_set_realm_permission_group_setting_to_anonymous_groups_test(
+        self, setting_name: str
+    ) -> None:
+        realm = self.user_profile.realm
+        system_user_groups_dict = get_role_based_system_groups_dict(
+            realm,
+        )
+
+        setting_permission_configuration = Realm.REALM_PERMISSION_GROUP_SETTINGS[setting_name]
+
+        default_group_name = setting_permission_configuration.default_group_name
+        default_group = system_user_groups_dict[default_group_name]
+
+        now = timezone_now()
+
+        do_change_realm_permission_group_setting(
+            realm,
+            setting_name,
+            default_group,
+            acting_user=self.user_profile,
+        )
+
+        self.assertEqual(
+            RealmAuditLog.objects.filter(
+                realm=realm,
+                event_type=RealmAuditLog.REALM_PROPERTY_CHANGED,
+                event_time__gte=now,
+                acting_user=self.user_profile,
+            ).count(),
+            1,
+        )
+
+        othello = self.example_user("othello")
+        admins_group = system_user_groups_dict[SystemGroups.ADMINISTRATORS]
+
+        setting_group = self.create_or_update_anonymous_group_for_setting([othello], [admins_group])
+        now = timezone_now()
+
+        with self.verify_action(state_change_expected=True, num_events=1) as events:
+            do_change_realm_permission_group_setting(
+                realm,
+                setting_name,
+                setting_group,
+                acting_user=self.user_profile,
+            )
+
+        self.assertEqual(
+            RealmAuditLog.objects.filter(
+                realm=realm,
+                event_type=RealmAuditLog.REALM_PROPERTY_CHANGED,
+                event_time__gte=now,
+                acting_user=self.user_profile,
+                extra_data={
+                    RealmAuditLog.OLD_VALUE: default_group.id,
+                    RealmAuditLog.NEW_VALUE: {
+                        "direct_members": [othello.id],
+                        "direct_subgroups": [admins_group.id],
+                    },
+                    "property": setting_name,
+                },
+            ).count(),
+            1,
+        )
+        check_realm_update_dict("events[0]", events[0])
+        self.assertEqual(
+            events[0]["data"][setting_name],
+            AnonymousSettingGroupDict(
+                direct_members=[othello.id], direct_subgroups=[admins_group.id]
+            ),
+        )
+
+        old_setting_api_value = get_group_setting_value_for_api(setting_group)
+        moderators_group = system_user_groups_dict[SystemGroups.MODERATORS]
+        setting_group = self.create_or_update_anonymous_group_for_setting(
+            [self.user_profile], [moderators_group], existing_setting_group=setting_group
+        )
+
+        # state_change_expected is False here because the initial state will
+        # also have the new setting value due to the setting group already
+        # being modified with the new members.
+        with self.verify_action(state_change_expected=False, num_events=1) as events:
+            do_change_realm_permission_group_setting(
+                realm,
+                setting_name,
+                setting_group,
+                old_setting_api_value=old_setting_api_value,
+                acting_user=self.user_profile,
+            )
+
+        self.assertEqual(
+            RealmAuditLog.objects.filter(
+                realm=realm,
+                event_type=RealmAuditLog.REALM_PROPERTY_CHANGED,
+                event_time__gte=now,
+                acting_user=self.user_profile,
+                extra_data={
+                    RealmAuditLog.OLD_VALUE: {
+                        "direct_members": [othello.id],
+                        "direct_subgroups": [admins_group.id],
+                    },
+                    RealmAuditLog.NEW_VALUE: {
+                        "direct_members": [self.user_profile.id],
+                        "direct_subgroups": [moderators_group.id],
+                    },
+                    "property": setting_name,
+                },
+            ).count(),
+            1,
+        )
+        check_realm_update_dict("events[0]", events[0])
+        self.assertEqual(
+            events[0]["data"][setting_name],
+            AnonymousSettingGroupDict(
+                direct_members=[self.user_profile.id], direct_subgroups=[moderators_group.id]
+            ),
+        )
+
+        with self.verify_action(state_change_expected=True, num_events=1) as events:
+            do_change_realm_permission_group_setting(
+                realm,
+                setting_name,
+                default_group,
+                acting_user=self.user_profile,
+            )
+
+        self.assertEqual(
+            RealmAuditLog.objects.filter(
+                realm=realm,
+                event_type=RealmAuditLog.REALM_PROPERTY_CHANGED,
+                event_time__gte=now,
+                acting_user=self.user_profile,
+                extra_data={
+                    RealmAuditLog.OLD_VALUE: {
+                        "direct_members": [self.user_profile.id],
+                        "direct_subgroups": [moderators_group.id],
+                    },
+                    RealmAuditLog.NEW_VALUE: default_group.id,
+                    "property": setting_name,
+                },
+            ).count(),
+            1,
+        )
+        check_realm_update_dict("events[0]", events[0])
+        self.assertEqual(events[0]["data"][setting_name], default_group.id)
+
     def test_change_realm_property(self) -> None:
         for prop in Realm.property_types:
             with self.settings(SEND_DIGEST_EMAILS=True):
@@ -3582,6 +3731,10 @@ class RealmPropertyActionTest(BaseAction):
             with self.settings(SEND_DIGEST_EMAILS=True):
                 self.do_set_realm_permission_group_setting_test(prop)
 
+        for prop in Realm.REALM_PERMISSION_GROUP_SETTINGS_WITH_NEW_API_FORMAT:
+            with self.settings(SEND_DIGEST_EMAILs=True):
+                self.do_set_realm_permission_group_setting_to_anonymous_groups_test(prop)
+
     def do_set_realm_user_default_setting_test(self, name: str) -> None:
         bool_tests: List[bool] = [True, False, True]
         test_values: Dict[str, Any] = dict(

zerver/tests/test_home.py

@@ -126,6 +126,7 @@ class HomeTest(ZulipTestCase):
         "realm_bot_domain",
         "realm_bots",
         "realm_can_access_all_users_group",
+        "realm_can_create_public_channel_group",
         "realm_create_multiuse_invite_group",
         "realm_create_private_stream_policy",
         "realm_create_public_stream_policy",

zerver/tests/test_realm.py

@@ -36,6 +36,7 @@ from zerver.actions.realm_settings import (
     do_set_realm_user_default_setting,
 )
 from zerver.actions.streams import do_deactivate_stream, merge_streams
+from zerver.actions.user_groups import check_add_user_group
 from zerver.lib.realm_description import get_realm_rendered_description, get_realm_text_description
 from zerver.lib.send_email import send_future_email
 from zerver.lib.streams import create_stream_if_needed
@@ -1541,6 +1542,14 @@ class RealmAPITest(ZulipTestCase):
         self.assertEqual(getattr(realm, setting_name), default_group.usergroup_ptr)
 
         for user_group in all_system_user_groups:
+            value = orjson.dumps(user_group.id).decode()
+            if setting_name in Realm.REALM_PERMISSION_GROUP_SETTINGS_WITH_NEW_API_FORMAT:
+                value = orjson.dumps(
+                    {
+                        "new": user_group.id,
+                    }
+                ).decode()
+
             if (
                 (
                     user_group.name == SystemGroups.EVERYONE_ON_INTERNET
@@ -1564,17 +1573,245 @@ class RealmAPITest(ZulipTestCase):
                     not in setting_permission_configuration.allowed_system_groups
                 )
             ):
-                value = orjson.dumps(user_group.id).decode()
-
                 result = self.client_patch("/json/realm", {setting_name: value})
                 self.assert_json_error(
                     result, f"'{setting_name}' setting cannot be set to '{user_group.name}' group."
                 )
                 continue
 
-            realm = self.update_with_api(setting_name, user_group.id)
+            realm = self.update_with_api(setting_name, value)
             self.assertEqual(getattr(realm, setting_name), user_group.usergroup_ptr)
 
+    def do_test_realm_permission_group_setting_update_api_with_anonymous_groups(
+        self, setting_name: str
+    ) -> None:
+        realm = get_realm("zulip")
+        othello = self.example_user("othello")
+        hamlet = self.example_user("hamlet")
+        leadership_group = check_add_user_group(realm, "leadership", [hamlet], acting_user=None)
+
+        moderators_group = NamedUserGroup.objects.get(
+            name=SystemGroups.MODERATORS, realm=realm, is_system_group=True
+        )
+
+        result = self.client_patch(
+            "/json/realm", {setting_name: orjson.dumps({"new": moderators_group.id}).decode()}
+        )
+        self.assert_json_success(result)
+        realm = get_realm("zulip")
+        self.assertEqual(getattr(realm, setting_name), moderators_group.usergroup_ptr)
+
+        # Try passing the old value as well.
+        admins_group = NamedUserGroup.objects.get(
+            name=SystemGroups.ADMINISTRATORS, realm=realm, is_system_group=True
+        )
+        result = self.client_patch(
+            "/json/realm",
+            {
+                setting_name: orjson.dumps(
+                    {"new": admins_group.id, "old": leadership_group.id}
+                ).decode()
+            },
+        )
+        self.assert_json_error(result, "'old' value does not match the expected value.")
+
+        result = self.client_patch(
+            "/json/realm",
+            {
+                setting_name: orjson.dumps(
+                    {
+                        "new": admins_group.id,
+                        "old": {
+                            "direct_members": [othello.id],
+                            "direct_subgroups": [moderators_group.id],
+                        },
+                    }
+                ).decode(),
+            },
+        )
+        self.assert_json_error(result, "'old' value does not match the expected value.")
+
+        result = self.client_patch(
+            "/json/realm",
+            {
+                setting_name: orjson.dumps(
+                    {"new": admins_group.id, "old": moderators_group.id}
+                ).decode()
+            },
+        )
+        realm = get_realm("zulip")
+        self.assertEqual(getattr(realm, setting_name), admins_group.usergroup_ptr)
+
+        result = self.client_patch(
+            "/json/realm",
+            {
+                setting_name: orjson.dumps(
+                    {
+                        "new": {
+                            "direct_members": [othello.id],
+                            "direct_subgroups": [leadership_group.id],
+                        }
+                    }
+                ).decode()
+            },
+        )
+        self.assert_json_success(result)
+        realm = get_realm("zulip")
+        self.assertCountEqual(list(getattr(realm, setting_name).direct_members.all()), [othello])
+        self.assertCountEqual(
+            list(getattr(realm, setting_name).direct_subgroups.all()), [leadership_group]
+        )
+
+        result = self.client_patch(
+            "/json/realm",
+            {
+                setting_name: orjson.dumps(
+                    {
+                        "new": {
+                            "direct_members": [hamlet.id],
+                            "direct_subgroups": [moderators_group.id],
+                        },
+                        "old": moderators_group.id,
+                    }
+                ).decode()
+            },
+        )
+        self.assert_json_error(result, "'old' value does not match the expected value.")
+
+        result = self.client_patch(
+            "/json/realm",
+            {
+                setting_name: orjson.dumps(
+                    {
+                        "new": {
+                            "direct_members": [hamlet.id],
+                            "direct_subgroups": [moderators_group.id],
+                        },
+                        "old": {
+                            "direct_members": [othello.id],
+                            "direct_subgroups": [moderators_group.id],
+                        },
+                    }
+                ).decode(),
+            },
+        )
+        self.assert_json_error(result, "'old' value does not match the expected value.")
+
+        result = self.client_patch(
+            "/json/realm",
+            {
+                setting_name: orjson.dumps(
+                    {
+                        "new": {
+                            "direct_members": [hamlet.id],
+                            "direct_subgroups": [moderators_group.id],
+                        },
+                        "old": {
+                            "direct_members": [othello.id],
+                            "direct_subgroups": [leadership_group.id],
+                        },
+                    }
+                ).decode()
+            },
+        )
+        self.assert_json_success(result)
+        realm = get_realm("zulip")
+        self.assertCountEqual(list(getattr(realm, setting_name).direct_members.all()), [hamlet])
+        self.assertCountEqual(
+            list(getattr(realm, setting_name).direct_subgroups.all()), [moderators_group]
+        )
+
+        result = self.client_patch(
+            "/json/realm",
+            {
+                setting_name: orjson.dumps(
+                    {
+                        "new": leadership_group.id,
+                        "old": {
+                            "direct_members": [hamlet.id],
+                            "direct_subgroups": [moderators_group.id],
+                        },
+                    }
+                ).decode()
+            },
+        )
+        self.assert_json_success(result)
+        realm = get_realm("zulip")
+        self.assertEqual(getattr(realm, setting_name), leadership_group.usergroup_ptr)
+
+        # Test that object with only one direct_subgroup is considered
+        # same as passing the named user group ID directly.
+        result = self.client_patch(
+            "/json/realm",
+            {
+                setting_name: orjson.dumps(
+                    {
+                        "new": {
+                            "direct_members": [],
+                            "direct_subgroups": [admins_group.id],
+                        },
+                        "old": {
+                            "direct_members": [],
+                            "direct_subgroups": [leadership_group.id],
+                        },
+                    }
+                ).decode()
+            },
+        )
+        self.assert_json_success(result)
+        realm = get_realm("zulip")
+        self.assertEqual(getattr(realm, setting_name), admins_group.usergroup_ptr)
+
+        # Test case when ALLOW_ANONYMOUS_GROUP_VALUED_SETTINGS is False.
+        with self.settings(ALLOW_ANONYMOUS_GROUP_VALUED_SETTINGS=False):
+            result = self.client_patch(
+                "/json/realm",
+                {
+                    setting_name: orjson.dumps(
+                        {
+                            "new": {
+                                "direct_members": [othello.id],
+                                "direct_subgroups": [moderators_group.id, leadership_group.id],
+                            }
+                        }
+                    ).decode()
+                },
+            )
+            self.assert_json_error(
+                result, f"{setting_name} can only be set to a single named user group."
+            )
+
+            result = self.client_patch(
+                "/json/realm",
+                {
+                    setting_name: orjson.dumps(
+                        {
+                            "new": {
+                                "direct_members": [],
+                                "direct_subgroups": [moderators_group.id],
+                            }
+                        }
+                    ).decode()
+                },
+            )
+            self.assert_json_success(result)
+            realm = get_realm("zulip")
+            self.assertEqual(getattr(realm, setting_name), moderators_group.usergroup_ptr)
+
+            result = self.client_patch(
+                "/json/realm",
+                {
+                    setting_name: orjson.dumps(
+                        {
+                            "new": leadership_group.id,
+                        }
+                    ).decode()
+                },
+            )
+            self.assert_json_success(result)
+            realm = get_realm("zulip")
+            self.assertEqual(getattr(realm, setting_name), leadership_group.usergroup_ptr)
+
     def test_update_realm_properties(self) -> None:
         for prop in Realm.property_types:
             # push_notifications_enabled is maintained by the server, not via the API.
@@ -1586,6 +1823,10 @@ class RealmAPITest(ZulipTestCase):
             with self.subTest(property=prop):
                 self.do_test_realm_permission_group_setting_update_api(prop)
 
+        for prop in Realm.REALM_PERMISSION_GROUP_SETTINGS_WITH_NEW_API_FORMAT:
+            with self.subTest(property=prop):
+                self.do_test_realm_permission_group_setting_update_api_with_anonymous_groups(prop)
+
     # Not in Realm.property_types because org_type has
     # a unique RealmAuditLog event_type.
     def test_update_realm_org_type(self) -> None:

zerver/views/realm.py

@@ -1,6 +1,7 @@
 from typing import Any, Dict, Mapping, Optional, Union
 
 from django.core.exceptions import ValidationError
+from django.db import transaction
 from django.http import HttpRequest, HttpResponse
 from django.shortcuts import render
 from django.utils.translation import gettext as _
@@ -35,7 +36,13 @@ from zerver.lib.response import json_success
 from zerver.lib.retention import parse_message_retention_days
 from zerver.lib.streams import access_stream_by_id
 from zerver.lib.typed_endpoint import ApiParamConfig, typed_endpoint
-from zerver.lib.user_groups import access_user_group_for_setting
+from zerver.lib.user_groups import (
+    GroupSettingChangeRequest,
+    access_user_group_for_setting,
+    get_group_setting_value_for_api,
+    parse_group_setting_value,
+    validate_group_setting_value_change,
+)
 from zerver.lib.validator import (
     check_bool,
     check_capped_url,
@@ -140,6 +147,7 @@ def update_realm(
     digest_emails_enabled: Optional[Json[bool]] = None,
     message_content_allowed_in_email_notifications: Optional[Json[bool]] = None,
     bot_creation_policy: Optional[Json[BotCreationPolicyEnum]] = None,
+    can_create_public_channel_group: Optional[Json[GroupSettingChangeRequest]] = None,
     create_public_stream_policy: Optional[Json[CommonPolicyEnum]] = None,
     create_private_stream_policy: Optional[Json[CommonPolicyEnum]] = None,
     create_web_public_stream_policy: Optional[Json[CreateWebPublicStreamPolicyEnum]] = None,
@@ -340,8 +348,8 @@ def update_realm(
         if k in realm.property_types:
             req_vars[k] = v
 
-        for permission_configuration in Realm.REALM_PERMISSION_GROUP_SETTINGS.values():
+        for setting_name, permission_configuration in Realm.REALM_PERMISSION_GROUP_SETTINGS.items():
-            if k == permission_configuration.id_field_name:
+            if k in [permission_configuration.id_field_name, setting_name]:
                 req_group_setting_vars[k] = v
 
     for k, v in req_vars.items():
@@ -355,22 +363,47 @@ def update_realm(
     for setting_name, permission_configuration in Realm.REALM_PERMISSION_GROUP_SETTINGS.items():
         setting_group_id_name = permission_configuration.id_field_name
 
-        assert setting_group_id_name in req_group_setting_vars
+        expected_current_setting_value = None
+        if setting_name in Realm.REALM_PERMISSION_GROUP_SETTINGS_WITH_NEW_API_FORMAT:
+            assert setting_name in req_group_setting_vars
+            if req_group_setting_vars[setting_name] is None:
+                continue
 
-        if req_group_setting_vars[setting_group_id_name] is not None and req_group_setting_vars[
+            setting_value = req_group_setting_vars[setting_name]
-            setting_group_id_name
+            new_setting_value = parse_group_setting_value(setting_value.new, setting_name)
-        ] != getattr(realm, setting_group_id_name):
+
-            user_group_id = req_group_setting_vars[setting_group_id_name]
+            if setting_value.old is not None:
+                expected_current_setting_value = parse_group_setting_value(
+                    setting_value.old, setting_name
+                )
+        else:
+            assert setting_group_id_name in req_group_setting_vars
+            if req_group_setting_vars[setting_group_id_name] is None:
+                continue
+            new_setting_value = req_group_setting_vars[setting_group_id_name]
+
+        current_value = getattr(realm, setting_name)
+        current_setting_api_value = get_group_setting_value_for_api(current_value)
+
+        if validate_group_setting_value_change(
+            current_setting_api_value, new_setting_value, expected_current_setting_value
+        ):
+            with transaction.atomic(durable=True):
                 user_group = access_user_group_for_setting(
-                user_group_id,
+                    new_setting_value,
                     user_profile,
                     setting_name=setting_name,
                     permission_configuration=permission_configuration,
+                    current_setting_value=current_value,
                 )
                 do_change_realm_permission_group_setting(
-                realm, setting_name, user_group, acting_user=user_profile
+                    realm,
+                    setting_name,
+                    user_group,
+                    old_setting_api_value=current_setting_api_value,
+                    acting_user=user_profile,
                 )
-            data[setting_name] = user_group_id
+            data[setting_name] = new_setting_value
 
     # The following realm properties do not fit the pattern above
     # authentication_methods is not supported by the do_set_realm_property