1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
"""
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.api.deps import get_current_user
from meshbay_hub.api.middleware import limiter
from meshbay_hub.db.engine import get_db
from meshbay_hub.api.netutil import client_ip
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] = {}
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).
node_group_ids = set(_node_groups.get(node_id, []))
if node_group_ids:
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 = await db.execute(
select(Group.id).where(
Group.id.in_(node_group_ids),
Group.join_policy == "open",
Group.status == "active",
))
if not has_open.first():
raise HTTPException(status_code=403, detail="Not a member of any group on this node")
else:
active = await db.execute(
select(Group.id).where(Group.id.in_(shared), Group.status == "active"))
if not active.first():
raise HTTPException(status_code=403, detail="Group is not active")
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
_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 asyncio.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)
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) -> None:
"""Called from the node WebSocket message loop when a webrtc_answer arrives."""
peer_id = msg.get("peer_id")
if not peer_id:
log.warning("webrtc_answer without peer_id")
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)
|