# HG changeset patch # User Goffi # Date 1685462360 -7200 # Node ID 08ee0e623e7e8613905ae848431c0a2b22e7a239 # Parent e12936318177363044eedc15097712af8d1821d7 plugin XEP-0338: "Jingle Grouping Framework" implementation: rel 440 diff -r e12936318177 -r 08ee0e623e7e sat/plugins/plugin_xep_0338.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/sat/plugins/plugin_xep_0338.py Tue May 30 17:59:20 2023 +0200 @@ -0,0 +1,156 @@ +#!/usr/bin/env python3 + +# Libervia plugin +# Copyright (C) 2009-2023 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 typing import List + +from twisted.words.protocols.jabber.xmlstream import XMPPHandler +from twisted.words.xish import domish +from wokkel import disco, iwokkel +from zope.interface import implementer + +from sat.core.constants import Const as C +from sat.core.i18n import _ +from sat.core.log import getLogger +from sat.core.core_types import SatXMPPEntity + +log = getLogger(__name__) + +NS_JINGLE_GROUPING = "urn:xmpp:jingle:apps:grouping:0" +NS_RFC_5888 = "urn:ietf:rfc:5888" + +PLUGIN_INFO = { + C.PI_NAME: "Jingle Grouping Framework", + C.PI_IMPORT_NAME: "XEP-0338", + C.PI_TYPE: "XEP", + C.PI_MODES: C.PLUG_MODE_BOTH, + C.PI_PROTOCOLS: ["XEP-0338"], + C.PI_DEPENDENCIES: ["XEP-0166", "XEP-0167"], + C.PI_RECOMMENDATIONS: [], + C.PI_MAIN: "XEP_0338", + C.PI_HANDLER: "yes", + C.PI_DESCRIPTION: _("""Jingle mapping of RFC 5888 SDP Grouping Framework"""), +} + + +class XEP_0338: + def __init__(self, host): + log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization") + self._j = host.plugins["XEP-0166"] + host.trigger.add("XEP-0167_parse_sdp_a", self._parse_sdp_a_trigger) + host.trigger.add( + "XEP-0167_generate_sdp_session", self._generate_sdp_session_trigger + ) + host.trigger.add("XEP-0167_jingle_session_init", self._jingle_session_init_trigger) + host.trigger.add("XEP-0167_jingle_handler", self._jingle_handler_trigger) + + def get_handler(self, client): + return XEP_0338_handler() + + def _parse_sdp_a_trigger( + self, + attribute: str, + parts: List[str], + call_data: dict, + metadata: dict, + media_type: str, + application_data: dict, + transport_data: dict, + ) -> None: + """Parse "group" attributes""" + if attribute == "group": + semantics = parts[0] + content_names = parts[1:] + metadata.setdefault("group", {})[semantics] = content_names + + def _generate_sdp_session_trigger( + self, + session: dict, + local: bool, + sdp_lines: List[str], + ) -> None: + """Generate "group" attributes""" + key = "metadata" if local else "peer_metadata" + group_data = session[key].get("group", {}) + + for semantics, content_names in group_data.items(): + sdp_lines.append(f"a=group:{semantics} {' '.join(content_names)}") + + def parse_group_element( + self, jingle_elt: domish.Element, session: dict + ) -> None: + """Parse the and elements""" + for group_elt in jingle_elt.elements(NS_JINGLE_GROUPING, "group"): + try: + metadata = session["peer_metadata"] + semantics = group_elt["semantics"] + group_content = metadata.setdefault("group", {})[semantics] = [] + for content_elt in group_elt.elements(NS_JINGLE_GROUPING, "content"): + group_content.append(content_elt["name"]) + except KeyError as e: + log.warning(f"Error while parsing : {e}\n{group_elt.toXml()}") + + def add_group_element( + self, jingle_elt: domish.Element, session: dict + ) -> None: + """Build the and elements if possible""" + for semantics, content_names in session["metadata"].get("group", {}).items(): + group_elt = jingle_elt.addElement((NS_JINGLE_GROUPING, "group")) + group_elt["semantics"] = semantics + for content_name in content_names: + content_elt = group_elt.addElement((NS_JINGLE_GROUPING, "content")) + content_elt["name"] = content_name + + def _jingle_session_init_trigger( + self, + client: SatXMPPEntity, + session: dict, + content_name: str, + media: str, + media_data: dict, + desc_elt: domish.Element, + ) -> None: + jingle_elt = session["jingle_elt"] + self.add_group_element(jingle_elt, session) + + def _jingle_handler_trigger( + self, + client: SatXMPPEntity, + action: str, + session: dict, + content_name: str, + desc_elt: domish.Element, + ) -> None: + # this is a session metadata, so we only generate it on the first content + if content_name == next(iter(session["contents"])) and action in ( + self._j.A_PREPARE_CONFIRMATION, + self._j.A_SESSION_INITIATE, + self._j.A_PREPARE_INITIATOR, + ): + jingle_elt = session["jingle_elt"] + self.parse_group_element(jingle_elt, session) + if action == self._j.A_SESSION_INITIATE: + self.add_group_element(jingle_elt, session) + + +@implementer(iwokkel.IDisco) +class XEP_0338_handler(XMPPHandler): + def getDiscoInfo(self, requestor, target, nodeIdentifier=""): + return [disco.DiscoFeature(NS_RFC_5888)] + + def getDiscoItems(self, requestor, target, nodeIdentifier=""): + return []