diff options
Diffstat (limited to 'packages/meshbay-node/tests/test_security_regressions.py')
| -rw-r--r-- | packages/meshbay-node/tests/test_security_regressions.py | 32 |
1 files changed, 21 insertions, 11 deletions
diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py index e13dec0..78a631a 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 conftest import one_root from meshbay_node.transport.webrtc_server import WebRTCPeerSession @@ -129,12 +130,24 @@ def test_the_node_never_generates_a_name_it_would_refuse(tmp_path): f"the node picked {chosen!r} and would then reject it on the next upload") +def _uploads_dir(session) -> Path: + """ + Where this session's uploads land: uploads/ inside the group's upload root. + + 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" + + def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession: """A peer session wired to a real shared root, with sending stubbed out.""" shared_root = tmp_path / "shared" shared_root.mkdir(exist_ok=True) index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) - ctx = {"shared_root": shared_root, "index": index, "sk_node": index.sk_node} + ctx = {"roots": one_root(shared_root), "index": index, "sk_node": index.sk_node} session = WebRTCPeerSession.__new__(WebRTCPeerSession) session._ctx = ctx @@ -161,9 +174,7 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path): this test now asserts — an existing file is never replaced. """ victim = _session(tmp_path, "victim-user") - shared_root = victim._ctx["shared_root"] - - uploads = shared_root / "uploads" + uploads = _uploads_dir(victim) uploads.mkdir() original = uploads / "important.mp4" original.write_bytes(b"operator's original content") @@ -190,7 +201,7 @@ def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path): session.sent.clear() session._do_file_upload(dict(payload)) - uploads = session._ctx["shared_root"] / "uploads" + uploads = _uploads_dir(session) assert (uploads / "movie.mp4").read_bytes() == b"first", ( "the first upload was replaced") assert (uploads / "movie (2).mp4").read_bytes() == b"first" @@ -221,7 +232,6 @@ def test_upload_ignores_any_directory_the_client_asks_for(tmp_path): client-chosen destination would open does not exist on this path. """ session = _session(tmp_path, "user-1") - shared_root = session._ctx["shared_root"] session._do_file_upload({ "filename": "note.txt", "dir": "../../etc", @@ -229,7 +239,7 @@ def test_upload_ignores_any_directory_the_client_asks_for(tmp_path): "data": base64.b64encode(b"x").decode(), }) - assert (shared_root / "uploads" / "note.txt").read_bytes() == b"x" + assert (_uploads_dir(session) / "note.txt").read_bytes() == b"x" assert not (tmp_path / "etc").exists() @@ -249,7 +259,7 @@ def test_two_members_can_send_the_same_filename(tmp_path): "data": base64.b64encode(b"second").decode(), }) - uploads = first._ctx["shared_root"] / "uploads" + uploads = _uploads_dir(first) assert (uploads / "IMG_1234.jpg").read_bytes() == b"first" assert (uploads / "IMG_1234 (2).jpg").read_bytes() == b"second" @@ -270,8 +280,8 @@ def test_chat_store_and_peers_are_per_group(tmp_path): index_a = GroupIndex(group_id="a" * 32, sk_node=Ed25519PrivateKey.generate()) index_b = GroupIndex(group_id="b" * 32, sk_node=Ed25519PrivateKey.generate()) groups = { - "a" * 32: {"chat_store": "STORE_A", "index": index_a, "shared_root": tmp_path}, - "b" * 32: {"chat_store": "STORE_B", "index": index_b, "shared_root": tmp_path}, + "a" * 32: {"chat_store": "STORE_A", "index": index_a, "roots": one_root(tmp_path / "a")}, + "b" * 32: {"chat_store": "STORE_B", "index": index_b, "roots": one_root(tmp_path / "b")}, } ctx = {"groups": groups} @@ -663,7 +673,7 @@ def test_admin_ui_escapes_filenames(tmp_path): html = _render_page({ "status": "running", - "groups_ctx": {"g" * 32: {"index": index, "shared_root": tmp_path}}, + "groups_ctx": {"g" * 32: {"index": index, "roots": one_root(tmp_path)}}, "indexes": {"g" * 32: index}, }) |