diff options
Diffstat (limited to 'packages/meshbay-node/tests')
| -rw-r--r-- | packages/meshbay-node/tests/test_security_regressions.py | 86 |
1 files changed, 68 insertions, 18 deletions
diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py index dcd9cf6..7627926 100644 --- a/packages/meshbay-node/tests/test_security_regressions.py +++ b/packages/meshbay-node/tests/test_security_regressions.py @@ -120,9 +120,14 @@ def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession: 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 files, - and — by becoming the recorded uploader of the replaced file — delete them - through the uploader path, bypassing the Ed25519 admin challenge entirely. + overwrite whatever was there. That let any member destroy the operator's + files, and — by becoming the recorded uploader of the replaced file — delete + them through the uploader path, bypassing the Ed25519 admin challenge. + + The per-user quarantine that fixed it was removed on 2026-08-14: files now go + where the member is looking, because a shared directory nobody can organise is + not a shared directory. What made the quarantine work is kept, and is what + this test now asserts — an existing file is never replaced. """ victim = _session(tmp_path, "victim-user") shared_root = victim._ctx["shared_root"] @@ -139,23 +144,14 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path): }) assert original.read_bytes() == b"operator's original content" - uploaded = shared_root / ".uploads" / "attacker-user" / "important.mp4" - assert uploaded.exists(), "upload should be quarantined, not dropped" - assert uploaded.read_bytes() == b"attacker content" - - -def test_upload_rejects_out_of_order_chunks(tmp_path): - """C5a: chunk_index > 0 used to append blindly to any .part file on disk.""" - session = _session(tmp_path, "user-1") - session._do_file_upload({ - "filename": "movie.mp4", "chunk_index": 3, "total_chunks": 5, - "data": base64.b64encode(b"spliced").decode(), - }) - assert any(m.get("type") == "error" for m in session.sent) + assert any(m.get("type") == "error" for m in attacker.sent), ( + "the upload must be refused outright, not silently dropped") + assert not (shared_root / "important.mp4.part").exists(), ( + "a refused upload must leave nothing behind") def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path): - """C5a: even the original uploader goes through a fresh name, not an overwrite.""" + """C5a: even the original uploader does not get to overwrite.""" session = _session(tmp_path, "user-1") payload = {"filename": "movie.mp4", "chunk_index": 0, "total_chunks": 1, "data": base64.b64encode(b"first").decode()} @@ -164,10 +160,64 @@ def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path): session._do_file_upload(dict(payload)) assert any(m.get("type") == "error" for m in session.sent) - stored = session._ctx["shared_root"] / ".uploads" / "user-1" / "movie.mp4" + stored = session._ctx["shared_root"] / "movie.mp4" assert stored.read_bytes() == b"first" +@pytest.mark.parametrize("bad_dir", [ + "..", "../..", "/etc", "a/../../b", "./../x", "sub/../../..", + "\\..\\..", "~", "a/./../..", +]) +def test_upload_cannot_escape_the_shared_root(tmp_path, bad_dir): + """ + The destination now arrives from the client, which is a path the node did not + choose. Every segment goes through the same allowlist as a filename and the + result must resolve inside the shared root. + """ + session = _session(tmp_path, "user-1") + outside = tmp_path / "outside.txt" + + session._do_file_upload({ + "filename": "outside.txt", "dir": bad_dir, + "chunk_index": 0, "total_chunks": 1, + "data": base64.b64encode(b"escaped").decode(), + }) + + assert any(m.get("type") == "error" for m in session.sent), bad_dir + assert not outside.exists(), f"upload escaped the shared root via {bad_dir!r}" + + +@pytest.mark.parametrize("bad", [ + {"dir": "..", "name": "evil"}, + {"dir": "", "name": ".."}, + {"dir": "", "name": "a/b"}, + {"dir": "/etc", "name": "evil"}, + {"dir": "", "name": ".hidden"}, +]) +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) + + assert any(m.get("type") == "error" for m in session.sent), bad + assert set(tmp_path.rglob("*")) == before, f"created something via {bad!r}" + + +def test_upload_still_refuses_unsafe_names_in_a_subdirectory(tmp_path): + """The name allowlist is not weakened by having somewhere to put the file.""" + session = _session(tmp_path, "user-1") + (session._ctx["shared_root"] / "docs").mkdir() + + session._do_file_upload({ + "filename": "../escape.txt", "dir": "docs", + "chunk_index": 0, "total_chunks": 1, + "data": base64.b64encode(b"x").decode(), + }) + assert any(m.get("type") == "error" for m in session.sent) + + # ── H1: group isolation ────────────────────────────────────────────────────── def test_chat_store_and_peers_are_per_group(tmp_path): |