1 files changed, 120 insertions, 0 deletions
diff --git a/packages/meshbay-hub/tests/test_node_ws_auth.py b/packages/meshbay-hub/tests/test_node_ws_auth.py
index 1391722..4ead0d7 100644
--- a/packages/meshbay-hub/tests/test_node_ws_auth.py
+++ b/packages/meshbay-hub/tests/test_node_ws_auth.py
@@ -344,3 +344,123 @@ async def test_announce_with_valid_proof_succeeds_and_is_idempotent(client):
     assert second.status_code == 201, second.text
     assert second.json()["node_id"] == first.json()["node_id"], (
         "re-announcing the same key must not create a second node record (M8)")
+
+
+# ── An empty claim is not a claim on everything ──────────────────────────────
+#
+# 2026-09-11, found on a live deployment. `_claimable` used to read
+# `set(claimed_groups or authorized)`, and the node omits `group_ids` entirely
+# when it hosts nothing — so "I host no groups" was read as "I host all of
+# yours". The node cannot serve any of them (no GEK, and its own handshake
+# refuses them), but `/v1/groups/{id}/nodes` lists nodes in registration order,
+# so whenever such a node won the reconnection race after a hub restart it
+# became `nodes[0]` and the group stopped opening for every member.
+
+
+@pytest.mark.asyncio
+async def test_ws_absent_claim_registers_no_groups(client):
+    """A node that declares nothing hosts nothing — it must not inherit the set."""
+    from meshbay_hub.api.revocation import _authorize_node_ws
+
+    user = await _make_user(client, "empty1")
+    node_id = await _announce_node(client, user)
+    for name in ("has-one", "has-two"):
+        r = await client.post(
+            "/v1/groups",
+            json={"name": name, "visibility": "private", "join_policy": "invite"},
+            headers={"Authorization": f"Bearer {user['token']}"},
+        )
+        assert r.status_code == 201, r.text
+
+    resolved, groups = await _authorize_node_ws(_node_token(user), node_id, None)
+    assert resolved == node_id
+    assert groups == [], (
+        "a node hosting nothing was registered as a host for its owner's groups")
+
+
+@pytest.mark.asyncio
+async def test_ws_explicit_empty_claim_registers_no_groups(client):
+    """And the same when the node says so out loud, which it now does."""
+    from meshbay_hub.api.revocation import _authorize_node_ws
+
+    user = await _make_user(client, "empty2")
+    node_id = await _announce_node(client, user)
+    r = await client.post(
+        "/v1/groups",
+        json={"name": "lonely", "visibility": "private", "join_policy": "invite"},
+        headers={"Authorization": f"Bearer {user['token']}"},
+    )
+    assert r.status_code == 201, r.text
+
+    resolved, groups = await _authorize_node_ws(_node_token(user), node_id, [])
+    assert resolved == node_id
+    assert groups == []
+
+
+@pytest.mark.asyncio
+async def test_empty_node_cannot_shadow_another_members_group(client):
+    """
+    The outage itself: two members, one group, and only one of them hosts it.
+    The other's node — running, configured with nothing — must not appear as a
+    source for that group, because it is the one clients would reach first.
+    """
+    from meshbay_hub.api.revocation import (
+        _authorize_node_ws, _node_groups, get_online_nodes_for_group)
+
+    host = await _make_user(client, "hoster")
+    guest = await _make_user(client, "guest")
+    host_node = await _announce_node(client, host)
+    guest_node = await _announce_node(client, guest)
+
+    r = await client.post(
+        "/v1/groups",
+        json={"name": "shared", "visibility": "private", "join_policy": "invite"},
+        headers={"Authorization": f"Bearer {host['token']}"},
+    )
+    assert r.status_code == 201, r.text
+    group_id = r.json()["group_id"]
+
+    r = await client.post(
+        f"/v1/groups/{group_id}/members/{'guest'}",
+        headers={"Authorization": f"Bearer {host['token']}"},
+    )
+    assert r.status_code == 201, r.text
+
+    # The guest's node registers first — the order that made this fatal.
+    _, guest_groups = await _authorize_node_ws(_node_token(guest), guest_node, None)
+    _, host_groups = await _authorize_node_ws(
+        _node_token(host), host_node, [group_id])
+    _node_groups[guest_node] = guest_groups
+    _node_groups[host_node] = host_groups
+    try:
+        assert get_online_nodes_for_group(group_id) == [host_node], (
+            "a member's empty node shadowed the node actually hosting the group")
+    finally:
+        _node_groups.pop(guest_node, None)
+        _node_groups.pop(host_node, None)
+
+
+@pytest.mark.asyncio
+async def test_update_groups_is_held_to_the_same_ceiling(client):
+    """
+    `update_groups` assigned the message's list verbatim, so the C2 ceiling held
+    at authentication could be stepped over one message later: a node had only
+    to reload to claim any group on the hub. It now goes through the same gate,
+    which is what this asserts — the socket loop itself needs a WebSocket the
+    ASGI harness has not got (see this module's docstring).
+    """
+    from meshbay_hub.api.revocation import _authorized_groups, _claimable
+
+    user = await _make_user(client, "reloader")
+    r = await client.post(
+        "/v1/groups",
+        json={"name": "owned", "visibility": "private", "join_policy": "invite"},
+        headers={"Authorization": f"Bearer {user['token']}"},
+    )
+    assert r.status_code == 201, r.text
+    own = r.json()["group_id"]
+
+    authorized = await _authorized_groups(user["user_id"])
+    assert _claimable([own, "someone-elses-group"], authorized) == [own]
+    assert _claimable([], authorized) == []
+    assert _claimable(None, authorized) == []