feat(node): the operator can close uploading to everyone but themselves

A group where every member may add files stays the default. Some groups want a
library the operator curates, and until now the only way to get one was to
designate no upload root at all — which refuses the operator too.

**The node enforces it; the interface merely stops offering it.** The Upload
button in the Files toolbar and the paperclip in the chat composer both
disappear, which is a courtesy to the people who are not trying. The control is
`_do_file_upload` refusing with `member_upload_off`, so a member on an old tab,
or one speaking MNP directly, gets the same answer. There is a test for each,
and the enforcement test is in the node package rather than beside the UI one so
nobody reads the hidden button as the mechanism.

**Changing it is a signed operator instruction** — `OP_MEMBER_UPLOAD`, on the
same path as removing a member. An unsigned one would let any member turn it
back on and make the setting a suggestion. The transcript's subject is `on` or
`off`: what the operator is shown before signing has to name the outcome, not
the operation.

**It lives on the node**, in a new `group_settings` table in `roster.db`. Not
the hub, which has no business deciding who may write to someone else's disk.
Not `node.toml` either: that file is hand-written and full of comments recording
decisions, `ops.py` appends to it rather than round-tripping it through a
writer, and a setting toggled from a panel must not rewrite the operator's file
or need a restart. The value is cached in the group context because the upload
path is synchronous, and the signed operation updates both — storing it without
applying it would make the panel say one thing while the node did another.

**Absent means allowed**, at every layer: no row in the table, no key in the
context, no field in `handshake_ack`. An older node and an older client both
behave exactly as before, and upgrading never silently closes a group. Each of
those three has its own test, because they fail independently.

The operator is always exempt — otherwise turning it off locks them out of their
own node with a config file and a restart as the only way back. `is_node_admin`
was being computed in two places by then and is now one function, since two
copies of "is this the operator" is how the ack and the gate come to disagree.

A change reaches everyone already connected via `member_upload_ack`, so the
button goes without a reconnection. That message is both a broadcast and the
reply to the request that caused it, which is why the client does not return
early on it.

Docs updated for a cold start: draft-v6 §2.1b and change 9, a new "Where Phase
13 stands" section in CLAUDE.md recording what is built, deployed and still
missing, the module map row, and desktop-client-v1 §10b on the Settings tab and
where group settings live.

883 tests pass.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

4 files changed, 327 insertions, 1 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index f4dcca5..45aca7a 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -256,6 +256,13 @@ class NodeDaemon:
                     # Admission policy comes from node.toml, never from the hub:
                     # a hub that could declare a group open would be handed its key.
                     "join_policy": group_cfg.join_policy,
+                    # Whether ordinary members may upload. Read once here, into
+                    # the context, because the upload handler is synchronous and
+                    # a database round trip per chunk would be absurd. The
+                    # signed operation that changes it updates this dict in
+                    # place, so the two never drift within a run.
+                    "member_upload": await self._roster.member_upload_allowed(
+                        group_cfg.id) if self._roster else True,
                 }
 
             if not groups_ctx:
diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py
index 226b784..c811016 100644
--- a/packages/meshbay-node/src/meshbay_node/roster.py
+++ b/packages/meshbay-node/src/meshbay_node/roster.py
@@ -100,6 +100,26 @@ CREATE TABLE IF NOT EXISTS members (
     PRIMARY KEY (group_id, user_id)
 );
 
