Mercurial > libervia-backend
view libervia/backend/plugins/plugin_comp_file_sharing_management.py @ 4334:111dce64dcb5
plugins XEP-0300, XEP-0446, XEP-0447, XEP0448 and others: Refactoring to use Pydantic:
Pydantic models are used more and more in Libervia, for the bridge API, and also to
convert `domish.Element` to internal representation.
Type hints have also been added in many places.
rel 453
author | Goffi <goffi@goffi.org> |
---|---|
date | Tue, 03 Dec 2024 00:12:38 +0100 |
parents | 0d7bb4df2343 |
children |
line wrap: on
line source
#!/usr/bin/env python3 # Libervia plugin to manage file sharing component through ad-hoc commands # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. import os.path from functools import partial from wokkel import data_form from twisted.internet import defer from twisted.words.protocols.jabber import jid from libervia.backend.core.i18n import _, D_ from libervia.backend.core import exceptions from libervia.backend.core.constants import Const as C from libervia.backend.core.log import getLogger from libervia.backend.tools.common import utils log = getLogger(__name__) PLUGIN_INFO = { C.PI_NAME: "File Sharing Management", C.PI_IMPORT_NAME: "FILE_SHARING_MANAGEMENT", C.PI_MODES: [C.PLUG_MODE_COMPONENT], C.PI_TYPE: "EXP", C.PI_PROTOCOLS: [], C.PI_DEPENDENCIES: ["XEP-0050", "XEP-0264"], C.PI_RECOMMENDATIONS: [], C.PI_MAIN: "FileSharingManagement", C.PI_HANDLER: "no", C.PI_DESCRIPTION: _( "Experimental handling of file management for file sharing. This plugins allows " "to change permissions of stored files/directories or remove them." ), } NS_FILE_MANAGEMENT = "https://salut-a-toi.org/protocol/file-management:0" NS_FILE_MANAGEMENT_PERM = "https://salut-a-toi.org/protocol/file-management:0#perm" NS_FILE_MANAGEMENT_DELETE = "https://salut-a-toi.org/protocol/file-management:0#delete" NS_FILE_MANAGEMENT_THUMB = "https://salut-a-toi.org/protocol/file-management:0#thumb" NS_FILE_MANAGEMENT_QUOTA = "https://salut-a-toi.org/protocol/file-management:0#quota" class WorkflowError(Exception): """Raised when workflow can't be completed""" def __init__(self, err_args): """ @param err_args(tuple): arguments to return to finish the command workflow """ Exception.__init__(self) self.err_args = err_args class FileSharingManagement(object): # This is a temporary way (Q&D) to handle stored files, a better way (using pubsub # syntax?) should be elaborated and proposed as a standard. def __init__(self, host): log.info(_("File Sharing Management plugin initialization")) self.host = host self._c = host.plugins["XEP-0050"] self._t = host.plugins["XEP-0264"] self.files_path = host.get_local_path(None, C.FILES_DIR) host.bridge.add_method( "file_sharing_delete", ".plugin", in_sign="ssss", out_sign="", method=self._delete, async_=True, ) def profile_connected(self, client): self._c.add_ad_hoc_command( client, self._on_change_file, "Change Permissions of File(s)", node=NS_FILE_MANAGEMENT_PERM, allowed_magics=C.ENTITY_ALL, ) self._c.add_ad_hoc_command( client, self._on_delete_file, "Delete File(s)", node=NS_FILE_MANAGEMENT_DELETE, allowed_magics=C.ENTITY_ALL, ) self._c.add_ad_hoc_command( client, self._on_gen_thumbnails, "Generate Thumbnails", node=NS_FILE_MANAGEMENT_THUMB, allowed_magics=C.ENTITY_ALL, ) self._c.add_ad_hoc_command( client, self._on_quota, "Get Quota", node=NS_FILE_MANAGEMENT_QUOTA, allowed_magics=C.ENTITY_ALL, ) def _delete(self, service_jid_s, path, namespace, profile): client = self.host.get_client(profile) service_jid = jid.JID(service_jid_s) if service_jid_s else None return defer.ensureDeferred( self._c.sequence( client, [{"path": path, "namespace": namespace}, {"confirm": True}], NS_FILE_MANAGEMENT_DELETE, service_jid, ) ) def _err(self, reason): """Helper method to get argument to return for error workflow will be interrupted with an error note @param reason(unicode): reason of the error @return (tuple): arguments to use in defer.returnValue """ status = self._c.STATUS.COMPLETED payload = None note = (self._c.NOTE.ERROR, reason) return payload, status, None, note def _get_root_args(self): """Create the form to select the file to use @return (tuple): arguments to use in defer.returnValue """ status = self._c.STATUS.EXECUTING form = data_form.Form( "form", title="File Management", formNamespace=NS_FILE_MANAGEMENT ) field = data_form.Field("text-single", "path", required=True) form.addField(field) field = data_form.Field("text-single", "namespace", required=False) form.addField(field) payload = form.toElement() return payload, status, None, None async def _get_file_data(self, client, session_data, command_form): """Retrieve field requested in root form "found_file" will also be set in session_data @param command_form(data_form.Form): response to root form @return (D(dict)): found file data @raise WorkflowError: something is wrong """ fields = command_form.fields try: path = fields["path"].value.strip() namespace = fields["namespace"].value or None except KeyError: self._c.ad_hoc_error(self._c.ERROR.BAD_PAYLOAD) if not path: self._c.ad_hoc_error(self._c.ERROR.BAD_PAYLOAD) requestor = session_data["requestor"] requestor_bare = requestor.userhostJID() path = path.rstrip("/") parent_path, basename = os.path.split(path) # TODO: if parent_path and basename are empty, we ask for root directory # this must be managed try: found_files = await self.host.memory.get_files( client, requestor_bare, path=parent_path, name=basename, namespace=namespace, ) found_file = found_files[0] except (exceptions.NotFound, IndexError): raise WorkflowError(self._err(_("file not found"))) except exceptions.PermissionError: raise WorkflowError(self._err(_("forbidden"))) if found_file["owner"] != requestor_bare: # only owner can manage files log.warning(_("Only owner can manage files")) raise WorkflowError(self._err(_("forbidden"))) session_data["found_file"] = found_file session_data["namespace"] = namespace return found_file def _update_read_permission(self, access, allowed_jids): if not allowed_jids: if C.ACCESS_PERM_READ in access: del access[C.ACCESS_PERM_READ] elif allowed_jids == "PUBLIC": access[C.ACCESS_PERM_READ] = {"type": C.ACCESS_TYPE_PUBLIC} else: access[C.ACCESS_PERM_READ] = { "type": C.ACCESS_TYPE_WHITELIST, "jids": [j.full() for j in allowed_jids], } async def _update_dir(self, client, requestor, namespace, file_data, allowed_jids): """Recursively update permission of a directory and all subdirectories @param file_data(dict): metadata of the file @param allowed_jids(list[jid.JID]): list of entities allowed to read the file """ assert file_data["type"] == C.FILE_TYPE_DIRECTORY files_data = await self.host.memory.get_files( client, requestor, parent=file_data["id"], namespace=namespace ) for file_data in files_data: if not file_data["access"].get(C.ACCESS_PERM_READ, {}): log.debug( "setting {perm} read permission for {name}".format( perm=allowed_jids, name=file_data["name"] ) ) await self.host.memory.file_update( file_data["id"], "access", partial(self._update_read_permission, allowed_jids=allowed_jids), ) if file_data["type"] == C.FILE_TYPE_DIRECTORY: await self._update_dir(client, requestor, namespace, file_data, "PUBLIC") async def _on_change_file(self, client, command_elt, session_data, action, node): try: x_elt = next(command_elt.elements(data_form.NS_X_DATA, "x")) command_form = data_form.Form.fromElement(x_elt) except StopIteration: command_form = None found_file = session_data.get("found_file") requestor = session_data["requestor"] requestor_bare = requestor.userhostJID() if command_form is None or len(command_form.fields) == 0: # root request return self._get_root_args() elif found_file is None: # file selected, we retrieve it and ask for permissions try: found_file = await self._get_file_data(client, session_data, command_form) except WorkflowError as e: return e.err_args # management request if found_file["type"] == C.FILE_TYPE_DIRECTORY: instructions = D_("Please select permissions for this directory") else: instructions = D_("Please select permissions for this file") form = data_form.Form( "form", title="File Management", instructions=[instructions], formNamespace=NS_FILE_MANAGEMENT, ) field = data_form.Field( "text-multi", "read_allowed", required=False, desc="list of jids allowed to read this file (beside yourself), or " '"PUBLIC" to let a public access', ) read_access = found_file["access"].get(C.ACCESS_PERM_READ, {}) access_type = read_access.get("type", C.ACCESS_TYPE_WHITELIST) if access_type == C.ACCESS_TYPE_PUBLIC: field.values = ["PUBLIC"] else: field.values = read_access.get("jids", []) form.addField(field) if found_file["type"] == C.FILE_TYPE_DIRECTORY: field = data_form.Field( "boolean", "recursive", value=False, required=False, desc="Files under it will be made public to follow this dir " "permission (only if they don't have already a permission set).", ) form.addField(field) status = self._c.STATUS.EXECUTING payload = form.toElement() return (payload, status, None, None) else: # final phase, we'll do permission change here try: read_allowed = command_form.fields["read_allowed"] except KeyError: self._c.ad_hoc_error(self._c.ERROR.BAD_PAYLOAD) if read_allowed.value == "PUBLIC": allowed_jids = "PUBLIC" elif read_allowed.value.strip() == "": allowed_jids = None else: try: allowed_jids = [ jid.JID(v.strip()) for v in read_allowed.values if v.strip() ] except RuntimeError as e: log.warning( _("Can't use read_allowed values: {reason}").format(reason=e) ) self._c.ad_hoc_error(self._c.ERROR.BAD_PAYLOAD) if found_file["type"] == C.FILE_TYPE_FILE: await self.host.memory.file_update( found_file["id"], "access", partial(self._update_read_permission, allowed_jids=allowed_jids), ) else: try: recursive = command_form.fields["recursive"] except KeyError: self._c.ad_hoc_error(self._c.ERROR.BAD_PAYLOAD) await self.host.memory.file_update( found_file["id"], "access", partial(self._update_read_permission, allowed_jids=allowed_jids), ) if recursive: # we set all file under the directory as public (if they haven't # already a permission set), so allowed entities of root directory # can read them. namespace = session_data["namespace"] await self._update_dir( client, requestor_bare, namespace, found_file, "PUBLIC" ) # job done, we can end the session status = self._c.STATUS.COMPLETED payload = None note = (self._c.NOTE.INFO, _("management session done")) return (payload, status, None, note) async def _on_delete_file(self, client, command_elt, session_data, action, node): try: x_elt = next(command_elt.elements(data_form.NS_X_DATA, "x")) command_form = data_form.Form.fromElement(x_elt) except StopIteration: command_form = None found_file = session_data.get("found_file") requestor = session_data["requestor"] requestor_bare = requestor.userhostJID() if command_form is None or len(command_form.fields) == 0: # root request return self._get_root_args() elif found_file is None: # file selected, we need confirmation before actually deleting try: found_file = await self._get_file_data(client, session_data, command_form) except WorkflowError as e: return e.err_args if found_file["type"] == C.FILE_TYPE_DIRECTORY: msg = D_( "Are you sure to delete directory {name} and all files and " "directories under it?" ).format(name=found_file["name"]) else: msg = D_( "Are you sure to delete file {name}?".format(name=found_file["name"]) ) form = data_form.Form( "form", title="File Management", instructions=[msg], formNamespace=NS_FILE_MANAGEMENT, ) field = data_form.Field( "boolean", "confirm", value=False, required=True, desc="check this box to confirm", ) form.addField(field) status = self._c.STATUS.EXECUTING payload = form.toElement() return (payload, status, None, None) else: # final phase, we'll do deletion here try: confirmed = C.bool(command_form.fields["confirm"].value) except KeyError: self._c.ad_hoc_error(self._c.ERROR.BAD_PAYLOAD) if not confirmed: note = None else: recursive = found_file["type"] == C.FILE_TYPE_DIRECTORY await self.host.memory.file_delete( client, requestor_bare, found_file["id"], recursive ) note = (self._c.NOTE.INFO, _("file deleted")) status = self._c.STATUS.COMPLETED payload = None return (payload, status, None, note) def _update_thumbs(self, extra, thumbnails): extra[C.KEY_THUMBNAILS] = thumbnails async def _gen_thumbs(self, client, requestor, namespace, file_data): """Recursively generate thumbnails @param file_data(dict): metadata of the file """ if file_data["type"] == C.FILE_TYPE_DIRECTORY: sub_files_data = await self.host.memory.get_files( client, requestor, parent=file_data["id"], namespace=namespace ) for sub_file_data in sub_files_data: await self._gen_thumbs(client, requestor, namespace, sub_file_data) elif file_data["type"] == C.FILE_TYPE_FILE: media_type = file_data["media_type"] file_path = os.path.join(self.files_path, file_data["file_hash"]) if media_type == "image": thumbnails = [] for max_thumb_size in self._t.SIZES: try: thumb_size, thumb_id = await self._t.generate_thumbnail( file_path, max_thumb_size, # we keep thumbnails for 6 months 60 * 60 * 24 * 31 * 6, ) except Exception as e: log.warning( _("Can't create thumbnail: {reason}").format(reason=e) ) break thumbnails.append({"id": thumb_id, "size": thumb_size}) await self.host.memory.file_update( file_data["id"], "extra", partial(self._update_thumbs, thumbnails=thumbnails), ) log.info( "thumbnails for [{file_name}] generated".format( file_name=file_data["name"] ) ) else: log.warning("unmanaged file type: {type_}".format(type_=file_data["type"])) async def _on_gen_thumbnails(self, client, command_elt, session_data, action, node): try: x_elt = next(command_elt.elements(data_form.NS_X_DATA, "x")) command_form = data_form.Form.fromElement(x_elt) except StopIteration: command_form = None found_file = session_data.get("found_file") requestor = session_data["requestor"] if command_form is None or len(command_form.fields) == 0: # root request return self._get_root_args() elif found_file is None: # file selected, we retrieve it and ask for permissions try: found_file = await self._get_file_data(client, session_data, command_form) except WorkflowError as e: return e.err_args log.info("Generating thumbnails as requested") await self._gen_thumbs(client, requestor, found_file["namespace"], found_file) # job done, we can end the session status = self._c.STATUS.COMPLETED payload = None note = (self._c.NOTE.INFO, _("thumbnails generated")) return (payload, status, None, note) async def _on_quota(self, client, command_elt, session_data, action, node): requestor = session_data["requestor"] quota = self.host.plugins["file_sharing"].get_quota(client, requestor) try: size_used = await self.host.memory.file_get_used_space(client, requestor) except exceptions.PermissionError: raise WorkflowError(self._err(_("forbidden"))) status = self._c.STATUS.COMPLETED form = data_form.Form("result") form.makeFields({"quota": quota, "user": size_used}) payload = form.toElement() note = ( self._c.NOTE.INFO, _("You are currently using {size_used} on {size_quota}").format( size_used=utils.get_human_size(size_used), size_quota=( _("unlimited quota") if quota is None else utils.get_human_size(quota) ), ), ) return (payload, status, None, note)