path: root/packages/meshbay-hub/src/meshbay_hub/api/signaling.py

"""
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 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
MAX_PENDING_PER_USER = 3           # concurrent in-flight offers per account

_pending_per_user: dict[str, int] = {}


@router.post("/{node_id}/webrtc/offer", response_model=WebRTCOfferResponse)
@limiter.limit("30/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}")

    if _pending_per_user.get(current_user.id, 0) >= MAX_PENDING_PER_USER:
        raise HTTPException(status_code=429, detail="Too many pending connections")

    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)