path: root/packages/meshbay-node/tests/test_root_writable_policy.py

"""
Who may write to the operator's disk, now that RO/RW on the root decides it.

One property per root — `writable` — and no second control anywhere: not a
group-wide switch, not a designated upload target. Two controls for one question
is a question answered differently depending on which is read first.

The properties this holds:

* 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.
"""

from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from conftest import sealed_upload
from meshbay_common.adminop import OP_ROOT_UPDATE, OP_ROOT_EJECT, OP_ROOT_PLUG
from meshbay_common.crypto import generate_gek
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,
        # Uploads are sealed under the group key since MNP 2.0.
        "gek": generate_gek(),
    }
    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(sealed_upload(
        session, filename=filename, data=body, dir="shared"))


def _uploads_dir(session) -> Path:
    # The root itself: the `uploads/` subdirectory the node used to create is
    # gone (see test_security_regressions._uploads_dir for why).
    return session._ctx["roots"].roots[0].path


# ── 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) / "clip.mp4").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"


async def test_a_member_cannot_create_a_folder_in_a_read_only_root(tmp_path):
    """
    Read-only has to mean read-only for every way of writing, not just for
    files. `_do_file_upload` gained this check with the RO/RW model and
    `_do_dir_create` did not, so a member refused a file in a published library
    could still leave empty directories all through it.

    Creating a folder stays unprivileged — the node's own words: "a member who
    can add a file can organise where it goes". What changed is that it now
    requires the same root to be writable that adding the file would have.
    """
    session = _session(tmp_path, "member-1", writable=False)
    session._do_dir_create({"dir": "shared", "name": "New folder"})

    refusal = [m for m in session.sent if m.get("type") == "error"]
    assert refusal and refusal[0].get("code") == "root_read_only"
    assert not (tmp_path / "shared" / "New folder").exists()


async def test_a_member_can_create_a_folder_in_a_writable_root(tmp_path):
    """The counter-property: it must stay unprivileged where it is allowed."""
    session = _session(tmp_path, "member-1", writable=True)
    session._do_dir_create({"dir": "shared", "name": "New folder"})

    assert not [m for m in session.sent if m.get("type") == "error"]
    assert (tmp_path / "shared" / "New folder").is_dir()


async def test_an_ejected_root_refuses_a_new_folder(tmp_path):
    """Writing to a drive somebody has their hand on, one level up from a file."""
    session = _session(tmp_path, "member-1", writable=True)
    roots = session._ctx["roots"]
    roots.roots[0].ejected = True
    roots.roots[0].available = False

    session._do_dir_create({"dir": "shared", "name": "New folder"})
    refusal = [m for m in session.sent if m.get("type") == "error"]
    assert refusal and refusal[0].get("code") == "root_unavailable"
    assert not (tmp_path / "shared" / "New folder").exists()


# ── 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"]


# ── There is no group-wide upload switch ─────────────────────────────────────

async def test_no_message_can_reopen_uploads_for_a_whole_group(tmp_path):
    """
    Whether a member may write is a property of each root, and there is no
    second way to say it.

    A group-wide switch is the thing this model replaced, and it cannot come
    back by accident: the type does not exist, so a peer asking for it is a peer
    the dispatcher logs and ignores. What must never happen is what a switch
    would have allowed — one instruction turning a published, read-only library
    into a writable one.
    """
    assert not hasattr(MNP, "MEMBER_UPLOAD"), (
        "a group-wide upload switch is back in the protocol")

    session = _session(tmp_path, "member-1", writable=False)
    session._has_admin_authority = lambda: True
    issued = _capture_challenges(session)

    session._dispatch_message({"type": "member_upload", "allowed": True})

    assert issued == [], "an unknown instruction asked to be signed"
    assert session._ctx["roots"].roots[0].writable is False
    assert not [m for m in session.sent if "upload" in str(m.get("type"))], (
        "the node answered an instruction it does not implement")

    # 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"