1 files changed, 203 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_root_writable_policy.py b/packages/meshbay-node/tests/test_root_writable_policy.py
new file mode 100644
index 0000000..da95032
--- /dev/null
+++ b/packages/meshbay-node/tests/test_root_writable_policy.py
@@ -0,0 +1,203 @@
+"""
+Who may write to the operator's disk, now that RO/RW on the root decides it.
+
+This replaces `test_member_upload_policy.py`. The old model had two orthogonal
+controls — one root designated as the upload target, and a group-wide
+`member_upload` switch — and collapsed into one property per root: `writable`.
+The properties worth keeping from the old file survive the change unaltered:
+
+* the interface hiding a control is a courtesy to the people who are not
+  trying; **the node refusing is the part that holds** against someone who is.
+  A member with an old tab open, or one speaking MNP directly, gets the same
+  answer. That half is pinned in `test_security_regressions.py`, next to the
+  overwrite properties it belongs with;
+* the setting is changed by a **signed** operator instruction, or it is a
+  suggestion any member can undo;
+* it is stored on the **node**, never the hub. A hub that could decide who
+  writes to the operator's disk would have authority over the node.
+
+And one that is new: the *old* message must no longer be able to change
+anything. A deprecated instruction that still works is not deprecated, and this
+one would reopen uploads group-wide.
+"""
+
+import base64
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.adminop import OP_ROOT_UPDATE, OP_ROOT_EJECT, OP_ROOT_PLUG
+from meshbay_common.protocol import MNP
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roots import RootSet
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+pytestmark = pytest.mark.asyncio
+
+
+def _session(tmp_path: Path, user_id: str, *,
+             writable: bool = True,
+             operator: str | None = None) -> WebRTCPeerSession:
+    shared_root = tmp_path / "shared"
+    shared_root.mkdir(exist_ok=True)
+    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
+    ctx = {
+        "roots": RootSet.build([{"path": str(shared_root), "writable": writable}]),
+        "index": index,
+        "sk_node": index.sk_node,
+        "node_user_id": operator,
+    }
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = ctx
+    session._group_id = "g" * 32
+    session._user_id = user_id
+    session._pk_user = ""
+    session._uploads = {}
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+def _upload(session, filename="clip.mp4", body=b"bytes"):
+    session._do_file_upload({
+        "filename": filename, "root": "shared",
+        "chunk_index": 0, "total_chunks": 1,
+        "data": base64.b64encode(body).decode(),
+    })
+
+
+def _uploads_dir(session) -> Path:
+    return session._ctx["roots"].roots[0].path / "uploads"
+
+
+# ── The door, not the button ─────────────────────────────────────────────────
+
+async def test_a_member_cannot_upload_to_a_read_only_root(tmp_path):
+    session = _session(tmp_path, "member-1", writable=False)
+    _upload(session)
+
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_read_only"
+    assert not _uploads_dir(session).exists()
+
+
+async def test_members_upload_normally_to_a_writable_root(tmp_path):
+    session = _session(tmp_path, "member-1", writable=True)
+    _upload(session)
+
+    assert not [m for m in session.sent if m.get("type") == "error"]
+    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
+
+
+async def test_read_only_binds_the_operator_too(tmp_path):
+    """
+    The old model exempted the operator, because the switch was about *members*.
+    RO is about the directory: a published library is read-only for everyone, and
+    an exception for admin authority is how a rule turns into a default.
+    """
+    session = _session(tmp_path, "the-operator", writable=False,
+                       operator="the-operator")
+    session._is_node_admin = lambda: True
+    _upload(session)
+
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_read_only"
+
+
+# ── Signed, or it is a suggestion ────────────────────────────────────────────
+
+def _capture_challenges(session) -> list[tuple[str, str]]:
+    issued: list[tuple[str, str]] = []
+
+    def issue(op, subject, **kw):
+        issued.append((op, subject))
+
+    session._issue_admin_challenge = issue
+    session._has_admin_authority = lambda: True
+    return issued
+
+
+async def test_changing_a_roots_flags_needs_a_signature(tmp_path):
+    """The flags are not applied by the request — only by the signed response."""
+    session = _session(tmp_path, "the-operator", operator="the-operator")
+    issued = _capture_challenges(session)
+
+    session._do_root_update({"group_id": "g" * 32, "root_name": "shared",
+                             "writable": False})
+
+    assert [op for op, _ in issued] == [OP_ROOT_UPDATE]
+    assert session._ctx["roots"].roots[0].writable is True, (
+        "applied before it was signed")
+
+
+async def test_the_subject_names_the_outcome_not_the_operation(tmp_path):
+    """
+    The operator is shown the subject before signing, so it has to say what will
+    be true afterwards. "shared" alone would have them authorize a change they
+    cannot see the direction of.
+    """
+    session = _session(tmp_path, "op", operator="op")
+    issued = _capture_challenges(session)
+
+    session._do_root_update({"group_id": "g" * 32, "root_name": "shared",
+                             "writable": True, "removable": True})
+
+    assert issued == [(OP_ROOT_UPDATE, "shared:rw=on,rem=on")]
+
+
+async def test_eject_and_plug_are_signed_too(tmp_path):
+    """
+    Hiding a group's whole library from every member is not a lesser act than
+    changing a flag. An unsigned one would let any member black out a group.
+    """
+    session = _session(tmp_path, "op", operator="op")
+    issued = _capture_challenges(session)
+
+    session._do_root_eject({"group_id": "g" * 32, "root_name": "shared"})
+    session._do_root_plug({"group_id": "g" * 32, "root_name": "shared"})
+
+    assert issued == [(OP_ROOT_EJECT, "shared"), (OP_ROOT_PLUG, "shared")]
+
+
+async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
+    """
+    An unpaired node has no key to check a signature against, so the challenge
+    is never issued rather than issued and then unverifiable.
+    """
+    session = _session(tmp_path, "member-1")
+    issued = _capture_challenges(session)
+    session._has_admin_authority = lambda: False
+
+    session._do_root_update({"group_id": "g" * 32, "root_name": "shared",
+                             "writable": True})
+
+    assert issued == []
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+# ── The deprecated message must not still work ───────────────────────────────
+
+async def test_the_old_member_upload_message_changes_nothing(tmp_path):
+    """
+    MNP still parses `member_upload` so an old client gets an answer instead of
+    a dropped request. What it must not do is act: this instruction could
+    reopen uploads for a whole group, and a client old enough to send it is
+    exactly one that knows nothing about read-only roots.
+    """
+    session = _session(tmp_path, "member-1", writable=False)
+    session._has_admin_authority = lambda: True
+    issued = _capture_challenges(session)
+
+    session._do_member_upload({"allowed": True})
+
+    assert issued == [], "a deprecated instruction asked to be signed"
+    assert session._ctx["roots"].roots[0].writable is False
+    acks = [m for m in session.sent if m.get("type") == MNP.MEMBER_UPLOAD_ACK]
+    assert acks and acks[0].get("deprecated") is True
+
+    # And the door is still shut.
+    _upload(session)
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_read_only"