OP#2801 : add addressbooks share APIs

Author: tkeriven

app/api/v1/contact/ApiContact.py

@@ -7,12 +7,13 @@
 from flask.typing import ResponseReturnValue
 from flask_smorest import Blueprint
 
+from app.config.settings.DomainSettings import UserModuleSettings
 from app.interface.contact.InterfaceApiContactContact import InterfaceApiContactContact
 from app.module.contact.ContactConst import IMPORT_MAX_BYTES
 from app.module.contact.source.ContactSourceDb import LIST_SORTABLE_COLUMNS, SORTABLE_COLUMNS
 from app.utils.api.ApiBaseResponse import create_api_base_response
 from app.utils.api.paginate_sort_filter import collection_paginate, CustomPaginateResponse
-from app.utils.errors import ERROR_CONTACT_IMPORT_NO_FILE, ERROR_CONTACT_IMPORT_TOO_LARGE
+from app.utils.errors import ERROR_CONTACT_IMPORT_NO_FILE, ERROR_CONTACT_IMPORT_TOO_LARGE, ERROR_CONTACT_SHARING_DISABLED
 from app.utils.logger.logger import logger_api
 from .schemas.addressbook import (
     AddressBookCreateSchema,
@@ -22,6 +23,10 @@
     ContactImportQueryArgsSchema,
     ContactImportUploadSchema,
     ContactJobResponseSchema,
+    ContactSharePatchSchema,
+    ContactSharePutSchema,
+    ContactSharePostSchema,
+    ContactShareResponseSchema,
 )
 from .schemas.contact import (
     ContactCreateSchema,
@@ -67,7 +72,14 @@
 
 
 @blp.before_request
-def init_contact_config() -> None:  # pylint: disable=missing-function-docstring
+def init_contact_config() -> ResponseReturnValue | None:  # pylint: disable=missing-function-docstring
+    if request.path.endswith("/share"):
+        user_domain_settings: dict = g.user_domain_settings
+        user_module_settings: dict = user_domain_settings.get(UserModuleSettings.subparent, {})
+        if "contact" in user_module_settings.get("SOGO_D_FOLDER_DISABLE_SHARING", []):
+            logger_api.debug("Access denied for %s: contact sharing is disabled", request.path)
+            return create_api_base_response(None, ERROR_CONTACT_SHARING_DISABLED)
+
     g.inter = InterfaceApiContactContact(
         process_setting=g.process_settings,
         user_domain_settings=g.user_domain_settings,
@@ -122,6 +134,49 @@ def delete(self, key: str) -> ResponseReturnValue:
         return interface.delete_addressbook(key)
 
 
+@blp.route("/addressbooks/<string:key>/share")
+class ApiAddressBookShare(MethodView):
+    """API to manage address book sharing and user permissions."""
+
+    @blp.response(200, ContactShareResponseSchema, example=ContactShareResponseSchema.example())
+    def get(self, key: str) -> ResponseReturnValue:
+        """Get all user permissions for an address book."""
+        logger_api.debug("GET /addressbooks/%s/share user=%s", key, g.user.uid)
+        interface: InterfaceApiContactContact = g.inter
+        return interface.get_addressbook_share(key)
+
+    @blp.arguments(ContactSharePatchSchema(many=True), example=ContactSharePatchSchema.example())  # type: ignore [arg-type]
+    @blp.response(200, ContactShareResponseSchema, example=ContactShareResponseSchema.example())
+    def patch(self, body: list[dict], key: str) -> ResponseReturnValue:
+        """Partially update user permissions for an address book.
+
+        Only the users specified in the request body are modified.
+        Other existing permissions remain unchanged.
+        """
+        logger_api.debug("PATCH /addressbooks/%s/share user=%s body=%s", key, g.user.uid, body)
+        interface: InterfaceApiContactContact = g.inter
+        return interface.patch_addressbook_share(key, body)
+
+    @blp.arguments(ContactSharePutSchema(many=True), example=ContactSharePutSchema.example())  # type: ignore [arg-type]
+    @blp.response(200, ContactShareResponseSchema, example=ContactShareResponseSchema.example())
+    def put(self, body: list[dict], key: str) -> ResponseReturnValue:
+        """Replace all user permissions for an address book.
+
+        All existing permissions are replaced by the users specified in the request body.
+        """
+        logger_api.debug("PUT /addressbooks/%s/share user=%s body=%s", key, g.user.uid, body)
+        interface: InterfaceApiContactContact = g.inter
+        return interface.put_addressbook_share(key, body)
+
+    @blp.arguments(ContactSharePostSchema(many=True), example=ContactSharePostSchema.example())  # type: ignore [arg-type]
+    @blp.response(200, ContactShareResponseSchema, example=ContactShareResponseSchema.example())
+    def post(self, body: list[dict], key: str) -> ResponseReturnValue:
+        """Grant full permissions to one or several users."""
+        logger_api.debug("POST /addressbooks/%s/share user=%s body=%s", key, g.user.uid, body)
+        interface: InterfaceApiContactContact = g.inter
+        return interface.post_addressbook_share(key, body)
+
+
 @blp.route("/addressbooks/<string:key>/contacts")
 class ApiAddressBookContactList(MethodView):
     """API to list (paginated) and create contacts within one address book."""

app/api/v1/contact/schemas/addressbook.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
-from marshmallow import Schema, fields, validate
+from typing import Any
+
+from marshmallow import Schema, fields, validate, validates_schema, ValidationError
 
 from app.utils.api.ApiBaseResponse import ApiBaseResponse
 
@@ -83,3 +85,164 @@ class ContactImportUploadSchema(Schema):
         metadata={"type": "string", "format": "binary",
                   "description": "The JSON (.json), vCard (.vcf) or LDIF (.ldif) file to import."},
     )
+
+
+class ContactShareRightsSchema(Schema):
+    """Permission rights for an address book share."""
+
+    can_view = fields.Boolean(required=True, metadata={"description": "Can view contacts and lists", "example": True})
+    can_create_objects = fields.Boolean(required=True, metadata={"description": "Can create contacts and lists", "example": True})
+    can_edit_objects = fields.Boolean(required=True, metadata={"description": "Can edit contacts and lists", "example": True})
+    can_erase_objects = fields.Boolean(required=True, metadata={"description": "Can delete contacts and lists", "example": False})
+
+
+class ContactShareUserSchema(Schema):
+    """User permission entry in address book sharing.
+
+    ``c_email`` and ``uid`` are required unless ``user_class`` is ``"anyone"``, in which case
+    they are ignored (the share applies to any authenticated user, not a specific one).
+    """
+
+    c_email = fields.String(required=False, allow_none=True, metadata={"description": "User email address", "example": "jdoe@example.org"})
+    uid = fields.String(required=False, allow_none=True, metadata={"description": "User UID", "example": "jdoe"})
+    user_class = fields.String(
+        required=True,
+        validate=validate.OneOf(["user", "anyone"]),
+    )
+    rights = fields.Nested(ContactShareRightsSchema, required=True, metadata={"description": "Permission rights for this user"})
+
+    @validates_schema
+    def validate_user_identity(self, data: dict[str, Any], **kwargs: Any) -> None:  # pylint: disable=unused-argument
+        """Require c_email and uid unless user_class is 'anyone'."""
+        if data.get("user_class") == "anyone":
+            return
+        errors: dict[str, list[str]] = {}
+        if not data.get("c_email"):
+            errors["c_email"] = ["Missing data for required field."]
+        if not data.get("uid"):
+            errors["uid"] = ["Missing data for required field."]
+        if errors:
+            raise ValidationError(errors)
+
+
+class ContactSharePatchSchema(ContactShareUserSchema):
+    """Request body item for PATCH /addressbooks/{key}/share - partial update of user permissions.
+
+    The endpoint expects a JSON list of these objects (use with ``many=True``).
+    Only the users specified in the request are modified. Other existing permissions remain unchanged.
+    """
+
+    class Meta:
+        ordered = True
+
+    @staticmethod
+    def example() -> list[dict[str, Any]]:
+        """Example data for Swagger documentation."""
+        return [
+            {
+                "c_email": "jdoe@example.org",
+                "uid": "jdoe",
+                "user_class": "user",
+                "rights": {
+                    "can_view": True,
+                    "can_create_objects": True,
+                    "can_edit_objects": True,
+                    "can_erase_objects": False
+                }
+            }
+        ]
+
+
+class ContactSharePutSchema(ContactShareUserSchema):
+    """Request body item for PUT /addressbooks/{key}/share - replace all user permissions.
+
+    The endpoint expects a JSON list of these objects (use with ``many=True``).
+    All existing permissions are replaced by the users specified in the request.
+    """
+
+    class Meta:
+        ordered = True
+
+    @staticmethod
+    def example() -> list[dict[str, Any]]:
+        """Example data for Swagger documentation."""
+        return [
+            {
+                "c_email": "jdoe@example.org",
+                "uid": "jdoe",
+                "user_class": "user",
+                "rights": {
+                    "can_view": True,
+                    "can_create_objects": True,
+                    "can_edit_objects": True,
+                    "can_erase_objects": False
+                }
+            },
+            {
+                "c_email": "alice@example.org",
+                "uid": "alice",
+                "user_class": "user",
+                "rights": {
+                    "can_view": True,
+                    "can_create_objects": True,
+                    "can_edit_objects": True,
+                    "can_erase_objects": True
+                }
+            }
+        ]
+
+
+class ContactSharePostSchema(ContactShareUserSchema):
+    """Request body item for POST /addressbooks/{key}/share - grant full permissions to users.
+
+    The endpoint expects a JSON list of these objects (use with ``many=True``).
+    Grants full view/create/edit/erase rights to the specified users, regardless of the rights
+    carried in the request body.
+    """
+
+    class Meta:
+        ordered = True
+
+    @staticmethod
+    def example() -> list[dict[str, Any]]:
+        """Example data for Swagger documentation."""
+        return [
+            {
+                "c_email": "jdoe@example.org",
+                "uid": "jdoe",
+                "user_class": "user",
+                "rights": {
+                    "can_view": True,
+                    "can_create_objects": True,
+                    "can_edit_objects": True,
+                    "can_erase_objects": True
+                }
+            }
+        ]
+
+
+class ContactShareResponseSchema(ApiBaseResponse):
+    """Response schema for address book sharing endpoints. ``data`` is a plain list of users."""
+
+    data = fields.List(fields.Nested(ContactShareUserSchema), allow_none=True)
+
+    @staticmethod
+    def example() -> dict[str, Any]:
+        """Example full envelope for Swagger documentation."""
+        return {
+            "data": [
+                {
+                    "c_email": "jdoe@example.org",
+                    "uid": "jdoe",
+                    "user_class": "user",
+                    "rights": {
+                        "can_view": True,
+                        "can_create_objects": True,
+                        "can_edit_objects": True,
+                        "can_erase_objects": False
+                    }
+                }
+            ],
+            "error_code": "S000000",
+            "error_msg": "No Error"
+        }

app/factory/share/shareContact.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from app.factory.share.share import Share
+from app.module.contact.model.enums.ContactShareLevel import ContactShareLevel
+from app.utils import constants as cs
+from app.utils.strings import get_domain_from_mail
+
+if TYPE_CHECKING:
+    from app.factory.share.RepositoryAcl import AclEntry
+
+# Discriminant stored in sogo6_acl.type for address book shares.
+CONTACT_RESOURCE_TYPE: str = "addressbook"
+
+# Rights blob granted by POST /addressbooks/{key}/share (full access, per the endpoint's contract).
+FULL_MODIFY_RIGHTS: dict = {
+    "can_view": True,
+    "can_create_objects": True,
+    "can_edit_objects": True,
+    "can_erase_objects": True,
+}
+
+
+class ShareContact(Share):
+    """Sharing for address books, backed by sogo6_acl (type='addressbook').
+
+    The rights blob stored per (addressbook key, to_user) matches the API's
+    ContactShareRightsSchema: ``{"can_view": bool, "can_create_objects": bool,
+    "can_edit_objects": bool, "can_erase_objects": bool}``.
+
+    ``rights_needed`` passed to ``check_permissions`` is the name of the right to check
+    (e.g. "can_view", "can_edit_objects").
+    """
+
+    resource_type: str = CONTACT_RESOURCE_TYPE
+
+    def get_user_or_anyone(self, for_user_uid: str, owner_uid: str, on_key: str) -> AclEntry | None:
+        """Resolve the ACL entry granting for_user_uid access to on_key.
+
+        Priority: an entry addressed specifically to for_user_uid; failing that, the "anyone"
+        pseudo entry (``cs.ANYONE_TO_USER``, "<default>") - but only when for_user_uid and
+        owner_uid belong to the same mail domain, since an "anyone" share only ever means
+        "anyone in the owner's domain".
+        """
+        entry: AclEntry | None = self.get_entry(for_user_uid, on_key)
+        if entry is not None:
+            return entry
+        user_domain: str | None = get_domain_from_mail(for_user_uid)
+        owner_domain: str | None = get_domain_from_mail(owner_uid)
+        if not user_domain or user_domain != owner_domain:
+            return None
+        return self.get_entry(cs.ANYONE_TO_USER, on_key)
+
+    @staticmethod
+    def to_share_level(rights: dict) -> ContactShareLevel | None:
+        """Convert a stored rights blob into a ContactShareLevel, for ContactAclEngine.
+
+        Any write flag (create/edit/erase) grants MODIFY (which also satisfies a VIEW check);
+        otherwise can_view alone grants VIEW; a rights blob granting nothing at all denies.
+        """
+        if rights.get("can_create_objects") or rights.get("can_edit_objects") or rights.get("can_erase_objects"):
+            return ContactShareLevel.MODIFY
+        if rights.get("can_view"):
+            return ContactShareLevel.VIEW
+        return None
+
+    def _rights_satisfy(self, rights: dict, rights_needed: str) -> bool:
+        return bool(rights.get(rights_needed, False))

app/interface/calendar/InterfaceApiCalendarCalendar.py

@@ -748,7 +748,7 @@ def _grant_folder_subs_keys(self, target_uids: Iterable[str], key: str) -> None:
         """
         for target_uid in target_uids:
             if target_uid == cs.ANYONE_TO_USER:
-                continue
+                continue # The "anyone" pseudo-user has no real folders to update, so skip it.
             self._user_module.add_folder_key(target_uid, "CALENDAR", key, owner_key="SUBS")
 
     def _serialize_share_entries(self, entries: list[AclEntry]) -> list[dict[str, Any]]: