class BaseGenericPermissionBackend(object):

    def get_all_permissions(self, user_obj, obj=None):
        if not user_obj.is_active or user_obj.is_anonymous:
            return set()
        disable_cache = getattr(settings, 'DISABLE_GENERIC_PERMISSION_CACHE', False)
        if user_obj.is_superuser:
            perms = getattr(self, '_all_permissions', None)
            if perms is None:
                perms = generate_perm_list(Permission.objects.all())
                if not disable_cache:
                    setattr(self, '_all_permissions', perms)
            return perms
        cache_field_name = get_cache_field_name(obj)
        # need this for testing purposes, do not enable this in production
        if not hasattr(user_obj, cache_field_name) or disable_cache:
            perms = get_all_user_permissions(user_obj, obj)
            setattr(user_obj, cache_field_name, perms)
        return getattr(user_obj, cache_field_name)

    def has_perm(self, user_obj, perm, obj=None):
        if not user_obj.is_active or user_obj.is_anonymous:
            return False
        disable_cache = getattr(settings, 'DISABLE_GENERIC_PERMISSION_CACHE', False)
        cache_field_name = get_cache_field_name(obj) + '_dict'
        if not hasattr(user_obj, cache_field_name) or disable_cache:
            setattr(user_obj, cache_field_name, {})
        perm_dict = getattr(user_obj, cache_field_name)

        # Check cached all permissions set.
        # It is not empty if has_perm() is called after get_all_permissions()
        if not disable_cache and not obj:
            all_perms_cache_field_name = get_cache_field_name()
            all_perms_set = getattr(user_obj, all_perms_cache_field_name, None)
            if all_perms_set is not None:
                perm_dict[perm] = perm in all_perms_set

        if perm not in perm_dict or disable_cache:
            check_result = check_single_permission(user_obj, perm, obj)
            perm_dict[perm] = check_result
        return perm_dict[perm]

    def has_module_perms(self, user_obj, app_label):
        if not user_obj.is_active:
            return False
        for perm in self.get_all_permissions(user_obj):
            if perm[:perm.index('.')] == app_label:
                return True
        return False

    def get_user(self, user_id):
        UserModel = get_user_model()
        try:
            return UserModel._default_manager.get(pk=user_id)
        except UserModel.DoesNotExist:
            return None class NoReasonSpecified(BaseException):
    def __init__(self, message='You should point the reason for this action'):
        super(NoReasonSpecified, self).__init__(message) class ImproperResponsibleInstancePassed(BaseException):
    def __init__(self, message='Responsible should be an instance of User model'):
        super(ImproperResponsibleInstancePassed, self).__init__(message) class GenericGlobalPerm(models.Model):
    """
        This model is for defining template-like permissions
        e.g. Every blog moderator could edit his blog
    """
    content_type = models.ForeignKey(
        ContentType, related_name='global_perms', verbose_name=_('content type'),
        on_delete=models.CASCADE, null=True
    )
    roles = models.IntegerField(verbose_name=_('roles'), default=DEFAULT_ROLE)
    permission = models.ForeignKey(
        to=Permission, verbose_name=_('permission'), on_delete=models.CASCADE
    )

    class Meta:
        verbose_name = _('global group permission')
        verbose_name_plural = _('global group permissions')
        unique_together = ('content_type', 'permission') class OwnerToPermission(AbstractOwnerToPermission):
    """
        This model is two-way generic many-to-many link from owner_object to owned object
        Multiple links from owner to object is supported i.e. different permissions
    """
    ADD_PERMISSION = ADD_PERMISSION_PERMISSION
    date_issued = models.DateTimeField(verbose_name=_('date issued'), auto_now_add=True)

    FIELDS_TO_IGNORE_FOR_HISTORY = ('id', 'date_issued', 'owner', 'content_object',)

    objects = OwnerToPermissionManager()

    class Meta:
        verbose_name = _('owner to permission link')
        verbose_name_plural = _('owner to permission links')
        index_together = (
            ['owner_content_type', 'owner_object_id'],
            ['content_type', 'object_id', 'permission'],
        )
        unique_together = (
            'content_type', 'object_id', 'owner_content_type', 'owner_object_id', 'permission'
        )
        permissions = (
            (ADD_PERMISSION_PERMISSION, _('add permission')),
            (VIEW_RESTRICTED_OBJECTS, _('view restricted objects')),
        )

    def __unicode__(self):
        if self.object_id is None:
            ctype = None
        else:
            ctype = self.content_type
        result = "{app}.{model}.{pk} ".format(
            app=self.owner_content_type.app_label,
            model=self.owner_content_type.model,
            pk=self.owner_object_id,
        )
        if self.object_id is not None:  # real object not global permission
            result += "- {app}.{model}.{pk}. ".format(
                app=ctype.app_label if ctype else '',
                model=ctype.model if ctype else '',
                pk=self.object_id or '',
            )
        if self.roles:
            result += "Roles {roles}. ".format(roles=self.roles)
        result += "Permission {perm}".format(perm=self.permission.codename)
        return result

    def values_to_save_for_history(self):
        # for related fields we want them to end on _id for more accurate perfomance
        fields_to_save = [
            f.name + '_id' if f.related_model else f.name for f in self._meta.get_fields()
            if f.name not in self.FIELDS_TO_IGNORE_FOR_HISTORY
        ]
        return {
            model_key: model_value for model_key, model_value in self.__dict__.items()
            if model_key in fields_to_save
        }

    @check_responsible_reason
    def delete(self, **kwargs):
        history_dict = self.values_to_save_for_history()
        history_dict.update({
            'reason': kwargs.get('reason'),
            'responsible': kwargs.get('responsible'),
            'change_type': HistoryOwnerToPermission.TYPE_REMOVE,
        })
        HistoryOwnerToPermission.objects.create(**history_dict)
        return super(OwnerToPermission, self).delete()

    @check_responsible_reason
    def save(self, *args, **kwargs):
        # This is made for cases when object_id or content_type are None,
        # as db engines do not take into account NULL fields when talking about uniqueness.
        model_fields = self.values_to_save_for_history()

        if OwnerToPermission.objects.filter(**model_fields).exists():
            raise IntegrityError('Duplicate with kwargs: {}'.format(model_fields))

        model_fields.update({
            'reason': kwargs.get('reason'),
            'change_type': HistoryOwnerToPermission.TYPE_ADD if not self.pk else HistoryOwnerToPermission.TYPE_CHANGE,
        })
        HistoryOwnerToPermission.objects.create(**model_fields)

        if self.owner_content_type == get_user_ctype():
            # Here is a bit of denormalization
            # User is a part of group of his own
            # This is done to drastically improve perm checking performance
            self_role = 1  # role == 1 in user model is assumed to mean user himself
            generic_user_to_group, created = GenericUserToGroup.objects.get_or_create(
                reason=kwargs.get('reason'),
                group_id=self.owner_object_id,
                group_content_type=self.owner_content_type,
                user_id=self.owner_object_id,
                defaults={
                    'responsible': kwargs.get('responsible'),
                    'roles': self_role,
                }
            )
            if not generic_user_to_group.roles & self_role:
                # If (for some reason) user is a part of group of his own without self role (1),
                # we ensure that self role is included
                generic_user_to_group.roles |= self_role
                generic_user_to_group.save(reason=SELF_GROUP_ROLE_UPDATED)
        try:
            del kwargs['reason']
        except KeyError:
            pass
        super(OwnerToPermission, self).save(*args, **kwargs) class HistoryOwnerToPermission(AbstractBaseHistory, AbstractOwnerToPermission):
    TYPE_ADD = 1
    TYPE_REMOVE = 2
    TYPE_CHANGE = 3

    CHANGE_TYPES = (
        (TYPE_ADD, 'add permission'),
        (TYPE_REMOVE, 'remove permission'),
        (TYPE_CHANGE, 'role changes'),
    )

    change_type = models.SmallIntegerField(
        choices=CHANGE_TYPES, null=False, blank=False,
    )

    class Meta:
        verbose_name = _('owner to permission history')
        verbose_name_plural = _('owner to permission histories')
        permissions = (
            (VIEW_OWNER_TO_PERM_HISTORY, _('view owner to permission history')),
        )

    def __str__(self):
        return (
            '{history_id} | initiated by {responsible}, '
            'action: {action_type} | {group_name} {group_id} for perm {permission}'
        ).format(
            history_id=self.id,
            responsible=self.responsible.username if self.responsible else '',
            action_type=self.change_type,
            group_name=self.owner_content_type,
            group_id=self.owner_object_id,
            permission=self.permission.codename if self.permission else '',
        )

    objects = models.Manager() class GenericUserToGroup(AbstractGenericUserToGroup):
    """
        This models is used for linking user to any possible group
        User can have only one link to group
        In case of multiple roles bitmasks is used
    """
    date_joined = models.DateTimeField(verbose_name=_('date joined'), auto_now_add=True)

    FIELDS_TO_IGNORE_FOR_HISTORY = ('id', 'date_joined', 'group',)

    objects = GenericUserToGroupManager()

    class Meta:
        verbose_name = _('user to group link')
        verbose_name_plural = _('user to group links')
        unique_together = ('group_content_type', 'group_id', 'user')

    def __unicode__(self):
        return "{app}.{model}.{group_id} - {username}".format(
            app=self.group_content_type.app_label,
            model=self.group_content_type.model,
            group_id=self.group_id,
            username=self.user.username
        )

    def values_to_save_for_history(self):
        # for related fields we want them to end on _id for more accurate perfomance
        fields_to_save = [
            f.name + '_id' if f.related_model else f.name for f in self._meta.get_fields()
            if f.name not in self.FIELDS_TO_IGNORE_FOR_HISTORY
        ]
        return {
            model_key: model_value for model_key, model_value in self.__dict__.items()
            if model_key in fields_to_save
        }

    @check_responsible_reason
    def delete(self, **kwargs):
        history_dict = self.values_to_save_for_history()
        history_dict.update({
            'reason': kwargs.get('reason'),
            'responsible': kwargs.get('responsible'),
            'change_type': HistoryGenericUserToGroup.TYPE_REMOVE,
        })
        HistoryGenericUserToGroup.objects.create(**history_dict)
        return super(GenericUserToGroup, self).delete()

    @check_responsible_reason
    def save(self, *args, **kwargs):
        model_fields = self.values_to_save_for_history()
        model_fields.update({
            'reason': kwargs.get('reason'),
            'change_type': HistoryGenericUserToGroup.TYPE_ADD if not self.pk else HistoryGenericUserToGroup.TYPE_CHANGE,
        })
        HistoryGenericUserToGroup.objects.create(**model_fields)
        try:
            del kwargs['reason']
        except KeyError:
            pass
        super(GenericUserToGroup, self).save(*args, **kwargs) class HistoryGenericUserToGroup(AbstractBaseHistory, AbstractGenericUserToGroup):
    TYPE_ADD = 1
    TYPE_REMOVE = 2
    TYPE_CHANGE = 3

    CHANGE_TYPES = (
        (TYPE_ADD, 'add user to group'),
        (TYPE_REMOVE, 'remove user from group'),
        (TYPE_CHANGE, 'role changes')
    )

    change_type = models.SmallIntegerField(
        choices=CHANGE_TYPES, null=False, blank=False,
    )

    class Meta:
        verbose_name = _('generic user to group history')
        verbose_name_plural = _('generic user to group histories')
        index_together = ('group_content_type', 'group_id', 'user')
        permissions = (
            (VIEW_GENERIC_GROUP_HISTORY, _('view generic group history')),
        )

    def __str__(self):
        return '{history_id} | initiated by {responsible}, action: {action_type} | {group_name} {group_id}'.\
            format(
                history_id=self.id,
                responsible=self.responsible.username if self.responsible else '',
                action_type=self.change_type,
                group_name=self.group_content_type,
                group_id=self.group_id,
            )

    objects = models.Manager() TEST_REASON = 'TEST REASON' def get_default_group_ctype():
    return ContentType.objects.get_by_natural_key(
        *settings.PROTECTOR_GENERIC_GROUP.lower().split('.')
    ) def get_user_ctype():
    return ContentType.objects.get_for_model(get_user_model()) def get_all_permission_owners(permission, include_superuser=False, obj=None):
    query = _get_permissions_query(obj)
    query.fields.append("gug.user_id AS id")
    query.conditions.append("op.permission_id = {perm_id!s}")
    query.params.update({'perm_id': permission.id})
    table_name = get_user_model()._meta.db_table
    condition = "{table_name!s}.id IN ({subquery!s})"
    if include_superuser:
        condition += " OR {table_name!s}.is_superuser"
    condition = condition.format(
        table_name=table_name, subquery=query.get_raw_query()
    )
    return get_user_model().objects.extra(where=[condition]) def get_permission_owners_of_type_for_object(permission, owner_content_type, content_object):
    OwnerToPermission = apps.get_model('protector', 'OwnerToPermission')
    qs = OwnerToPermission.objects.filter(
        content_type=ContentType.objects.get_for_model(content_object._meta.model),
        object_id=content_object.pk,
        owner_content_type=owner_content_type,
        permission=permission
    )
    return owner_content_type.model_class().objects.filter(
        id__in=qs.values_list('owner_object_id', flat=True)
    ) def filter_object_id_list(object_list, user_id, permission_id):
    # object_list list is a list of tuples (ctype_id, object_id)
    query = """
        SELECT ctype_id, object_id FROM ({ids_query}) AS ids
        WHERE EXISTS (SELECT op.id FROM {permission_owners} WHERE {filter_condition})
    """
    query = query.format(
        ids_query=generate_obj_list_query(object_list),
        permission_owners=get_permission_owners_query(),
        filter_condition=_generate_filter_condition(
            user_id, permission_id, 'ids.ctype_id', 'ids.object_id'
        )
    )
    cursor = connection.cursor()
    cursor.execute(query)
    return [(row[0], row[1]) for row in cursor.fetchall()] def is_user_having_perm_on_any_object(user, permission):
    if user.is_superuser:
        return True
    perm_id = get_permission_id_by_name(permission)
    if perm_id is None:
        return False
    query = """
        SELECT op.id FROM {permission_owners}
        WHERE op.permission_id = {permission_id} AND gug.user_id = {user_id}
        LIMIT 1;
    """
    query = query.format(
        permission_owners=get_permission_owners_query(),
        permission_id=perm_id,
        user_id=user.id
    )
    cursor = connection.cursor()
    cursor.execute(query)
    return len(cursor.fetchall()) > 0 def check_single_permission(user, permission, obj=None):
    if user.is_superuser:
        return True
    if obj is not None and obj.id is not None:
        ctype_id = ContentType.objects.get_for_model(obj).id
        obj_id = obj.id
    else:
        ctype_id = None
        obj_id = None
    perm_id = get_permission_id_by_name(permission)
    if perm_id is None:
        return False
    query = "SELECT op.id FROM {permission_owners} WHERE {filter_condition} LIMIT 1"
    query = query.format(
        permission_owners=get_permission_owners_query(),
        filter_condition=_generate_filter_condition(
            user.id, perm_id, ctype_id, obj_id
        )
    )
    cursor = connection.cursor()
    cursor.execute(query)
    return len(cursor.fetchall()) > 0
