core (memory): getFileAffiliations and setFileAffiliations implementations:

those methods allow to change files permissions using affiliations similar to the ones
used with PubSub.

sat/memory/memory.py

@@ -22,7 +22,8 @@ import copy
 import shortuuid
 import mimetypes
 import time
-from typing import Optional, Tuple
+from functools import partial
+from typing import Optional, Tuple, Dict
 from pathlib import Path
 from uuid import uuid4
 from collections import namedtuple
@@ -1310,6 +1311,104 @@ class Memory(object):
                 parent = directory["id"]
         defer.returnValue((parent, []))
 
+    def getFileAffiliations(self, file_data: dict) -> Dict[jid.JID, str]:
+        """Convert file access to pubsub like affiliations"""
+        affiliations = {}
+        access_data = file_data['access']
+
+        read_data = access_data.get(C.ACCESS_PERM_READ, {})
+        if read_data.get('type') == C.ACCESS_TYPE_WHITELIST:
+            for entity_jid_s in read_data['jids']:
+                entity_jid = jid.JID(entity_jid_s)
+                affiliations[entity_jid] = 'member'
+
+        write_data = access_data.get(C.ACCESS_PERM_WRITE, {})
+        if write_data.get('type') == C.ACCESS_TYPE_WHITELIST:
+            for entity_jid_s in write_data['jids']:
+                entity_jid = jid.JID(entity_jid_s)
+                affiliations[entity_jid] = 'publisher'
+
+        owner = file_data.get('owner')
+        if owner:
+            affiliations[owner] = 'owner'
+
+        return affiliations
+
+    def _setFileAffiliationsUpdate(
+        self,
+        access: dict,
+        file_data: dict,
+        affiliations: Dict[jid.JID, str]
+    ) -> None:
+        read_data = access.setdefault(C.ACCESS_PERM_READ, {})
+        if read_data.get('type') != C.ACCESS_TYPE_WHITELIST:
+            read_data['type'] == C.ACCESS_TYPE_WHITELIST
+            if 'jids' not in read_data:
+                read_data['jids'] = []
+        read_whitelist = read_data['jids']
+        write_data = access.setdefault(C.ACCESS_PERM_WRITE, {})
+        if write_data.get('type') != C.ACCESS_TYPE_WHITELIST:
+            write_data['type'] == C.ACCESS_TYPE_WHITELIST
+            if 'jids' not in write_data:
+                write_data['jids'] = []
+        write_whitelist = write_data['jids']
+        for entity_jid, affiliation in affiliations.items():
+            entity_jid_s = entity_jid.full()
+            if affiliation == "none":
+                try:
+                    read_whitelist.remove(entity_jid_s)
+                except ValueError:
+                    log.warning(
+                        "removing affiliation from an entity without read permission: "
+                        f"{entity_jid}"
+                    )
+                try:
+                    write_whitelist.remove(entity_jid_s)
+                except ValueError:
+                    pass
+            elif affiliation == "publisher":
+                if entity_jid_s not in read_whitelist:
+                    read_whitelist.append(entity_jid_s)
+                if entity_jid_s not in write_whitelist:
+                    write_whitelist.append(entity_jid_s)
+            elif affiliation == "member":
+                if entity_jid_s not in read_whitelist:
+                    read_whitelist.append(entity_jid_s)
+                try:
+                    write_whitelist.remove(entity_jid_s)
+                except ValueError:
+                    pass
+            elif affiliation == "owner":
+                raise NotImplementedError('"owner" affiliation can\'t be set')
+            else:
+                raise ValueError(f"unknown affiliation: {affiliation!r}")
+
+    async def setFileAffiliations(
+        self,
+        client,
+        file_data: dict,
+        affiliations: Dict[jid.JID, str]
+    ) -> None:
+        """Apply pubsub like affiliation to file_data
+
+        Affiliations are converted to access types, then set in a whitelist.
+        Affiliation are mapped as follow:
+            - "owner" can't be set (for now)
+            - "publisher" gives read and write permissions
+            - "member" gives read permission only
+            - "none" removes both read and write permissions
+        """
+        file_id = file_data['id']
+        await self.fileUpdate(
+            file_id,
+            'access',
+            update_cb=partial(
+                self._setFileAffiliationsUpdate,
+                file_data=file_data,
+                affiliations=affiliations
+            ),
+        )
+
     @defer.inlineCallbacks
     def getFiles(
         self, client, peer_jid, file_id=None, version=None, parent=None, path=None,