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
138
139
140
141
|
"""
Which folder (possibly a subfolder of a shared root) is the Videos app's
entry point for a group. Same shape as test_apps_enabled_policy.py: a
signed operator instruction, per-group (unlike tmdb_config, which is
node-wide), stored via roster.py's group_settings table.
Specific to this one: a non-empty path must resolve to a real, readable
directory inside one of the group's own roots before a challenge is ever
issued — refusing a typo up front, the same way an empty apps set is
refused up front rather than round-tripped to the operator's browser.
"""
from pathlib import Path
import pytest
from meshbay_common.adminop import OP_VIDEO_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 / "Movies").mkdir()
(shared_root / "Shows").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_video_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_video_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_video_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_video_root({"path": "shared/Movies"})
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 'the whole group index' — always valid, nothing to resolve."""
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_video_root({"path": ""})
assert issued == [(OP_VIDEO_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_video_root({"path": "shared/Movies"})
assert issued == [(OP_VIDEO_ROOT, "shared/Movies")]
# ── 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.video_root("g1") == "", "absent must mean the whole group index"
await roster.set_video_root("g1", "shared/Movies", set_by="op")
assert await roster.video_root("g1") == "shared/Movies"
finally:
await roster.close()
reopened = Roster(db_path=tmp_path / "roster.db")
await reopened.open()
try:
assert await reopened.video_root("g1") == "shared/Movies"
assert await reopened.video_root("g2") == "", "one group's setting must not answer for another"
finally:
await reopened.close()
|