import mock
from django.test import TestCase
from django.db import IntegrityError
from django.contrib.auth.models import Permission
from django.contrib.auth import get_user_model
from django.contrib.contenttypes.models import ContentType
from django.test.utils import override_settings
from protector.backends import BaseGenericPermissionBackend
from protector.exceptions import NoReasonSpecified, ImproperResponsibleInstancePassed
from protector.models import (
    GenericGlobalPerm,
    OwnerToPermission,
    HistoryOwnerToPermission,
    GenericUserToGroup,
    HistoryGenericUserToGroup,
)
from protector.reserved_reasons import TEST_REASON
from protector.internals import get_default_group_ctype, get_user_ctype
from protector.helpers import (
    get_all_permission_owners, get_permission_owners_of_type_for_object,
    filter_object_id_list, is_user_having_perm_on_any_object, check_single_permission
)
        self.assertTrue(
            self.user2.has_perms([self.permission_key, self.permission2_key])
        )
        self.assertEqual(self.HistoryOwnerToPermission.objects.count(), 2)

    def test_has_module_perms(self):
        app_label = get_user_model()._meta.app_label
        self.assertFalse(
            self.user2.has_module_perms(app_label)
        )
        self.user2.permissions.add(self.permission, TEST_REASON)
        self.assertTrue(
            self.user2.has_module_perms(app_label)
        )
        self.assertEqual(self.HistoryOwnerToPermission.objects.count(), 1)

    def test_object_list_filter(self):
        group_ctype = ContentType.objects.get_for_model(self.TestGroup)
        obj_list = (
            (group_ctype.id, self.group.id),
            (group_ctype.id, self.group2.id)
        )
        self.assertEquals(
            len(filter_object_id_list(
                obj_list, self.user2.id, self.TestGroup.get_view_permission().id
            )), 0
        )
        self.user2.permissions.add(
            self.TestGroup.get_view_permission(), TEST_REASON, self.group2, responsible=self.responsible_user
        )
        self.assertEquals(
            filter_object_id_list(
                obj_list, self.user2.id, self.TestGroup.get_view_permission().id
            ), [(group_ctype.id, self.group2.id)]
        )
        self.assertEqual(self.HistoryOwnerToPermission.objects.count(), 1)

    def test_user_has_perm_on_any_object(self):
        self.assertFalse(
            is_user_having_perm_on_any_object(self.user, self.permission_key)
        )
        self.user.permissions.add(self.permission, TEST_REASON, self.group2, responsible=self.responsible_user)
        self.assertTrue(
            is_user_having_perm_on_any_object(self.user, self.permission_key)
        )
        self.assertFalse(
            is_user_having_perm_on_any_object(self.user, self.permission2_key)
        )
        self.user.permissions.add(self.permission, TEST_REASON, self.group2)
        self.user.permissions.remove(self.permission, TEST_REASON, self.group2)
        self.assertFalse(
            is_user_having_perm_on_any_object(self.user, self.permission_key)
        )
        self.user.permissions.add(self.permission, TEST_REASON, responsible=self.responsible_user)
        self.assertTrue(
            is_user_having_perm_on_any_object(self.user, self.permission_key)
        )
        self.assertEqual(self.HistoryOwnerToPermission.objects.count(), 3)

    def test_superuser_has_perm_on_any_object(self):
        self.assertFalse(
            is_user_having_perm_on_any_object(self.user, self.permission_key)
        )
        self.user.is_superuser = True
        self.user.save()
        self.assertTrue(
            is_user_having_perm_on_any_object(self.user, self.permission_key)
        )

    def test_single_permission_helper_global(self):
        self.assertFalse(
            check_single_permission(self.user, self.permission_key, self.group)
        )
        self.user.permissions.add(self.permission, TEST_REASON, responsible=self.responsible_user)
        self.assertTrue(
            check_single_permission(self.user, self.permission_key, self.group)
        )
        self.assertEqual(self.HistoryOwnerToPermission.objects.count(), 1)

    def test_single_permission_helper_on_object(self):
        self.assertFalse(
            check_single_permission(self.user, self.permission_key, self.group)
        )
        self.user.permissions.add(self.permission, TEST_REASON, obj=self.group, responsible=self.responsible_user)
        self.assertTrue(
            check_single_permission(self.user, self.permission_key, self.group)
        )
        self.assertEqual(self.HistoryOwnerToPermission.objects.count(), 1)

    def test_single_permission_global(self):
        self.assertFalse(
            check_single_permission(self.user, self.permission_key)
        )
        self.user.permissions.add(self.permission, TEST_REASON, obj=self.group)
        self.assertFalse(
            check_single_permission(self.user, self.permission_key)
        )
        self.user.permissions.add(self.permission, TEST_REASON)
        self.assertTrue(
            check_single_permission(self.user, self.permission_key)
        )
        self.assertEqual(self.HistoryOwnerToPermission.objects.count(), 2)

    def test_non_existing_permission(self):
        self.assertFalse(
            check_single_permission(self.user, 'not.exist', self.group)
        )
        self.assertFalse(
            check_single_permission(self.user, 'not.exist')
        )

    def test_no_owner_duplicates_allowed(self):
        create_test_dict = {
            'owner': self.user,
            'permission': self.permission,
            'reason': TEST_REASON,
        }
        try:
            # Intentionally forgetting to point reason
            OwnerToPermission.objects.create(owner=self.user, permission=self.permission)

        except NoReasonSpecified: