forked from Github-Mirrors/canaille
refactor: factorized User.load_permissions
This commit is contained in:
parent
e02b54f327
commit
fe809161ff
4 changed files with 36 additions and 51 deletions
|
@ -54,6 +54,13 @@ class User(canaille.core.models.User, LDAPObject):
|
|||
)
|
||||
return cls.get(filter=filter, **kwargs)
|
||||
|
||||
def match_filter(self, filter):
|
||||
conn = Backend.get().connection
|
||||
filter_ = self.acl_filter_to_ldap_filter(filter)
|
||||
return not filter_ or (
|
||||
self.dn and conn.search_s(self.dn, ldap.SCOPE_SUBTREE, filter_)
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def acl_filter_to_ldap_filter(cls, filter_):
|
||||
if isinstance(filter_, dict):
|
||||
|
@ -171,21 +178,6 @@ class User(canaille.core.models.User, LDAPObject):
|
|||
|
||||
self.state[group_attr] = new_groups
|
||||
|
||||
def load_permissions(self):
|
||||
conn = Backend.get().connection
|
||||
self._permissions = set()
|
||||
self._readable_fields = set()
|
||||
self._writable_fields = set()
|
||||
|
||||
for details in current_app.config["CANAILLE"]["ACL"].values():
|
||||
filter_ = self.acl_filter_to_ldap_filter(details["FILTER"])
|
||||
if not filter_ or (
|
||||
self.dn and conn.search_s(self.dn, ldap.SCOPE_SUBTREE, filter_)
|
||||
):
|
||||
self._permissions |= set(details["PERMISSIONS"])
|
||||
self._readable_fields |= set(details["READ"])
|
||||
self._writable_fields |= set(details["WRITE"])
|
||||
|
||||
|
||||
class Group(canaille.core.models.Group, LDAPObject):
|
||||
attribute_map = {
|
||||
|
|
|
@ -3,8 +3,6 @@ import datetime
|
|||
import typing
|
||||
import uuid
|
||||
|
||||
from flask import current_app
|
||||
|
||||
import canaille.core.models
|
||||
import canaille.oidc.models
|
||||
from canaille.app import models
|
||||
|
@ -235,31 +233,6 @@ class MemoryModel(BackendModel):
|
|||
def identifier(self):
|
||||
return getattr(self, self.identifier_attribute)
|
||||
|
||||
|
||||
class User(canaille.core.models.User, MemoryModel):
|
||||
identifier_attribute = "user_name"
|
||||
model_attributes = {
|
||||
"groups": ("Group", "members"),
|
||||
}
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.load_permissions()
|
||||
|
||||
def reload(self):
|
||||
super().reload()
|
||||
self.load_permissions()
|
||||
|
||||
def load_permissions(self):
|
||||
self._permissions = set()
|
||||
self._readable_fields = set()
|
||||
self._writable_fields = set()
|
||||
for details in current_app.config["CANAILLE"]["ACL"].values():
|
||||
if self.match_filter(details["FILTER"]):
|
||||
self._permissions |= set(details["PERMISSIONS"])
|
||||
self._readable_fields |= set(details["READ"])
|
||||
self._writable_fields |= set(details["WRITE"])
|
||||
|
||||
def match_filter(self, filter):
|
||||
if filter is None:
|
||||
return True
|
||||
|
@ -280,6 +253,21 @@ class User(canaille.core.models.User, MemoryModel):
|
|||
|
||||
return any(self.match_filter(subfilter) for subfilter in filter)
|
||||
|
||||
|
||||
class User(canaille.core.models.User, MemoryModel):
|
||||
identifier_attribute = "user_name"
|
||||
model_attributes = {
|
||||
"groups": ("Group", "members"),
|
||||
}
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.load_permissions()
|
||||
|
||||
def reload(self):
|
||||
super().reload()
|
||||
self.load_permissions()
|
||||
|
||||
@classmethod
|
||||
def get_from_login(cls, login=None, **kwargs):
|
||||
return User.get(user_name=login)
|
||||
|
|
|
@ -3,7 +3,6 @@ import typing
|
|||
import uuid
|
||||
from typing import List
|
||||
|
||||
from flask import current_app
|
||||
from sqlalchemy import Boolean
|
||||
from sqlalchemy import Column
|
||||
from sqlalchemy import ForeignKey
|
||||
|
@ -180,14 +179,7 @@ class User(canaille.core.models.User, Base, SqlAlchemyModel):
|
|||
|
||||
@reconstructor
|
||||
def load_permissions(self):
|
||||
self._permissions = set()
|
||||
self._readable_fields = set()
|
||||
self._writable_fields = set()
|
||||
for details in current_app.config["CANAILLE"]["ACL"].values():
|
||||
if self.match_filter(details["FILTER"]):
|
||||
self._permissions |= set(details["PERMISSIONS"])
|
||||
self._readable_fields |= set(details["READ"])
|
||||
self._writable_fields |= set(details["WRITE"])
|
||||
super().load_permissions()
|
||||
|
||||
def normalize_filter_value(self, attribute, value):
|
||||
# not super generic, but we can improve this when we have
|
||||
|
|
|
@ -2,6 +2,8 @@ import datetime
|
|||
from typing import List
|
||||
from typing import Optional
|
||||
|
||||
from flask import current_app
|
||||
|
||||
from canaille.backends.models import Model
|
||||
|
||||
|
||||
|
@ -280,6 +282,17 @@ class User(Model):
|
|||
datetime.timezone.utc
|
||||
)
|
||||
|
||||
def load_permissions(self):
|
||||
self._permissions = set()
|
||||
self._readable_fields = set()
|
||||
self._writable_fields = set()
|
||||
acls = current_app.config["CANAILLE"]["ACL"].values()
|
||||
for details in acls:
|
||||
if self.match_filter(details["FILTER"]):
|
||||
self._permissions |= set(details["PERMISSIONS"])
|
||||
self._readable_fields |= set(details["READ"])
|
||||
self._writable_fields |= set(details["WRITE"])
|
||||
|
||||
|
||||
class Group(Model):
|
||||
"""
|
||||
|
|
Loading…
Reference in a new issue