diff options
Diffstat (limited to 'packages/meshbay-node/tests/test_security_regressions.py')
| -rw-r--r-- | packages/meshbay-node/tests/test_security_regressions.py | 167 |
1 files changed, 155 insertions, 12 deletions
diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py index 7f71da5..1a318f7 100644 --- a/packages/meshbay-node/tests/test_security_regressions.py +++ b/packages/meshbay-node/tests/test_security_regressions.py @@ -18,6 +18,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common.protocol import IndexEntry from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roots import RootSet from conftest import one_root from meshbay_node.transport.webrtc_server import WebRTCPeerSession @@ -132,14 +133,22 @@ def test_the_node_never_generates_a_name_it_would_refuse(tmp_path): def _uploads_dir(session) -> Path: """ - Where this session's uploads land: uploads/ inside the group's upload root. + Where an unaddressed upload lands: the first writable root itself. + + There is no `uploads/` subdirectory any more. It was the last of v5's + quarantine — the per-user layer went on 2026-08-14 — and it went for the + same reason: a folder appearing beside the operator's library because + somebody sent a file is the node deciding how their disk is arranged. The + protections that made the quarantine worth having are the allowlist, the + size cap, the chunk ordering and the no-overwrite rule, and every one of + them is asserted below, unchanged. Asked of the root set rather than assembled by hand, so a test cannot pass while agreeing with a wrong answer the code also produced. """ - root = session._ctx["roots"].upload_root - assert root is not None, "the fixture must designate an upload root" - return root.path / "uploads" + writable = session._ctx["roots"].writable_roots + assert writable, "the fixture must give the group a writable root" + return writable[0].path def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession: @@ -175,7 +184,6 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path): """ victim = _session(tmp_path, "victim-user") uploads = _uploads_dir(victim) - uploads.mkdir() original = uploads / "important.mp4" original.write_bytes(b"operator's original content") @@ -225,22 +233,157 @@ 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_upload_ignores_any_directory_the_client_asks_for(tmp_path): +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 + choice is *resolved against the group's own roots* rather than joined to + one. + + `RootSet.resolve()` refuses `..`, absolute segments and anything whose + resolved form escapes its root, symlinks included. So "which of this + group's folders" is answerable by a member and "which path on the + operator's disk" is not. + """ + session = _session(tmp_path, "user-1") + (session._ctx["roots"].roots[0].path / "sub").mkdir() + before = set(tmp_path.rglob("*")) + + for bad in ("../../etc", "/etc", "shared/../..", "shared/../../etc", + "nope", "shared/missing"): + session.sent.clear() + session._do_file_upload({ + "filename": "note.txt", "dir": bad, + "chunk_index": 0, "total_chunks": 1, + "data": base64.b64encode(b"x").decode(), + }) + refusal = [m for m in session.sent if m.get("type") == "error"] + assert refusal, f"{bad!r} was accepted" + assert refusal[0].get("code") in ("no_such_root", "no_such_directory"), bad + + assert set(tmp_path.rglob("*")) == before, "a refused upload still wrote" + + +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 + at expects it to be in that folder. + """ + session = _session(tmp_path, "user-1") + root = session._ctx["roots"].roots[0] + (root.path / "Albums").mkdir() + + session._do_file_upload({ + "filename": "note.txt", "dir": f"{root.name}/Albums", + "chunk_index": 0, "total_chunks": 1, + "data": base64.b64encode(b"x").decode(), + }) + + assert (root.path / "Albums" / "note.txt").read_bytes() == b"x" + assert not (root.path / "Albums" / "uploads").exists(), ( + "the node invented a subdirectory in the operator's library") + assert not (root.path / "uploads").exists() + + +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 + node picking one meant a file uploaded from a folder on screen landed in a + different one — the same "uploads went somewhere else" the single upload + root was never allowed to guess about. + """ + media = tmp_path / "Media" + incoming = tmp_path / "Incoming" + media.mkdir() + incoming.mkdir() + session = _session(tmp_path, "user-1") + session._ctx["roots"] = RootSet.build([ + {"path": str(media), "writable": True}, + {"path": str(incoming), "writable": True}, + ]) + + session._do_file_upload({ + "filename": "note.txt", "dir": "Incoming", + "chunk_index": 0, "total_chunks": 1, + "data": base64.b64encode(b"x").decode(), + }) + + 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): + """ + 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 + exception for whoever happens to hold admin authority is the sort of + carve-out that later reads as the rule. + """ + published = tmp_path / "Published" + published.mkdir() + session = _session(tmp_path, "user-1") + session._ctx["roots"] = RootSet.build([{"path": str(published)}]) + session._is_node_admin = lambda: True + + session._do_file_upload({ + "filename": "note.txt", "dir": "Published", + "chunk_index": 0, "total_chunks": 1, + "data": base64.b64encode(b"x").decode(), + }) + + refusal = [m for m in session.sent if m.get("type") == "error"] + assert refusal and refusal[0].get("code") == "root_read_only" + assert not (published / "note.txt").exists() + + +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 + than write into whatever root happens to come first. + """ + published = tmp_path / "Published" + published.mkdir() + session = _session(tmp_path, "user-1") + session._ctx["roots"] = RootSet.build([{"path": str(published)}]) + + session._do_file_upload({ + "filename": "note.txt", + "chunk_index": 0, "total_chunks": 1, + "data": base64.b64encode(b"x").decode(), + }) + + refusal = [m for m in session.sent if m.get("type") == "error"] + assert refusal and refusal[0].get("code") == "no_writable_root" + assert not (published / "note.txt").exists() + + +def test_an_ejected_root_refuses_an_upload(tmp_path): """ - Uploads land in uploads/, chosen by the node. A client that names somewhere - else — or nowhere at all — changes nothing, so the traversal surface that a - client-chosen destination would open does not exist on this 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 + has to be checked separately, which is what an earlier version conflated. """ + usb = tmp_path / "USB" + usb.mkdir() session = _session(tmp_path, "user-1") + roots = RootSet.build([{"path": str(usb), "writable": True, + "removable": True}]) + roots.roots[0].ejected = True + roots.roots[0].available = False + session._ctx["roots"] = roots session._do_file_upload({ - "filename": "note.txt", "dir": "../../etc", + "filename": "note.txt", "dir": "USB", "chunk_index": 0, "total_chunks": 1, "data": base64.b64encode(b"x").decode(), }) - assert (_uploads_dir(session) / "note.txt").read_bytes() == b"x" - assert not (tmp_path / "etc").exists() + refusal = [m for m in session.sent if m.get("type") == "error"] + assert refusal and refusal[0].get("code") == "root_unavailable" + assert not (usb / "note.txt").exists() def test_two_members_can_send_the_same_filename(tmp_path): |