From ab26a56c1da82700c0276742f9e04745b89f8d41 Mon Sep 17 00:00:00 2001 From: Goffi Date: Tue, 6 Oct 2026 19:27:53 +0200 Subject: [PATCH] plugin XEP-0166: support all content actions + fixes: - all content actions are now handled - XEP-0167: the new contents of an offered SDP are added with a single content-add, the contents sharing a single transport are bundled (XEP-0338) - XEP-0167: the generated SDP declares the BUNDLE group of its own contents - XEP-0176: one transport-info IQ per content. ICE data are keyed by content names. - fix session-level ICE attributes and fingerprint --- .../plugins/plugin_misc_remote_control.py | 2 +- .../plugins/plugin_xep_0166/__init__.py | 325 +++++++++++++- .../backend/plugins/plugin_xep_0166/models.py | 57 ++- .../plugins/plugin_xep_0167/__init__.py | 372 +++++++++++++--- .../plugins/plugin_xep_0167/mapping.py | 278 ++++++++---- .../backend/plugins/plugin_xep_0167/models.py | 30 ++ libervia/backend/plugins/plugin_xep_0176.py | 190 +++++---- libervia/backend/plugins/plugin_xep_0338.py | 101 ++++- libervia/backend/plugins/plugin_xep_0343.py | 2 +- tests/unit/conftest.py | 1 + tests/unit/test_plugin_xep_0166.py | 94 +++++ tests/unit/test_plugin_xep_0167.py | 398 +++++++++++++++++- tests/unit/test_plugin_xep_0176.py | 7 +- tests/unit/test_plugin_xep_0272.py | 27 +- tests/unit/test_plugin_xep_0338.py | 175 +++++++- 15 files changed, 1801 insertions(+), 258 deletions(-) create mode 100644 libervia/backend/plugins/plugin_xep_0167/models.py create mode 100644 tests/unit/test_plugin_xep_0166.py diff --git a/libervia/backend/plugins/plugin_misc_remote_control.py b/libervia/backend/plugins/plugin_misc_remote_control.py index e9f9f206..a418f015 100644 --- a/libervia/backend/plugins/plugin_misc_remote_control.py +++ b/libervia/backend/plugins/plugin_misc_remote_control.py @@ -119,7 +119,7 @@ class RemoteControl(BaseApplicationHandler): except KeyError: raise exceptions.DataError('"call_data" must be set in "extra".') - metadata = self._rtp.parse_call_data(call_data) + metadata = self._rtp.parse_call_data(call_data, self._j.ROLE_INITIATOR) try: application_data = call_data["application"] except KeyError: diff --git a/libervia/backend/plugins/plugin_xep_0166/__init__.py b/libervia/backend/plugins/plugin_xep_0166/__init__.py index 21322236..7a5bb196 100644 --- a/libervia/backend/plugins/plugin_xep_0166/__init__.py +++ b/libervia/backend/plugins/plugin_xep_0166/__init__.py @@ -18,7 +18,17 @@ import time -from typing import Any, Callable, Dict, Final, List, Optional, Tuple +from typing import ( + TYPE_CHECKING, + Any, + Callable, + Dict, + Final, + List, + Optional, + Tuple, + cast, +) import uuid from twisted.internet import defer @@ -39,10 +49,16 @@ from libervia.backend.core.log import getLogger from libervia.backend.tools import xml_tools from libervia.backend.tools import utils +if TYPE_CHECKING: + from libervia.backend.core.main import LiberviaBackend + from .models import ( ApplicationData, BaseApplicationHandler, BaseTransportHandler, + ContentChange, + ContentChangeAction, + ContentChangeItem, ContentData, TransportData, ) @@ -96,6 +112,7 @@ class XEP_0166: A_SESSION_TERMINATE: Final = "session-terminate" A_SESSION_INFO: Final = "session-info" A_CONTENT_ADD: Final = "content-add" + A_CONTENT_ACCEPT: Final = "content-accept" A_CONTENT_MODIFY: Final = "content-modify" A_CONTENT_REJECT: Final = "content-reject" A_CONTENT_REMOVE: Final = "content-remove" @@ -120,7 +137,7 @@ class XEP_0166: #: cleaning operations A_DESTROY: Final = "destroy" - def __init__(self, host): + def __init__(self, host: "LiberviaBackend"): log.info(_("plugin Jingle initialization")) self.host = host self._applications = {} # key: namespace, value: application data @@ -139,6 +156,9 @@ class XEP_0166: async_=True, ) + # args: serialised ContentChange payload, profile + host.bridge.add_signal("jingle_content_change", ".plugin", signature="ss") + def profile_connected(self, client): client.jingle_sessions = {} # key = sid, value = session_data @@ -829,6 +849,14 @@ class XEP_0166: await self.on_session_info(client, request, jingle_elt, session) elif action == XEP_0166.A_CONTENT_ADD: await self.on_content_add(client, request, jingle_elt, session) + elif action == XEP_0166.A_CONTENT_MODIFY: + await self.on_content_modify(client, request, jingle_elt, session) + elif action == XEP_0166.A_CONTENT_ACCEPT: + await self.on_content_accept(client, request, jingle_elt, session) + elif action == XEP_0166.A_CONTENT_REJECT: + await self.on_content_reject(client, request, jingle_elt, session) + elif action == XEP_0166.A_CONTENT_REMOVE: + await self.on_content_remove(client, request, jingle_elt, session) elif action == XEP_0166.A_TRANSPORT_INFO: self.on_transport_info(client, request, jingle_elt, session) elif action == XEP_0166.A_TRANSPORT_REPLACE: @@ -986,6 +1014,75 @@ class XEP_0166: """ return elt + def _emit_content_change( + self, + client: SatXMPPEntity, + action: str, + session: dict, + content_names: list[str] | None = None, + ) -> None: + """Emit the ``jingle_content_change`` signal for a content change + + The "get_content_change_payload" method from application handler will be + used for each content of the stanza, so that the frontend can apply all + the changes in a single WebRTC negotiation. + @param content_names: names of the contents of the current stanza; by default, all + the contents of the session pending the action are used + """ + content_names_key = { + self.A_CONTENT_ADD: "added_content_names", + self.A_CONTENT_MODIFY: "modified_content_names", + self.A_CONTENT_ACCEPT: "accepted_content_names", + self.A_CONTENT_REMOVE: "removed_content_names", + self.A_CONTENT_REJECT: "rejected_content_names", + }[action] + if content_names is None: + content_names = session.get(content_names_key) or [] + changes = [] + for content_name in content_names: + if content_name not in session["contents"]: + log.warning( + f"unknown content name {content_name!r} in session {session['id']}" + ) + continue + content_data = session["contents"][content_name] + try: + payload = content_data["application"].handler.get_content_change_payload( + client, action, session, content_name + ) + except Exception: + log.exception( + f"can't get the payload of content {content_name!r} " + f"in session {session['id']}" + ) + continue + changes.append( + ContentChangeItem( + content=content_name, + senders=content_data["senders"], + payload=payload, + ) + ) + if not changes: + return + info = ContentChange( + session_id=session["id"], + action=cast(ContentChangeAction, action), + changes=changes, + ) + if not client.is_component: + self.host.bridge.jingle_content_change( + info.model_dump_json(exclude_none=True), client.profile + ) + self.host.trigger.point( + "XEP-0166_content_change", + client, + action, + session, + [change.content for change in changes], + triggers_no_cancel=True, + ) + async def _call_plugins( self, client: SatXMPPEntity, @@ -1157,6 +1254,13 @@ class XEP_0166: try: for content_name, content_data in session["contents"].items(): + # contents not from the initial offer are not part of the session-accept + # (they are negotiated with a content-add/accept) + if ( + content_data.get("added_by_peer") + or content_data.get("creator") != XEP_0166.ROLE_INITIATOR + ): + continue content_elt = jingle_elt.addElement("content") content_elt["creator"] = XEP_0166.ROLE_INITIATOR content_elt["name"] = content_name @@ -1194,17 +1298,28 @@ class XEP_0166: app_method_name=None, elements=False, ) + self.host.trigger.point( + "XEP-0166_session_accept", + client, + session, + jingle_elt, + triggers_no_cancel=True, + ) session.pop("jingle_elt") await iq_elt.send() + except Exception as e: + await self._jingle_error_cb(e, session, jingle_elt, client) + return - session["state"] = XEP_0166.STATE_ACTIVE - + session["state"] = XEP_0166.STATE_ACTIVE + try: await self._call_plugins( client, XEP_0166.A_ACCEPTED_ACK, session, elements=False ) except Exception: - log.exception("Error while sending jingle stanza") - self.delete_session(client, session["id"]) + # the session-accept has been sent, an error in the ack handling must not kill + # an active session + log.exception("Error in the accepted-ack handling") def get_reason_elt(self, parent_elt: domish.Element) -> domish.Element: """Find a element in parent_elt @@ -1262,6 +1377,10 @@ class XEP_0166: force_element=reason_elt, ) + await self.host.trigger.async_point( + "XEP-0166_session_terminate", client, session, reason_elt + ) + self.delete_session(client, session["id"]) await client.a_send(xmlstream.toResponse(request, "result")) @@ -1346,16 +1465,18 @@ class XEP_0166: ) -> None: """Called on content-add action - The "jingle_request_confirmation" method of each application will be called - (or self.jingle_request_confirmation_default if the former doesn't exist). - The session is only accepted if all application are confirmed. - The application must manage itself multiple contents scenari (e.g. audio/video). + The contents are stored in the session, pending a content-accept. @param client: %(doc_client)s @param request(domish.Element): full request @param jingle_elt(domish.Element): element @param session(dict): session data """ - return + # the received contents are created by the peer (the session's other role) + creator = ( + XEP_0166.ROLE_RESPONDER + if session["role"] == XEP_0166.ROLE_INITIATOR + else XEP_0166.ROLE_INITIATOR + ) try: contents_dict = self._parse_elements( jingle_elt, @@ -1363,7 +1484,7 @@ class XEP_0166: request, client, True, - XEP_0166.ROLE_INITIATOR, + creator, ) except exceptions.CancelError: return @@ -1374,20 +1495,194 @@ class XEP_0166: return session["contents_new"] = contents_dict + # a former content-add may still be pending an accept + session.setdefault("added_content_names", []).extend(contents_dict) # at this point we can send the result to confirm reception of the request client.send(xmlstream.toResponse(request, "result")) - assert "jingle_elt" not in session + # the session may already have a jingle_elt (e.g. the session-accept one) session["jingle_elt"] = jingle_elt await self._call_plugins(client, XEP_0166.A_CONTENT_ADD, session, delete=False) + self._emit_content_change( + client, XEP_0166.A_CONTENT_ADD, session, list(contents_dict) + ) + + async def on_content_modify( + self, + client: SatXMPPEntity, + request: domish.Element, + jingle_elt: domish.Element, + session: dict[str, Any], + ) -> None: + """Called on content-modify action + + The "senders" attribute of the modified contents is updated and the frontend is + notified. An application can refuse the change from its "jingle_handler" + (reverting the "senders" attribute and sending a contrary content-modify). + @param client: the client instance + @param request: full request + @param jingle_elt: element + @param session: session data + """ + try: + contents_dict = self._parse_elements( + jingle_elt, + session, + request, + client, + False, + XEP_0166.ROLE_INITIATOR, + with_application=False, + with_transport=False, + ) + except exceptions.CancelError: + return + + # update the senders attribute of the modified content(s) + applied_names: list[str] = [] + for content_elt in jingle_elt.elements(NS_JINGLE, "content"): + name = content_elt["name"] + content_data = contents_dict[name] + content_data["senders"] = content_elt.attributes.get("senders", "both") + applied_names.append(name) + + # acknowledge the request (per spec, no content-accept for content-modify) + client.send(xmlstream.toResponse(request, "result")) + if not applied_names: + return + + session["modified_content_names"] = applied_names + session["jingle_elt"] = jingle_elt + await self._call_plugins( + client, XEP_0166.A_CONTENT_MODIFY, session, delete=False, elements=False + ) + self._emit_content_change(client, XEP_0166.A_CONTENT_MODIFY, session) + + async def on_content_accept( + self, + client: SatXMPPEntity, + request: domish.Element, + jingle_elt: domish.Element, + session: dict[str, Any], + ) -> None: + """Called on content-accept action (the peer accepts a content we added) + + The peer answer (description + transport) is stored in the session, and the + frontend is notified so it can set the remote description. + @param client: the client instance + @param request: full request + @param jingle_elt: element + @param session: session data + """ + try: + self._parse_elements( + jingle_elt, + session, + request, + client, + False, + XEP_0166.ROLE_RESPONDER, + with_application=True, + with_transport=True, + ) + except exceptions.CancelError: + return + + session["accepted_content_names"] = [ + elt["name"] for elt in jingle_elt.elements(NS_JINGLE, "content") + ] + + # acknowledge the request + client.send(xmlstream.toResponse(request, "result")) + + self._emit_content_change(client, XEP_0166.A_CONTENT_ACCEPT, session) + + async def _on_content_removal( + self, + client: SatXMPPEntity, + request: domish.Element, + jingle_elt: domish.Element, + session: dict[str, Any], + action: str, + ) -> None: + """Handle a content-reject or content-remove action + + @param client: the client entity. + @param request: full request. + @param jingle_elt: the element. + @param session: the session data. + @param action: the action (content-reject or content-remove). + """ + log.debug(f"{action} received for session {session['id']}") + try: + self._parse_elements( + jingle_elt, + session, + request, + client, + False, + XEP_0166.ROLE_INITIATOR, + with_application=False, + with_transport=False, + ) + except exceptions.CancelError: + return + + prefix = "removed" if action == self.A_CONTENT_REMOVE else "rejected" + names_key = f"{prefix}_content_names" + session[names_key] = [ + elt["name"] for elt in jingle_elt.elements(NS_JINGLE, "content") + ] + + # acknowledge the request + client.send(xmlstream.toResponse(request, "result")) + self._emit_content_change(client, action, session) + for content_name in session[names_key]: + del session["contents"][content_name] + + async def on_content_reject( + self, + client: SatXMPPEntity, + request: domish.Element, + jingle_elt: domish.Element, + session: dict[str, Any], + ) -> None: + """Called on content-reject action (the peer rejects a content we added) + + @param client: the client entity. + @param request: full request. + @param jingle_elt: the element. + @param session: the session data. + """ + await self._on_content_removal( + client, request, jingle_elt, session, self.A_CONTENT_REJECT + ) + + async def on_content_remove( + self, + client: SatXMPPEntity, + request: domish.Element, + jingle_elt: domish.Element, + session: dict[str, Any], + ) -> None: + """Called on content-remove action (the creator removes one of its contents) + + @param client: the client entity. + @param request: full request. + @param jingle_elt: the element. + @param session: the session data. + """ + await self._on_content_removal( + client, request, jingle_elt, session, self.A_CONTENT_REMOVE + ) async def on_transport_replace(self, client, request, jingle_elt, session): """A transport change is requested The request is parsed, and jingle_handler is called on concerned transport plugin(s) - @param client: %(doc_client)s + @param client: the client instance @param request(domish.Element): full request @param jingle_elt(domish.Element): the element @param session(dict): session data @@ -1540,7 +1835,7 @@ class XEP_0166: @param jingle_elt: the element @param session: session data """ - log.debug(f"Jingle session {session['id']} has been accepted") + log.debug(f"received a transport-info for Jingle session {session['id']}") try: parsed_contents = self._parse_elements( diff --git a/libervia/backend/plugins/plugin_xep_0166/models.py b/libervia/backend/plugins/plugin_xep_0166/models.py index 55706622..61cf67f2 100644 --- a/libervia/backend/plugins/plugin_xep_0166/models.py +++ b/libervia/backend/plugins/plugin_xep_0166/models.py @@ -19,8 +19,9 @@ import abc from dataclasses import dataclass -from typing import Awaitable, Callable, Union +from typing import Awaitable, Callable, Literal, Union +from pydantic import BaseModel, ConfigDict, Field, SerializeAsAny from twisted.internet import defer from twisted.words.xish import domish @@ -28,6 +29,50 @@ from libervia.backend.core import exceptions from libervia.backend.core.core_types import SatXMPPEntity from libervia.backend.core.i18n import _ +ContentChangeAction = Literal[ + "content-add", + "content-modify", + "content-accept", + "content-remove", + "content-reject", +] + + +class ContentChangePayload(BaseModel): + """Base payload of the ``jingle_content_change`` signal, to be subclassed""" + + model_config = ConfigDict(extra="allow") + + +class ContentChangeItem(BaseModel): + """One changed content of the ``jingle_content_change`` signal""" + + content: str = Field(description="Name of the concerned content") + senders: Literal["initiator", "responder", "both", "none"] = Field( + description="Senders value of the content" + ) + # "SerializeAsAny" allows the use of subclasses (cf. + # https://stackoverflow.com/q/77049825, thanks!)) + payload: SerializeAsAny[ContentChangePayload] = Field( + default_factory=ContentChangePayload, + description=( + "Application-specific payload, provided by the application handler " + "(``get_content_change_payload``)" + ), + ) + + +class ContentChange(BaseModel): + """Payload of the ``jingle_content_change`` signal.""" + + session_id: str = Field(description="ID of the Jingle session") + action: ContentChangeAction = Field( + description="Jingle action that triggered the change" + ) + changes: list[ContentChangeItem] = Field( + description="The changed contents of the stanza" + ) + class BaseApplicationHandler(abc.ABC): @@ -118,6 +163,16 @@ class BaseApplicationHandler(abc.ABC): """ pass + def get_content_change_payload( + self, client: SatXMPPEntity, action: str, session: dict, content_name: str + ) -> ContentChangePayload: + """Build the application-specific payload of the content change signal + + @return: payload to store in the ``payload`` field of the + ``jingle_content_change`` signal + """ + return ContentChangePayload() + @abc.abstractmethod def jingle_terminate( self, diff --git a/libervia/backend/plugins/plugin_xep_0167/__init__.py b/libervia/backend/plugins/plugin_xep_0167/__init__.py index e26a11d8..13fb1500 100644 --- a/libervia/backend/plugins/plugin_xep_0167/__init__.py +++ b/libervia/backend/plugins/plugin_xep_0167/__init__.py @@ -16,7 +16,7 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -from typing import Optional +from typing import Literal, Optional from twisted.internet import reactor from twisted.internet import defer @@ -37,8 +37,8 @@ from libervia.backend.tools.common import data_format from . import mapping from ..plugin_xep_0166 import BaseApplicationHandler +from .models import RTPContentChangePayload from .constants import ( - NS_AV_CONFERENCES, NS_JINGLE_RTP, NS_JINGLE_RTP_AUDIO, NS_JINGLE_RTP_INFO, @@ -81,8 +81,6 @@ class XEP_0167(BaseApplicationHandler): def __init__(self, host): log.info(f'Plugin "{PLUGIN_INFO[C.PI_NAME]}" initialization') self.host = host - # FIXME: to be removed once host is accessible from global var - mapping.host = host self._j = host.plugins["XEP-0166"] self._j.register_application(NS_JINGLE_RTP, self) host.register_namespace("jingle-rtp", NS_JINGLE_RTP) @@ -110,6 +108,30 @@ class XEP_0167(BaseApplicationHandler): method=self._call_answer_sdp, async_=False, ) + host.bridge.add_method( + "call_content_accept", + ".plugin", + in_sign="ssss", + out_sign="", + method=self._call_content_accept, + async_=False, + ) + host.bridge.add_method( + "call_content_modify", + ".plugin", + in_sign="ssss", + out_sign="", + method=self._call_content_modify, + async_=False, + ) + host.bridge.add_method( + "call_content_add", + ".plugin", + in_sign="sss", + out_sign="", + method=self._call_content_add, + async_=False, + ) host.bridge.add_method( "call_info", ".plugin", @@ -128,6 +150,8 @@ class XEP_0167(BaseApplicationHandler): # args: session_id, serialised setup data (dict with keys "role" and "sdp"), # profile + host.trigger.add("XEP-0166_content_change", self.on_content_change_trigger) + host.bridge.add_signal("call_setup", ".plugin", signature="sss") # args: session_id, serialised update data, profile @@ -142,6 +166,39 @@ class XEP_0167(BaseApplicationHandler): def get_handler(self, client): return XEP_0167_handler() + def get_content_change_payload( + self, client: SatXMPPEntity, action: str, session: dict, content_name: str + ) -> RTPContentChangePayload: + """Return payload of the content change signal (media and SDP)""" + content_data = session["contents"][content_name] + application_data = content_data["application_data"] + payload = RTPContentChangePayload(media=application_data["media"]) + if action == self._j.A_CONTENT_ADD: + # answer the content-add with the local description of the new media + payload.sdp = mapping.generate_content_add_sdp( + content_data, session, content_name + ) + payload.origin = application_data.get("origin") + elif action == self._j.A_CONTENT_ACCEPT: + # parse the peer answer (description + transport), so that the peer m-line can + # be generated + content_data["application_data"]["peer_data"] = mapping.parse_description( + content_data["desc_elt"] + ) + peer_ice_data = content_data["transport"].handler.parse_transport( + content_data["transport_elt"] + ) + content_data["transport_data"]["peer_ice_data"] = peer_ice_data + payload.sdp = mapping.generate_content_add_sdp( + content_data, session, content_name + ) + elif action == self._j.A_CONTENT_MODIFY: + # new direction to apply to the remote m-line (peer perspective) + payload.direction = mapping.senders_to_sdp( + content_data["senders"], session + ).removeprefix("a=") + return payload + # bridge methods def _call_start( @@ -168,12 +225,18 @@ class XEP_0167(BaseApplicationHandler): self.call_update(client, session_id, data_format.deserialise(call_data_s)) ) - def parse_call_data(self, call_data: dict) -> dict: - """Parse ``call_data`` and return corresponding contents end metadata""" + def parse_call_data( + self, call_data: dict, role: Literal["initiator", "responder"] + ) -> dict: + """Parse ``call_data`` and return corresponding contents and metadata + + @param call_data: call data (see [call_start] for the expected structure) + @param role: role of the entity which created the SDP + """ metadata = call_data.get("metadata") or {} if "sdp" in call_data: - sdp_data = mapping.parse_sdp(call_data["sdp"], self._j.ROLE_INITIATOR) + sdp_data = mapping.parse_sdp(call_data["sdp"], role) to_delete = set() for media, data in sdp_data.items(): if media not in ("audio", "video", "application"): @@ -261,8 +324,8 @@ class XEP_0167(BaseApplicationHandler): """Initiate a call session with the given peer. @param peer_jid: JID of the peer to initiate a call session with. - @param call_data: Dictionary containing data for the call. Must include SDP information. - The dict can have the following keys: + @param call_data: Dictionary containing data for the call. Must include SDP + information. The dict can have the following keys: - sdp (str): SDP data for the call. - metadata (dict): Additional metadata for the call (optional). Each media type ("audio" and "video") in the SDP should have: @@ -285,7 +348,7 @@ class XEP_0167(BaseApplicationHandler): "WebRTC is disabled when using Tor to avoid leaking IP address." ) - metadata = self.parse_call_data(call_data) + metadata = self.parse_call_data(call_data, self._j.ROLE_INITIATOR) contents = self.get_contents(call_data, metadata) if not contents: raise exceptions.DataError("no valid media data found: {call_data}") @@ -312,65 +375,82 @@ class XEP_0167(BaseApplicationHandler): client: SatXMPPEntity, session_id: str, call_data: dict, - ) -> None: - """Update a running call session. + role: Literal["initiator", "responder"] | None = None, + jingle_child_elts: list[domish.Element] | None = None, + ) -> list[str]: + """Add new contents to a running call session. @param session_id: ID of the Jingle session to update. - @param call_data: Dictionary containing updated data for the call. Must include SDP information. - The dict can have the following keys: - - sdp (str): SDP data for the call. - - metadata (dict): Additional metadata for the call (optional). - Each media type ("audio" and "video") in the SDP should have: - - application_data (dict): Data about the media. - - fingerprint (str): Security fingerprint data (optional). - - id (str): Identifier for the media (optional). - - ice-candidates: ICE candidates for media transport. - - And other transport specific data. - - - @raises exceptions.DataError: If media data is invalid or duplicate content name - (mid) is found. + @param call_data: offered data, must include an "sdp" key (see call_start). + @param role: role of the entity which created the SDP. + @param jingle_child_elts: elements to add to the element. + @return: names of the contents added to the session. """ session = self._j.get_session(client, session_id) - try: - new_offer_sdp = call_data["sdp"] - except KeyError: + if "sdp" not in call_data: raise exceptions.DataError(f"New SDP offer is missing: {call_data}") - metadata = self.parse_call_data(call_data) + metadata = self.parse_call_data(call_data, role or session["role"]) contents = self.get_contents(call_data, metadata) if not contents: - raise exceptions.DataError("no valid media data found: {call_data}") + raise exceptions.DataError(f"no valid media data found: {call_data}") - call_type = ( - C.META_SUBTYPE_CALL_VIDEO - if "video" in call_data - else C.META_SUBTYPE_CALL_AUDIO - ) + iq_elt = None + jingle_elt = None + added: list[str] = [] for content_args in contents: - content = content_args["app_kwargs"] - content["app_ns"] = NS_JINGLE_RTP - content["name"] = (content_args["name"],) - content["transport_type"] = self._j.TRANSPORT_DATAGRAM - media = content["media"] - media_data = content["media_data"].copy() - media_data["transport_data"] = content_args["transport_data"][ - "local_ice_data" - ] - desc_elt = mapping.build_description(media, media_data, {}) - iq_elt, __ = self._j.build_action( + name = content_args["name"] + if name is None: + raise exceptions.DataError( + f"a media is missing its ID (mid): {content_args}" + ) + if name in session["contents"]: + log.debug(f"content {name!r} already exists, skipping") + continue + media = content_args["app_kwargs"]["media"] + media_data = content_args["app_kwargs"]["media_data"] + ice_data = content_args["transport_data"]["local_ice_data"] + desc_elt = mapping.build_description(media, media_data, session) + content_data = self._j.get_content_data(content_args) + transport = self._j.get_transport(client, content_args, content_data) + transport_elt = transport.handler.build_transport(ice_data) + iq_elt, context_elt = self._j.build_action( client, self._j.A_CONTENT_ADD, session, - content_args["name"], + name, context_elt=desc_elt, + iq_elt=iq_elt, ) - content_data = self._j.get_content_data(content) - transport = self._j.get_transport(client, content, content_data) - transport_elt = transport.handler.build_transport( - media_data["transport_data"] + content_elt = context_elt.parent + jingle_elt = content_elt.parent + content_elt["senders"] = content_args.get("senders", "both") + content_elt.addChild(transport_elt) + session["contents"][name] = { + "application": content_data.application, + "application_data": { + "media": media, + "local_data": media_data, + }, + "transport": transport, + "transport_data": content_args["transport_data"], + "creator": session["role"], + "senders": content_args.get("senders", "both"), + } + added.append(name) + if iq_elt is not None: + assert jingle_elt is not None + for child in jingle_child_elts or []: + jingle_elt.addChild(child) + self.host.trigger.point( + "XEP-0167_content_add_elt_built", + client, + session, + jingle_elt, + added, + triggers_no_cancel=True, ) - iq_elt.jingle.content.addChild(transport_elt) await iq_elt.send() + return added def _call_answer_sdp(self, session_id: str, answer_sdp: str, profile: str) -> None: client = self.host.get_client(profile) @@ -383,6 +463,150 @@ class XEP_0167(BaseApplicationHandler): ) answer_sdp_d.callback(answer_sdp) + def _call_content_accept( + self, session_id: str, content_name: str, mline_sdp: str, profile: str + ) -> None: + """Send a content-accept IQ to the peer (in response to a content-add) + + The frontend answers the m-line offered by the peer, and sends it back. It is + converted to Jingle and elements. + + @param session_id: Jingle session ID + @param content_name: name of the content added by the peer + @param mline_sdp: SDP m-line of the frontend answer for the new media + @param profile: profile key + """ + client = self.host.get_client(profile) + session = self._j.get_session(client, session_id) + try: + pending_names = session["added_content_names"] + pending_names.remove(content_name) + except (KeyError, ValueError): + raise exceptions.NotFound( + f"Content {content_name!r} is not pending in session {session_id!r}" + ) + if not pending_names: + del session["added_content_names"] + + content_data = session["contents"][content_name] + media = content_data["application_data"]["media"] + + # the answer is from our perspective, whatever our role is in the session + _, media_data = mapping.parse_mline(mline_sdp, session["role"]) + local_data = media_data["application_data"] + local_ice_data = media_data["transport_data"] + + # store the local data in the session (used for SDP generation) + content_data["application_data"]["local_data"] = local_data + content_data["transport_data"]["local_ice_data"] = local_ice_data + + desc_elt = mapping.build_description(media, local_data, session) + transport_elt = content_data["transport"].handler.build_transport(local_ice_data) + + iq_elt, jingle_elt = self._j._build_jingle_elt( + client, session, self._j.A_CONTENT_ACCEPT + ) + content_elt = jingle_elt.addElement("content", self._j.namespace) + # the content has been created by the peer + content_elt["creator"] = content_data.get("creator", self._j.ROLE_INITIATOR) + content_elt["name"] = content_name + content_elt["senders"] = content_data["senders"] + content_elt.addChild(desc_elt) + content_elt.addChild(transport_elt) + + client.send(iq_elt) + + def _call_content_modify( + self, session_id: str, media: str, senders: str, profile: str + ) -> None: + """Send a content-modify IQ to the peer + + Used when the frontend wants to also send media on a content added by the peer. + + @param session_id: Jingle session ID + @param media: media type of the content to modify (e.g. "video") + @param senders: new senders value (e.g. "both") + @param profile: profile key + """ + client = self.host.get_client(profile) + session = self._j.get_session(client, session_id) + content_name: str | None = None + for name, content_data in session["contents"].items(): + if content_data["application_data"].get("media") == media: + content_name = name + break + if content_name is None: + raise exceptions.NotFound( + f"No content with media {media!r} in session {session_id!r}" + ) + content_data = session["contents"][content_name] + + iq_elt, jingle_elt = self._j._build_jingle_elt( + client, session, self._j.A_CONTENT_MODIFY + ) + content_elt = jingle_elt.addElement("content", self._j.namespace) + content_elt["creator"] = content_data.get("creator", self._j.ROLE_INITIATOR) + content_elt["name"] = content_name + content_elt["senders"] = senders + client.send(iq_elt) + + def _call_content_add(self, session_id: str, mline_sdp: str, profile: str) -> None: + """Send a content-add IQ to the peer (add a new media content to the session) + + @param session_id: Jingle session ID + @param mline_sdp: SDP m-line of the new media (our offer) + @param profile: profile key + """ + client = self.host.get_client(profile) + session = self._j.get_session(client, session_id) + + media, media_data = mapping.parse_mline(mline_sdp, session["role"]) + local_data = media_data["application_data"] + local_ice_data = media_data["transport_data"] + + # generate a unique content name (a content-add must not reuse an + # existing name, and the session entry must not be overwritten) + content_name = media + suffix = 2 + while content_name in session["contents"]: + content_name = f"{media}_{suffix}" + suffix += 1 + + # The transport must be a registered one (not None): contents are iterated on + # plugin handler calls, where the transport handler is accessed + content_data = self._j.get_content_data( + { + "app_ns": NS_JINGLE_RTP, + "name": content_name, + "transport_data": {"local_ice_data": local_ice_data}, + } + ) + session["contents"][content_name] = content_data = { + "application": self._j.get_application(NS_JINGLE_RTP), + "application_data": {"media": media, "local_data": local_data}, + "transport": self._j.get_transport( + client, {"transport_type": self._j.TRANSPORT_DATAGRAM}, content_data + ), + "transport_data": {"local_ice_data": local_ice_data}, + "creator": session["role"], + "senders": session["role"], + } + + desc_elt = mapping.build_description(media, local_data, session) + transport_elt = content_data["transport"].handler.build_transport(local_ice_data) + + iq_elt, jingle_elt = self._j._build_jingle_elt( + client, session, self._j.A_CONTENT_ADD + ) + content_elt = jingle_elt.addElement("content", self._j.namespace) + content_elt["creator"] = session["role"] + content_elt["name"] = content_name + content_elt["senders"] = session["role"] + content_elt.addChild(desc_elt) + content_elt.addChild(transport_elt) + + client.send(iq_elt) + def _call_end( self, session_id: str, @@ -406,7 +630,12 @@ class XEP_0167(BaseApplicationHandler): @param data: optional extra data, may be used to indicate the reason to end the call """ - session = self._j.get_session(client, session_id) + try: + session = self._j.get_session(client, session_id) + except exceptions.NotFound: + # the session is already gone (e.g. duplicated call_end): nothing to do + log.warning(f"No session with SID {session_id} found, call_end ignored.") + return await self._j.terminate(client, self._j.REASON_SUCCESS, session) # jingle callbacks @@ -644,6 +873,9 @@ class XEP_0167(BaseApplicationHandler): def propagate_data(self, session: dict, parsed_answer: dict) -> None: """Propagate local SDP data to other contents""" for media in ("audio", "video", "application"): + if media not in parsed_answer: + # the frontend may not answer all the contents + continue for content in session["contents"].values(): try: application_data = content["application_data"] @@ -723,9 +955,14 @@ class XEP_0167(BaseApplicationHandler): raise exceptions.ConflictError( f"There is already a {content_name!r} content." ) + # the content will be answered with a content-accept + content_data["added_by_peer"] = True current_contents[content_name] = content_data application_data["media"] = desc_elt["media"] application_data["peer_data"] = mapping.parse_description(desc_elt) + elif action == self._j.A_CONTENT_MODIFY: + # the senders attribute has been updated in "on_content_modify" + pass else: log.warning(f"FIXME: unmanaged action {action}") @@ -740,6 +977,32 @@ class XEP_0167(BaseApplicationHandler): ) return desc_elt + def on_content_change_trigger( + self, + client: SatXMPPEntity, + action: str, + session: dict, + content_names: list[str], + ) -> None: + """Pass the peer answer SDP of the accepted contents to the session. + + @param client: the client entity. + @param action: the content action (content-accept). + @param session: the session data. If the "content_accept_cb" entry is set, it is + called with the peer answer SDP of the accepted contents; a BUNDLE group is + added when several contents are accepted at once. + @param content_names: names of the accepted contents. + """ + if action != self._j.A_CONTENT_ACCEPT: + return + cb = session.get("content_accept_cb") + if cb is None: + return + sdp = mapping.generate_sdp_from_session(session, content_names=content_names) + result = cb(client, session, sdp, content_names) + if result is not None: + defer.ensureDeferred(result) + def jingle_session_info( self, client: SatXMPPEntity, @@ -811,7 +1074,6 @@ class XEP_0167_handler(XMPPHandler): disco.DiscoFeature(NS_JINGLE_RTP), disco.DiscoFeature(NS_JINGLE_RTP_AUDIO), disco.DiscoFeature(NS_JINGLE_RTP_VIDEO), - disco.DiscoFeature(NS_AV_CONFERENCES), ] def getDiscoItems(self, requestor, target, nodeIdentifier=""): diff --git a/libervia/backend/plugins/plugin_xep_0167/mapping.py b/libervia/backend/plugins/plugin_xep_0167/mapping.py index 3eef9536..2de45428 100644 --- a/libervia/backend/plugins/plugin_xep_0167/mapping.py +++ b/libervia/backend/plugins/plugin_xep_0167/mapping.py @@ -17,10 +17,11 @@ # along with this program. If not, see . import base64 -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Literal, Optional, Tuple from twisted.words.xish import domish +from libervia.backend import G from libervia.backend.core.constants import Const as C from libervia.backend.core.log import getLogger @@ -28,11 +29,13 @@ from .constants import NS_JINGLE_RTP log = getLogger(__name__) -host = None - def senders_to_sdp(senders: str, session: dict) -> str: - """Returns appropriate SDP attribute corresponding to Jingle senders attribute""" + """Returns the SDP direction attribute corresponding to a Jingle senders attribute + + @param senders: value of the Jingle "senders" attribute. + @param session: session data (the "role" key is used). + """ if senders == "both": return "a=sendrecv" elif senders == "none": @@ -43,7 +46,12 @@ def senders_to_sdp(senders: str, session: dict) -> str: return "a=recvonly" -def generate_sdp_from_session(session: dict, local: bool = False, port: int = 9) -> str: +def generate_sdp_from_session( + session: dict, + local: bool = False, + port: int = 9, + content_names: list[str] | None = None, +) -> str: """Generate an SDP string from session data. @param session: A dictionary containing the session data. It should have the @@ -73,10 +81,12 @@ def generate_sdp_from_session(session: dict, local: bool = False, port: int = 9) otherwise for the peer entity. Generally the local SDP is received from frontends and not needed in backend, except for debugging purpose. @param port: The preferred port for communications. - + @param content_names: if set, only the contents with these names are included. @return: The generated SDP string. """ contents = session["contents"] + if content_names is not None: + contents = {name: contents[name] for name in content_names} sdp_lines = ["v=0"] # Add originator (o=) line after the version (v=) line @@ -97,18 +107,19 @@ def generate_sdp_from_session(session: dict, local: bool = False, port: int = 9) sdp_lines.append("t=0 0") # stream direction - all_senders = {c["senders"] for c in session["contents"].values()} + all_senders = {c["senders"] for c in contents.values()} # if we don't have a common senders for all contents, we set them at media level senders = all_senders.pop() if len(all_senders) == 1 else None sdp_lines.append("a=msid-semantic:WMS *") sdp_lines.append("a=ice-options:trickle") - host.trigger.point( + G.host.trigger.point( "XEP-0167_generate_sdp_session", session, local, sdp_lines, + content_names, triggers_no_cancel=True, ) @@ -196,7 +207,7 @@ def generate_sdp_from_session(session: dict, local: bool = False, port: int = 9) crypto_line += f" {session_params}" sdp_lines.append(crypto_line) - host.trigger.point( + G.host.trigger.point( "XEP-0167_generate_sdp_content", session, local, @@ -214,6 +225,62 @@ def generate_sdp_from_session(session: dict, local: bool = False, port: int = 9) return "\r\n".join(sdp_lines) + "\r\n" +def generate_content_add_sdp(content_data: dict, session: dict, content_name: str) -> str: + """Generate the SDP m-line of a content, from the peer description and + transport data, to be used as remote description by the frontend + + @param content_data: The content data, as stored in ``session["contents"]``. + @param session: session data. + @param content_name: name of the content (used as media ID, as per JSEP). + @return: The SDP m-line (with its a= lines), without the session header. + """ + application_data = content_data["application_data"] + media = application_data["media"] + media_data = application_data["peer_data"] + ice_data = content_data["transport_data"]["peer_ice_data"] + payload_types = media_data.get("payload_types", {}) + + sdp_lines = [ + f"m={media} 9 UDP/TLS/RTP/SAVPF " + + " ".join(str(pt_id) for pt_id in payload_types) + ] + sdp_lines.append(f"a=mid:{content_name}") + sdp_lines.append("c=IN IP4 0.0.0.0") + sdp_lines.append(senders_to_sdp(content_data["senders"], session)) + + for pt_id, pt in payload_types.items(): + name = pt["name"] + clockrate = pt.get("clockrate", "") + channels = pt.get("channels") + if channels: + sdp_lines.append(f"a=rtpmap:{pt_id} {name}/{clockrate}/{channels}") + else: + sdp_lines.append(f"a=rtpmap:{pt_id} {name}/{clockrate}") + if "parameters" in pt: + fmtp_params = ";".join(f"{k}={v}" for k, v in pt["parameters"].items()) + sdp_lines.append(f"a=fmtp:{pt_id} {fmtp_params}") + + if media_data.get("rtcp-mux"): + sdp_lines.append("a=rtcp-mux") + + fingerprint_data = ice_data.get("fingerprint") + if fingerprint_data: + sdp_lines.append( + f"a=fingerprint:{fingerprint_data['hash']} " + f"{fingerprint_data['fingerprint']}" + ) + if "setup" in fingerprint_data: + sdp_lines.append(f"a=setup:{fingerprint_data['setup']}") + + sdp_lines.append(f"a=ice-ufrag:{ice_data['ufrag']}") + sdp_lines.append(f"a=ice-pwd:{ice_data['pwd']}") + + for candidate in ice_data.get("candidates", []): + sdp_lines.append(f"a={generate_candidate_line(candidate)}") + + return "\r\n".join(sdp_lines) + "\r\n" + + def generate_candidate_line(candidate: dict) -> str: """Generate a ``candidate:`` attribute line from candidate data. @@ -274,30 +341,39 @@ def parse_candidate(parts: list[str]) -> dict: return candidate -def parse_sdp(sdp: str, role: str) -> dict: - """Parse SDP string. +def parse_mline( + mline_sdp: str, + role: Literal["initiator", "responder"], + session_attributes: Optional[Dict[str, Any]] = None, + call_data: Optional[Dict[str, Any]] = None, + metadata: Optional[Dict[str, Any]] = None, +) -> Tuple[str, Dict[str, Any]]: + """Parse an SDP media block (m= line and its a= lines) - @param sdp: The SDP string to parse. - @param role: Role of the entities which produces the SDP. - @return: A dictionary containing parsed session data. + Can be used on a media section of a full SDP (see parse_sdp) or on a + standalone m-line block (as sent by the frontend). + @param mline_sdp: the m= line and its a= lines + @param role: role of the entity which produces the SDP + @param session_attributes: session-level attributes to apply to the media + @param call_data: passed to the ``XEP-0167_parse_sdp_a`` trigger + @param metadata: passed to the ``XEP-0167_parse_sdp_a`` trigger + @return: (media type, media data) """ - assert host is not None - lines = sdp.strip().split("\r\n") - metadata: Dict[str, Any] = {} - call_data = {"metadata": metadata} - session_attributes: Dict[str, Any] = {} + if call_data is None: + call_data = {"metadata": metadata if metadata is not None else {}} + + senders = "both" + for attr, value in (session_attributes or {}).items(): + if attr in ("sendrecv", "sendonly", "recvonly", "inactive"): + senders = value media_type = None media_data: Optional[Dict[str, Any]] = None application_data: Optional[Dict[str, Any]] = None transport_data: Optional[Dict[str, Any]] = None - fingerprint_data: Optional[Dict[str, str]] = None - ice_pwd: Optional[str] = None - ice_ufrag: Optional[str] = None payload_types: Optional[Dict[int, dict]] = None - senders: str = "both" - for line in lines: + for line in mline_sdp.strip().split("\r\n"): try: parts = line.split() prefix = parts[0][:2] @@ -319,22 +395,22 @@ def parse_sdp(sdp: str, role: str) -> dict: ) transport_data = {"port": port} - if fingerprint_data is not None: - transport_data["fingerprint"] = fingerprint_data - if ice_pwd is not None: - transport_data["pwd"] = ice_pwd - if ice_ufrag is not None: - transport_data["ufrag"] = ice_ufrag - media_data = call_data[media_type] = { + media_data = { "application_data": application_data, "transport_data": transport_data, "senders": senders, } - # Apply session attributes to the new media - for attr, value in session_attributes.items(): - if attr in ("fingerprint", "ice-pwd", "ice-ufrag"): - transport_data[attr] = value + # Apply session attributes to the new media, unless overridden at + # the media level (RFC 4566 §5.7). The fingerprint dict is copied, + # as the media-level a= lines update it in place. + for attr, value in (session_attributes or {}).items(): + if attr == "fingerprint": + transport_data[attr] = dict(value) + elif attr == "ice-pwd": + transport_data["pwd"] = value + elif attr == "ice-ufrag": + transport_data["ufrag"] = value elif attr in ("sendrecv", "sendonly", "recvonly", "inactive"): application_data["senders"] = value else: @@ -346,31 +422,7 @@ def parse_sdp(sdp: str, role: str) -> dict: else: attribute = parts[0] - if media_type is None: - # This is a session-level attribute - if attribute == "fingerprint": - algorithm, fingerprint = parts[0], parts[1] - session_attributes["fingerprint"] = { - "hash": algorithm, - "fingerprint": fingerprint, - } - elif attribute == "ice-pwd": - session_attributes["ice-pwd"] = parts[0] - elif attribute == "ice-ufrag": - session_attributes["ice-ufrag"] = parts[0] - elif attribute in ("sendrecv", "sendonly", "recvonly", "inactive"): - if attribute == "sendrecv": - value = "both" - elif attribute == "sendonly": - value = role - elif attribute == "recvonly": - value = "responder" if role == "initiator" else "initiator" - else: - value = "none" - session_attributes[attribute] = value - else: - session_attributes[attribute] = parts[0] if parts else True - else: + if media_type is not None: # This is a media-level attribute if attribute == "mid": assert media_data is not None @@ -424,10 +476,9 @@ def parse_sdp(sdp: str, role: str) -> dict: elif attribute == "fingerprint": algorithm, fingerprint = parts[0], parts[1] - fingerprint_data = {"hash": algorithm, "fingerprint": fingerprint} if transport_data is not None: transport_data.setdefault("fingerprint", {}).update( - fingerprint_data + {"hash": algorithm, "fingerprint": fingerprint} ) elif attribute == "setup": assert transport_data is not None @@ -466,7 +517,7 @@ def parse_sdp(sdp: str, role: str) -> dict: else: application_data["senders"] = value - host.trigger.point( + G.host.trigger.point( "XEP-0167_parse_sdp_a", attribute, parts, @@ -483,6 +534,93 @@ def parse_sdp(sdp: str, role: str) -> dict: except IndexError as e: raise IndexError(f"Incomplete line. Missing data: {line}") from e + assert media_type is not None and media_data is not None + return media_type, media_data + + +def parse_sdp(sdp: str, role: Literal["initiator", "responder"]) -> dict: + """Parse SDP string. + + @param sdp: The SDP string to parse. + @param role: Role of the entities which produces the SDP. + @return: A dictionary containing parsed session data. + """ + lines = sdp.strip().split("\r\n") + metadata: Dict[str, Any] = {} + call_data = {"metadata": metadata} + session_attributes: Dict[str, Any] = {} + media_block: List[str] = [] + + def flush_media_block() -> None: + if media_block: + media_type, media_data = parse_mline( + "\r\n".join(media_block), role, session_attributes, call_data, metadata + ) + call_data[media_type] = media_data + media_block.clear() + + for line in lines: + if line.startswith("m="): + flush_media_block() + media_block.append(line) + elif media_block and line.startswith("a="): + media_block.append(line) + else: + # session-level line + try: + parts = line.split() + prefix = parts[0][:2] + parts[0] = parts[0][2:] + + if prefix == "a=": + if ":" in parts[0]: + attribute, parts[0] = parts[0].split(":", 1) + else: + attribute = parts[0] + + if attribute == "fingerprint": + algorithm, fingerprint = parts[0], parts[1] + session_attributes["fingerprint"] = { + "hash": algorithm, + "fingerprint": fingerprint, + } + elif attribute == "ice-pwd": + session_attributes["ice-pwd"] = parts[0] + elif attribute == "ice-ufrag": + session_attributes["ice-ufrag"] = parts[0] + elif attribute in ("sendrecv", "sendonly", "recvonly", "inactive"): + if attribute == "sendrecv": + value = "both" + elif attribute == "sendonly": + value = role + elif attribute == "recvonly": + value = "responder" if role == "initiator" else "initiator" + else: + value = "none" + session_attributes[attribute] = value + else: + session_attributes[attribute] = parts[0] if parts else True + + G.host.trigger.point( + "XEP-0167_parse_sdp_a", + attribute, + parts, + call_data, + metadata, + None, + None, + None, + triggers_no_cancel=True, + ) + except ValueError as e: + raise ValueError( + f"Could not parse line. Invalid format ({e}): {line}" + ) from e + except IndexError as e: + raise IndexError(f"Incomplete line. Missing data: {line}") from e + + flush_media_block() + # we remove private data (data starting with _, used by some plugins (e.g. XEP-0294) # to handle session data at media level)) for key in [k for k in call_data if k.startswith("_")]: @@ -536,9 +674,6 @@ def build_description(media: str, media_data: dict, session: dict) -> domish.Ele @return: A element. """ - # FIXME: to be removed once host is accessible from global var - assert host is not None - desc_elt = domish.Element((NS_JINGLE_RTP, "description"), attribs={"media": media}) for pt_id, pt_data in media_data.get("payload_types", {}).items(): @@ -553,7 +688,7 @@ def build_description(media: str, media_data: dict, session: dict) -> domish.Ele param_elt = payload_type_elt.addElement("parameter") param_elt["name"] = param_name param_elt["value"] = param_value - host.trigger.point( + G.host.trigger.point( "XEP-0167_build_description_payload_type", desc_elt, media_data, @@ -580,7 +715,7 @@ def build_description(media: str, media_data: dict, session: dict) -> domish.Ele if attr in enc_data: crypto_elt[attr] = enc_data[attr] - host.trigger.point( + G.host.trigger.point( "XEP-0167_build_description", desc_elt, media_data, @@ -597,9 +732,6 @@ def parse_description(desc_elt: domish.Element) -> dict: @param desc_elt: element @return: media data as in [build_description] """ - # FIXME: to be removed once host is accessible from global var - assert host is not None - media_data = {} if desc_elt.hasAttribute("ssrc"): media_data.setdefault("ssrc", {})[desc_elt["ssrc"]] = {} @@ -637,7 +769,7 @@ def parse_description(desc_elt: domish.Element) -> dict: if parameters: payload_type_data["parameters"] = parameters - host.trigger.point( + G.host.trigger.point( "XEP-0167_parse_description_payload_type", desc_elt, media_data, @@ -688,7 +820,7 @@ def parse_description(desc_elt: domish.Element) -> dict: if encryption_data: media_data["encryption"] = encryption_data - host.trigger.point( + G.host.trigger.point( "XEP-0167_parse_description", desc_elt, media_data, triggers_no_cancel=True ) diff --git a/libervia/backend/plugins/plugin_xep_0167/models.py b/libervia/backend/plugins/plugin_xep_0167/models.py new file mode 100644 index 00000000..9c364676 --- /dev/null +++ b/libervia/backend/plugins/plugin_xep_0167/models.py @@ -0,0 +1,30 @@ +#!/usr/bin/env python3 + +# Libervia plugin for Jingle RTP (XEP-0167) +# Copyright (C) 2009-2026 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 libervia.backend.plugins.plugin_xep_0166.models import ContentChangePayload + + +class RTPContentChangePayload(ContentChangePayload): + """Payload for the Jingle RTP application""" + + media: str + sdp: str | None = None + # new direction of the content (content-modify), from the peer perspective + direction: str | None = None + # origin of the added content, an entity URL (e.g. in a conference) + origin: str | None = None diff --git a/libervia/backend/plugins/plugin_xep_0176.py b/libervia/backend/plugins/plugin_xep_0176.py index 2dda5b24..1ad4c162 100644 --- a/libervia/backend/plugins/plugin_xep_0176.py +++ b/libervia/backend/plugins/plugin_xep_0176.py @@ -16,7 +16,7 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -from typing import Dict, List, Optional +from typing import List import uuid from twisted.internet import defer @@ -253,39 +253,41 @@ class XEP_0176(BaseTransportHandler): @param session: Jingle session. @param content_name: Name (ID) of the content. @param content_data: Data of the content. - @param media_type: Type of media of the content. """ media_type = content_data["application_data"].get("media") # we check if we have any buffered ICE candidates, and send them if it's the case - try: - buffer = session["XEP-0176_ice_buffer"] - buffered_ice_data = buffer.pop(media_type) - except KeyError: - pass - else: - if not buffer: - del session["XEP-0176_ice_buffer"] - local_jid = session["local_jid"] - peer_jid = session["peer_jid"] - log.debug(f"Replaying ICE buffer from {local_jid} to {peer_jid}") - transport_elt = self.build_transport(buffered_ice_data) - iq_elt, __ = self._j.build_action( - client, - self._j.A_TRANSPORT_INFO, - session, - content_name, - context_elt=transport_elt, - ) - self.host.trigger.point( - "XEP-0176_jingle_handler_send_buffer", - client, - session, - content_name, - content_data, - transport_elt, - iq_elt, - ) - await iq_elt.send() + buffer = session.get("XEP-0176_ice_buffer") + if not buffer: + return + buffered_ice_data = buffer.pop(content_name, None) + if buffered_ice_data is None: + # legacy frontends key the buffer by media type + buffered_ice_data = buffer.pop(media_type, None) + if buffered_ice_data is None: + return + if not buffer: + del session["XEP-0176_ice_buffer"] + local_jid = session["local_jid"] + peer_jid = session["peer_jid"] + log.debug(f"Replaying ICE buffer from {local_jid} to {peer_jid}") + transport_elt = self.build_transport(buffered_ice_data) + iq_elt, __ = self._j.build_action( + client, + self._j.A_TRANSPORT_INFO, + session, + content_name, + context_elt=transport_elt, + ) + self.host.trigger.point( + "XEP-0176_jingle_handler_send_buffer", + client, + session, + content_name, + content_data, + transport_elt, + iq_elt, + ) + await iq_elt.send() async def jingle_handler( self, @@ -308,10 +310,19 @@ class XEP_0176(BaseTransportHandler): """ content_data = session["contents"][content_name] transport_data = content_data["transport_data"] - if action in (self._j.A_PREPARE_CONFIRMATION, self._j.A_PREPARE_INITIATOR): + if action in ( + self._j.A_PREPARE_CONFIRMATION, + self._j.A_PREPARE_INITIATOR, + # a content-add comes with the peer transport, we parse it as well + self._j.A_CONTENT_ADD, + ): peer_ice_data = self.parse_transport(transport_elt) transport_data["peer_ice_data"] = peer_ice_data + elif action == self._j.A_CONTENT_MODIFY: + # only the senders attribute is changed, nothing to do here + pass + elif action == self._j.A_ACCEPTED_ACK: buffer = session.pop("XEP-0176_handler_buffer", None) if buffer: @@ -343,7 +354,6 @@ class XEP_0176(BaseTransportHandler): buffer.append([client, action, session, content_name, transport_elt]) return transport_elt - media_type = content_data["application_data"].get("media") new_ice_data = self.parse_transport(transport_elt) restart = self.update_candidates(transport_data, new_ice_data, local=False) if restart: @@ -354,7 +364,9 @@ class XEP_0176(BaseTransportHandler): self.host.bridge.ice_restart(session["id"], "peer", client.profile) ice_candidates_new_cb = session.get("ice_candidates_new_cb") - ice_candidates_data = {media_type: new_ice_data} + # the content name is used as key, as it is unique in the session (unlike the + # media type, which can be duplicated) + ice_candidates_data = {content_name: new_ice_data} if ice_candidates_new_cb is None: self.host.bridge.ice_candidates_new( session["id"], @@ -409,75 +421,85 @@ class XEP_0176(BaseTransportHandler): return False async def ice_candidates_add( - self, client: SatXMPPEntity, session_id: str, media_ice_data: dict[str, dict] + self, client: SatXMPPEntity, session_id: str, content_ice_data: dict[str, dict] ) -> None: - """Called when a new ICE candidates are available for a session + """Called when new ICE candidates are available for a session - @param session_id: Session ID - @param media_ice_data: a map from media type (audio, video) to ICE data - ICE data must be in the same format as in [self.parse_transport] + @param content_ice_data: a map from content name to ICE data (see + [self.parse_transport]); a media type can also be used as key. """ + # FIXME: frontends still use media types as key, they need to be updated session = self._j.get_session(client, session_id) - iq_elt: Optional[domish.Element] = None - content_name: str | None = None - content_data: dict | None = None + # FIXME: we use one transport-info IQ per content, as multi-content stanzas are + # rejected by some clients (e.g. Dino) + iqs: list[tuple[dict, str, dict, domish.Element]] = [] + + for key, new_ice_data in content_ice_data.items(): + if key in session["contents"]: + content_names = [key] + else: + # legacy: key is a media type + content_names = [ + name + for name, content_data in session["contents"].items() + if content_data["application_data"].get("media") == key + ] + if not content_names: + log.warning(f"no content found for {key!r}") + continue - for media_type, new_ice_data in media_ice_data.items(): if session["state"] == self._j.STATE_PENDING: log.debug(f"session not active, buffering") buffer = session.setdefault("XEP-0176_ice_buffer", {}) - media_buffer = buffer.setdefault(media_type, {}) + for content_name in content_names: + content_buffer = buffer.setdefault(content_name, {}) - for key in ["ufrag", "pwd"]: - if key not in media_buffer: - media_buffer[key] = new_ice_data[key] - else: - if media_buffer[key] != new_ice_data[key]: - log.warning( - f"{key} conflict, new value will replace old one\n" - f"buffer={media_buffer[key]!r}\n" - f"new={new_ice_data[key]!r}" - ) - media_buffer[key] = new_ice_data[key] + for ice_key in ["ufrag", "pwd"]: + if ice_key not in content_buffer: + content_buffer[ice_key] = new_ice_data[ice_key] + else: + if content_buffer[ice_key] != new_ice_data[ice_key]: + log.warning( + f"{ice_key} conflict, new value will replace old" + " one\n" + f"buffer={content_buffer[ice_key]!r}\n" + f"new={new_ice_data[ice_key]!r}" + ) + content_buffer[ice_key] = new_ice_data[ice_key] - media_buffer.setdefault("candidates", []).extend( - new_ice_data["candidates"] - ) + content_buffer.setdefault("candidates", []).extend( + new_ice_data["candidates"] + ) continue - for content_name, content_data in session["contents"].items(): - if content_data["application_data"].get("media") == media_type: - break - else: - log.warning(f"no media of type {media_type} has been found") - continue - restart = self.update_candidates( - content_data["transport_data"], new_ice_data, True - ) - if restart: - log.debug( - f"Local ICE restart detected on session {session['id']} " - f"[{client.profile}]" + for content_name in content_names: + content_data = session["contents"][content_name] + restart = self.update_candidates( + content_data["transport_data"], new_ice_data, True ) - self.host.bridge.ice_restart(session["id"], "local", client.profile) - transport_elt = self.build_transport(new_ice_data) - iq_elt, __ = self._j.build_action( - client, - self._j.A_TRANSPORT_INFO, - session, - content_name, - iq_elt=iq_elt, - context_elt=transport_elt, - ) + if restart: + log.debug( + f"Local ICE restart detected on session {session['id']} " + f"[{client.profile}]" + ) + self.host.bridge.ice_restart(session["id"], "local", client.profile) + transport_elt = self.build_transport(new_ice_data) + iq_elt, __ = self._j.build_action( + client, + self._j.A_TRANSPORT_INFO, + session, + content_name, + context_elt=transport_elt, + ) + iqs.append((new_ice_data, content_name, content_data, iq_elt)) - if iq_elt is not None: - assert content_name is not None and content_data is not None + for new_ice_data, content_name, content_data, iq_elt in iqs: try: self.host.trigger.point( "XEP-0176_ice_candidate_send", client, session, - media_ice_data, + new_ice_data, content_name, content_data, iq_elt, diff --git a/libervia/backend/plugins/plugin_xep_0338.py b/libervia/backend/plugins/plugin_xep_0338.py index b0f1d5bf..2a637ffa 100644 --- a/libervia/backend/plugins/plugin_xep_0338.py +++ b/libervia/backend/plugins/plugin_xep_0338.py @@ -58,6 +58,9 @@ class XEP_0338: host.trigger.add( "XEP-0167_jingle_session_init", self._jingle_session_init_trigger ) + host.trigger.add( + "XEP-0167_content_add_elt_built", self._content_add_elt_built_trigger + ) host.trigger.add("XEP-0167_jingle_handler", self._jingle_handler_trigger) def get_handler(self, client): @@ -84,13 +87,41 @@ class XEP_0338: session: dict, local: bool, sdp_lines: List[str], + content_names: List[str] | None = None, ) -> None: """Generate "group" attributes""" + if content_names is not None: + # the session groups describe the full session: only the generated m-lines + # sharing a single transport are bundled + ice_key = "local_ice_data" if local else "peer_ice_data" + for names in self._group_by_transport( + session, content_names, ice_key + ).values(): + if len(names) > 1: + sdp_lines.append(f"a=group:BUNDLE {' '.join(names)}") + return 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)}") + for semantics, names in group_data.items(): + sdp_lines.append(f"a=group:{semantics} {' '.join(names)}") + + def _group_by_transport( + self, session: dict, content_names: List[str], ice_key: str + ) -> dict[tuple[str, str], List[str]]: + """Group content names sharing a single transport + + @param session: session data + @param content_names: names of the contents to group + @param ice_key: transport data key holding the credentials ("local_ice_data" or + "peer_ice_data") + @return: dict mapping (ufrag, pwd) to the names sharing the transport + """ + groups: dict[tuple[str, str], List[str]] = {} + for name in content_names: + ice_data = session["contents"][name]["transport_data"][ice_key] + groups.setdefault((ice_data["ufrag"], ice_data["pwd"]), []).append(name) + return groups def parse_group_element(self, jingle_elt: domish.Element, session: dict) -> None: """Parse the and elements""" @@ -104,18 +135,60 @@ class XEP_0338: 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(): - # always add the element as first child of element to work - # around https://github.com/movim/movim/issues/1233 - group_elt = domish.Element((NS_JINGLE_GROUPING, "group")) - jingle_elt.children.insert(0, group_elt) + def _add_group_elt( + self, + jingle_elt: domish.Element, + semantics: str, + content_names: List[str], + ) -> None: + """Add a element with the given contents - group_elt["semantics"] = semantics - for content_name in content_names: - content_elt = group_elt.addElement((NS_JINGLE_GROUPING, "content")) - content_elt["name"] = content_name + @param jingle_elt: element the group is added to + @param semantics: grouping semantics (e.g. "BUNDLE") + @param content_names: names of the grouped contents + """ + group_elt = domish.Element((NS_JINGLE_GROUPING, "group")) + jingle_elt.addChild(group_elt) + + 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 add_group_element( + self, jingle_elt: domish.Element, session: dict, source: str = "metadata" + ) -> None: + """Build the and elements if possible + + @param jingle_elt: element the groups are added to + @param session: session data + @param source: session key holding the group data ("metadata" for the local + groups, "peer_metadata" to echo the peer's groups) + """ + for semantics, content_names in session.get(source, {}).get("group", {}).items(): + self._add_group_elt(jingle_elt, semantics, content_names) + + def _content_add_elt_built_trigger( + self, + client: SatXMPPEntity, + session: dict, + jingle_elt: domish.Element, + added: List[str], + ) -> None: + """Add the BUNDLE groups of a sent content-add + + The added contents are bundled with the other contents sharing their transport. + @param client: the client entity + @param session: session data + @param jingle_elt: element of the content-add to be sent + @param added: names of the contents added to the session + """ + for names in self._group_by_transport( + session, list(session["contents"]), "local_ice_data" + ).values(): + # the element lists the full bundle, not only the added contents + if len(names) > 1 and any(name in names for name in added): + self._add_group_elt(jingle_elt, "BUNDLE", names) def _jingle_session_init_trigger( self, @@ -149,7 +222,7 @@ class XEP_0338: 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) + self.add_group_element(jingle_elt, session, source="peer_metadata") @implementer(itx_xmpp.IDisco) diff --git a/libervia/backend/plugins/plugin_xep_0343.py b/libervia/backend/plugins/plugin_xep_0343.py index 880d88a4..b21e2038 100644 --- a/libervia/backend/plugins/plugin_xep_0343.py +++ b/libervia/backend/plugins/plugin_xep_0343.py @@ -177,7 +177,7 @@ class XEP_0343(BaseTransportHandler): ) -> None: call_data = content["app_kwargs"]["extra"].pop("call_data", None) if call_data: - metadata = self._rtp.parse_call_data(call_data) + metadata = self._rtp.parse_call_data(call_data, self._j.ROLE_INITIATOR) try: application_data = call_data["application"] except KeyError: diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index 5fe73b7d..188178a1 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -141,6 +141,7 @@ def client_IQ(): @fixture def client(): client = MagicMock() + client.is_component = False client.jid = jid.JID("test_user@test.example/123") client.server_jid = jid.JID("test.example") client.pubsub_service = jid.JID("pubsub.test.example") diff --git a/tests/unit/test_plugin_xep_0166.py b/tests/unit/test_plugin_xep_0166.py new file mode 100644 index 00000000..e78030df --- /dev/null +++ b/tests/unit/test_plugin_xep_0166.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 + +# Libervia: an XMPP client +# Copyright (C) 2009-2026 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 unittest.mock import MagicMock + +from libervia.backend.plugins.plugin_xep_0166 import XEP_0166 +from libervia.backend.tools.common import data_format + + +class TestXEP0166ContentChangeSignal: + def _get_session(self, content_name="video"): + mock_application = MagicMock() + mock_application.handler.get_content_change_payload.return_value = { + "media": "video", + } + return { + "id": "123", + "added_content_names": [content_name], + "contents": { + content_name: { + "senders": "initiator", + "application": mock_application, + } + }, + } + + def test_emit_content_change(self, host, client): + """The signal carries the session data and the application extra data""" + xep_0166 = XEP_0166(host) + mock_signal = MagicMock() + host.bridge.jingle_content_change = mock_signal + session = self._get_session() + + xep_0166._emit_content_change(client, XEP_0166.A_CONTENT_ADD, session) + + args = mock_signal.call_args.args + assert args[1] == client.profile + info = data_format.deserialise(args[0]) + assert info["session_id"] == "123" + assert info["action"] == XEP_0166.A_CONTENT_ADD + assert info["changes"] == [ + {"content": "video", "senders": "initiator", "payload": {"media": "video"}} + ] + session["contents"]["video"][ + "application" + ].handler.get_content_change_payload.assert_called_once_with( + client, XEP_0166.A_CONTENT_ADD, session, "video" + ) + + def test_emit_content_change_multi_content(self, host, client): + """A stanza with several contents emits a single signal with all the changes""" + xep_0166 = XEP_0166(host) + mock_signal = MagicMock() + host.bridge.jingle_content_change = mock_signal + session = self._get_session("video") + session["added_content_names"] = ["video", "audio"] + session["contents"]["audio"] = session["contents"]["video"] + + xep_0166._emit_content_change(client, XEP_0166.A_CONTENT_ADD, session) + + assert mock_signal.call_count == 1 + info = data_format.deserialise(mock_signal.call_args.args[0]) + changes = [change["content"] for change in info["changes"]] + assert changes == ["video", "audio"] + + def test_emit_content_change_unknown_content(self, host, client): + """When the content is unknown, no signal is emitted and no exception raised""" + xep_0166 = XEP_0166(host) + mock_signal = MagicMock() + host.bridge.jingle_content_change = mock_signal + session = { + "id": "123", + "added_content_names": ["video"], + "contents": {}, + } + + xep_0166._emit_content_change(client, XEP_0166.A_CONTENT_ADD, session) + + mock_signal.assert_not_called() diff --git a/tests/unit/test_plugin_xep_0167.py b/tests/unit/test_plugin_xep_0167.py index c2882449..5fdc08a9 100644 --- a/tests/unit/test_plugin_xep_0167.py +++ b/tests/unit/test_plugin_xep_0167.py @@ -22,13 +22,19 @@ from unittest.mock import MagicMock, patch from pytest import fixture from pytest import raises from twisted.words.protocols.jabber import jid +from twisted.words.xish import domish -from libervia.backend.plugins.plugin_xep_0166 import XEP_0166 +from libervia.backend.plugins.plugin_xep_0166 import XEP_0166, NS_JINGLE +from libervia.backend.plugins.plugin_xep_0166.models import ( + ContentChangePayload, + TransportData, +) from libervia.backend.plugins.plugin_xep_0167 import XEP_0167, mapping from libervia.backend.plugins.plugin_xep_0167.constants import ( NS_JINGLE_RTP, NS_JINGLE_RTP_INFO, ) +from libervia.backend.plugins.plugin_xep_0176 import NS_JINGLE_ICE_UDP from libervia.backend.tools import xml_tools from libervia.backend.tools.common import data_format @@ -40,10 +46,6 @@ def no_application_register(monkeypatch): class TestXEP0167Mapping: - @fixture(scope="class", autouse=True) - def set_mapping_host(self, host): - mapping.host = host - def test_senders_to_sdp(self): """Senders are mapped to SDP attribute""" assert mapping.senders_to_sdp("both", {"role": "initiator"}) == "a=sendrecv" @@ -112,6 +114,68 @@ class TestXEP0167Mapping: assert mapping.generate_sdp_from_session(session, True) == expected_sdp + def test_generate_content_add_sdp(self): + """The m-line of a content added by the peer is correctly generated""" + content_data = { + "application_data": { + "media": "video", + "peer_data": { + "payload_types": { + 96: { + "name": "VP8", + "clockrate": 90000, + "parameters": {"max-fs": "12288"}, + } + }, + "rtcp-mux": True, + }, + }, + "transport_data": { + "peer_ice_data": { + "ufrag": "ufrag", + "pwd": "pwd", + "fingerprint": { + "hash": "sha-256", + "setup": "actpass", + "fingerprint": "AB:CD:EF", + }, + "candidates": [ + { + "foundation": "1", + "component_id": 1, + "transport": "UDP", + "priority": 1, + "address": "10.0.0.2", + "port": 54321, + "type": "host", + } + ], + } + }, + "senders": "initiator", + } + session = {"role": "responder"} + + expected_sdp = ( + "m=video 9 UDP/TLS/RTP/SAVPF 96\r\n" + "a=mid:video\r\n" + "c=IN IP4 0.0.0.0\r\n" + "a=sendonly\r\n" + "a=rtpmap:96 VP8/90000\r\n" + "a=fmtp:96 max-fs=12288\r\n" + "a=rtcp-mux\r\n" + "a=fingerprint:sha-256 AB:CD:EF\r\n" + "a=setup:actpass\r\n" + "a=ice-ufrag:ufrag\r\n" + "a=ice-pwd:pwd\r\n" + "a=candidate:1 1 UDP 1 10.0.0.2 54321 typ host\r\n" + ) + + assert ( + mapping.generate_content_add_sdp(content_data, session, "video") + == expected_sdp + ) + def test_parse_sdp(self): """SDP is correctly parsed to session data""" sdp = ( @@ -170,6 +234,68 @@ class TestXEP0167Mapping: assert mapping.parse_sdp(sdp, "responder") == expected_session + def test_parse_sdp_session_level_ice(self): + """Session-level transport attributes apply to all media + + unless overridden at the media level + media + """ + sdp = ( + "v=0\r\no=- 0 0 IN IP4 0.0.0.0\r\nt=0 0\r\n" + "a=fingerprint:sha-256 00:11:22\r\n" + "a=ice-pwd:spwd\r\n" + "m=audio 1 UDP/TLS/RTP/SAVPF 96\r\n" + "a=fingerprint:sha-256 33:44:55\r\n" + "a=ice-ufrag:sufrag\r\n" + "m=video 2 UDP/TLS/RTP/SAVPF 97\r\n" + "a=ice-ufrag:sufrag\r\n" + ) + parsed = mapping.parse_sdp(sdp, "initiator") + # audio: the media-level fingerprint wins + assert parsed["audio"]["transport_data"]["fingerprint"] == { + "hash": "sha-256", + "fingerprint": "33:44:55", + } + # video: no media-level fingerprint, the session one applies (the audio + # override must not have polluted it) + assert parsed["video"]["transport_data"]["fingerprint"] == { + "hash": "sha-256", + "fingerprint": "00:11:22", + } + # the session-level ice-pwd applies to both media + assert parsed["audio"]["transport_data"]["pwd"] == "spwd" + assert parsed["video"]["transport_data"]["pwd"] == "spwd" + + def test_parse_mline(self): + """Test parse_mline on a standalone m-line block""" + mline = ( + "m=audio 49170 UDP/TLS/RTP/SAVPF 0 96\r\n" + "c=IN IP4 0.0.0.0\r\n" + "a=rtcp-mux\r\n" + "a=sendonly\r\n" + "a=rtpmap:96 opus/48000/2\r\n" + "a=ice-ufrag:abc123\r\n" + "a=ice-pwd:def456\r\n" + "a=fingerprint:sha-256 00:11:22\r\n" + ) + media, media_data = mapping.parse_mline(mline, "responder") + assert media == "audio" + assert media_data["application_data"]["senders"] == "responder" + assert media_data["application_data"]["payload_types"][96]["name"] == "opus" + assert media_data["transport_data"]["port"] == 49170 + assert media_data["transport_data"]["ufrag"] == "abc123" + assert media_data["transport_data"]["pwd"] == "def456" + assert media_data["transport_data"]["fingerprint"]["fingerprint"] == "00:11:22" + + # session attributes (as forwarded by parse_sdp) are applied to the media: + # with no media-level direction, the session one is used (a media-level + # a=sendonly would override it, remapped with the role) + mline_no_dir = mline.replace("a=sendonly\r\n", "") + media, media_data = mapping.parse_mline( + mline_no_dir, "responder", {"sendonly": "initiator"} + ) + assert media_data["application_data"]["senders"] == "initiator" + def test_build_description(self): """ element is generated from media data""" session = {"metadata": {}} @@ -374,3 +500,265 @@ class TestXEP0167: with raises(ValueError, match="Unkown info type 'invalid_action'"): xep_0167.send_info(client, session_id, "invalid_action", extra) + + def test_content_change_payload_content_add(self, host, client): + """For a content-add, the offered m-line SDP is returned in the payload""" + xep_0167 = XEP_0167(host) + session = { + "id": "123", + "role": "responder", + "contents": { + "video": { + "application_data": { + "media": "video", + "peer_data": { + "payload_types": {96: {"name": "VP8", "clockrate": 90000}}, + }, + }, + "transport_data": { + "peer_ice_data": { + "ufrag": "uf", + "pwd": "pw", + "candidates": [], + } + }, + "senders": "initiator", + } + }, + } + + payload = xep_0167.get_content_change_payload( + client, XEP_0166.A_CONTENT_ADD, session, "video" + ) + + assert payload.media == "video" + assert payload.sdp + assert "m=video 9 UDP/TLS/RTP/SAVPF 96" in payload.sdp + assert "a=rtpmap:96 VP8/90000" in payload.sdp + assert "a=ice-ufrag:uf" in payload.sdp + + def test_content_change_payload_content_modify(self, host, client): + """For a content-modify, the media and new direction are returned (no SDP)""" + xep_0167 = XEP_0167(host) + session = { + "id": "123", + "role": "responder", + "contents": { + "video": { + "application_data": {"media": "video"}, + "senders": "both", + } + }, + } + + payload = xep_0167.get_content_change_payload( + client, XEP_0166.A_CONTENT_MODIFY, session, "video" + ) + + assert payload.model_dump(exclude_none=True) == { + "media": "video", + "direction": "sendrecv", + } + + async def test_content_modify(self, host, client, client_IQ): + """The modified senders attribute is updated and acknowledged""" + xep_0166 = XEP_0166(host) + client.IQ = client_IQ() + mock_application = MagicMock() + mock_application.handler.get_content_change_payload.return_value = ( + ContentChangePayload() + ) + session = { + "id": "123", + "local_jid": jid.JID("client1@example.org/1"), + "peer_jid": jid.JID("client2@example.net/2"), + "role": "responder", + "contents": { + "video": { + "senders": "initiator", + "creator": XEP_0166.ROLE_INITIATOR, + "application": mock_application, + "transport": MagicMock(), + } + }, + } + mock_send = MagicMock() + client.send = mock_send + + request = domish.Element((None, "iq")) + request["type"] = "set" + jingle_elt = request.addElement("jingle", NS_JINGLE) + jingle_elt["sid"] = "123" + jingle_elt["action"] = XEP_0166.A_CONTENT_MODIFY + content_elt = jingle_elt.addElement("content", NS_JINGLE) + content_elt["name"] = "video" + content_elt["senders"] = "both" + + await xep_0166.on_content_modify(client, request, jingle_elt, session) + + # the direction is updated + assert session["contents"]["video"]["senders"] == "both" + assert session["modified_content_names"] == ["video"] + # no contrary content-modify is sent, only the IQ result + stanzas = [call.args[0] for call in mock_send.call_args_list] + assert stanzas[0]["type"] == "result" + + async def test_call_update(self, host, client, client_IQ): + """New m-lines of the offered SDP are added with a content-add IQ""" + xep_0167 = XEP_0167(host) + client.IQ = client_IQ() + session_id = "123" + session = { + "id": session_id, + "local_jid": client.jid, + "peer_jid": jid.JID("room@conference.test.example"), + "role": "initiator", + "contents": {}, + } + client.jingle_sessions = {session_id: session} + lines = [ + "v=0", + "o=- 0 0 IN IP4 127.0.0.1", + "s=-", + "t=0 0", + "m=audio 9 UDP/TLS/RTP/SAVPF 111", + "c=IN IP4 0.0.0.0", + "a=rtcp-mux", + "a=mid:1-0", + "a=sendonly", + "a=rtpmap:111 opus/48000/2", + "a=ice-ufrag:uf1", + "a=ice-pwd:pw1", + ] + added = await xep_0167.call_update( + client, session_id, {"sdp": "\r\n".join(lines)} + ) + assert added == ["1-0"] + # the new content is sent with a content-add IQ + iq_elt = client.IQ.sent_iqs[0] + jingle_elt = list(iq_elt.elements(NS_JINGLE, "jingle"))[0] + assert jingle_elt["action"] == XEP_0166.A_CONTENT_ADD + content_elts = list(jingle_elt.elements(NS_JINGLE, "content")) + assert [c["name"] for c in content_elts] == ["1-0"] + + def test_content_change_payload_content_accept(self, host, client): + """For a content-accept, the answer m-line is generated""" + xep_0167 = XEP_0167(host) + # the real ICE-UDP transport handler, so the whole parsing chain is tested + # (XEP-0176 parsing + XEP-0320 fingerprint) + transport = TransportData(NS_JINGLE_ICE_UDP, host.plugins["XEP-0176"], 100) + desc_elt = domish.Element( + (NS_JINGLE_RTP, "description"), attribs={"media": "video"} + ) + pt_elt = desc_elt.addElement("payload-type") + pt_elt["id"] = "96" + pt_elt["name"] = "VP8" + pt_elt["clockrate"] = "90000" + # the peer transport element, parsed by the XEP-0176 handler (fingerprint + # included by the XEP-0320 trigger) + peer_transport_elt = domish.Element( + (NS_JINGLE_ICE_UDP, "transport"), attribs={"ufrag": "uf", "pwd": "pw"} + ) + fingerprint_elt = peer_transport_elt.addElement( + ("urn:xmpp:jingle:apps:dtls:0", "fingerprint"), content="AB:CD" + ) + fingerprint_elt["hash"] = "sha-256" + fingerprint_elt["setup"] = "actpass" + session = { + "id": "123", + "role": "responder", + "contents": { + "video": { + "application_data": {"media": "video"}, + "transport": transport, + "transport_data": {}, + "desc_elt": desc_elt, + "transport_elt": peer_transport_elt, + "senders": "responder", + } + }, + } + + payload = xep_0167.get_content_change_payload( + client, XEP_0166.A_CONTENT_ACCEPT, session, "video" + ) + + assert payload.media == "video" + # the peer data are parsed and stored in the session + assert session["contents"]["video"]["application_data"]["peer_data"] + assert session["contents"]["video"]["transport_data"]["peer_ice_data"] + assert payload.sdp + assert "m=video 9 UDP/TLS/RTP/SAVPF 96" in payload.sdp + # the direction is strict (senders=responder, we are the responder: the peer receives) + assert "a=recvonly" in payload.sdp + assert "a=fingerprint:sha-256 AB:CD" in payload.sdp + assert "a=ice-ufrag:uf" in payload.sdp + + def test_call_content_accept(self, host, client, client_IQ): + """A content-accept IQ is sent, built from the frontend answer m-line""" + xep_0167 = XEP_0167(host) + transport_elt = domish.Element( + (NS_JINGLE_ICE_UDP, "transport"), attribs={"ufrag": "uf", "pwd": "pw"} + ) + mock_transport = MagicMock() + mock_transport.handler.build_transport = MagicMock(return_value=transport_elt) + client.IQ = client_IQ() + session = { + "id": "123", + "added_content_names": ["video"], + "local_jid": jid.JID("client1@example.org/1"), + "peer_jid": jid.JID("client2@example.net/2"), + "role": "responder", + "contents": { + "video": { + "application_data": {"media": "video"}, + "transport": mock_transport, + "transport_data": {}, + "senders": "initiator", + } + }, + } + client.jingle_sessions = {"123": session} + host.get_client = MagicMock(return_value=client) + mock_send = MagicMock() + client.send = mock_send + mline = ( + "m=video 9 UDP/TLS/RTP/SAVPF 96\r\n" + "c=IN IP4 0.0.0.0\r\n" + "a=recvonly\r\n" + "a=rtpmap:96 VP8/90000\r\n" + "a=rtcp-mux\r\n" + "a=fingerprint:sha-256 AA:BB\r\n" + "a=setup:active\r\n" + "a=ice-ufrag:uf\r\n" + "a=ice-pwd:pw\r\n" + "a=candidate:1 1 UDP 1 10.0.0.1 12345 typ host\r\n" + ) + + xep_0167._call_content_accept("123", "video", mline, client.profile) + + assert "added_content_names" not in session + # local data are stored in the session + assert session["contents"]["video"]["application_data"]["local_data"] + assert ( + session["contents"]["video"]["transport_data"]["local_ice_data"]["ufrag"] + == "uf" + ) + sent_iq = mock_send.call_args.args[0] + jingle_elt = sent_iq.firstChildElement() + assert jingle_elt["action"] == XEP_0166.A_CONTENT_ACCEPT + content_elt = jingle_elt.firstChildElement() + assert content_elt["name"] == "video" + assert content_elt["creator"] == XEP_0166.ROLE_INITIATOR + assert content_elt["senders"] == "initiator" + desc_elts = list(content_elt.elements(NS_JINGLE_RTP, "description")) + assert len(desc_elts) == 1 + desc_elt = desc_elts[0] + assert desc_elt["media"] == "video" + payload_elts = list(desc_elt.elements(NS_JINGLE_RTP, "payload-type")) + assert len(payload_elts) == 1 + assert payload_elts[0]["name"] == "VP8" + transport_elts = content_elt.elements(NS_JINGLE_ICE_UDP, "transport") + transport_elt = next(transport_elts) + assert transport_elt["ufrag"] == "uf" + assert transport_elt["pwd"] == "pw" diff --git a/tests/unit/test_plugin_xep_0176.py b/tests/unit/test_plugin_xep_0176.py index 8e7352a7..99629ad5 100644 --- a/tests/unit/test_plugin_xep_0176.py +++ b/tests/unit/test_plugin_xep_0176.py @@ -251,7 +251,7 @@ class TestXEP0176: assert returned_transport_elt.toXml() == expected_transport_elt.toXml() @ed - async def test_ice_candidates_add(self, host, client): + async def test_ice_candidates_add(self, host, client, monkeypatch): """local new ICE candidates are added, IQ is sent, bridge signal emitted""" xep_0176 = XEP_0176(host) @@ -278,8 +278,9 @@ class TestXEP0176: } mock_session = self.create_mock_session(content_name) - xep_0176._j.get_session = MagicMock(return_value=mock_session) - xep_0176._j.build_action = MagicMock() + get_session = MagicMock(return_value=mock_session) + monkeypatch.setattr(xep_0176._j, "get_session", get_session) + monkeypatch.setattr(xep_0176._j, "build_action", MagicMock()) iq_elt = AsyncMock() xep_0176._j.build_action.return_value = (iq_elt, None) xep_0176.host.bridge.ice_restart = MagicMock() diff --git a/tests/unit/test_plugin_xep_0272.py b/tests/unit/test_plugin_xep_0272.py index 8a42a63b..825e47c0 100644 --- a/tests/unit/test_plugin_xep_0272.py +++ b/tests/unit/test_plugin_xep_0272.py @@ -47,7 +47,8 @@ def xep_0272(host) -> XEP_0272: "XEP-0045": MagicMock(), "XEP-0166": MagicMock(), "XEP-0167": MagicMock(), - "XEP-0249": MagicMock(), + "XEP-0402": MagicMock(), + "XEP-0482": MagicMock(), } return XEP_0272(host) @@ -236,6 +237,30 @@ class TestXEP0272: xep_0272.try_to_finish_preparation(client, room, muji_data) mock_call_group_setup.assert_not_called() + @ed + async def test_call_group_end(self, host, client): + """call_group_end removes the MUJI presence data and the room bookmark.""" + xep_0272 = XEP_0272(host) + room_jid = jid.JID("room@example.com") + room = MagicMock() + room.roomJID = room_jid + muji_data = {"to_call": {jid.JID("peer@example.com")}, "done_preparing": True} + + with ( + patch.object(xep_0272._muc, "get_room", return_value=room), + patch.object(xep_0272, "get_muji_data", return_value=muji_data), + patch.object( + client.presence, "send", return_value=defer.succeed(None) + ) as mock_presence_send, + patch.object( + xep_0272._bookmarks, "remove", return_value=defer.succeed(None) + ) as mock_remove, + ): + await xep_0272.call_group_end(client, room_jid) + mock_presence_send.assert_called_once() + assert muji_data == {"to_call": set()} + mock_remove.assert_called_once_with(client, room_jid) + @ed async def test_call_group_data_set(self, host, client): """call_group_data_set sends correct presence and muji data.""" diff --git a/tests/unit/test_plugin_xep_0338.py b/tests/unit/test_plugin_xep_0338.py index 4220a26c..9ece6869 100644 --- a/tests/unit/test_plugin_xep_0338.py +++ b/tests/unit/test_plugin_xep_0338.py @@ -17,16 +17,25 @@ # along with this program. If not, see . +from pytest import fixture +from twisted.words.protocols.jabber import jid from twisted.words.xish import domish -from libervia.backend.plugins.plugin_xep_0338 import NS_JINGLE_GROUPING, XEP_0338 +from libervia.backend.plugins.plugin_xep_0166 import XEP_0166, NS_JINGLE +from libervia.backend.plugins.plugin_xep_0338 import NS_JINGLE_GROUPING from libervia.backend.tools.xml_tools import parse +@fixture(autouse=True) +def no_application_register(monkeypatch): + """Do not register the application in XEP-0166""" + monkeypatch.setattr(XEP_0166, "register_application", lambda *a, **kw: None) + + class TestXEP0338: def test_parse_sdp(self, host): """'group' attribute in SDP is correctly parsed""" - xep_0338 = XEP_0338(host) + xep_0338 = host.plugins["XEP-0338"] call_data = {} metadata = {} @@ -52,7 +61,7 @@ class TestXEP0338: def test_generate_sdp(self, host): """'group' attribute in SDP is correctly generated""" - xep_0338 = XEP_0338(host) + xep_0338 = host.plugins["XEP-0338"] session = {"metadata": {"group": {"BUNDLE": ["audio", "video"]}}} sdp_lines = [] @@ -64,7 +73,7 @@ class TestXEP0338: def test_group_building(self, host, client): """ element are built from session in session init trigger""" - xep_0338 = XEP_0338(host) + xep_0338 = host.plugins["XEP-0338"] session = { "contents": {"audio": {}}, @@ -92,7 +101,7 @@ class TestXEP0338: async def test_group_parsing(self, host, client): """ elements are correctly parsed in jingle_handler trigger""" - xep_0338 = XEP_0338(host) + xep_0338 = host.plugins["XEP-0338"] action = xep_0338._j.A_SESSION_INITIATE session = { @@ -138,3 +147,159 @@ class TestXEP0338: for content_elt in group_elt.elements(NS_JINGLE_GROUPING, "content") ] assert content_names == ["audio", "video"] + + async def test_content_add_bundle_group(self, host, client, client_IQ): + """A BUNDLE group is declared with all the contents sharing the transport""" + xep_0167 = host.plugins["XEP-0167"] + client.IQ = client_IQ() + session_id = "123" + session = { + "id": session_id, + "local_jid": client.jid, + "peer_jid": jid.JID("room@conference.test.example"), + "role": "initiator", + # an existing content already using the transport (e.g. a single transport + # shared by all the streams of the session) + "contents": { + "0-0": { + "application_data": {"media": "audio"}, + "transport_data": { + "local_ice_data": {"ufrag": "uf1", "pwd": "pw1", "candidates": []} + }, + } + }, + } + client.jingle_sessions = {session_id: session} + lines = [ + "v=0", + "o=- 0 0 IN IP4 127.0.0.1", + "s=-", + "t=0 0", + "m=audio 9 UDP/TLS/RTP/SAVPF 111", + "c=IN IP4 0.0.0.0", + "a=rtcp-mux", + "a=mid:1-0", + "a=sendonly", + "a=rtpmap:111 opus/48000/2", + "a=ice-ufrag:uf1", + "a=ice-pwd:pw1", + "m=video 9 UDP/TLS/RTP/SAVPF 96", + "c=IN IP4 0.0.0.0", + "a=rtcp-mux", + "a=mid:1-1", + "a=sendonly", + "a=rtpmap:96 VP8/90000", + "a=ice-ufrag:uf1", + "a=ice-pwd:pw1", + ] + added = await xep_0167.call_update( + client, session_id, {"sdp": "\r\n".join(lines)} + ) + assert added == ["1-0", "1-1"] + # the BUNDLE group is declared for the added contents + iq_elt = client.IQ.sent_iqs[0] + jingle_elt = list(iq_elt.elements(NS_JINGLE, "jingle"))[0] + group_elts = list(jingle_elt.elements(NS_JINGLE_GROUPING, "group")) + assert len(group_elts) == 1 + group_elt = group_elts[0] + assert group_elt["semantics"] == "BUNDLE" + content_names = [ + c["name"] for c in group_elt.elements(NS_JINGLE_GROUPING, "content") + ] + # the existing content sharing the transport is included in the group + assert content_names == ["0-0", "1-0", "1-1"] + + def test_sdp_bundle_group(self, host, client): + """The stream answer SDP uses the stream's own BUNDLE group, not the uplink's""" + xep_0167 = host.plugins["XEP-0167"] + session = { + "id": "123", + "local_jid": client.jid, + "role": "responder", + "peer_metadata": {"group": {"BUNDLE": ["video0", "audio1"]}}, + "contents": { + "1-0": { + "application_data": { + "media": "audio", + "peer_data": { + "payload_types": {111: {"name": "opus", "clockrate": 48000}} + }, + }, + "transport_data": { + "peer_ice_data": {"ufrag": "uf", "pwd": "pw", "candidates": []} + }, + "senders": "responder", + }, + "1-1": { + "application_data": { + "media": "video", + "peer_data": { + "payload_types": {96: {"name": "VP8", "clockrate": 90000}} + }, + }, + "transport_data": { + "peer_ice_data": {"ufrag": "uf", "pwd": "pw", "candidates": []} + }, + "senders": "responder", + }, + }, + } + sdps = [] + session["content_accept_cb"] = lambda *a: sdps.append(a[2]) + xep_0167.on_content_change_trigger( + client, XEP_0166.A_CONTENT_ACCEPT, session, ["1-0", "1-1"] + ) + assert len(sdps) == 1 + assert "a=group:BUNDLE 1-0 1-1" in sdps[0] + assert "video0" not in sdps[0] + + def test_content_change_sdp_no_bundle_group(self, host, client): + """Accepted contents with different transports are not bundled""" + xep_0167 = host.plugins["XEP-0167"] + session = { + "id": "123", + "local_jid": client.jid, + "role": "responder", + "peer_metadata": {"group": {"BUNDLE": ["video0", "audio1"]}}, + "contents": { + "1-0": { + "application_data": { + "media": "audio", + "peer_data": { + "payload_types": {111: {"name": "opus", "clockrate": 48000}} + }, + }, + "transport_data": { + "peer_ice_data": { + "ufrag": "uf1", + "pwd": "pw1", + "candidates": [], + } + }, + "senders": "responder", + }, + "1-1": { + "application_data": { + "media": "video", + "peer_data": { + "payload_types": {96: {"name": "VP8", "clockrate": 90000}} + }, + }, + "transport_data": { + "peer_ice_data": { + "ufrag": "uf2", + "pwd": "pw2", + "candidates": [], + } + }, + "senders": "responder", + }, + }, + } + sdps = [] + session["content_accept_cb"] = lambda *a: sdps.append(a[2]) + xep_0167.on_content_change_trigger( + client, XEP_0166.A_CONTENT_ACCEPT, session, ["1-0", "1-1"] + ) + assert len(sdps) == 1 + assert "a=group:" not in sdps[0]