Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_xep_0294.py @ 4071:4b842c1fb686

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
refactoring: renamed `sat` package to `libervia.backend`
author Goffi <goffi@goffi.org>
date Fri, 02 Jun 2023 11:49:51 +0200
parents sat/plugins/plugin_xep_0294.py@18719058a914
children
comparison
equal deleted inserted replaced
4070:d10748475025 4071:4b842c1fb686
1 #!/usr/bin/env python3
2
3 # Libervia plugin
4 # Copyright (C) 2009-2023 Jérôme Poisson (goffi@goffi.org)
5
6 # This program is free software: you can redistribute it and/or modify
7 # it under the terms of the GNU Affero General Public License as published by
8 # the Free Software Foundation, either version 3 of the License, or
9 # (at your option) any later version.
10
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 # GNU Affero General Public License for more details.
15
16 # You should have received a copy of the GNU Affero General Public License
17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
18
19 from typing import Dict, List, Optional, Union
20
21 from twisted.words.protocols.jabber.xmlstream import XMPPHandler
22 from twisted.words.xish import domish
23 from wokkel import disco, iwokkel
24 from zope.interface import implementer
25
26 from libervia.backend.core import exceptions
27 from libervia.backend.core.constants import Const as C
28 from libervia.backend.core.i18n import _
29 from libervia.backend.core.log import getLogger
30
31 log = getLogger(__name__)
32
33 NS_JINGLE_RTP_HDREXT = "urn:xmpp:jingle:apps:rtp:rtp-hdrext:0"
34
35 PLUGIN_INFO = {
36 C.PI_NAME: "Jingle RTP Header Extensions Negotiation",
37 C.PI_IMPORT_NAME: "XEP-0294",
38 C.PI_TYPE: "XEP",
39 C.PI_MODES: C.PLUG_MODE_BOTH,
40 C.PI_PROTOCOLS: ["XEP-0294"],
41 C.PI_DEPENDENCIES: ["XEP-0167"],
42 C.PI_RECOMMENDATIONS: [],
43 C.PI_MAIN: "XEP_0294",
44 C.PI_HANDLER: "yes",
45 C.PI_DESCRIPTION: _("""Jingle RTP Header Extensions Negotiation"""),
46 }
47
48
49 class XEP_0294:
50 def __init__(self, host):
51 log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization")
52 host.trigger.add("XEP-0167_parse_sdp_a", self._parse_sdp_a_trigger)
53 host.trigger.add(
54 "XEP-0167_generate_sdp_content", self._generate_sdp_content_trigger
55 )
56 host.trigger.add("XEP-0167_parse_description", self._parse_description_trigger)
57 host.trigger.add("XEP-0167_build_description", self._build_description_trigger)
58
59 def get_handler(self, client):
60 return XEP_0294_handler()
61
62 def _parse_extmap(self, parts: List[str], application_data: dict) -> None:
63 """Parse an individual extmap line and fill application_data accordingly"""
64 if "/" in parts[0]:
65 id_, direction = parts[0].split("/", 1)
66 else:
67 id_ = parts[0]
68 direction = None
69 uri = parts[1]
70 attributes = parts[2:]
71
72 if direction in (None, "sendrecv"):
73 senders = "both"
74 elif direction == "sendonly":
75 senders = "initiator"
76 elif direction == "recvonly":
77 senders = "responder"
78 elif direction == "inactive":
79 senders = "none"
80 else:
81 log.warning(f"invalid direction for extmap: {direction}")
82 senders = "sendrecv"
83
84 rtp_hdr_ext_data: Dict[str, Union[str, dict]] = {
85 "id": id_,
86 "uri": uri,
87 "senders": senders,
88 }
89
90 if attributes:
91 parameters = {}
92 for attribute in attributes:
93 name, *value = attribute.split("=", 1)
94 parameters[name] = value[0] if value else None
95 rtp_hdr_ext_data["parameters"] = parameters
96
97 application_data.setdefault("rtp-hdrext", {})[id_] = rtp_hdr_ext_data
98
99 def _parse_sdp_a_trigger(
100 self,
101 attribute: str,
102 parts: List[str],
103 call_data: dict,
104 metadata: dict,
105 media_type: str,
106 application_data: Optional[dict],
107 transport_data: dict,
108 ) -> None:
109 """Parse "extmap" and "extmap-allow-mixed" attributes"""
110 if attribute == "extmap":
111 if application_data is None:
112 call_data.setdefault("_extmaps", []).append(parts)
113 else:
114 self._parse_extmap(parts, application_data)
115 elif attribute == "extmap-allow-mixed":
116 if application_data is None:
117 call_data["_extmap-allow-mixed"] = True
118 else:
119 application_data["extmap-allow-mixed"] = True
120 elif (
121 application_data is not None
122 and "_extmaps" in call_data
123 and "rtp-hdrext" not in application_data
124 ):
125 extmaps = call_data.pop("_extmaps")
126 for parts in extmaps:
127 self._parse_extmap(parts, application_data)
128 elif (
129 application_data is not None
130 and "_extmap-allow-mixed" in call_data
131 and "extmap-allow-mixed" not in application_data
132 ):
133 value = call_data.pop("_extmap-allow-mixed")
134 application_data["extmap-allow-mixed"] = value
135
136 def _generate_sdp_content_trigger(
137 self,
138 session: dict,
139 local: bool,
140 idx: int,
141 content_data: dict,
142 sdp_lines: List[str],
143 application_data: dict,
144 app_data_key: str,
145 media_data: dict,
146 media: str,
147 ) -> None:
148 """Generate "extmap" and "extmap-allow-mixed" attributes"""
149 rtp_hdrext_dict = media_data.get("rtp-hdrext", {})
150
151 for id_, ext_data in rtp_hdrext_dict.items():
152 senders = ext_data.get("senders")
153 if senders in (None, "both"):
154 direction = "sendrecv"
155 elif senders == "initiator":
156 direction = "sendonly"
157 elif senders == "responder":
158 direction = "recvonly"
159 elif senders == "none":
160 direction = "inactive"
161 else:
162 raise exceptions.InternalError(
163 f"Invalid senders value for extmap: {ext_data.get('senders')}"
164 )
165
166 parameters_str = ""
167 if "parameters" in ext_data:
168 parameters_str = " " + " ".join(
169 f"{k}={v}" if v is not None else f"{k}"
170 for k, v in ext_data["parameters"].items()
171 )
172
173 sdp_lines.append(
174 f"a=extmap:{id_}/{direction} {ext_data['uri']}{parameters_str}"
175 )
176
177 if media_data.get("extmap-allow-mixed", False):
178 sdp_lines.append("a=extmap-allow-mixed")
179
180 def _parse_description_trigger(
181 self, desc_elt: domish.Element, media_data: dict
182 ) -> None:
183 """Parse the <rtp-hdrext> and <extmap-allow-mixed> elements"""
184 for rtp_hdrext_elt in desc_elt.elements(NS_JINGLE_RTP_HDREXT, "rtp-hdrext"):
185 id_ = rtp_hdrext_elt["id"]
186 uri = rtp_hdrext_elt["uri"]
187 senders = rtp_hdrext_elt.getAttribute("senders", "both")
188 # FIXME: workaround for Movim bug https://github.com/movim/movim/issues/1212
189 if senders in ("sendonly", "recvonly", "sendrecv", "inactive"):
190 log.warning("Movim bug workaround for wrong extmap value")
191 if senders == "sendonly":
192 senders = "initiator"
193 elif senders == "recvonly":
194 senders = "responder"
195 elif senders == "sendrecv":
196 senders = "both"
197 else:
198 senders = "none"
199
200 media_data.setdefault("rtp-hdrext", {})[id_] = {
201 "id": id_,
202 "uri": uri,
203 "senders": senders,
204 }
205
206 parameters = {}
207 for param_elt in rtp_hdrext_elt.elements(NS_JINGLE_RTP_HDREXT, "parameter"):
208 try:
209 parameters[param_elt["name"]] = param_elt.getAttribute("value")
210 except KeyError:
211 log.warning(f"invalid parameters (missing name): {param_elt.toXml()}")
212
213 if parameters:
214 media_data["rtp-hdrext"][id_]["parameters"] = parameters
215
216 try:
217 next(desc_elt.elements(NS_JINGLE_RTP_HDREXT, "extmap-allow-mixed"))
218 except StopIteration:
219 pass
220 else:
221 media_data["extmap-allow-mixed"] = True
222
223 def _build_description_trigger(
224 self, desc_elt: domish.Element, media_data: dict, session: dict
225 ) -> None:
226 """Build the <rtp-hdrext> and <extmap-allow-mixed> elements if possible"""
227 for id_, hdrext_data in media_data.get("rtp-hdrext", {}).items():
228 rtp_hdrext_elt = desc_elt.addElement((NS_JINGLE_RTP_HDREXT, "rtp-hdrext"))
229 rtp_hdrext_elt["id"] = id_
230 rtp_hdrext_elt["uri"] = hdrext_data["uri"]
231 senders = hdrext_data.get("senders", "both")
232 if senders != "both":
233 # we must not set "both" senders otherwise calls will fail with Movim due
234 # to https://github.com/movim/movim/issues/1213
235 rtp_hdrext_elt["senders"] = senders
236
237 for name, value in hdrext_data.get("parameters", {}).items():
238 param_elt = rtp_hdrext_elt.addElement((NS_JINGLE_RTP_HDREXT, "parameter"))
239 param_elt["name"] = name
240 if value is not None:
241 param_elt["value"] = value
242
243 if media_data.get("extmap-allow-mixed", False):
244 desc_elt.addElement((NS_JINGLE_RTP_HDREXT, "extmap-allow-mixed"))
245
246
247 @implementer(iwokkel.IDisco)
248 class XEP_0294_handler(XMPPHandler):
249 def getDiscoInfo(self, requestor, target, nodeIdentifier=""):
250 return [disco.DiscoFeature(NS_JINGLE_RTP_HDREXT)]
251
252 def getDiscoItems(self, requestor, target, nodeIdentifier=""):
253 return []