Merge branch 'issue-30-model-attributes' into 'main'

models attributes cardinality is closer to SCIM models

See merge request yaal/canaille!156

Author: azmeuk

CHANGES.rst

@@ -3,6 +3,11 @@ All notable changes to this project will be documented in this file.
 The format is based on `Keep a Changelog <https://keepachangelog.com/en/1.0.0/>`_,
 and this project adheres to `Semantic Versioning <https://semver.org/spec/v2.0.0.html>`_.
 
+Changed
+*******
+
+- Model attributes cardinality is closer to SCIM model. :pr:`155`
+
 [0.0.34] - 2023-10-02
 =====================
 

canaille/app/__init__.py

@@ -58,3 +58,11 @@ def validate_uri(value):
         re.IGNORECASE,
     )
     return re.match(regex, value) is not None
+
+
+class classproperty:
+    def __init__(self, f):
+        self.f = f
+
+    def __get__(self, obj, owner):
+        return self.f(owner)

canaille/backends/ldap/ldapobject.py

@@ -6,6 +6,7 @@
 from canaille.backends.models import Model
 
 from .backend import Backend
+from .utils import cardinalize_attribute
 from .utils import ldap_to_python
 from .utils import listify
 from .utils import python_to_ldap
@@ -109,7 +110,7 @@ class LDAPObject(Model, metaclass=LDAPObjectMetaclass):
     base = None
     root_dn = None
     rdn_attribute = None
-    attributes = None
+    attribute_map = None
     ldap_object_class = None
 
     def __init__(self, dn=None, **kwargs):
@@ -121,38 +122,38 @@ def __init__(self, dn=None, **kwargs):
             setattr(self, name, value)
 
     def __repr__(self):
-        reverse_attributes = {v: k for k, v in (self.attributes or {}).items()}
-        attribute_name = reverse_attributes.get(self.rdn_attribute, self.rdn_attribute)
+        attribute_name = self.ldap_attribute_to_python(self.rdn_attribute)
         return (
             f"<{self.__class__.__name__} {attribute_name}={self.rdn_value}>"
             if self.rdn_attribute
             else "<LDAPOBject>"
         )
 
     def __eq__(self, other):
+        ldap_attributes = self.may() + self.must()
         if not (
             isinstance(other, self.__class__)
             and self.may() == other.may()
             and self.must() == other.must()
             and all(
-                hasattr(self, attr) == hasattr(other, attr)
-                for attr in self.may() + self.must()
+                self.has_ldap_attribute(attr) == other.has_ldap_attribute(attr)
+                for attr in ldap_attributes
             )
         ):
             return False
 
         self_attributes = python_attrs_to_ldap(
             {
-                attr: getattr(self, attr)
-                for attr in self.may() + self.must()
-                if hasattr(self, attr)
+                attr: self.get_ldap_attribute(attr)
+                for attr in ldap_attributes
+                if self.has_ldap_attribute(attr)
             }
         )
         other_attributes = python_attrs_to_ldap(
             {
-                attr: getattr(other, attr)
-                for attr in self.may() + self.must()
-                if hasattr(self, attr)
+                attr: other.get_ldap_attribute(attr)
+                for attr in ldap_attributes
+                if other.has_ldap_attribute(attr)
             }
         )
         return self_attributes == other_attributes
@@ -161,17 +162,40 @@ def __hash__(self):
         return hash(self.id)
 
     def __getattr__(self, name):
-        name = self.attributes.get(name, name)
-
-        if name not in self.ldap_object_attributes():
+        if name not in self.attributes:
             return super().__getattribute__(name)
 
-        single_value = self.ldap_object_attributes()[name].single_value
+        ldap_name = self.python_attribute_to_ldap(name)
+
+        if ldap_name == "dn":
+            return self.dn_for(self.rdn_value)
+
+        python_single_value = "List" not in str(self.__annotations__[name])
+        ldap_value = self.get_ldap_attribute(ldap_name)
+        return cardinalize_attribute(python_single_value, ldap_value)
+
+    def __setattr__(self, name, value):
+        if name not in self.attributes:
+            super().__setattr__(name, value)
+
+        ldap_name = self.python_attribute_to_ldap(name)
+        self.set_ldap_attribute(ldap_name, value)
+
+    def __delattr__(self, name):
+        ldap_name = self.python_attribute_to_ldap(name)
+        self.delete_ldap_attribute(ldap_name)
+
+    def has_ldap_attribute(self, name):
+        return name in self.ldap_object_attributes() and (
+            name in self.changes or name in self.state
+        )
+
+    def get_ldap_attribute(self, name):
         if name in self.changes:
