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
|
"""
What of the operator's disk is shared, and who may write there, is decided on
the node's own machine — never over MNP (MNP 6.0).
A signed operation proves that the operator's key signed, not that the operator
meant it: in a browser the key is driven by code the hub serves, and in the
desktop application by a renderer that parses content from nodes. `root_add`,
`root_update` and `group_attach` let either of them share any folder on the
machine, or open one to writes, from anywhere. They are gone; the loopback API
(behind a native dialog in the desktop application) and the CLI remain.
And the consequence that has to hold for the operator's list to stay true: a
flag changed through the loopback API reaches every connected page, which the
MNP ack used to do.
"""
from dataclasses import asdict
from pathlib import Path
from types import SimpleNamespace
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common import adminop
from meshbay_common.protocol import MNP
from meshbay_node import ops
from meshbay_node.config import GroupConfig, RootSpec
from meshbay_node.indexer.indexer import DirectoryIndexer
from meshbay_node.roots import RootSet
from meshbay_node.transport.webrtc.admin import _ADMIN_EXECUTORS
from meshbay_node.transport.webrtc.dispatch import _HANDLERS
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
from test_admin_challenge_bounds import _PC, _Channel
GROUP = "g" * 32
GONE = ("root_add", "root_update", "group_attach")
@pytest.mark.parametrize("mtype", GONE)
def test_no_message_widens_what_the_node_shares(mtype):
assert mtype not in _HANDLERS, f"{mtype} is dispatched again"
assert mtype not in _ADMIN_EXECUTORS, f"{mtype} can be executed again"
assert mtype not in vars(MNP).values(), f"{mtype} is back in the protocol"
assert mtype not in vars(adminop).values(), f"{mtype} is a signed op again"
@pytest.mark.parametrize("mtype", GONE)
def test_an_older_client_asking_is_issued_nothing_to_sign(mtype):
"""A 5.x client still sends these. Nothing it sends may come back as a
challenge — a challenge is what a compromised page needs signed."""
ctx = {"sk_node": Ed25519PrivateKey.from_private_bytes(b"\x01" * 32),
"groups": {GROUP: {}}, "has_admin_authority": True}
s = WebRTCPeerSession(_PC(), ctx, peer_id="peer")
s._channel = _Channel()
s._audit = lambda *a, **k: None
s._user_id, s._group_id = "operator", GROUP
s._dispatch_message({"type": mtype, "group_id": GROUP, "path": "/home/someone",
"shared_dir": "/home/someone", "name": "x",
"root_name": "x", "writable": True})
assert s._admin_ops == {}
assert not [m for m in s._channel.sent if m.get("type") == "admin_challenge"]
def _state(tmp_path: Path, indexer_roots: RootSet | None = None):
(tmp_path / "media").mkdir()
cfg = GroupConfig(id=GROUP, name="plop",
roots=[RootSpec(path=str(tmp_path / "media"), name="media")])
conf = tmp_path / "node.toml"
conf.write_text(
f'[[groups]]\nid = "{GROUP}"\nname = "plop"\n\n'
f' [[groups.roots]]\n path = "{(tmp_path / "media").as_posix()}"\n'
f' name = "media"\n', encoding="utf-8")
live = RootSet.build([asdict(r) for r in cfg.roots])
pushed: list[list[dict]] = []
async def on_change(indexer):
pushed.append(indexer.index.roots)
indexer = DirectoryIndexer(indexer_roots or live, GROUP,
Ed25519PrivateKey.generate(), None,
on_change=on_change)
state = {"config": SimpleNamespace(groups=[cfg]), "config_path": str(conf),
"groups_ctx": {GROUP: {"roots": live}},
"indexers": {GROUP: indexer}}
return state, pushed
async def test_a_flag_changed_on_the_node_reaches_every_open_page(tmp_path):
state, pushed = _state(tmp_path)
await ops.update_root(state, GROUP, "media", writable=True)
assert pushed, "nothing was pushed — open pages keep the old flag"
assert pushed[-1][0]["writable"] is True
async def test_the_pushed_table_is_right_when_the_indexer_holds_its_own_set(tmp_path):
"""The indexer and the group context normally share one RootSet; when they
do not, the table peers receive is the indexer's, and must carry the flag."""
(tmp_path / "media").mkdir()
own = RootSet.build([{"path": str(tmp_path / "media"), "name": "media"}])
(tmp_path / "media").rmdir()
state, pushed = _state(tmp_path, indexer_roots=own)
await ops.update_root(state, GROUP, "media", removable=True)
assert pushed[-1][0]["removable"] is True
assert state["groups_ctx"][GROUP]["roots"].by_name("media").removable is True
|