plugin XEP-0482: Call Invites implementation

libervia/backend/plugins/plugin_xep_0482.py

@@ -0,0 +1,134 @@
+#!/usr/bin/env python3
+
+# Libervia plugin for call invites (XEP-0482)
+# 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 <http://www.gnu.org/licenses/>.
+
+import uuid
+
+from twisted.words.protocols.jabber import jid
+from twisted.words.xish import domish
+
+from libervia.backend import G
+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__)
+
+NS_CALL_INVITES = "urn:xmpp:call-invites:0"
+
+PLUGIN_INFO = {
+    C.PI_NAME: "Call Invites",
+    C.PI_IMPORT_NAME: "XEP-0482",
+    C.PI_TYPE: "XEP",
+    C.PI_PROTOCOLS: ["XEP-0482"],
+    C.PI_DEPENDENCIES: [],
+    C.PI_MAIN: "XEP_0482",
+    C.PI_HANDLER: "no",
+    C.PI_DESCRIPTION: _("Invitation handling for calls."),
+}
+
+
+class XEP_0482:
+
+    def __init__(self, host):
+        log.info(f'Plugin "{PLUGIN_INFO[C.PI_NAME]}" initialization')
+        host.register_namespace("call-invites", NS_CALL_INVITES)
+        host.trigger.add("message_received", self._message_received_trigger)
+
+    def _build_reply(self, name: str, id_s: str) -> tuple[domish.Element, domish.Element]:
+        """Build a <accept> or <reject> message
+
+        @param name: name of the call-invites element to add
+        @param id_s: id of the invite message
+        @return: the message element and the answer element
+        """
+        message_elt = domish.Element((None, "message"))
+        message_elt["type"] = "chat"
+        answer_elt = message_elt.addElement((NS_CALL_INVITES, name))
+        answer_elt["id"] = id_s
+        return message_elt, answer_elt
+
+    def invite(
+        self,
+        client: SatXMPPEntity,
+        to_jid: jid.JID,
+        method_elts: list[domish.Element],
+    ) -> None:
+        """Send a call invite to a given JID
+
+        @param client: The client instance.
+        @param to_jid: JID of the user to invite
+        @param method_elts: elements describing the ways to join the call, they are added
+            as children of the <invite> element
+        """
+        message_elt = domish.Element((None, "message"))
+        message_elt["to"] = to_jid.full()
+        message_elt["type"] = "chat"
+        message_id_s = str(uuid.uuid4())
+        message_elt["id"] = message_id_s
+        invite_elt = message_elt.addElement((NS_CALL_INVITES, "invite"))
+        # FIXME: workaround for Dino bug (https://github.com/dino/dino/issues/1902)
+        invite_elt["id"] = message_id_s
+        for method_elt in method_elts:
+            invite_elt.addChild(method_elt)
+        client.send(message_elt)
+
+    def accept(
+        self,
+        client: SatXMPPEntity,
+        to_jid: jid.JID,
+        invite_id_s: str,
+        method_elt: domish.Element,
+    ) -> None:
+        """Accept a call invite
+
+        @param client: The client instance.
+        @param to_jid: JID of the user who sent the invite
+        @param invite_id_s: id of the invite message
+        @param method_elt: element describing the connection method used to join, as
+            defined by the specification of this method
+        """
+        message_elt, answer_elt = self._build_reply("accept", invite_id_s)
+        message_elt["to"] = to_jid.full()
+        answer_elt.addChild(method_elt)
+        client.send(message_elt)
+
+    def reject(self, client: SatXMPPEntity, to_jid: jid.JID, invite_id_s: str) -> None:
+        """Reject a call invite
+
+        @param client: The client instance.
+        @param to_jid: JID of the user who sent the invite
+        @param invite_id_s: id of the invite message
+        """
+        message_elt, _ = self._build_reply("reject", invite_id_s)
+        message_elt["to"] = to_jid.full()
+        client.send(message_elt)
+
+    def _message_received_trigger(self, client, message_elt, post_treat) -> bool:
+        """Check if a call invite is in the message"""
+        invite_elt = next(message_elt.elements(NS_CALL_INVITES, "invite"), None)
+        if invite_elt is None:
+            return True
+        log.info(f"Call invite received from {message_elt['from']} [{client.profile}]")
+        if G.host.trigger.point(
+            "XEP-0482_invite_received", client, message_elt, invite_elt
+        ):
+            log.warning(
+                f"Call invite received, but no plugin handled it: {message_elt.toXml()}"
+            )
+        return False