paulmataruso/zulip
1
0
Fork 0
mirror of https://github.com/zulip/zulip.git synced 2025-10-23 16:14:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

streams: Extract channel creation permissions check logic.

Browse Source 
Prep commit to add the channel creation API
endpoint.
Signed-off-by: apoorvapendse <apoorvavpendse@gmail.com>
This commit is contained in:
apoorvapendse
2025-06-08 16:28:56 +05:30
committed by Tim Abbott
parent 3fecbe41c5
commit e2f3b64b35
1 changed files with 38 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
zerver/lib/streams.py
View File

@@ -172,6 +172,35 @@ def send_stream_creation_event(
send_event_on_commit(realm, event, user_ids)
def check_channel_creation_permissions(
user_profile: UserProfile,
*,
is_default_stream: bool,
invite_only: bool,
is_web_public: bool,
message_retention_days: str | int | None,
) -> None:
if invite_only and not user_profile.can_create_private_streams():
raise JsonableError(_("Insufficient permission"))
if not invite_only and not user_profile.can_create_public_streams():
raise JsonableError(_("Insufficient permission"))
if is_default_stream and not user_profile.is_realm_admin:
raise JsonableError(_("Insufficient permission"))
if invite_only and is_default_stream:
raise JsonableError(_("A default channel cannot be private."))
if is_web_public:
if not user_profile.realm.web_public_streams_enabled():
raise JsonableError(_("Web-public channels are not enabled."))
if not user_profile.can_create_web_public_streams():
# We set can_create_web_public_channel_group to allow only organization
# owners to create web-public streams, because of their sensitive nature.
raise JsonableError(_("Insufficient permission"))
if message_retention_days is not None:
if not user_profile.is_realm_owner:
raise OrganizationOwnerRequiredError
user_profile.realm.ensure_not_on_limited_plan()
def get_stream_permission_default_group(
setting_name: str,
system_groups_name_dict: dict[str, NamedUserGroup],
@@ -1459,18 +1488,11 @@ def list_to_streams(
):
raise JsonableError(_("Insufficient permission"))
message_retention_days_not_none = False
web_public_stream_requested = False
for stream_dict in streams_raw:
stream_name = stream_dict["name"]
stream = existing_stream_map.get(stream_name.lower())
if stream is None:
if stream_dict.get("message_retention_days", None) is not None:
message_retention_days_not_none = True
missing_stream_dicts.append(stream_dict)
if autocreate and stream_dict["is_web_public"]:
web_public_stream_requested = True
else:
existing_streams.append(stream)
@@ -1479,18 +1501,6 @@ def list_to_streams(
# streams to exist already.
created_streams: list[Stream] = []
else:
# autocreate=True path starts here
for stream_dict in missing_stream_dicts:
invite_only = stream_dict.get("invite_only", False)
if invite_only and not user_profile.can_create_private_streams():
raise JsonableError(_("Insufficient permission"))
if not invite_only and not user_profile.can_create_public_streams():
raise JsonableError(_("Insufficient permission"))
if is_default_stream and not user_profile.is_realm_admin:
raise JsonableError(_("Insufficient permission"))
if invite_only and is_default_stream:
raise JsonableError(_("A default channel cannot be private."))
if not autocreate:
raise JsonableError(
_("Channel(s) ({channel_names}) do not exist").format(
@@ -1499,20 +1509,15 @@ def list_to_streams(
),
)
)
if web_public_stream_requested:
if not user_profile.realm.web_public_streams_enabled():
raise JsonableError(_("Web-public channels are not enabled."))
if not user_profile.can_create_web_public_streams():
# We set can_create_web_public_channel_group to allow only organization
# owners to create web-public streams, because of their sensitive nature.
raise JsonableError(_("Insufficient permission"))
if message_retention_days_not_none:
if not user_profile.is_realm_owner:
raise OrganizationOwnerRequiredError
user_profile.realm.ensure_not_on_limited_plan()
# autocreate=True path starts here
for stream_dict in missing_stream_dicts:
check_channel_creation_permissions(
user_profile,
is_default_stream=is_default_stream,
invite_only=stream_dict.get("invite_only", False),
is_web_public=stream_dict["is_web_public"],
message_retention_days=stream_dict.get("message_retention_days", None),
)
# We already filtered out existing streams, so dup_streams
# will normally be an empty list below, but we protect against somebody