-            return self.changes[name][0] if single_value else self.changes[name]
+            return self.changes[name]
 
         if not self.state.get(name):
-            return None if single_value else []
+            return None
 
         # Lazy conversion from ldap format to python format
         if any(isinstance(value, bytes) for value in self.state[name]):
@@ -180,35 +204,23 @@ def __getattr__(self, name):
                 ldap_to_python(value, syntax) for value in self.state[name]
             ]
 
-        if single_value:
-            return self.state.get(name)[0]
-        else:
-            return [value for value in self.state.get(name) if value is not None]
+        return self.state.get(name)
 
-    def __setattr__(self, name, value):
-        if self.attributes:
-            name = self.attributes.get(name, name)
+    def set_ldap_attribute(self, name, value):
+        if name not in self.ldap_object_attributes():
+            return
 
-        if name in self.ldap_object_attributes():
-            value = listify(value)
-            self.changes[name] = value
+        value = listify(value)
+        self.changes[name] = value
 
-        else:
-            super().__setattr__(name, value)
-
-    def __delattr__(self, name):
-        name = self.attributes.get(name, name)
+    def delete_ldap_attribute(self, name):
         self.changes[name] = [None]
 
     @property
     def rdn_value(self):
-        value = getattr(self, self.rdn_attribute)
+        value = self.get_ldap_attribute(self.rdn_attribute)
         return (value[0] if isinstance(value, list) else value).strip()
 
-    @property
-    def dn(self):
-        return self.dn_for(self.rdn_value)
-
     @classmethod
     def dn_for(cls, rdn):
         return f"{cls.rdn_attribute}={ldap.dn.escape_dn_chars(rdn)},{cls.base},{cls.root_dn}"
