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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
"""
Which folder (possibly a subfolder of a shared root) is the Music app's
entry point for a group. Same shape as test_video_root_policy.py — a
signed operator instruction, per-group, stored via roster.py's
group_settings table, added later once a real messy library showed
musicbay.md's original "no root, whole shared tree" call was wrong.
"""
from pathlib import Path
import pytest
from meshbay_common.adminop import OP_AUDIO_ROOT
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roster import Roster
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from conftest import one_root
pytestmark = pytest.mark.asyncio
def _session(tmp_path: Path, user_id: str, *, operator: str | None = None) -> WebRTCPeerSession:
shared_root = tmp_path / "shared"
shared_root.mkdir(exist_ok=True)
(shared_root / "Music").mkdir()
(shared_root / "Podcasts").mkdir()
index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
ctx = {
"roots": one_root(shared_root),
"index": index,
"sk_node": index.sk_node,
"node_user_id": operator,
}
session = WebRTCPeerSession.__new__(WebRTCPeerSession)
session._ctx = ctx
session._group_id = None
session._user_id = user_id
session._pk_user = ""
session.sent = []
session._send = session.sent.append
session._audit = lambda *a, **k: None
return session
# ── Refused before a challenge is even issued ───────────────────────────────
async def test_missing_path_is_refused(tmp_path):
session = _session(tmp_path, "op", operator="op")
session._has_admin_authority = lambda: True
issued = []
session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
session._do_audio_root({})
assert not issued
assert [m for m in session.sent if m.get("type") == "error"]
async def test_a_nonexistent_folder_is_refused(tmp_path):
session = _session(tmp_path, "op", operator="op")
session._has_admin_authority = lambda: True
issued = []
session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
session._do_audio_root({"path": "shared/Nonexistent"})
assert not issued, "a mistyped path must be refused before a signature round trip"
assert [m for m in session.sent if m.get("type") == "error"]
async def test_path_traversal_is_refused(tmp_path):
session = _session(tmp_path, "op", operator="op")
session._has_admin_authority = lambda: True
issued = []
session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
session._do_audio_root({"path": "../../etc"})
assert not issued
assert [m for m in session.sent if m.get("type") == "error"]
async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
session = _session(tmp_path, "member-1", operator="the-operator")
session._has_admin_authority = lambda: False
session._do_audio_root({"path": "shared/Music"})
assert [m for m in session.sent if m.get("type") == "error"]
# ── Accepted cases ───────────────────────────────────────────────────────────
async def test_an_empty_path_is_always_accepted(tmp_path):
"""Empty means 'unset' — Music shows nothing yet, always valid to clear."""
session = _session(tmp_path, "op", operator="op")
session._has_admin_authority = lambda: True
issued = []
session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
session._do_audio_root({"path": ""})
assert issued == [(OP_AUDIO_ROOT, "")]
async def test_a_real_subfolder_is_accepted_and_signed(tmp_path):
session = _session(tmp_path, "op", operator="op")
session._has_admin_authority = lambda: True
issued = []
session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
session._do_audio_root({"path": "shared/Music"})
assert issued == [(OP_AUDIO_ROOT, "shared/Music")]
# ── Where it is stored ──────────────────────────────────────────────────────
async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
roster = Roster(db_path=tmp_path / "roster.db")
await roster.open()
try:
assert await roster.audio_root("g1") == "", "absent must mean unset"
await roster.set_audio_root("g1", "shared/Music", set_by="op")
assert await roster.audio_root("g1") == "shared/Music"
finally:
await roster.close()
reopened = Roster(db_path=tmp_path / "roster.db")
await reopened.open()
try:
assert await reopened.audio_root("g1") == "shared/Music"
assert await reopened.audio_root("g2") == "", "one group's setting must not answer for another"
finally:
await reopened.close()
|