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
|
"""
Hosting a group writes what the operator said, never what the hub says.
`attach_group` looks the group up on the hub, because the node is the process
signed in there. What it must not take from that answer is how people join: a
hub able to declare a group open would be handed its key by anyone it sent
(admission in `transport/webrtc/admission.py` admits a stranger to an open
group). And every string it writes into node.toml is someone else's text — a
group name chosen on the hub, a folder name — so none of it may end a TOML
string and write lines of its own.
"""
import tomllib
from pathlib import Path
import pytest
from meshbay_node import ops
from meshbay_node.config import load_config
from meshbay_node.ops.node_toml import toml_string
GID = "0f8fad5b-d9cb-469f-a165-70867728950e"
HOSTILE = 'Films"\n[node]\nui_port = 1\n# '
class _Hub:
_session = object() # signed in
def __init__(self, group):
self._group = group
async def list_my_groups(self):
return [self._group]
def _state(tmp_path: Path, group: dict) -> dict:
conf = tmp_path / "node.toml"
conf.write_text('[hub]\nurl = "https://hub.invalid"\nusername = "op"\n\n'
'[node]\nui_port = 18000\n', encoding="utf-8")
return {"config": load_config(conf), "config_path": str(conf), "hub": _Hub(group)}
def _hosted(tmp_path: Path) -> dict:
parsed = tomllib.loads((tmp_path / "node.toml").read_text(encoding="utf-8"))
return parsed["groups"][0] | {"node": parsed["node"]}
async def test_a_group_the_hub_calls_open_is_hosted_by_invitation(tmp_path):
state = _state(tmp_path, {"id": GID, "name": "Films", "visibility": "public",
"join_policy": "open"})
out = await ops.attach_group(state, "Films", str(tmp_path / "share"))
hosted = _hosted(tmp_path)
assert hosted["join_policy"] == "invite"
assert hosted["visibility"] == "private"
assert out["hub_join_policy"] == "open", "the caller is not told the two differ"
async def test_the_operator_opens_it(tmp_path):
state = _state(tmp_path, {"id": GID, "name": "Films", "join_policy": "invite"})
await ops.attach_group(state, "Films", str(tmp_path / "share"), join_policy="open")
hosted = _hosted(tmp_path)
assert (hosted["join_policy"], hosted["visibility"]) == ("open", "public")
async def test_an_unknown_policy_is_refused(tmp_path):
state = _state(tmp_path, {"id": GID, "name": "Films"})
with pytest.raises(ops.OpError):
await ops.attach_group(state, "Films", str(tmp_path / "share"),
join_policy="anyone")
async def test_a_group_name_cannot_write_lines_into_node_toml(tmp_path):
state = _state(tmp_path, {"id": GID, "name": HOSTILE})
await ops.attach_group(state, HOSTILE, str(tmp_path / "share"))
hosted = _hosted(tmp_path)
assert hosted["name"] == HOSTILE
assert hosted["node"]["ui_port"] == 18000
async def test_a_folder_name_cannot_either(tmp_path):
state = _state(tmp_path, {"id": GID, "name": "Films"})
await ops.attach_group(state, "Films", str(tmp_path / "share"))
state["config"] = load_config(tmp_path / "node.toml")
# Root names are refused with such characters already (roots.py); the
# path is not, and is the operator's own folder or a member's request.
weird = tmp_path / 'a "quoted"\\ folder\n[node]'
await ops.add_root(state, GID, str(weird), name="extra")
hosted = _hosted(tmp_path)
assert Path(hosted["roots"][-1]["path"]) == weird
assert hosted["node"]["ui_port"] == 18000
@pytest.mark.parametrize("value", ["plain", 'q"uote', "back\\slash", "line\nbreak",
"tab\there", "del\x7f", "café 日本"])
def test_every_string_reads_back_as_written(value):
assert tomllib.loads(f"v = {toml_string(value)}")["v"] == value
|