Mercurial > libervia-backend

comparison sat/memory/memory.py @ 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 7550ae9cfbac
children 30779935c0aa
comparison
equal deleted inserted replaced
3498:d78b5eae912a 3499:a83a04b7394b
1449 file_data=file_data, 1449 file_data=file_data,
1450 access_model=access_model 1450 access_model=access_model
1451 ), 1451 ),
1452 ) 1452 )
1453 1453
1454 def getFilesOwner(
1455 self,
1456 client,
1457 owner: Optional[jid.JID],
1458 peer_jid: Optional[jid.JID],
1459 file_id: Optional[str],
1460 parent: Optional[str]
1461 ) -> jid.JID:
1462 """Get owner to use for a file operation
1463
1464 if owner is not explicitely set, a suitable one will be used (client.jid for
1465 clients, peer_jid for components).
1466 @raise exception.InternalError: we are one a component, and neither owner nor
1467 peer_jid are set
1468 """
1469 if owner is not None:
1470 return owner.userhostJID()
1471 if client is None:
1472 # client may be None when looking for file with public_id
1473 return None
1474 if file_id or parent:
1475 # owner has already been filtered on parent file
1476 return None
1477 if not client.is_component:
1478 return client.jid.userhostJID()
1479 if peer_jid is None:
1480 raise exceptions.InternalError(
1481 "Owner must be set for component if peer_jid is None"
1482 )
1483 return peer_jid.userhostJID()
1484
1454 @defer.inlineCallbacks 1485 @defer.inlineCallbacks
1455 def getFiles( 1486 def getFiles(
1456 self, client, peer_jid, file_id=None, version=None, parent=None, path=None, 1487 self, client, peer_jid, file_id=None, version=None, parent=None, path=None,
1457 type_=None, file_hash=None, hash_algo=None, name=None, namespace=None, 1488 type_=None, file_hash=None, hash_algo=None, name=None, namespace=None,
1458 mime_type=None, public_id=None, owner=None, access=None, projection=None, 1489 mime_type=None, public_id=None, owner=None, access=None, projection=None,
1483 @param projection(list[unicode], None): name of columns to retrieve 1514 @param projection(list[unicode], None): name of columns to retrieve
1484 None to retrieve all 1515 None to retrieve all
1485 @param unique(bool): if True will remove duplicates 1516 @param unique(bool): if True will remove duplicates
1486 @param perms_to_check(tuple[unicode],None): permission to check 1517 @param perms_to_check(tuple[unicode],None): permission to check
1487 must be a tuple of C.ACCESS_PERM_* or None 1518 must be a tuple of C.ACCESS_PERM_* or None
1488 if None, permission will no be checked (peer_jid must be None too in this case) 1519 if None, permission will no be checked (peer_jid must be None too in this
1520 case)
1489 other params are the same as for [setFile] 1521 other params are the same as for [setFile]
1490 @return (list[dict]): files corresponding to filters 1522 @return (list[dict]): files corresponding to filters
1491 @raise exceptions.NotFound: parent directory not found (when path is specified) 1523 @raise exceptions.NotFound: parent directory not found (when path is specified)
1492 @raise exceptions.PermissionError: peer_jid can't use perms_to_check for one of 1524 @raise exceptions.PermissionError: peer_jid can't use perms_to_check for one of
1493 the file 1525 the file
1496 if peer_jid is None and perms_to_check or perms_to_check is None and peer_jid: 1528 if peer_jid is None and perms_to_check or perms_to_check is None and peer_jid:
1497 raise exceptions.InternalError( 1529 raise exceptions.InternalError(
1498 "if you want to disable permission check, both peer_jid and " 1530 "if you want to disable permission check, both peer_jid and "
1499 "perms_to_check must be None" 1531 "perms_to_check must be None"
1500 ) 1532 )
1501 if owner is not None: 1533 owner = self.getFilesOwner(client, owner, peer_jid, file_id, parent)
1502 owner = owner.userhostJID()
1503 if path is not None: 1534 if path is not None:
1504 path = str(path) 1535 path = str(path)
1505 # permission are checked by _getParentDir 1536 # permission are checked by _getParentDir
1506 parent, remaining_path_elts = yield self._getParentDir( 1537 parent, remaining_path_elts = yield self._getParentDir(
1507 client, path, parent, namespace, owner, peer_jid, perms_to_check 1538 client, path, parent, namespace, owner, peer_jid, perms_to_check
1598 will be encoded to json in database 1629 will be encoded to json in database
1599 @param extra(dict, None): serialisable dictionary of any extra data 1630 @param extra(dict, None): serialisable dictionary of any extra data
1600 will be encoded to json in database 1631 will be encoded to json in database
1601 @param perms_to_check(tuple[unicode],None): permission to check 1632 @param perms_to_check(tuple[unicode],None): permission to check
1602 must be a tuple of C.ACCESS_PERM_* or None 1633 must be a tuple of C.ACCESS_PERM_* or None
1603 if None, permission will no be checked (peer_jid must be None too in this 1634 if None, permission will not be checked (peer_jid must be None too in this
1604 case) 1635 case)
1605 @param profile(unicode): profile owning the file 1636 @param profile(unicode): profile owning the file
1606 """ 1637 """
1607 if "/" in name: 1638 if "/" in name:
1608 raise ValueError('name must not contain a slash ("/")') 1639 raise ValueError('name must not contain a slash ("/")')
1628 if type_ == C.FILE_TYPE_DIRECTORY: 1659 if type_ == C.FILE_TYPE_DIRECTORY:
1629 if any((version, file_hash, size, mime_type)): 1660 if any((version, file_hash, size, mime_type)):
1630 raise ValueError( 1661 raise ValueError(
1631 "version, file_hash, size and mime_type can't be set for a directory" 1662 "version, file_hash, size and mime_type can't be set for a directory"
1632 ) 1663 )
1633 if owner is not None: 1664 owner = self.getFilesOwner(client, owner, peer_jid, file_id, parent)
1634 owner = owner.userhostJID()
1635 1665
1636 if path is not None: 1666 if path is not None:
1637 path = str(path) 1667 path = str(path)
1638 # _getParentDir will check permissions if peer_jid is set, so we use owner 1668 # _getParentDir will check permissions if peer_jid is set, so we use owner
1639 parent, remaining_path_elts = yield self._getParentDir( 1669 parent, remaining_path_elts = yield self._getParentDir(