component conferences: update to manage the new single Jingle session and the ICE restart:

The Jingle Audio/Video Conferences spec has been updated to work with a single Jingle
session. This patch updates the component to handle it, and implements ICE restart
management.

Closing is better handled, and session are replaced in case of duplication (happens
notably in web frontend on tab reload).

libervia/backend/plugins/plugin_comp_conferences/__init__.py

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

tests/unit/test_plugin_comp_conferences.py

@@ -0,0 +1,318 @@
+"""Unit tests for the conferences component (ICE restart paths)."""
+
+from pathlib import Path
+from types import SimpleNamespace
+
+from pytest import fixture
+from twisted.internet import defer
+from twisted.words.protocols.jabber import jid
+
+from libervia.backend.plugins.plugin_comp_conferences import Conference
+
+# the SDP line endings are CRLF, as in real offers (RFC 4566)
+UPLINK_OFFER_SDP = ("""v=0
+o=- 0 0 IN IP4 127.0.0.1
+s=-
+t=0 0
+a=group:BUNDLE 0 1
+m=audio 9 UDP/TLS/RTP/SAVPF 111
+c=IN IP4 0.0.0.0
+a=mid:0
+a=sendonly
+a=ice-ufrag:uf1
+a=ice-pwd:pw1
+m=video 9 UDP/TLS/RTP/SAVPF 96
+c=IN IP4 0.0.0.0
+a=mid:1
+a=sendonly
+a=ice-ufrag:uf1
+a=ice-pwd:pw1
+""").replace("\n", "\r\n")
+
+# answer of the renegotiated uplink offer (an ICE restart by the user): new ICE
+# credentials, with inline candidates
+UPLINK_ANSWER_SDP = ("""v=0
+o=- 0 0 IN IP4 127.0.0.1
+s=-
+t=0 0
+a=group:BUNDLE 0 1
+m=audio 9 UDP/TLS/RTP/SAVPF 111
+c=IN IP4 0.0.0.0
+a=mid:0
+a=recvonly
+a=ice-ufrag:uf3
+a=ice-pwd:pw3
+a=candidate:1 1 udp 2122260223 192.168.1.1 30000 typ host
+m=video 9 UDP/TLS/RTP/SAVPF 96
+c=IN IP4 0.0.0.0
+a=mid:1
+a=recvonly
+a=ice-ufrag:uf3
+a=ice-pwd:pw3
+a=candidate:2 1 udp 2122260223 192.168.1.1 30001 typ host
+""").replace("\n", "\r\n")
+
+# new offer for an existing stream (an ICE restart by the SFU): new ICE credentials, with
+# inline candidates
+DOWNLINK_REOFFER_SDP = ("""v=0
+o=- 0 0 IN IP4 127.0.0.1
+s=-
+t=0 0
+a=group:BUNDLE 0 1
+m=audio 9 UDP/TLS/RTP/SAVPF 111
+c=IN IP4 0.0.0.0
+a=mid:0
+a=sendonly
+a=ice-ufrag:uf9
+a=ice-pwd:pw9
+a=candidate:1 1 udp 2122260223 192.168.1.1 30000 typ host
+m=video 9 UDP/TLS/RTP/SAVPF 96
+c=IN IP4 0.0.0.0
+a=mid:1
+a=sendonly
+a=ice-ufrag:uf9
+a=ice-pwd:pw9
+a=candidate:2 1 udp 2122260223 192.168.1.1 30001 typ host
+""").replace("\n", "\r\n")
+
+
+class MockProtocol:
+    """Records the data sent to the SFU."""
+
+    client_id = "c0ffee"
+
+    def __init__(self):
+        self.sent = []
+
+    def send_data(self, data):
+        self.sent.append(data)
+
+    def sendClose(self):
+        pass
+
+
+class MockIceUdp:
+    """Records the calls to ice_candidates_add (the transport-info send path)."""
+
+    def __init__(self, session: dict):
+        self.session = session
+        self.calls = []
+
+    def ice_candidates_add(self, client, session_id, content_ice_data):
+        # like XEP-0176: the session's local ICE data is updated with the new credentials
+        for name, ice_data in content_ice_data.items():
+            local_ice_data = (
+                self.session["contents"]
+                .get(name, {})
+                .get("transport_data", {})
+                .get("local_ice_data")
+            )
+            if local_ice_data is not None:
+                local_ice_data.update(ufrag=ice_data["ufrag"], pwd=ice_data["pwd"])
+        self.calls.append((client, session_id, content_ice_data))
+        return defer.succeed(None)
+
+
+@fixture
+def conference():
+    """A Conference with mocked parent (SFU protocol and XEP-0176)."""
+    session = {
+        "id": "sess123",
+        "local_jid": jid.JID("conferences.test.example"),
+        "peer_jid": jid.JID("user@test.example"),
+        "role": "responder",
+        "contents": {},
+        "streams": {},
+        "uplink_offer_sdp": UPLINK_OFFER_SDP,
+    }
+    parent = SimpleNamespace(
+        client=SimpleNamespace(get_virtual_client=lambda jid_: object()),
+        _ice_udp=MockIceUdp(session),
+    )
+    conf = Conference(
+        parent,
+        "testroom",
+        Path("testroom.json"),
+        "ws://localhost",
+        "http://localhost",
+        session,
+    )
+    conf._protocol = MockProtocol()
+    return conf
+
+
+def test_restart_uplink_same_credentials(conference):
+    """No renegotiation if the uplink credentials are unchanged."""
+    conference._restart_uplink(
+        conference.session, {"ufrag": "uf1", "pwd": "pw1", "candidates": []}
+    )
+    assert conference._protocol.sent == []
+
+
+def test_restart_uplink_new_credentials(conference):
+    """New uplink credentials trigger a renegotiation offer to the SFU."""
+    conference._restart_uplink(
+        conference.session, {"ufrag": "uf2", "pwd": "pw2", "candidates": []}
+    )
+    (offer,) = conference._protocol.sent
+    assert offer["type"] == "offer"
+    assert offer["id"] == conference.galene_id
+    assert offer["sdp"].count("a=ice-ufrag:uf2") == 2
+    assert offer["sdp"].count("a=ice-pwd:pw2") == 2
+    # the CRLF line endings are preserved (a mangled SDP would not be parsable)
+    assert offer["sdp"].count("\r\n") == UPLINK_OFFER_SDP.count("\r\n")
+    # the offer is stored for the next restart
+    assert conference.session["uplink_offer_sdp"] == offer["sdp"]
+    assert isinstance(conference.session["answer_sdp_d"], defer.Deferred)
+    # the renegotiation is pending: the answer will carry new credentials
+    assert conference.session["uplink_renegotiating"] is True
+
+
+def test_restart_uplink_no_stored_offer(conference):
+    """Without the stored uplink offer, nothing is re-offered."""
+    del conference.session["uplink_offer_sdp"]
+    conference._restart_uplink(
+        conference.session, {"ufrag": "uf2", "pwd": "pw2", "candidates": []}
+    )
+    assert conference._protocol.sent == []
+
+
+def test_restart_uplink_answer_consumed(conference):
+    """The answer of the renegotiation offer is consumed by on_answer()."""
+    conference._restart_uplink(
+        conference.session, {"ufrag": "uf2", "pwd": "pw2", "candidates": []}
+    )
+    conference.on_answer({"sdp": "v=0\n"})
+    assert conference.session.pop("answer_sdp_d", None) is None
+
+
+def test_restart_uplink_answer_credentials_forwarded(conference):
+    """The answer's ICE credentials and inline candidates are forwarded to the user."""
+    session = conference.session
+    for name in ("0", "1"):
+        session["contents"][name] = {
+            "transport_data": {
+                "local_ice_data": {
+                    "ufrag": "uf_old",
+                    "pwd": "pw_old",
+                    "candidates": [],
+                }
+            }
+        }
+    conference._restart_uplink(session, {"ufrag": "uf2", "pwd": "pw2", "candidates": []})
+    conference.on_answer({"sdp": UPLINK_ANSWER_SDP})
+    assert "uplink_renegotiating" not in session
+    calls = conference.parent._ice_udp.calls
+    # two transport-info with the new credentials (one per content), then the two inline
+    # candidates
+    assert [next(iter(data)) for data in (c[2] for c in calls)] == [
+        "0",
+        "1",
+        "0",
+        "1",
+    ]
+    for _, session_id, data in calls:
+        assert session_id == "sess123"
+        ((name, ice_data),) = data.items()
+        assert (ice_data["ufrag"], ice_data["pwd"]) == ("uf3", "pw3")
+    assert calls[2][2]["0"]["candidates"]
+    assert calls[3][2]["1"]["candidates"]
+
+
+def test_on_answer_initial_no_credentials_forwarded(conference):
+    """Without an ICE restart, the answer is not forwarded to the user."""
+    session = conference.session
+    for name in ("0", "1"):
+        session["contents"][name] = {
+            "transport_data": {
+                "local_ice_data": {
+                    "ufrag": "uf3",
+                    "pwd": "pw3",
+                    "candidates": [],
+                }
+            }
+        }
+    session["answer_sdp_d"] = defer.Deferred()
+    conference.on_answer({"sdp": UPLINK_ANSWER_SDP})
+    assert conference.parent._ice_udp.calls == []
+
+
+async def test_renegotiation_downlink(conference):
+    """A re-offered stream: new credentials to the user, stored answer to the SFU."""
+    session = conference.session
+    session["streams"] = {"down1": ["2-0", "2-1"]}
+    session["stream_answer_sdp"] = {"down1": "ANSWER_SDP"}
+    # the contents of the stream, with the new local ICE data (set in production by the
+    # ice_candidates_add calls of _handle_renegotiation)
+    for name, media in (("2-0", "audio"), ("2-1", "video")):
+        session["contents"][name] = {
+            "application_data": {"media": media},
+            "transport_data": {
+                "local_ice_data": {"ufrag": "uf9", "pwd": "pw9", "candidates": []}
+            },
+        }
+    await conference._handle_renegotiation({"id": "down1", "sdp": DOWNLINK_REOFFER_SDP})
+    # two transport-info with the new credentials (one per content), then the two inline
+    # candidates
+    calls = conference.parent._ice_udp.calls
+    assert len(calls) == 4
+    for _, session_id, data in calls:
+        assert session_id == "sess123"
+        ((name, ice_data),) = data.items()
+        assert (ice_data["ufrag"], ice_data["pwd"]) == ("uf9", "pw9")
+    assert [next(iter(data)) for data in (c[2] for c in calls)] == [
+        "2-0",
+        "2-1",
+        "2-0",
+        "2-1",
+    ]
+    assert calls[2][2]["2-0"]["candidates"]
+    assert calls[3][2]["2-1"]["candidates"]
+    answer = [m for m in conference._protocol.sent if m["type"] == "answer"]
+    assert len(answer) == 1
+    assert answer[0]["id"] == "down1"
+    assert answer[0]["sdp"] == "ANSWER_SDP"
+
+
+async def test_renegotiation_downlink_no_stored_answer(conference):
+    """Without the stored answer, no answer is re-sent to the SFU."""
+    session = conference.session
+    session["streams"] = {"down1": ["2-0", "2-1"]}
+    for name, media in (("2-0", "audio"), ("2-1", "video")):
+        session["contents"][name] = {
+            "application_data": {"media": media},
+            "transport_data": {
+                "local_ice_data": {"ufrag": "uf9", "pwd": "pw9", "candidates": []}
+            },
+        }
+    await conference._handle_renegotiation({"id": "down1", "sdp": DOWNLINK_REOFFER_SDP})
+    assert all(m["type"] != "answer" for m in conference._protocol.sent)
+
+
+async def test_offer_mids_renamed(conference):
+    """The mids of a new downlink stream are prefixed with the stream counter."""
+    session = conference.session
+    added = []
+
+    class MockRtp:
+        async def call_update(
+            self, client, session_id, call_data, jingle_child_elts=None
+        ):
+            added.append(call_data)
+            return ["1-0", "1-1"]
+
+    class MockAvConf:
+        def build_origin_elt(self, origin):
+            return None
+
+    conference.parent._rtp = MockRtp()
+    conference.parent._av_conf = MockAvConf()
+    data = {"id": "down1", "sdp": DOWNLINK_REOFFER_SDP, "username": "user@test.example"}
+    await conference.on_offer(data)
+    (call_data,) = added
+    sdp = call_data["sdp"]
+    assert "a=mid:1-0" in sdp
+    assert "a=mid:1-1" in sdp
+    # the CRLF line endings are preserved (a mangled SDP would not be parsable)
+    assert sdp.count("\r\n") == DOWNLINK_REOFFER_SDP.count("\r\n")
+    assert session["streams"] == {"down1": ["1-0", "1-1"]}