Permalink
Browse files

auth: cleanup 2

  • Loading branch information...
interDist committed Jul 4, 2017
1 parent 8961637 commit 0eae092d2673349cce910c3f18b44366e94b9b89
View
@@ -31,7 +31,6 @@
auth_log = logging.getLogger('PasportaServo.auth')
dprint = auth_log.debug
class SupervisorAuthBackend(ModelBackend):
@@ -43,34 +42,34 @@ def get_user_supervisor_of(self, user_obj, obj=None, code=False):
Calculate responsibilities, globally or for an optional object.
The given object may be an iterable of countries, a single country, or a profile.
"""
dprint("\tcalculating countries")
auth_log.debug("\tcalculating countries")
cache_name = '_countrygroup_cache'
if not hasattr(user_obj, cache_name):
dprint("\t\t ... storing in cache %s ... ", cache_name)
auth_log.debug("\t\t ... storing in cache %s ... ", cache_name)
user_groups = user_obj.groups.all() if not user_obj.is_superuser else Group.objects.all()
user_countries = frozenset(Country(g.name) for g in user_groups if len(g.name) == 2)
setattr(user_obj, cache_name, user_countries)
supervised = getattr(user_obj, cache_name)
dprint("\tobject is %s", repr(obj))
auth_log.debug("\tobject is %s", repr(obj))
if obj is not None:
if isinstance(obj, Country):
countries = [obj]
dprint("\t\tGot a Country, %s", countries)
auth_log.debug("\t\tGot a Country, %s", countries)
elif isinstance(obj, Profile):
countries = obj.owned_places.filter(deleted=False).values_list('country', flat=True)
dprint("\t\tGot a Profile, %s", countries)
auth_log.debug("\t\tGot a Profile, %s", countries)
elif isinstance(obj, Place):
countries = [obj.country]
dprint("\t\tGot a Place, %s", countries)
auth_log.debug("\t\tGot a Place, %s", countries)
elif hasattr(obj, '__iter__') and not isinstance(obj, str):
countries = obj # assume an iterable of countries
dprint("\t\tGot an iterable, %s", countries)
auth_log.debug("\t\tGot an iterable, %s", countries)
else:
raise ImproperlyConfigured(
"Supervisor check needs either a profile, a country, or a list of countries."
)
dprint("\t\trequested: %s supervised: %s\n\t\tresult: %s",
set(countries), set(supervised), set(supervised) & set(countries))
auth_log.debug("\t\trequested: %s supervised: %s\n\t\tresult: %s",
set(countries), set(supervised), set(supervised) & set(countries))
supervised = set(supervised) & set(countries)
return supervised if code else [c.name for c in supervised]
@@ -87,15 +86,15 @@ def has_perm(self, user_obj, perm, obj=None):
Verify if this user has permission (to an optional object).
Short-circuits when resposibility is not satisfied.
"""
dprint("checking permission: %s [ %s ] for %s",
perm, user_obj, "%s %s" % ("object", repr(obj)) if obj else "any records")
auth_log.debug("checking permission: %s [ %s ] for %s",
perm, user_obj, "%s %s" % ("object", repr(obj)) if obj else "any records")
if perm == PERM_SUPERVISOR and obj is not None:
all_perms = self.get_all_permissions(user_obj, obj)
allowed = any(self._perm_sv_particular_re.match(p) for p in all_perms)
else:
allowed = super().has_perm(user_obj, perm, obj)
if perm == PERM_SUPERVISOR and not allowed:
dprint("permission to supervise not granted")
auth_log.debug("permission to supervise not granted")
raise PermissionDenied
return allowed
@@ -111,25 +110,25 @@ def get_group_permissions(self, user_obj, obj=None):
If an object is passed in, only permissions matching this object are returned.
"""
perms = super().get_group_permissions(user_obj, obj)
dprint("\tUser's built in perms: %s", perms)
auth_log.debug("\tUser's built in perms: %s", perms)
groups = set(self.get_user_supervisor_of(user_obj, code=True))
if any(groups):
dprint("\tUser's groups: %s", groups)
auth_log.debug("\tUser's groups: %s", groups)
if obj is None:
perms.update([PERM_SUPERVISOR])
cache_name = '_countrygroup_perm_cache'
if not hasattr(user_obj, cache_name):
dprint("\t\t ... storing in cache %s ... ", cache_name)
auth_log.debug("\t\t ... storing in cache %s ... ", cache_name)
setattr(user_obj, cache_name, frozenset("%s.%s" % (PERM_SUPERVISOR, g) for g in groups))
dprint("\tUser's group perms: %s", set(getattr(user_obj, cache_name)))
auth_log.debug("\tUser's group perms: %s", set(getattr(user_obj, cache_name)))
if obj is None:
perms.update(getattr(user_obj, cache_name))
else:
groups_for_obj = set(self.get_user_supervisor_of(user_obj, obj, code=True))
perms_for_obj = set("%s.%s" % (PERM_SUPERVISOR, g) for g in groups_for_obj)
dprint("\tUser's perms for object: %s", perms_for_obj)
auth_log.debug("\tUser's perms for object: %s", perms_for_obj)
perms.update(getattr(user_obj, cache_name) & perms_for_obj)
dprint("\tUser's all perms: %s", perms)
auth_log.debug("\tUser's all perms: %s", perms)
return perms
@@ -215,8 +214,9 @@ def _auth_verify(self, object, context_omitted=False):
if settings.DEBUG:
view_name = camel_case_split(self.__class__.__name__)
raise PermissionDenied(
"Not allowed to {0} this {1}.".format(view_name[-2].lower(), " ".join(view_name[0:-2]).lower()),
self
"Not allowed to {action} this {obj}.".format(
action=view_name[-2].lower(), obj=" ".join(view_name[0:-2]).lower()
), self
)
elif self.display_permission_denied and self.request.user.has_perm(PERM_SUPERVISOR):
raise PermissionDenied(self.get_permission_denied_message(object, context_omitted), self)
View
@@ -219,7 +219,7 @@ def user_link(self, obj):
def supervisor(self, obj):
country_list = CustomGroupAdmin.CountryGroup.objects.filter(user__pk=obj.user.id if obj.user else -1)
if len(country_list):
if country_list:
return format_html(",  ".join(map(str, country_list)))
else:
return self.get_empty_value_display()
@@ -9,7 +9,7 @@
class Migration(migrations.Migration):
dependencies = [
('hosting', '0037_auto_20170408_1510'),
('hosting', '0037_add_countrygroup_proxy_model'),
]
operations = [
@@ -0,0 +1,16 @@
# -*- coding: utf-8 -*-
# Generated by Django 1.10.7 on 2017-06-13 13:41
from __future__ import unicode_literals
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('hosting', '0038_add_approver_proxy_model'),
('hosting', '0035_place_blocking_dates'),
]
operations = [
]
@@ -13,7 +13,7 @@
register = template.Library()
dprint = logging.getLogger('PasportaServo.auth').debug
auth_log = logging.getLogger('PasportaServo.auth')
def _convert_profile_to_user(profile_obj):
@@ -26,17 +26,17 @@ def _convert_profile_to_user(profile_obj):
@register.filter
def is_supervisor(user_or_profile):
user = _convert_profile_to_user(user_or_profile)
dprint("* checking if supervising... [ %s %s]",
user, "<~ '%s' " % user_or_profile if user != user_or_profile else "")
auth_log.debug("* checking if supervising... [ %s %s]",
user, "<~ '%s' " % user_or_profile if user != user_or_profile else "")
return user.has_perm(PERM_SUPERVISOR)
@register.filter
def is_supervisor_of(user_or_profile, profile_or_countries):
user = _convert_profile_to_user(user_or_profile)
dprint("* checking if object is supervised... [ %s %s] [ %s ]",
user, "<~ '%s' " % user_or_profile if user != user_or_profile else "",
repr(profile_or_countries))
auth_log.debug("* checking if object is supervised... [ %s %s] [ %s ]",
user, "<~ '%s' " % user_or_profile if user != user_or_profile else "",
repr(profile_or_countries))
if isinstance(profile_or_countries, int):
try:
profile_or_countries = Profile.objects.get(pk=profile_or_countries)
@@ -62,8 +62,8 @@ def is_supervisor_of(user_or_profile, profile_or_countries):
@register.filter
def supervisor_of(user_or_profile):
user = _convert_profile_to_user(user_or_profile)
dprint("* searching supervised objects... [ %s %s]",
user, "<~ '%s' " % user_or_profile if user != user_or_profile else "")
auth_log.debug("* searching supervised objects... [ %s %s]",
user, "<~ '%s' " % user_or_profile if user != user_or_profile else "")
for backend in auth.get_backends():
try:
return sorted(backend.get_user_supervisor_of(user))
View
@@ -52,6 +52,14 @@ def generate_stats(self, request, response):
def custom_permission_denied_view(request, exception, template_name=ERROR_403_TEMPLATE_NAME):
"""
The Permission Denied view normally lacks information about the view that triggered the
exception, unless this information was provided in the exception object manually (as the
second parameter). This custom view attempts to include the relevant information if it
is available.
It is used, among others, by the Auth mixin to provide data about the offending view to
the Debug toolbar.
"""
response = permission_denied(request, exception.args[0] if exception.args else exception, template_name)
try:
response.context_data = getattr(response, 'context_data', {})

0 comments on commit 0eae092

Please sign in to comment.