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

libervia/backend/plugins/plugin_misc_download.py

@@ -277,12 +277,12 @@ class DownloadPlugin(object):
         callback: Callable[
             [
                 SatXMPPEntity,
-                Dict[str, Any],
-                Dict[str, Any],
+                dict[str, Any],
+                dict[str, Any],
                 Union[str, Path],
-                Dict[str, Any],
+                dict[str, Any],
             ],
-            Tuple[str, defer.Deferred],
+            tuple[str, defer.Deferred],
         ],
     ) -> None:
         """Register a handler to manage a type of attachment source
@@ -295,11 +295,11 @@ class DownloadPlugin(object):
             source used.
             The callabke must return a tuple with:
                 - progress ID
-                - a Deferred which fire whant the file is fully downloaded
+                - a Deferred which fire when the file is fully downloaded
         """
-        if source_type is self._download_callbacks:
+        if source_type in self._download_callbacks:
             raise exceptions.ConflictError(
-                f"The is already a callback registered for source type {source_type!r}"
+                f"There is already a callback registered for source type {source_type!r}"
             )
         self._download_callbacks[source_type] = callback
 

libervia/backend/plugins/plugin_misc_text_syntaxes.py

@@ -43,7 +43,7 @@ try:
 except ImportError:
     raise exceptions.MissingModule(
         'Missing module "lxml_html_clean". Please download and install it from '
-        'http://lxml.de/. Note that this is a separate package to install in addition '
+        "http://lxml.de/. Note that this is a separate package to install in addition "
         'to "lxml".'
     )
 

libervia/backend/plugins/plugin_misc_uri_finder.py

@@ -74,12 +74,12 @@ class URIFinder(object):
         @return (dict[unicode, unicode]): map from key to found uri
         """
         keys_re = "|".join(keys)
-        label_re = r'\"(?P<label>[^\"]+)\"'
+        label_re = r"\"(?P<label>[^\"]+)\""
         uri_re = re.compile(
             r"(?P<key>{keys_re})[ :]? +(?P<uri>xmpp:\S+)(?:.*use {label_re} label)?".format(
                 keys_re=keys_re, label_re=label_re
             ),
-            re.IGNORECASE
+            re.IGNORECASE,
         )
         path = os.path.normpath(path)
         if not os.path.isdir(path) or not os.path.isabs(path):

libervia/backend/plugins/plugin_xep_0048.py

@@ -422,10 +422,7 @@ class XEP_0048(object):
             raise NotImplementedError
 
     def bookmarks_list(
-            self,
-            type_: str,
-            storage_location: str,
-            profile_key: str = C.PROF_KEY_NONE
+        self, type_: str, storage_location: str, profile_key: str = C.PROF_KEY_NONE
     ) -> defer.Deferred[dict]:
         """Return stored bookmarks
 

libervia/backend/plugins/plugin_xep_0060.py

@@ -711,10 +711,10 @@ class XEP_0060(object):
         client: SatXMPPEntity,
         service: jid.JID,
         nodeIdentifier: str,
-        items: list[pubsub.Item]|None = None,
-        options: dict|None = None,
-        sender: jid.JID|None = None,
-        extra: dict[str, Any]|None = None,
+        items: list[pubsub.Item] | None = None,
+        options: dict | None = None,
+        sender: jid.JID | None = None,
+        extra: dict[str, Any] | None = None,
     ) -> domish.Element:
         """Publish pubsub items
 
@@ -807,13 +807,13 @@ class XEP_0060(object):
     async def get_items(
         self,
         client: SatXMPPEntity,
-        service: jid.JID|None,
+        service: jid.JID | None,
         node: str,
-        max_items: int|None = None,
-        item_ids: list[str]|None = None,
-        sub_id: str|None = None,
-        rsm_request: rsm.RSMRequest|None = None,
-        extra: dict|None = None,
+        max_items: int | None = None,
+        item_ids: list[str] | None = None,
+        sub_id: str | None = None,
+        rsm_request: rsm.RSMRequest | None = None,
+        extra: dict | None = None,
     ) -> tuple[list[domish.Element], dict]:
         """Retrieve pubsub items from a node.
 

libervia/backend/plugins/plugin_xep_0103.py

@@ -15,18 +15,18 @@
 # 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/>.
 
-from typing import Dict, Any
+from typing import Self
 
+from pydantic import BaseModel, Field
 from twisted.words.xish import domish
 
+from libervia.backend.core import exceptions
 from libervia.backend.core.constants import Const as C
 from libervia.backend.core.i18n import _
 from libervia.backend.core.log import getLogger
-from libervia.backend.core import exceptions
 
 log = getLogger(__name__)
 