+-- Per-group settings the operator changes while the node runs.
+--
+-- Not node.toml: that file is hand-written, full of comments explaining
+-- decisions, and `ops.py` deliberately appends to it rather than round-tripping
+-- it through a TOML writer. A setting toggled from a panel has to take effect
+-- without an edit to the operator's file and without a restart, so it lives
+-- here, where the node already keeps what it decided rather than what it was
+-- configured with.
+--
+-- Absent means default. Nothing writes a row until someone changes something,
+-- so an existing node has the same behaviour it had before this table existed.
+CREATE TABLE IF NOT EXISTS group_settings (
+    group_id TEXT NOT NULL,
+    key      TEXT NOT NULL,
+    value    TEXT NOT NULL,
+    set_by   TEXT NOT NULL DEFAULT '',
+    set_at   TEXT NOT NULL DEFAULT '',
+    PRIMARY KEY (group_id, key)
+);
+
 CREATE TABLE IF NOT EXISTS invites (
     code_hash  TEXT PRIMARY KEY,
     group_id   TEXT NOT NULL,
@@ -518,6 +538,43 @@ class Roster:
 
     # ── Invites ──────────────────────────────────────────────────────────────
 
+    # ── Group settings ──────────────────────────────────────────────────────
+
+    # Whether members who are not the operator may upload. Default is yes: a
+    # group that nobody may add to is the unusual case, and an existing node
+    # must not change behaviour because a table was added under it.
+    SETTING_MEMBER_UPLOAD = "member_upload"
+
+    async def get_setting(self, group_id: str, key: str,
+                          default: str | None = None) -> str | None:
+        async with self._db.execute(
+                "SELECT value FROM group_settings WHERE group_id = ? AND key = ?",
+                (group_id, key)) as cur:
+            row = await cur.fetchone()
+        return row["value"] if row else default
+
+    async def set_setting(self, group_id: str, key: str, value: str,
+                          set_by: str = "") -> None:
+        await self._db.execute(
+            "INSERT INTO group_settings (group_id, key, value, set_by, set_at) "
+            "VALUES (?, ?, ?, ?, ?) "
+            "ON CONFLICT(group_id, key) DO UPDATE SET "
+            "value = excluded.value, set_by = excluded.set_by, "
+            "set_at = excluded.set_at",
+            (group_id, key, value, set_by, _now()))
+        await self._db.commit()
+
+    async def member_upload_allowed(self, group_id: str) -> bool:
+        """Whether an ordinary member may upload to this group."""
+        value = await self.get_setting(group_id, self.SETTING_MEMBER_UPLOAD, "1")
+        return value != "0"
+
+    async def set_member_upload(self, group_id: str, allowed: bool,
+                                set_by: str = "") -> bool:
+        await self.set_setting(group_id, self.SETTING_MEMBER_UPLOAD,
+                               "1" if allowed else "0", set_by)
+        return allowed
+
     async def create_invite(
         self,
         group_id: str,
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 9d16f82..22e5e15 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -62,6 +62,7 @@ from meshbay_common.adminop import (
     OP_MEMBER_REVOKE,
     OP_GEK_ROTATE,
     OP_MEMBER_UNPIN,
+    OP_MEMBER_UPLOAD,
     admin_transcript,
 )
 from meshbay_common.crypto import pk_to_b64, wrap_gek_aes
@@ -469,6 +470,8 @@ class WebRTCPeerSession:
                 self._spawn(self._do_device_list(msg))
             elif mtype == MNP.DEVICE_REVOKE:
                 self._spawn(self._do_device_revoke(msg))
+            elif mtype == MNP.MEMBER_UPLOAD:
+                self._do_member_upload(msg)
             elif mtype == MNP.MEMBER_UNPIN:
                 self._do_member_unpin(msg)
             elif mtype == MNP.GEK_ROTATE:
@@ -687,7 +690,11 @@ class WebRTCPeerSession:
             "proof": base64.b64encode(node_proof).decode(),
             "sig": base64.b64encode(
                 self._ctx["sk_node"].sign(node_transcript)).decode(),
-            "is_node_admin": bool(node_user_id and self._user_id == node_user_id),
+            "is_node_admin": self._is_node_admin(),
+            # So the interface knows whether to offer uploading at all. Not a
+            # permission — the node refuses regardless — but without it the
+            # only way to discover the answer is to try.
+            "member_upload": bool(self._group_ctx().get("member_upload", True)),
         }
         if node_user_id:
             ack["node_user_id"] = node_user_id
@@ -1568,6 +1575,58 @@ class WebRTCPeerSession:
         self._send({"type": MNP.MEMBER_UNPIN_ACK, "v": MNP_VERSION,
                     "user_id": user_id})
 
+    def _do_member_upload(self, msg: dict) -> None:
+        """
+        Turn uploading by ordinary members on or off, for this group.
+
+        Signed like every other operator action. The setting decides who may
+        write to the operator's disk, so a node that took it from an unsigned
+        message would let any member turn it back on for everyone — the control
+        would be a suggestion.
+        """
+        if "allowed" not in msg:
+            self._send({"type": "error", "detail": "Missing allowed"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        # The subject is what the operator is shown before signing, so it has to
+        # name the outcome rather than the operation.
+        self._issue_admin_challenge(
+            OP_MEMBER_UPLOAD, "on" if msg.get("allowed") else "off")
+
+    async def _admin_exec_member_upload(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        allowed = pending["subject"] == "on"
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"member_upload:{pending['subject']}")
+            return
+        roster = self._ctx.get("roster")
+        if roster is None:
+            self._send({"type": "error", "detail": "No roster on this node"})
+            return
+        await roster.set_member_upload(self._group_id or "", allowed,
+                                       set_by=self._user_id)
+        # Stored *and* applied. The upload path is synchronous and reads this
+        # dict; leaving it to the next restart would make the panel say one
+        # thing while the node did another.
+        self._group_ctx()["member_upload"] = allowed
+        self._audit("member_upload", pending["subject"])
+
+        # Everyone already connected is told, rather than finding out by having
+        # an upload refused. Enforcement does not depend on this reaching them —
+        # it is the node that refuses — but a button that stays visible until
+        # the next reconnection is a button people press.
+        notice = {"type": MNP.MEMBER_UPLOAD_ACK, "v": MNP_VERSION,
+                  "allowed": allowed}
+        for uid, session in list(self._peer_registry().items()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
     async def _run_op(self, fn, *args, **kwargs):
         """
         Call an operation from `meshbay_node.ops` with the daemon's own view.
@@ -1995,6 +2054,19 @@ class WebRTCPeerSession:
                         "filename": filename})
             return
 
+        # The operator can close uploading to everyone but themselves. Enforced
+        # here rather than by hiding a button: the button is a courtesy to the
+        # people who are not trying, and this is the part that holds against
+        # someone who is. `is_node_admin` is computed from the identity this
+        # node pinned, never from a hub claim.
+        if not ctx.get("member_upload", True) and not self._is_node_admin():
+            self._send({"type": "error",
+                        "detail": "Uploading is turned off for this group",
+                        "code": "member_upload_off",
+                        "filename": filename})
+            self._audit("upload_refused", filename[:64])
+            return
+
         roots: RootSet | None = ctx.get("roots")
         upload_root = roots.upload_root if roots else None
         if upload_root is None:
@@ -2189,6 +2261,17 @@ class WebRTCPeerSession:
         if ident:
             self._pinned_pk = ident["pk_ed25519"]
 
+    def _is_node_admin(self) -> bool:
+        """
+        Whether the peer on this connection is the node's operator.
+
+        Was written out twice — once in the handshake ack and once at the gate
+        below it — which is how the two come to disagree. From the node's own
+        record of who it belongs to, never from a hub claim.
+        """
+        node_user_id = self._ctx.get("node_user_id")
+        return bool(node_user_id and self._user_id == node_user_id)
+
     def _has_admin_authority(self) -> bool:
         """
         Cheap synchronous pre-check: is there anyone who could authorize this?
@@ -2269,6 +2352,9 @@ class WebRTCPeerSession:
         elif pending["op"] == OP_MEMBER_UNPIN:
             self._spawn(
                 self._admin_exec_member_unpin(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_MEMBER_UPLOAD:
+            self._spawn(
+                self._admin_exec_member_upload(pending, transcript, sig_bytes))
         else:
             self._send({"type": "error", "detail": "Unknown admin operation"})
 
diff --git a/packages/meshbay-node/tests/test_member_upload_policy.py b/packages/meshbay-node/tests/test_member_upload_policy.py
new file mode 100644
index 0000000..b1dc0cb
--- /dev/null
+++ b/packages/meshbay-node/tests/test_member_upload_policy.py
@@ -0,0 +1,176 @@
+"""
+The operator can close uploading to everyone but themselves.
+
+The point of these tests is the difference between a hidden button and a closed
+door. The interface stops offering the control, which is a courtesy to the
+people who are not trying; **the node refuses the upload**, which is the part
+that holds against someone who is. A member who kept an old tab open, or who
+speaks MNP directly, gets the same answer as everyone else.
+
+Two further things are worth holding:
+
+* the setting is changed by a **signed** operator instruction. A node that took
+  it from an unsigned message would let any member turn it back on, and the
+  control would be a suggestion;
+* it is stored on the **node**, not the hub. A hub that could decide who may
+  write to the operator's disk is a hub with authority over the node, which is
+  the thing this whole design is arranged to avoid.
+"""
+
+import base64
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.adminop import OP_MEMBER_UPLOAD
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roster import Roster
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+from conftest import one_root
+
+pytestmark = pytest.mark.asyncio
+
+
+def _session(tmp_path: Path, user_id: str, *, member_upload: bool,
+             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": one_root(shared_root),
+        "index": index,
+        "sk_node": index.sk_node,
+        "member_upload": member_upload,
+        "node_user_id": operator,
+    }
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = ctx
+    session._group_id = None
+    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, "chunk_index": 0, "total_chunks": 1,
+        "data": base64.b64encode(body).decode(),
+    })
+
+
+def _uploads_dir(session) -> Path:
+    return session._ctx["roots"].upload_root.path / "uploads"
+
+
+# ── The door, not the button ────────────────────────────────────────────────
+
+async def test_a_member_cannot_upload_when_it_is_turned_off(tmp_path):
+    session = _session(tmp_path, "member-1", member_upload=False,
+                       operator="the-operator")
+    _upload(session)
+
+    assert not (_uploads_dir(session) / "clip.mp4").exists(), (
+        "the file was written even though uploading is off — the setting is "
+        "decorative and the hidden button was the whole control")
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "member_upload_off"
+
+
+async def test_the_operator_can_still_upload(tmp_path):
+    """Otherwise turning it off locks the operator out of their own node, and
+    the only way back is a config file and a restart."""
+    session = _session(tmp_path, "the-operator", member_upload=False,
+                       operator="the-operator")
+    _upload(session)
+
+    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
+
+
+async def test_members_upload_normally_when_it_is_on(tmp_path):
+    session = _session(tmp_path, "member-1", member_upload=True,
+                       operator="the-operator")
+    _upload(session)
+
+    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
+
+
+async def test_a_node_that_never_heard_of_the_setting_still_accepts_uploads(tmp_path):
+    """An existing node's context has no such key. The absence must read as
+    "allowed", or upgrading the node silently closes every group."""
+    session = _session(tmp_path, "member-1", member_upload=True,
+                       operator="the-operator")
+    del session._ctx["member_upload"]
+    _upload(session)
+
+    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
+
+
+# ── Who may change it ───────────────────────────────────────────────────────
+
+async def test_changing_it_needs_a_signature(tmp_path):
+    """
+    The request only ever produces a challenge. Nothing is applied until a
+    signature over the transcript verifies — the same path as removing a member.
+    """
+    session = _session(tmp_path, "member-1", member_upload=True,
+                       operator="the-operator")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_member_upload({"allowed": False})
+
+    assert issued == [(OP_MEMBER_UPLOAD, "off")]
+    assert session._ctx["member_upload"] 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. "member_upload" tells
+    them nothing; "off" tells them what they are about to do."""
+    session = _session(tmp_path, "op", member_upload=False, operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_member_upload({"allowed": True})
+
+    assert issued == [(OP_MEMBER_UPLOAD, "on")]
+
+
+async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
+    session = _session(tmp_path, "member-1", member_upload=True,
+                       operator="the-operator")
+    session._has_admin_authority = lambda: False
+
+    session._do_member_upload({"allowed": False})
+
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+# ── Where it is stored ──────────────────────────────────────────────────────
+
+async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        assert await roster.member_upload_allowed("g1") is True, (
+            "absent must mean allowed, or an upgrade closes every group")
+        await roster.set_member_upload("g1", False, set_by="op")
+        assert await roster.member_upload_allowed("g1") is False
+    finally:
+        await roster.close()
+
+    reopened = Roster(db_path=tmp_path / "roster.db")
+    await reopened.open()
+    try:
+        assert await reopened.member_upload_allowed("g1") is False
+        assert await reopened.member_upload_allowed("g2") is True, (
+            "one group's setting must not answer for another")
+    finally:
+        await reopened.close()