From 5e272710f36fac100ea318a4b1d455630cf27d03 Mon Sep 17 00:00:00 2001 From: Goffi Date: Sat, 4 Jul 2026 16:45:48 +0200 Subject: [PATCH] plugin XEP-0066: "Out of Band Data" implementation --- libervia/backend/plugins/plugin_xep_0066.py | 175 ++++++++++++++++++++ 1 file changed, 175 insertions(+) create mode 100644 libervia/backend/plugins/plugin_xep_0066.py diff --git a/libervia/backend/plugins/plugin_xep_0066.py b/libervia/backend/plugins/plugin_xep_0066.py new file mode 100644 index 00000000..47324b24 --- /dev/null +++ b/libervia/backend/plugins/plugin_xep_0066.py @@ -0,0 +1,175 @@ +#!/usr/bin/env python3 + +# Copyright (C) 2009-2026 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 . + +from collections.abc import Iterator +from typing import Any, Final, Self + +from pydantic import BaseModel, Field, RootModel +from twisted.internet import defer +from twisted.words.protocols.jabber import xmlstream +from twisted.words.xish import domish +from tx_xmpp import disco, itx_xmpp +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 + +log = getLogger(__name__) + +EXTRA_KEY: Final = "oob" + + +PLUGIN_INFO = { + C.PI_NAME: "Out of Band Data", + C.PI_IMPORT_NAME: "XEP-0066", + C.PI_TYPE: "XEP", + C.PI_MODES: C.PLUG_MODE_BOTH, + C.PI_PROTOCOLS: ["XEP-0066"], + C.PI_MAIN: "XEP_0066", + C.PI_HANDLER: "yes", + C.PI_DESCRIPTION: _("""Implementation of XEP-0066 (Out of Band Data)"""), +} + +NS_OOB = "jabber:x:oob" + + +class OOB(BaseModel): + """OOB data""" + + url: str = Field(description="URI of the resource.") + desc: str | None = Field( + default=None, + description="Description of the resource.", + ) + + def to_element(self) -> domish.Element: + """Build the ```` element from this instance's data. + + @return: OOB element. + """ + x_elt = domish.Element((NS_OOB, "x")) + url_elt = x_elt.addElement("url") + url_elt.addContent(self.url) + if self.desc is not None: + desc_elt = x_elt.addElement("desc") + desc_elt.addContent(self.desc) + return x_elt + + @classmethod + def from_element(cls, element: domish.Element) -> Self: + """Create an OOB instance from an element. + + @param element: OOB element. + @return: OOB instance. + @raise exceptions.DataError: The element is missing. + """ + url_elt = next(element.elements(NS_OOB, "url"), None) + if url_elt is None: + raise exceptions.DataError(" element is missing") + + desc_elt = next(element.elements(NS_OOB, "desc"), None) + return cls(url=str(url_elt), desc=str(desc_elt) if desc_elt is not None else None) + + +class OOBList(RootModel): + """List of OOB data.""" + + root: list[OOB] + + def __iter__(self) -> Iterator[OOB]: # type: ignore[override] + return iter(self.root) + + def __getitem__(self, item: int) -> OOB: + return self.root[item] + + def __len__(self) -> int: + return len(self.root) + + def append(self, item: OOB) -> None: + self.root.append(item) + + @classmethod + def from_element(cls, parent_elt: domish.Element) -> Self: + """Create an OOBList instance from a parent element. + + @param parent_elt: Parent element containing OOB children. + @return: OOBList instance. + """ + return cls([OOB.from_element(elt) for elt in parent_elt.elements(NS_OOB, "x")]) + + +class XEP_0066: + + def __init__(self, host): + log.info(f"Plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization.") + host.register_namespace("oob", NS_OOB) + host.trigger.add("message_received", self._message_received_trigger) + + def generate_oob(self, url: str, desc: str | None = None) -> domish.Element: + """Generate an OOB element. + + @param url: URI of the resource. + @param desc: Optional description of the resource. + @return: ```` element. + """ + return OOB(url=url, desc=desc).to_element() + + def _message_received_trigger( + self, + client: SatXMPPEntity, + message_elt: domish.Element, + post_treat: defer.Deferred, + ) -> bool: + """Check presence of OOB in message. + + @param client: client instance. + @param message_elt: the element. + @param post_treat: deferred to add post-treatments to. + @return: True to continue message processing. + """ + oob_list = OOBList.from_element(message_elt) + if oob_list: + post_treat.addCallback(self._add_oob, oob_list) + return True + + def _add_oob(self, data: dict[str, Any], oob_list: OOBList) -> dict[str, Any]: + """Add OOB data to message extra. + + @param data: message data dict. + @param oob_list: parsed OOB list. + @return: updated message data dict. + """ + data["extra"][EXTRA_KEY] = oob_list.model_dump() + return data + + def get_handler(self, client: SatXMPPEntity) -> "XEP_0066Handler": + return XEP_0066Handler() + + +@implementer(itx_xmpp.IDisco) +class XEP_0066Handler(xmlstream.XMPPHandler): + + def getDiscoInfo(self, requestor: str, target: str, nodeIdentifier: str = "") -> list: + return [disco.DiscoFeature(NS_OOB)] + + def getDiscoItems( + self, requestor: str, target: str, nodeIdentifier: str = "" + ) -> list: + return []