Mercurial > libervia-backend

changeset 3499:a83a04b7394b

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
memory: new `getFilesOwner` method: this method is used by `getFiles` and `setFile` and fill `owner` when it's missing to avoid getting accidentally files from other users. `owner` is not filled in some case (no client, `file_id` or `parent` specified) and an exception is raised if a component has neither `owner` nor `peer_jid` specified.
author Goffi <goffi@goffi.org>
date Fri, 16 Apr 2021 18:32:37 +0200
parents d78b5eae912a
children 73b8a8d938be
files sat/memory/memory.py
diffstat 1 files changed, 36 insertions(+), 6 deletions(-) [+]
line wrap: on
line diff
--- a/sat/memory/memory.py	Fri Apr 16 18:32:34 2021 +0200
+++ b/sat/memory/memory.py	Fri Apr 16 18:32:37 2021 +0200
@@ -1451,6 +1451,37 @@
             ),
         )
 
+    def getFilesOwner(
+            self,
+            client,
+            owner: Optional[jid.JID],
+            peer_jid: Optional[jid.JID],
+            file_id: Optional[str],
+            parent: Optional[str]
+    ) -> jid.JID:
+        """Get owner to use for a file operation
+
+        if owner is not explicitely set, a suitable one will be used (client.jid for
+        clients, peer_jid for components).
+        @raise exception.InternalError: we are one a component, and neither owner nor
+            peer_jid are set
+        """
+        if owner is not None:
+            return owner.userhostJID()
+        if client is None:
+            # client may be None when looking for file with public_id
+            return None
+        if file_id or parent:
+            # owner has already been filtered on parent file
+            return None
+        if not client.is_component:
+            return client.jid.userhostJID()
+        if peer_jid is None:
+            raise exceptions.InternalError(
+                "Owner must be set for component if peer_jid is None"
+            )
+        return peer_jid.userhostJID()
+
     @defer.inlineCallbacks
     def getFiles(
         self, client, peer_jid, file_id=None, version=None, parent=None, path=None,
@@ -1485,7 +1516,8 @@
         @param unique(bool): if True will remove duplicates
         @param perms_to_check(tuple[unicode],None): permission to check
             must be a tuple of C.ACCESS_PERM_* or None
-            if None, permission will no be checked (peer_jid must be None too in this case)
+            if None, permission will no be checked (peer_jid must be None too in this
+            case)
         other params are the same as for [setFile]
         @return (list[dict]): files corresponding to filters
         @raise exceptions.NotFound: parent directory not found (when path is specified)
@@ -1498,8 +1530,7 @@
                 "if you want to disable permission check, both peer_jid and "
                 "perms_to_check must be None"
             )
-        if owner is not None:
-            owner = owner.userhostJID()
+        owner = self.getFilesOwner(client, owner, peer_jid, file_id, parent)
         if path is not None:
             path = str(path)
             # permission are checked by _getParentDir
@@ -1600,7 +1631,7 @@
             will be encoded to json in database
         @param perms_to_check(tuple[unicode],None): permission to check
             must be a tuple of C.ACCESS_PERM_* or None
-            if None, permission will no be checked (peer_jid must be None too in this
+            if None, permission will not be checked (peer_jid must be None too in this
             case)
         @param profile(unicode): profile owning the file
         """
@@ -1630,8 +1661,7 @@
                 raise ValueError(
                     "version, file_hash, size and mime_type can't be set for a directory"
                 )
-        if owner is not None:
-            owner = owner.userhostJID()
+        owner = self.getFilesOwner(client, owner, peer_jid, file_id, parent)
 
         if path is not None:
             path = str(path)