1 files changed, 137 insertions, 8 deletions
diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py
index 7f71da5..9db8ac1 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,14 @@ 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 this session's uploads land: uploads/ inside its first writable 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"
+    writable = session._ctx["roots"].writable_roots
+    assert writable, "the fixture must give the group a writable root"
+    return writable[0].path / "uploads"
 
 
 def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession:
@@ -227,14 +228,17 @@ def test_dir_create_cannot_escape_the_shared_root(tmp_path, bad):
 
 def test_upload_ignores_any_directory_the_client_asks_for(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.
+    The destination inside a root is the node's decision, and stays so.
+
+    A client now names the *root* it is uploading into — it has to, once a group
+    can have several writable ones — but that is a name looked up in the root
+    table, never a path. Everything below the root is still chosen here, so the
+    traversal surface a client-chosen destination would open does not exist.
     """
     session = _session(tmp_path, "user-1")
 
     session._do_file_upload({
-        "filename": "note.txt", "dir": "../../etc",
+        "filename": "note.txt", "dir": "../../etc", "path": "/etc",
         "chunk_index": 0, "total_chunks": 1,
         "data": base64.b64encode(b"x").decode(),
     })
@@ -243,6 +247,131 @@ def test_upload_ignores_any_directory_the_client_asks_for(tmp_path):
     assert not (tmp_path / "etc").exists()
 
 
+@pytest.mark.parametrize("named_root", [
+    "../../etc", "/etc", "shared/../..", "Shared/uploads", "nope",
+])
+def test_a_root_name_is_looked_up_never_joined(tmp_path, named_root):
+    """
+    The name the client sends is matched against the group's root table and
+    refused when it matches nothing. A version that joined it to a path — or
+    that quietly fell back to the first writable root — would turn "which
+    directory" into either a traversal or a file on a disk the operator did
+    not intend, and the second is discovered weeks later.
+    """
+    session = _session(tmp_path, "user-1")
+    before = set(tmp_path.rglob("*"))
+
+    session._do_file_upload({
+        "filename": "note.txt", "root": named_root,
+        "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_such_root", named_root
+    assert set(tmp_path.rglob("*")) == before, f"wrote something via {named_root!r}"
+
+
+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", "root": "Incoming",
+        "chunk_index": 0, "total_chunks": 1,
+        "data": base64.b64encode(b"x").decode(),
+    })
+
+    assert (incoming / "uploads" / "note.txt").read_bytes() == b"x"
+    assert not (media / "uploads").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", "root": "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 / "uploads").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 / "uploads").exists()
+
+
+def test_an_ejected_root_refuses_an_upload(tmp_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", "root": "USB",
+        "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_unavailable"
+    assert not (usb / "uploads").exists()
+
+
 def test_two_members_can_send_the_same_filename(tmp_path):
     """
     One shared uploads/ means collisions are ordinary — every camera produces