Mercurial > libervia-backend

comparison sat/plugins/plugin_xep_0176.py @ 4045:ae756bf7c3e8

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
plugin XEP-0176: Jingle ICE-UDP Transport Method implementation: rel 419
author Goffi <goffi@goffi.org>
date Mon, 15 May 2023 16:23:36 +0200
parents
children 2ced30f6d5de
comparison
equal deleted inserted replaced
4044:3900626bc100 4045:ae756bf7c3e8
1 #!/usr/bin/env python3
2
3 # Libervia plugin for Jingle (XEP-0176)
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
20 import uuid
21
22 from twisted.internet import defer
23 from twisted.words.protocols.jabber.xmlstream import XMPPHandler
24 from twisted.words.xish import domish
25 from wokkel import disco, iwokkel
26 from zope.interface import implementer
27
28 from sat.core import exceptions
29 from sat.core.constants import Const as C
30 from sat.core.core_types import SatXMPPEntity
31 from sat.core.i18n import _
32 from sat.core.log import getLogger
33 from sat.tools.common import data_format
34
35 from .plugin_xep_0166 import BaseTransportHandler
36
37 log = getLogger(__name__)
38
39 NS_JINGLE_ICE_UDP= "urn:xmpp:jingle:transports:ice-udp:1"
40
41 PLUGIN_INFO = {
42 C.PI_NAME: "Jingle ICE-UDP Transport Method",
43 C.PI_IMPORT_NAME: "XEP-0176",
44 C.PI_TYPE: "XEP",
45 C.PI_MODES: C.PLUG_MODE_BOTH,
46 C.PI_PROTOCOLS: ["XEP-0176"],
47 C.PI_DEPENDENCIES: ["XEP-0166"],
48 C.PI_RECOMMENDATIONS: [],
49 C.PI_MAIN: "XEP_0176",
50 C.PI_HANDLER: "yes",
51 C.PI_DESCRIPTION: _("""Implementation of Jingle ICE-UDP transport"""),
52 }
53
54
55 class XEP_0176(BaseTransportHandler):
56
57 def __init__(self, host):
58 log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization")
59 self.host = host
60 self._j = host.plugins["XEP-0166"] # shortcut to access jingle
61 self._j.register_transport(
62 NS_JINGLE_ICE_UDP, self._j.TRANSPORT_DATAGRAM, self, 100
63 )
64 host.bridge.add_method(
65 "ice_candidates_add",
66 ".plugin",
67 in_sign="sss",
68 out_sign="",
69 method=self._ice_candidates_add,
70 async_=True,
71 )
72 host.bridge.add_signal(
73 "ice_candidates_new", ".plugin", signature="sss"
74 ) # args: jingle_sid, candidates_serialised, profile
75 host.bridge.add_signal(
76 "ice_restart", ".plugin", signature="sss"
77 ) # args: jingle_sid, side ("local" or "peer"), profile
78
79 def get_handler(self, client):
80 return XEP_0176_handler()
81
82 def _ice_candidates_add(
83 self,
84 session_id: str,
85 media_ice_data_s: str,
86 profile_key: str,
87 ):
88 client = self.host.get_client(profile_key)
89 return defer.ensureDeferred(self.ice_candidates_add(
90 client,
91 session_id,
92 data_format.deserialise(media_ice_data_s),
93 ))
94
95 def build_transport(self, ice_data: dict) -> domish.Element:
96 """Generate <transport> element from ICE data
97
98 @param ice_data: a dict containing the following keys:
99 - "ufrag" (str): The ICE username fragment.
100 - "pwd" (str): The ICE password.
101 - "candidates" (List[dict]): A list of ICE candidate dictionaries, each
102 containing:
103 - "component_id" (int): The component ID.
104 - "foundation" (str): The candidate foundation.
105 - "address" (str): The candidate IP address.
106 - "port" (int): The candidate port.
107 - "priority" (int): The candidate priority.
108 - "transport" (str): The candidate transport protocol, e.g., "udp".
109 - "type" (str): The candidate type, e.g., "host", "srflx", "prflx", or
110 "relay".
111 - "generation" (str, optional): The candidate generation. Defaults to "0".
112 - "network" (str, optional): The candidate network. Defaults to "0".
113 - "rel_addr" (str, optional): The related address for the candidate, if
114 any.
115 - "rel_port" (int, optional): The related port for the candidate, if any.
116
117 @return: A <transport> element.
118 """
119 try:
120 ufrag: str = ice_data["ufrag"]
121 pwd: str = ice_data["pwd"]
122 candidates: List[dict] = ice_data["candidates"]
123 except KeyError as e:
124 raise exceptions.DataError(f"ICE {e} must be provided")
125
126 candidates.sort(key=lambda c: int(c.get("priority", 0)), reverse=True)
127 transport_elt = domish.Element(
128 (NS_JINGLE_ICE_UDP, "transport"),
129 attribs={"ufrag": ufrag, "pwd": pwd}
130 )
131
132 for candidate in candidates:
133 try:
134 candidate_elt = transport_elt.addElement("candidate")
135 candidate_elt["component"] = str(candidate["component_id"])
136 candidate_elt["foundation"] = candidate["foundation"]
137 candidate_elt["generation"] = str(candidate.get("generation", "0"))
138 candidate_elt["id"] = candidate.get("id") or str(uuid.uuid4())
139 candidate_elt["ip"] = candidate["address"]
140 candidate_elt["network"] = str(candidate.get("network", "0"))
141 candidate_elt["port"] = str(candidate["port"])
142 candidate_elt["priority"] = str(candidate["priority"])
143 candidate_elt["protocol"] = candidate["transport"]
144 candidate_elt["type"] = candidate["type"]
145 except KeyError as e:
146 raise exceptions.DataError(
147 f"Mandatory ICE candidate attribute {e} is missing"
148 )
149
150 if "rel_addr" in candidate and "rel_port" in candidate:
151 candidate_elt["rel-addr"] = candidate["rel_addr"]
152 candidate_elt["rel-port"] = str(candidate["rel_port"])
153
154 self.host.trigger.point("XEP-0176_build_transport", transport_elt, ice_data)
155
156 return transport_elt
157
158 def parse_transport(self, transport_elt: domish.Element) -> dict:
159 """Parse <transport> to a dict
160
161 @param transport_elt: <transport> element
162 @return: ICE data (as in [build_description])
163 """
164 try:
165 ice_data = {
166 "ufrag": transport_elt["ufrag"],
167 "pwd": transport_elt["pwd"]
168 }
169 except KeyError as e:
170 raise exceptions.DataError(
171 f"<transport> is missing mandatory attribute {e}: {transport_elt.toXml()}"
172 )
173 ice_data["candidates"] = ice_candidates = []
174
175 for candidate_elt in transport_elt.elements(NS_JINGLE_ICE_UDP, "candidate"):
176 try:
177 candidate = {
178 "component_id": int(candidate_elt["component"]),
179 "foundation": candidate_elt["foundation"],
180 "address": candidate_elt["ip"],
181 "port": int(candidate_elt["port"]),
182 "priority": int(candidate_elt["priority"]),
183 "transport": candidate_elt["protocol"],
184 "type": candidate_elt["type"],
185 }
186 except KeyError as e:
187 raise exceptions.DataError(
188 f"Mandatory attribute {e} is missing in candidate element"
189 )
190
191 if candidate_elt.hasAttribute("generation"):
192 candidate["generation"] = candidate_elt["generation"]
193
194 if candidate_elt.hasAttribute("network"):
195 candidate["network"] = candidate_elt["network"]
196
197 if candidate_elt.hasAttribute("rel-addr"):
198 candidate["rel_addr"] = candidate_elt["rel-addr"]
199
200 if candidate_elt.hasAttribute("rel-port"):
201 candidate["rel_port"] = int(candidate_elt["rel-port"])
202
203 ice_candidates.append(candidate)
204
205 self.host.trigger.point("XEP-0176_parse_transport", transport_elt, ice_data)
206
207 return ice_data
208
209 async def jingle_session_init(
210 self,
211 client: SatXMPPEntity,
212 session: dict,
213 content_name: str,
214 ) -> domish.Element:
215 """Create a Jingle session initiation transport element with ICE candidates.
216
217 @param client: SatXMPPEntity object representing the client.
218 @param session: Dictionary containing session data.
219 @param content_name: Name of the content.
220 @param ufrag: ICE username fragment.
221 @param pwd: ICE password.
222 @param candidates: List of ICE candidate dictionaries parsed from the
223 parse_ice_candidate method.
224
225 @return: domish.Element representing the Jingle transport element.
226
227 @raise exceptions.DataError: If mandatory data is missing from the candidates.
228 """
229 content_data = session["contents"][content_name]
230 transport_data = content_data["transport_data"]
231 ice_data = transport_data["local_ice_data"]
232 return self.build_transport(ice_data)
233
234 async def jingle_handler(
235 self,
236 client: SatXMPPEntity,
237 action: str,
238 session: dict,
239 content_name: str,
240 transport_elt: domish.Element,
241 ) -> domish.Element:
242 """Handle Jingle requests
243
244 @param client: The SatXMPPEntity instance.
245 @param action: The action to be performed with the session.
246 @param session: A dictionary containing the session information.
247 @param content_name: The name of the content.
248 @param transport_elt: The domish.Element instance representing the transport
249 element.
250
251 @return: <transport> element
252 """
253 content_data = session["contents"][content_name]
254 transport_data = content_data["transport_data"]
255 if action in (self._j.A_PREPARE_CONFIRMATION, self._j.A_PREPARE_INITIATOR):
256 peer_ice_data = self.parse_transport(transport_elt)
257 transport_data["peer_ice_data"] = peer_ice_data
258
259 elif action in (self._j.A_ACCEPTED_ACK, self._j.A_PREPARE_RESPONDER):
260 pass
261
262 elif action == self._j.A_SESSION_ACCEPT:
263 pass
264
265 elif action == self._j.A_START:
266 pass
267
268 elif action == self._j.A_SESSION_INITIATE:
269 # responder side, we give our candidates
270 transport_elt = self.build_transport(transport_data["local_ice_data"])
271 elif action == self._j.A_TRANSPORT_INFO:
272 new_ice_data = self.parse_transport(transport_elt)
273 restart = self.update_candidates(transport_data, new_ice_data, local=False)
274 if restart:
275 log.debug(
276 f"Peer ICE restart detected on session {session['id']} "
277 f"[{client.profile}]"
278 )
279 self.host.bridge.ice_restart(session["id"], "peer", client.profile)
280
281 self.host.bridge.ice_candidates_new(
282 session["id"],
283 data_format.serialise(new_ice_data["candidates"]),
284 client.profile
285 )
286 elif action == self._j.A_DESTROY:
287 pass
288 else:
289 log.warning("FIXME: unmanaged action {}".format(action))
290
291 return transport_elt
292
293 def jingle_terminate(self, client, action, session, content_name, reason_elt):
294 log.debug("ICE-UDP session terminated")
295
296 def update_candidates(
297 self,
298 transport_data: dict,
299 new_ice_data: dict,
300 local: bool
301 ) -> bool:
302 """Update ICE candidates when new one are received
303
304 @param transport_data: transport_data of the content linked to the candidates
305 @param new_ice_data: new ICE data, in the same format as returned
306 by [self.parse_transport]
307 @param local: True if it's our candidates, False if it's peer ones
308 @return: True if there is a ICE restart
309 """
310 key = "local_ice_data" if local else "peer_ice_data"
311 try:
312 ice_data = transport_data[key]
313 except KeyError:
314 log.warning(
315 f"no {key} available"
316 )
317 transport_data[key] = new_ice_data
318 else:
319 if (
320 new_ice_data["ufrag"] != ice_data["ufrag"]
321 or new_ice_data["pwd"] != ice_data["pwd"]
322 ):
323 ice_data["ufrag"] = new_ice_data["ufrag"]
324 ice_data["pwd"] = new_ice_data["pwd"]
325 ice_data["candidates"] = new_ice_data["candidates"]
326 return True
327 return False
328
329 async def ice_candidates_add(
330 self,
331 client: SatXMPPEntity,
332 session_id: str,
333 media_ice_data: Dict[str, dict]
334 ) -> None:
335 """Called when a new ICE candidates are available for a session
336
337 @param session_id: Session ID
338 @param candidates: a map from media type (audio, video) to ICE data
339 ICE data must be in the same format as in [self.parse_transport]
340 """
341 session = self._j.get_session(client, session_id)
342 iq_elt: Optional[domish.Element] = None
343
344 for media_type, new_ice_data in media_ice_data.items():
345 for content_name, content_data in session["contents"].items():
346 if content_data["application_data"].get("media") == media_type:
347 break
348 else:
349 log.warning(
350 "no media of type {media_type} has been found"
351 )
352 continue
353 restart = self.update_candidates(
354 content_data["transport_data"], new_ice_data, True
355 )
356 if restart:
357 log.debug(
358 f"Local ICE restart detected on session {session['id']} "
359 f"[{client.profile}]"
360 )
361 self.host.bridge.ice_restart(session["id"], "local", client.profile)
362 transport_elt = self.build_transport(new_ice_data)
363 iq_elt, __ = self._j.build_action(
364 client, self._j.A_TRANSPORT_INFO, session, content_name, iq_elt=iq_elt,
365 transport_elt=transport_elt
366 )
367
368 if iq_elt is not None:
369 try:
370 await iq_elt.send()
371 except Exception as e:
372 log.warning(f"Could not send new ICE candidates: {e}")
373
374 else:
375 log.error("Could not find any content to apply new ICE candidates")
376
377
378 @implementer(iwokkel.IDisco)
379 class XEP_0176_handler(XMPPHandler):
380
381 def getDiscoInfo(self, requestor, target, nodeIdentifier=""):
382 return [disco.DiscoFeature(NS_JINGLE_ICE_UDP)]
383
384 def getDiscoItems(self, requestor, target, nodeIdentifier=""):
385 return []