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
|
"""
Who owns an uploaded file, and who may therefore delete it.
MESHBAY_DESIGN.md §5.4 grants `file_delete` to "the operator, **or any
non-revoked device of the uploading account**". That second half needs the
index entry to record an uploader, and nothing recorded one: the transport
tagged the entry at the end of the upload, walking `ctx["index"]` for the name
it had just written — at a moment when, by construction, no such entry exists.
The file was a `.part` until the rename on the line above (excluded from the
index), and the watchdog that will index it debounces for two seconds and then
hashes. The walk matched nothing, returned silently, and every uploaded file in
every group was owned by nobody: `_do_file_delete` refuses a caller with no
admin authority when the entry records no uploader, so an ordinary member could
not delete what they had just sent.
The suite did not see it because every test of ownership sets `uploader_id` on
an entry by hand — which tests `_verify_uploader_sig`, and nothing about how a
real upload ever comes to have an uploader. These tests cross that seam: a real
`file_upload` through the real handler, a real `DirectoryIndexer` over the same
directory, and the entry it produces.
The second property is the one a restart decides. The index is rebuilt from
disk at every start, so an attribution held in memory is an owner the node
forgets overnight — the file would be deletable by its uploader today and not
tomorrow, which is worse than never having offered it.
"""
import asyncio
from pathlib import Path
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.crypto import generate_gek
from meshbay_node.indexer.cache import IndexCache
from meshbay_node.indexer.indexer import DirectoryIndexer
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
from conftest import one_root, sealed_upload
GROUP = "g" * 32
UPLOADER = "user-1"
UPLOADER_PK = "cGluc2V0LWtleQ==" # the key this node pinned, base64
async def _indexer(shared: Path, cache: IndexCache) -> DirectoryIndexer:
"""A real indexer over `shared`, watching, with a short debounce.
Short but not zero: the debounce is the thing that puts the entry's creation
after the upload's last chunk, which is the whole subject here.
"""
indexer = DirectoryIndexer(
roots=one_root(shared), group_id=GROUP,
sk_node=Ed25519PrivateKey.generate(), gek=generate_gek(),
cache=cache, debounce_secs=0.2,
)
await indexer.initial_scan()
await indexer.start()
return indexer
def _session(shared: Path, indexer: DirectoryIndexer, gek: bytes,
*, user_id: str = UPLOADER) -> WebRTCPeerSession:
session = WebRTCPeerSession.__new__(WebRTCPeerSession)
session._ctx = {
"roots": one_root(shared),
"index": indexer.index,
"sk_node": indexer.sk_node,
"gek": gek,
# The seam under test: the transport hands the record to the indexer,
# which stamps the entry when it finally creates it.
#
# `getattr` rather than the attribute, deliberately: against the source
# this test was written for there is no such seam at all, and a test
# that dies of AttributeError there proves only that a method is
# missing. Tolerating its absence makes the pre-fix run reach the
# assertions and fail on the property — no uploader on the entry —
# which is the thing being guarded.
"record_upload": getattr(indexer, "record_upload", None),
}
session._group_id = GROUP
session._user_id = user_id
session._pinned_pk = UPLOADER_PK
session._pk_user = ""
session._uploads = {}
session._tasks = set()
session.sent = []
session._send = session.sent.append
session._audit = lambda *a, **k: None
return session
async def _upload(session, shared: Path, name: str, data: bytes) -> None:
root_name = shared.name
session._do_file_upload(
sealed_upload(session, filename=name, data=data, dir=root_name))
errors = [m for m in session.sent if m.get("type") == "error"]
assert not errors, errors
# The record is written by a task the handler spawned.
await asyncio.gather(*list(session._tasks))
async def _entry_for(indexer: DirectoryIndexer, name: str, timeout: float = 5.0):
"""The index entry for a file, once the indexer has got to it."""
deadline = asyncio.get_event_loop().time() + timeout
while asyncio.get_event_loop().time() < deadline:
for entry in indexer.index.entries:
if entry.name == name:
return entry
await asyncio.sleep(0.05)
return None
@pytest.mark.asyncio
async def test_an_uploaded_file_records_who_sent_it(tmp_path):
shared = tmp_path / "shared"
shared.mkdir()
gek = generate_gek()
async with IndexCache(tmp_path / "cache.db") as cache:
indexer = await _indexer(shared, cache)
try:
session = _session(shared, indexer, gek)
await _upload(session, shared, "holiday.jpg", b"JPEGDATA" * 64)
entry = await _entry_for(indexer, "holiday.jpg")
assert entry is not None, "the file was never indexed at all"
assert entry.uploader_id == UPLOADER, (
"an uploaded file with no uploader is a file its own sender "
"cannot delete — §5.4 grants that to the uploading account")
assert entry.uploader_pk == UPLOADER_PK
finally:
await indexer.stop()
@pytest.mark.asyncio
async def test_the_uploader_survives_a_restart(tmp_path):
"""A second indexer over the same directory and the same cache.
This is what a node restart is: the index is rebuilt from disk, and every
field not on the disk has to come from somewhere durable.
"""
shared = tmp_path / "shared"
shared.mkdir()
gek = generate_gek()
async with IndexCache(tmp_path / "cache.db") as cache:
first = await _indexer(shared, cache)
try:
session = _session(shared, first, gek)
await _upload(session, shared, "report.txt", b"TEXT" * 64)
assert await _entry_for(first, "report.txt") is not None
finally:
await first.stop()
second = await _indexer(shared, cache)
try:
entry = await _entry_for(second, "report.txt")
assert entry is not None
assert entry.uploader_id == UPLOADER, (
"the attribution did not survive the rebuild, so the uploader "
"could delete their file today and not tomorrow")
finally:
await second.stop()
@pytest.mark.asyncio
async def test_a_different_file_at_the_same_path_inherits_nothing(tmp_path):
"""The record is keyed by path, and a path is not an identity.
A member uploads, the operator deletes it and puts a file of their own
there under the same name. Nothing about that second file was sent by the
member, and crediting them would hand them the right to delete it.
"""
shared = tmp_path / "shared"
shared.mkdir()
gek = generate_gek()
async with IndexCache(tmp_path / "cache.db") as cache:
indexer = await _indexer(shared, cache)
try:
session = _session(shared, indexer, gek)
await _upload(session, shared, "notes.txt", b"SENT" * 64)
assert await _entry_for(indexer, "notes.txt") is not None
(shared / "notes.txt").unlink()
(shared / "notes.txt").write_bytes(b"THE OPERATOR'S OWN FILE")
await asyncio.sleep(0.6)
entry = await _entry_for(indexer, "notes.txt")
assert entry is not None
assert not entry.uploader_id, (
"a file the operator put there is not the member's to delete")
finally:
await indexer.stop()
@pytest.mark.asyncio
async def test_a_file_nobody_uploaded_has_no_uploader(tmp_path):
"""The operator's own library is not attributed to anyone."""
shared = tmp_path / "shared"
shared.mkdir()
(shared / "already-here.txt").write_bytes(b"ON DISK BEFORE ANY MEMBER")
async with IndexCache(tmp_path / "cache.db") as cache:
indexer = await _indexer(shared, cache)
try:
entry = await _entry_for(indexer, "already-here.txt")
assert entry is not None
assert not entry.uploader_id and not entry.uploader_pk
finally:
await indexer.stop()
|