1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
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
|