erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

plugin XEP-0272: update to v0.2.0 + fixes:

Browse source 
- use the proper `urn:xmpp:jingle:muji:0` namespace
- replace XEP-0249 invitations with XEP-0482
- the dedicated `META_TYPE_GROUP_CALL` subtype is now used for the confirm action (instead
  of `META_TYPE_MUC_INVITATION`)
- handle MUJI invitation: on accept, join the room, start preparing and answer with the
  connection method element
- add the `<muji>` element to outgoing Jingle IQs
- propagate the call type (audio/video) in the MUJI invites (the `type` attribute), so the
  invited clients set up the proper local stream
- the joining participant initiates the sessions with the other participants, using their
  real JIDs, extracted from the presence (the room is set non-anonymous for that)
- only trust the `<muji>` element of a `session-initiate` from a member of the room
- add a `call_group_end` bridge method to clean the MUJI session
- typo fix in `META_TYPE_MUC_INVIRATION` (`R` instead of `T`)
This commit is contained in:
Goffi 2026-09-24 12:05:33 +02:00
commit 0c6f852ce9
4 changed files with 358 additions and 94 deletions
Download patch file Download diff file Expand all files Collapse all files

171
tests/unit/test_plugin_xep_0272.py
View file

@ -16,14 +16,19 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import pytest
from unittest.mock import MagicMock, patch
from pytest_twisted import ensureDeferred as ed
from pytest import fixture
from libervia.backend.core import exceptions
from twisted.internet import defer
from twisted.words.protocols.jabber import jid
from twisted.words.xish import domish
from tx_xmpp import muc
from libervia.backend.plugins.plugin_xep_0272 import (
NS_MUJI,
PRESENCE_MUJI,
@ -37,7 +42,7 @@ TEST_ROOM_JID = jid.JID("room@example.com/user")
@fixture
def xep_0272(host) -> XEP_0272:
"""Fixture for initializing XEP-0272 plugin."""
"""XEP-0272 plugin with mocked dependencies."""
host.plugins = {
"XEP-0045": MagicMock(),
"XEP-0166": MagicMock(),
@ -48,6 +53,20 @@ def xep_0272(host) -> XEP_0272:
class TestXEP0272:
def test_get_presence_user_jid(self, host, client):
"""The real JID is extracted from the muc#user item of the presence."""
xep_0272 = XEP_0272(host)
presence_elt = domish.Element((None, "presence"))
x_elt = presence_elt.addElement((muc.NS_MUC_USER, "x"))
item_elt = x_elt.addElement((muc.NS_MUC_USER, "item"))
item_elt["jid"] = "peer@example.com/res"
assert xep_0272._get_presence_user_jid(presence_elt) == jid.JID(
"peer@example.com/res"
)
# no muc#user item
assert xep_0272._get_presence_user_jid(domish.Element((None, "presence"))) is None
def test_get_handler(self, host, client):
"""XEP_0272_handler is instantiated and returned."""
xep_0272 = XEP_0272(host)
@ -56,56 +75,138 @@ class TestXEP0272:
assert handler.plugin_parent == xep_0272
def test_on_muji_request_preparing_state(self, host, client):
"""try_to_finish_preparation is called when preparing_state is True."""
"""Our own preparing broadcast finishes the data collection."""
xep_0272 = XEP_0272(host)
presence_elt = domish.Element((None, "presence"))
presence_elt["from"] = TEST_ROOM_JID.full()
muji_elt = presence_elt.addElement((NS_MUJI, "muji"))
muji_elt.addElement("preparing")
muji_data = {"preparing_jids": set()}
muji_data = {"preparing_jids": set(), "to_call": set()}
with patch.object(xep_0272._muc, "get_room_user_jid", return_value=TEST_ROOM_JID):
with patch.object(xep_0272, "get_muji_data", return_value=muji_data):
with patch.object(
xep_0272, "try_to_finish_preparation"
) as mock_try_to_finish_preparation:
xep_0272.on_muji_request(presence_elt, client)
assert muji_data["done_collecting"] is True
mock_try_to_finish_preparation.assert_called_once()
def test_on_muji_request_not_preparing_state(self, host, client):
"""try_to_finish_preparation is not called when preparing_state is False."""
def test_on_muji_request_peer_data(self, host, client):
"""The real JID of a data-advertising peer is added to to_call."""
xep_0272 = XEP_0272(host)
presence_elt = domish.Element((None, "presence"))
presence_elt["from"] = "room@example.com/user"
presence_elt["from"] = "room@example.com/peer"
presence_elt.addElement((NS_MUJI, "muji"))
x_elt = presence_elt.addElement((muc.NS_MUC_USER, "x"))
item_elt = x_elt.addElement((muc.NS_MUC_USER, "item"))
item_elt["jid"] = "peer@example.com/res"
muji_data = {"done_preparing": True, "preparing_jids": set(), "to_call": set()}
with patch.object(xep_0272, "get_muji_data", return_value=muji_data):
with patch.object(
xep_0272, "try_to_finish_preparation"
) as mock_try_to_finish_preparation:
xep_0272.on_muji_request(presence_elt, client)
mock_try_to_finish_preparation.assert_not_called()
room = MagicMock()
room.roomJID = TEST_ROOM_JID.userhostJID()
muji_data = {"preparing_jids": set(), "to_call": set()}
with (
patch.object(xep_0272._muc, "get_room", return_value=room),
patch.object(xep_0272._muc, "get_room_user_jid", return_value=TEST_ROOM_JID),
patch.object(xep_0272, "get_muji_data", return_value=muji_data),
patch.object(host.bridge, "call_group_setup") as mock_call_group_setup,
):
# the peer's data is collected, but the preparation is not finished yet
# (our own preparation broadcast has not been received)
xep_0272.on_muji_request(presence_elt, client)
assert "done_preparing" not in muji_data
mock_call_group_setup.assert_not_called()
# our own preparation broadcast: the data collection is done, finish and
# ask the frontend to initiate the sessions
own_presence = domish.Element((None, "presence"))
own_presence["from"] = TEST_ROOM_JID.full()
muji_elt = own_presence.addElement((NS_MUJI, "muji"))
muji_elt.addElement("preparing")
xep_0272.on_muji_request(own_presence, client)
assert muji_data["done_preparing"] is True
mock_call_group_setup.assert_called_once()
to_call = data_format.deserialise(mock_call_group_setup.call_args.args[1])[
"to_call"
]
assert to_call == ["peer@example.com/res"]
# the preparation is already done: a data presence received now is
# ignored, no new setup is emitted
mock_call_group_setup.reset_mock()
xep_0272.on_muji_request(presence_elt, client)
mock_call_group_setup.assert_not_called()
@ed
async def test_on_session_initiate(self, host, client):
"""The MUJI room is only trusted if the sender is a member of the room."""
xep_0272 = XEP_0272(host)
jingle_elt = domish.Element((None, "jingle"))
muji_elt = jingle_elt.addElement((NS_MUJI, "muji"))
muji_elt["room"] = "room@example.com"
# sender is a member: the room is stored
with patch.object(xep_0272._muc, "is_jid_in_room", return_value=True):
session = {"peer_jid": jid.JID("peer@example.com/res")}
assert await xep_0272._on_session_initiate(client, session, None, jingle_elt)
assert session["muji_room"] == "room@example.com"
# sender is not a member: the room is ignored
with patch.object(xep_0272._muc, "is_jid_in_room", return_value=False):
session = {"peer_jid": jid.JID("stranger@example.com")}
assert await xep_0272._on_session_initiate(client, session, None, jingle_elt)
assert "muji_room" not in session
def test_on_muji_request_semi_anonymous_room(self, host, client):
"""A room which does not reveal the real JIDs can't be used for MUJI."""
xep_0272 = XEP_0272(host)
presence_elt = domish.Element((None, "presence"))
presence_elt["from"] = "room@example.com/peer"
presence_elt.addElement((NS_MUJI, "muji"))
# no muc#user item: the sender's real JID is not available
room = MagicMock()
room.roomJID = TEST_ROOM_JID.userhostJID()
muji_data = {"preparing_jids": set(), "to_call": set()}
with (
patch.object(xep_0272._muc, "get_room", return_value=room),
patch.object(xep_0272._muc, "get_room_user_jid", return_value=TEST_ROOM_JID),
patch.object(xep_0272, "get_muji_data", return_value=muji_data),
pytest.raises(exceptions.ExternalRequestError),
):
xep_0272.on_muji_request(presence_elt, client)
def test_on_muji_request_own_jid(self, host, client):
"""try_to_finish_preparation is not called when the presence is from own JID."""
"""try_to_finish_preparation is not called for a presence from own JID."""
xep_0272 = XEP_0272(host)
presence_elt = domish.Element((None, "presence"))
presence_elt["from"] = "room@example.com/user"
presence_elt.addElement((NS_MUJI, "muji"))
client.jid = jid.JID("room@example.com/user")
with patch.object(
xep_0272, "try_to_finish_preparation"
) as mock_try_to_finish_preparation:
muji_data = {"preparing_jids": set(), "to_call": set()}
with (
patch.object(
xep_0272._muc,
"get_room_user_jid",
return_value=jid.JID("room@example.com/user"),
),
patch.object(xep_0272, "get_muji_data", return_value=muji_data),
patch.object(
xep_0272, "try_to_finish_preparation"
) as mock_try_to_finish_preparation,
):
xep_0272.on_muji_request(presence_elt, client)
mock_try_to_finish_preparation.assert_not_called()
def test_try_to_finish_preparation(self, host, client):
"""try_to_finish_preparation sets done_preparing to True and emits the signal."""
"""call_group_setup is only emitted once the data collection is done and no
other preparation is left to wait."""
xep_0272 = XEP_0272(host)
room = MagicMock()
muji_data = {"preparing_jids": set(), "to_call": set()}
room.roomJID = TEST_ROOM_JID.userhostJID()
muji_data = {
"preparing_jids": set(),
"to_call": {jid.JID("early@example.com")},
}
# signals are synchronous (see _Bridge.send_signal): use a plain mock, not the
# async one of the bridge fixture
@ -113,9 +214,27 @@ class TestXEP0272:
patch.object(xep_0272, "get_muji_data", return_value=muji_data),
patch.object(host.bridge, "call_group_setup") as mock_call_group_setup,
):
# our own preparation broadcast has not been received yet: nothing is done
xep_0272.try_to_finish_preparation(client, room, muji_data)
mock_call_group_setup.assert_not_called()
assert "done_preparing" not in muji_data
# data collection done, no preparation left: finish and ask the frontend to
# initiate the sessions
muji_data["done_collecting"] = True
xep_0272.try_to_finish_preparation(client, room, muji_data)
assert muji_data["done_preparing"] is True
mock_call_group_setup.assert_called_once()
to_call = data_format.deserialise(mock_call_group_setup.call_args.args[1])[
"to_call"
]
assert to_call == ["early@example.com"]
# a preparation is still in progress: nothing more is done
muji_data["preparing_jids"] = {jid.JID("room@example.com/other")}
mock_call_group_setup.reset_mock()
xep_0272.try_to_finish_preparation(client, room, muji_data)
mock_call_group_setup.assert_not_called()
@ed
async def test_call_group_data_set(self, host, client):
@ -130,10 +249,10 @@ class TestXEP0272:
return_value=(MagicMock(), MagicMock()),
):
with patch.object(
client, "a_send", return_value=defer.succeed(None)
) as mock_a_send:
client.presence, "send", return_value=defer.succeed(None)
) as mock_presence_send:
await xep_0272.call_group_data_set(client, room_jid, call_data)
mock_a_send.assert_called()
mock_presence_send.assert_called()
@ed
async def test_start_preparation(self, host, client):
@ -147,10 +266,10 @@ class TestXEP0272:
return_value=(MagicMock(), MagicMock()),
):
with patch.object(
client, "a_send", return_value=defer.succeed(None)
) as mock_a_send:
client.presence, "send", return_value=defer.succeed(None)
) as mock_presence_send:
await xep_0272.start_preparation(client, room)
mock_a_send.assert_called()
mock_presence_send.assert_called()
class TestXEP0272Handler: