lldap: automatic provision + system users + stalwart + whatever

machines/gerd/services/lldap/module/bootstrap/attributes.py

@@ -0,0 +1,161 @@
+from typing import Any
+
+import gql
+from gql.dsl import DSLQuery, DSLSchema, dsl_gql, DSLMutation
+from gql.transport.aiohttp import AIOHTTPTransport
+
+from .utils import to_camelcase, to_snakecase
+
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+class AttributeTypes:
+    STRING = "STRING"
+    INTEGER = "INTEGER"
+    JPEG_PHOTO = "JPEG_PHOTO"
+    DATE_TIME = "DATE_TIME"
+
+
+class AttributeSchema:
+    def __init__(self, raw_vals: dict[str, Any]):
+        self.name: str = raw_vals["name"]
+        self.attributeType: str = raw_vals["attributeType"]
+        self.isList: bool = bool(raw_vals["isList"])
+        self.isVisible: bool = bool(raw_vals["isVisible"])
+        self.isEditable: bool = bool(raw_vals["isEditable"])
+        self.isReadonly: bool = bool(raw_vals["isReadonly"])
+        self.isHardcoded: bool = bool(raw_vals["isHardcoded"])
+
+    def __repr__(self):
+        return f"<AttributeSchema name={self.name} attributeType={self.attributeType} isList={self.isList} isVisible={self.isVisible} isEditable={self.isEditable} isReadonly={self.isReadonly} isHardcoded={self.isHardcoded} />"
+
+
+class AttributeValue:
+    def __init__(self, raw_attribute: dict[str, Any]):
+        self.name: str = raw_attribute["name"]
+
+        tmpValue = raw_attribute.get("value", [])
+        if isinstance(tmpValue, str):
+            tmpValue = [tmpValue]
+
+        self.value: list[str] = tmpValue
+
+    def __repr__(self):
+        return f"<AttributeValue name={self.name} value={self.value} />"
+
+
+class LLDAPAttributes:
+    def __init__(
+        self,
+        client: gql.Client,
+        using_user_attributes: bool = False,
+        using_group_attributes: bool = False,
+    ):
+        self._client = client
+
+        self._using_user_attributes = using_user_attributes
+        self._using_group_attributes = using_group_attributes
+
+        if self._using_user_attributes and self._using_group_attributes:
+            raise Exception("can not both use user attributes and group attributes")
+
+        if not self._using_user_attributes and not self._using_group_attributes:
+            raise Exception("neither user attributes and group attributes specified")
+
+    def list_all(self) -> dict[str, AttributeSchema]:
+        ds = DSLSchema(self._client.schema)
+
+        if self._using_user_attributes:
+            querySchema = ds.Schema.userSchema
+            querySchemaStr = "userSchema"
+        else:
+            querySchema = ds.Schema.groupSchema
+            querySchemaStr = "groupSchema"
+
+        query = dsl_gql(
+            DSLQuery(
+                ds.Query.schema().select(
+                    querySchema.select(
+                        ds.AttributeList.attributes.select(
+                            ds.AttributeSchema.name,
+                            ds.AttributeSchema.attributeType,
+                            ds.AttributeSchema.isList,
+                            ds.AttributeSchema.isVisible,
+                            ds.AttributeSchema.isEditable,
+                            ds.AttributeSchema.isReadonly,
+                            ds.AttributeSchema.isHardcoded,
+                        )
+                    )
+                ),
+            ),
+        )
+
+        result = self._client.execute(query)
+
+        attrs: dict[str, AttributeSchema] = {}
+        for raw_attribute in result["schema"][querySchemaStr]["attributes"]:
+            attr = AttributeSchema(raw_attribute)
+            attrs[attr.name] = attr
+
+        return attrs
+
+    def create(
+        self,
+        name: str,
+        attributeType: str,
+        isList: bool,
+        isVisible: bool,
+        isEditable: bool,
+    ):
+        logger.debug(
+            f"adding attribute, name:'{name}' attributeType:'{attributeType}' isList:'{isList}' isVisible:'{isVisible}' isEditable:'{isEditable}'"
+        )
+
+        ds = DSLSchema(self._client.schema)
+
+        if self._using_user_attributes:
+            mutationSchema = ds.Mutation.addUserAttribute
+        else:
+            mutationSchema = ds.Mutation.addGroupAttribute
+
+        query = dsl_gql(
+            DSLMutation(
+                mutationSchema.args(
+                    name=name,
+                    attributeType=attributeType,
+                    isList=isList,
+                    isVisible=isVisible,
+                    isEditable=isEditable,
+                ).select(ds.Success.ok)
+            ),
+        )
+        self._client.execute(query)
+
+    def get(self, name: str) -> AttributeSchema | None:
+        attrs = self.list_all()
+        return attrs.get(name)
+
+    def update(self, name: str):
+        raise Exception("unable to update attribute")
+
+    def delete(self, name: str):
+        logger.debug(f"deleting attribute '{name}'")
+
+        ds = DSLSchema(self._client.schema)
+
+        if self._using_user_attributes:
+            mutationSchema = ds.Mutation.deleteUserAttribute
+        else:
+            mutationSchema = ds.Mutation.deleteGroupAttribute
+
+        query = dsl_gql(
+            DSLMutation(
+                mutationSchema.args(
+                    name=name,
+                ).select(ds.Success.ok)
+            ),
+        )
+        self._client.execute(query)