feat: let the operator purge a group's chat (MNP 6.1)

Signed chat_purge from the Chat settings deletes every stored message;
epoch keys and attachments stay. The ack is broadcast so open chat
panels empty.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 110 insertions, 0 deletions

diff --git a/packages/meshbay-node/tests/test_chat_purge.py b/packages/meshbay-node/tests/test_chat_purge.py
new file mode 100644
index 0000000..37bd954
--- /dev/null
+++ b/packages/meshbay-node/tests/test_chat_purge.py
@@ -0,0 +1,110 @@
+"""
+The operator's chat purge (`chat_purge`, MNP 6.1): every stored message of the
+group goes, signed like every operator instruction, and every connected member
+is told so their open chat empties.
+
+Driven through `_do_admin_response`, as in `test_admin_ops_mnp.py`: the node
+rebuilds the transcript from the operation it holds, and skipping that would
+test nothing.
+"""
+
+import base64
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_common.adminop import OP_CHAT_PURGE, admin_transcript
+from meshbay_node import ops
+from meshbay_node.chat import FORMAT_SEALED_V1, ChatStore
+from meshbay_node.roster import open_roster
+from test_admin_ops_mnp import GROUP, _drain, _last, _session
+
+
+@pytest.fixture
+async def roster(tmp_path):
+    r = await open_roster(tmp_path)
+    yield r
+    await r.close()
+
+
+async def _with_chat(tmp_path, roster, *, operator=True, messages=2):
+    session = await _session(tmp_path, roster, operator=operator)
+    store = ChatStore(tmp_path / "chat.db")
+    await store.open()
+    for i in range(messages):
+        await store.save_message(
+            "grenet", 0, b"ct", format=FORMAT_SEALED_V1, epoch=1,
+            device=b"d" * 32, nonce=bytes([i]) * 12, sig=b"s" * 64)
+    session.state["groups_ctx"][GROUP]["chat_store"] = store
+    # A second member connected to the group, to see the broadcast arrive.
+    other = []
+    session.state["groups_ctx"][GROUP]["_peers"] = {
+        "grenet": session, "bob": type("Peer", (), {"_send": other.append})()}
+    return session, store, other
+
+
+def _sign(session, challenge, sk=None):
+    transcript = admin_transcript(
+        op=OP_CHAT_PURGE, node_pk_b64=session._node_pk_b64(), group_id=GROUP,
+        subject=challenge["subject"], nonce=base64.b64decode(challenge["nonce"]),
+        ts=challenge["ts"])
+    sig = (sk or session.sk_op).sign(transcript)
+    session._do_admin_response({"op_id": challenge["op_id"],
+                                "signature": base64.b64encode(sig).decode()})
+
+
+async def test_a_signed_purge_empties_the_chat_and_tells_the_group(tmp_path, roster):
+    session, store, other = await _with_chat(tmp_path, roster)
+
+    session._do_chat_purge({})
+    challenge = _last(session)
+    assert challenge["type"] == "admin_challenge"
+    assert challenge["op"] == OP_CHAT_PURGE and challenge["subject"] == GROUP
+    _sign(session, challenge)
+    await _drain(session)
+
+    assert await store.message_count() == 0
+    assert other[-1]["type"] == "chat_purge_ack" and other[-1]["removed"] == 2
+    assert _last(session)["type"] == "chat_purge_ack"
+    await store.close()
+
+
+async def test_the_purge_names_the_connection_group_and_no_other(tmp_path, roster):
+    """A group named in the message is not the one purged: the ack goes to this
+    connection's group, and so does the operation."""
+    session, store, _ = await _with_chat(tmp_path, roster)
+
+    session._do_chat_purge({"group_id": "h" * 32})
+
+    assert _last(session)["subject"] == GROUP
+    await store.close()
+
+
+async def test_a_signature_from_a_non_operator_key_purges_nothing(tmp_path, roster):
+    session, store, other = await _with_chat(tmp_path, roster)
+
+    session._do_chat_purge({})
+    _sign(session, _last(session), sk=Ed25519PrivateKey.generate())
+    await _drain(session)
+
+    assert _last(session)["type"] == "error"
+    assert await store.message_count() == 2
+    assert other == []
+    await store.close()
+
+
+async def test_no_challenge_without_an_operator(tmp_path, roster):
+    session, store, _ = await _with_chat(tmp_path, roster, operator=False)
+
+    session._do_chat_purge({})
+
+    assert _last(session)["type"] == "error"
+    assert session._admin_ops == {}
+    await store.close()
+
+
+async def test_a_group_without_a_chat_store_says_so(tmp_path, roster):
+    session = await _session(tmp_path, roster, operator=True)
+
+    with pytest.raises(ops.OpError) as e:
+        await ops.purge_chat(session.state, GROUP)
+    assert e.value.status == 404