Split acls models module into models, managers, classes and api
This commit is contained in:
Roberto Rosario
@@ -5,7 +5,7 @@ from django.utils.translation import ugettext_lazy as _
|
||||
from navigation.api import register_links, register_multi_item_links
|
||||
from project_setup.api import register_setup
|
||||
|
||||
from .models import AccessHolder, AccessObjectClass, ClassAccessHolder
|
||||
from .classes import AccessHolder, AccessObjectClass, ClassAccessHolder
|
||||
from .permissions import (ACLS_EDIT_ACL, ACLS_VIEW_ACL,
|
||||
ACLS_CLASS_EDIT_ACL, ACLS_CLASS_VIEW_ACL)
|
||||
|
||||
|
||||
13
apps/acls/api.py
Normal file
13
apps/acls/api.py
Normal file
@@ -0,0 +1,13 @@
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
|
||||
_class_permissions = {}
|
||||
|
||||
|
||||
def class_permissions(cls, permission_list):
|
||||
stored_permissions = _class_permissions.setdefault(cls, [])
|
||||
stored_permissions.extend(permission_list)
|
||||
|
||||
|
||||
def get_class_permissions_for(obj):
|
||||
content_type = ContentType.objects.get_for_model(obj)
|
||||
return _class_permissions.get(content_type.model_class(), [])
|
||||
136
apps/acls/classes.py
Normal file
136
apps/acls/classes.py
Normal file
@@ -0,0 +1,136 @@
|
||||
import logging
|
||||
import sys
|
||||
import types
|
||||
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.db.models.base import ModelBase
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_cache = {}
|
||||
|
||||
|
||||
class EncapsulatedObject(object):
|
||||
source_object_name = u'source_object'
|
||||
|
||||
@classmethod
|
||||
def add_to_class(cls, name, value):
|
||||
if hasattr(value, 'contribute_to_class'):
|
||||
value.contribute_to_class(cls, name)
|
||||
else:
|
||||
setattr(cls, name, value)
|
||||
|
||||
@classmethod
|
||||
def set_source_object_name(cls, new_name):
|
||||
cls.source_object_name = new_name
|
||||
|
||||
@classmethod
|
||||
def encapsulate(cls, source_object=None, app_label=None, model=None, pk=None):
|
||||
if source_object:
|
||||
content_type = ContentType.objects.get_for_model(source_object)
|
||||
elif app_label and model:
|
||||
try:
|
||||
content_type = ContentType.objects.get(app_label=app_label, model=model)
|
||||
source_object_model_class = content_type.model_class()
|
||||
if pk:
|
||||
source_object = content_type.get_object_for_this_type(pk=pk)
|
||||
else:
|
||||
source_object = source_object_model_class
|
||||
except ContentType.DoesNotExist:
|
||||
#cls.add_to_class('DoesNotExist', subclass_exception('DoesNotExist', (ObjectDoesNotExist,), cls.__name__))
|
||||
#raise cls.DoesNotExist("%s matching query does not exist." % ContentType._meta.object_name)
|
||||
raise ObjectDoesNotExist("%s matching query does not exist." % ContentType._meta.object_name)
|
||||
except source_object_model_class.DoesNotExist:
|
||||
#cls.add_to_class('DoesNotExist', subclass_exception('DoesNotExist', (ObjectDoesNotExist,), cls.__name__))
|
||||
#raise cls.DoesNotExist("%s matching query does not exist." % source_object_model_class._meta.object_name)
|
||||
raise ObjectDoesNotExist("%s matching query does not exist." % source_object_model_class._meta.object_name)
|
||||
|
||||
if hasattr(source_object, 'pk'):
|
||||
# Object
|
||||
object_key = '%s.%s.%s.%s' % (cls.__name__, content_type.app_label, content_type.model, source_object.pk)
|
||||
else:
|
||||
# Class
|
||||
object_key = '%s.%s.%s' % (cls.__name__, content_type.app_label, content_type.model)
|
||||
|
||||
try:
|
||||
return _cache[object_key]
|
||||
except KeyError:
|
||||
encapsulated_object = cls(source_object)
|
||||
_cache[object_key] = encapsulated_object
|
||||
return encapsulated_object
|
||||
|
||||
@classmethod
|
||||
def get(cls, gid):
|
||||
elements = gid.split('.')
|
||||
if len(elements) == 3:
|
||||
app_label, model, pk = elements[0], elements[1], elements[2]
|
||||
object_key = '%s.%s.%s.%s' % (cls.__name__, app_label, model, pk)
|
||||
elif len(elements) == 2:
|
||||
app_label, model = elements[0], elements[1]
|
||||
pk = None
|
||||
object_key = '%s.%s.%s' % (cls.__name__, app_label, model)
|
||||
|
||||
try:
|
||||
return _cache[object_key]
|
||||
except KeyError:
|
||||
if pk:
|
||||
return cls.encapsulate(app_label=app_label, model=model, pk=pk)
|
||||
else:
|
||||
return cls.encapsulate(app_label=app_label, model=model)
|
||||
|
||||
def __init__(self, source_object):
|
||||
self.content_type = ContentType.objects.get_for_model(source_object)
|
||||
self.ct_fullname = '%s.%s' % (self.content_type.app_label, self.content_type.name)
|
||||
|
||||
if isinstance(source_object, ModelBase):
|
||||
# Class
|
||||
self.gid = '%s.%s' % (self.content_type.app_label, self.content_type.model)
|
||||
else:
|
||||
# Object
|
||||
self.gid = '%s.%s.%s' % (self.content_type.app_label, self.content_type.model, source_object.pk)
|
||||
|
||||
setattr(self, self.__class__.source_object_name, source_object)
|
||||
|
||||
def __unicode__(self):
|
||||
if isinstance(self.source_object, ModelBase):
|
||||
return capfirst(unicode(self.source_object._meta.verbose_name_plural))
|
||||
|
||||
elif self.ct_fullname == 'auth.user':
|
||||
return u'%s %s' % (self.source_object._meta.verbose_name, self.source_object.get_full_name())
|
||||
else:
|
||||
return u'%s %s' % (self.source_object._meta.verbose_name, self.source_object)
|
||||
|
||||
def __repr__(self):
|
||||
return self.__unicode__()
|
||||
|
||||
@property
|
||||
def source_object(self):
|
||||
return getattr(self, self.__class__.source_object_name, None)
|
||||
|
||||
def get_class_permissions(self):
|
||||
return _class_permissions.get(self.content_type.model_class(), [])
|
||||
|
||||
|
||||
class AccessHolder(EncapsulatedObject):
|
||||
source_object_name = u'holder_object'
|
||||
|
||||
|
||||
class AccessObject(EncapsulatedObject):
|
||||
source_object_name = u'obj'
|
||||
|
||||
|
||||
class AccessObjectClass(EncapsulatedObject):
|
||||
source_object_name = u'cls'
|
||||
|
||||
|
||||
class ClassAccessHolder(EncapsulatedObject):
|
||||
source_object_name = u'class_holder'
|
||||
|
||||
|
||||
if sys.version_info < (2, 5):
|
||||
# Prior to Python 2.5, Exception was an old-style class
|
||||
def subclass_exception(name, parents, unused):
|
||||
return types.ClassType(name, parents, {})
|
||||
else:
|
||||
def subclass_exception(name, parents, module):
|
||||
return type(name, parents, {'__module__': module})
|
||||
@@ -7,7 +7,7 @@ from django.contrib.auth.models import User, Group
|
||||
from permissions.models import Permission, Role
|
||||
from common.utils import generate_choices_w_labels, encapsulate, get_object_name
|
||||
|
||||
from .models import AccessHolder
|
||||
from .classes import AccessHolder
|
||||
|
||||
|
||||
def _as_choice_list(holders):
|
||||
|
||||
236
apps/acls/managers.py
Normal file
236
apps/acls/managers.py
Normal file
@@ -0,0 +1,236 @@
|
||||
from __future__ import absolute_import
|
||||
|
||||
import logging
|
||||
|
||||
from django.db import models
|
||||
from django.utils.translation import ugettext_lazy as _
|
||||
from django.utils.translation import ugettext
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.contrib.auth.models import User
|
||||
from django.core.exceptions import PermissionDenied
|
||||
|
||||
from .classes import EncapsulatedObject, AccessHolder, ClassAccessHolder
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class AccessEntryManager(models.Manager):
|
||||
def source_object(self, obj):
|
||||
if isinstance(obj, EncapsulatedObject):
|
||||
return obj.source_object
|
||||
else:
|
||||
return obj
|
||||
|
||||
def grant(self, permission, actor, obj):
|
||||
'''
|
||||
Grant a permission (what), (to) an actor, (on) a specific object
|
||||
'''
|
||||
obj = self.source_object(obj)
|
||||
actor = self.source_object(actor)
|
||||
|
||||
access_entry, created = self.model.objects.get_or_create(
|
||||
permission=permission,
|
||||
holder_type=ContentType.objects.get_for_model(actor),
|
||||
holder_id=actor.pk,
|
||||
content_type=ContentType.objects.get_for_model(obj),
|
||||
object_id=obj.pk
|
||||
)
|
||||
return created
|
||||
|
||||
def revoke(self, permission, actor, obj):
|
||||
'''
|
||||
Revoke a permission (what), (from) an actor, (on) a specific object
|
||||
'''
|
||||
obj = self.source_object(obj)
|
||||
actor = self.source_object(actor)
|
||||
|
||||
try:
|
||||
access_entry = self.model.objects.get(
|
||||
permission=permission,
|
||||
holder_type=ContentType.objects.get_for_model(actor),
|
||||
holder_id=actor.pk,
|
||||
content_type=ContentType.objects.get_for_model(obj),
|
||||
object_id=obj.pk
|
||||
)
|
||||
access_entry.delete()
|
||||
return True
|
||||
except self.model.DoesNotExist:
|
||||
return False
|
||||
|
||||
def has_access(self, permission, actor, obj):
|
||||
obj = self.source_object(obj)
|
||||
actor = self.source_object(actor)
|
||||
|
||||
if isinstance(actor, User):
|
||||
if actor.is_superuser or actor.is_staff:
|
||||
return True
|
||||
|
||||
try:
|
||||
access_entry = self.model.objects.get(
|
||||
permission=permission.get_stored_permission(),
|
||||
holder_type=ContentType.objects.get_for_model(actor),
|
||||
holder_id=actor.pk,
|
||||
content_type=ContentType.objects.get_for_model(obj),
|
||||
object_id=obj.pk
|
||||
)
|
||||
return True
|
||||
except self.model.DoesNotExist:
|
||||
return False
|
||||
|
||||
def check_access(self, permission, actor, obj):
|
||||
obj = self.source_object(obj)
|
||||
actor = self.source_object(actor)
|
||||
|
||||
if self.has_access(permission, actor, obj):
|
||||
return True
|
||||
else:
|
||||
raise PermissionDenied(ugettext(u'Insufficient access.'))
|
||||
|
||||
def check_accesses(self, permission_list, actor, obj):
|
||||
obj = self.source_object(obj)
|
||||
actor = self.source_object(actor)
|
||||
for permission in permission_list:
|
||||
if self.has_access(permission, actor, obj):
|
||||
return True
|
||||
|
||||
raise PermissionDenied(ugettext(u'Insufficient access.'))
|
||||
|
||||
def get_allowed_class_objects(self, permission, actor, cls):
|
||||
actor_type = ContentType.objects.get_for_model(actor)
|
||||
content_type = ContentType.objects.get_for_model(cls)
|
||||
|
||||
return (obj.content_object for obj in self.model.objects.filter(holder_type=actor_type, holder_id=actor.pk, content_type=content_type, permission=permission.get_stored_permission))
|
||||
|
||||
def get_acl_url(self, obj):
|
||||
content_type = ContentType.objects.get_for_model(obj)
|
||||
return reverse('acl_list', args=[content_type.app_label, content_type.model, obj.pk])
|
||||
|
||||
def get_new_holder_url(self, obj):
|
||||
content_type = ContentType.objects.get_for_model(obj)
|
||||
return reverse('acl_new_holder_for', args=[content_type.app_label, content_type.model, obj.pk])
|
||||
|
||||
def get_holders_for(self, obj):
|
||||
content_type = ContentType.objects.get_for_model(obj)
|
||||
holder_list = []
|
||||
for access_entry in self.model.objects.filter(content_type=content_type, object_id=obj.pk):
|
||||
entry = AccessHolder.encapsulate(access_entry.holder_object)
|
||||
|
||||
if entry not in holder_list:
|
||||
holder_list.append(entry)
|
||||
|
||||
return holder_list
|
||||
|
||||
def get_holder_permissions_for(self, obj, actor):
|
||||
logger.debug('obj: %s' % obj)
|
||||
logger.debug('actor: %s' % actor)
|
||||
|
||||
if isinstance(actor, User):
|
||||
if actor.is_superuser or actor.is_staff:
|
||||
return Permission.objects.all()
|
||||
|
||||
actor_type = ContentType.objects.get_for_model(actor)
|
||||
content_type = ContentType.objects.get_for_model(obj)
|
||||
return (access.permission for access in self.model.objects.filter(content_type=content_type, object_id=obj.pk, holder_type=actor_type, holder_id=actor.pk))
|
||||
|
||||
def filter_objects_by_access(self, permission, actor, object_list, exception_on_empty=False):
|
||||
logger.debug('exception_on_empty: %s' % exception_on_empty)
|
||||
logger.debug('object_list: %s' % object_list)
|
||||
|
||||
if isinstance(actor, User):
|
||||
if actor.is_superuser or actor.is_staff:
|
||||
return object_list
|
||||
|
||||
try:
|
||||
if object_list.count() == 0:
|
||||
return object_list
|
||||
except TypeError:
|
||||
# object_list is not a queryset
|
||||
if len(object_list) == 0:
|
||||
return object_list
|
||||
|
||||
try:
|
||||
# Try to process as a QuerySet
|
||||
qs = object_list.filter(pk__in=[obj.pk for obj in self.get_allowed_class_objects(permission, actor, object_list[0])])
|
||||
logger.debug('qs: %s' % qs)
|
||||
|
||||
if qs.count() == 0 and exception_on_empty == True:
|
||||
raise PermissionDenied
|
||||
|
||||
return qs
|
||||
except AttributeError:
|
||||
# Fallback to a list filtered list
|
||||
obj_list = list(set(object_list) & set(self.get_allowed_class_objects(permission, actor, object_list[0])))
|
||||
logger.debug('obj_list: %s' % obj_list)
|
||||
if len(obj_list) == 0 and exception_on_empty == True:
|
||||
raise PermissionDenied
|
||||
|
||||
return obj_list
|
||||
|
||||
|
||||
class DefaultAccessEntryManager(models.Manager):
|
||||
def get_holders_for(self, cls):
|
||||
if isinstance(cls, EncapsulatedObject):
|
||||
cls = cls.source_object
|
||||
|
||||
content_type = ContentType.objects.get_for_model(cls)
|
||||
holder_list = []
|
||||
for access_entry in self.model.objects.filter(content_type=content_type):
|
||||
entry = ClassAccessHolder.encapsulate(access_entry.holder_object)
|
||||
|
||||
if entry not in holder_list:
|
||||
holder_list.append(entry)
|
||||
|
||||
return holder_list
|
||||
|
||||
def has_access(self, permission, actor, cls):
|
||||
if isinstance(actor, User):
|
||||
if actor.is_superuser or actor.is_staff:
|
||||
return True
|
||||
|
||||
try:
|
||||
access_entry = self.model.objects.get(
|
||||
permission=permission.get_stored_permission(),
|
||||
holder_type=ContentType.objects.get_for_model(actor),
|
||||
holder_id=actor.pk,
|
||||
content_type=ContentType.objects.get_for_model(cls),
|
||||
)
|
||||
return True
|
||||
except self.model.DoesNotExist:
|
||||
return False
|
||||
|
||||
def grant(self, permission, actor, cls):
|
||||
'''
|
||||
Grant a permission (what), (to) an actor, (on) a specific class
|
||||
'''
|
||||
access_entry, created = self.model.objects.get_or_create(
|
||||
permission=permission,
|
||||
holder_type=ContentType.objects.get_for_model(actor),
|
||||
holder_id=actor.pk,
|
||||
content_type=ContentType.objects.get_for_model(cls),
|
||||
)
|
||||
return created
|
||||
|
||||
def revoke(self, permission, actor, cls):
|
||||
'''
|
||||
Revoke a permission (what), (from) an actor, (on) a specific class
|
||||
'''
|
||||
try:
|
||||
access_entry = self.model.objects.get(
|
||||
permission=permission,
|
||||
holder_type=ContentType.objects.get_for_model(actor),
|
||||
holder_id=actor.pk,
|
||||
content_type=ContentType.objects.get_for_model(cls),
|
||||
)
|
||||
access_entry.delete()
|
||||
return True
|
||||
except self.model.DoesNotExist:
|
||||
return False
|
||||
|
||||
def get_holder_permissions_for(self, cls, actor):
|
||||
if isinstance(actor, User):
|
||||
if actor.is_superuser or actor.is_staff:
|
||||
return Permission.objects.all()
|
||||
|
||||
actor_type = ContentType.objects.get_for_model(actor)
|
||||
content_type = ContentType.objects.get_for_model(cls)
|
||||
return [access.permission for access in self.model.objects.filter(content_type=content_type, holder_type=actor_type, holder_id=actor.pk)]
|
||||
@@ -1,7 +1,5 @@
|
||||
from __future__ import absolute_import
|
||||
|
||||
import sys
|
||||
import types
|
||||
import logging
|
||||
|
||||
from django.db import models
|
||||
@@ -9,303 +7,24 @@ from django.utils.translation import ugettext_lazy as _
|
||||
from django.utils.translation import ugettext
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.contrib.contenttypes import generic
|
||||
from django.contrib.auth.models import User
|
||||
from django.core.exceptions import PermissionDenied
|
||||
from django.core.urlresolvers import reverse
|
||||
from django.core.exceptions import ObjectDoesNotExist
|
||||
from django.shortcuts import get_object_or_404
|
||||
from django.db.models.base import ModelBase
|
||||
from django.template.defaultfilters import capfirst
|
||||
|
||||
from permissions.models import StoredPermission
|
||||
|
||||
_cache = {}
|
||||
|
||||
_class_permissions = {}
|
||||
from .managers import AccessEntryManager, DefaultAccessEntryManager
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def class_permissions(cls, permission_list):
|
||||
stored_permissions = _class_permissions.setdefault(cls, [])
|
||||
stored_permissions.extend(permission_list)
|
||||
|
||||
|
||||
class EncapsulatedObject(object):
|
||||
source_object_name = u'source_object'
|
||||
|
||||
@classmethod
|
||||
def add_to_class(cls, name, value):
|
||||
if hasattr(value, 'contribute_to_class'):
|
||||
value.contribute_to_class(cls, name)
|
||||
else:
|
||||
setattr(cls, name, value)
|
||||
|
||||
@classmethod
|
||||
def set_source_object_name(cls, new_name):
|
||||
cls.source_object_name = new_name
|
||||
|
||||
@classmethod
|
||||
def encapsulate(cls, source_object=None, app_label=None, model=None, pk=None):
|
||||
if source_object:
|
||||
content_type = ContentType.objects.get_for_model(source_object)
|
||||
elif app_label and model:
|
||||
try:
|
||||
content_type = ContentType.objects.get(app_label=app_label, model=model)
|
||||
source_object_model_class = content_type.model_class()
|
||||
if pk:
|
||||
source_object = content_type.get_object_for_this_type(pk=pk)
|
||||
else:
|
||||
source_object = source_object_model_class
|
||||
except ContentType.DoesNotExist:
|
||||
#cls.add_to_class('DoesNotExist', subclass_exception('DoesNotExist', (ObjectDoesNotExist,), cls.__name__))
|
||||
#raise cls.DoesNotExist("%s matching query does not exist." % ContentType._meta.object_name)
|
||||
raise ObjectDoesNotExist("%s matching query does not exist." % ContentType._meta.object_name)
|
||||
except source_object_model_class.DoesNotExist:
|
||||
#cls.add_to_class('DoesNotExist', subclass_exception('DoesNotExist', (ObjectDoesNotExist,), cls.__name__))
|
||||
#raise cls.DoesNotExist("%s matching query does not exist." % source_object_model_class._meta.object_name)
|
||||
raise ObjectDoesNotExist("%s matching query does not exist." % source_object_model_class._meta.object_name)
|
||||
|
||||
if hasattr(source_object, 'pk'):
|
||||
# Object
|
||||
object_key = '%s.%s.%s.%s' % (cls.__name__, content_type.app_label, content_type.model, source_object.pk)
|
||||
else:
|
||||
# Class
|
||||
object_key = '%s.%s.%s' % (cls.__name__, content_type.app_label, content_type.model)
|
||||
|
||||
try:
|
||||
return _cache[object_key]
|
||||
except KeyError:
|
||||
encapsulated_object = cls(source_object)
|
||||
_cache[object_key] = encapsulated_object
|
||||
return encapsulated_object
|
||||
|
||||
@classmethod
|
||||
def get(cls, gid):
|
||||
elements = gid.split('.')
|
||||
if len(elements) == 3:
|
||||
app_label, model, pk = elements[0], elements[1], elements[2]
|
||||
object_key = '%s.%s.%s.%s' % (cls.__name__, app_label, model, pk)
|
||||
elif len(elements) == 2:
|
||||
app_label, model = elements[0], elements[1]
|
||||
pk = None
|
||||
object_key = '%s.%s.%s' % (cls.__name__, app_label, model)
|
||||
|
||||
try:
|
||||
return _cache[object_key]
|
||||
except KeyError:
|
||||
if pk:
|
||||
return cls.encapsulate(app_label=app_label, model=model, pk=pk)
|
||||
else:
|
||||
return cls.encapsulate(app_label=app_label, model=model)
|
||||
|
||||
def __init__(self, source_object):
|
||||
self.content_type = ContentType.objects.get_for_model(source_object)
|
||||
self.ct_fullname = '%s.%s' % (self.content_type.app_label, self.content_type.name)
|
||||
|
||||
if isinstance(source_object, ModelBase):
|
||||
# Class
|
||||
self.gid = '%s.%s' % (self.content_type.app_label, self.content_type.model)
|
||||
else:
|
||||
# Object
|
||||
self.gid = '%s.%s.%s' % (self.content_type.app_label, self.content_type.model, source_object.pk)
|
||||
|
||||
setattr(self, self.__class__.source_object_name, source_object)
|
||||
|
||||
def __unicode__(self):
|
||||
if isinstance(self.source_object, ModelBase):
|
||||
return capfirst(unicode(self.source_object._meta.verbose_name_plural))
|
||||
|
||||
elif self.ct_fullname == 'auth.user':
|
||||
return u'%s %s' % (self.source_object._meta.verbose_name, self.source_object.get_full_name())
|
||||
else:
|
||||
#label = unicode(obj)
|
||||
return u'%s %s' % (self.source_object._meta.verbose_name, self.source_object)
|
||||
|
||||
#return unicode(getattr(self, self.__class__.source_object_name, None))
|
||||
|
||||
def __repr__(self):
|
||||
return self.__unicode__()
|
||||
|
||||
@property
|
||||
def source_object(self):
|
||||
return getattr(self, self.__class__.source_object_name, None)
|
||||
|
||||
def get_class_permissions(self):
|
||||
return _class_permissions.get(self.content_type.model_class(), [])
|
||||
|
||||
|
||||
class AccessHolder(EncapsulatedObject):
|
||||
source_object_name = u'holder_object'
|
||||
|
||||
|
||||
class AccessObject(EncapsulatedObject):
|
||||
source_object_name = u'obj'
|
||||
|
||||
|
||||
class AccessObjectClass(EncapsulatedObject):
|
||||
source_object_name = u'cls'
|
||||
|
||||
|
||||
class ClassAccessHolder(EncapsulatedObject):
|
||||
source_object_name = u'class_holder'
|
||||
|
||||
|
||||
class AccessEntryManager(models.Manager):
|
||||
def source_object(self, obj):
|
||||
if isinstance(obj, EncapsulatedObject):
|
||||
return obj.source_object
|
||||
else:
|
||||
return obj
|
||||
|
||||
def grant(self, permission, actor, obj):
|
||||
'''
|
||||
Grant a permission (what), (to) a requester, (on) a specific object
|
||||
'''
|
||||
obj = self.source_object(obj)
|
||||
actor = self.source_object(actor)
|
||||
|
||||
access_entry, created = self.model.objects.get_or_create(
|
||||
permission=permission,
|
||||
holder_type=ContentType.objects.get_for_model(actor),
|
||||
holder_id=actor.pk,
|
||||
content_type=ContentType.objects.get_for_model(obj),
|
||||
object_id=obj.pk
|
||||
)
|
||||
return created
|
||||
|
||||
def revoke(self, permission, actor, obj):
|
||||
obj = self.source_object(obj)
|
||||
actor = self.source_object(actor)
|
||||
|
||||
try:
|
||||
access_entry = self.model.objects.get(
|
||||
permission=permission,
|
||||
holder_type=ContentType.objects.get_for_model(actor),
|
||||
holder_id=actor.pk,
|
||||
content_type=ContentType.objects.get_for_model(obj),
|
||||
object_id=obj.pk
|
||||
)
|
||||
access_entry.delete()
|
||||
return True
|
||||
except self.model.DoesNotExist:
|
||||
return False
|
||||
|
||||
def has_access(self, permission, actor, obj):
|
||||
obj = self.source_object(obj)
|
||||
actor = self.source_object(actor)
|
||||
|
||||
if isinstance(actor, User):
|
||||
if actor.is_superuser or actor.is_staff:
|
||||
return True
|
||||
|
||||
try:
|
||||
access_entry = self.model.objects.get(
|
||||
permission=permission.get_stored_permission(),
|
||||
holder_type=ContentType.objects.get_for_model(actor),
|
||||
holder_id=actor.pk,
|
||||
content_type=ContentType.objects.get_for_model(obj),
|
||||
object_id=obj.pk
|
||||
)
|
||||
return True
|
||||
except self.model.DoesNotExist:
|
||||
return False
|
||||
|
||||
def check_access(self, permission, actor, obj):
|
||||
obj = self.source_object(obj)
|
||||
actor = self.source_object(actor)
|
||||
|
||||
if self.has_access(permission, actor, obj):
|
||||
return True
|
||||
else:
|
||||
raise PermissionDenied(ugettext(u'Insufficient access.'))
|
||||
|
||||
def check_accesses(self, permission_list, actor, obj):
|
||||
obj = self.source_object(obj)
|
||||
actor = self.source_object(actor)
|
||||
for permission in permission_list:
|
||||
if self.has_access(permission, actor, obj):
|
||||
return True
|
||||
|
||||
raise PermissionDenied(ugettext(u'Insufficient access.'))
|
||||
|
||||
def get_allowed_class_objects(self, permission, actor, cls):
|
||||
actor_type = ContentType.objects.get_for_model(actor)
|
||||
content_type = ContentType.objects.get_for_model(cls)
|
||||
|
||||
return (obj.content_object for obj in self.model.objects.filter(holder_type=actor_type, holder_id=actor.pk, content_type=content_type, permission=permission.get_stored_permission))
|
||||
|
||||
def get_acl_url(self, obj):
|
||||
content_type = ContentType.objects.get_for_model(obj)
|
||||
return reverse('acl_list', args=[content_type.app_label, content_type.model, obj.pk])
|
||||
|
||||
def get_new_holder_url(self, obj):
|
||||
content_type = ContentType.objects.get_for_model(obj)
|
||||
return reverse('acl_new_holder_for', args=[content_type.app_label, content_type.model, obj.pk])
|
||||
|
||||
def get_holders_for(self, obj):
|
||||
content_type = ContentType.objects.get_for_model(obj)
|
||||
holder_list = []
|
||||
for access_entry in self.model.objects.filter(content_type=content_type, object_id=obj.pk):
|
||||
entry = AccessHolder.encapsulate(access_entry.holder_object)
|
||||
|
||||
if entry not in holder_list:
|
||||
holder_list.append(entry)
|
||||
|
||||
return holder_list
|
||||
|
||||
def get_holder_permissions_for(self, obj, actor):
|
||||
logger.debug('obj: %s' % obj)
|
||||
logger.debug('actor: %s' % actor)
|
||||
|
||||
if isinstance(actor, User):
|
||||
if actor.is_superuser or actor.is_staff:
|
||||
return Permission.objects.all()
|
||||
|
||||
actor_type = ContentType.objects.get_for_model(actor)
|
||||
content_type = ContentType.objects.get_for_model(obj)
|
||||
return (access.permission for access in self.model.objects.filter(content_type=content_type, object_id=obj.pk, holder_type=actor_type, holder_id=actor.pk))
|
||||
|
||||
def filter_objects_by_access(self, permission, actor, object_list, exception_on_empty=False):
|
||||
logger.debug('exception_on_empty: %s' % exception_on_empty)
|
||||
logger.debug('object_list: %s' % object_list)
|
||||
|
||||
if isinstance(actor, User):
|
||||
if actor.is_superuser or actor.is_staff:
|
||||
return object_list
|
||||
|
||||
try:
|
||||
if object_list.count() == 0:
|
||||
return object_list
|
||||
except TypeError:
|
||||
# object_list is not a queryset
|
||||
if len(object_list) == 0:
|
||||
return object_list
|
||||
|
||||
try:
|
||||
# Try to process as a QuerySet
|
||||
qs = object_list.filter(pk__in=[obj.pk for obj in self.get_allowed_class_objects(permission, actor, object_list[0])])
|
||||
logger.debug('qs: %s' % qs)
|
||||
|
||||
if qs.count() == 0 and exception_on_empty == True:
|
||||
raise PermissionDenied
|
||||
|
||||
return qs
|
||||
except AttributeError:
|
||||
# Fallback to a list filtered list
|
||||
obj_list = list(set(object_list) & set(self.get_allowed_class_objects(permission, actor, object_list[0])))
|
||||
logger.debug('obj_list: %s' % obj_list)
|
||||
if len(obj_list) == 0 and exception_on_empty == True:
|
||||
raise PermissionDenied
|
||||
|
||||
return obj_list
|
||||
|
||||
def get_class_permissions_for(self, obj):
|
||||
content_type = ContentType.objects.get_for_model(obj)
|
||||
return _class_permissions.get(content_type.model_class(), [])
|
||||
|
||||
|
||||
class AccessEntry(models.Model):
|
||||
'''
|
||||
Model that hold the permission, object, actor relationship
|
||||
'''
|
||||
|
||||
permission = models.ForeignKey(StoredPermission, verbose_name=_(u'permission'))
|
||||
|
||||
holder_type = models.ForeignKey(
|
||||
@@ -339,73 +58,12 @@ class AccessEntry(models.Model):
|
||||
return u'%s: %s' % (self.content_type, self.content_object)
|
||||
|
||||
|
||||
class DefaultAccessEntryManager(models.Manager):
|
||||
def get_holders_for(self, cls):
|
||||
if isinstance(cls, EncapsulatedObject):
|
||||
cls = cls.source_object
|
||||
|
||||
content_type = ContentType.objects.get_for_model(cls)
|
||||
holder_list = []
|
||||
for access_entry in self.model.objects.filter(content_type=content_type):
|
||||
entry = ClassAccessHolder.encapsulate(access_entry.holder_object)
|
||||
|
||||
if entry not in holder_list:
|
||||
holder_list.append(entry)
|
||||
|
||||
return holder_list
|
||||
|
||||
def has_access(self, permission, actor, cls):
|
||||
if isinstance(actor, User):
|
||||
if actor.is_superuser or actor.is_staff:
|
||||
return True
|
||||
|
||||
try:
|
||||
access_entry = self.model.objects.get(
|
||||
permission=permission.get_stored_permission(),
|
||||
holder_type=ContentType.objects.get_for_model(actor),
|
||||
holder_id=actor.pk,
|
||||
content_type=ContentType.objects.get_for_model(cls),
|
||||
)
|
||||
return True
|
||||
except self.model.DoesNotExist:
|
||||
return False
|
||||
|
||||
def grant(self, permission, actor, cls):
|
||||
'''
|
||||
Grant a permission (what), (to) a requester, (on) a specific class
|
||||
'''
|
||||
access_entry, created = self.model.objects.get_or_create(
|
||||
permission=permission,
|
||||
holder_type=ContentType.objects.get_for_model(actor),
|
||||
holder_id=actor.pk,
|
||||
content_type=ContentType.objects.get_for_model(cls),
|
||||
)
|
||||
return created
|
||||
|
||||
def revoke(self, permission, actor, cls):
|
||||
try:
|
||||
access_entry = self.model.objects.get(
|
||||
permission=permission,
|
||||
holder_type=ContentType.objects.get_for_model(actor),
|
||||
holder_id=actor.pk,
|
||||
content_type=ContentType.objects.get_for_model(cls),
|
||||
)
|
||||
access_entry.delete()
|
||||
return True
|
||||
except self.model.DoesNotExist:
|
||||
return False
|
||||
|
||||
def get_holder_permissions_for(self, cls, actor):
|
||||
if isinstance(actor, User):
|
||||
if actor.is_superuser or actor.is_staff:
|
||||
return Permission.objects.all()
|
||||
|
||||
actor_type = ContentType.objects.get_for_model(actor)
|
||||
content_type = ContentType.objects.get_for_model(cls)
|
||||
return [access.permission for access in self.model.objects.filter(content_type=content_type, holder_type=actor_type, holder_id=actor.pk)]
|
||||
|
||||
|
||||
class DefaultAccessEntry(models.Model):
|
||||
'''
|
||||
Model that holds the permission, class, actor relationship, that will
|
||||
be added upon the creation of an instance of said class
|
||||
'''
|
||||
|
||||
@classmethod
|
||||
def get_classes(cls):
|
||||
return [AccessObjectClass.encapsulate(cls) for cls in _class_permissions.keys()]
|
||||
@@ -436,13 +94,3 @@ class DefaultAccessEntry(models.Model):
|
||||
|
||||
def __unicode__(self):
|
||||
return u'%s: %s' % (self.content_type, self.content_object)
|
||||
|
||||
|
||||
if sys.version_info < (2, 5):
|
||||
# Prior to Python 2.5, Exception was an old-style class
|
||||
def subclass_exception(name, parents, unused):
|
||||
return types.ClassType(name, parents, {})
|
||||
else:
|
||||
def subclass_exception(name, parents, module):
|
||||
return type(name, parents, {'__module__': module})
|
||||
|
||||
|
||||
@@ -24,10 +24,12 @@ from common.widgets import two_state_template
|
||||
|
||||
from .permissions import (ACLS_EDIT_ACL, ACLS_VIEW_ACL,
|
||||
ACLS_CLASS_EDIT_ACL, ACLS_CLASS_VIEW_ACL)
|
||||
from .models import (AccessEntry, AccessObject, AccessHolder,
|
||||
DefaultAccessEntry, AccessObjectClass, ClassAccessHolder)
|
||||
from .models import AccessEntry, DefaultAccessEntry
|
||||
from .classes import (AccessHolder, AccessObject, AccessObjectClass,
|
||||
ClassAccessHolder)
|
||||
from .widgets import object_w_content_type_icon
|
||||
from .forms import HolderSelectionForm
|
||||
from .api import get_class_permissions_for
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -89,8 +91,7 @@ def acl_detail_for(request, actor, obj, navigation_object=None):
|
||||
except PermissionDenied:
|
||||
AccessEntry.objects.check_accesses([ACLS_VIEW_ACL, ACLS_EDIT_ACL], actor, obj)
|
||||
|
||||
#permission_list = list(obj.get_class_permissions())
|
||||
permission_list = AccessEntry.objects.get_class_permissions_for(obj)
|
||||
permission_list = get_class_permissions_for(obj)
|
||||
|
||||
#TODO : get all globally assigned permission, new function get_permissions_for_holder (roles aware)
|
||||
subtemplates_list = [
|
||||
|
||||
Reference in New Issue
Block a user