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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
|
"""
One implementation, several front doors.
The point of `meshbay_node.ops` is not tidiness. C1 and C6 were both "a second
path into the node with its own weaker handshake", and two implementations of
`revoke` with two authorization checks is that shape one size down. So the tests
that matter here are the ones that would fail if a second implementation
appeared: the adapters must be thin, and the operations must not decide who may
call them.
"""
import inspect
from pathlib import Path
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from conftest import one_root
from meshbay_node import ops
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.transport.quic_server import Denylist
def _state(tmp_path: Path) -> dict:
shared = tmp_path / "shared"
shared.mkdir(exist_ok=True)
index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
return {
"groups_ctx": {"g" * 32: {"index": index, "roots": one_root(shared),
"gek": None}},
"denylist": Denylist(path=tmp_path / "denylist.json"),
"indexes": {"g" * 32: index},
}
# ── The adapters stay thin ───────────────────────────────────────────────────
def test_operations_take_state_and_nothing_web_shaped():
"""
An operation that knew about HTTP could not be called from MNP without a
second copy. Every public operation therefore takes `state` first and
returns plain data.
"""
public = [(n, f) for n, f in vars(ops).items()
if inspect.iscoroutinefunction(f) and not n.startswith("_")]
assert public, "no operations found — did the module move?"
for name, fn in public:
params = list(inspect.signature(fn).parameters)
assert params and params[0] == "state", (
f"{name} does not take state first — an adapter would have to "
f"assemble something for it, which is where a second implementation "
f"begins")
def test_the_http_adapter_adds_no_logic():
"""
Each loopback handler should be a call into `ops` and nothing else. A
handler that grew a check of its own would be a rule the MNP path does not
have.
"""
import meshbay_node.ui.app as ui
source = inspect.getsource(ui)
# Every endpoint that performs an operation routes through _op(...).
for endpoint in ("operator_pair", "create_invite", "revoke_member",
"unpin_member", "init_gek", "attach_group", "delete_file",
"add_root", "remove_root", "set_member_upload",
"reload_config"):
start = source.index(f"async def {endpoint}(")
body = source[start:start + 700]
assert "_op(" in body.split("\n\n")[0] + body, (
f"{endpoint} does not go through the shared adapter")
assert "roster.set_status" not in body and "generate_gek" not in body, (
f"{endpoint} performs the operation itself instead of calling ops")
def test_op_errors_carry_a_status_without_importing_http():
source = inspect.getsource(ops)
for forbidden in ("JSONResponse", "fastapi", "starlette", "HTTPException"):
assert forbidden not in source, (
f"ops imports {forbidden} — it must not know which adapter called it")
# ── Denylist (14.10) ─────────────────────────────────────────────────────────
async def test_denylist_reports_what_it_refuses(tmp_path):
state = _state(tmp_path)
state["denylist"].deny_user("alice")
state["denylist"].deny_group("g" * 32)
out = await ops.read_denylist(state)
assert out["users"] == ["alice"]
assert out["groups"] == ["g" * 32]
assert out["count"] == 2
async def test_clearing_one_subject_leaves_the_rest(tmp_path):
state = _state(tmp_path)
state["denylist"].deny_user("alice")
state["denylist"].deny_user("bob")
out = await ops.clear_denylist(state, subject="alice")
assert out["removed"] == 1
assert (await ops.read_denylist(state))["users"] == ["bob"]
async def test_clearing_everything_says_how_much(tmp_path):
"""The count is the point: it tells the operator whether they undid one
revocation or all of them."""
state = _state(tmp_path)
for name in ("a", "b", "c"):
state["denylist"].deny_user(name)
out = await ops.clear_denylist(state)
assert out["removed"] == 3
assert (await ops.read_denylist(state))["count"] == 0
async def test_denylist_survives_a_restart(tmp_path):
"""Finding H4: revocations used to live only in memory, so a restart
silently un-revoked everyone."""
state = _state(tmp_path)
state["denylist"].deny_user("alice")
reopened = Denylist(path=tmp_path / "denylist.json")
assert reopened.is_denied("alice", jti="", group_id="")
# ── File deletion (14.11) ────────────────────────────────────────────────────
async def test_deleting_a_file_removes_it_from_disk_and_index(tmp_path):
state = _state(tmp_path)
ctx = state["groups_ctx"]["g" * 32]
target = ctx["roots"].roots[0].path / "gone.txt"
target.write_text("x")
from meshbay_common.protocol import IndexEntry
ctx["index"].add_entry(IndexEntry(id="a" * 64, name="gone.txt", path="shared",
size=1, type="other", added_at=0))
out = await ops.delete_file(state, "g" * 32, "a" * 64)
assert out["status"] == "deleted"
assert not target.exists()
assert ctx["index"].get_entry("a" * 64) is None
async def test_deleting_from_an_unavailable_root_is_refused(tmp_path):
"""
A frozen root's files are still listed. Deleting one would either fail
obscurely or — worse, once the drive returns — leave the index and the disk
disagreeing.
"""
state = _state(tmp_path)
ctx = state["groups_ctx"]["g" * 32]
from meshbay_common.protocol import IndexEntry
ctx["index"].add_entry(IndexEntry(id="a" * 64, name="frozen.txt", path="shared",
size=1, type="other", added_at=0))
ctx["roots"].roots[0].available = False
with pytest.raises(ops.OpError, match="frozen, not gone"):
await ops.delete_file(state, "g" * 32, "a" * 64)
assert ctx["index"].get_entry("a" * 64) is not None
async def test_deleting_an_unknown_file_says_so(tmp_path):
state = _state(tmp_path)
with pytest.raises(ops.OpError, match="No such file"):
await ops.delete_file(state, "g" * 32, "f" * 64)
async def test_an_unhosted_group_offers_what_it_does_host(tmp_path):
"""A bare "no such group" leaves an operator guessing at a UUID."""
state = _state(tmp_path)
with pytest.raises(ops.OpError) as exc:
await ops.delete_file(state, "z" * 32, "a" * 64)
assert exc.value.status == 404
assert exc.value.extra.get("available")
# ── Upload policy (set_member_upload) ───────────────────────────────────────
async def test_set_member_upload_toggles_and_persists(tmp_path):
from meshbay_node.roster import Roster
state = _state(tmp_path)
roster = Roster(db_path=tmp_path / "roster.db")
await roster.open()
state["roster"] = roster
state["node_user_id"] = "operator"
out = await ops.set_member_upload(state, "g" * 32, True)
assert out["allowed"] is True
assert state["groups_ctx"]["g" * 32]["member_upload"] is True
out2 = await ops.set_member_upload(state, "g" * 32, False)
assert out2["allowed"] is False
assert state["groups_ctx"]["g" * 32]["member_upload"] is False
# ── Reload ──────────────────────────────────────────────────────────────────
async def test_reload_config_calls_reload_fn(tmp_path):
state = _state(tmp_path)
called = []
async def fake_reload():
called.append(True)
state["reload_fn"] = fake_reload
out = await ops.reload_config(state)
assert out["status"] == "reloaded"
assert called
async def test_reload_config_without_fn_is_refused(tmp_path):
state = _state(tmp_path)
with pytest.raises(ops.OpError, match="Reload not available"):
await ops.reload_config(state)
|