diff options
Diffstat (limited to 'packages/meshbay-node')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 22 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_root_writable_policy.py | 41 |
2 files changed, 63 insertions, 0 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 61458f2..0e2d418 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -1595,6 +1595,28 @@ class WebRTCPeerSession: "detail": "Choose a folder to create this in"}) return + # Read-only means read-only, and creating a folder writes to the + # operator's disk. `_do_file_upload` gained this check with the RO/RW + # model and this one did not — so a member could not add a file to a + # published library but could still leave empty directories in it. + owner = roots.split(parent_rel) + if owner is None: + self._send({"type": "error", "detail": "Invalid directory"}) + return + parent_root, _tail = owner + if not parent_root.writable: + self._send({"type": "error", + "detail": f"Directory '{parent_root.name}' is read-only", + "code": "root_read_only"}) + self._audit("dir_create_refused", parent_rel[:64]) + return + if not parent_root.available: + self._send({"type": "error", + "detail": f"Directory '{parent_root.name}' is " + f"currently unavailable", + "code": "root_unavailable"}) + return + parent = safe_subdir(roots, parent_rel) if parent is None or not parent.is_dir(): self._send({"type": "error", "detail": "Invalid directory"}) diff --git a/packages/meshbay-node/tests/test_root_writable_policy.py b/packages/meshbay-node/tests/test_root_writable_policy.py index d7f2666..7eb75fd 100644 --- a/packages/meshbay-node/tests/test_root_writable_policy.py +++ b/packages/meshbay-node/tests/test_root_writable_policy.py @@ -108,6 +108,47 @@ async def test_read_only_binds_the_operator_too(tmp_path): assert refusal and refusal[0].get("code") == "root_read_only" +async def test_a_member_cannot_create_a_folder_in_a_read_only_root(tmp_path): + """ + Read-only has to mean read-only for every way of writing, not just for + files. `_do_file_upload` gained this check with the RO/RW model and + `_do_dir_create` did not, so a member refused a file in a published library + could still leave empty directories all through it. + + Creating a folder stays unprivileged — the node's own words: "a member who + can add a file can organise where it goes". What changed is that it now + 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"}) + + refusal = [m for m in session.sent if m.get("type") == "error"] + assert refusal and refusal[0].get("code") == "root_read_only" + assert not (tmp_path / "shared" / "New folder").exists() + + +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"}) + + assert not [m for m in session.sent if m.get("type") == "error"] + assert (tmp_path / "shared" / "New folder").is_dir() + + +async def test_an_ejected_root_refuses_a_new_folder(tmp_path): + """Writing to a drive somebody has their hand on, one level up from a file.""" + session = _session(tmp_path, "member-1", writable=True) + roots = session._ctx["roots"] + roots.roots[0].ejected = True + roots.roots[0].available = False + + 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() + + # ── Signed, or it is a suggestion ──────────────────────────────────────────── def _capture_challenges(session) -> list[tuple[str, str]]: |