diff options
Diffstat (limited to 'packages/meshbay-hub/src/meshbay_hub/api/revocation.py')
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/api/revocation.py | 77 |
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 ───────────────────────────────────────────────── |