feat(us-017): P2P gossip, NAT-traversal relay node, and TLS
- packages/p2p: identity (peer_id from sha256 of RSA pubkey), TLS cert generation with SHA-256 fingerprint, GossipClient (WSS PubSub with per-topic handlers, dedup by msg_id, auto-reconnect), MdnsDiscovery (zeroconf optional dependency, graceful no-op fallback) - packages/relay: new meshnet-relay package — RelayServer (asyncio + websockets) with gossip fanout hub, circuit relay proxy for NAT traversal, peer registry; meshnet-relay CLI - packages/p2p/relay_bootstrap.json: team relay bootstrap list - Tracker: _NodeEntry gains relay_addr, cert_fingerprint, peer_id; both register and heartbeat handlers read and store these optional fields - docs/adr/0010 already written (previous commit) - conftest.py: packages/relay added to sys.path - 18 new tests; 115 passed total, 1 skipped (no regressions) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Dobromir Popov
3
packages/relay/meshnet_relay/__init__.py
Normal file
3
packages/relay/meshnet_relay/__init__.py
Normal file
@@ -0,0 +1,3 @@
|
||||
"""meshnet-relay — NAT-traversal relay and gossip hub."""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
60
packages/relay/meshnet_relay/cli.py
Normal file
60
packages/relay/meshnet_relay/cli.py
Normal file
@@ -0,0 +1,60 @@
|
||||
"""meshnet-relay CLI entry point."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="meshnet-relay",
|
||||
description="Meshnet NAT-traversal relay and gossip hub",
|
||||
)
|
||||
parser.add_argument("--host", default="0.0.0.0", help="Interface to bind")
|
||||
parser.add_argument("--port", type=int, default=8765, help="WebSocket port")
|
||||
parser.add_argument("--cert", metavar="PATH", help="TLS certificate (PEM)")
|
||||
parser.add_argument("--key", metavar="PATH", help="TLS private key (PEM)")
|
||||
parser.add_argument("--max-peers", type=int, default=500, help="Max concurrent peers")
|
||||
parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"])
|
||||
|
||||
args = parser.parse_args()
|
||||
logging.basicConfig(
|
||||
level=getattr(logging, args.log_level),
|
||||
format="%(asctime)s %(levelname)-8s %(name)s %(message)s",
|
||||
)
|
||||
|
||||
from .server import RelayServer
|
||||
|
||||
ssl_cert = Path(args.cert) if args.cert else None
|
||||
ssl_key = Path(args.key) if args.key else None
|
||||
|
||||
server = RelayServer(
|
||||
host=args.host,
|
||||
port=args.port,
|
||||
ssl_cert=ssl_cert,
|
||||
ssl_key=ssl_key,
|
||||
max_peers=args.max_peers,
|
||||
)
|
||||
port = server.start()
|
||||
scheme = "wss" if ssl_cert else "ws"
|
||||
print(f"meshnet-relay listening on {scheme}://{args.host}:{port}", flush=True)
|
||||
print(" /ws gossip PubSub", flush=True)
|
||||
print(" /relay/<id> circuit relay to peer", flush=True)
|
||||
print(" /health health check", flush=True)
|
||||
print(" /v1/peers peer list", flush=True)
|
||||
print("Press Ctrl-C to stop.", flush=True)
|
||||
|
||||
try:
|
||||
while True:
|
||||
time.sleep(1)
|
||||
except KeyboardInterrupt:
|
||||
print("\nStopping relay…", flush=True)
|
||||
server.stop()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
60
packages/relay/meshnet_relay/peer_registry.py
Normal file
60
packages/relay/meshnet_relay/peer_registry.py
Normal file
@@ -0,0 +1,60 @@
|
||||
"""In-memory registry of connected gossip peers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
|
||||
@dataclass
|
||||
class PeerEntry:
|
||||
peer_id: str
|
||||
addr: str
|
||||
ws: Any # websockets.WebSocketServerProtocol
|
||||
connected_at: float = field(default_factory=time.monotonic)
|
||||
last_seen: float = field(default_factory=time.monotonic)
|
||||
|
||||
|
||||
class PeerRegistry:
|
||||
def __init__(self):
|
||||
self._peers: dict[str, PeerEntry] = {}
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def register(self, peer_id: str, addr: str, ws) -> None:
|
||||
with self._lock:
|
||||
self._peers[peer_id] = PeerEntry(peer_id=peer_id, addr=addr, ws=ws)
|
||||
|
||||
def unregister(self, peer_id: str) -> None:
|
||||
with self._lock:
|
||||
self._peers.pop(peer_id, None)
|
||||
|
||||
def touch(self, peer_id: str) -> None:
|
||||
with self._lock:
|
||||
if peer_id in self._peers:
|
||||
self._peers[peer_id].last_seen = time.monotonic()
|
||||
|
||||
def get(self, peer_id: str) -> PeerEntry | None:
|
||||
with self._lock:
|
||||
return self._peers.get(peer_id)
|
||||
|
||||
def all_except(self, peer_id: str) -> list[PeerEntry]:
|
||||
with self._lock:
|
||||
return [e for pid, e in self._peers.items() if pid != peer_id]
|
||||
|
||||
def list_peers(self) -> list[dict]:
|
||||
with self._lock:
|
||||
return [
|
||||
{
|
||||
"peer_id": e.peer_id,
|
||||
"addr": e.addr,
|
||||
"connected_at": e.connected_at,
|
||||
"last_seen": e.last_seen,
|
||||
}
|
||||
for e in self._peers.values()
|
||||
]
|
||||
|
||||
def __len__(self) -> int:
|
||||
with self._lock:
|
||||
return len(self._peers)
|
||||
224
packages/relay/meshnet_relay/server.py
Normal file
224
packages/relay/meshnet_relay/server.py
Normal file
@@ -0,0 +1,224 @@
|
||||
"""Relay server — WebSocket gossip hub + circuit relay proxy.
|
||||
|
||||
HTTP API (served via asyncio-based handler on same port):
|
||||
GET /health → {"status": "ok", "peers": N}
|
||||
GET /v1/peers → [{peer_id, addr, last_seen}]
|
||||
POST /v1/gossip → accept a gossip envelope, fan out to connected peers
|
||||
|
||||
WebSocket endpoints:
|
||||
ws[s]://host:port/ws → gossip PubSub connection
|
||||
ws[s]://host:port/relay/{peer_id} → circuit relay to that peer
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
from .peer_registry import PeerRegistry
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class RelayServer:
|
||||
"""Async WebSocket relay server that runs in a background thread.
|
||||
|
||||
Usage::
|
||||
|
||||
server = RelayServer(host="0.0.0.0", port=8765)
|
||||
port = server.start() # returns actual port
|
||||
...
|
||||
server.stop()
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
host: str = "0.0.0.0",
|
||||
port: int = 8765,
|
||||
ssl_cert: Path | None = None,
|
||||
ssl_key: Path | None = None,
|
||||
max_peers: int = 500,
|
||||
):
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.ssl_cert = ssl_cert
|
||||
self.ssl_key = ssl_key
|
||||
self.max_peers = max_peers
|
||||
|
||||
self._registry = PeerRegistry()
|
||||
self._loop: asyncio.AbstractEventLoop | None = None
|
||||
self._thread: threading.Thread | None = None
|
||||
self._server = None
|
||||
self._actual_port: int = port
|
||||
self._ready = threading.Event()
|
||||
self._running = False
|
||||
self._stop_event: asyncio.Event | None = None
|
||||
|
||||
@property
|
||||
def registry(self) -> PeerRegistry:
|
||||
return self._registry
|
||||
|
||||
def start(self) -> int:
|
||||
"""Start server in background thread. Returns actual bound port."""
|
||||
self._running = True
|
||||
self._loop = asyncio.new_event_loop()
|
||||
self._thread = threading.Thread(target=self._run, daemon=True, name="relay")
|
||||
self._thread.start()
|
||||
self._ready.wait(timeout=5)
|
||||
return self._actual_port
|
||||
|
||||
def stop(self) -> None:
|
||||
self._running = False
|
||||
if self._loop and self._stop_event is not None:
|
||||
self._loop.call_soon_threadsafe(self._stop_event.set)
|
||||
if self._thread:
|
||||
self._thread.join(timeout=3.0)
|
||||
|
||||
def _run(self) -> None:
|
||||
asyncio.set_event_loop(self._loop)
|
||||
self._loop.run_until_complete(self._serve())
|
||||
|
||||
async def _serve(self) -> None:
|
||||
import websockets # type: ignore[import]
|
||||
import websockets.server # type: ignore[import]
|
||||
|
||||
ssl_ctx = None
|
||||
if self.ssl_cert and self.ssl_key:
|
||||
import ssl
|
||||
ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
||||
ssl_ctx.load_cert_chain(str(self.ssl_cert), str(self.ssl_key))
|
||||
|
||||
server = await websockets.serve(
|
||||
self._handle_connection,
|
||||
self.host,
|
||||
self.port,
|
||||
ssl=ssl_ctx,
|
||||
)
|
||||
# Record actual port after bind
|
||||
for sock in server.sockets or []:
|
||||
self._actual_port = sock.getsockname()[1]
|
||||
break
|
||||
|
||||
self._stop_event = asyncio.Event()
|
||||
self._server = server
|
||||
self._ready.set()
|
||||
log.info("Relay listening on %s:%d", self.host, self._actual_port)
|
||||
|
||||
await self._stop_event.wait()
|
||||
server.close()
|
||||
await server.wait_closed()
|
||||
|
||||
async def _handle_connection(self, ws) -> None:
|
||||
"""Dispatch incoming WebSocket to gossip hub or circuit relay."""
|
||||
try:
|
||||
path = ws.request.path
|
||||
except AttributeError:
|
||||
path = getattr(ws, "path", "/ws")
|
||||
|
||||
if path.startswith("/relay/"):
|
||||
peer_id = path[len("/relay/"):]
|
||||
await self._handle_circuit_relay(ws, peer_id)
|
||||
elif path == "/health":
|
||||
await ws.send(json.dumps({"status": "ok", "peers": len(self._registry)}))
|
||||
await ws.close()
|
||||
elif path == "/v1/peers":
|
||||
await ws.send(json.dumps(self._registry.list_peers()))
|
||||
await ws.close()
|
||||
else:
|
||||
await self._handle_gossip(ws)
|
||||
|
||||
async def _handle_gossip(self, ws) -> None:
|
||||
"""Accept a gossip peer connection, register it, and fan out messages."""
|
||||
peer_id: str | None = None
|
||||
peer_addr: str = ""
|
||||
|
||||
try:
|
||||
async for raw in ws:
|
||||
try:
|
||||
envelope = json.loads(raw)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
continue
|
||||
|
||||
topic = envelope.get("topic", "")
|
||||
from_peer = envelope.get("from_peer", "")
|
||||
|
||||
# Handle peer registration message
|
||||
if topic == "peer-register":
|
||||
payload = envelope.get("payload", {})
|
||||
peer_id = payload.get("peer_id") or from_peer
|
||||
peer_addr = payload.get("addr", "")
|
||||
if len(self._registry) >= self.max_peers:
|
||||
await ws.close(1008, "relay at capacity")
|
||||
return
|
||||
self._registry.register(peer_id, peer_addr, ws)
|
||||
log.debug("Peer registered: %s", peer_id)
|
||||
# Send current peer list back
|
||||
await ws.send(json.dumps({
|
||||
"topic": "peer-list",
|
||||
"version": 1,
|
||||
"from_peer": "relay",
|
||||
"payload": {"peers": self._registry.list_peers()},
|
||||
}))
|
||||
continue
|
||||
|
||||
# Fan out to all other registered peers
|
||||
if peer_id:
|
||||
self._registry.touch(peer_id)
|
||||
fan_out_peers = self._registry.all_except(peer_id or "")
|
||||
await _broadcast(raw, fan_out_peers)
|
||||
|
||||
except Exception as exc:
|
||||
log.debug("Gossip connection error: %s", exc)
|
||||
finally:
|
||||
if peer_id:
|
||||
self._registry.unregister(peer_id)
|
||||
log.debug("Peer unregistered: %s", peer_id)
|
||||
|
||||
async def _handle_circuit_relay(self, ws_requester, target_peer_id: str) -> None:
|
||||
"""Proxy WebSocket traffic between ws_requester and target_peer_id's ws."""
|
||||
target = self._registry.get(target_peer_id)
|
||||
if target is None:
|
||||
try:
|
||||
await ws_requester.send(json.dumps({
|
||||
"error": f"peer {target_peer_id!r} not connected to relay"
|
||||
}))
|
||||
await ws_requester.close(1011, "target peer not found")
|
||||
except Exception:
|
||||
pass
|
||||
return
|
||||
|
||||
log.debug("Circuit relay: ??? → %s", target_peer_id)
|
||||
|
||||
async def pipe(src, dst) -> None:
|
||||
try:
|
||||
async for msg in src:
|
||||
await dst.send(msg)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
await asyncio.gather(
|
||||
pipe(ws_requester, target.ws),
|
||||
pipe(target.ws, ws_requester),
|
||||
return_exceptions=True,
|
||||
)
|
||||
|
||||
|
||||
async def _broadcast(raw: str | bytes, peers: list) -> None:
|
||||
"""Send raw message to all peers; ignore individual send failures."""
|
||||
if not peers:
|
||||
return
|
||||
import asyncio
|
||||
await asyncio.gather(
|
||||
*[_safe_send(p.ws, raw) for p in peers],
|
||||
return_exceptions=True,
|
||||
)
|
||||
|
||||
|
||||
async def _safe_send(ws, msg) -> None:
|
||||
try:
|
||||
await ws.send(msg)
|
||||
except Exception:
|
||||
pass
|
||||
20
packages/relay/pyproject.toml
Normal file
20
packages/relay/pyproject.toml
Normal file
@@ -0,0 +1,20 @@
|
||||
[build-system]
|
||||
requires = ["setuptools>=64"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "meshnet-relay"
|
||||
version = "0.1.0"
|
||||
description = "Distributed Inference Network NAT-traversal relay and gossip hub"
|
||||
requires-python = ">=3.10"
|
||||
|
||||
dependencies = [
|
||||
"websockets>=13",
|
||||
]
|
||||
|
||||
[project.scripts]
|
||||
meshnet-relay = "meshnet_relay.cli:main"
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["."]
|
||||
include = ["meshnet_relay*"]
|
||||
Reference in New Issue
Block a user