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
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
|
"""A connection process: peer connections held for the node, and nothing else.
The WebRTC stack (ICE, DTLS, SCTP) is Python and saturates one core at about
35 MB/s summed over every connection (docs/MESHBAY_DESIGN.md §5.5). This process
runs that stack and only that. It holds no node key, no group key and no index:
what crosses it is what crosses the wire, bytes of the data channel in both
directions, and the node proper decides what they mean. `workers.py` is the
other end and documents the frames.
Started by the node with one JSON line on stdin: the loopback port to call
back, the token that proves it is the process the node started, and the ICE
settings. It ends when that socket does, so it cannot outlive its node.
"""
import asyncio
import json
import logging
import struct
import sys
HEADER = struct.Struct(">IBI") # length of what follows, frame type, connection
MAX_FRAME = 64 * 1024 * 1024
# node -> process
OFFER, SEND, CLOSE, CONFIG, QUIT = 1, 2, 3, 4, 5
# process -> node
ANSWER, CHANNEL, MESSAGE, STATE, FLOW, STATS, LOG = 16, 17, 18, 19, 20, 21, 22
# How often the node learns what a channel has put on the wire. The download
# path waits for room in steps of 50 ms (files.py), so anything well under that
# is as good as reading `bufferedAmount` in place.
FLOW_INTERVAL = 0.02
STATS_EVERY = 50 # flow ticks between two SCTP summaries
def pack(kind: int, conn: int, payload: bytes = b"") -> bytes:
return HEADER.pack(len(payload) + 5, kind, conn) + payload
async def read_frame(reader: asyncio.StreamReader) -> tuple[int, int, bytes]:
length, kind, conn = HEADER.unpack(await reader.readexactly(HEADER.size))
if not 5 <= length <= MAX_FRAME:
raise ValueError(f"frame of {length} bytes")
return kind, conn, await reader.readexactly(length - 5)
class _Forward(logging.Handler):
"""Hands every record to the node, which owns the log file."""
def __init__(self, write) -> None:
super().__init__()
self._write = write
def emit(self, record: logging.LogRecord) -> None:
try:
body = {"name": record.name, "level": record.levelno,
"text": record.getMessage()}
self._write(pack(LOG, 0, json.dumps(body).encode()))
except Exception:
pass
class _Conn:
def __init__(self, pc) -> None:
self.pc = pc
self.channel = None
self.taken = 0 # bytes of SEND frames handed to the channel
self.reported = -1
self.stats = ""
class Worker:
def __init__(self, reader, writer) -> None:
self._reader = reader
self._writer = writer
self._conns: dict[int, _Conn] = {}
self._stun: list[str] = []
self._tasks: set[asyncio.Task] = set()
def _put(self, kind: int, conn: int, payload: bytes = b"") -> None:
if not self._writer.is_closing():
self._writer.write(pack(kind, conn, payload))
def _put_json(self, kind: int, conn: int, body: dict) -> None:
self._put(kind, conn, json.dumps(body).encode())
def _spawn(self, coro) -> None:
task = asyncio.ensure_future(coro)
self._tasks.add(task)
task.add_done_callback(self._tasks.discard)
def configure(self, cfg: dict) -> None:
from meshbay_node.transport import ice_filter, stun_multi
if "ice_interfaces" in cfg:
ice_filter.install(cfg["ice_interfaces"] or None)
if "stun" in cfg:
self._stun = list(cfg["stun"] or [])
stun_multi.install(self._stun or None)
async def run(self) -> None:
flow = asyncio.ensure_future(self._report_flow())
try:
while True:
kind, conn, payload = await read_frame(self._reader)
if kind == SEND:
c = self._conns.get(conn)
if c is not None and c.channel is not None \
and c.channel.readyState == "open":
c.channel.send(payload)
if c is not None:
# Counted whether or not it went out: the node's view of
# what is buffered must come back to zero on a channel
# that has closed, or its sender waits for room forever.
c.taken += len(payload)
elif kind == OFFER:
self._spawn(self._offer(conn, json.loads(payload)))
elif kind == CLOSE:
self._spawn(self._close(conn))
elif kind == CONFIG:
self.configure(json.loads(payload))
elif kind == QUIT:
break
except (asyncio.IncompleteReadError, ConnectionError):
pass
finally:
flow.cancel()
for conn in list(self._conns):
await self._close(conn)
async def _offer(self, conn: int, body: dict) -> None:
from aiortc import (
RTCConfiguration,
RTCIceServer,
RTCPeerConnection,
RTCSessionDescription,
)
from meshbay_node.transport.webrtc.channel import _get_remote_ip
try:
config = RTCConfiguration(
iceServers=[RTCIceServer(urls=s) for s in self._stun])
pc = RTCPeerConnection(configuration=config)
c = self._conns[conn] = _Conn(pc)
def state() -> None:
self._put_json(STATE, conn, {
"connection": pc.connectionState, "ice": pc.iceConnectionState,
"ip": _get_remote_ip(pc),
"channel": c.channel.readyState if c.channel else ""})
@pc.on("datachannel")
def on_datachannel(channel) -> None:
c.channel = channel
self._put_json(CHANNEL, conn, {"label": channel.label})
@channel.on("message")
def on_message(message) -> None:
if isinstance(message, str):
message = message.encode()
self._put(MESSAGE, conn, message)
channel.on("close", state)
@pc.on("connectionstatechange")
async def on_state() -> None:
state()
if pc.connectionState in ("failed", "closed"):
if self._conns.pop(conn, None) is not None \
and pc.connectionState == "failed":
await pc.close()
pc.on("iceconnectionstatechange", state)
await pc.setRemoteDescription(
RTCSessionDescription(sdp=body["sdp"], type="offer"))
await pc.setLocalDescription(await pc.createAnswer())
self._put_json(ANSWER, conn, {"sdp": pc.localDescription.sdp})
except Exception as e:
gone = self._conns.pop(conn, None)
self._put_json(ANSWER, conn, {"error": str(e) or type(e).__name__})
if gone is not None:
try:
await gone.pc.close()
except Exception:
pass
async def _close(self, conn: int) -> None:
c = self._conns.pop(conn, None)
if c is None:
return
try:
await c.pc.close()
except Exception:
pass
self._put_json(STATE, conn, {"connection": "closed", "ice": "closed",
"ip": "", "channel": "closed"})
async def _report_flow(self) -> None:
from meshbay_node.transport.webrtc import sctp_stats
tick = 0
while True:
await asyncio.sleep(FLOW_INTERVAL)
tick += 1
for conn, c in list(self._conns.items()):
if c.channel is None:
continue
sent = c.taken - c.channel.bufferedAmount \
if c.channel.readyState == "open" else c.taken
if sent != c.reported:
c.reported = sent
self._put(FLOW, conn, struct.pack(">Q", max(0, sent)))
if tick % STATS_EVERY == 0:
line = sctp_stats.describe(c.pc)
if line != c.stats:
c.stats = line
self._put(STATS, conn, line.encode())
async def _main() -> None:
hello = json.loads(sys.stdin.readline())
reader, writer = await asyncio.open_connection("127.0.0.1", hello["port"])
writer.write(bytes.fromhex(hello["token"]))
root = logging.getLogger()
root.handlers[:] = [_Forward(writer.write)]
root.setLevel(hello.get("level", logging.INFO))
# Patches aiortc on import, as the node does for the connections it holds
# itself.
from meshbay_node.transport.webrtc import sctp_fix, sctp_stats # noqa: F401
worker = Worker(reader, writer)
worker.configure(hello)
await worker.run()
writer.close()
def main() -> None:
try:
asyncio.run(_main())
except KeyboardInterrupt:
pass
if __name__ == "__main__":
main()
|