diff --git a/libervia/backend/plugins/plugin_comp_conferences/__init__.py b/libervia/backend/plugins/plugin_comp_conferences/__init__.py index 5538b743..b1bab127 100644 --- a/libervia/backend/plugins/plugin_comp_conferences/__init__.py +++ b/libervia/backend/plugins/plugin_comp_conferences/__init__.py @@ -18,8 +18,9 @@ import json import os +import re from pathlib import Path -from typing import Self +from typing import Self, cast from urllib.parse import quote from autobahn.twisted.websocket import WebSocketClientFactory, WebSocketClientProtocol @@ -42,11 +43,16 @@ from libervia.backend import G from libervia.backend.core import exceptions from libervia.backend.core.constants import Const as C from libervia.backend.core.core_types import SatXMPPEntity +from libervia.backend.core.xmpp import SatXMPPComponent from libervia.backend.core.i18n import _ from libervia.backend.core.log import getLogger from libervia.backend.plugins.plugin_xep_0106 import XEP_0106 from libervia.backend.plugins.plugin_xep_0166 import XEP_0166 -from libervia.backend.plugins.plugin_xep_0167 import XEP_0167, mapping, NS_AV_CONFERENCES +from libervia.backend.plugins.plugin_xep_0167 import XEP_0167, mapping +from libervia.backend.plugins.plugin_misc_av_conferences import ( + AVConferences, + NS_AV_CONFERENCES, +) from libervia.backend.plugins.plugin_xep_0176 import XEP_0176 from libervia.backend.plugins.plugin_xep_0298 import XEP_0298 from libervia.backend.tools.common import async_process, regex @@ -63,7 +69,14 @@ PLUGIN_INFO = { C.PI_MODES: [C.PLUG_MODE_COMPONENT], C.PI_TYPE: C.PLUG_TYPE_ENTRY_POINT, C.PI_PROTOCOLS: [], - C.PI_DEPENDENCIES: ["XEP-0106", "XEP-0166", "XEP-0167", "XEP-0176", "XEP-0298"], + C.PI_DEPENDENCIES: [ + "AV_CONFERENCES", + "XEP-0106", + "XEP-0166", + "XEP-0167", + "XEP-0176", + "XEP-0298", + ], C.PI_RECOMMENDATIONS: [], C.PI_MAIN: "ConferencesComponent", C.PI_HANDLER: C.BOOL_TRUE, @@ -92,6 +105,8 @@ class ConferenceSettings(ComponentSettings, plugin_info=PLUGIN_INFO): class Conference: + _PENDING_ICE_TIMEOUT: float = 10 + def __init__( self, parent: "ConferencesComponent", @@ -141,7 +156,11 @@ class Conference: return self.parent._ice_udp @property - def client(self) -> SatXMPPEntity: + def _av_conf(self) -> AVConferences: + return self.parent._av_conf + + @property + def client(self) -> SatXMPPComponent: client = self.parent.client assert client is not None return client @@ -151,19 +170,6 @@ class Conference: """ID to use with Galène API.""" return f"{self.client_id}-{self.session['id']}" - async def on_call_setup( - self, - client: SatXMPPEntity, - session: dict, - call_data: dict, - ) -> None: - if call_data["role"] == self._j.ROLE_INITIATOR: - self.send_answer(session["id"], call_data["sdp"]) - else: - raise exceptions.InternalError( - '"role" should be "{self._j.ROLE_INITIATOR}" here.' - ) - def join(self, user_jid: jid.JID) -> None: self.protocol.send_data( { @@ -175,6 +181,15 @@ class Conference: } ) + def close(self) -> None: + """Close the SFU connection (the session is over).""" + protocol = self._protocol + self._protocol = None + self.parent._conferences.pop(self.session["id"], None) + if protocol is not None: + log.debug(f"Closing SFU connection {protocol.client_id!r}.") + protocol.sendClose() + def send_request(self, requested: dict | None = None) -> None: if requested is None: requested = {"": ["audio", "video"]} @@ -191,27 +206,34 @@ class Conference: } ) - def send_answer(self, session_id: str, sdp: str) -> None: - data_id = session_id[len(self.client_id) + 1 :] + def send_answer(self, stream_id: str, sdp: str) -> None: + """Send the answer SDP for a stream (our uplink or a downlink PC). + + @param stream_id: ID of the stream to answer. + @param sdp: answer SDP to send. + """ self.protocol.send_data( { "type": "answer", - "id": data_id, - "label": "camera", - "username": self.session["peer_jid"].userhost(), + "id": stream_id, "sdp": sdp, } ) - def add_candidate(self, candidate: dict, sdp_mid: str, sdp_mline_index: int) -> None: + def add_candidate( + self, candidate: dict, stream_id: str, sdp_mid: str, sdp_mline_index: int + ) -> None: """Add an ICE candidate. @param candidate: ICE candidate, SDP format. + @param stream_id: ID of the stream the candidate belongs to. + @param sdp_mid: mid of the m-line, as in the stream's SDP. + @param sdp_mline_index: index of the m-line the candidate belongs to. """ self.protocol.send_data( { "type": "ice", - "id": self.galene_id, + "id": stream_id, "candidate": { "candidate": mapping.generate_candidate_line(candidate), "sdpMid": sdp_mid, @@ -245,64 +267,256 @@ class Conference: '"answer_sdp_d" should be available in session.' ) else: + log.debug(f"Uplink answer received: {self._ice_credentials(data['sdp'])}") answer_sdp_d.callback(data["sdp"]) + if self.session.pop("uplink_renegotiating", False): + # an ICE restart: forward the SFU's new ICE credentials (XEP-0176) + defer.ensureDeferred( + self._forward_new_credentials( + data["sdp"], self.get_uplink_content_names() + ) + ) + # the pending uplink ICE candidates (if any) can be processed now + self._flush_pending_ice() - def on_offer(self, data: dict) -> None: - """Called when a new SFP offer has been received""" - defer.ensureDeferred(self.a_on_offer(data)) + def get_uplink_content_names(self) -> list[str]: + """Get the names of the user's own (uplink) contents.""" + streams = self.session.get("streams", {}) + return [ + name + for name in self.session["contents"] + if not any(name in names for names in streams.values()) + ] - async def a_on_offer(self, data: dict) -> None: - client: SatXMPPEntity = self.client.get_virtual_client(self.session["local_jid"]) - call_data = {"sdp": data["sdp"]} + @staticmethod + def _origin_entity(data: dict) -> str | None: + """Get the entity URI of the origin of a stream + + @param data: the SFU "offer" data. + @return: entity URI of the origin, or None if it can't be determined. + """ + username = data.get("username", "") try: - # FIXME: This assume that all username are coming from XMPP, "source" should - # be used instead to be sure to match XMPP users, and other ones should use - # a non "xmpp:" URL scheme. - user_jid = jid.JID(data["username"]) - if not user_jid.user: - raise ValueError("Missing user part.") - except Exception as e: - log.warning("Username is not a JID: {data['username']=}, {e}") - else: - call_data["metadata"] = {"user": user_jid} - sid = await self._rtp.call_start( - client, - self.session["peer_jid"], - call_data, - session_id=f"{self.client_id}-{data['id']}", - ) - session = self._j.get_session(client, sid) - session["call_setup_cb"] = self.on_call_setup + user_jid = jid.JID(username) + if user_jid.user: + # FIXME: "source" should be checked to be sure it is an XMPP user + return f"xmpp:{user_jid.userhost()}" + except Exception: + pass + # non-XMPP origin (e.g. user of the SFU's own interface) + source = data.get("source") or username + return f"galene-source:{source}" if source else None - def on_ice(self, data: dict) -> None: - if data["id"] == self.galene_id: - session = self.session - else: - session = self._j.get_session(self.client, f"{self.client_id}-{data['id']}") + async def on_offer(self, data: dict) -> None: + """Handle a stream offer from the SFU (a re-offer is an ICE restart). + + @param data: the SFU "offer" data. + """ + stream_id = data["id"] + if stream_id == self.galene_id: + # our own uplink stream, nothing to do + return + session = self.session + if stream_id in session.get("streams", {}): + # a re-offer of an existing stream: an ICE restart (contents unchanged) + await self._handle_renegotiation(data) + return + counter = session.get("stream_counter", 0) + 1 + session["stream_counter"] = counter + # Galene mids are unique only per stream, this can clash as we use a single jingle + # session + sdp = re.sub(r"(?m)^a=mid:(\S+)", rf"a=mid:{counter}-\1", data["sdp"]) + jingle_child_elts: list[domish.Element] = [] + if (origin := self._origin_entity(data)) is not None: + # one per added stream, as a child of the element + jingle_child_elts.append(self._av_conf.build_origin_elt(origin)) + client: SatXMPPEntity = self.client.get_virtual_client(session["local_jid"]) log.debug( - f"ICE candidate for session {session['id']}, peer_jid={session['peer_jid']}." + f"Adding stream {stream_id!r} (origin {origin!r}, counter {counter}) " + f"to session {session['id']!r}." ) - candidate_data = data["candidate"] - contents = session["contents"] - try: - content_id = list(contents)[candidate_data["sdpMLineIndex"]] - except IndexError: - log.error( - f"Can't find any content at index {candidate_data['sdpMLineIndex']}." + added = await self._rtp.call_update( + client, session["id"], {"sdp": sdp}, jingle_child_elts=jingle_child_elts + ) + if added: + session.setdefault("streams", {})[stream_id] = added + log.debug(f"Stream {stream_id!r} added to session {session['id']!r}: {added}") + # the contents are now available, the pending ICE candidates can be processed + self._flush_pending_ice() + + async def _handle_renegotiation(self, data: dict) -> None: + """The SFU re-offered an existing stream (an ICE restart, XEP-0176). + + @param data: the SFU renegotiation data. + """ + stream_id = data["id"] + session = self.session + names = session["streams"][stream_id] + if not await self._forward_new_credentials(data["sdp"], names): + return + answer_sdp = session.get("stream_answer_sdp", {}).get(stream_id) + if answer_sdp is None: + log.warning( + f"No stored answer for stream {stream_id!r}, can't re-answer the SFU." ) return - content = contents[content_id] - local_ice_data = content["transport_data"]["local_ice_data"] + self.send_answer(stream_id, answer_sdp) + self._flush_pending_ice() - media = content["application_data"]["media"] + @staticmethod + def _ice_credentials(sdp: str) -> tuple[str, str] | None: + """ICE credentials (ufrag, pwd) of an SDP (or None if not found). + + @param sdp: the SDP to parse. + """ + ufrag_m = re.search(r"(?m)^a=ice-ufrag:(\S+)", sdp) + pwd_m = re.search(r"(?m)^a=ice-pwd:(\S+)", sdp) + if ufrag_m is None or pwd_m is None: + return None + return ufrag_m.group(1), pwd_m.group(1) + + def _restart_uplink(self, session: dict, ice_data: dict) -> None: + """Re-offer the uplink to the SFU with the new ICE credentials. + + A same-ID offer is a renegotiation, its answer is forwarded to the user. + @param session: the session to restart the uplink of. + @param ice_data: new ICE data of the uplink (ufrag, pwd). + """ + offer_sdp = session.get("uplink_offer_sdp") + if offer_sdp is None: + return + if self._ice_credentials(offer_sdp) == (ice_data["ufrag"], ice_data["pwd"]): + # unchanged credentials: a plain trickle, or another content of the transport + return + # [^\r\n] (not .*): a bare $ would consume the CR of a CRLF line ending, mangle + # the SDP and break its parsing + new_sdp = re.sub( + r"(?m)^a=ice-ufrag:[^\r\n]*", f"a=ice-ufrag:{ice_data['ufrag']}", offer_sdp + ) + new_sdp = re.sub( + r"(?m)^a=ice-pwd:[^\r\n]*", f"a=ice-pwd:{ice_data['pwd']}", new_sdp + ) + session["uplink_offer_sdp"] = new_sdp + session["uplink_renegotiating"] = True + # the SFU's answer is forwarded to on_answer(), nothing more to do with it + session["answer_sdp_d"] = defer.Deferred() + log.debug( + f"Uplink ICE restart, re-offering stream {self.galene_id!r} to the SFU." + ) + self.send_offer(new_sdp) + + async def _forward_new_credentials(self, sdp: str, names: list[str]) -> bool: + """Forward the SFU's new ICE credentials (and inline candidates) to the user. + + The user completes its side of the ICE restart with them (XEP-0176). + @param sdp: the SFU (re-)offer SDP carrying the new credentials. + @param names: content names of the stream, in m-line order. + @return: True if the credentials have been forwarded. + """ + credentials = self._ice_credentials(sdp) + if credentials is None: + log.warning(f"No ICE credentials in the SFU SDP for contents {names!r}.") + return False + ufrag, pwd = credentials + session = self.session + client: SatXMPPEntity = self.client.get_virtual_client(session["local_jid"]) + # the m-lines share a single transport: all the contents get the same credentials + for name in names: + await self._ice_udp.ice_candidates_add( + client, + session["id"], + {name: {"ufrag": ufrag, "pwd": pwd, "candidates": []}}, + ) + self._send_inline_candidates(sdp, names) + return True + + def _send_inline_candidates(self, sdp: str, names: list[str]) -> None: + """Send the candidates included in an SDP. + + The SFU also trickles them, a duplicate is harmless for the user's client. + @param sdp: the SDP containing the candidates. + @param names: content names, in m-line order. + """ + mlines = re.split(r"(?m)^(?=m=)", sdp)[1:] + for mline_index, (name, section) in enumerate(zip(names, mlines)): + for line in section.splitlines(): + if line.startswith("a=candidate:"): + self._send_ice_candidate( + name, + { + "candidate": line.removeprefix("a="), + "sdpMLineIndex": mline_index, + }, + ) + + def on_ice(self, data: dict) -> None: + """Called when an ICE candidate has been received from the SFU + + @param data: the SFU "ice" data. + """ + stream_id = data["id"] + session = self.session + candidate_data = data["candidate"] + mline_index = candidate_data["sdpMLineIndex"] + log.debug(f"on_ice: stream {stream_id!r}, mline {mline_index}") + content_name = self._resolve_ice_content(stream_id, mline_index) + if content_name is None or self._get_local_ice_data(content_name) is None: + # content (or local ICE data) not available yet: buffer the candidate + session.setdefault("pending_ice", []).append( + (reactor.seconds(), stream_id, mline_index, candidate_data) + ) + if session.get("pending_ice_retry") is None: + session["pending_ice_retry"] = reactor.callLater( + 0.5, self._flush_pending_ice + ) + return + self._send_ice_candidate(content_name, candidate_data) + + def _resolve_ice_content(self, stream_id: str, mline_index: int) -> str | None: + """Name of the session content an ICE candidate belongs to + + @param stream_id: ID of the stream the candidate belongs to. + @param mline_index: index of the m-line in the stream SDP. + @return: content name, or None if the stream is not (yet) available. + """ + if stream_id == self.galene_id: + # uplink stream: the mline index is on the user's own contents + try: + return self.get_uplink_content_names()[mline_index] + except IndexError: + return None + try: + return self.session["streams"][stream_id][mline_index] + except (KeyError, IndexError): + return None + + def _get_local_ice_data(self, content_name: str) -> dict | None: + """Local ICE data of a session content (or None if not available yet). + + @param content_name: name of the session content. + """ + return self.session["contents"][content_name]["transport_data"].get( + "local_ice_data" + ) + + def _send_ice_candidate(self, content_name: str, candidate_data: dict) -> None: + """Send an ICE candidate (received from the SFU) to the user + + @param content_name: name of the session content the candidate belongs to. + @param candidate_data: candidate data (as provided by the SFU). + """ + session = self.session + local_ice_data = self._get_local_ice_data(content_name) + candidate = mapping.parse_candidate( + candidate_data["candidate"].removeprefix("candidate:").split() + ) client: SatXMPPEntity = self.client.get_virtual_client(session["local_jid"]) - candidate = mapping.parse_candidate(candidate_data["candidate"][10:].split()) defer.ensureDeferred( self._ice_udp.ice_candidates_add( client, session["id"], { - media: { + content_name: { "candidates": [candidate], "ufrag": local_ice_data["ufrag"], "pwd": local_ice_data["pwd"], @@ -311,6 +525,72 @@ class Conference: ) ) + def _flush_pending_ice(self) -> None: + """Process the ICE candidates buffered in on_ice().""" + session = self.session + session["pending_ice_retry"] = None + pending = session.get("pending_ice") + if not pending: + return + now = reactor.seconds() + kept: list = [] + for timestamp, stream_id, mline_index, candidate_data in pending: + if now - timestamp > self._PENDING_ICE_TIMEOUT: + log.debug( + f"Timeout, dropping pending ICE candidate for stream " + f"{stream_id!r}." + ) + continue + content_name = self._resolve_ice_content(stream_id, mline_index) + if content_name is None or self._get_local_ice_data(content_name) is None: + kept.append((timestamp, stream_id, mline_index, candidate_data)) + continue + self._send_ice_candidate(content_name, candidate_data) + if kept: + session["pending_ice"] = kept + session["pending_ice_retry"] = reactor.callLater(0.5, self._flush_pending_ice) + else: + del session["pending_ice"] + + def on_close(self, data: dict) -> None: + """Called when a stream has been closed by the SFU + + @param data: the SFU "close" data. + """ + stream_id = data["id"] + if stream_id == self.galene_id: + # the uplink stream is closed (the user left), handled at the XMPP level + log.debug("Uplink stream closed.") + return + session = self.session + names = session.get("streams", {}).pop(stream_id, None) + if not names: + # an SFU race (close before offer processed): a later offer is still handled + log.debug(f"No contents found for closed stream {stream_id!r}.") + return + session.get("partial_stream_answers", {}).pop(stream_id, None) + session.get("stream_answer_sdp", {}).pop(stream_id, None) + iq_elt, jingle_elt = self._j._build_jingle_elt( + self.client, session, self._j.A_CONTENT_REMOVE + ) + for name in names: + content_elt = jingle_elt.addElement("content", self._j.namespace) + content_elt["creator"] = session["role"] + content_elt["name"] = name + self.client.send(iq_elt) + for name in names: + session["contents"].pop(name, None) + log.debug(f"Contents {names} of stream {stream_id!r} removed.") + + def on_renegotiate(self, data: dict) -> None: + """The SFU requested a renegotiation (only logged). + + The user's client handles the ICE restart itself (see _restart_uplink). + @param data: the SFU "renegotiate" data. + """ + stream_id = data.get("id", "") + log.warning(f"The SFU requested a renegotiation for stream {stream_id!r}.") + class GaleneClientFactory(WebSocketClientFactory): @@ -337,6 +617,9 @@ class GaleneProtocol(WebSocketClientProtocol): def connectionLost(self, reason: failure.Failure = ConnectionDone) -> None: super().connectionLost(reason) + # clear the reference only (the conference is closed on session end) + if self.conference._protocol is self: + self.conference._protocol = None def onOpen(self): handshake_data = {"type": "handshake", "version": ["2"], "id": self.client_id} @@ -373,9 +656,18 @@ class GaleneProtocol(WebSocketClientProtocol): case "ping": log.debug("pong") self.send_data({"type": "pong"}) - case "joined" | "answer" | "ice" | "offer" as data_type: + case "offer": + defer.ensureDeferred(self.conference.on_offer(data)) + case "joined" | "answer" | "ice" as data_type: method = getattr(self.conference, f"on_{data_type}") method(data) + case "close" | "abort": + self.conference.on_close(data) + case "renegotiate": + self.conference.on_renegotiate(data) + case "user": + # user roster update (ignored, stream changes come separately) + log.debug(f"User update: {data}") case _: log.debug(f"Unhandled message: {data}") except (KeyError, IndexError): @@ -388,13 +680,17 @@ class ConferencesComponent: def __init__(self, host): self.host = host - self.client: SatXMPPEntity | None = None + self.client: SatXMPPComponent | None = None self._e: XEP_0106 = host.plugins["XEP-0106"] self._j: XEP_0166 = host.plugins["XEP-0166"] self._rtp: XEP_0167 = host.plugins["XEP-0167"] self._ice_udp: XEP_0176 = host.plugins["XEP-0176"] self._coin: XEP_0298 = host.plugins["XEP-0298"] + self._av_conf: AVConferences = host.plugins["AV_CONFERENCES"] + self._conferences: dict[str, Conference] = {} host.trigger.add("XEP-0167_jingle_handler", self._jingle_handler_trigger) + host.trigger.add("XEP-0166_session_terminate", self._session_terminated) + host.trigger.add("disconnected", self._client_disconnected) self.initalized = False def _init(self) -> None: @@ -450,7 +746,7 @@ class ConferencesComponent: return ConferencesHandler() def profile_connecting(self, client: SatXMPPEntity) -> None: - self.client = client + self.client = cast(SatXMPPComponent, client) if not self.initalized: self._init() self.initalized = True @@ -500,9 +796,42 @@ class ConferencesComponent: # factory.setProtocolOptions(logOctets=True) factory.protocol = GaleneProtocol reactor.connectTCP("127.0.0.1", int(self.galene_http_port), factory) + self._conferences[session["id"]] = conference return conference + def _session_terminated( + self, client: SatXMPPEntity, session: dict, reason_elt: domish.Element + ) -> None: + """The peer terminated the Jingle session: close the conference, if any. + + @param client: the client entity. + @param session: the terminated session. + @param reason_elt: the element of the session-terminate action. + """ + conference = session.get("conference") + if conference is not None: + conference.close() + + def _client_disconnected(self, client: SatXMPPEntity, reason) -> bool: + """A client has disconnected: close the conferences of its sessions. + + The sessions are not terminated, so the conferences must be closed here. + @param client: the client entity. + @param reason: disconnect reason. + @return: True, to let the (cancellable) workflow continue. + """ + peer_bare = client.jid.userhostJID() + for conference in list(self._conferences.values()): + peer_jid = conference.session.get("peer_jid") + if peer_jid is not None and peer_jid.userhostJID() == peer_bare: + log.debug( + f"Client {client.jid.full()} disconnected, closing " + f"conference for session {conference.session['id']!r}." + ) + conference.close() + return True + async def _jingle_handler_trigger( self, client: SatXMPPEntity, @@ -522,10 +851,24 @@ class ConferencesComponent: raise StanzaError("forbidden", "A room name must be specified.") group_name = self._e.unescape(local_jid.user) + # dedup: an existing session for the same JID (e.g. a page reload) is replaced + peer_jid = session["peer_jid"] + for conf in list(self._conferences.values()): + if conf.group_name == group_name and conf.session["peer_jid"] == peer_jid: + log.info( + f"Dedup: closing existing session {conf.session['id']} for " + f"{peer_jid} in {group_name!r} (replaced by {session['id']})." + ) + # notify the client (if still connected) that its session was replaced + await self._j.terminate(client, self._j.REASON_SUCCESS, conf.session) + conf.close() + break + session["conference"] = await self.attach_to_group(session, group_name) session["pre_accepted"] = True session["call_setup_cb"] = self.on_call_setup session["ice_candidates_new_cb"] = self.on_ice_candidates_new + session["content_accept_cb"] = self.on_content_accept async def on_call_setup( self, @@ -543,6 +886,8 @@ class ConferencesComponent: if call_data["role"] == self._j.ROLE_RESPONDER: await conference.ready conference.join(session["peer_jid"]) + # keep the uplink offer: re-sent to the SFU on an ICE restart + session["uplink_offer_sdp"] = call_data["sdp"] conference.send_offer(call_data["sdp"]) else: raise exceptions.InternalError( @@ -555,22 +900,125 @@ class ConferencesComponent: session: dict, ice_candidates_data: dict[str, dict], ) -> None: + """The user's client sent new ICE candidates (trickle) + + @param client: the client entity. + @param session: the session the candidates belong to. + @param ice_candidates_data: ICE data, keyed by content name. + """ try: conference = session["conference"] except KeyError: raise exceptions.InternalError("Conference data is missing.") - for media, media_data in ice_candidates_data.items(): - for idx, (content_id, content) in enumerate(session["contents"].items()): - if content["application_data"]["media"] == media: - break + streams = session.get("streams", {}) + uplink_names = conference.get_uplink_content_names() + for content_name, ice_data in ice_candidates_data.items(): + if content_name in uplink_names: + conference._restart_uplink(session, ice_data) + stream_id = conference.galene_id + sdp_mid = content_name + sdp_mline_index = uplink_names.index(content_name) else: - log.error(f"Can't find corresponding content for {media!r}") - continue - sdp_mline_index: int = idx - sdp_mid: str = content_id + stream_id = next( + (sid for sid, names in streams.items() if content_name in names), + None, + ) + if stream_id is None: + log.error(f"Can't find corresponding stream for {content_name!r}") + continue + # the SFU uses the m-line index as MID in the stream offer + sdp_mline_index = streams[stream_id].index(content_name) + sdp_mid = str(sdp_mline_index) - for candidate in media_data["candidates"]: - conference.add_candidate(candidate, sdp_mid, sdp_mline_index) + for candidate in ice_data["candidates"]: + conference.add_candidate(candidate, stream_id, sdp_mid, sdp_mline_index) + + def on_content_accept( + self, + client: SatXMPPEntity, + session: dict, + sdp: str, + content_names: list[str], + ) -> None: + """The user accepted added contents: forward the answer to the SFU. + + Each stream is answered as a whole, when all its m-lines are received. + @param client: the client entity. + @param session: the session the contents belong to. + @param sdp: the user's answer SDP. + @param content_names: names of the accepted contents. + """ + conference = session.get("conference") + streams = session.get("streams", {}) + if conference is None or not streams: + return + answers = session.setdefault("partial_stream_answers", {}) + for name in content_names: + stream_id = next( + (sid for sid, names in streams.items() if name in names), None + ) + if stream_id is None: + log.warning(f"No stream found for content {name!r}.") + continue + try: + mline = self._get_mline_section(sdp, name) + except ValueError as e: + log.warning(f"{e}, can't answer stream {stream_id!r}.") + continue + answers.setdefault(stream_id, {})[name] = mline + if all(n in answers[stream_id] for n in streams[stream_id]): + names = streams[stream_id] + mlines = [answers[stream_id][n] for n in names] + del answers[stream_id] + # content names are "{counter}-{mid}": use the original mids in the answer + orig_mids = [n.split("-", 1)[1] for n in names] + answer_sdp = self._build_answer_sdp(names, orig_mids, mlines) + # keep the answer: it is re-sent to the SFU on an ICE restart + session.setdefault("stream_answer_sdp", {})[stream_id] = answer_sdp + conference.send_answer(stream_id, answer_sdp) + + @staticmethod + def _get_mline_section(sdp: str, content_name: str) -> str: + """Extract the m-line section of a content from an SDP string + + @param sdp: the SDP to parse. + @param content_name: name of the content (a=mid). + """ + for section in re.split(r"(?m)^(?=m=)", sdp): + if f"a=mid:{content_name}" in section: + return section + raise ValueError(f"content {content_name!r} not found in SDP") + + @staticmethod + def _build_answer_sdp( + names: list[str], orig_mids: list[str], mlines: list[str] + ) -> str: + """Build the answer SDP to send to the SFU, from the user's m-line sections. + + The answer mids must match the offer's, so they are renamed accordingly. + @param names: content names, in m-line order. + @param orig_mids: the mids of the SFU offer (used for the BUNDLE group). + @param mlines: the user's answer m-line sections, in the same order. + """ + sections = [] + for name, orig_mid, mline in zip(names, orig_mids, mlines): + # normalize CRLF line endings (browser SDP), so the mid rename matches + mline = mline.replace("\r\n", "\n").replace("\r", "\n") + sections.append( + re.sub( + rf"(?m)^a=mid:{re.escape(name)}$", + f"a=mid:{orig_mid}", + mline, + ) + ) + # the answer must declare the BUNDLE group (single transport), with the offer mids + bundle = "a=group:BUNDLE " + " ".join(orig_mids) + return ( + "\n".join( + ["v=0", "o=- 0 0 IN IP4 127.0.0.1", "s=-", "t=0 0", bundle, *sections] + ) + + "\n" + ) def _galene_process_errback(self, failure_: Failure) -> None: log.error(f"Can't run Galene process: {failure_.value}") diff --git a/tests/unit/test_plugin_comp_conferences.py b/tests/unit/test_plugin_comp_conferences.py new file mode 100644 index 00000000..4f6a606b --- /dev/null +++ b/tests/unit/test_plugin_comp_conferences.py @@ -0,0 +1,318 @@ +"""Unit tests for the conferences component (ICE restart paths).""" + +from pathlib import Path +from types import SimpleNamespace + +from pytest import fixture +from twisted.internet import defer +from twisted.words.protocols.jabber import jid + +from libervia.backend.plugins.plugin_comp_conferences import Conference + +# the SDP line endings are CRLF, as in real offers (RFC 4566) +UPLINK_OFFER_SDP = ("""v=0 +o=- 0 0 IN IP4 127.0.0.1 +s=- +t=0 0 +a=group:BUNDLE 0 1 +m=audio 9 UDP/TLS/RTP/SAVPF 111 +c=IN IP4 0.0.0.0 +a=mid:0 +a=sendonly +a=ice-ufrag:uf1 +a=ice-pwd:pw1 +m=video 9 UDP/TLS/RTP/SAVPF 96 +c=IN IP4 0.0.0.0 +a=mid:1 +a=sendonly +a=ice-ufrag:uf1 +a=ice-pwd:pw1 +""").replace("\n", "\r\n") + +# answer of the renegotiated uplink offer (an ICE restart by the user): new ICE +# credentials, with inline candidates +UPLINK_ANSWER_SDP = ("""v=0 +o=- 0 0 IN IP4 127.0.0.1 +s=- +t=0 0 +a=group:BUNDLE 0 1 +m=audio 9 UDP/TLS/RTP/SAVPF 111 +c=IN IP4 0.0.0.0 +a=mid:0 +a=recvonly +a=ice-ufrag:uf3 +a=ice-pwd:pw3 +a=candidate:1 1 udp 2122260223 192.168.1.1 30000 typ host +m=video 9 UDP/TLS/RTP/SAVPF 96 +c=IN IP4 0.0.0.0 +a=mid:1 +a=recvonly +a=ice-ufrag:uf3 +a=ice-pwd:pw3 +a=candidate:2 1 udp 2122260223 192.168.1.1 30001 typ host +""").replace("\n", "\r\n") + +# new offer for an existing stream (an ICE restart by the SFU): new ICE credentials, with +# inline candidates +DOWNLINK_REOFFER_SDP = ("""v=0 +o=- 0 0 IN IP4 127.0.0.1 +s=- +t=0 0 +a=group:BUNDLE 0 1 +m=audio 9 UDP/TLS/RTP/SAVPF 111 +c=IN IP4 0.0.0.0 +a=mid:0 +a=sendonly +a=ice-ufrag:uf9 +a=ice-pwd:pw9 +a=candidate:1 1 udp 2122260223 192.168.1.1 30000 typ host +m=video 9 UDP/TLS/RTP/SAVPF 96 +c=IN IP4 0.0.0.0 +a=mid:1 +a=sendonly +a=ice-ufrag:uf9 +a=ice-pwd:pw9 +a=candidate:2 1 udp 2122260223 192.168.1.1 30001 typ host +""").replace("\n", "\r\n") + + +class MockProtocol: + """Records the data sent to the SFU.""" + + client_id = "c0ffee" + + def __init__(self): + self.sent = [] + + def send_data(self, data): + self.sent.append(data) + + def sendClose(self): + pass + + +class MockIceUdp: + """Records the calls to ice_candidates_add (the transport-info send path).""" + + def __init__(self, session: dict): + self.session = session + self.calls = [] + + def ice_candidates_add(self, client, session_id, content_ice_data): + # like XEP-0176: the session's local ICE data is updated with the new credentials + for name, ice_data in content_ice_data.items(): + local_ice_data = ( + self.session["contents"] + .get(name, {}) + .get("transport_data", {}) + .get("local_ice_data") + ) + if local_ice_data is not None: + local_ice_data.update(ufrag=ice_data["ufrag"], pwd=ice_data["pwd"]) + self.calls.append((client, session_id, content_ice_data)) + return defer.succeed(None) + + +@fixture +def conference(): + """A Conference with mocked parent (SFU protocol and XEP-0176).""" + session = { + "id": "sess123", + "local_jid": jid.JID("conferences.test.example"), + "peer_jid": jid.JID("user@test.example"), + "role": "responder", + "contents": {}, + "streams": {}, + "uplink_offer_sdp": UPLINK_OFFER_SDP, + } + parent = SimpleNamespace( + client=SimpleNamespace(get_virtual_client=lambda jid_: object()), + _ice_udp=MockIceUdp(session), + ) + conf = Conference( + parent, + "testroom", + Path("testroom.json"), + "ws://localhost", + "http://localhost", + session, + ) + conf._protocol = MockProtocol() + return conf + + +def test_restart_uplink_same_credentials(conference): + """No renegotiation if the uplink credentials are unchanged.""" + conference._restart_uplink( + conference.session, {"ufrag": "uf1", "pwd": "pw1", "candidates": []} + ) + assert conference._protocol.sent == [] + + +def test_restart_uplink_new_credentials(conference): + """New uplink credentials trigger a renegotiation offer to the SFU.""" + conference._restart_uplink( + conference.session, {"ufrag": "uf2", "pwd": "pw2", "candidates": []} + ) + (offer,) = conference._protocol.sent + assert offer["type"] == "offer" + assert offer["id"] == conference.galene_id + assert offer["sdp"].count("a=ice-ufrag:uf2") == 2 + assert offer["sdp"].count("a=ice-pwd:pw2") == 2 + # the CRLF line endings are preserved (a mangled SDP would not be parsable) + assert offer["sdp"].count("\r\n") == UPLINK_OFFER_SDP.count("\r\n") + # the offer is stored for the next restart + assert conference.session["uplink_offer_sdp"] == offer["sdp"] + assert isinstance(conference.session["answer_sdp_d"], defer.Deferred) + # the renegotiation is pending: the answer will carry new credentials + assert conference.session["uplink_renegotiating"] is True + + +def test_restart_uplink_no_stored_offer(conference): + """Without the stored uplink offer, nothing is re-offered.""" + del conference.session["uplink_offer_sdp"] + conference._restart_uplink( + conference.session, {"ufrag": "uf2", "pwd": "pw2", "candidates": []} + ) + assert conference._protocol.sent == [] + + +def test_restart_uplink_answer_consumed(conference): + """The answer of the renegotiation offer is consumed by on_answer().""" + conference._restart_uplink( + conference.session, {"ufrag": "uf2", "pwd": "pw2", "candidates": []} + ) + conference.on_answer({"sdp": "v=0\n"}) + assert conference.session.pop("answer_sdp_d", None) is None + + +def test_restart_uplink_answer_credentials_forwarded(conference): + """The answer's ICE credentials and inline candidates are forwarded to the user.""" + session = conference.session + for name in ("0", "1"): + session["contents"][name] = { + "transport_data": { + "local_ice_data": { + "ufrag": "uf_old", + "pwd": "pw_old", + "candidates": [], + } + } + } + conference._restart_uplink(session, {"ufrag": "uf2", "pwd": "pw2", "candidates": []}) + conference.on_answer({"sdp": UPLINK_ANSWER_SDP}) + assert "uplink_renegotiating" not in session + calls = conference.parent._ice_udp.calls + # two transport-info with the new credentials (one per content), then the two inline + # candidates + assert [next(iter(data)) for data in (c[2] for c in calls)] == [ + "0", + "1", + "0", + "1", + ] + for _, session_id, data in calls: + assert session_id == "sess123" + ((name, ice_data),) = data.items() + assert (ice_data["ufrag"], ice_data["pwd"]) == ("uf3", "pw3") + assert calls[2][2]["0"]["candidates"] + assert calls[3][2]["1"]["candidates"] + + +def test_on_answer_initial_no_credentials_forwarded(conference): + """Without an ICE restart, the answer is not forwarded to the user.""" + session = conference.session + for name in ("0", "1"): + session["contents"][name] = { + "transport_data": { + "local_ice_data": { + "ufrag": "uf3", + "pwd": "pw3", + "candidates": [], + } + } + } + session["answer_sdp_d"] = defer.Deferred() + conference.on_answer({"sdp": UPLINK_ANSWER_SDP}) + assert conference.parent._ice_udp.calls == [] + + +async def test_renegotiation_downlink(conference): + """A re-offered stream: new credentials to the user, stored answer to the SFU.""" + session = conference.session + session["streams"] = {"down1": ["2-0", "2-1"]} + session["stream_answer_sdp"] = {"down1": "ANSWER_SDP"} + # the contents of the stream, with the new local ICE data (set in production by the + # ice_candidates_add calls of _handle_renegotiation) + for name, media in (("2-0", "audio"), ("2-1", "video")): + session["contents"][name] = { + "application_data": {"media": media}, + "transport_data": { + "local_ice_data": {"ufrag": "uf9", "pwd": "pw9", "candidates": []} + }, + } + await conference._handle_renegotiation({"id": "down1", "sdp": DOWNLINK_REOFFER_SDP}) + # two transport-info with the new credentials (one per content), then the two inline + # candidates + calls = conference.parent._ice_udp.calls + assert len(calls) == 4 + for _, session_id, data in calls: + assert session_id == "sess123" + ((name, ice_data),) = data.items() + assert (ice_data["ufrag"], ice_data["pwd"]) == ("uf9", "pw9") + assert [next(iter(data)) for data in (c[2] for c in calls)] == [ + "2-0", + "2-1", + "2-0", + "2-1", + ] + assert calls[2][2]["2-0"]["candidates"] + assert calls[3][2]["2-1"]["candidates"] + answer = [m for m in conference._protocol.sent if m["type"] == "answer"] + assert len(answer) == 1 + assert answer[0]["id"] == "down1" + assert answer[0]["sdp"] == "ANSWER_SDP" + + +async def test_renegotiation_downlink_no_stored_answer(conference): + """Without the stored answer, no answer is re-sent to the SFU.""" + session = conference.session + session["streams"] = {"down1": ["2-0", "2-1"]} + for name, media in (("2-0", "audio"), ("2-1", "video")): + session["contents"][name] = { + "application_data": {"media": media}, + "transport_data": { + "local_ice_data": {"ufrag": "uf9", "pwd": "pw9", "candidates": []} + }, + } + await conference._handle_renegotiation({"id": "down1", "sdp": DOWNLINK_REOFFER_SDP}) + assert all(m["type"] != "answer" for m in conference._protocol.sent) + + +async def test_offer_mids_renamed(conference): + """The mids of a new downlink stream are prefixed with the stream counter.""" + session = conference.session + added = [] + + class MockRtp: + async def call_update( + self, client, session_id, call_data, jingle_child_elts=None + ): + added.append(call_data) + return ["1-0", "1-1"] + + class MockAvConf: + def build_origin_elt(self, origin): + return None + + conference.parent._rtp = MockRtp() + conference.parent._av_conf = MockAvConf() + data = {"id": "down1", "sdp": DOWNLINK_REOFFER_SDP, "username": "user@test.example"} + await conference.on_offer(data) + (call_data,) = added + sdp = call_data["sdp"] + assert "a=mid:1-0" in sdp + assert "a=mid:1-1" in sdp + # the CRLF line endings are preserved (a mangled SDP would not be parsable) + assert sdp.count("\r\n") == DOWNLINK_REOFFER_SDP.count("\r\n") + assert session["streams"] == {"down1": ["1-0", "1-1"]}