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
|
"""The data channel itself: framing, the reply correlation id, the peer's
address and DTLS fingerprint."""
import contextvars
import struct
import msgpack
from aiortc import RTCPeerConnection
from meshbay_node.transport.webrtc.limits import MAX_MSG
def _extract_dtls_fingerprint(sdp: str) -> bytes:
"""Extract the DTLS SHA-256 fingerprint from SDP as raw 32 bytes."""
for line in sdp.splitlines():
if line.startswith("a=fingerprint:sha-256 "):
hex_str = line.split(" ", 1)[1].replace(":", "")
return bytes.fromhex(hex_str)
return b""
def _pack(obj: dict) -> bytes:
data = msgpack.packb(obj, use_bin_type=True)
return struct.pack(">I", len(data)) + data
# The request this session is currently answering, as (session, req_id).
#
# MNP has never carried a correlation id: a reply named its own type and
# nothing else, so a client with more than one request outstanding had to guess
# which one a message answered — by arrival order, for every reply the client
# could not key off a field of its own. The guess is wrong whenever two replies
# reorder, and catastrophically wrong for the replies that name *nothing*: this
# module sends `{"type": "error"}` from 240 places and two of them name what
# they are about. A refusal therefore reached no caller at all, and the request
# it belonged to waited out the client's 30s timeout while some unrelated
# request was resolved with the refusal instead. Live symptom, found 2026-09-06:
# the Chat composer is disabled while a send is in flight, so a chat message
# whose reply went astray froze the tab for 30 seconds.
#
# `req_id` closes it: whatever the caller put on the request is stamped on the
# reply. A ContextVar rather than a parameter because the alternative is
# threading an argument through all 240 send sites — and asyncio copies the
# current context into a task, so a handler that `_spawn`s its real work still
# answers under the id of the request that started it.
#
# The session is held alongside the id because a handler may send to *other*
# sessions as well as its own (a chat broadcast, an index push): those are not
# replies to anything and must not be stamped. _send checks the owner.
_REPLY_TO: contextvars.ContextVar[tuple] = contextvars.ContextVar(
"meshbay_reply_to", default=(None, None))
class _DataChannelBuffer:
"""
Accumulate DataChannel messages and extract length-prefixed msgpack.
Finding H6: the limit was a flat 64 MB applied even before the handshake, so an
unauthenticated peer could announce a 64 MB frame and dribble bytes into it,
holding that much memory per connection. Until a peer has proved GEK
possession it gets a small budget; the large one is for file uploads.
"""
def __init__(self, max_message: int = MAX_MSG):
self._buf = bytearray()
self.max_message = max_message
def feed(self, data: bytes):
self._buf.extend(data)
def messages(self):
while len(self._buf) >= 4:
length = struct.unpack(">I", self._buf[:4])[0]
if length > self.max_message:
raise ValueError(f"Message too large: {length}")
if len(self._buf) < 4 + length:
break
msg_bytes = bytes(self._buf[4:4 + length])
del self._buf[:4 + length]
yield msgpack.unpackb(msg_bytes, raw=False)
def _get_remote_ip(pc: RTCPeerConnection) -> str:
"""Best-effort extraction of the remote peer IP from the ICE transport."""
try:
dtls = pc.sctp and pc.sctp.transport
ice = dtls and dtls.transport
conn = ice and ice._connection
if conn and hasattr(conn, '_nominated') and conn._nominated:
for pair in conn._nominated.values():
return pair.remote_candidate.host
if conn and conn.remote_candidates:
return conn.remote_candidates[0].host
except Exception:
pass
return ""
|