Misago/misago/threads/permissions/privatethreads.py

from django import forms
from django.core.exceptions import PermissionDenied
from django.http import Http404
from django.utils.translation import ugettext_lazy as _

from misago.acl import algebra
from misago.acl.decorators import return_boolean
from misago.acl.models import Role
from misago.categories import PRIVATE_THREADS_ROOT_NAME
from misago.categories.models import Category
from misago.core.forms import YesNoSwitch
from misago.threads.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',
]
"""
Admin Permissions Form
"""


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 him 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
    else:
        return None


"""
ACL Builder
"""


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']:
        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, 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, thread),
        'can_add_participants': can_add_participants(user, thread),
    })


def register_with(registry):
    registry.acl_annotator(Thread, add_acl_to_thread)


"""
ACL tests
"""


def allow_use_private_threads(user):
    if user.is_anonymous:
        raise PermissionDenied(_("You have to sign in to use private threads."))
    if not user.acl_cache['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, target):
    if user.acl_cache['can_moderate_private_threads']:
        can_see_reported = target.has_reported_posts
    else:
        can_see_reported = False

    can_see_participating = user in [p.user 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, target):
    is_moderator = user.acl_cache['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, target):
    is_moderator = user.acl_cache['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_cache['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, thread, target):
    if user.acl_cache['can_moderate_private_threads']:
        return

    if user == target:
        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, target):
    message_format = {'user': target.username}

    if not can_use_private_threads(target):
        raise PermissionDenied(
            _("%(user)s can't participate in private threads.") % message_format
        )

    if user.acl_cache['can_add_everyone_to_private_threads']:
        return

    if user.acl_cache['can_be_blocked'] and target.is_blocking(user):
        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):
        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, target):
    allow_use_private_threads(user)
    allow_add_participant(user, target)


can_message_user = return_boolean(allow_message_user)