@@ -317,7 +329,7 @@ def query(cls, id=None, filter=None, conn=None, **kwargs):
         arg_filter = ""
         kwargs = python_attrs_to_ldap(
             {
-                (cls.attributes or {}).get(name, name): values
+                cls.python_attribute_to_ldap(name): values
                 for name, values in kwargs.items()
             },
             encode=False,
@@ -350,7 +362,7 @@ def query(cls, id=None, filter=None, conn=None, **kwargs):
     def fuzzy(cls, query, attributes=None, **kwargs):
         query = ldap.filter.escape_filter_chars(query)
         attributes = attributes or cls.may() + cls.must()
-        attributes = [cls.attributes.get(name, name) for name in attributes]
+        attributes = [cls.python_attribute_to_ldap(name) for name in attributes]
         filter = (
             "(|" + "".join(f"({attribute}=*{query}*)" for attribute in attributes) + ")"
         )
@@ -380,6 +392,15 @@ def update_ldap_attributes(cls):
         cls._may = list(set(cls._may))
         cls._must = list(set(cls._must))
 
+    @classmethod
+    def ldap_attribute_to_python(cls, name):
+        reverse_attribute_map = {v: k for k, v in (cls.attribute_map or {}).items()}
+        return reverse_attribute_map.get(name, name)
+
+    @classmethod
+    def python_attribute_to_ldap(cls, name):
+        return cls.attribute_map.get(name, name) if cls.attribute_map else None
+
     def reload(self, conn=None):
         conn = conn or Backend.get().connection
         result = conn.search_s(self.id, ldap.SCOPE_SUBTREE, None, ["+", "*"])
@@ -389,7 +410,7 @@ def reload(self, conn=None):
     def save(self, conn=None):
         conn = conn or Backend.get().connection
 
-        setattr(self, "objectClass", self.ldap_object_class)
+        self.set_ldap_attribute("objectClass", self.ldap_object_class)
 
         # Object already exists in the LDAP database
         if self.exists:
@@ -429,10 +450,6 @@ def save(self, conn=None):
         self.state = {**self.state, **self.changes}
         self.changes = {}
 
-    def update(self, **kwargs):
-        for k, v in kwargs.items():
-            self.__setattr__(k, v)
-
     def delete(self, conn=None):
         conn = conn or Backend.get().connection
         conn.delete_s(self.id)

canaille/backends/ldap/models.py

@@ -15,7 +15,7 @@ class User(canaille.core.models.User, LDAPObject):
     DEFAULT_FILTER = "(|(uid={{ login }})(mail={{ login }}))"
     DEFAULT_RDN = "cn"
 
-    attributes = {
+    attribute_map = {
         "id": "dn",
         "user_name": "uid",
         "password": "userPassword",
@@ -66,7 +66,7 @@ def acl_filter_to_ldap_filter(cls, filter_):
                 filter_["groups"] = Group.dn_for(filter_["groups"])
 
             base = "".join(
-                f"({cls.attributes.get(key, key)}={value})"
+                f"({cls.python_attribute_to_ldap(key)}={value})"
                 for key, value in filter_.items()
             )
             return f"(&{base})" if len(filter_) > 1 else base
@@ -147,7 +147,7 @@ def reload(self):
         self.load_permissions()
 
     def save(self, *args, **kwargs):
-        group_attr = self.attributes.get("groups", "groups")
+        group_attr = self.python_attribute_to_ldap("groups")
         new_groups = self.changes.get(group_attr)
         if not new_groups:
             return super().save(*args, **kwargs)
@@ -194,7 +194,7 @@ class Group(canaille.core.models.Group, LDAPObject):
     DEFAULT_NAME_ATTRIBUTE = "cn"
     DEFAULT_USER_FILTER = "member={user.id}"
 
-    attributes = {
+    attribute_map = {
         "id": "dn",
         "display_name": "cn",
         "members": "member",
@@ -243,9 +243,8 @@ class Client(canaille.oidc.models.Client, LDAPObject):
         "software_version": "oauthSoftwareVersion",
     }
 
-    attributes = {
+    attribute_map = {
         "id": "dn",
-        "description": "description",
         "preconsent": "oauthPreconsent",
         # post_logout_redirect_uris is not yet supported by authlib
         "post_logout_redirect_uris": "oauthPostLogoutRedirectURI",
@@ -263,10 +262,9 @@ class AuthorizationCode(canaille.oidc.models.AuthorizationCode, LDAPObject):
     ldap_object_class = ["oauthAuthorizationCode"]
     base = "ou=authorizations,ou=oauth"
     rdn_attribute = "oauthAuthorizationCodeID"
-    attributes = {
+    attribute_map = {
         "id": "dn",
         "authorization_code_id": "oauthAuthorizationCodeID",
-        "description": "description",
         "code": "oauthCode",
         "client": "oauthClient",
         "subject": "oauthSubject",
@@ -290,11 +288,10 @@ class Token(canaille.oidc.models.Token, LDAPObject):
     ldap_object_class = ["oauthToken"]
     base = "ou=tokens,ou=oauth"
     rdn_attribute = "oauthTokenID"
-    attributes = {
+    attribute_map = {
         "id": "dn",
         "token_id": "oauthTokenID",
         "access_token": "oauthAccessToken",
-        "description": "description",
         "client": "oauthClient",
         "subject": "oauthSubject",
         "type": "oauthTokenType",
@@ -315,7 +312,7 @@ class Consent(canaille.oidc.models.Consent, LDAPObject):
     ldap_object_class = ["oauthConsent"]
     base = "ou=consents,ou=oauth"
     rdn_attribute = "cn"
-    attributes = {
+    attribute_map = {
         "id": "dn",
         "consent_id": "cn",
         "subject": "oauthSubject",

canaille/backends/ldap/utils.py

@@ -85,3 +85,13 @@ def python_to_ldap(value, syntax, encode=True):
 
 def listify(value):
     return value if isinstance(value, list) else [value]
+
+
+def cardinalize_attribute(python_unique, value):
+    if not value:
+        return None if python_unique else []
+
+    if python_unique:
+        return value[0]
+
+    return [v for v in value if v is not None]