nk20/apps/permission/models.py

244 lines
9.4 KiB
Python
Raw Normal View History

2020-03-07 12:12:17 +00:00
# Copyright (C) 2018-2020 by BDE ENS Paris-Saclay
# SPDX-License-Identifier: GPL-3.0-or-later
2020-02-09 16:35:15 +00:00
import functools
2019-09-18 12:26:42 +00:00
import json
2020-02-09 16:35:15 +00:00
import operator
2019-09-18 12:26:42 +00:00
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from django.db import models
2020-02-13 14:59:19 +00:00
from django.db.models import F, Q
2019-09-18 12:26:42 +00:00
from django.utils.translation import gettext_lazy as _
class InstancedPermission:
2020-02-09 16:35:15 +00:00
def __init__(self, model, query, type, field):
2019-09-18 12:26:42 +00:00
self.model = model
2020-02-09 16:35:15 +00:00
self.query = query
2019-09-18 12:26:42 +00:00
self.type = type
self.field = field
def applies(self, obj, permission_type, field_name=None):
2020-02-09 16:35:15 +00:00
"""
Returns True if the permission applies to
the field `field_name` object `obj`
"""
2020-03-18 14:49:52 +00:00
if ContentType.objects.get_for_model(obj) != self.model:
# The permission does not apply to the model
return False
2020-03-07 08:30:22 +00:00
if self.type == 'add':
if permission_type == self.type:
return self.query(obj)
2020-03-07 12:12:17 +00:00
if permission_type == self.type:
if self.field and field_name != self.field:
2019-09-18 12:26:42 +00:00
return False
2020-03-07 12:12:17 +00:00
return obj in self.model.model_class().objects.filter(self.query).all()
2020-02-09 16:35:15 +00:00
else:
return False
2019-09-18 12:26:42 +00:00
def __repr__(self):
if self.field:
2020-03-07 12:12:17 +00:00
return _("Can {type} {model}.{field} in {query}").format(type=self.type, model=self.model, field=self.field, query=self.query)
2019-09-18 12:26:42 +00:00
else:
2020-03-07 12:12:17 +00:00
return _("Can {type} {model} in {query}").format(type=self.type, model=self.model, query=self.query)
def __str__(self):
return self.__repr__()
2019-09-18 12:26:42 +00:00
class Permission(models.Model):
PERMISSION_TYPES = [
('add', 'add'),
('view', 'view'),
('change', 'change'),
('delete', 'delete')
2019-09-18 12:26:42 +00:00
]
model = models.ForeignKey(ContentType, on_delete=models.CASCADE, related_name='+')
2020-02-09 16:35:15 +00:00
# A json encoded Q object with the following grammar
# query -> [] | {} (the empty query representing all objects)
2020-03-07 12:12:17 +00:00
# query -> ["AND", query, …] AND multiple queries
# | ["OR", query, …] OR multiple queries
# | ["NOT", query] Opposite of query
2020-02-13 14:59:19 +00:00
# query -> {key: value, …} A list of fields and values of a Q object
# key -> string A field name
# value -> int | string | bool | null Literal values
# | [parameter] A parameter
2020-03-07 12:12:17 +00:00
# | {"F": oper} An F object
2020-02-13 14:59:19 +00:00
# oper -> [string] A parameter
2020-03-07 12:12:17 +00:00
# | ["ADD", oper, …] Sum multiple F objects or literal
# | ["SUB", oper, oper] Substract two F objects or literal
# | ["MUL", oper, …] Multiply F objects or literals
2020-02-13 14:59:19 +00:00
# | int | string | bool | null Literal values
2020-03-07 12:12:17 +00:00
# | ["F", string] A field
2020-02-09 16:35:15 +00:00
#
# Examples:
2020-03-07 12:12:17 +00:00
# Q(is_superuser=True) := {"is_superuser": true}
# ~Q(is_superuser=True) := ["NOT", {"is_superuser": true}]
query = models.TextField()
2019-09-18 12:26:42 +00:00
type = models.CharField(max_length=15, choices=PERMISSION_TYPES)
2019-09-18 12:26:42 +00:00
field = models.CharField(max_length=255, blank=True)
description = models.CharField(max_length=255, blank=True)
2019-09-18 12:26:42 +00:00
class Meta:
unique_together = ('model', 'query', 'type', 'field')
2019-09-18 12:26:42 +00:00
def clean(self):
self.query = json.dumps(json.loads(self.query))
if self.field and self.type not in {'view', 'change'}:
2019-09-18 12:26:42 +00:00
raise ValidationError(_("Specifying field applies only to view and change permission types."))
2020-03-07 12:12:17 +00:00
def save(self, **kwargs):
2019-09-18 12:26:42 +00:00
self.full_clean()
super().save()
2020-02-13 14:59:19 +00:00
@staticmethod
2020-03-07 12:12:17 +00:00
def compute_f(oper, **kwargs):
2020-02-13 14:59:19 +00:00
if isinstance(oper, list):
if oper[0] == 'ADD':
return functools.reduce(operator.add, [Permission.compute_f(oper, **kwargs) for oper in oper[1:]])
elif oper[0] == 'SUB':
return Permission.compute_f(oper[1], **kwargs) - Permission.compute_f(oper[2], **kwargs)
elif oper[0] == 'MUL':
return functools.reduce(operator.mul, [Permission.compute_f(oper, **kwargs) for oper in oper[1:]])
elif oper[0] == 'F':
return F(oper[1])
else:
field = kwargs[oper[0]]
for i in range(1, len(oper)):
field = getattr(field, oper[i])
return field
2020-02-13 14:59:19 +00:00
else:
return oper
@staticmethod
def compute_param(value, **kwargs):
if not isinstance(value, list):
return value
field = kwargs[value[0]]
for i in range(1, len(value)):
if isinstance(value[i], list):
field = getattr(field, value[i][0])
params = []
call_kwargs = {}
for j in range(1, len(value[i])):
param = Permission.compute_param(value[i][j], **kwargs)
if isinstance(param, dict):
for key in param:
val = Permission.compute_param(param[key], **kwargs)
call_kwargs[key] = val
else:
params.append(param)
field = field(*params, **call_kwargs)
else:
field = getattr(field, value[i])
return field
2020-02-13 14:59:19 +00:00
2020-03-07 12:12:17 +00:00
def _about(self, query, **kwargs):
2020-03-07 09:48:38 +00:00
if self.type == 'add':
2020-03-07 08:30:22 +00:00
# Handle add permission differently
return self._about_add(query, **kwargs)
if len(query) == 0:
# The query is either [] or {} and
2020-02-09 16:35:15 +00:00
# applies to all objects of the model
# to represent this we return a trivial request
return Q(pk=F("pk"))
if isinstance(query, list):
if query[0] == 'AND':
return functools.reduce(operator.and_, [self._about(query, **kwargs) for query in query[1:]])
elif query[0] == 'OR':
return functools.reduce(operator.or_, [self._about(query, **kwargs) for query in query[1:]])
elif query[0] == 'NOT':
return ~self._about(query[1], **kwargs)
elif isinstance(query, dict):
2020-02-09 16:35:15 +00:00
q_kwargs = {}
for key in query:
value = query[key]
2020-02-09 16:35:15 +00:00
if isinstance(value, list):
# It is a parameter we query its return value
q_kwargs[key] = Permission.compute_param(value, **kwargs)
2020-02-13 14:59:19 +00:00
elif isinstance(value, dict):
# It is an F object
q_kwargs[key] = Permission.compute_f(value['F'], **kwargs)
2020-02-09 16:35:15 +00:00
else:
q_kwargs[key] = value
return Q(**q_kwargs)
2019-09-18 12:26:42 +00:00
else:
2020-02-09 16:35:15 +00:00
# TODO: find a better way to crash here
raise Exception("query {} is wrong".format(self.query))
2019-09-18 12:26:42 +00:00
2020-03-07 12:12:17 +00:00
def _about_add(self, _query, **kwargs):
2020-03-07 08:30:22 +00:00
query = _query
if len(query) == 0:
return lambda _: True
if isinstance(query, list):
if query[0] == 'AND':
2020-03-07 12:12:17 +00:00
return lambda obj: functools.reduce(operator.and_, [self._about_add(q, **kwargs)(obj) for q in query[1:]])
2020-03-07 08:30:22 +00:00
elif query[0] == 'OR':
2020-03-07 12:12:17 +00:00
return lambda obj: functools.reduce(operator.or_, [self._about_add(q, **kwargs)(obj) for q in query[1:]])
2020-03-07 08:30:22 +00:00
elif query[0] == 'NOT':
return lambda obj: not self._about_add(query[1], **kwargs)(obj)
elif isinstance(query, dict):
q_kwargs = {}
for key in query:
value = query[key]
if isinstance(value, list):
# It is a parameter we query its primary key
q_kwargs[key] = Permission.compute_param(value, **kwargs)
2020-03-07 08:30:22 +00:00
elif isinstance(value, dict):
# It is an F object
2020-03-07 12:12:17 +00:00
q_kwargs[key] = Permission.compute_f(query['F'], **kwargs)
2020-03-07 08:30:22 +00:00
else:
q_kwargs[key] = value
def func(obj):
nonlocal q_kwargs
for arg in q_kwargs:
2020-03-18 14:49:52 +00:00
spl = arg.split('__')
value = obj
last = None
for s in spl:
if not hasattr(obj, s):
last = s
break
value = getattr(obj, s)
if last == "lte": # TODO Add more filters
if value > q_kwargs[arg]:
return False
elif value != q_kwargs[arg]:
2020-03-07 08:30:22 +00:00
return False
return True
return func
2019-09-18 12:26:42 +00:00
def about(self, **kwargs):
"""
Return an InstancedPermission with the parameters
replaced by their values and the query interpreted
"""
query = json.loads(self.query)
query = self._about(query, **kwargs)
2020-02-09 16:35:15 +00:00
return InstancedPermission(self.model, query, self.type, self.field)
2019-09-18 12:26:42 +00:00
def __str__(self):
if self.field:
return _("Can {type} {model}.{field} in {query}").format(type=self.type, model=self.model, field=self.field, query=self.query)
2019-09-18 12:26:42 +00:00
else:
return _("Can {type} {model} in {query}").format(type=self.type, model=self.model, query=self.query)
2019-09-18 12:26:42 +00:00
class UserPermission(models.Model):
user = models.ForeignKey('auth.User', on_delete=models.CASCADE)
permission = models.ForeignKey(Permission, on_delete=models.CASCADE)