From 14dbca530b9b877d2915b2b2da72feb941fea735 Mon Sep 17 00:00:00 2001 From: Goffi Date: Tue, 18 Aug 2026 18:44:19 +0200 Subject: [PATCH] plugin XEP-0313: check message origin: Check that "from" attribute of the forwarded message is coming from the service. --- libervia/backend/plugins/plugin_xep_0313.py | 20 +++++ tests/unit/test_plugin_xep_0313.py | 98 +++++++++++++++++++++ 2 files changed, 118 insertions(+) create mode 100644 tests/unit/test_plugin_xep_0313.py diff --git a/libervia/backend/plugins/plugin_xep_0313.py b/libervia/backend/plugins/plugin_xep_0313.py index 0462ae75..b056212f 100644 --- a/libervia/backend/plugins/plugin_xep_0313.py +++ b/libervia/backend/plugins/plugin_xep_0313.py @@ -407,6 +407,26 @@ class XEP_0313: f"{mess_elt.toXml()}" ) raise exceptions.DataError("Invalid element") + if service is not None: + # the forwarded message must come from the queried service, else a + # malicious service could inject messages from arbitrary JIDs in its + # archive (e.g., in a MUC archive, "from" MUST be the sender's occupant + # JID) + try: + fwd_from_jid = jid.JID(fwd_message_elt["from"]) + except (KeyError, jid.InvalidFormat): + log.warning( + f'MAM message from {service_jid} has an invalid "from", ' + f"hack attempt?\n{mess_elt.toXml()}" + ) + raise exceptions.DataError("Invalid element") + if fwd_from_jid.userhostJID() != service_jid: + log.warning( + f'MAM message from {service_jid} has a "from" which is not from ' + f"the service ({fwd_from_jid.full()}), hack attempt?\n" + f"{mess_elt.toXml()}" + ) + raise exceptions.DataError("Invalid element") stanza_id = self._sid.get_stanza_id(fwd_message_elt, service_jid) if stanza_id is None: # not stanza-id element is present, we add one so message diff --git a/tests/unit/test_plugin_xep_0313.py b/tests/unit/test_plugin_xep_0313.py new file mode 100644 index 00000000..1684e594 --- /dev/null +++ b/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 . + +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 (the MAM service) + @param fwd_from: "from" of the forwarded + @param body: forwarded message body + @return: the result 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"