1 files changed, 121 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py
index 988aa46..fa55ff6 100644
--- a/packages/meshbay-node/tests/test_security_regressions.py
+++ b/packages/meshbay-node/tests/test_security_regressions.py
@@ -837,3 +837,124 @@ def test_node_control_api_serves_no_html():
     for gone in ("_render_page", "_render_audit_page", "_render_roster",
                  "_AUDIT_HTML"):
         assert not hasattr(ui_app, gone), f"{gone} came back — HTML surface"
+
+
+# ── NS4 / M3: the node's own controls take no authority from the hub ─────────
+
+async def _owner_session(tmp_path, roster, *, device: str = "",
+                         confirmed: bool = False):
+    """A session whose token says it is the account this node belongs to.
+
+    Which is all a hub can decide: `_user_id` is the `sub` of a JWT it issued,
+    so this is what an active hub forging a token arrives holding. Whether the
+    *device* on the connection is one the node pinned as an operator is the
+    other half, and no token can assert it.
+    """
+    from meshbay_node.indexer.group_index import GroupIndex
+
+    shared = tmp_path / "shared"
+    shared.mkdir(exist_ok=True)
+    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = {
+        "roots": one_root(shared), "index": index, "sk_node": index.sk_node,
+        "gek": generate_gek(), "roster": roster,
+        "node_user_id": "the-owner",
+        "daemon_state": {"groups_ctx": {}, "reload_fn": None},
+    }
+    session._group_id = "g" * 32
+    session._user_id = "the-owner"
+    session._username = "the-owner"
+    session._pinned_pk = device
+    session._device_confirmed = confirmed
+    session._pk_user = ""
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+@pytest.mark.asyncio
+async def test_a_token_naming_the_owner_is_not_node_authority(tmp_path):
+    """
+    The hub holds no user keys, so it cannot countersign a device — but it does
+    choose what a token says. Six node-wide controls used to be gated on the
+    account id alone, which is the hub's to decide: `node_status` (every group
+    on the machine, with the operator's absolute paths), `node_settings_set`,
+    `roster_read`, `denylist_read`, `denylist_clear` (the persisted revocation
+    H4 exists to keep) and `node_reload`.
+
+    An active hub reaches a completed handshake wherever it can also obtain the
+    group key, which §3.5 concedes it can in an open-join group. From there,
+    "the hub says you are the owner" was the whole of the check.
+    """
+    from meshbay_node.roster import Roster
+
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        forged = await _owner_session(tmp_path, roster)
+        await forged._do_node_status({})
+        assert forged.sent[-1]["type"] == "error"
+        assert forged.sent[-1]["code"] == "not_operator"
+
+        forged.sent.clear()
+        await forged._do_denylist_read({})
+        assert forged.sent[-1]["type"] == "error"
+
+        forged.sent.clear()
+        await forged._do_denylist_clear({"subject": ""})
+        assert forged.sent[-1]["type"] == "error", (
+            "clearing the denylist undoes a revocation every node enforces")
+    finally:
+        await roster.close()
+
+
+@pytest.mark.asyncio
+async def test_an_identified_device_that_is_not_an_operator_is_refused(tmp_path):
+    """Being a pinned member of the group is not being the node's operator.
+
+    The device proof is real here; what it proves is an ordinary member's key,
+    and `operator_pks()` is rebuilt from the roster on every call so a revoked
+    one stops working at once.
+    """
+    from meshbay_node.roster import Roster
+    from meshbay_common.crypto import pk_to_b64
+
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        sk = Ed25519PrivateKey.generate()
+        member_pk = pk_to_b64(sk.public_key())
+        await roster.pin_identity("the-owner", "the-owner", member_pk,
+                                  member_pk, "code")
+        session = await _owner_session(tmp_path, roster, device=member_pk,
+                                       confirmed=True)
+        await session._do_node_status({})
+        assert session.sent[-1]["type"] == "error"
+    finally:
+        await roster.close()
+
+
+@pytest.mark.asyncio
+async def test_a_paired_operator_device_is_what_opens_it(tmp_path):
+    """The positive case, so the test above is about authority and not about
+    everything being refused."""
+    from meshbay_node.roster import Roster
+    from meshbay_common.join import ROLE_OPERATOR
+    from meshbay_common.crypto import pk_to_b64
+
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        sk = Ed25519PrivateKey.generate()
+        pk = pk_to_b64(sk.public_key())
+        await roster.pin_identity("the-owner", "the-owner", pk, pk, "code")
+        await roster.set_member("", "the-owner", ROLE_OPERATOR, "active",
+                                "local-cli")
+        session = await _owner_session(tmp_path, roster, device=pk,
+                                       confirmed=True)
+        await session._do_denylist_read({})
+        assert session.sent[-1]["type"] != "error", session.sent[-1]
+    finally:
+        await roster.close()