diff options
Diffstat (limited to 'packages/meshbay-node/tests')
8 files changed, 179 insertions, 78 deletions
diff --git a/packages/meshbay-node/tests/test_disk_io_off_loop.py b/packages/meshbay-node/tests/test_disk_io_off_loop.py index 206b77b..2811836 100644 --- a/packages/meshbay-node/tests/test_disk_io_off_loop.py +++ b/packages/meshbay-node/tests/test_disk_io_off_loop.py @@ -29,12 +29,15 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common.crypto import generate_gek from meshbay_common.protocol import MNP from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes +from meshbay_node.indexer.group_index import GroupIndex from meshbay_node.indexer.indexer import DirectoryIndexer -from meshbay_node.roots import RootSet +from meshbay_node.roots import Root, RootSet from meshbay_node.transfers import LeaselessReads from meshbay_node.transport import webrtc_server from meshbay_node.transport.webrtc_server import WebRTCPeerSession +from conftest import one_root, sealed_upload + GROUP = "g" * 32 # Long enough that a blocked loop is unmistakable, short enough to keep the @@ -222,3 +225,96 @@ def test_no_handler_resolves_a_path_on_the_loop(): assert not stray, ( f"{len(stray)} call(s) to entry_abs_path outside _locate: " f"resolve a path through `off_disk(roots, _locate, ...)` instead") + + +async def test_the_availability_poll_does_not_stop_the_loop(tmp_path, monkeypatch): + """ + The poll is the one that runs whether anybody asked for anything. + + `refresh_availability` stats every root, and reconcile calls it on a timer. + On the loop, a node with a sleeping disk stalls once per tick for as long as + the spin-up takes — and the stat is also what keeps waking the disk, so the + node pays for a library nobody is reading. + """ + root = tmp_path / "films" + root.mkdir() + (root / "clip.bin").write_bytes(CONTENT) + roots = RootSet.build([{"path": str(root), "name": "films"}]) + monkeypatch.setattr(Root, "is_live", _slow(Root.is_live)) + + idx = DirectoryIndexer(roots=roots, group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=None) + try: + with _Ticker() as ticker: + await idx.initial_scan() + finally: + await idx.stop() + roots.close_io() + + assert ticker.ticks > SLOW_S / TICK_S / 2, ( + f"the loop was blocked: {ticker.ticks} wake-ups during a {SLOW_S}s poll") + + +def _upload_session(tmp_path): + """One connection into a group with one writable root, as a node has.""" + shared = tmp_path / "shared" + shared.mkdir(exist_ok=True) + ctx = {"roots": one_root(shared), + "index": GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()), + "gek": generate_gek()} + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = ctx + session._group_id = GROUP + session._user_id = "user-1" + session._pk_user = "" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session, shared + + +async def test_a_slow_upload_write_does_not_stop_the_loop(tmp_path, monkeypatch): + session, shared = _upload_session(tmp_path) + monkeypatch.setattr(webrtc_server, "_append_chunk", + _slow(webrtc_server._append_chunk)) + + with _Ticker() as ticker: + await session._do_file_upload(sealed_upload( + session, filename="clip.bin", data=CONTENT)) + + assert ticker.ticks > SLOW_S / TICK_S / 2, ( + f"the loop was blocked: {ticker.ticks} wake-ups during a {SLOW_S}s write") + assert (shared / "clip.bin").read_bytes() == CONTENT + + +async def test_chunks_of_one_upload_keep_their_order_under_a_slow_disk(tmp_path, + monkeypatch): + """ + The rule that used to hold for free. + + `chunk_index != state.next_index` is refused, and nothing could come between + that check and the `advance` answering it while the handler was synchronous. + Awaiting the write opens the gap: chunk 1 arriving while chunk 0 is still in + the disk thread reads a position that has not moved yet and is refused as + out of order — an upload that fails on a slow disk and nowhere else. The + lock closes it, and the disk made slow here is what makes the gap wide + enough to fall into. + """ + session, shared = _upload_session(tmp_path) + pieces = [b"first-", b"second-", b"third-", b"fourth"] + + monkeypatch.setattr(webrtc_server, "_append_chunk", + _slow(webrtc_server._append_chunk)) + + # Fired together and in order, which is what the dispatcher does: it creates + # one task per message as it arrives. + await asyncio.gather(*[ + session._do_file_upload(sealed_upload( + session, filename="clip.bin", data=piece, + chunk_index=i, total_chunks=len(pieces))) + for i, piece in enumerate(pieces) + ]) + + refusals = [m for m in session.sent if m.get("type") == "error"] + assert not refusals, f"a chunk was refused: {refusals}" + assert (shared / "clip.bin").read_bytes() == b"".join(pieces) diff --git a/packages/meshbay-node/tests/test_lease_enforcement.py b/packages/meshbay-node/tests/test_lease_enforcement.py index c516b5a..2b1014b 100644 --- a/packages/meshbay-node/tests/test_lease_enforcement.py +++ b/packages/meshbay-node/tests/test_lease_enforcement.py @@ -236,7 +236,7 @@ async def test_a_queued_upload_is_refused_before_anything_is_written(group, msg = sealed_upload(second, filename="sent.bin", data=b"DATA", dir=Path(group["roots"].roots[0].path).name) msg["tr"] = "u2" - second._do_file_upload(msg) + await second._do_file_upload(msg) assert second.errors()[-1]["code"] == "lease_not_granted" assert not list(Path(group["roots"].roots[0].path).glob("sent*")), ( diff --git a/packages/meshbay-node/tests/test_ops.py b/packages/meshbay-node/tests/test_ops.py index c118b5a..083f31e 100644 --- a/packages/meshbay-node/tests/test_ops.py +++ b/packages/meshbay-node/tests/test_ops.py @@ -44,9 +44,14 @@ def test_operations_take_state_and_nothing_web_shaped(): second copy. Every public operation therefore takes `state` first and returns plain data. """ + # Defined here, not merely visible here: an async helper imported into the + # module (`off_disk`, say) is not an operation, and reporting it as one says + # nothing about a second implementation appearing. Every real operation is + # written in this module, so nothing is lost by asking. 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?" + if inspect.iscoroutinefunction(f) and not n.startswith("_") + and getattr(f, "__module__", None) == ops.__name__] + assert len(public) > 30, "operations have gone missing — did the module move?" for name, fn in public: params = list(inspect.signature(fn).parameters) assert params and params[0] == "state", ( diff --git a/packages/meshbay-node/tests/test_partial_uploads.py b/packages/meshbay-node/tests/test_partial_uploads.py index f5b6602..0270a69 100644 --- a/packages/meshbay-node/tests/test_partial_uploads.py +++ b/packages/meshbay-node/tests/test_partial_uploads.py @@ -305,7 +305,7 @@ def _errors(session): return [m for m in session.sent if m.get("type") == "error"] -def test_an_upload_survives_the_connection_that_started_it(tmp_path): +async def test_an_upload_survives_the_connection_that_started_it(tmp_path): """The defect this stage exists to fix. The state used to live on the session, so the second connection saw no @@ -315,14 +315,14 @@ def test_an_upload_survives_the_connection_that_started_it(tmp_path): """ ctx = _group_ctx(tmp_path) first = _peer(ctx) - first._do_file_upload(sealed_upload(first, filename="film.mkv", + await first._do_file_upload(sealed_upload(first, filename="film.mkv", data=b"first-half", chunk_index=0, total_chunks=2)) assert _errors(first) == [] # The link drops; the client comes back on a new connection and carries on. second = _peer(ctx) - second._do_file_upload(sealed_upload(second, filename="film.mkv", + await second._do_file_upload(sealed_upload(second, filename="film.mkv", data=b"second-half", chunk_index=1, total_chunks=2)) assert _errors(second) == [], _errors(second) @@ -331,31 +331,31 @@ def test_an_upload_survives_the_connection_that_started_it(tmp_path): assert (root.path / "film.mkv").read_bytes() == b"first-halfsecond-half" -def test_another_member_cannot_continue_somebody_elses_upload(tmp_path): +async def test_another_member_cannot_continue_somebody_elses_upload(tmp_path): """The key includes the member for a reason. Without it, a second person sending the same name into the same folder would append their chunks to the first person's file — which a shared folder makes an ordinary accident, not only an attack.""" ctx = _group_ctx(tmp_path) alice = _peer(ctx, "alice") - alice._do_file_upload(sealed_upload(alice, filename="IMG_1234.jpg", + await alice._do_file_upload(sealed_upload(alice, filename="IMG_1234.jpg", data=b"hers", chunk_index=0, total_chunks=2)) assert _errors(alice) == [] bob = _peer(ctx, "bob") - bob._do_file_upload(sealed_upload(bob, filename="IMG_1234.jpg", + await bob._do_file_upload(sealed_upload(bob, filename="IMG_1234.jpg", data=b"his", chunk_index=1, total_chunks=2)) assert [m.get("code") for m in _errors(bob)] == ["not_started"] -def test_an_upload_in_flight_is_known_to_the_reaper(tmp_path): +async def test_an_upload_in_flight_is_known_to_the_reaper(tmp_path): """The two halves of this stage meeting: the state the node keeps is what stops the janitor deleting a file somebody is still sending.""" ctx = _group_ctx(tmp_path) peer = _peer(ctx) - peer._do_file_upload(sealed_upload(peer, filename="film.mkv", + await peer._do_file_upload(sealed_upload(peer, filename="film.mkv", data=b"half", chunk_index=0, total_chunks=2)) live = ctx["partial_uploads"].live_paths() @@ -379,38 +379,38 @@ def _probe(session, filename: str) -> dict: chunk_index=UPLOAD_PROBE_INDEX, total_chunks=1) -def test_a_probe_for_an_unknown_file_says_start_at_the_beginning(tmp_path): +async def test_a_probe_for_an_unknown_file_says_start_at_the_beginning(tmp_path): ctx = _group_ctx(tmp_path) peer = _peer(ctx) - peer._do_file_upload(_probe(peer, "film.mkv")) + await peer._do_file_upload(_probe(peer, "film.mkv")) assert _errors(peer) == [] assert _acks(peer, ctx)[0]["resume_from"] == 0 -def test_a_probe_reports_what_the_node_already_holds(tmp_path): +async def test_a_probe_reports_what_the_node_already_holds(tmp_path): """The point of the whole stage: the client learns it has 2 chunks there and sends the third, instead of sending a film again.""" ctx = _group_ctx(tmp_path) first = _peer(ctx) for i in range(2): - first._do_file_upload(sealed_upload(first, filename="film.mkv", + await first._do_file_upload(sealed_upload(first, filename="film.mkv", data=b"xxxx", chunk_index=i, total_chunks=5)) assert _errors(first) == [] reconnected = _peer(ctx) - reconnected._do_file_upload(_probe(reconnected, "film.mkv")) + await reconnected._do_file_upload(_probe(reconnected, "film.mkv")) ack = _acks(reconnected, ctx)[0] assert ack["resume_from"] == 2 assert ack["stored_as"] == "film.mkv" -def test_a_probe_writes_nothing_and_reserves_nothing(tmp_path): +async def test_a_probe_writes_nothing_and_reserves_nothing(tmp_path): """It has to be free of consequence: a client that asks and goes away must leave no file, no state and no name taken.""" ctx = _group_ctx(tmp_path) peer = _peer(ctx) - peer._do_file_upload(_probe(peer, "film.mkv")) + await peer._do_file_upload(_probe(peer, "film.mkv")) root = ctx["roots"].roots[0] assert list(root.path.iterdir()) == [] assert len(ctx.get("partial_uploads") or []) == 0 @@ -418,36 +418,36 @@ def test_a_probe_writes_nothing_and_reserves_nothing(tmp_path): assert _acks(peer, ctx)[0]["stored_as"] == "" -def test_a_probe_answers_only_about_the_member_who_asks(tmp_path): +async def test_a_probe_answers_only_about_the_member_who_asks(tmp_path): """Same keying as the upload itself. Otherwise one member could measure another's progress on a file they never sent — and worse, resume it.""" ctx = _group_ctx(tmp_path) alice = _peer(ctx, "alice") - alice._do_file_upload(sealed_upload(alice, filename="film.mkv", + await alice._do_file_upload(sealed_upload(alice, filename="film.mkv", data=b"xxxx", chunk_index=0, total_chunks=5)) bob = _peer(ctx, "bob") - bob._do_file_upload(_probe(bob, "film.mkv")) + await bob._do_file_upload(_probe(bob, "film.mkv")) assert _acks(bob, ctx)[0]["resume_from"] == 0 -def test_an_ordinary_ack_carries_no_resume_field(tmp_path): +async def test_an_ordinary_ack_carries_no_resume_field(tmp_path): """So a client can tell a probe's answer from a chunk's without looking at the index it echoed.""" ctx = _group_ctx(tmp_path) peer = _peer(ctx) - peer._do_file_upload(sealed_upload(peer, filename="a.bin", data=b"x", + await peer._do_file_upload(sealed_upload(peer, filename="a.bin", data=b"x", chunk_index=0, total_chunks=2)) assert "resume_from" not in _acks(peer, ctx)[0] -def test_a_probe_is_refused_where_an_upload_would_be(tmp_path): +async def test_a_probe_is_refused_where_an_upload_would_be(tmp_path): """Every check the write path makes has already run when the probe is answered, so it cannot be used to ask questions about somewhere the caller may not write.""" ctx = _group_ctx(tmp_path) peer = _peer(ctx) - peer._do_file_upload(sealed_upload(peer, filename="../escape", + await peer._do_file_upload(sealed_upload(peer, filename="../escape", data=b"", chunk_index=UPLOAD_PROBE_INDEX, total_chunks=1)) assert [m.get("code") for m in _errors(peer)] == ["invalid_filename"] @@ -456,7 +456,7 @@ def test_a_probe_is_refused_where_an_upload_would_be(tmp_path): # ── the slot an upload holds ──────────────────────────────────────────────── -def test_an_upload_chunk_says_its_slot_is_in_use(tmp_path): +async def test_an_upload_chunk_says_its_slot_is_in_use(tmp_path): """A grant nobody takes up is reclaimed after thirty seconds and abandoned on the third miss. Uploads are not gated by the lease, so the file arrived anyway — but the widget follows the lease, and a 3.5 GB upload therefore @@ -482,7 +482,7 @@ def test_an_upload_chunk_says_its_slot_is_in_use(tmp_path): msg = sealed_upload(peer, filename="film.mkv", data=b"xxxx", chunk_index=0, total_chunks=2) msg["tr"] = "up-1" - peer._do_file_upload(msg) + await peer._do_file_upload(msg) assert _errors(peer) == [] assert slots.leases["up-1"].used is True, ( diff --git a/packages/meshbay-node/tests/test_root_writable_policy.py b/packages/meshbay-node/tests/test_root_writable_policy.py index 730e636..23b55cb 100644 --- a/packages/meshbay-node/tests/test_root_writable_policy.py +++ b/packages/meshbay-node/tests/test_root_writable_policy.py @@ -64,8 +64,8 @@ def _session(tmp_path: Path, user_id: str, *, return session -def _upload(session, filename="clip.mp4", body=b"bytes"): - session._do_file_upload(sealed_upload( +async def _upload(session, filename="clip.mp4", body=b"bytes"): + await session._do_file_upload(sealed_upload( session, filename=filename, data=body, dir="shared")) @@ -79,7 +79,7 @@ def _uploads_dir(session) -> Path: async def test_a_member_cannot_upload_to_a_read_only_root(tmp_path): session = _session(tmp_path, "member-1", writable=False) - _upload(session) + await _upload(session) refusal = [m for m in session.sent if m.get("type") == "error"] assert refusal and refusal[0].get("code") == "root_read_only" @@ -88,7 +88,7 @@ async def test_a_member_cannot_upload_to_a_read_only_root(tmp_path): async def test_members_upload_normally_to_a_writable_root(tmp_path): session = _session(tmp_path, "member-1", writable=True) - _upload(session) + await _upload(session) assert not [m for m in session.sent if m.get("type") == "error"] assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes" @@ -103,7 +103,7 @@ async def test_read_only_binds_the_operator_too(tmp_path): session = _session(tmp_path, "the-operator", writable=False, operator="the-operator") session._is_node_admin = lambda: True - _upload(session) + await _upload(session) refusal = [m for m in session.sent if m.get("type") == "error"] assert refusal and refusal[0].get("code") == "root_read_only" @@ -249,6 +249,6 @@ async def test_no_message_can_reopen_uploads_for_a_whole_group(tmp_path): "the node answered an instruction it does not implement") # And the door is still shut. - _upload(session) + await _upload(session) refusal = [m for m in session.sent if m.get("type") == "error"] assert refusal and refusal[0].get("code") == "root_read_only" diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py index fa55ff6..e203811 100644 --- a/packages/meshbay-node/tests/test_security_regressions.py +++ b/packages/meshbay-node/tests/test_security_regressions.py @@ -174,7 +174,7 @@ def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession: return session -def test_upload_cannot_overwrite_another_members_file(tmp_path): +async def test_upload_cannot_overwrite_another_members_file(tmp_path): """ C5a: uploads used to land in the shared root under a client-chosen name and overwrite whatever was there. That let any member destroy the operator's @@ -192,7 +192,7 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path): original.write_bytes(b"operator's original content") attacker = _session(tmp_path, "attacker-user") - attacker._do_file_upload(sealed_upload( + await attacker._do_file_upload(sealed_upload( attacker, filename="important.mp4", data=b"attacker content")) assert original.read_bytes() == b"operator's original content", ( @@ -200,18 +200,18 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path): assert (uploads / "important (2).mp4").read_bytes() == b"attacker content" -def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path): +async def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path): """C5a: even the original uploader does not get to overwrite.""" session = _session(tmp_path, "user-1") - def _send_it(): + async def _send_it(): # Sealed afresh each time: a nonce is drawn per message, so re-sending # the same dict would be a replay rather than a second upload. - session._do_file_upload(sealed_upload( + await session._do_file_upload(sealed_upload( session, filename="movie.mp4", data=b"first")) - _send_it() + await _send_it() session.sent.clear() - _send_it() + await _send_it() uploads = _uploads_dir(session) assert (uploads / "movie.mp4").read_bytes() == b"first", ( "the first upload was replaced") @@ -236,7 +236,7 @@ def test_dir_create_cannot_escape_the_shared_root(tmp_path, bad): assert set(tmp_path.rglob("*")) == before, f"created something via {bad!r}" -def test_the_client_names_a_folder_and_never_a_filesystem_path(tmp_path): +async def test_the_client_names_a_folder_and_never_a_filesystem_path(tmp_path): """ The destination is now the folder the sender is looking at, which means the client does choose it — and the whole of what keeps that safe is that the @@ -255,7 +255,7 @@ def test_the_client_names_a_folder_and_never_a_filesystem_path(tmp_path): for bad in ("../../etc", "/etc", "shared/../..", "shared/../../etc", "nope", "shared/missing"): session.sent.clear() - session._do_file_upload(sealed_upload( + await session._do_file_upload(sealed_upload( session, filename="note.txt", data=b"x", dir=bad)) refusal = [m for m in session.sent if m.get("type") == "error"] assert refusal, f"{bad!r} was accepted" @@ -264,7 +264,7 @@ def test_the_client_names_a_folder_and_never_a_filesystem_path(tmp_path): assert set(tmp_path.rglob("*")) == before, "a refused upload still wrote" -def test_an_upload_lands_in_the_folder_it_names(tmp_path): +async def test_an_upload_lands_in_the_folder_it_names(tmp_path): """ And in that folder itself — the `uploads/` subdirectory the node used to create is gone. Somebody dropping a file into the folder they are looking @@ -274,7 +274,7 @@ def test_an_upload_lands_in_the_folder_it_names(tmp_path): root = session._ctx["roots"].roots[0] (root.path / "Albums").mkdir() - session._do_file_upload(sealed_upload( + await session._do_file_upload(sealed_upload( session, filename="note.txt", data=b"x", dir=f"{root.name}/Albums")) assert (root.path / "Albums" / "note.txt").read_bytes() == b"x" @@ -283,7 +283,7 @@ def test_an_upload_lands_in_the_folder_it_names(tmp_path): assert not (root.path / "uploads").exists() -def test_an_upload_goes_to_the_root_it_names(tmp_path): +async def test_an_upload_goes_to_the_root_it_names(tmp_path): """ With two writable roots there is no defensible default, and the client is the only party that knows which directory the person is looking at. The @@ -301,14 +301,14 @@ def test_an_upload_goes_to_the_root_it_names(tmp_path): {"path": str(incoming), "writable": True}, ]) - session._do_file_upload(sealed_upload( + await session._do_file_upload(sealed_upload( session, filename="note.txt", data=b"x", dir="Incoming")) assert (incoming / "note.txt").read_bytes() == b"x" assert not (media / "note.txt").exists(), "it went to the first root instead" -def test_a_read_only_root_refuses_an_upload(tmp_path): +async def test_a_read_only_root_refuses_an_upload(tmp_path): """ RO is the mechanism now, not a hidden button. It binds the operator too: "read-only for everyone" is what makes a published library one, and an @@ -321,7 +321,7 @@ def test_a_read_only_root_refuses_an_upload(tmp_path): session._ctx["roots"] = RootSet.build([{"path": str(published)}]) session._is_node_admin = lambda: True - session._do_file_upload(sealed_upload( + await session._do_file_upload(sealed_upload( session, filename="note.txt", data=b"x", dir="Published")) refusal = [m for m in session.sent if m.get("type") == "error"] @@ -329,7 +329,7 @@ def test_a_read_only_root_refuses_an_upload(tmp_path): assert not (published / "note.txt").exists() -def test_a_fully_read_only_group_refuses_an_unaddressed_upload(tmp_path): +async def test_a_fully_read_only_group_refuses_an_unaddressed_upload(tmp_path): """ An MNP 1.0 client names no root, so the node falls back to the first writable one. There isn't one here, and the fallback must refuse rather @@ -340,7 +340,7 @@ def test_a_fully_read_only_group_refuses_an_unaddressed_upload(tmp_path): session = _session(tmp_path, "user-1") session._ctx["roots"] = RootSet.build([{"path": str(published)}]) - session._do_file_upload(sealed_upload( + await session._do_file_upload(sealed_upload( session, filename="note.txt", data=b"x")) refusal = [m for m in session.sent if m.get("type") == "error"] @@ -348,7 +348,7 @@ def test_a_fully_read_only_group_refuses_an_unaddressed_upload(tmp_path): assert not (published / "note.txt").exists() -def test_an_ejected_root_refuses_an_upload(tmp_path): +async def test_an_ejected_root_refuses_an_upload(tmp_path): """ Writing to a drive somebody has their hand on is the thing eject exists to stop. `writable` is still true — that is configuration — so availability @@ -363,7 +363,7 @@ def test_an_ejected_root_refuses_an_upload(tmp_path): roots.roots[0].available = False session._ctx["roots"] = roots - session._do_file_upload(sealed_upload( + await session._do_file_upload(sealed_upload( session, filename="note.txt", data=b"x", dir="USB")) refusal = [m for m in session.sent if m.get("type") == "error"] @@ -371,19 +371,19 @@ def test_an_ejected_root_refuses_an_upload(tmp_path): assert not (usb / "note.txt").exists() -def test_two_members_can_send_the_same_filename(tmp_path): +async def test_two_members_can_send_the_same_filename(tmp_path): """ One shared uploads/ means collisions are ordinary — every camera produces IMG_1234.jpg. The second gets a free name; neither replaces the other. """ first = _session(tmp_path, "user-1") - first._do_file_upload(sealed_upload( + await first._do_file_upload(sealed_upload( first, filename="IMG_1234.jpg", data=b"first")) second = _session(tmp_path, "user-2") # Same group, so the same key: `_session` builds one per call, and two # members of one group do not have two. second._ctx["gek"] = first._ctx["gek"] - second._do_file_upload(sealed_upload( + await second._do_file_upload(sealed_upload( second, filename="IMG_1234.jpg", data=b"second")) uploads = _uploads_dir(first) diff --git a/packages/meshbay-node/tests/test_upload_attribution.py b/packages/meshbay-node/tests/test_upload_attribution.py index dc8d989..63762e9 100644 --- a/packages/meshbay-node/tests/test_upload_attribution.py +++ b/packages/meshbay-node/tests/test_upload_attribution.py @@ -91,7 +91,7 @@ def _session(shared: Path, indexer: DirectoryIndexer, gek: bytes, async def _upload(session, shared: Path, name: str, data: bytes) -> None: root_name = shared.name - session._do_file_upload( + await 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 diff --git a/packages/meshbay-node/tests/test_upload_sealed.py b/packages/meshbay-node/tests/test_upload_sealed.py index 7c1be96..7f78ba7 100644 --- a/packages/meshbay-node/tests/test_upload_sealed.py +++ b/packages/meshbay-node/tests/test_upload_sealed.py @@ -80,14 +80,14 @@ def test_the_wire_message_carries_no_filename_and_no_plaintext(tmp_path): assert b"JPEGDATA" not in blob, "the file content is on the wire in clear" -def test_the_ack_carries_no_stored_name(tmp_path): +async def test_the_ack_carries_no_stored_name(tmp_path): """ `stored_as` is the name the node settled on — it finds a free one rather than replacing anything — and naming it in clear would hand back exactly what the request took the trouble to hide. """ session = _session(tmp_path) - session._do_file_upload(sealed_upload( + await session._do_file_upload(sealed_upload( session, filename="holiday.jpg", data=b"x", dir=_root(session).name)) ack = [m for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK][-1] @@ -98,7 +98,7 @@ def test_the_ack_carries_no_stored_name(tmp_path): "dir": _root(session).name} -def test_the_ack_names_the_upload_so_one_refusal_fails_one_upload(tmp_path): +async def test_the_ack_names_the_upload_so_one_refusal_fails_one_upload(tmp_path): """ `filename` used to be the correlation key on both sides. It cannot be one any more, and `upload_id` replaces it — a client-chosen label, opaque to @@ -107,10 +107,10 @@ def test_the_ack_names_the_upload_so_one_refusal_fails_one_upload(tmp_path): for one file used to fail every upload in flight. """ session = _session(tmp_path) - session._do_file_upload(sealed_upload( + await session._do_file_upload(sealed_upload( session, filename="a.txt", data=b"x", dir=_root(session).name, upload_id="upload-A")) - session._do_file_upload(sealed_upload( + await session._do_file_upload(sealed_upload( session, filename="../evil", data=b"x", dir=_root(session).name, upload_id="upload-B")) @@ -121,14 +121,14 @@ def test_the_ack_names_the_upload_so_one_refusal_fails_one_upload(tmp_path): # ── What is refused ────────────────────────────────────────────────────────── -def test_a_plaintext_upload_is_refused(tmp_path): +async def test_a_plaintext_upload_is_refused(tmp_path): """ The MNP 1.x shape, which is what an un-updated client sends. Refused with a code and a message saying which side is old — never accepted "just this once", because a path that still takes plaintext is not a sealed path. """ session = _session(tmp_path) - session._do_file_upload({ + await session._do_file_upload({ "filename": "note.txt", "dir": _root(session).name, "chunk_index": 0, "total_chunks": 1, "data": b"x", }) @@ -137,7 +137,7 @@ def test_a_plaintext_upload_is_refused(tmp_path): assert not _wrote_anything(tmp_path) -def test_a_tampered_chunk_is_refused(tmp_path): +async def test_a_tampered_chunk_is_refused(tmp_path): """ AES-GCM's tag, asserted where it matters: a flipped bit in the ciphertext must stop the upload, not produce a corrupt file with a plausible name. @@ -146,13 +146,13 @@ def test_a_tampered_chunk_is_refused(tmp_path): msg = sealed_upload(session, filename="note.txt", data=b"x" * 64, dir=_root(session).name) msg["ct"] = bytes([msg["ct"][0] ^ 0x01]) + msg["ct"][1:] - session._do_file_upload(msg) + await session._do_file_upload(msg) assert _errors(session)[0]["code"] == "upload_not_sealed" assert not _wrote_anything(tmp_path) -def test_an_upload_sealed_for_another_group_is_refused(tmp_path): +async def test_an_upload_sealed_for_another_group_is_refused(tmp_path): """ The group is the AAD, so a node hosting two groups cannot have a chunk moved between them — and a member of one cannot write into the other by @@ -163,20 +163,20 @@ def test_an_upload_sealed_for_another_group_is_refused(tmp_path): session._ctx["gek"], "some-other-group", upload_id="u1", chunk_index=0, total_chunks=1, filename="note.txt", data=b"x", dir=_root(session).name) - session._do_file_upload(msg) + await session._do_file_upload(msg) assert _errors(session)[0]["code"] == "upload_not_sealed" assert not _wrote_anything(tmp_path) -def test_an_upload_under_another_key_is_refused(tmp_path): +async def test_an_upload_under_another_key_is_refused(tmp_path): """A peer past the handshake with the wrong GEK still writes nothing.""" session = _session(tmp_path) msg = file_upload_wire( generate_gek(), GROUP, upload_id="u1", chunk_index=0, total_chunks=1, filename="note.txt", data=b"x", dir=_root(session).name) - session._do_file_upload(msg) + await session._do_file_upload(msg) assert _errors(session)[0]["code"] == "upload_not_sealed" assert not _wrote_anything(tmp_path) @@ -199,7 +199,7 @@ def test_an_ack_replayed_as_a_request_does_not_open(tmp_path): file_upload_payload(session._ctx["gek"], GROUP, ack) -def test_a_group_with_no_key_refuses_rather_than_falling_back(tmp_path): +async def test_a_group_with_no_key_refuses_rather_than_falling_back(tmp_path): """ A node whose group has no GEK yet cannot open anything. It must say so, not read the message as though it were the old plaintext shape. @@ -208,7 +208,7 @@ def test_a_group_with_no_key_refuses_rather_than_falling_back(tmp_path): msg = sealed_upload(session, filename="note.txt", data=b"x", dir=_root(session).name) session._ctx["gek"] = b"" - session._do_file_upload(msg) + await session._do_file_upload(msg) assert _errors(session)[0]["code"] == "no_group_key" assert not _wrote_anything(tmp_path) @@ -216,7 +216,7 @@ def test_a_group_with_no_key_refuses_rather_than_falling_back(tmp_path): # ── What must still work ───────────────────────────────────────────────────── -def test_a_multi_chunk_upload_reassembles(tmp_path): +async def test_a_multi_chunk_upload_reassembles(tmp_path): """ Every chunk is sealed under its own nonce, and the node appends in order. Nothing about the seal may change what lands on disk. @@ -225,7 +225,7 @@ def test_a_multi_chunk_upload_reassembles(tmp_path): body = bytes(range(256)) * 40 parts = [body[i:i + 1024] for i in range(0, len(body), 1024)] for i, part in enumerate(parts): - session._do_file_upload(sealed_upload( + await session._do_file_upload(sealed_upload( session, filename="blob.bin", data=part, chunk_index=i, total_chunks=len(parts), dir=_root(session).name)) @@ -247,7 +247,7 @@ def test_two_identical_chunks_do_not_reuse_a_nonce(tmp_path): assert a["ct"] != b["ct"] -def test_a_sealed_payload_is_authenticated_not_validated(tmp_path): +async def test_a_sealed_payload_is_authenticated_not_validated(tmp_path): """ Opening a payload proves a member wrote it, not that they wrote something sensible. A member can seal anything, so the fields still need their types @@ -261,7 +261,7 @@ def test_a_sealed_payload_is_authenticated_not_validated(tmp_path): {"filename": "note.txt", "data": "not bytes"}, {"filename": "note.txt"}): session.sent.clear() - session._do_file_upload({ + await session._do_file_upload({ "type": MNP.FILE_UPLOAD, "v": "2.0", "upload_id": "u1", "chunk_index": 0, "total_chunks": 1, **seal(session._ctx["gek"], PURPOSE_UPLOAD, MNP.FILE_UPLOAD, @@ -273,13 +273,13 @@ def test_a_sealed_payload_is_authenticated_not_validated(tmp_path): assert not _wrote_anything(tmp_path) -def test_a_peer_controlled_chunk_index_cannot_crash_the_handler(tmp_path): +async def test_a_peer_controlled_chunk_index_cannot_crash_the_handler(tmp_path): """`chunk_index` is outside the seal by necessity, so it is unchecked input.""" session = _session(tmp_path) msg = sealed_upload(session, filename="note.txt", data=b"x", dir=_root(session).name) msg["chunk_index"] = "zero" - session._do_file_upload(msg) + await session._do_file_upload(msg) assert _errors(session)[0]["code"] == "bad_chunk_index" assert not _wrote_anything(tmp_path) |