plugin XEP-0313: check message origin:

Check that "from" attribute of the forwarded message is coming from the service.

tests/unit/test_plugin_xep_0313.py

@@ -0,0 +1,98 @@
+#!/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 <http://www.gnu.org/licenses/>.
+
+from unittest.mock import patch
+
+from pytest import fixture, raises
+from twisted.words.protocols.jabber import jid
+from twisted.words.xish import domish
+from tx_xmpp import mam
+
+from libervia.backend.core import exceptions
+from libervia.backend.core.constants import Const as C
+from libervia.backend.plugins.plugin_xep_0059 import XEP_0059
+from libervia.backend.plugins.plugin_xep_0313 import XEP_0313
+from libervia.backend.plugins.plugin_xep_0359 import XEP_0359
+
+ROOM_JID = jid.JID("room@conference.test.example")
+
+
+def make_mam_result(query_id, from_jid, fwd_from, body="Hello"):
+    """Helper to build a MAM result element wrapping a forwarded message.
+
+    @param query_id: MAM query id to put in the result element
+    @param from_jid: "from" of the result <message/> (the MAM service)
+    @param fwd_from: "from" of the forwarded <message/>
+    @param body: forwarded message body
+    @return: the result <message/> element
+    """
+    fwd_message = domish.Element(
+        (C.NS_CLIENT, "message"),
+        attribs={"from": fwd_from, "type": "groupchat"},
+    )
+    fwd_message.addElement((C.NS_CLIENT, "body"), content=body)
+    forwarded = domish.Element((C.NS_FORWARD, "forwarded"))
+    forwarded.addChild(
+        domish.Element((C.NS_DELAY, "delay"), attribs={"stamp": "2025-06-01T10:00:00Z"})
+    )
+    forwarded.addChild(fwd_message)
+    result = domish.Element(
+        (mam.NS_MAM, "result"),
+        attribs={"queryid": query_id, "id": "archive-1"},
+    )
+    result.addChild(forwarded)
+    mess = domish.Element((None, "message"), attribs={"from": from_jid})
+    mess.addChild(result)
+    return mess
+
+
+@fixture
+def xep_0313(host):
+    """Create XEP-0313 plugin instance with its dependencies."""
+    with patch.dict(
+        host.plugins,
+        {"XEP-0059": XEP_0059(host), "XEP-0359": XEP_0359(host)},
+    ):
+        yield XEP_0313(host)
+
+
+class TestMamResultValidation:
+    """Validation of forwarded messages in MAM results."""
+
+    def test_muc_result_from_room_occupant(self, xep_0313, client):
+        """A MUC archive message from a room occupant is accepted."""
+        mam_req = mam.MAMRequest(query_id="q1")
+        mess_elt = make_mam_result("q1", ROOM_JID.full(), f"{ROOM_JID.full()}/romeo")
+        fwd_message = xep_0313.get_message_from_result(
+            client, mess_elt, mam_req, service=ROOM_JID
+        )
+        assert fwd_message.getAttribute("from") == f"{ROOM_JID.full()}/romeo"
+
+    def test_muc_result_from_arbitrary_jid_rejected(self, xep_0313, client):
+        """A MUC archive message from a third-party JID is rejected."""
+        mam_req = mam.MAMRequest(query_id="q1")
+        mess_elt = make_mam_result("q1", ROOM_JID.full(), "malicious@example.com")
+        with raises(exceptions.DataError):
+            xep_0313.get_message_from_result(client, mess_elt, mam_req, service=ROOM_JID)
+
+    def test_user_result_from_contact_accepted(self, xep_0313, client):
+        """A user archive message from a contact is accepted."""
+        mam_req = mam.MAMRequest(query_id="q1")
+        mess_elt = make_mam_result("q1", client.jid.userhost(), "romeo@example.com")
+        fwd_message = xep_0313.get_message_from_result(client, mess_elt, mam_req)
+        assert fwd_message.getAttribute("from") == "romeo@example.com"