"""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 ( # noqa: F401 sctp_fix, sctp_stats, udp_buffer, ) 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()