vnctalk-prosody/tests/test_02_muc.py

"""MUC-specific tests for vnctalk requirements."""
import asyncio
import uuid
import pytest
import slixmpp
import xml.etree.ElementTree as ET


@pytest.mark.asyncio
class TestMuc:

    async def test_muc_component_disco(self, xmpp_client, xmpp_config):
        """MUC component must advertise MUC and MAM features."""
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")
        res = await xmpp_client.disco_info(to=muc_domain)
        features = [f.get("var") for f in res.xml.findall(".//{http://jabber.org/protocol/disco#info}feature")]
        assert "http://jabber.org/protocol/muc" in features, "MUC feature missing"
        assert "urn:xmpp:mam:2" in features, "MUC MAM feature missing"

    async def test_room_hidden_by_default(self, xmpp_client, xmpp_config):
        """Priority 4 from AUDIT: newly created rooms must not appear in disco#items."""
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")

        room_jid = f"hiddentest_{uuid.uuid4().hex[:8]}@{muc_domain}"
        nick = "creator"

        # Create room by joining
        await xmpp_client.join_muc(room_jid, nick)
        await asyncio.sleep(1)

        try:
            # Query MUC service disco#items
            res = await xmpp_client.disco_items(to=muc_domain)
            items = [i.get("jid") for i in res.xml.findall(".//{http://jabber.org/protocol/disco#items}item")]
            assert room_jid not in items, f"Newly created room {room_jid} appeared in public disco#items (should be hidden by default)"

            # Query room disco#info — should NOT advertise muc_public
            res2 = await xmpp_client.disco_info(to=room_jid)
            features = [f.get("var") for f in res2.xml.findall(".//{http://jabber.org/protocol/disco#info}feature")]
            assert "muc_public" not in features, "Room incorrectly advertises muc_public"
            assert "muc_hidden" in features, "Room should advertise muc_hidden"
        finally:
            await xmpp_client.leave_muc(room_jid, nick)
            await xmpp_client.destroy_muc(room_jid)

    async def test_muc_mam_query_returns_groupchat_only(self, xmpp_client, xmpp_config):
        """Priority 5 from AUDIT: MUC MAM archives groupchat messages, not presence."""
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")

        room_jid = f"mamcontent_{uuid.uuid4().hex[:8]}@{muc_domain}"
        nick = "testuser"

        await xmpp_client.join_muc(room_jid, nick)
        await asyncio.sleep(1)

        # Send a groupchat message
        msg_id = f"muc-msg-{uuid.uuid4().hex[:8]}"
        msg = xmpp_client.make_message(mto=room_jid, mtype="groupchat")
        msg["id"] = msg_id
        msg["body"] = f"Test MUC MAM message {msg_id}"  # slixmpp: dict-style, not msg.body
        msg.send()  # message.send() returns None, not awaitable
        await asyncio.sleep(1)

        try:
            # Query the room's MAM (MUC archive lives at the room, not the user)
            results = await xmpp_client.query_mam_and_collect(timeout=10, to=room_jid)
            msg_ids = []
            for r in results:
                forwarded = r.find("{urn:xmpp:forward:0}forwarded")
                if forwarded is not None:
                    m = next(
                        (c for c in forwarded if c.tag in (
                            "{jabber:client}message",
                            "{urn:xmpp:forward:0}message",
                            "message",
                        )),
                        None,
                    )
                    if m is not None:
                        msg_ids.append(m.get("id"))

            assert msg_id in msg_ids, f"Groupchat message {msg_id} not found in MUC MAM archive"
        finally:
            await xmpp_client.leave_muc(room_jid, nick)
            await xmpp_client.destroy_muc(room_jid)

    async def test_muc_broadcast_strips_spoofed_stanza_id(self, xmpp_client, xmpp_config, second_client):
        """Priority 5 from AUDIT: a client-injected <stanza-id> claiming the room
        JID is stripped from broadcast copies (anti-spoofing).

        Per XEP-0359 the room legitimately adds its OWN <stanza-id by=room>, so the
        contract is not "no stanza-id at all" — it is that a *spoofed* value
        supplied by the sender must not survive. mod_muc_mam strips by-room
        stanza-ids on muc-broadcast-message, then save_to_history adds the real one.
        """
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")

        # second_client must be a different account for this test to be meaningful
        if xmpp_client.boundjid.bare == second_client.boundjid.bare:
            pytest.skip("Need a second distinct account to observe broadcast copies")

        room_jid = f"nostanzaid_{uuid.uuid4().hex[:8]}@{muc_domain}"
        nick1 = "user1"
        nick2 = "user2"

        await xmpp_client.join_muc(room_jid, nick1)
        # unlock the freshly-created (locked) room so the second user can join
        await xmpp_client.configure_muc(room_jid)
        await second_client.join_muc(room_jid, nick2)
        await asyncio.sleep(1)

        spoof_id = f"spoofed-{uuid.uuid4().hex[:12]}"
        received = asyncio.Event()
        copies = []

        def on_message(msg):
            if msg["type"] == "groupchat" and str(msg["from"]).startswith(room_jid):
                copies.append(ET.tostring(msg.xml, encoding="unicode"))
                received.set()

        second_client.add_event_handler("message", on_message)
        try:
            msg = xmpp_client.make_message(mto=room_jid, mtype="groupchat")
            msg["body"] = "stanza-id spoof test"  # slixmpp: dict-style, not msg.body
            # inject a spoofed stanza-id claiming to come from the room
            sid = ET.SubElement(msg.xml, "{urn:xmpp:sid:0}stanza-id")
            sid.set("by", room_jid)
            sid.set("id", spoof_id)
            msg.send()  # message.send() returns None, not awaitable
            await asyncio.wait_for(received.wait(), timeout=5)
            assert not any(spoof_id in c for c in copies), (
                "Spoofed stanza-id survived broadcast — anti-spoof stripping contract broken"
            )
        finally:
            second_client.del_event_handler("message", on_message)
            await xmpp_client.leave_muc(room_jid, nick1)
            await second_client.leave_muc(room_jid, nick2)
            await xmpp_client.destroy_muc(room_jid)

    async def test_muc_automember_with_distinct_account(self, xmpp_client, xmpp_config, second_client):
        """Priority 7 from AUDIT: invite a distinct account; they must gain member affiliation automatically."""
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")

        # This test is only meaningful with two distinct bare JIDs
        if xmpp_client.boundjid.bare == second_client.boundjid.bare:
            pytest.xfail("Need --xmpp-jid2 with a distinct account to test automember meaningfully")

        room_jid = f"automem_{uuid.uuid4().hex[:8]}@{muc_domain}"
        nick1 = "owner"
        nick2 = "invited"

        await xmpp_client.join_muc(room_jid, nick1)
        # unlock the freshly-created (locked) room so the invitee can join
        await xmpp_client.configure_muc(room_jid)
        await asyncio.sleep(1)

        try:
            # Send mediated invite from owner to second account
            invite_msg = xmpp_client.make_message(mto=room_jid, mtype="normal")
            x = ET.SubElement(invite_msg.xml, "{http://jabber.org/protocol/muc#user}x")
            invite = ET.SubElement(x, "{http://jabber.org/protocol/muc#user}invite")
            invite.set("to", second_client.boundjid.bare)
            invite_msg.send()  # message.send() returns None, not awaitable
            await asyncio.sleep(1)

            # Second client joins
            await second_client.join_muc(room_jid, nick2)
            await asyncio.sleep(1)

            # Query member affiliations
            members = await xmpp_client.get_muc_affiliations(room_jid, "member")
            member_jids = [m.get("jid") for m in members]
            assert second_client.boundjid.bare in member_jids, "Automember did not grant membership after invite"
        finally:
            await xmpp_client.leave_muc(room_jid, nick1)
            await second_client.leave_muc(room_jid, nick2)
            await xmpp_client.destroy_muc(room_jid)

    async def test_muc_unregister_iq_removes_affiliation(self, xmpp_client, xmpp_config, second_client):
        """Priority 6 from AUDIT: xmpp:vnctalk:unregister:query removes affiliation."""
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")

        if xmpp_client.boundjid.bare == second_client.boundjid.bare:
            pytest.xfail("Need --xmpp-jid2 with a distinct account")

        room_jid = f"unreg_{uuid.uuid4().hex[:8]}@{muc_domain}"
        nick1 = "owner"
        nick2 = "member"

        await xmpp_client.join_muc(room_jid, nick1)
        # unlock the freshly-created (locked) room before affiliating others
        await xmpp_client.configure_muc(room_jid)
        await asyncio.sleep(0.5)
        # Make second client a member
        await xmpp_client.set_muc_affiliation(room_jid, second_client.boundjid.bare, "member")
        await asyncio.sleep(0.5)

        try:
            # Verify membership exists
            members_before = await xmpp_client.get_muc_affiliations(room_jid, "member")
            assert any(m.get("jid") == second_client.boundjid.bare for m in members_before), "Precondition: second client must be a member"

            # Send unregister IQ as the member.
            # handle_unregister_iq performs the removal but returns no IQ reply
            # (fire-and-forget), so the send will time out — that is expected;
            # we verify the side effect (affiliation removed) instead.
            iq = second_client.make_iq_set(ito=room_jid)
            ET.SubElement(iq.xml, "{xmpp:vnctalk:unregister}query")
            try:
                await iq.send(timeout=5)
            except slixmpp.exceptions.IqTimeout:
                pass
            await asyncio.sleep(1)

            # Verify affiliation removed
            members_after = await xmpp_client.get_muc_affiliations(room_jid, "member")
            assert not any(m.get("jid") == second_client.boundjid.bare for m in members_after), "Unregister IQ did not remove affiliation"
        finally:
            await xmpp_client.leave_muc(room_jid, nick1)
            await xmpp_client.destroy_muc(room_jid)

    async def test_muc_mam_presence_not_archived(self, xmpp_client, xmpp_config):
        """Priority 1 from AUDIT: presence stanzas sent to a room must NOT appear in MUC MAM."""
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")

        room_jid = f"mampres_{uuid.uuid4().hex[:8]}@{muc_domain}"
        nick = "testuser"

        await xmpp_client.join_muc(room_jid, nick)
        await asyncio.sleep(1)

        # Send a presence update to the room (not a join, just an update)
        presence = xmpp_client.make_presence(pto=f"{room_jid}/{nick}")
        presence.send()  # presence.send() returns None, not awaitable
        await asyncio.sleep(1)

        try:
            # Query MUC MAM
            results = await xmpp_client.query_mam_and_collect(timeout=10)
            presence_count = 0
            for r in results:
                forwarded = r.find("{urn:xmpp:forward:0}forwarded")
                if forwarded is not None:
                    m = forwarded.find("{jabber:client}presence")
                    if m is not None:
                        presence_count += 1

            assert presence_count == 0, (
                f"Found {presence_count} presence stanza(s) in MUC MAM archive — "
                "mod_muc_mam.lua presence-exclusion contract broken"
            )
        finally:
            await xmpp_client.leave_muc(room_jid, nick)
            await xmpp_client.destroy_muc(room_jid)

    async def test_hidden_lib_rejects_public_override(self, xmpp_client, xmpp_config):
        """Priority 2 from AUDIT: when restrict_public is true, config form must not contain publicroom field,
        and submitting publicroom=true must not make the room public (unless actor is server admin)."""
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")

        room_jid = f"hiderej_{uuid.uuid4().hex[:8]}@{muc_domain}"
        nick = "owner"

        await xmpp_client.join_muc(room_jid, nick)
        await asyncio.sleep(1)

        try:
            # 1. Request config form — publicroom field should be absent
            iq = xmpp_client.make_iq_get(ito=room_jid)
            ET.SubElement(iq.xml, "{http://jabber.org/protocol/muc#owner}query")
            res = await iq.send()
            form = res.xml.find(".//{jabber:x:data}x")
            assert form is not None, "No config form returned"
            fields = [f.get("var") for f in form.findall("{jabber:x:data}field")]
            if "muc#roomconfig_publicroom" in fields:
                # hidden.lib only hides this field when restrict_public is true
                # (muc_room_allow_public = false). This deployment allows public
                # rooms, so the restriction-override scenario does not apply.
                pytest.skip(
                    "Server does not restrict public rooms (muc_room_allow_public not false); "
                    "restrict_public override test is not applicable"
                )

            # 2. Try to submit publicroom=true anyway (defense-in-depth)
            iq_set = xmpp_client.make_iq_set(ito=room_jid)
            query = ET.SubElement(iq_set.xml, "{http://jabber.org/protocol/muc#owner}query")
            x = ET.SubElement(query, "{jabber:x:data}x")
            x.set("type", "submit")
            field = ET.SubElement(x, "{jabber:x:data}field")
            field.set("var", "muc#roomconfig_publicroom")
            field.set("type", "boolean")
            ET.SubElement(field, "{jabber:x:data}value").text = "1"
            await iq_set.send()
            await asyncio.sleep(0.5)

            # 3. Verify room state
            res2 = await xmpp_client.disco_info(to=room_jid)
            features = [f.get("var") for f in res2.xml.findall(".//{http://jabber.org/protocol/disco#info}feature")]
            if "muc_public" in features:
                # The override succeeded — this is allowed for server admins.
                # We cannot distinguish admin vs non-admin from the client side,
                # so mark as expected failure for admin accounts.
                pytest.xfail(
                    "Room became public after override submission — test account may be a server admin, "
                    "which is allowed to override. Non-admin accounts should be rejected."
                )
            assert "muc_hidden" in features, "Room should still advertise muc_hidden after rejected override"
        finally:
            await xmpp_client.leave_muc(room_jid, nick)
            await xmpp_client.destroy_muc(room_jid)

    async def test_muc_vdata_disco(self, xmpp_client, xmpp_config):
        """MUC disco#info should contain vdata form field if mod_vnc_muc_data is loaded."""
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")
        room_jid = f"vdata_{uuid.uuid4().hex[:8]}@{muc_domain}"
        nick = "test"
        await xmpp_client.join_muc(room_jid, nick)
        await asyncio.sleep(1)

        try:
            res = await xmpp_client.disco_info(to=room_jid)
            # the data form is nested inside <query>, not a direct child of <iq>
            x = res.xml.find(".//{jabber:x:data}x")
            if x is None:
                pytest.fail("No extended disco info form returned")
            fields = [f.get("var") for f in x.findall("{jabber:x:data}field")]
            assert "muc#roominfo_vdata" in fields, "mod_vnc_muc_data field missing — module may not be loaded"
        finally:
            await xmpp_client.leave_muc(room_jid, nick)
            await xmpp_client.destroy_muc(room_jid)


