diff options
Diffstat (limited to 'packages/meshbay-node')
5 files changed, 122 insertions, 53 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 92f2951..3bb0df7 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -646,9 +646,9 @@ class WebRTCPeerSession: elif mtype == MNP.FILE_UPLOAD: self._spawn(self._do_file_upload(msg)) elif mtype == MNP.DIR_CREATE: - self._do_dir_create(msg) + self._spawn(self._do_dir_create(msg)) elif mtype == MNP.DIR_DELETE: - self._do_dir_delete(msg) + self._spawn(self._do_dir_delete(msg)) elif mtype == MNP.FILE_DELETE: self._do_file_delete(msg) elif mtype == MNP.ADMIN_RESPONSE: @@ -2034,7 +2034,7 @@ class WebRTCPeerSession: self._send(reply) self._audit_join("gek_wrapped", f"group={group_id[:8]}") - def _do_dir_create(self, msg: dict) -> None: + async def _do_dir_create(self, msg: dict) -> None: """ Create a directory, for any member of the group. @@ -2085,20 +2085,19 @@ class WebRTCPeerSession: "code": "root_unavailable"}) return - parent = safe_subdir(roots, parent_rel) - if parent is None or not parent.is_dir(): + parent = await off_disk(roots, safe_subdir, roots, parent_rel) + if parent is None or not await off_disk(roots, parent.is_dir): self._send({"type": "error", "detail": "Invalid directory"}) return - target = safe_subdir(roots, f"{parent_rel}/{name}") + target = await off_disk(roots, safe_subdir, roots, f"{parent_rel}/{name}") if target is None: self._send({"type": "error", "detail": "Invalid directory"}) return - if target.exists(): - self._send({"type": "error", "detail": "Already exists"}) + refusal = await off_disk(roots, _mkdir_if_absent, target) + if refusal is not None: + self._send({"type": "error", "detail": refusal}) return - - target.mkdir(parents=False) virtual = roots.virtual_of(target) or f"{parent_rel}/{name}" log.info("Directory created by %s: %s", self._user_id[:8], virtual) self._audit("dir_create", virtual) @@ -2113,7 +2112,7 @@ class WebRTCPeerSession: found = roots.split(rel or "") return found is not None and not found[1] - def _do_dir_delete(self, msg: dict) -> None: + async def _do_dir_delete(self, msg: dict) -> None: """ Remove an empty directory, for the node operator. @@ -2131,17 +2130,17 @@ class WebRTCPeerSession: return rel = (msg.get("dir") or "").strip("/") - target = safe_subdir(roots, rel) + target = await off_disk(roots, safe_subdir, roots, rel) # A root itself is not deletable here: removing one is a configuration # change, and doing it through a file operation would leave the group # config naming a directory nobody can reach. if target is None or self._names_a_root(roots, rel): self._send({"type": "error", "detail": "Invalid directory"}) return - if not target.is_dir(): + if not await off_disk(roots, target.is_dir): self._send({"type": "error", "detail": "Not a directory"}) return - if any(target.iterdir()): + if not await off_disk(roots, _is_empty_dir, target): self._send({"type": "error", "detail": "Directory is not empty"}) return if not self._has_admin_authority(): @@ -2157,9 +2156,9 @@ class WebRTCPeerSession: rel = pending["subject"] ctx = self._group_ctx() roots: RootSet | None = ctx.get("roots") - target = safe_subdir(roots, rel) if roots else None + target = await off_disk(roots, safe_subdir, roots, rel) if roots else None if (target is None or self._names_a_root(roots, rel) - or not target.is_dir()): + or not await off_disk(roots, target.is_dir)): self._send({"type": "error", "detail": "Not a directory"}) return @@ -2173,11 +2172,9 @@ class WebRTCPeerSession: # Checked again after the signature: the emptiness test that let this # through happened before a round trip to the operator's browser, and a # file could have landed in the meantime. - if any(target.iterdir()): + if not await off_disk(roots, _rmdir_if_empty, target): self._send({"type": "error", "detail": "Directory is not empty"}) return - - target.rmdir() log.info("Directory removed by %s: %s", self._user_id[:8], rel) self._audit("dir_delete", rel) self._send({"type": MNP.DIR_DELETE_ACK, "v": MNP_VERSION, "dir": rel}) @@ -6607,6 +6604,42 @@ class WebRTCPeerSession: await self._pc.close() +def _mkdir_if_absent(target: Path) -> str | None: + """ + Create a directory unless it is already there, or say why not. + + Both in one call, not a check awaited and then an act: the disk thread is + one worker, so nothing can slip between them. Split across two awaits, two + members creating the same name would both find nothing there and the second + `mkdir` would raise where a refusal was meant. Blocking; called through + `off_disk`. + """ + if target.exists(): + return "Already exists" + target.mkdir(parents=False) + return None + + +def _is_empty_dir(target: Path) -> bool: + """Blocking; called through `off_disk`.""" + return not any(target.iterdir()) + + +def _rmdir_if_empty(target: Path) -> bool: + """ + Remove a directory if nothing is in it. False if something is. + + The emptiness test and the removal are one call for the reason the caller + re-tests at all: the first test happened before a round trip to the + operator's browser, and a file can land in between. Two awaits here would + reopen the same window one size smaller. Blocking; called through `off_disk`. + """ + if any(target.iterdir()): + return False + target.rmdir() + return True + + def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]: """ Where an entry is, and whether it is readable — or the refusal to send. 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 2811836..2179e66 100644 --- a/packages/meshbay-node/tests/test_disk_io_off_loop.py +++ b/packages/meshbay-node/tests/test_disk_io_off_loop.py @@ -19,8 +19,8 @@ means two reads of the same file can never be inside it at once, and that is what makes the single `f.seek()`/`f.read()` pair safe without a lock. """ +import ast import asyncio -import re import threading import time from pathlib import Path @@ -204,29 +204,6 @@ async def test_one_root_set_reads_one_chunk_at_a_time(tmp_path): assert peak == 1, f"{peak} reads of one root set were inside the disk at once" -def test_no_handler_resolves_a_path_on_the_loop(): - """ - `entry_abs_path` is `Path.resolve()`, which is syscalls — it belongs on the - disk thread with everything else. - - This reads the source because there is nothing else to read: a handler added - later that resolves an entry inline would pass every test above, since those - only exercise the handlers that exist today. `_locate` is the one place - allowed to call it, and `off_disk` is how `_locate` is reached. - """ - src = Path(webrtc_server.__file__).read_text() - # Every call site, not the first one: a guard that stops at the first - # occurrence stops guarding the moment a new call is inserted above it. - calls = [m.start() for m in re.finditer(r"\bentry_abs_path\(", src)] - body = re.search(r"\ndef _locate\(.*?\n(?=\n\ndef |\n\nclass )", src, re.S) - assert body, "_locate is gone or has been renamed — this guard needs rewriting" - allowed = range(body.start(), body.end()) - stray = [c for c in calls if c not in allowed] - 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. @@ -255,6 +232,65 @@ async def test_the_availability_poll_does_not_stop_the_loop(tmp_path, monkeypatc f"the loop was blocked: {ticker.ticks} wake-ups during a {SLOW_S}s poll") +def test_no_handler_touches_the_disk_on_the_loop(): + """ + The whole class, not the calls that were fixed. + + Every measured test above exercises a handler that exists today; a new one + that stats a root inline would pass all of them. So this walks the module's + syntax tree instead and fails on any filesystem call outside the few + functions written to be run through `off_disk`. + + `entry_abs_path` and `safe_subdir` are in the list because both are + `Path.resolve()` underneath, and a resolve is syscalls whatever it is + called. + """ + blocking = {"is_dir", "exists", "mkdir", "unlink", "rename", "rmdir", + "iterdir", "read_bytes", "write_bytes", "stat", + "resolve", "entry_abs_path"} + # Written to block, and reached only through `off_disk`. + on_the_disk_thread = {"_locate", "_append_chunk", "_read_and_encrypt", + "_mkdir_if_absent", "_is_empty_dir", "_rmdir_if_empty", + "safe_subdir"} + # ffmpeg's own output, under `tempfile.mkstemp` on the system disk — not a + # group root, so not what spins down. Listed rather than silently allowed: + # these still read a whole transcode into memory from the loop, and the day + # that matters it is a different measurement from this one. + ffmpeg_scratch = {"_transcode_audio_to_aac", "_seek_lands_at", + "_extract_subtitle_to_webvtt"} + allowed = on_the_disk_thread | ffmpeg_scratch + + found = [] + + def visit(node, owner): + for child in ast.iter_child_nodes(node): + if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)): + visit(child, child.name) + continue + if isinstance(child, ast.Call): + fn = child.func + name = (fn.attr if isinstance(fn, ast.Attribute) + else getattr(fn, "id", "")) + if name in blocking and owner not in allowed: + found.append(f"{owner} calls {name}() at line {child.lineno}") + visit(child, owner) + + tree = ast.parse(Path(webrtc_server.__file__).read_text()) + for node in tree.body: + if isinstance(node, ast.ClassDef): + for member in node.body: + if isinstance(member, (ast.FunctionDef, ast.AsyncFunctionDef)): + visit(member, member.name) + elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + visit(node, node.name) + + assert not found, ( + "filesystem calls made from the event loop:\n " + + "\n ".join(found) + + "\nRun them through `off_disk(roots, ...)`, or put the call in a " + "helper that is only reached that way.") + + def _upload_session(tmp_path): """One connection into a group with one writable root, as a node has.""" shared = tmp_path / "shared" diff --git a/packages/meshbay-node/tests/test_root_writable_policy.py b/packages/meshbay-node/tests/test_root_writable_policy.py index 23b55cb..3c8a837 100644 --- a/packages/meshbay-node/tests/test_root_writable_policy.py +++ b/packages/meshbay-node/tests/test_root_writable_policy.py @@ -121,7 +121,7 @@ async def test_a_member_cannot_create_a_folder_in_a_read_only_root(tmp_path): requires the same root to be writable that adding the file would have. """ session = _session(tmp_path, "member-1", writable=False) - session._do_dir_create({"dir": "shared", "name": "New folder"}) + await session._do_dir_create({"dir": "shared", "name": "New folder"}) refusal = [m for m in session.sent if m.get("type") == "error"] assert refusal and refusal[0].get("code") == "root_read_only" @@ -131,7 +131,7 @@ async def test_a_member_cannot_create_a_folder_in_a_read_only_root(tmp_path): async def test_a_member_can_create_a_folder_in_a_writable_root(tmp_path): """The counter-property: it must stay unprivileged where it is allowed.""" session = _session(tmp_path, "member-1", writable=True) - session._do_dir_create({"dir": "shared", "name": "New folder"}) + await session._do_dir_create({"dir": "shared", "name": "New folder"}) assert not [m for m in session.sent if m.get("type") == "error"] assert (tmp_path / "shared" / "New folder").is_dir() @@ -144,7 +144,7 @@ async def test_an_ejected_root_refuses_a_new_folder(tmp_path): roots.roots[0].ejected = True roots.roots[0].available = False - session._do_dir_create({"dir": "shared", "name": "New folder"}) + await session._do_dir_create({"dir": "shared", "name": "New folder"}) refusal = [m for m in session.sent if m.get("type") == "error"] assert refusal and refusal[0].get("code") == "root_unavailable" assert not (tmp_path / "shared" / "New folder").exists() diff --git a/packages/meshbay-node/tests/test_roster_pairing.py b/packages/meshbay-node/tests/test_roster_pairing.py index eb13c61..14285e7 100644 --- a/packages/meshbay-node/tests/test_roster_pairing.py +++ b/packages/meshbay-node/tests/test_roster_pairing.py @@ -984,7 +984,7 @@ async def test_a_directory_with_anything_in_it_is_refused(tmp_path, roster): full.mkdir() (full / "keep.txt").write_text("still here") - session._do_dir_delete({"dir": "shared/full"}) + await session._do_dir_delete({"dir": "shared/full"}) assert _last(session).get("detail") == "Directory is not empty" assert full.exists() and (full / "keep.txt").exists() @@ -996,7 +996,7 @@ async def test_no_challenge_is_issued_without_an_operator(tmp_path, roster): session._ctx["has_admin_authority"] = False (tmp_path / "shared" / "empty").mkdir() - session._do_dir_delete({"dir": "shared/empty"}) + await session._do_dir_delete({"dir": "shared/empty"}) assert _last(session).get("detail") == "No authorized key for deletion" assert (tmp_path / "shared" / "empty").exists() @@ -1013,7 +1013,7 @@ async def test_a_root_itself_is_not_a_target(tmp_path, roster): """ session = await _dir_session(tmp_path, roster) for attempt in ("", ".", "/", "../shared", "shared", "shared/", "SHARED"): - session._do_dir_delete({"dir": attempt}) + await session._do_dir_delete({"dir": attempt}) assert _last(session).get("type") == "error", f"{attempt!r} was accepted" assert (tmp_path / "shared").is_dir() @@ -1028,7 +1028,7 @@ async def test_escaping_the_shared_root_is_refused(tmp_path, roster): for attempt in ("../outside", "../../outside", "sub/../../outside", "shared/../outside", "shared/../../outside", "shared/sub/../../outside"): - session._do_dir_delete({"dir": attempt}) + await session._do_dir_delete({"dir": attempt}) assert _last(session).get("type") == "error", f"{attempt!r} was accepted" assert outside.is_dir(), "a path leaving the shared root removed a directory" @@ -1044,7 +1044,7 @@ async def test_an_empty_directory_needs_a_signature_and_then_goes(tmp_path, rost session = await _dir_session(tmp_path, roster) (tmp_path / "shared" / "gone").mkdir() - session._do_dir_delete({"dir": "shared/gone"}) + await session._do_dir_delete({"dir": "shared/gone"}) challenge = _last(session) assert challenge["type"] == "admin_challenge" assert challenge["op"] == "dir_delete" diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py index e203811..a229153 100644 --- a/packages/meshbay-node/tests/test_security_regressions.py +++ b/packages/meshbay-node/tests/test_security_regressions.py @@ -225,12 +225,12 @@ async def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path) {"dir": "/etc", "name": "evil"}, {"dir": "", "name": ".hidden"}, ]) -def test_dir_create_cannot_escape_the_shared_root(tmp_path, bad): +async def test_dir_create_cannot_escape_the_shared_root(tmp_path, bad): """Creating a directory is not privileged, but it still writes to a disk.""" session = _session(tmp_path, "user-1") before = set(tmp_path.rglob("*")) - session._do_dir_create(bad) + await session._do_dir_create(bad) assert any(m.get("type") == "error" for m in session.sent), bad assert set(tmp_path.rglob("*")) == before, f"created something via {bad!r}" |