stream: Guest users cannot get metadata access to channel via groups.

This commit is contained in:
Shubham Padia
2025-02-12 11:43:31 +00:00
committed by Tim Abbott
parent 77d3029ec5
commit 1db2487f1c
4 changed files with 52 additions and 8 deletions

View File

@@ -574,10 +574,19 @@ def access_stream_for_delete_or_update_requiring_metadata_access(
def has_metadata_access_to_channel_via_groups(
user_profile: UserProfile,
user_recursive_group_ids: set[int],
can_administer_channel_group_id: int,
can_add_subscribers_group_id: int,
) -> bool:
for setting_name in Stream.stream_permission_group_settings_granting_metadata_access:
permission_configuration = Stream.stream_permission_group_settings[setting_name]
if not permission_configuration.allow_everyone_group and user_profile.is_guest:
return False
# It's best to just check the variables directly here since it
# becomes complicated to create an automated loop for both settings
# and values because of https://github.com/python/mypy/issues/5382.
return (
can_administer_channel_group_id in user_recursive_group_ids
or can_add_subscribers_group_id in user_recursive_group_ids
@@ -614,6 +623,7 @@ def check_basic_stream_access(
) # nocoverage
) # nocoverage
if has_metadata_access_to_channel_via_groups(
user_profile,
user_group_membership_details.user_recursive_group_ids,
stream.can_administer_channel_group_id,
stream.can_add_subscribers_group_id,

View File

@@ -418,6 +418,7 @@ def validate_user_access_to_subscribers_helper(
)
if has_metadata_access_to_channel_via_groups(
user_profile,
user_group_membership_details.user_recursive_group_ids,
stream_dict["can_administer_channel_group_id"],
stream_dict["can_add_subscribers_group_id"],
@@ -591,7 +592,10 @@ def has_metadata_access_to_previously_subscribed_stream(
if stream_dict["invite_only"]:
return user_profile.is_realm_admin or has_metadata_access_to_channel_via_groups(
user_recursive_group_ids, can_administer_channel_group_id, can_add_subscribers_group_id
user_profile,
user_recursive_group_ids,
can_administer_channel_group_id,
can_add_subscribers_group_id,
)
return True
@@ -733,7 +737,10 @@ def gather_subscriptions_helper(
can_administer_channel_group_id = raw_stream_dict["can_administer_channel_group_id"]
can_add_subscribers_group_id = raw_stream_dict["can_add_subscribers_group_id"]
has_metadata_access = has_metadata_access_to_channel_via_groups(
user_recursive_group_ids, can_administer_channel_group_id, can_add_subscribers_group_id
user_profile,
user_recursive_group_ids,
can_administer_channel_group_id,
can_add_subscribers_group_id,
)
if is_public or user_profile.is_realm_admin or has_metadata_access:
slim_stream_dict = build_stream_dict_for_never_sub(

View File

@@ -184,6 +184,14 @@ class Stream(models.Model):
stream_permission_group_settings.keys()
)
stream_permission_group_settings_granting_metadata_access = [
"can_add_subscribers_group",
"can_administer_channel_group",
]
assert set(stream_permission_group_settings_granting_metadata_access).issubset(
stream_permission_group_settings.keys()
)
class Meta:
indexes = [
models.Index(Upper("name"), name="upper_stream_name_idx"),

View File

@@ -7840,6 +7840,7 @@ class AccessStreamTest(ZulipTestCase):
othello = self.example_user("othello")
iago = self.example_user("iago")
polonius = self.example_user("polonius")
# Realm admin cannot access the private stream
with self.assertRaisesRegex(JsonableError, "Invalid channel ID"):
@@ -7865,8 +7866,8 @@ class AccessStreamTest(ZulipTestCase):
with self.assertRaisesRegex(JsonableError, "Invalid channel name 'new_private_stream'"):
access_stream_by_name(othello, stream.name, require_content_access=False)
othello_group = check_add_user_group(
othello.realm, "user_profile_group", [othello], acting_user=othello
polonius_and_othello_group = check_add_user_group(
othello.realm, "user_profile_group", [othello, polonius], acting_user=othello
)
nobody_group = NamedUserGroup.objects.get(
name="role:nobody", is_system_group=True, realm=othello.realm
@@ -7875,13 +7876,19 @@ class AccessStreamTest(ZulipTestCase):
do_change_stream_group_based_setting(
stream,
"can_administer_channel_group",
othello_group,
polonius_and_othello_group,
acting_user=None,
)
# Channel admins can access private stream if
# require_content_access is set to False
access_stream_by_id(othello, stream.id, require_content_access=False)
access_stream_by_name(othello, stream.name, require_content_access=False)
# Guest user who is a channel admin cannot access a stream via
# groups if they are not subscribed to it.
with self.assertRaisesRegex(JsonableError, "Invalid channel ID"):
access_stream_by_id(polonius, stream.id, require_content_access=False)
with self.assertRaisesRegex(JsonableError, "Invalid channel name 'new_private_stream'"):
access_stream_by_name(polonius, stream.name, require_content_access=False)
do_change_stream_group_based_setting(
stream,
"can_administer_channel_group",
@@ -7892,13 +7899,25 @@ class AccessStreamTest(ZulipTestCase):
do_change_stream_group_based_setting(
stream,
"can_add_subscribers_group",
othello_group,
polonius_and_othello_group,
acting_user=None,
)
# Users in `can_add_subscribers_group` can access private
# stream if require_content_access is set to True
access_stream_by_id(othello, stream.id, require_content_access=False)
access_stream_by_name(othello, stream.name, require_content_access=False)
# Users in `can_add_subscribers_group` can access private
# stream if require_content_access is set to True
access_stream_by_id(othello, stream.id, require_content_access=True)
access_stream_by_name(othello, stream.name, require_content_access=True)
# Guest user who cannot access a stream via groups if they are
# part of `can_add_subscribers_group` but not subscribed to it.
with self.assertRaisesRegex(JsonableError, "Invalid channel ID"):
access_stream_by_id(polonius, stream.id, require_content_access=False)
with self.assertRaisesRegex(JsonableError, "Invalid channel name 'new_private_stream'"):
access_stream_by_name(polonius, stream.name, require_content_access=False)
with self.assertRaisesRegex(JsonableError, "Invalid channel ID"):
access_stream_by_id(polonius, stream.id, require_content_access=True)
with self.assertRaisesRegex(JsonableError, "Invalid channel name 'new_private_stream'"):
access_stream_by_name(polonius, stream.name, require_content_access=True)
def test_stream_access_by_guest(self) -> None:
guest_user_profile = self.example_user("polonius")