@pytest.mark.asyncio
class TestVcardMuc:
    """mod_vcard_muc: room vCard get/set persistence."""

    async def test_muc_vcard_get_set(self, xmpp_client, xmpp_config):
        muc_domain = xmpp_config.get("muc_domain")
        if not muc_domain:
            pytest.skip("No MUC domain configured")

        room_jid = f"vcardmuc_{uuid.uuid4().hex[:8]}@{muc_domain}"
        nick = "owner"
        await xmpp_client.join_muc(room_jid, nick)
        await xmpp_client.configure_muc(room_jid)
        await asyncio.sleep(0.5)

        try:
            # A brand-new room has no vCard → item-not-found.
            iq = xmpp_client.make_iq_get(ito=room_jid)
            iq.append(ET.Element("{vcard-temp}vCard"))
            try:
                await iq.send(timeout=8)
                # Some builds return an empty vCard instead of an error; that's
                # also acceptable as the "no vCard yet" state.
            except slixmpp.exceptions.IqError as e:
                assert e.condition == "item-not-found", (
                    f"expected item-not-found for empty room vCard, got {e.condition}"
                )

            # Set a vCard with FN (owner affiliation is required to set).
            vcard = ET.Element("{vcard-temp}vCard")
            fn = ET.SubElement(vcard, "{vcard-temp}FN")
            fn.text = f"Room Display Name {uuid.uuid4().hex[:6]}"
            iq = xmpp_client.make_iq_set(ito=room_jid)
            iq.append(vcard)
            res = await iq.send(timeout=8)
            assert res["type"] == "result", f"vCard set failed: {res['type']}"

            # Re-query and assert FN persisted.
            iq = xmpp_client.make_iq_get(ito=room_jid)
            iq.append(ET.Element("{vcard-temp}vCard"))
            res = await iq.send(timeout=8)
            fn_el = res.xml.find(".//{vcard-temp}FN")
            assert fn_el is not None and fn_el.text == fn.text, (
                f"room vCard FN did not persist; got {fn_el.text if fn_el is not None else None!r}"
            )
        finally:
            await xmpp_client.leave_muc(room_jid, nick)
            await xmpp_client.destroy_muc(room_jid)