from django import forms from django.core.exceptions import PermissionDenied from django.http import Http404 from django.utils.translation import gettext_lazy as _ from ...acl import algebra from ...acl.decorators import return_boolean from ...acl.models import Role from ...admin.forms import YesNoSwitch from ...categories import PRIVATE_THREADS_ROOT_NAME from ...categories.models import Category from ..models import Thread __all__ = [ "allow_use_private_threads", "can_use_private_threads", "allow_see_private_thread", "can_see_private_thread", "allow_change_owner", "can_change_owner", "allow_add_participants", "can_add_participants", "allow_remove_participant", "can_remove_participant", "allow_add_participant", "can_add_participant", "allow_message_user", "can_message_user", ] class PermissionsForm(forms.Form): legend = _("Private threads") can_use_private_threads = YesNoSwitch(label=_("Can use private threads")) can_start_private_threads = YesNoSwitch(label=_("Can start private threads")) max_private_thread_participants = forms.IntegerField( label=_("Max number of users invited to private thread"), help_text=_("Enter 0 to don't limit number of participants."), initial=3, min_value=0, ) can_add_everyone_to_private_threads = YesNoSwitch( label=_("Can add everyone to threads"), help_text=_( "Allows user to add users that are blocking them to private threads." ), ) can_report_private_threads = YesNoSwitch( label=_("Can report private threads"), help_text=_( "Allows user to report private threads they are " "participating in, making them accessible to moderators." ), ) can_moderate_private_threads = YesNoSwitch( label=_("Can moderate private threads"), help_text=_( "Allows user to read, reply, edit and delete content " "in reported private threads." ), ) def change_permissions_form(role): if isinstance(role, Role) and role.special_role != "anonymous": return PermissionsForm def build_acl(acl, roles, key_name): new_acl = { "can_use_private_threads": 0, "can_start_private_threads": 0, "max_private_thread_participants": 3, "can_add_everyone_to_private_threads": 0, "can_report_private_threads": 0, "can_moderate_private_threads": 0, } new_acl.update(acl) algebra.sum_acls( new_acl, roles=roles, key=key_name, can_use_private_threads=algebra.greater, can_start_private_threads=algebra.greater, max_private_thread_participants=algebra.greater_or_zero, can_add_everyone_to_private_threads=algebra.greater, can_report_private_threads=algebra.greater, can_moderate_private_threads=algebra.greater, ) if not new_acl["can_use_private_threads"]: new_acl["can_start_private_threads"] = 0 return new_acl private_category = Category.objects.private_threads() new_acl["visible_categories"].append(private_category.pk) new_acl["browseable_categories"].append(private_category.pk) if new_acl["can_moderate_private_threads"]: new_acl["can_see_reports"].append(private_category.pk) category_acl = { "can_see": 1, "can_browse": 1, "can_see_all_threads": 1, "can_see_own_threads": 0, "can_start_threads": new_acl["can_start_private_threads"], "can_reply_threads": 1, "can_edit_threads": 1, "can_edit_posts": 1, "can_hide_own_threads": 0, "can_hide_own_posts": 1, "thread_edit_time": 0, "post_edit_time": 0, "can_hide_threads": 0, "can_hide_posts": 0, "can_protect_posts": 0, "can_move_posts": 0, "can_merge_posts": 0, "can_pin_threads": 0, "can_close_threads": 0, "can_move_threads": 0, "can_merge_threads": 0, "can_approve_content": 0, "can_report_content": new_acl["can_report_private_threads"], "can_see_reports": 0, "can_see_posts_likes": 0, "can_like_posts": 0, "can_hide_events": 0, } if new_acl["can_moderate_private_threads"]: category_acl.update( { "can_edit_threads": 2, "can_edit_posts": 2, "can_hide_threads": 2, "can_hide_posts": 2, "can_protect_posts": 1, "can_merge_posts": 1, "can_see_reports": 1, "can_close_threads": 1, "can_hide_events": 2, } ) new_acl["categories"][private_category.pk] = category_acl return new_acl def add_acl_to_thread(user_acl, thread): if thread.thread_type.root_name != PRIVATE_THREADS_ROOT_NAME: return if not hasattr(thread, "participant"): thread.participants_list = [] thread.participant = None thread.acl.update( { "can_start_poll": False, "can_change_owner": can_change_owner(user_acl, thread), "can_add_participants": can_add_participants(user_acl, thread), } ) def register_with(registry): registry.acl_annotator(Thread, add_acl_to_thread) def allow_use_private_threads(user_acl): if user_acl["is_anonymous"]: raise PermissionDenied(_("You have to sign in to use private threads.")) if not user_acl["can_use_private_threads"]: raise PermissionDenied(_("You can't use private threads.")) can_use_private_threads = return_boolean(allow_use_private_threads) def allow_see_private_thread(user_acl, target): if user_acl["can_moderate_private_threads"]: can_see_reported = target.has_reported_posts else: can_see_reported = False can_see_participating = user_acl["user_id"] in [ p.user_id for p in target.participants_list ] if not (can_see_participating or can_see_reported): raise Http404() can_see_private_thread = return_boolean(allow_see_private_thread) def allow_change_owner(user_acl, target): is_moderator = user_acl["can_moderate_private_threads"] is_owner = target.participant and target.participant.is_owner if not (is_owner or is_moderator): raise PermissionDenied( _("Only thread owner and moderators can change threads owners.") ) if not is_moderator and target.is_closed: raise PermissionDenied(_("Only moderators can change closed threads owners.")) can_change_owner = return_boolean(allow_change_owner) def allow_add_participants(user_acl, target): is_moderator = user_acl["can_moderate_private_threads"] if not is_moderator: if not target.participant or not target.participant.is_owner: raise PermissionDenied( _("You have to be thread owner to add new participants to it.") ) if target.is_closed: raise PermissionDenied( _("Only moderators can add participants to closed threads.") ) max_participants = user_acl["max_private_thread_participants"] current_participants = len(target.participants_list) - 1 if current_participants >= max_participants: raise PermissionDenied(_("You can't add any more new users to this thread.")) can_add_participants = return_boolean(allow_add_participants) def allow_remove_participant(user_acl, thread, target): if user_acl["can_moderate_private_threads"]: return if user_acl["user_id"] == target.id: return # we can always remove ourselves if thread.is_closed: raise PermissionDenied( _("Only moderators can remove participants from closed threads.") ) if not thread.participant or not thread.participant.is_owner: raise PermissionDenied( _("You have to be thread owner to remove participants from it.") ) can_remove_participant = return_boolean(allow_remove_participant) def allow_add_participant(user_acl, target, target_acl): message_format = {"user": target.username} if not can_use_private_threads(target_acl): raise PermissionDenied( _("%(user)s can't participate in private threads.") % message_format ) if user_acl["can_add_everyone_to_private_threads"]: return if user_acl["can_be_blocked"] and target.is_blocking(user_acl["user_id"]): raise PermissionDenied(_("%(user)s is blocking you.") % message_format) if target.can_be_messaged_by_nobody: raise PermissionDenied( _("%(user)s is not allowing invitations to private threads.") % message_format ) if target.can_be_messaged_by_followed and not target.is_following( user_acl["user_id"] ): message = _("%(user)s limits invitations to private threads to followed users.") raise PermissionDenied(message % message_format) can_add_participant = return_boolean(allow_add_participant) def allow_message_user(user_acl, target, target_acl): allow_use_private_threads(user_acl) allow_add_participant(user_acl, target, target_acl) can_message_user = return_boolean(allow_message_user)