saml: Implement group sync.

Adds support for syncing group memberships for a user when logging in
via SAML. The list of group memberships is passed by the IdP in the
zulip_groups SAML attribute in the SAMLResponse.

docs/production/authentication-methods.md

@@ -610,6 +610,7 @@ other IdPs (identity providers). You can configure it as follows:
 
 [saml-help-center]: https://zulip.com/help/saml-authentication
 [user-role-help-center]: https://zulip.com/help/user-roles
+[user-groups-help-center]: https://zulip.com/help/user-groups
 
 ### IdP-initiated SSO
 
@@ -656,7 +657,7 @@ to the root and `engineering` subdomains:
 </saml2:Attribute>
 ```
 
-### Synchronizing user role or custom profile fields during login
+### Synchronizing data during login
 
 In contrast with SCIM or LDAP, the SAML protocol only allows Zulip to
 access data about a user when that user authenticates to Zulip using
@@ -670,15 +671,17 @@ offer SCIM or the fields one is interested in syncing change rarely
 enough that asking users to logout and then login again to resync
 their metadata might feel reasonable.
 
-Specifically, Zulip supports synchronizing the [user
+Specifically, Zulip supports synchronizing
+[group memberships][user-groups-help-center], the [user
 role][user-role-help-center] and [custom profile
 fields][custom-profile-fields] from the SAML provider.
 
-In order to use this functionality, configure
-`SOCIAL_AUTH_SYNC_ATTRS_DICT` in `/etc/zulip/settings.py` according to
-the instructions in the inline documentation in the file. Servers
-installed before Zulip 10.0 may want to [update inline comment
-documentation][update-inline-comments] first in order to access it.
+In order to use this functionality, configure `SOCIAL_AUTH_SYNC_ATTRS_DICT` in
+`/etc/zulip/settings.py` according to the instructions in the inline
+documentation in the file. Servers installed before Zulip 10.0 may want to
+[update inline comment documentation][update-inline-comments] first in order to
+access it. For configuring syncing of groups see
+[below][configure-saml-group-sync].
 
 Custom profile fields are only synchronized during login, not during
 account creation; we consider this [a
@@ -691,6 +694,46 @@ user who's coming from an invitation link, the IdP-provided role will
 take precedence over the role set in the invitation.
 :::
 
+[configure-saml-group-sync]: #synchronizing-group-membership-with-saml
+
+#### Synchronizing group membership with SAML
+
+Zulip 11.0+ includes support for syncing group memberships upon user
+login. To activate this feature, uncomment the `groups` field in the
+config in `SOCIAL_AUTH_SYNC_ATTRS_DICT` and configure the list as
+explained below. An example configuration might look like this:
+
+```python
+SOCIAL_AUTH_SYNC_ATTRS_DICT = {
+    "your_subdomain": {
+        "saml": {
+            "groups": ["group1", ("samlgroup2", "zulipgroup2"), "group3"],
+        }
+    }
+}
+```
+
+The tuple syntax (`("samlgroup2", "zulipgroup2")`) should be used when
+the Zulip group that you'd like to sync does not have exactly the same
+name as the SAML group.
+
+Your SAML IdP will need to provide the list of SAML group names in the
+`zulip_groups` attribute of the `SAMLResponse`. When a user logs in
+using SAML, groups are synced as follows:
+
+1. If a Zulip group name does not occur in the
+   `SOCIAL_AUTH_SYNC_ATTRS_DICT` groups list, that group's membership
+   is managed entirely in Zulip.
+1. Otherwise, if the group appears in `zulip_groups` in the
+   `SAMLResponse`, the user is added to that group (if not already a
+   member).
+1. Otherwise, the user is removed from that group (if currently a
+   member).
+
+Only direct membership of groups is synced through this protocol;
+subgroups of Zulip groups are managed entirely [inside
+Zulip](https://zulip.com/help/manage-user-groups#add-user-groups-to-a-group).
+
 ### SCIM
 
 Many SAML IdPs also offer SCIM provisioning to manage automatically

zerver/tests/test_auth_backends.py

@@ -3408,17 +3408,349 @@ class SAMLAuthBackendTest(SocialAuthBase):
         self.user_profile.refresh_from_db()
         self.assertEqual(self.user_profile.role, UserProfile.ROLE_REALM_OWNER)
 
+    def test_social_auth_group_sync(self) -> None:
+        realm = get_realm("zulip")
+        hamlet = self.example_user("hamlet")
+        testgroup1 = create_user_group_in_database("testgroup1", [], realm, acting_user=hamlet)
+        testgroup2 = create_user_group_in_database("testgroup2", [], realm, acting_user=hamlet)
+
+        sync_custom_attrs_dict = {
+            "zulip": {
+                "saml": {
+                    "role": "zulip_role",
+                    "groups": ["testgroup1", ("samlgroup2", "testgroup2")],
+                }
+            }
+        }
+
+        with (
+            self.settings(
+                SOCIAL_AUTH_SYNC_ATTRS_DICT=sync_custom_attrs_dict,
+            ),
+            self.assertLogs(self.logger_string) as mock_log,
+        ):
+            account_data_dict = self.get_account_data_dict(email=self.email, name=self.name)
+            result = self.social_auth_test(
+                account_data_dict,
+                subdomain="zulip",
+                extra_attributes=dict(zulip_groups=["testgroup1", "samlgroup2"]),
+            )
+        data = load_subdomain_token(result)
+        self.assertEqual(data["email"], self.email)
+        self.assertEqual(result.status_code, 302)
+
+        self.assertTrue(
+            is_user_in_group(
+                testgroup1.id,
+                hamlet,
+                direct_member_only=True,
+            )
+        )
+        self.assertTrue(
+            is_user_in_group(
+                testgroup2.id,
+                hamlet,
+                direct_member_only=True,
+            )
+        )
+        # Verify the expected log line revealing the internal details of the incoming groups -> Zulip groups translation.
+        self.assertIn(
+            self.logger_output(
+                f"social_auth_sync_user_attributes:<user:{hamlet.id}>: received group names: ['samlgroup2', 'testgroup1']|intended Zulip groups: ['testgroup1', 'testgroup2']. group mapping used: {{'testgroup1': 'testgroup1', 'samlgroup2': 'testgroup2'}}",
+                type="info",
+            ),
+            mock_log.output,
+        )
+
+        with self.settings(
+            SOCIAL_AUTH_SYNC_ATTRS_DICT=sync_custom_attrs_dict,
+        ):
+            account_data_dict = self.get_account_data_dict(email=self.email, name=self.name)
+            result = self.social_auth_test(
+                account_data_dict,
+                subdomain="zulip",
+                extra_attributes=dict(zulip_groups=["testgroup1"]),
+            )
+        data = load_subdomain_token(result)
+        self.assertEqual(data["email"], self.email)
+        self.assertEqual(result.status_code, 302)
+
+        self.assertTrue(
+            is_user_in_group(
+                testgroup1.id,
+                hamlet,
+                direct_member_only=True,
+            )
+        )
+        self.assertFalse(
+            is_user_in_group(
+                testgroup2.id,
+                hamlet,
+                direct_member_only=True,
+            )
+        )
+
+        with self.settings(
+            SOCIAL_AUTH_SYNC_ATTRS_DICT=sync_custom_attrs_dict,
+        ):
+            account_data_dict = self.get_account_data_dict(email=self.email, name=self.name)
+            result = self.social_auth_test(
+                account_data_dict,
+                subdomain="zulip",
+                extra_attributes=dict(zulip_groups=[]),
+            )
+        data = load_subdomain_token(result)
+        self.assertEqual(data["email"], self.email)
+        self.assertEqual(result.status_code, 302)
+
+        self.assertFalse(
+            is_user_in_group(
+                testgroup1.id,
+                hamlet,
+                direct_member_only=True,
+            )
+        )
+        self.assertFalse(
+            is_user_in_group(
+                testgroup2.id,
+                hamlet,
+                direct_member_only=True,
+            )
+        )
+
+        bulk_add_members_to_user_groups([testgroup1, testgroup2], [hamlet.id], acting_user=None)
+
+        with self.settings(
+            # If the realm is not configured for group sync, group memberships of course should be
+            # unaffected by zulip_groups attr.
+            SOCIAL_AUTH_SYNC_ATTRS_DICT={"zulip": {"saml": {}}},
+        ):
+            account_data_dict = self.get_account_data_dict(email=self.email, name=self.name)
+            result = self.social_auth_test(
+                account_data_dict,
+                subdomain="zulip",
+                extra_attributes=dict(zulip_groups=[]),
+            )
+        data = load_subdomain_token(result)
+        self.assertEqual(data["email"], self.email)
+        self.assertEqual(result.status_code, 302)
+
+        self.assertTrue(
+            is_user_in_group(
+                testgroup1.id,
+                hamlet,
+                direct_member_only=True,
+            )
+        )
+        self.assertTrue(
+            is_user_in_group(
+                testgroup2.id,
+                hamlet,
+                direct_member_only=True,
+            )
+        )
+
+        with self.settings(
+            SOCIAL_AUTH_SYNC_ATTRS_DICT=sync_custom_attrs_dict,
+        ):
+            account_data_dict = self.get_account_data_dict(email=self.email, name=self.name)
+            # Simulate a SAMLResponse without zulip_groups attribute being specified in it at all.
+            # As the realm is configured for group sync, that should be treated as
+            # "user should not be a member of any of the groups configured for sync"
+            result = self.social_auth_test(
+                account_data_dict,
+                subdomain="zulip",
+            )
+        data = load_subdomain_token(result)
+        self.assertEqual(data["email"], self.email)
+        self.assertEqual(result.status_code, 302)
+
+        self.assertFalse(
+            is_user_in_group(
+                testgroup1.id,
+                hamlet,
+                direct_member_only=True,
+            )
+        )
+        self.assertFalse(
+            is_user_in_group(
+                testgroup2.id,
+                hamlet,
+                direct_member_only=True,
+            )
+        )
+
     @override_settings(TERMS_OF_SERVICE_VERSION=None)
-    def test_social_auth_create_user_with_synced_role(self) -> None:
+    def test_social_auth_create_user_with_synced_role_and_groups(self) -> None:
         email = "newuser@zulip.com"
         name = "Full Name"
         subdomain = "zulip"
+        desdemona = self.example_user("desdemona")
         realm = get_realm("zulip")
 
         account_data_dict = self.get_account_data_dict(email=email, name=name)
         idps_dict = copy.deepcopy(settings.SOCIAL_AUTH_SAML_ENABLED_IDPS)
         idps_dict["test_idp"]["extra_attrs"] = ["zulip_role"]
 
+        testgroup1 = create_user_group_in_database("testgroup1", [], realm, acting_user=desdemona)
+        testgroup2 = create_user_group_in_database("testgroup2", [], realm, acting_user=desdemona)
+
+        sync_custom_attrs_dict = {
+            "zulip": {
+                "saml": {
+                    "role": "zulip_role",
+                    "groups": ["testgroup1", ("samlgroup2", "testgroup2")],
+                }
+            }
+        }
+
+        with (
+            self.settings(
+                SOCIAL_AUTH_SAML_ENABLED_IDPS=idps_dict,
+                SOCIAL_AUTH_SYNC_ATTRS_DICT=sync_custom_attrs_dict,
+            ),
+            self.assertLogs(self.logger_string, level="INFO") as mock_logger,
+        ):
+            result = self.social_auth_test(
+                account_data_dict,
+                subdomain="zulip",
+                is_signup=True,
+                extra_attributes=dict(
+                    mobilePhone=["123412341234"],
+                    birthday=["2021-01-01"],
+                    zulip_role=["owner"],
+                    zulip_groups=["testgroup1"],
+                ),
+            )
+
+        with self.assertLogs("", level="INFO") as mock_root_logger:
+            self.stage_two_of_registration(
+                result, realm, subdomain, email, name, name, self.BACKEND_CLASS.full_name_validated
+            )
+        user_profile = get_user_by_delivery_email(email, realm)
+        self.assertEqual(user_profile.role, UserProfile.ROLE_REALM_OWNER)
+        self.assertTrue(
+            is_user_in_group(
+                testgroup1.id,
+                user_profile,
+                direct_member_only=True,
+            )
+        )
+        self.assertFalse(
+            is_user_in_group(
+                testgroup2.id,
+                user_profile,
+                direct_member_only=True,
+            )
+        )
+
+        self.assertEqual(
+            mock_logger.output[0],
+            self.logger_output(
+                "social_auth_sync_user_attributes:<new user signup>: received group names: ['testgroup1']|intended Zulip groups: ['testgroup1']. group mapping used: {'testgroup1': 'testgroup1', 'samlgroup2': 'testgroup2'}",
+                type="info",
+            ),
+        )
+        self.assertEqual(
+            mock_logger.output[1],
+            self.logger_output("Returning role owner for user creation", type="info"),
+        )
+
+        self.assertIn(
+            f"INFO:root:Syncing groups post-registration for new user {user_profile.id} in realm {realm.id}",
+            mock_root_logger.output[0],
+        )
+
+    @override_settings(TERMS_OF_SERVICE_VERSION=None)
+    def test_social_auth_create_user_from_multiuse_invite_role_and_group_sync(self) -> None:
+        email = "newuser@zulip.com"
+        name = "Full Name"
+        subdomain = "zulip"
+        desdemona = self.example_user("desdemona")
+        realm = get_realm("zulip")
+
+        account_data_dict = self.get_account_data_dict(email=email, name=name)
+        idps_dict = copy.deepcopy(settings.SOCIAL_AUTH_SAML_ENABLED_IDPS)
+        idps_dict["test_idp"]["extra_attrs"] = ["zulip_role"]
+
+        testgroup1 = create_user_group_in_database("testgroup1", [], realm, acting_user=desdemona)
+        testgroup2 = create_user_group_in_database("testgroup2", [], realm, acting_user=desdemona)
+
+        sync_custom_attrs_dict = {
+            "zulip": {
+                "saml": {
+                    "role": "zulip_role",
+                    "groups": ["testgroup1", ("samlgroup2", "testgroup2")],
+                }
+            }
+        }
+
+        invite = MultiuseInvite.objects.create(
+            realm=realm,
+            referred_by=desdemona,
+            # Set a role on the invite to verify that it gets ignored in favor
+            # of the role implied by the zulip_role attribute.
+            invited_as=PreregistrationUser.INVITE_AS["REALM_ADMIN"],
+        )
+        invite.groups.set([testgroup1, testgroup2])
+        create_confirmation_link(invite, Confirmation.MULTIUSE_INVITE)
+        multiuse_confirmation = Confirmation.objects.all().last()
+        assert multiuse_confirmation is not None
+        multiuse_object_key = multiuse_confirmation.confirmation_key
+        account_data_dict = self.get_account_data_dict(email=email, name=name)
+        with (
+            self.settings(
+                SOCIAL_AUTH_SAML_ENABLED_IDPS=idps_dict,
+                SOCIAL_AUTH_SYNC_ATTRS_DICT=sync_custom_attrs_dict,
+            ),
+        ):
+            result = self.social_auth_test(
+                account_data_dict,
+                subdomain="zulip",
+                is_signup=True,
+                multiuse_object_key=multiuse_object_key,
+                extra_attributes=dict(
+                    zulip_role=["member"],
+                    zulip_groups=["testgroup1"],
+                ),
+            )
+        with self.assertLogs("", level="INFO") as mock_root_logger:
+            self.stage_two_of_registration(
+                result, realm, subdomain, email, name, name, self.BACKEND_CLASS.full_name_validated
+            )
+        user_profile = get_user_by_delivery_email(email, realm)
+        self.assertEqual(user_profile.role, UserProfile.ROLE_MEMBER)
+        self.assertTrue(
+            is_user_in_group(
+                testgroup1.id,
+                user_profile,
+                direct_member_only=True,
+            )
+        )
+        self.assertFalse(
+            is_user_in_group(
+                testgroup2.id,
+                user_profile,
+                direct_member_only=True,
+            )
+        )
+
+        self.assertIn(
+            f"INFO:root:Syncing groups post-registration for new user {user_profile.id} in realm {realm.id}",
+            mock_root_logger.output[0],
+        )
+
+    @override_settings(TERMS_OF_SERVICE_VERSION=None)
+    def test_social_auth_create_user_with_synced_role_only(self) -> None:
+        email = "newuser@zulip.com"
+        name = "Full Name"
+        subdomain = "zulip"
+        realm = get_realm("zulip")
+
+        account_data_dict = self.get_account_data_dict(email=email, name=name)
+
+        idps_dict = copy.deepcopy(settings.SOCIAL_AUTH_SAML_ENABLED_IDPS)
+        idps_dict["test_idp"]["extra_attrs"] = ["zulip_role"]
         sync_custom_attrs_dict = {
             "zulip": {
                 "saml": {
@@ -3432,24 +3764,28 @@ class SAMLAuthBackendTest(SocialAuthBase):
                 SOCIAL_AUTH_SAML_ENABLED_IDPS=idps_dict,
                 SOCIAL_AUTH_SYNC_ATTRS_DICT=sync_custom_attrs_dict,
             ),
-            self.assertLogs(self.logger_string, level="INFO") as m,
+            self.assertLogs(self.logger_string, level="INFO") as mock_logger,
         ):
             result = self.social_auth_test(
                 account_data_dict,
                 subdomain="zulip",
                 is_signup=True,
                 extra_attributes=dict(
-                    mobilePhone=["123412341234"], birthday=["2021-01-01"], zulip_role=["owner"]
+                    zulip_role=["owner"],
+                    # Groups won't get synced, despite being passed - group sync
+                    # is not configured.
+                    zulip_groups=["samlgroup1"],
                 ),
             )
-
             self.stage_two_of_registration(
                 result, realm, subdomain, email, name, name, self.BACKEND_CLASS.full_name_validated
             )
+
         user_profile = get_user_by_delivery_email(email, realm)
         self.assertEqual(user_profile.role, UserProfile.ROLE_REALM_OWNER)
         self.assertEqual(
-            m.output[0], self.logger_output("Returning role owner for user creation", type="info")
+            mock_logger.output[0],
+            self.logger_output("Returning role owner for user creation", type="info"),
         )
 
     def test_social_auth_sync_field_not_existing(self) -> None:

zerver/views/auth.py

@@ -165,6 +165,7 @@ def maybe_send_to_registration(
     desktop_flow_otp: str | None = None,
     full_name: str = "",
     full_name_validated: bool = False,
+    group_memberships_sync_map: dict[str, bool] | None = None,
     is_signup: bool = False,
     mobile_flow_otp: str | None = None,
     multiuse_object_key: str = "",
@@ -227,6 +228,18 @@ def maybe_send_to_registration(
                 expiry_seconds=EXPIRABLE_SESSION_VAR_DEFAULT_EXPIRY_SECS,
             )
 
+    if group_memberships_sync_map:
+        set_expirable_session_var(
+            request.session,
+            "registration_group_memberships_sync_map",
+            orjson.dumps(group_memberships_sync_map).decode(),
+            expiry_seconds=EXPIRABLE_SESSION_VAR_DEFAULT_EXPIRY_SECS,
+        )
+    elif "registration_group_memberships_sync_map" in request.session:  # nocoverage
+        # Ensure it isn't possible to leak this state across
+        # registration attempts.
+        del request.session["registration_group_memberships_sync_map"]
+
     try:
         # TODO: This should use get_realm_from_request, but a bunch of tests
         # rely on mocking get_subdomain here, so they'll need to be tweaked first.
@@ -360,6 +373,7 @@ def register_remote_user(request: HttpRequest, result: ExternalAuthResult) -> Ht
         "email",
         "full_name",
         "role",
+        "group_memberships_sync_map",
         "mobile_flow_otp",
         "desktop_flow_otp",
         "is_signup",

zerver/views/registration.py

@@ -135,6 +135,7 @@ from zproject.backends import (
     get_external_method_dicts,
     ldap_auth_enabled,
     password_auth_enabled,
+    sync_groups,
 )
 
 logger = logging.getLogger("")
@@ -842,6 +843,8 @@ def registration_helper(
                 # duplicate email address.  Redirect them to the login
                 # form.
                 return redirect_to_email_login_url(email)
+            else:
+                sync_groups_post_registration(request=request, user_profile=user_profile)
 
         if realm_creation:
             # Because for realm creation, registration happens on the
@@ -914,6 +917,34 @@ def registration_helper(
     return TemplateResponse(request, "zerver/register.html", context=context)
 
 
+def sync_groups_post_registration(request: HttpRequest, user_profile: UserProfile) -> None:
+    group_memberships_sync_data = get_expirable_session_var(
+        request.session, "registration_group_memberships_sync_map", delete=True
+    )
+    if not group_memberships_sync_data:
+        return
+
+    group_memberships_sync_map = orjson.loads(group_memberships_sync_data)
+    if group_memberships_sync_map:
+        logger.info(
+            "Syncing groups post-registration for new user %s in realm %s: %s",
+            user_profile.id,
+            user_profile.realm_id,
+            group_memberships_sync_map,
+        )
+        assert isinstance(group_memberships_sync_map, dict)
+        sync_groups(
+            all_group_names=set(group_memberships_sync_map.keys()),
+            intended_group_names={
+                group_name
+                for group_name, is_member in group_memberships_sync_map.items()
+                if is_member
+            },
+            user_profile=user_profile,
+            logger=logger,
+        )
+
+
 def login_and_go_to_home(request: HttpRequest, user_profile: UserProfile) -> HttpResponse:
     mobile_flow_otp = get_expirable_session_var(
         request.session, "registration_mobile_flow_otp", delete=True

zproject/backends.py

@@ -17,6 +17,7 @@ import json
 import logging
 from abc import ABC, abstractmethod
 from collections.abc import Callable
+from dataclasses import dataclass
 from email.headerregistry import Address
 from typing import Any, TypedDict, TypeVar, cast
 
@@ -1434,6 +1435,7 @@ class ExternalAuthDataDict(TypedDict, total=False):
     desktop_flow_otp: str | None
     multiuse_object_key: str
     full_name_validated: bool
+    group_memberships_sync_map: dict[str, bool] | None
     # The mobile app doesn't actually use a session, so this
     # data is not applicable there.
     params_to_store_in_authenticated_session: dict[str, str]
@@ -1696,9 +1698,15 @@ def sync_groups(
     logger.debug("Finished group sync for user %s", user_id)
 
 
+@dataclass
+class SocialAuthSyncNewUserInfo:
+    role: int | None
+    group_memberships_sync_map: dict[str, bool]
+
+
 def social_auth_sync_user_attributes(
     realm: Realm, user_profile: UserProfile | None, extra_attrs: dict[str, Any], backend: Any
-) -> int | None:
+) -> SocialAuthSyncNewUserInfo | None:
     """
     Syncs user attributes based on the SOCIAL_AUTH_SYNC_ATTRS_DICT setting.
     Only supports:
@@ -1714,11 +1722,81 @@ def social_auth_sync_user_attributes(
     # Unlike LDAP or SCIM, this hook can only do syncing during the authentication
     # flow, as that's when the data is provided and we don't have a way to query
     # for it otherwise.
-    assert backend.name == "saml"
+    if backend.name != "saml":
+        assert not extra_attrs
+        return None
 
     attrs_by_backend = settings.SOCIAL_AUTH_SYNC_ATTRS_DICT.get(realm.subdomain, {})
-    profile_field_name_to_attr_name = attrs_by_backend.get(backend.name, {})
-    if not extra_attrs or not profile_field_name_to_attr_name:
+    attrs_config = attrs_by_backend.get(backend.name, {})
+    external_group_name_to_zulip_group_name: dict[str, str] = {}
+    groups_config_list = cast(list[str | tuple[str, str]], attrs_config.get("groups", []))
+    if groups_config_list:
+        # Group sync is only supported for SAML for the foreseeable time.
+        assert backend.name == "saml"
+
+        for group_name in groups_config_list:
+            # the objects in the config are either straight-forward group names
+            # or tuples (<saml group name>, <zulip group name>) indicating the
+            # obvious mapping.
+            if isinstance(group_name, str):
+                external_group_name_to_zulip_group_name[group_name] = group_name
+            else:
+                saml_group_name, zulip_group_name = group_name
+                external_group_name_to_zulip_group_name[saml_group_name] = zulip_group_name
+
+    should_sync_groups = bool(external_group_name_to_zulip_group_name)
+
+    profile_field_name_to_attr_name = cast(
+        dict[str, str], {key: value for key, value in attrs_config.items() if key != "groups"}
+    )
+
+    if should_sync_groups:
+        syncable_group_names = set(external_group_name_to_zulip_group_name.values())
+
+        # pop zulip_groups from extra_attrs, so that only user attribute sync
+        # values remain there.
+        # zulip_groups being absent should be treated as if no group memberships
+        # are desired. That's because Okta doesn't send the SAML attribute at all
+        # if the user has no group memberships. Thus we treat this absence as
+        # an empty list.
+        all_received_group_names = extra_attrs.pop("zulip_groups", [])
+        external_group_names = [
+            name
+            for name in all_received_group_names
+            # Ignore group names which aren't configured.
+            if name in external_group_name_to_zulip_group_name
+        ]
+        intended_group_names = {
+            external_group_name_to_zulip_group_name[external_name]
+            for external_name in external_group_names
+        }
+
+        # It's important to log the information about what we received in the request and what Zulip groups
+        # that was translated to based on the configuration - otherwise debugging a misconfiguration would be
+        # a very painful process.
+        #
+        # Notably, it's generally expected for the list of received groups to be shorter than the "translated"
+        # list of intended Zulip groups. The expected way for admins to configure what is sent in zulip_groups
+        # is to send *all* the groups the user belongs to in their user directory. That is very likely to contain
+        # some groups that are irrelevant to their Zulip setup and thus shouldn't be translated into any
+        # Zulip group memberships.
+        # For example, Okta sends the Everyone group, which is just a built-in group inside of Okta described
+        # in their documentation as
+        # "catch-all group where every single user in the Okta instance automatically lands".
+        # That obviously will be irrelevant to Zulip group memberships in almost all configurations.
+        user_string = (
+            f"<user:{user_profile.id}>" if user_profile is not None else "<new user signup>"
+        )
+        backend.logger.info(
+            "social_auth_sync_user_attributes:%s: received group names: %s|intended Zulip groups: %s. group mapping used: %s",
+            user_string,
+            sorted(all_received_group_names),
+            sorted(intended_group_names),
+            external_group_name_to_zulip_group_name,
+        )
+
+    should_sync_user_attrs = extra_attrs and profile_field_name_to_attr_name
+    if not (should_sync_user_attrs or should_sync_groups):
         return None
 
     user_id = None
@@ -1753,12 +1831,22 @@ def social_auth_sync_user_attributes(
 
     if user_profile is None:
         # We don't support user creation with custom profile fields, so just
-        # return role so that it can be plumbed through to the signup flow.
+        # return role and group memberships so that it can be plumbed through to the signup flow.
         if new_role is not None:
             backend.logger.info(
                 "Returning role %s for user creation", UserProfile.ROLE_ID_TO_API_NAME[new_role]
             )
-        return new_role
+        if should_sync_groups:
+            group_memberships_sync_map = {
+                group_name: (group_name in intended_group_names)
+                for group_name in syncable_group_names
+            }
+        else:
+            group_memberships_sync_map = {}
+
+        return SocialAuthSyncNewUserInfo(
+            role=new_role, group_memberships_sync_map=group_memberships_sync_map
+        )
 
     # Based on the information collected above, sync what's needed for the user_profile.
     old_role = user_profile.role
@@ -1777,6 +1865,14 @@ def social_auth_sync_user_attributes(
             str(e),
         )
 
+    if should_sync_groups:
+        sync_groups(
+            all_group_names=syncable_group_names,
+            intended_group_names=intended_group_names,
+            user_profile=user_profile,
+            logger=backend.logger,
+        )
+
     return None
 
 
@@ -2037,9 +2133,7 @@ def social_auth_finish(
         is_signup = False
 
     extra_attrs = return_data.get("extra_attrs", {})
-    role_for_new_user = None
-    if extra_attrs:
-        role_for_new_user = social_auth_sync_user_attributes(
+    social_auth_sync_new_user_info = social_auth_sync_user_attributes(
         realm, user_profile, extra_attrs, backend
     )
 
@@ -2101,7 +2195,19 @@ def social_auth_finish(
         params_to_store_in_authenticated_session=backend.get_params_to_store_in_authenticated_session(),
     )
     if user_profile is None:
-        data_dict.update(dict(full_name=full_name, email=email_address, role=role_for_new_user))
+        data_dict.update(
+            dict(
+                full_name=full_name,
+                email=email_address,
+            )
+        )
+        if social_auth_sync_new_user_info is not None:
+            data_dict.update(
+                dict(
+                    role=social_auth_sync_new_user_info.role,
+                    group_memberships_sync_map=social_auth_sync_new_user_info.group_memberships_sync_map,
+                )
+            )
 
     result = ExternalAuthResult(user_profile=user_profile, data_dict=data_dict)
 
@@ -2547,6 +2653,10 @@ class ZulipSAMLIdentityProvider(SAMLIdentityProvider):
 
         extra_attr_names = self.conf.get("extra_attrs", [])
         result["extra_attrs"] = {}
+
+        if (groups_list := attributes.get("zulip_groups")) is not None:
+            result["extra_attrs"]["zulip_groups"] = groups_list
+
         for extra_attr_name in extra_attr_names:
             result["extra_attrs"][extra_attr_name] = self.get_attr(
                 attributes=attributes, conf_key=None, default_attribute=extra_attr_name

zproject/computed_settings.py

@@ -1215,6 +1215,13 @@ def ensure_dict_path(d: dict[str, Any], keys: list[str]) -> None:
         d = d[key]
 
 
+for dict_for_subdomain in SOCIAL_AUTH_SYNC_ATTRS_DICT.values():
+    for attrs_map in dict_for_subdomain.values():
+        if "zulip_groups" in attrs_map.values():
+            raise AssertionError(
+                "zulip_groups can't be listed as a SAML attribute in SOCIAL_AUTH_SYNC_ATTRS_DICT"
+            )
+
 SOCIAL_AUTH_PIPELINE = [
     "social_core.pipeline.social_auth.social_details",
     "zproject.backends.social_auth_associate_user",

zproject/default_settings.py

@@ -93,6 +93,7 @@ SOCIAL_AUTH_SAML_TECHNICAL_CONTACT: dict[str, str] | None = None
 SOCIAL_AUTH_SAML_SUPPORT_CONTACT: dict[str, str] | None = None
 SOCIAL_AUTH_SAML_ENABLED_IDPS: dict[str, SAMLIdPConfigDict] = {}
 SOCIAL_AUTH_SAML_SECURITY_CONFIG: dict[str, Any] = {}
+
 # Set this to True to enforce that any configured IdP needs to specify
 # the limit_to_subdomains setting to be considered valid:
 SAML_REQUIRE_LIMIT_TO_SUBDOMAINS = False
@@ -112,7 +113,7 @@ SOCIAL_AUTH_APPLE_EMAIL_AS_USERNAME = True
 SOCIAL_AUTH_OIDC_ENABLED_IDPS: dict[str, OIDCIdPConfigDict] = {}
 SOCIAL_AUTH_OIDC_FULL_NAME_VALIDATED = False
 
-SOCIAL_AUTH_SYNC_ATTRS_DICT: dict[str, dict[str, dict[str, str]]] = {}
+SOCIAL_AUTH_SYNC_ATTRS_DICT: dict[str, dict[str, dict[str, str | list[str | tuple[str, str]]]]] = {}
 
 # Other auth
 SSO_APPEND_DOMAIN: str | None = None

zproject/prod_settings_template.py

@@ -519,6 +519,10 @@ SOCIAL_AUTH_SAML_SUPPORT_CONTACT = {
 #             # Specify custom profile fields with a custom__ prefix for the
 #             # Zulip field name.
 #             "custom__title": "title",
+#             # Sync the membership of the listed Zulip groups with
+#             # the list of group names sent in the "zulip_groups"
+#             # attribute in the SAMLResponse.
+#             "groups": ["group1", "group2", ("samlgroup3", "zulipgroup3"), "group4"],
 #         }
 #     }
 # }