diff --git a/sat/plugins/plugin_xep_0339.py b/sat/plugins/plugin_xep_0339.py new file mode 100644 index 00000000..37f54f2a --- /dev/null +++ b/sat/plugins/plugin_xep_0339.py @@ -0,0 +1,199 @@ +#!/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 import exceptions +from sat.core.constants import Const as C +from sat.core.core_types import SatXMPPEntity +from sat.core.i18n import _ +from sat.core.log import getLogger +from sat.tools import xml_tools + +log = getLogger(__name__) + +NS_JINGLE_RTP_SSMA = "urn:xmpp:jingle:apps:rtp:ssma:0" + +PLUGIN_INFO = { + C.PI_NAME: "Source-Specific Media Attributes in Jingle", + C.PI_IMPORT_NAME: "XEP-0339", + C.PI_TYPE: "XEP", + C.PI_MODES: C.PLUG_MODE_BOTH, + C.PI_PROTOCOLS: ["XEP-0339"], + C.PI_DEPENDENCIES: ["XEP-0092", "XEP-0167"], + C.PI_RECOMMENDATIONS: [], + C.PI_MAIN: "XEP_0339", + C.PI_HANDLER: "yes", + C.PI_DESCRIPTION: _("""Source-Specific Media Attributes in Jingle"""), +} + + +class XEP_0339: + def __init__(self, host): + log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization") + self.host = host + host.trigger.add("XEP-0167_parse_sdp_a", self._parse_sdp_a_trigger) + host.trigger.add( + "XEP-0167_generate_sdp_content", self._generate_sdp_content_trigger + ) + host.trigger.add("XEP-0167_parse_description", self._parse_description_trigger) + host.trigger.add("XEP-0167_build_description", self._build_description_trigger) + + def get_handler(self, client): + return XEP_0339_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 "ssrc" attributes""" + if attribute == "ssrc": + assert application_data is not None + ssrc_id = int(parts[0]) + + if len(parts) > 1: + name, *values = " ".join(parts[1:]).split(":", 1) + if values: + value = values[0] or None + else: + value = None + application_data.setdefault("ssrc", {}).setdefault(ssrc_id, {})[ + name + ] = value + else: + log.warning(f"no attribute in ssrc: {' '.join(parts)}") + application_data.setdefault("ssrc", {}).setdefault(ssrc_id, {}) + elif attribute == "ssrc-group": + assert application_data is not None + semantics, *ssrc_ids = parts + ssrc_ids = [int(ssrc_id) for ssrc_id in ssrc_ids] + application_data.setdefault("ssrc-group", {})[semantics] = ssrc_ids + elif attribute == "msid": + assert application_data is not None + application_data["msid"] = " ".join(parts) + + + def _generate_sdp_content_trigger( + self, + session: dict, + local: bool, + idx: int, + content_data: dict, + sdp_lines: List[str], + application_data: dict, + app_data_key: str, + media_data: dict, + media: str + ) -> None: + """Generate "msid" and "ssrc" attributes""" + if "msid" in media_data: + sdp_lines.append(f"a=msid:{media_data['msid']}") + + ssrc_data = media_data.get("ssrc", {}) + ssrc_group_data = media_data.get("ssrc-group", {}) + + for ssrc_id, attributes in ssrc_data.items(): + if not attributes: + # there are no attributes for this SSRC ID, we add a simple line with only + # the SSRC ID + sdp_lines.append(f"a=ssrc:{ssrc_id}") + else: + for attr_name, attr_value in attributes.items(): + if attr_value is not None: + sdp_lines.append(f"a=ssrc:{ssrc_id} {attr_name}:{attr_value}") + else: + sdp_lines.append(f"a=ssrc:{ssrc_id} {attr_name}") + for semantics, ssrc_ids in ssrc_group_data.items(): + ssrc_lines = " ".join(str(ssrc_id) for ssrc_id in ssrc_ids) + sdp_lines.append(f"a=ssrc-group:{semantics} {ssrc_lines}") + + def _parse_description_trigger( + self, desc_elt: domish.Element, media_data: dict + ) -> bool: + """Parse the and elements""" + for source_elt in desc_elt.elements(NS_JINGLE_RTP_SSMA, "source"): + try: + ssrc_id = int(source_elt["ssrc"]) + media_data.setdefault("ssrc", {})[ssrc_id] = {} + for param_elt in source_elt.elements(NS_JINGLE_RTP_SSMA, "parameter"): + name = param_elt["name"] + value = param_elt.getAttribute("value") + media_data["ssrc"][ssrc_id][name] = value + if name == "msid" and "msid" not in media_data: + media_data["msid"] = value + except (KeyError, ValueError) as e: + log.warning(f"Error while parsing : {e}\n{source_elt.toXml()}") + + for ssrc_group_elt in desc_elt.elements(NS_JINGLE_RTP_SSMA, "ssrc-group"): + try: + semantics = ssrc_group_elt["semantics"] + semantic_ids = media_data.setdefault("ssrc-group", {})[semantics] = [] + for source_elt in ssrc_group_elt.elements(NS_JINGLE_RTP_SSMA, "source"): + semantic_ids.append( + int(source_elt["ssrc"]) + ) + except (KeyError, ValueError) as e: + log.warning( + f"Error while parsing : {e}\n{ssrc_group_elt.toXml()}" + ) + + return True + + def _build_description_trigger( + self, desc_elt: domish.Element, media_data: dict, session: dict + ) -> bool: + """Build the and elements if possible""" + for ssrc_id, parameters in media_data.get("ssrc", {}).items(): + if "msid" not in parameters and "msid" in media_data: + parameters["msid"] = media_data["msid"] + source_elt = desc_elt.addElement((NS_JINGLE_RTP_SSMA, "source")) + source_elt["ssrc"] = str(ssrc_id) + for name, value in parameters.items(): + param_elt = source_elt.addElement((NS_JINGLE_RTP_SSMA, "parameter")) + param_elt["name"] = name + if value is not None: + param_elt["value"] = value + + for semantics, ssrc_ids in media_data.get("ssrc-group", {}).items(): + ssrc_group_elt = desc_elt.addElement((NS_JINGLE_RTP_SSMA, "ssrc-group")) + ssrc_group_elt["semantics"] = semantics + for ssrc_id in ssrc_ids: + source_elt = ssrc_group_elt.addElement((NS_JINGLE_RTP_SSMA, "source")) + source_elt["ssrc"] = str(ssrc_id) + + return True + + +@implementer(iwokkel.IDisco) +class XEP_0339_handler(XMPPHandler): + def getDiscoInfo(self, requestor, target, nodeIdentifier=""): + return [disco.DiscoFeature(NS_JINGLE_RTP_SSMA)] + + def getDiscoItems(self, requestor, target, nodeIdentifier=""): + return []