fix: bound what one member can cost the others

An availability review, prompted by the group claim above: a participant
supplies input — who else bears the cost? Six answers where the cost fell on
someone other than the sender, and none of them needs an attacker.

  AV3  `chat_notify` carried a `group_id` the hub believed, so any connected
       node could write a notification to every member of any group on the
       hub, carrying a display string of its choosing, with its account
       having no relation to that group. This is the group claim again, two
       hundred lines further down the same socket. Gated on what the node is
       registered for, and metered: the fan-out is one write per member. The
       budget expires by time rather than on disconnect, or reconnecting
       would refill it and a node token is good for an hour.

  AV4  A swarm source named its own `endpoint` as free text documented as
       "ip:port", so an account could publish a third party's address — H6's
       `peer_ip` defect, never applied here. Nothing dials a swarm source
       today, which is the only reason it was not already a reflection
       primitive. It is a transport and a port now, never a host, and the
       number of hashes one account may claim is bounded: rows were keyed
       (hash, account) with no cap at all.

  AV5  `handle_webrtc_answer` resolved any pending `peer_id` from any node's
       socket. The answer is the SDP a browser then connects to. That this
       had not happened rested on a uuid4 being unguessable.

  AV6  `relay_register` had no authentication of any kind: it compared
       `pk_relay` against the approved value, which is a *public* key, so
       anyone who could read it could rewrite where the hub tells nodes to
       send relayed traffic. The module docstring promised signed JWTs and
       `jwt` was imported and never used.

  AV7  The node held unlimited peer connections and kept one that never
       completed a handshake for the life of the daemon. H6 bounded what one
       unauthenticated peer costs; the hub's cap is three offers in flight
       per *account*, a limit on each caller and not on the machine, so an
       operator's exposure grew with the size of their groups.

  AV8  `invite-notify` put a request-supplied `group_name` into the subject
       of an email the hub sends under its own domain, to any account, with
       no rate limit. The name comes from the group row now.

The tests are two accounts each, in one file that says why: a one-member test
proves a one-member property, and every finding here needed a second person
to exist at all. Each was checked against the unfixed code. Two did not
survive that check and were rewritten — one re-enacted the disconnect path
instead of running it (hence `forget_node`), the other called the reaper
itself and would have passed with the call removed from `handle_offer`.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01T4YmK41VsEURWFdop4EEeT

1 files changed, 72 insertions, 5 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
index 7c37a1e..0e274f9 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
@@ -68,6 +68,53 @@ def get_online_nodes_for_group(group_id: str) -> list[str]:
     return [nid for nid, gids in _node_groups.items() if group_id in gids]
 
 
