""" WebRTC signaling — relay SDP/ICE between browser and node. The hub NEVER touches content. This is pure signaling: < 1 KB per message, stateless relay. After the SDP exchange completes, the browser and node communicate P2P via WebRTC DataChannel — hub is out of the loop. Flow: Browser → Hub : POST /v1/nodes/{node_id}/webrtc/offer {sdp, ice_candidates} Hub → Node : WS push {type: "webrtc_offer", sdp, ice_candidates, peer_id} Node → Hub : WS reply {type: "webrtc_answer", sdp, ice_candidates, peer_id} Hub → Browser : HTTP response {sdp, ice_candidates} """ import asyncio import json import logging import math import time import uuid from fastapi import APIRouter, Depends, HTTPException, Request from pydantic import BaseModel from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from meshbay_hub import hub_settings from meshbay_hub.api.deps import get_current_user from meshbay_hub.api.middleware import limiter from meshbay_hub.api.netutil import client_ip from meshbay_hub.db.engine import get_db from meshbay_hub.db.models import Group, GroupMember, IPLog, User log = logging.getLogger(__name__) router = APIRouter(prefix="/v1/nodes", tags=["signaling"]) _webrtc_answers: dict[str, asyncio.Future] = {} # peer_id → the node the offer was relayed to. An answer is only accepted # from that node (see handle_webrtc_answer). _answer_owner: dict[str, str] = {} class WebRTCOfferRequest(BaseModel): sdp: str ice_candidates: list[dict] = [] class WebRTCOfferResponse(BaseModel): sdp: str ice_candidates: list[dict] = [] peer_id: str MAX_SDP_BYTES = 16 * 1024 # an SDP offer is ~2 KB # Offers of one account waiting for their node's answer, across all its devices # and all nodes. An offer is pending for the node's round trip — well under a # second — or up to the 15 s answer timeout when the node is connected and # silent. Sized for an account with twenty groups on three devices: Search # negotiates six at once per page (search-page.js), so three pages are eighteen, # plus a phone's pooled connections all reconnecting as it wakes. It was 3, which # is exactly what one Search page dialled at once, so the first tile or the first # reconnection beside a sweep was refused and its group reported unreachable. MAX_PENDING_PER_USER = 32 # Offers one account may send one node: a burst of OFFER_BURST, refilled at # OFFER_REFILL_PER_S. This is what bounds a member's cost to a node (H6) — each # offer makes it allocate a peer connection — and it is counted per account # rather than per address, because a 4G carrier puts many subscribers behind one # IPv4 address and hands each a whole IPv6 /64. The burst covers twenty groups # hosted on one node, reloaded several times over on more than one device; the # refill is two a second, sustained. The node bounds its own total separately # (MAX_PEER_SESSIONS in webrtc_server.py). OFFER_BURST = 120 OFFER_REFILL_PER_S = 2.0 _pending_per_user: dict[str, int] = {} # (user id, node id) -> (tokens left, when they were counted) _offer_buckets: dict[tuple[str, str], tuple[float, float]] = {} _OFFER_BUCKETS_PRUNE_AT = 4096 def _take_offer(user_id: str, node_id: str, now: float) -> float | None: """Spend one of this account's offers to this node. None when it may go ahead, otherwise how many seconds until it may. """ if len(_offer_buckets) > _OFFER_BUCKETS_PRUNE_AT: # A bucket that would have refilled completely carries no information. full_after = OFFER_BURST / OFFER_REFILL_PER_S for key, (_, at) in list(_offer_buckets.items()): if now - at >= full_after: del _offer_buckets[key] key = (user_id, node_id) tokens, at = _offer_buckets.get(key, (float(OFFER_BURST), now)) tokens = min(float(OFFER_BURST), tokens + (now - at) * OFFER_REFILL_PER_S) if tokens < 1.0: _offer_buckets[key] = (tokens, now) return (1.0 - tokens) / OFFER_REFILL_PER_S _offer_buckets[key] = (tokens - 1.0, now) return None @router.post("/{node_id}/webrtc/offer", response_model=WebRTCOfferResponse) # Per address and per node, and only a coarse guard in front of authentication: # the account's budget above is the limit that means something. 600 because an # IPv4 address on a mobile network is shared by many subscribers. @limiter.limit("600/minute") async def webrtc_offer( node_id: str, body: WebRTCOfferRequest, request: Request, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """ Browser sends WebRTC SDP offer for a node. Hub relays via WebSocket. Returns the node's SDP answer once received. Finding H6: this was reachable by any authenticated user, for any node, with no rate limit and no membership check. Each call makes the node allocate an aiortc RTCPeerConnection and gather ICE, so it was a remote resource-exhaustion primitive against an arbitrary third party's machine. Finding H4: it also ignored group status, so "suspend a group" did not stop new connections from being brokered to nodes hosting it. """ from meshbay_hub.api.revocation import _connected_nodes, _node_groups if len(body.sdp) > MAX_SDP_BYTES: raise HTTPException(status_code=413, detail="SDP too large") # Logged here because this is the moment a browser starts a peer connection, # and the address it starts it from is this one — the hub's own view of the # TCP connection. Whatever address the peers then discover through STUN is # theirs to negotiate and is not what a log should record. db.add(IPLog(user_id=current_user.id, event="webrtc_offer", ip_address=client_ip(request), detail=node_id[:8])) await db.commit() ws = _connected_nodes.get(node_id) if not ws: raise HTTPException(status_code=404, detail="Node not connected") # The caller must share at least one active group with the target node, # OR the node must host at least one open-join group (public groups admit # anyone — the node's MNP handshake handles authorization). # # That second path is exactly what "public groups" means, so it is gated by # the instance switch: with public groups off, a non-member is not brokered a # connection to a node just because it happens to host an open group. Members # of that group are unaffected — they match `shared` below. node_group_ids = set(_node_groups.get(node_id, [])) # A node registered for no group shares no group with anybody, which is this # check's own answer — and `if node_group_ids:` used to skip the whole thing, # membership, group status and the public-group gate together. Since AV1 made # an empty claim mean "no groups" rather than "all of my owner's", that is # the *normal* registration of a node hosting nothing: exactly the # unconfigured node left running that took a group down on 2026-09-11. So the # machine least able to defend itself was the one any authenticated account # could make allocate a peer connection and gather ICE, which is H6 restored # in the one case AV1 made common. # # Nothing legitimate is lost by refusing here: a browser cannot complete a # handshake with such a node anyway — `group_id` is mandatory (M1) and a node # holding no group key refuses outright (NS8) — so this only declines work # the node would decline one step later, at its own expense. if not node_group_ids: raise HTTPException(status_code=403, detail="Not a member of any group on this node") result = await db.execute( select(GroupMember.group_id).where( GroupMember.user_id == current_user.id, GroupMember.group_id.in_(node_group_ids), )) shared = [gid for (gid,) in result.all()] if not shared: has_open = None if await hub_settings.public_groups_allowed(db): has_open = (await db.execute( select(Group.id).where( Group.id.in_(node_group_ids), Group.join_policy == "open", Group.status == "active", ))).first() if not has_open: raise HTTPException(status_code=403, detail="Not a member of any group on this node") else: statuses = set((await db.execute( select(Group.status).where(Group.id.in_(shared)))).scalars().all()) if "active" not in statuses: # Report the strongest state present — "revoked" is the signed, # node-enforced one; "suspended" is the reversible hub flag. state = "revoked" if "revoked" in statuses else next(iter(statuses), "suspended") raise HTTPException(status_code=403, detail=f"Group is {state}") # Both refusals say when to come back, and transport.js does: a 429 here is # the hub being busy, never the node being down. # The pending check comes first so that an offer refused by it does not # spend the node's budget: retries of a busy moment would otherwise drain it. if _pending_per_user.get(current_user.id, 0) >= MAX_PENDING_PER_USER: raise HTTPException(status_code=429, detail="Too many pending connections", headers={"Retry-After": "1"}) wait = _take_offer(current_user.id, node_id, time.monotonic()) if wait is not None: raise HTTPException(status_code=429, detail="Too many connections to this node", headers={"Retry-After": str(max(1, math.ceil(wait)))}) peer_id = str(uuid.uuid4()) answer_future: asyncio.Future = asyncio.get_event_loop().create_future() _webrtc_answers[peer_id] = answer_future _answer_owner[peer_id] = node_id _pending_per_user[current_user.id] = _pending_per_user.get(current_user.id, 0) + 1 try: await ws.send_text(json.dumps({ "type": "webrtc_offer", "peer_id": peer_id, "user_id": current_user.id, "sdp": body.sdp, "ice_candidates": body.ice_candidates, })) try: answer = await asyncio.wait_for(answer_future, timeout=15.0) except TimeoutError: raise HTTPException( status_code=504, detail="Node did not respond with WebRTC answer") return WebRTCOfferResponse( sdp=answer["sdp"], ice_candidates=answer.get("ice_candidates", []), peer_id=peer_id, ) finally: _webrtc_answers.pop(peer_id, None) _answer_owner.pop(peer_id, None) remaining = _pending_per_user.get(current_user.id, 1) - 1 if remaining > 0: _pending_per_user[current_user.id] = remaining else: _pending_per_user.pop(current_user.id, None) def handle_webrtc_answer(msg: dict, node_id: str) -> None: """Called from the node WebSocket message loop when a webrtc_answer arrives. `node_id` is the socket this arrived on, and the answer is accepted only for a `peer_id` the hub issued to **that** node. The answer carries the SDP the browser then connects to, so without the check any connected node could resolve any pending offer and stand in for the node the client asked for. That it had not happened rested on a uuid4 being unguessable, which is a reason it was hard, not a reason it was refused. """ peer_id = msg.get("peer_id") if not peer_id: log.warning("webrtc_answer without peer_id") return if _answer_owner.get(peer_id) != node_id: log.warning("Node %s answered an offer it was never sent (peer=%s)", (node_id or "?")[:8], str(peer_id)[:8]) return future = _webrtc_answers.get(peer_id) if future and not future.done(): future.set_result({ "sdp": msg.get("sdp", ""), "ice_candidates": msg.get("ice_candidates", []), }) else: log.warning("webrtc_answer for unknown peer_id: %s", peer_id)