-
 PLUGIN_INFO = {
     C.PI_NAME: "URL Address Information",
     C.PI_IMPORT_NAME: "XEP-0103",
@@ -37,45 +37,87 @@ PLUGIN_INFO = {
     C.PI_HANDLER: "no",
     C.PI_DESCRIPTION: _("""Implementation of XEP-0103 (URL Address Information)"""),
 }
-
 NS_URL_DATA = "http://jabber.org/protocol/url-data"
 
 
+class Desc(BaseModel):
+    """
+    Model for the <desc/> element.
+    """
+
+    content: str
+    xml_lang: str | None = None
+
+
+class URLData(BaseModel):
+    """
+    Model for the <url-data/> element.
+    """
+
+    url: str
+    desc: list[Desc] = Field(default_factory=list)
+
+    @classmethod
+    def from_element(cls, element: domish.Element) -> Self:
+        """Create a URLData instance from a <url-data> element or its parent.
+
+        @param url_data_elt: The <url-data> element or a parent element.
+        @return: URLData instance.
+        @raise exceptions.NotFound: If the <url-data> element is not found.
+        """
+        if element.uri != NS_URL_DATA or element.name != "url-data":
+            child_url_data_elt = next(element.elements(NS_URL_DATA, "url-data"), None)
+            if child_url_data_elt is None:
+                raise exceptions.NotFound("<url-data> element not found")
+            else:
+                element = child_url_data_elt
+        kwargs = {
+            "url": element["target"],
+            "desc": [
+                Desc(content=str(desc_elt), xml_lang=desc_elt.getAttribute("xml:lang"))
+                for desc_elt in element.elements(NS_URL_DATA, "desc")
+            ],
+        }
+        return cls(**kwargs)
+
+    def to_element(self) -> domish.Element:
+        """Build the <url-data> element from this instance's data.
+
+        @return: <url-data> element.
+        """
+        url_data_elt = domish.Element((NS_URL_DATA, "url-data"))
+        url_data_elt["target"] = self.url
+        for desc in self.desc:
+            desc_elt = url_data_elt.addElement((NS_URL_DATA, "desc"))
+            if desc.xml_lang:
+                desc_elt["xml:lang"] = desc.xml_lang
+            desc_elt.addContent(desc.content)
+        return url_data_elt
+
+
 class XEP_0103:
     namespace = NS_URL_DATA
 
     def __init__(self, host):
-        log.info(_("XEP-0103 (URL Address Information) plugin initialization"))
+        log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization")
         host.register_namespace("url-data", NS_URL_DATA)
 
-    def get_url_data_elt(self, url: str, **kwargs) -> domish.Element:
+    def generate_url_data(self, url: str, **kwargs) -> URLData:
         """Generate the element describing the URL
 
         @param url: URL to use
         @param extra: extra metadata describing how to access the URL
         @return: ``<url-data/>`` element
         """
-        url_data_elt = domish.Element((NS_URL_DATA, "url-data"))
-        url_data_elt["target"] = url
-        return url_data_elt
+        url_data = URLData(url=url, **kwargs)
+        return url_data
 
-    def parse_url_data_elt(self, url_data_elt: domish.Element) -> Dict[str, Any]:
+    def parse_url_data_elt(self, url_data_elt: domish.Element) -> URLData:
         """Parse <url-data/> element
 
         @param url_data_elt: <url-data/> element
             a parent element can also be used
-        @return: url-data data. It's a dict whose keys correspond to
-            [get_url_data_elt] parameters
+        @return: URLData instance
         @raise exceptions.NotFound: no <url-data/> element has been found
         """
-        if url_data_elt.name != "url-data":
-            try:
-                url_data_elt = next(url_data_elt.elements(NS_URL_DATA, "url-data"))
-            except StopIteration:
-                raise exceptions.NotFound
-        try:
-            data: Dict[str, Any] = {"url": url_data_elt["target"]}
-        except KeyError:
-            raise ValueError(f'"target" attribute is missing: {url_data_elt.toXml}')
-
-        return data
+        return URLData.from_element(url_data_elt)

libervia/backend/plugins/plugin_xep_0234.py

@@ -437,14 +437,14 @@ class XEP_0234(BaseApplicationHandler):
 
     def _file_jingle_request(
         self,
-        peer_jid,
-        filepath,
-        name="",
-        file_hash="",
-        hash_algo="",
-        extra=None,
-        profile=C.PROF_KEY_NONE,
-    ):
+        peer_jid: str,
+        filepath: str,
+        name: str = "",
+        file_hash: str = "",
+        hash_algo: str = "",
+        extra: dict | None = None,
+        profile: str = C.PROF_KEY_NONE,
+    ) -> defer.Deferred[str]:
         client = self.host.get_client(profile)
         return defer.ensureDeferred(
             self.file_jingle_request(
@@ -460,21 +460,22 @@ class XEP_0234(BaseApplicationHandler):
 
     async def file_jingle_request(
         self,
-        client,
-        peer_jid,
-        filepath,
-        name=None,
-        file_hash=None,
-        hash_algo=None,
-        extra=None,
-    ):
+        client: SatXMPPEntity,
+        peer_jid: jid.JID,
+        filepath: str,
+        name: str | None = None,
+        file_hash: str | None = None,
+        hash_algo: str | None = None,
+        extra: dict | None = None,
+    ) -> str:
         """Request a file using jingle file transfer
 
-        @param peer_jid(jid.JID): destinee jid
-        @param filepath(str): absolute path where the file will be downloaded
-        @param name(unicode, None): name of the file
-        @param file_hash(unicode, None): hash of the file
-        @return (D(unicode)): progress id
+        @param client: Profile session.
+        @param peer_jid: destinee jid
+        @param filepath: absolute path where the file will be downloaded
+        @param name: name of the file
+        @param file_hash: hash of the file
+        @return: progress id
         """
         progress_id_d = defer.Deferred()
         if extra is None:

libervia/backend/plugins/plugin_xep_0300.py

@@ -1,8 +1,7 @@
 #!/usr/bin/env python3
 
-
-# SAT plugin for Hash functions (XEP-0300)
-# Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org)
+# Libervia plugin for Hash functions (XEP-0300)
+# Copyright (C) 2009-2024 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
@@ -17,13 +16,16 @@
 # 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/>.
 
-from typing import Tuple
 import base64
 from collections import OrderedDict
 import hashlib
+from typing import BinaryIO, Callable, Self, TYPE_CHECKING
 
+from _hashlib import HASH
+from pydantic import BaseModel, Field
 from twisted.internet import threads
 from twisted.internet import defer
+from twisted.words.protocols.jabber import jid
 from twisted.words.protocols.jabber.xmlstream import XMPPHandler
 from twisted.words.xish import domish
 from wokkel import disco, iwokkel
@@ -31,9 +33,13 @@ from zope.interface import implementer
 
 from libervia.backend.core import exceptions
 from libervia.backend.core.constants import Const as C
+from libervia.backend.core.core_types import SatXMPPEntity
 from libervia.backend.core.i18n import _
 from libervia.backend.core.log import getLogger
 
+if TYPE_CHECKING:
+    from libervia.backend.core.main import LiberviaBackend
+
 log = getLogger(__name__)
 
 
@@ -54,9 +60,91 @@ BUFFER_SIZE = 2**12
 ALGO_DEFAULT = "sha-256"
 
 
-class XEP_0300(object):
+class Hash(BaseModel):
+    """
+    Model for hash data.
+    """
+
+    algo: str = Field(description="The algorithm used for hashing.")
+    hash_: str = Field(min_length=16, description="The base64-encoded hash value.")
+
+    @classmethod
+    def from_element(cls, hash_elt: domish.Element) -> Self:
+        """
+        Create a HashModel instance from a <hash> element.
+
+        @param hash_elt: The <hash> element.
+        @return: HashModel instance.
+        @raise exceptions.NotFound: If the <hash> element is not found.
+        """
+        if hash_elt.uri != NS_HASHES or hash_elt.name != "hash":
+            raise exceptions.NotFound("<hash> element not found")
+        algo = hash_elt.getAttribute("algo")
+        hash_value = str(hash_elt)
+        return cls(algo=algo, hash_=hash_value)
+
+    @classmethod
+    def from_parent(cls, parent_elt: domish.Element) -> list[Self]:
+        """Find and return child <hash> element in given parent.
+
+        @param parent_elt: Element which may content child <hash> elements.
+        @return: list of Hash corresponding to found elements
+        """
+        return [
+            cls.from_element(hash_elt)
+            for hash_elt in parent_elt.elements(NS_HASHES, "hash")
+        ]
+
+    def to_element(self) -> domish.Element:
+        """Build the <hash> element from this instance's data.
+
+        @return: <hash> element.
+        """
+        hash_elt = domish.Element((NS_HASHES, "hash"))
+        hash_elt["algo"] = self.algo
+        hash_elt.addContent(self.hash_)
+        return hash_elt
+
+
+class HashUsed(BaseModel):
+    """
+    Model for hash-used data.
+    """
+
+    algo: str = Field(description="The algorithm used for hashing.")
+
+    @classmethod
+    def from_element(cls, hash_used_elt: domish.Element) -> Self:
+        """Create a HashUsedModel instance from a <hash-used> element.
+
+        @param hash_used_elt: The <hash-used> element.
+        @return: HashUsedModel instance.
+        @raise exceptions.NotFound: If the <hash-used> element is not found.
+        """
+        if hash_used_elt.uri != NS_HASHES or hash_used_elt.name != "hash-used":
+            child_hash_used_elt = next(
+                hash_used_elt.elements(NS_HASHES, "hash-used"), None
+            )
+            if child_hash_used_elt is None:
+                raise exceptions.NotFound("<hash-used> element not found")
+            else:
+                hash_used_elt = child_hash_used_elt
+        algo = hash_used_elt.getAttribute("algo")
+        return cls(algo=algo)
+
+    def to_element(self) -> domish.Element:
+        """Build the <hash-used> element from this instance's data.
+
+        @return: <hash-used> element.
+        """
+        hash_used_elt = domish.Element((NS_HASHES, "hash-used"))
+        hash_used_elt["algo"] = self.algo
+        return hash_used_elt
+
+
+class XEP_0300:
     # TODO: add blake after moving to Python 3
-    ALGOS = OrderedDict(
+    ALGOS: OrderedDict[str, Callable] = OrderedDict(
         (
             ("md5", hashlib.md5),
             ("sha-1", hashlib.sha1),
@@ -66,38 +154,38 @@ class XEP_0300(object):
     )
     ALGO_DEFAULT = ALGO_DEFAULT
 
-    def __init__(self, host):
+    def __init__(self, host: "LiberviaBackend"):
         log.info(_("plugin Hashes initialization"))
         host.register_namespace("hashes", NS_HASHES)
+        self.host = host
 
-    def get_handler(self, client):
+    def get_handler(self, client: SatXMPPEntity) -> XMPPHandler:
         return XEP_0300_handler()
 
-    def get_hasher(self, algo=ALGO_DEFAULT):
+    def get_hasher(self, algo: str = ALGO_DEFAULT) -> Callable:
         """Return hasher instance
 
-        @param algo(unicode): one of the XEP_300.ALGOS keys
-        @return (hash object): same object s in hashlib.
-           update method need to be called for each chunh
-           diget or hexdigest can be used at the end
+        @param algo: one of the XEP_300.ALGOS keys
+        @return: same object s in hashlib.
+           update method need to be called for each chunk
+           digest or hexdigest can be used at the end
         """
         return self.ALGOS[algo]()
 
-    def get_default_algo(self):
+    def get_default_algo(self) -> str:
         return ALGO_DEFAULT
 
-    @defer.inlineCallbacks
-    def get_best_peer_algo(self, to_jid, profile):
-        """Return the best available hashing algorith of other peer
+    async def get_best_peer_algo(self, to_jid: jid.JID, profile: str) -> str | None:
+        """Return the best available hashing algorithm of other peer
 
-        @param to_jid(jid.JID): peer jid
-        @parm profile: %(doc_profile)s
-        @return (D(unicode, None)): best available algorithm,
+        @param to_jid: peer jid
+        @param profile: %(doc_profile)s
+        @return: best available algorithm,
            or None if hashing is not possible
         """
         client = self.host.get_client(profile)
         for algo in reversed(XEP_0300.ALGOS):
-            has_feature = yield self.host.hasFeature(
+            has_feature = await self.host.hasFeature(
                 client, NS_HASHES_FUNCTIONS.format(algo), to_jid
             )
             if has_feature:
@@ -106,15 +194,15 @@ class XEP_0300(object):
                         jid=to_jid.full(), algo=algo
                     )
                 )
-                defer.returnValue(algo)
+                return algo
 
-    def _calculate_hash_blocking(self, file_obj, hasher):
+    def _calculate_hash_blocking(self, file_obj: BinaryIO, hasher: HASH) -> str:
         """Calculate hash in a blocking way
 
         /!\\ blocking method, please use calculate_hash instead
-        @param file_obj(file): a file-like object
-        @param hasher(hash object): the method to call to initialise hash object
-        @return (str): the hex digest of the hash
+        @param file_obj: a file-like object
+        @param hasher: the method to call to initialise hash object
+        @return: the hex digest of the hash
         """
         while True:
             buf = file_obj.read(BUFFER_SIZE)
@@ -123,64 +211,49 @@ class XEP_0300(object):
             hasher.update(buf)
         return hasher.hexdigest()
 
-    def calculate_hash(self, file_obj, hasher):
+    def calculate_hash(self, file_obj: BinaryIO, hasher: HASH) -> defer.Deferred[str]:
         return threads.deferToThread(self._calculate_hash_blocking, file_obj, hasher)
 
-    def calculate_hash_elt(self, file_obj=None, algo=ALGO_DEFAULT):
+    async def calculate_hash_elt(
+        self, file_obj: BinaryIO, algo: str = ALGO_DEFAULT
+    ) -> domish.Element:
         """Compute hash and build hash element
 
-        @param file_obj(file, None): file-like object to use to calculate the hash
-        @param algo(unicode): algorithme to use, must be a key of XEP_0300.ALGOS
-        @return (D(domish.Element)): hash element
+        @param file_obj: file-like object to use to calculate the hash
+        @param algo: algorithm to use, must be a key of XEP_0300.ALGOS
+        @return: hash element
         """
-
-        def hash_calculated(hash_):
-            return self.build_hash_elt(hash_, algo)
-
         hasher = self.get_hasher(algo)
-        hash_d = self.calculate_hash(file_obj, hasher)
-        hash_d.addCallback(hash_calculated)
-        return hash_d
+        hash_ = await self.calculate_hash(file_obj, hasher)
+        return self.build_hash_elt(hash_, algo)
 
-    def build_hash_used_elt(self, algo=ALGO_DEFAULT):
-        hash_used_elt = domish.Element((NS_HASHES, "hash-used"))
-        hash_used_elt["algo"] = algo
-        return hash_used_elt
+    def build_hash_used_elt(self, algo: str = ALGO_DEFAULT) -> domish.Element:
+        hash_used_model = HashUsed(algo=algo)
+        return hash_used_model.to_element()
 
-    def parse_hash_used_elt(self, parent):
+    def parse_hash_used_elt(self, parent_elt: domish.Element) -> str:
         """Find and parse a hash-used element
 
-        @param (domish.Element): parent of <hash/> element
-        @return (unicode): hash algorithm used
+        @param parent: parent of <hash-used/> element
+        @return: hash algorithm used
         @raise exceptions.NotFound: the element is not present
         @raise exceptions.DataError: the element is invalid
         """
-        try:
-            hash_used_elt = next(parent.elements(NS_HASHES, "hash-used"))
-        except StopIteration:
-            raise exceptions.NotFound
-        algo = hash_used_elt["algo"]
-        if not algo:
-            raise exceptions.DataError
-        return algo
+        hash_used_model = HashUsed.from_element(parent_elt)
+        return hash_used_model.algo
 
-    def build_hash_elt(self, hash_, algo=ALGO_DEFAULT):
+    def build_hash_elt(self, hash_hex: str, algo: str = ALGO_DEFAULT) -> domish.Element:
         """Compute hash and build hash element
 
-        @param hash_(str): hash to use
-        @param algo(unicode): algorithme to use, must be a key of XEP_0300.ALGOS
-        @return (domish.Element): computed hash
+        @param hash_: Hexadecimal representation of hash to use.
+        @param algo: Algorithm to use, must be a key of XEP_0300.ALGOS.
+        @return: <hash> element
         """
-        assert hash_
-        assert algo
-        hash_elt = domish.Element((NS_HASHES, "hash"))
-        if hash_ is not None:
-            b64_hash = base64.b64encode(hash_.encode("utf-8")).decode("utf-8")
-            hash_elt.addContent(b64_hash)
-        hash_elt["algo"] = algo
-        return hash_elt
+        b64_hash = base64.b64encode(hash_hex.encode()).decode()
+        hash_model = Hash(algo=algo, hash_=b64_hash)
+        return hash_model.to_element()
 
-    def parse_hash_elt(self, parent: domish.Element) -> Tuple[str, bytes]:
+    def parse_hash_elt(self, parent: domish.Element) -> tuple[str, str]:
         """Find and parse a hash element
 
         if multiple elements are found, the strongest managed one is returned
@@ -195,7 +268,8 @@ class XEP_0300(object):
         best_algo = None
         best_value = None
         for hash_elt in parent.elements(NS_HASHES, "hash"):
-            algo = hash_elt.getAttribute("algo")
+            hash_model = Hash.from_element(hash_elt)
+            algo = hash_model.algo
             try:
                 idx = algos.index(algo)
             except ValueError:
@@ -205,7 +279,7 @@ class XEP_0300(object):
 
             if best_algo is None or algos.index(best_algo) < idx:
                 best_algo = algo
-                best_value = base64.b64decode(str(hash_elt)).decode("utf-8")
+                best_value = base64.b64decode(hash_model.hash_).decode()
 
         if not hash_elt:
             raise exceptions.NotFound
@@ -217,12 +291,16 @@ class XEP_0300(object):
 @implementer(iwokkel.IDisco)
 class XEP_0300_handler(XMPPHandler):
 
-    def getDiscoInfo(self, requestor, target, nodeIdentifier=""):
+    def getDiscoInfo(
+        self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = ""
+    ) -> list[disco.DiscoFeature]:
         hash_functions_names = [
             disco.DiscoFeature(NS_HASHES_FUNCTIONS.format(algo))
             for algo in XEP_0300.ALGOS
         ]
         return [disco.DiscoFeature(NS_HASHES)] + hash_functions_names
 
-    def getDiscoItems(self, requestor, target, nodeIdentifier=""):
+    def getDiscoItems(
+        self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = ""
+    ) -> list[disco.DiscoItem]:
         return []

libervia/backend/plugins/plugin_xep_0373.py

@@ -787,23 +787,17 @@ class GPGME_GPGProvider(GPGProvider):
                     algorithm=self.ALGORITHM,
                     expires=False,
                     certify=True,
-                    force=True
+                    force=True,
                 )
 
                 primary_key_obj = c.get_key(primary_create_result.fpr, secret=True)
 
                 c.create_subkey(
-                    primary_key_obj,
-                    algorithm=self.ALGORITHM,
-                    expires=False,
-                    sign=True
+                    primary_key_obj, algorithm=self.ALGORITHM, expires=False, sign=True
                 )
 
                 c.create_subkey(
-                    primary_key_obj,
-                    algorithm=self.ALGORITHM,
-                    expires=False,
-                    encrypt=True
+                    primary_key_obj, algorithm=self.ALGORITHM, expires=False, encrypt=True
                 )
             except gpg.errors.GPGMEError as e:
                 raise GPGProviderError("Internal GPGME error") from e

libervia/backend/plugins/plugin_xep_0446.py

@@ -15,16 +15,17 @@
 # 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/>.
 
-from logging import exception
-from typing import Optional, Union, Tuple, Dict, Any
 from pathlib import Path
+from typing import Self, cast
 
+from pydantic import BaseModel, Field
 from twisted.words.xish import domish
 
+from libervia.backend.core import exceptions
 from libervia.backend.core.constants import Const as C
 from libervia.backend.core.i18n import _
 from libervia.backend.core.log import getLogger
-from libervia.backend.core import exceptions
+from libervia.backend.plugins.plugin_xep_0300 import XEP_0300
 from libervia.backend.tools import utils
 
 log = getLogger(__name__)
@@ -45,26 +46,171 @@ PLUGIN_INFO = {
 NS_FILE_METADATA = "urn:xmpp:file:metadata:0"
 
 
-class XEP_0446:
+class FileMetadata(BaseModel):
+    """
+    Model for file metadata.
+    """
 
+    name: str | None = Field(None, description="Name of the file.")
+    media_type: str | None = Field(None, description="Media type of the file.")
+    desc: str | None = Field(None, description="Description of the file.")
+    size: int | None = Field(None, description="Size of the file in bytes.")
+    file_hash: tuple[str, str] | None = Field(
+        None, description="File hash as a tuple of (algo, hash)."
+    )
+    date: float | int | None = Field(
+        None, description="Timestamp of the last modification datetime."
+    )
+    width: int | None = Field(None, description="Image or video width in pixels.")
+    height: int | None = Field(None, description="Image or video height in pixels.")
+    length: int | None = Field(None, description="Video or audio length in milliseconds.")
+    thumbnail: str | None = Field(None, description="URL to a thumbnail.")
+
+    _hash: XEP_0300 | None = None
+
+    @classmethod
+    def from_element(cls, file_metadata_elt: domish.Element) -> Self:
+        """Create a FileMetadata instance from a <file> element or its parent.
+
+        @param file_metadata_elt: The <file> element or a parent element.
+        @return: FileMetadata instance.
+        @raise exceptions.NotFound: If the <file> element is not found.
+        """
+        assert cls._hash is not None, "_hash attribute is not set"
+
+        if file_metadata_elt.uri != NS_FILE_METADATA or file_metadata_elt.name != "file":
+            child_file_metadata_elt = next(
+                file_metadata_elt.elements(NS_FILE_METADATA, "file"), None
+            )
+            if child_file_metadata_elt is None:
+                raise exceptions.NotFound("<file> element not found")
+            else:
+                file_metadata_elt = child_file_metadata_elt
+
+        kwargs = {}
+        for key in (
+            "name",
+            "media_type",
+            "desc",
+            "size",
+            "date",
+            "width",
+            "height",
+            "length",
+        ):
+            elt = next(file_metadata_elt.elements(NS_FILE_METADATA, key), None)
+            if elt is not None:
+                if key in ("name", "media_type", "desc"):
+                    content = str(elt)
+                    if key == "name":
+                        content = Path(content).name
+                    kwargs[key] = content
+                elif key in ("size", "width", "height", "length"):
+                    kwargs[key] = int(str(elt))
+                elif key == "date":
+                    kwargs[key] = utils.parse_xmpp_date(str(elt))
+
+        hash_elt = next(file_metadata_elt.elements(NS_FILE_METADATA, "hash"), None)
+        if hash_elt:
+            try:
+                algo, hash_ = cls._hash.parse_hash_elt(hash_elt)
+            except exceptions.NotFound:
+                pass
+            except exceptions.DataError:
+                from libervia.backend.tools.xml_tools import p_fmt_elt
+
+                log.warning(f"invalid <hash/> element:\n{p_fmt_elt(file_metadata_elt)}")
+            else:
+                kwargs["file_hash"] = (algo, hash_)
+
+        return cls(**kwargs)
+
+    def to_element(self) -> domish.Element:
+        """Build the <file> element from this instance's data.
+
+        @return: <file> element.
+        """
+        assert self._hash is not None, "_hash attribute is not set"
+
+        file_elt = domish.Element((NS_FILE_METADATA, "file"))
+
+        for key, value in (
+            ("name", self.name),
+            ("media-type", self.media_type),
+            ("desc", self.desc),
+            ("size", self.size),
+            ("width", self.width),
+            ("height", self.height),
+            ("length", self.length),
+        ):
+            if value is not None:
+                file_elt.addElement(key, content=str(value))
+
+        if self.file_hash is not None:
+            hash_algo, hash_ = self.file_hash
+            file_elt.addChild(self._hash.build_hash_elt(hash_, hash_algo))
+
+        if self.date is not None:
+            file_elt.addElement("date", content=utils.xmpp_date(self.date))
+
+        if self.thumbnail is not None:
+            log.warning("thumbnail is not implemented yet")
+
+        return file_elt
+
+    @classmethod
+    def from_filedata_dict(cls, file_data: dict) -> Self:
+        """Create an instance of FileMetadata from data dict as returned by ``memory``.
+
+        @param data: A filedata dict as returned by ``memory.get_files``
+        @return: An instance of FileMetadata.
+        """
+
+        # Extracting relevant fields
+        name = file_data["name"]
+        media_type = f'{file_data["media_type"]}/{file_data["media_subtype"]}'
+        desc = None  # TODO
+        size = file_data["size"]
+        file_hash = (file_data["hash_algo"], file_data["file_hash"])
+        date = file_data.get("modified") or file_data["created"]
+        width = None  # TODO
+        height = None  # TODO
+        length = None  # TODO
+        thumbnail = None  # TODO
+
+        return cls(
+            name=name,
+            media_type=media_type,
+            desc=desc,
+            size=size,
+            file_hash=file_hash,
+            date=date,
+            width=width,
+            height=height,
+            length=length,
+            thumbnail=thumbnail,
+        )
+
+
+class XEP_0446:
     def __init__(self, host):
         log.info(_("XEP-0446 (File Metadata Element) plugin initialization"))
         host.register_namespace("file-metadata", NS_FILE_METADATA)
-        self._hash = host.plugins["XEP-0300"]
+        FileMetadata._hash = cast(XEP_0300, host.plugins["XEP-0300"])
 
-    def get_file_metadata_elt(
+    def generate_file_metadata(
         self,
-        name: Optional[str] = None,
-        media_type: Optional[str] = None,
-        desc: Optional[str] = None,
-        size: Optional[int] = None,
-        file_hash: Optional[Tuple[str, str]] = None,
-        date: Optional[Union[float, int]] = None,
-        width: Optional[int] = None,
-        height: Optional[int] = None,
-        length: Optional[int] = None,
-        thumbnail: Optional[str] = None,
-    ) -> domish.Element:
+        name: str | None = None,
+        media_type: str | None = None,
+        desc: str | None = None,
+        size: int | None = None,
+        file_hash: tuple[str, str] | None = None,
+        date: float | int | None = None,
+        width: int | None = None,
+        height: int | None = None,
+        length: int | None = None,
+        thumbnail: str | None = None,
+    ) -> FileMetadata:
         """Generate the element describing a file
 
         @param name: name of the file
@@ -79,87 +225,25 @@ class XEP_0446:
         @param thumbnail: URL to a thumbnail
         @return: ``<file/>`` element
         """
-        if name:
-            name = Path(name).name
-        file_elt = domish.Element((NS_FILE_METADATA, "file"))
-        for name, value in (
-            ("name", name),
-            ("media-type", media_type),
-            ("desc", desc),
-            ("size", size),
-            ("width", width),
-            ("height", height),
-            ("length", length),
-        ):
-            if value is not None:
-                file_elt.addElement(name, content=str(value))
-        if file_hash is not None:
-            hash_algo, hash_ = file_hash
-            file_elt.addChild(self._hash.build_hash_elt(hash_, hash_algo))
-        if date is not None:
-            file_elt.addElement("date", utils.xmpp_date(date))
-        if thumbnail is not None:
-            # TODO: implement thumbnails
-            log.warning("thumbnail is not implemented yet")
-        return file_elt
+        return FileMetadata(
+            name=name,
+            media_type=media_type,
+            desc=desc,
+            size=size,
+            file_hash=file_hash,
+            date=date,
+            width=width,
+            height=height,
+            length=length,
+            thumbnail=thumbnail,
+        )
 
-    def parse_file_metadata_elt(
-        self, file_metadata_elt: domish.Element
-    ) -> Dict[str, Any]:
+    def parse_file_metadata_elt(self, file_metadata_elt: domish.Element) -> FileMetadata:
         """Parse <file/> element
 
         @param file_metadata_elt: <file/> element
             a parent element can also be used
-        @return: file metadata. It's a dict whose keys correspond to
-            [get_file_metadata_elt] parameters
+        @return: FileMetadata instance.
         @raise exceptions.NotFound: no <file/> element has been found
         """
-
-        if file_metadata_elt.name != "file":
-            try:
-                file_metadata_elt = next(
-                    file_metadata_elt.elements(NS_FILE_METADATA, "file")
-                )
-            except StopIteration:
-                raise exceptions.NotFound
-        data: Dict[str, Any] = {}
-
-        for key, type_ in (
-            ("name", str),
-            ("media-type", str),
-            ("desc", str),
-            ("size", int),
-            ("date", "timestamp"),
-            ("width", int),
-            ("height", int),
-            ("length", int),
-        ):
-            elt = next(file_metadata_elt.elements(NS_FILE_METADATA, key), None)
-            if elt is not None:
-                if type_ in (str, int):
-                    content = str(elt)
-                    if key == "name":
-                        # we avoid malformed names or names containing path elements
-                        content = Path(content).name
-                    elif key == "media-type":
-                        key = "media_type"
-                    data[key] = type_(content)
-                elif type == "timestamp":
-                    data[key] = utils.parse_xmpp_date(str(elt))
-                else:
-                    raise exceptions.InternalError
-
-        try:
-            algo, hash_ = self._hash.parse_hash_elt(file_metadata_elt)
-        except exceptions.NotFound:
-            pass
-        except exceptions.DataError:
-            from libervia.backend.tools.xml_tools import p_fmt_elt
-
-            log.warning("invalid <hash/> element:\n{p_fmt_elt(file_metadata_elt)}")
-        else:
-            data["file_hash"] = (algo, hash_)
-
-        # TODO: thumbnails
-
-        return data
+        return FileMetadata.from_element(file_metadata_elt)

libervia/backend/plugins/plugin_xep_0447.py

@@ -15,12 +15,28 @@
 # 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/>.
 
+from abc import ABC, abstractmethod
 from collections import namedtuple
 from functools import partial
 import mimetypes
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional, Tuple, Union
+from typing import (
+    Any,
+    Callable,
+    ClassVar,
+    Dict,
+    Final,
+    List,
+    Literal,
+    NamedTuple,
+    Optional,
+    Self,
+    Tuple,
+    Union,
+    cast,
+)
 
+from pydantic import BaseModel, Field
 import treq
 from twisted.internet import defer
 from twisted.words.xish import domish
@@ -30,6 +46,8 @@ from libervia.backend.core.constants import Const as C
 from libervia.backend.core.core_types import SatXMPPEntity
 from libervia.backend.core.i18n import _
 from libervia.backend.core.log import getLogger
+from libervia.backend.plugins.plugin_xep_0103 import URLData, XEP_0103
+from libervia.backend.plugins.plugin_xep_0446 import FileMetadata, XEP_0446
 from libervia.backend.tools import stream
 from libervia.backend.tools.web import treq_client_no_ssl
 
@@ -42,7 +60,13 @@ PLUGIN_INFO = {
     C.PI_TYPE: "XEP",
     C.PI_MODES: C.PLUG_MODE_BOTH,
     C.PI_PROTOCOLS: ["XEP-0447"],
-    C.PI_DEPENDENCIES: ["XEP-0103", "XEP-0334", "XEP-0446", "ATTACH", "DOWNLOAD"],
+    C.PI_DEPENDENCIES: [
+        "XEP-0103",
+        "XEP-0334",
+        "XEP-0446",
+        "ATTACH",
+        "DOWNLOAD",
+    ],
     C.PI_RECOMMENDATIONS: ["XEP-0363"],
     C.PI_MAIN: "XEP_0447",
     C.PI_HANDLER: "no",
@@ -50,7 +74,116 @@ PLUGIN_INFO = {
 }
 
 NS_SFS = "urn:xmpp:sfs:0"
-SourceHandler = namedtuple("SourceHandler", ["callback", "encrypted"])
+
+
+class Source(ABC, BaseModel):
+
+    type: ClassVar[str]
+    encrypted: ClassVar[bool] = False
+
+    def __init_subclass__(cls) -> None:
+        super().__init_subclass__()
+        if not hasattr(cls, "type"):
+            raise TypeError(
+                f'Can\'t instantiate {cls.__name__} without "type" class attribute.'
+            )
+
+    @classmethod
+    @abstractmethod
+    def from_element(cls, element: domish.Element) -> Self:
+        """Parse an element and return corresponding model
+
+        @param element: element to parse
+        @raise exceptions.DataError: the element is invalid
+        """
+
+    @abstractmethod
+    def to_element(self) -> domish.Element:
+        """Convert model to an element
+
+        @return: domish.Element representing the model
+        """
+
+
+class FileSharing(BaseModel):
+    """
+    Model for handling XEP-0447 <file-sharing> element.
+    """
+
+    file: FileMetadata
+    sources: list[Source]
+    disposition: str | None = Field(
+        default=None,
+        description="Disposition of the file, either 'attachment' or 'inline'.",
+    )
+    id: str | None = Field(
+        default=None, description="Unique identifier for the file-sharing element."
+    )
+    _sfs: "XEP_0447 | None" = None
+
+    def to_element(self) -> domish.Element:
+        """Build the <file-sharing> element from this instance's data.
+
+        @return: <file-sharing> element.
+        """
+        file_sharing_elt = domish.Element((NS_SFS, "file-sharing"))
+
+        if self.disposition:
+            file_sharing_elt["disposition"] = self.disposition
+
+        if self.id:
+            file_sharing_elt["id"] = self.id
+
+        file_sharing_elt.addChild(self.file.to_element())
+
+        sources_elt = file_sharing_elt.addElement("sources")
+        for source in self.sources:
+            sources_elt.addChild(source.to_element())
+
+        return file_sharing_elt
+
+    @classmethod
+    def from_element(cls, file_sharing_elt: domish.Element) -> Self:
+        """Create a FileSharing instance from a <file-sharing> element or its parent.
+
+        @param file_sharing_elt: The <file-sharing> element or a parent element.
+        @return: FileSharing instance.
+        @raise exceptions.NotFound: If the <file-sharing> element is not found.
+        """
+        assert cls._sfs is not None
+        if file_sharing_elt.uri != NS_SFS or file_sharing_elt.name != "file-sharing":
+            child_file_sharing_elt = next(
+                file_sharing_elt.elements(NS_SFS, "file-sharing"), None
+            )
+            if child_file_sharing_elt is None:
+                raise exceptions.NotFound("<file-sharing> element not found")
+            else:
+                file_sharing_elt = child_file_sharing_elt
+
+        kwargs = {}
+        disposition = file_sharing_elt.getAttribute("disposition")
+        if disposition:
+            kwargs["disposition"] = disposition
+
+        file_id = file_sharing_elt.getAttribute("id")
+        if file_id:
+            kwargs["id"] = file_id
+
+        kwargs["file"] = FileMetadata.from_element(file_sharing_elt)
+        kwargs["sources"] = cls._sfs.parse_sources_elt(file_sharing_elt)
+
+        return cls(**kwargs)
+
+
+class URLDataSource(URLData, Source):
+    type = "url"
+
+    @classmethod
+    def from_element(cls, element: domish.Element) -> Self:
+        return super().from_element(element)
+
+    def to_element(self) -> domish.Element:
+        return super().to_element()
 
 
 class XEP_0447:
@@ -60,27 +193,34 @@ class XEP_0447:
         self.host = host
         log.info(_("XEP-0447 (Stateless File Sharing) plugin initialization"))
         host.register_namespace("sfs", NS_SFS)
-        self._sources_handlers = {}
-        self._u = host.plugins["XEP-0103"]
+        FileSharing._sfs = self
+        self._sources_handlers: dict[tuple[str, str], type[Source]] = {}
+        self._u = cast(XEP_0103, host.plugins["XEP-0103"])
         self._hints = host.plugins["XEP-0334"]
-        self._m = host.plugins["XEP-0446"]
+        self._m = cast(XEP_0446, host.plugins["XEP-0446"])
         self._http_upload = host.plugins.get("XEP-0363")
         self._attach = host.plugins["ATTACH"]
         self._attach.register(self.can_handle_attachment, self.attach, priority=1000)
-        self.register_source_handler(
-            self._u.namespace, "url-data", self._u.parse_url_data_elt
+        self.register_source(
+            self._u.namespace,
+            "url-data",
+            URLDataSource,
+        )
+        self.register_source(
+            self._jp.namespace,
+            "jinglepub",
+            JinglePubSource,
         )
         host.plugins["DOWNLOAD"].register_download_handler(
             self._u.namespace, self.download
         )
         host.trigger.add("message_received", self._message_received_trigger)
 
-    def register_source_handler(
+    def register_source(
         self,
         namespace: str,
         element_name: str,
-        callback: Callable[[domish.Element], Dict[str, Any]],
-        encrypted: bool = False,
+        source: type[Source],
     ) -> None:
         """Register a handler for file source
 
@@ -97,7 +237,7 @@ class XEP_0447:
                 f"There is already a resource handler for namespace {namespace!r} and "
                 f"name {element_name!r}"
             )
-        self._sources_handlers[key] = SourceHandler(callback, encrypted)
+        self._sources_handlers[key] = source
 
     async def download(
         self,
@@ -193,7 +333,7 @@ class XEP_0447:
         if media_type is None and name:
             media_type = mimetypes.guess_type(name, strict=False)[0]
         file_sharing_elt.addChild(
-            self._m.get_file_metadata_elt(
+            self._m.generate_file_metadata(
                 name=name,
                 media_type=media_type,
                 desc=desc,
@@ -204,19 +344,21 @@ class XEP_0447:
                 height=height,
                 length=length,
                 thumbnail=thumbnail,
-            )
+            ).to_element()
         )
         sources_elt = self.get_sources_elt()
         file_sharing_elt.addChild(sources_elt)
         for source_data in sources:
             if "url" in source_data:
-                sources_elt.addChild(self._u.get_url_data_elt(**source_data))
+                sources_elt.addChild(
+                    self._u.generate_url_data(**source_data).to_element()
+                )
             else:
                 raise NotImplementedError(f"source data not implemented: {source_data}")
 
         return file_sharing_elt
 
-    def parse_sources_elt(self, sources_elt: domish.Element) -> List[Dict[str, Any]]:
+    def parse_sources_elt(self, sources_elt: domish.Element) -> List[Source]:
         """Parse <sources/> element
 
         @param sources_elt: <sources/> element, or a direct parent element
@@ -242,12 +384,8 @@ class XEP_0447:
                 log.warning(f"unmanaged file sharing element: {elt.toXml}")
                 continue
             else:
-                source_data = source_handler.callback(elt)
-                if source_handler.encrypted:
-                    source_data[C.MESS_KEY_ENCRYPTED] = True
-                if "type" not in source_data:
-                    source_data["type"] = elt.uri
-                sources.append(source_data)
+                source = source_handler.from_element(elt)
+                sources.append(source)
         return sources
 
     def parse_file_sharing_elt(self, file_sharing_elt: domish.Element) -> Dict[str, Any]:
@@ -263,7 +401,7 @@ class XEP_0447:
             except StopIteration:
                 raise exceptions.NotFound
         try:
-            data = self._m.parse_file_metadata_elt(file_sharing_elt)
+            data = self._m.parse_file_metadata_elt(file_sharing_elt).model_dump()
         except exceptions.NotFound:
             data = {}
         disposition = file_sharing_elt.getAttribute("disposition")

libervia/backend/plugins/plugin_xep_0448.py

@@ -21,13 +21,14 @@ from functools import partial
 from pathlib import Path
 import secrets
 from textwrap import dedent
-from typing import Any, Dict, Optional, Tuple, Union
+from typing import Any, Dict, Optional, Self, Tuple, Union, cast
 
 from cryptography.exceptions import AlreadyFinalized
 from cryptography.hazmat import backends
 from cryptography.hazmat.primitives import ciphers
 from cryptography.hazmat.primitives.ciphers import CipherContext, modes
 from cryptography.hazmat.primitives.padding import PKCS7, PaddingContext
+from pydantic import BaseModel, ValidationError
 import treq
 from twisted.internet import defer
 from twisted.words.protocols.jabber.xmlstream import XMPPHandler
@@ -40,6 +41,10 @@ from libervia.backend.core.constants import Const as C
 from libervia.backend.core.core_types import SatXMPPEntity
 from libervia.backend.core.i18n import _
 from libervia.backend.core.log import getLogger
+from libervia.backend.plugins.plugin_misc_download import DownloadPlugin
+from libervia.backend.plugins.plugin_xep_0103 import XEP_0103
+from libervia.backend.plugins.plugin_xep_0300 import NS_HASHES, XEP_0300, Hash
+from libervia.backend.plugins.plugin_xep_0447 import XEP_0447, Source
 from libervia.backend.tools import stream
 from libervia.backend.tools.web import treq_client_no_ssl
 
@@ -79,90 +84,121 @@ NS_AES_256_GCM = "urn:xmpp:ciphers:aes-256-gcm-nopadding:0"
 NS_AES_256_CBC = "urn:xmpp:ciphers:aes-256-cbc-pkcs7:0"
 
 
+class EncryptedSource(Source):
+    type = "encrypted"
+    encrypted = True
+    cipher: str
+    key: str
+    iv: str
+    hashes: list[Hash]
+    sources: list[Source]
+    _hash: XEP_0300 | None = None
+    _sfs: XEP_0447 | None = None
+
+    @classmethod
+    def from_element(cls, element: domish.Element) -> Self:
+        """Parse an <encrypted> element and return corresponding EncryptedData model
+
+        @param encrypted_elt: element to parse
+        @raise exceptions.DataError: the element is invalid
+
+        """
+        assert cls._hash is not None, "_hash attribute is not set"
+        assert cls._sfs is not None, "_sfs attribute is not set"
+        try:
+            cipher = element["cipher"]
+            key = str(next(element.elements(NS_ESFS, "key")))
+            iv = str(next(element.elements(NS_ESFS, "iv")))
+        except (KeyError, StopIteration):
+            raise exceptions.DataError(
+                "invalid <encrypted/> element: {encrypted_elt.toXml()}"
+            )
+        sources = cls._sfs.parse_sources_elt(element)
+        if not sources:
+            raise exceptions.DataError(f"Sources are missing in {element.toXml()}")
+
+        if any(isinstance(source, cls) for source in sources):
+            raise exceptions.DataError(
+                f"EncryptedData is used as a source of another EncryptedData"
+            )
+
+        encrypted_data = {
+            "cipher": cipher,
+            "key": key,
+            "iv": iv,
+            "hashes": Hash.from_parent(element),
+            "sources": sources,
+        }
+
+        return cls(**encrypted_data)
+
+    def to_element(self) -> domish.Element:
+        """Convert EncryptedData model to an <encrypted> element
+
+        @return: domish.Element representing the encrypted data
+
+        """
+        assert self._hash is not None, "_hash attribute is not set"
+        encrypted_elt = domish.Element((NS_ESFS, "encrypted"))
+        encrypted_elt["cipher"] = self.cipher
+        encrypted_elt.addElement("key").addContent(self.key)
+        encrypted_elt.addElement("iv").addContent(self.iv)
+        for hash_ in self.hashes:
+            encrypted_elt.addChild(hash_.to_element())
+
+        return encrypted_elt
+
+
 class XEP_0448:
 
     def __init__(self, host):
         self.host = host
         log.info(_("XEP_0448 plugin initialization"))
         host.register_namespace("esfs", NS_ESFS)
-        self._u = host.plugins["XEP-0103"]
-        self._h = host.plugins["XEP-0300"]
+        self._u = cast(XEP_0103, host.plugins["XEP-0103"])
+        self._h = cast(XEP_0300, host.plugins["XEP-0300"])
         self._hints = host.plugins["XEP-0334"]
         self._http_upload = host.plugins["XEP-0363"]
         self._o = host.plugins["XEP-0384"]
-        self._sfs = host.plugins["XEP-0447"]
-        self._sfs.register_source_handler(
-            NS_ESFS, "encrypted", self.parse_encrypted_elt, encrypted=True
-        )
+        self._sfs = cast(XEP_0447, host.plugins["XEP-0447"])
+        self._sfs.register_source(NS_ESFS, "encrypted", EncryptedSource)
         self._attach = host.plugins["ATTACH"]
         self._attach.register(
             self.can_handle_attachment, self.attach, encrypted=True, priority=1000
         )
-        host.plugins["DOWNLOAD"].register_download_handler(NS_ESFS, self.download)
+        EncryptedSource._hash = self._h
+        EncryptedSource._sfs = self._sfs
+        download = cast(DownloadPlugin, host.plugins["DOWNLOAD"])
+        download.register_download_handler(NS_ESFS, self.download)
         host.trigger.add("XEP-0363_upload_pre_slot", self._upload_pre_slot)
         host.trigger.add("XEP-0363_upload", self._upload_trigger)
 
     def get_handler(self, client):
         return XEP0448Handler()
 
-    def parse_encrypted_elt(self, encrypted_elt: domish.Element) -> Dict[str, Any]:
-        """Parse an <encrypted> element and return corresponding source data
-
-        @param encrypted_elt: element to parse
-        @raise exceptions.DataError: the element is invalid
-
-        """
-        sources = self._sfs.parse_sources_elt(encrypted_elt)
-        if not sources:
-            raise exceptions.NotFound("sources are missing in {encrypted_elt.toXml()}")
-        if len(sources) > 1:
-            log.debug(
-                "more that one sources has been found, this is not expected, only the "
-                "first one will be used"
-            )
-        source = sources[0]
-        source["type"] = NS_ESFS
-        try:
-            encrypted_data = source["encrypted_data"] = {
-                "cipher": encrypted_elt["cipher"],
-                "key": str(next(encrypted_elt.elements(NS_ESFS, "key"))),
-                "iv": str(next(encrypted_elt.elements(NS_ESFS, "iv"))),
-            }
-        except (KeyError, StopIteration):
-            raise exceptions.DataError(
-                "invalid <encrypted/> element: {encrypted_elt.toXml()}"
-            )
-        try:
-            hash_algo, hash_value = self._h.parse_hash_elt(encrypted_elt)
-        except exceptions.NotFound:
-            pass
-        else:
-            encrypted_data["hash_algo"] = hash_algo
-            encrypted_data["hash"] = base64.b64encode(hash_value.encode()).decode()
-        return source
-
     async def download(
         self,
         client: SatXMPPEntity,
-        attachment: Dict[str, Any],
-        source: Dict[str, Any],
+        attachment: dict[str, Any],
+        source: dict[str, Any],
         dest_path: Union[Path, str],
-        extra: Optional[Dict[str, Any]] = None,
-    ) -> Tuple[str, defer.Deferred]:
+        extra: dict[str, Any] | None = None,
+    ) -> tuple[str, defer.Deferred]:
         # TODO: check hash
         if extra is None:
             extra = {}
+        assert source["type"] == "encrypted"
         try:
-            encrypted_data = source["encrypted_data"]
-            cipher = encrypted_data["cipher"]
-            iv = base64.b64decode(encrypted_data["iv"])
-            key = base64.b64decode(encrypted_data["key"])
+            cipher = source["cipher"]
+            iv = base64.b64decode(source["iv"])
+            key = base64.b64decode(source["key"])
         except KeyError as e:
-            raise ValueError(f"{source} has incomplete encryption data: {e}")
+            raise ValueError(f"{source} has incomplete encryption data: {e}") from e
+
         try:
-            download_url = source["url"]
-        except KeyError:
-            raise ValueError(f"{source} has missing URL")
+            download_url = source["sources"][0]["url"]
+        except (IndexError, KeyError) as e:
+            raise ValueError(f"{source} has missing URL") from e
 
         if extra.get("ignore_tls_errors", False):
             log.warning("TLS certificate check disabled, this is highly insecure")
@@ -294,9 +330,9 @@ class XEP_0448:
                     size=attachment["size"],
                     file_hash=file_hash,
                 )
-                encrypted_elt = file_sharing_elt.sources.addElement(
-                    (NS_ESFS, "encrypted")
-                )
+                sources_elt = file_sharing_elt.sources
+                assert sources_elt is not None
+                encrypted_elt = sources_elt.addElement((NS_ESFS, "encrypted"))
                 encrypted_elt["cipher"] = NS_AES_256_GCM
                 encrypted_elt.addElement(
                     "key", content=base64.b64encode(encryption_data["key"]).decode()
@@ -311,7 +347,7 @@ class XEP_0448:
                 )
                 encrypted_elt.addChild(
                     self._sfs.get_sources_elt(
-                        [self._u.get_url_data_elt(attachment["url"])]
+                        [self._u.generate_url_data(attachment["url"]).to_element()]
                     )
                 )
                 data["xml"].addChild(file_sharing_elt)