+# A chat_notify costs one query per member of the group plus a write for each,
+# and nothing on the node's side paces it. Without a budget, one node can keep
+# the hub's database busy on behalf of a group it belongs to — a cost borne by
+# every other group on the instance. Generous enough that a lively conversation
+# never meets it: notifications aggregate to one row per person per group, so
+# the useful rate is far below this.
+NOTIFY_BURST = 30            # messages
+NOTIFY_WINDOW_SECONDS = 60
+
+_notify_window: dict[str, tuple[float, int]] = {}   # node_id → (window start, count)
+
+
+def forget_node(node_id: str) -> None:
+    """Drop everything a disconnected node's socket owned.
+
+    One function rather than three lines in a `finally`, so that what a
+    disconnect does can be asserted by running it instead of by re-enacting it
+    — a test that re-enacts cleanup tests its own re-enactment, and would not
+    have noticed `_notify_window` being added here.
+
+    Which is the point: `_notify_window` is **not** cleared. Dropping it would
+    make reconnecting the way to refill the budget, and a node's token stays
+    valid for an hour. It expires by time, in `_notify_budget`.
+    """
+    _connected_nodes.pop(node_id, None)
+    _node_groups.pop(node_id, None)
+
+
+def _notify_budget(node_id: str) -> bool:
+    """True if this node may send one more chat_notify now."""
+    now = time.monotonic()
+    if len(_notify_window) > 1000:
+        # Swept here rather than on disconnect, which would let a node refill
+        # its budget by reconnecting — the same token stays valid for an hour.
+        for nid, (started, _) in list(_notify_window.items()):
+            if now - started >= NOTIFY_WINDOW_SECONDS:
+                _notify_window.pop(nid, None)
+    start, count = _notify_window.get(node_id, (now, 0))
+    if now - start >= NOTIFY_WINDOW_SECONDS:
+        start, count = now, 0
+    if count >= NOTIFY_BURST:
+        _notify_window[node_id] = (start, count)
+        return False
+    _notify_window[node_id] = (start, count + 1)
+    return True
+
+
 
 async def _mark_hosted(group_ids: list[str]) -> None:
     """Stamp the first time a node announced it hosts each of these groups.
@@ -133,10 +180,27 @@ def _sign_revocation(target: str, target_id: str, reason: str) -> str:
 
 # ── WebSocket endpoint ────────────────────────────────────────────────────────
 
-async def _handle_chat_notify(group_id: str, sender_name: str, sender_user_id: str) -> None:
-    """Node informs hub that a chat message was posted — create notifications for offline members."""
+async def _handle_chat_notify(group_id: str, sender_name: str, sender_user_id: str,
+                              *, node_id: str) -> None:
+    """Node informs hub that a chat message was posted — create notifications for offline members.
+
+    `group_id` arrives in the node's own message and is checked against what
+    that node is registered for. Without the check, any connected node could
+    write a notification to every member of **any** group on the hub, carrying
+    a display string of its choosing, with its account having no relation to
+    that group at all. Same shape as the empty group claim: something believed
+    about a group the sender has nothing to do with.
+
+    `node_id` is keyword-**required** rather than defaulted. A default here
+    would mean "unchecked when the caller forgets", which is the failure this
+    whole review is about.
+    """
     if not group_id:
         return
+    if group_id not in _node_groups.get(node_id, ()):
+        log.warning("Node %s sent chat_notify for a group it does not host",
+                    (node_id or "?")[:8])
+        return
     try:
         from meshbay_hub.db.engine import get_session_factory
         from meshbay_hub.db.models import GroupMember, Group
@@ -318,7 +382,7 @@ async def node_websocket(ws: WebSocket):
                     event.set()
             elif msg.get("type") == "webrtc_answer":
                 from meshbay_hub.api.signaling import handle_webrtc_answer
-                handle_webrtc_answer(msg)
+                handle_webrtc_answer(msg, node_id)
             elif msg.get("type") == "update_groups":
                 # Through the same gate as the registration above. This used to
                 # assign the message's list verbatim, so the ceiling that makes
@@ -330,6 +394,9 @@ async def node_websocket(ws: WebSocket):
                 await _mark_hosted(new_gids)
                 log.info("Node %s updated groups: %d", node_id[:8], len(new_gids))
             elif msg.get("type") == "chat_notify":
+                if not _notify_budget(node_id):
+                    log.warning("Node %s exceeded its chat_notify rate", node_id[:8])
+                    continue
                 asyncio.ensure_future(_handle_chat_notify(
                     msg.get("group_id", ""),
                     msg.get("sender_name", ""),
@@ -339,6 +406,7 @@ async def node_websocket(ws: WebSocket):
                     # that lied here could only suppress one notification, which
                     # is the same power it has by not sending the message at all.
                     msg.get("sender_user_id", ""),
+                    node_id=node_id,
                 ))
 
     except WebSocketDisconnect:
@@ -347,8 +415,7 @@ async def node_websocket(ws: WebSocket):
         log.error("Node WS error: %s", e)
     finally:
         if node_id:
-            _connected_nodes.pop(node_id, None)
-            _node_groups.pop(node_id, None)
+            forget_node(node_id)
 
 
 # ── Admin revocation endpoint ─────────────────────────────────────────────────