e62f911810
E2E-verified both directions dev-box<->x250 over real RF (5KB image ~60s):
1. Sender daemon never called link.identify() — receiver's
get_remote_identity() was None, so every arrived transfer carried an
empty source_hash and the Rust side dropped it (now also warns
instead of silently vanishing it).
2. Receiver treated resource.data as bytes, but RNS hands a concluded
Resource's data as a file-like BufferedReader — b64encode raised
TypeError and the transfer was lost even when attributed.
3. Radio twins of merged contacts carry the peer's Archipelago ed25519
key as pubkey_hex, not an RNS hash — prefix lookup could never match
('Unknown Reticulum prefix', observed live). resolve_dest_hash now
falls back to matching the announce-bound arch_pubkey_hex.
4. The daemon RPC socket kept asyncio's default 64KiB line limit; any
attachment >~48KB overflowed it and tore down the whole daemon
connection ('reticulum-daemon is gone'). Raised to 16MiB.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
735 lines
34 KiB
Python
735 lines
34 KiB
Python
#!/usr/bin/env python3
|
|
"""Archipelago Reticulum daemon — host-supervised RNS + LXMF bridge.
|
|
|
|
archipelago spawns and supervises one of these per active Reticulum (RNode) radio.
|
|
It owns the serial port, runs the Reticulum stack + an LXMF router whose identity is
|
|
**derived deterministically from the Archy node key**, and exposes a tiny
|
|
line-delimited JSON-RPC over a Unix domain socket (0600) for the Rust side to drive.
|
|
|
|
Security posture (see the plan's "most secure way" section):
|
|
* Runs as the same rootless archipelago user; no root, no network control plane.
|
|
* RPC is a Unix socket only; the identity key never leaves the host and is never
|
|
logged. The daemon executes ONLY the fixed verb set below.
|
|
|
|
RPC (one JSON object per line, both directions):
|
|
in : {"cmd":"send","dest_hash":"<hex16>","content":"…","title":"…","method":"direct|opportunistic"}
|
|
{"cmd":"announce"}
|
|
{"cmd":"set_name","name":"…"}
|
|
{"cmd":"status"}
|
|
{"cmd":"send_resource","id":"<correlation>","dest_hash":"<hex16>","data_b64":"…"}
|
|
{"cmd":"shutdown"}
|
|
out: {"event":"ready","dest_hash":"<hex16>","display_name":"…"}
|
|
{"event":"recv","source_hash":"<hex16>","content":"…","title":"…","fields":{…},"app_data":"<hex>","rssi":n,"snr":n,"stamp":t}
|
|
{"event":"announce","dest_hash":"<hex16>","app_data":"<hex>","display_name":"…"|null,"archy_blob":"ARCHY:2:…"|null}
|
|
{"event":"delivered","dest_hash":"<hex16>","state":"delivered|failed","id":"<hex>"}
|
|
{"event":"status","connected":bool,"dest_hash":"<hex16>","interfaces":[…]}
|
|
{"event":"resource_progress","id":"<correlation>","transferred":n,"total":n}
|
|
{"event":"resource_sent","id":"<correlation>"}
|
|
{"event":"resource_failed","id":"<correlation>","reason":"…"}
|
|
{"event":"resource_recv","source_hash":"<hex16>","data_b64":"…"}
|
|
|
|
``send_resource`` is for large (>~2.3KB) binary payloads that don't fit the small
|
|
LXMF-message path — it uses RNS's native Resource transfer protocol over a `RNS.Link`
|
|
to the peer's *resource* destination (a separate aspect from the LXMF delivery
|
|
destination, so it doesn't disturb normal messaging). Built for sending compressed
|
|
photos/files/voice-messages directly over LoRa instead of always falling back to Tor
|
|
past the small-message size cap. ``resource_recv``'s `data_b64` is the same
|
|
CBOR-encoded payload format already used for the small-inline LoRa path, so the Rust
|
|
side decodes it identically regardless of which path delivered it.
|
|
|
|
This is the Phase-1 skeleton: the identity/LXMF wiring and RPC loop are real and
|
|
exercised by ``--check`` / ``--selftest`` (no radio). The live LoRa message path is
|
|
validated in the Phase-0 hardware gate on .116/.228.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import base64
|
|
import json
|
|
import os
|
|
import signal
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from archy_rns_identity import lxmf_destination_hash, load_identity
|
|
|
|
# Lazy heavy imports (RNS/LXMF) happen in run(); --check stays import-light.
|
|
|
|
|
|
# ─────────────────────────── RNS config generation ───────────────────────────
|
|
|
|
def _require_loopback(host: str) -> None:
|
|
"""Bind-scope guard: TCP server mode is dev/verification only for now. WAN/LAN
|
|
exposure is a separate future decision needing its own security review — archy
|
|
is Tor-first for inter-node traffic (see README/CLAUDE.md), and a plain-TCP
|
|
Reticulum listener bound beyond loopback would bypass that entirely. Defense in
|
|
depth: the Rust side enforces the same rule (reticulum.rs::is_loopback_host),
|
|
but this daemon can also be invoked directly, so it re-checks here too."""
|
|
if host not in ("127.0.0.1", "::1", "localhost"):
|
|
raise SystemExit(
|
|
f"--tcp-listen host must be loopback-only (got {host!r}); "
|
|
"WAN/LAN bind is out of scope for this daemon build"
|
|
)
|
|
|
|
|
|
def _write_rns_config(
|
|
configdir: Path,
|
|
*,
|
|
serial_port: str | None,
|
|
lora: dict,
|
|
no_radio: bool,
|
|
tcp_listen: str | None = None,
|
|
tcp_connect: list[str] | None = None,
|
|
) -> None:
|
|
"""Materialise an RNS config file. RNode interface for real radios; plain-TCP
|
|
server/client interface(s) for radio-less dev/verification (e.g. Aurora
|
|
interop testing); a loopback-only disabled config for --selftest so the stack
|
|
comes up with zero interfaces at all."""
|
|
configdir.mkdir(parents=True, exist_ok=True)
|
|
cfg = configdir / "config"
|
|
if no_radio:
|
|
interfaces = (
|
|
" [[Default Interface]]\n"
|
|
" type = AutoInterface\n"
|
|
" enabled = no\n"
|
|
)
|
|
elif serial_port:
|
|
interfaces = (
|
|
" [[RNode LoRa]]\n"
|
|
" type = RNodeInterface\n"
|
|
" enabled = yes\n"
|
|
f" port = {serial_port}\n"
|
|
f" frequency = {lora['frequency']}\n"
|
|
f" bandwidth = {lora['bandwidth']}\n"
|
|
f" txpower = {lora['txpower']}\n"
|
|
f" spreadingfactor = {lora['spreadingfactor']}\n"
|
|
f" codingrate = {lora['codingrate']}\n"
|
|
)
|
|
elif tcp_listen or tcp_connect:
|
|
parts = []
|
|
if tcp_listen:
|
|
host, _, port = tcp_listen.rpartition(":")
|
|
_require_loopback(host)
|
|
parts.append(
|
|
" [[Reticulum TCP Server]]\n"
|
|
" type = TCPServerInterface\n"
|
|
" enabled = yes\n"
|
|
f" listen_ip = {host}\n"
|
|
f" listen_port = {port}\n"
|
|
)
|
|
for i, target in enumerate(tcp_connect or []):
|
|
host, _, port = target.rpartition(":")
|
|
parts.append(
|
|
f" [[Reticulum TCP Client {i}]]\n"
|
|
" type = TCPClientInterface\n"
|
|
" enabled = yes\n"
|
|
f" target_host = {host}\n"
|
|
f" target_port = {port}\n"
|
|
)
|
|
interfaces = "\n".join(parts)
|
|
else:
|
|
raise SystemExit(
|
|
"no interface configured: need --serial-port, --tcp-listen/--tcp-connect, "
|
|
"or --no-radio"
|
|
)
|
|
cfg.write_text(
|
|
"[reticulum]\n"
|
|
" enable_transport = no\n"
|
|
" share_instance = no\n"
|
|
" panic_on_interface_error = no\n\n"
|
|
"[interfaces]\n" + interfaces
|
|
)
|
|
os.chmod(cfg, 0o600)
|
|
|
|
|
|
# ─────────────────────────────── the daemon ──────────────────────────────────
|
|
|
|
class ReticulumDaemon:
|
|
def __init__(self, args):
|
|
self.args = args
|
|
self.seed = self._read_seed(Path(args.identity_key))
|
|
self.loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(self.loop)
|
|
self.clients: set[asyncio.StreamWriter] = set()
|
|
self.reticulum = None
|
|
self.router = None
|
|
self.delivery_destination = None
|
|
self.identity = None
|
|
self.dest_hash_hex = lxmf_destination_hash(self.seed).hex()
|
|
# Resource transfer (large binary payloads over a dedicated Link, separate
|
|
# from LXMF delivery — see module docstring). Keyed by the peer's *resource*
|
|
# destination hash (bytes), not their LXMF delivery hash.
|
|
self.resource_destination = None
|
|
self.links: dict[bytes, "RNS.Link"] = {}
|
|
self.pending_resource_sends: dict[bytes, list[tuple[bytes, str]]] = {}
|
|
|
|
@staticmethod
|
|
def _read_seed(path: Path) -> bytes:
|
|
seed = path.read_bytes()
|
|
if len(seed) != 32:
|
|
raise SystemExit(f"identity key {path} must be 32 bytes, got {len(seed)}")
|
|
return seed
|
|
|
|
# ---- RNS / LXMF bring-up ----
|
|
def bring_up(self):
|
|
import RNS
|
|
import LXMF
|
|
|
|
configdir = Path(self.args.rns_config)
|
|
_write_rns_config(
|
|
configdir,
|
|
serial_port=self.args.serial_port,
|
|
lora={
|
|
"frequency": self.args.frequency,
|
|
"bandwidth": self.args.bandwidth,
|
|
"txpower": self.args.txpower,
|
|
"spreadingfactor": self.args.spreadingfactor,
|
|
"codingrate": self.args.codingrate,
|
|
},
|
|
no_radio=self.args.no_radio,
|
|
tcp_listen=self.args.tcp_listen,
|
|
tcp_connect=self.args.tcp_connect,
|
|
)
|
|
self.reticulum = RNS.Reticulum(configdir=str(configdir))
|
|
self.identity = load_identity(self.seed)
|
|
|
|
storagepath = str(configdir / "lxmf")
|
|
Path(storagepath).mkdir(parents=True, exist_ok=True)
|
|
self.router = LXMF.LXMRouter(identity=self.identity, storagepath=storagepath)
|
|
self.delivery_destination = self.router.register_delivery_identity(
|
|
self.identity, display_name=self.args.display_name
|
|
)
|
|
self.router.register_delivery_callback(self._on_lxmf_delivery)
|
|
|
|
# Hear other LXMF nodes' announces so we can surface peers + bind contacts.
|
|
RNS.Transport.register_announce_handler(_AnnounceHandler(self))
|
|
|
|
assert self.delivery_destination.hash.hex() == self.dest_hash_hex, (
|
|
"derived dest hash diverged from LXMF's — aspect/identity mismatch"
|
|
)
|
|
|
|
# Separate destination/aspect for inbound Resource transfers (large
|
|
# binary payloads), so it never collides with LXMF delivery traffic.
|
|
# Every peer running this same daemon code can derive our resource
|
|
# destination hash from our (already-announced) identity via
|
|
# RNS.Destination.hash(identity, "archy", "resource") — see
|
|
# _resource_dest_hash_for, used on the sending side.
|
|
self.resource_destination = RNS.Destination(
|
|
self.identity, RNS.Destination.IN, RNS.Destination.SINGLE,
|
|
"archy", "resource",
|
|
)
|
|
self.resource_destination.set_link_established_callback(
|
|
self._on_resource_link_established
|
|
)
|
|
|
|
def announce(self):
|
|
if self.delivery_destination is not None:
|
|
self.delivery_destination.announce(app_data=self._announce_app_data())
|
|
|
|
def _archy_identity_blob(self):
|
|
"""The ``ARCHY:2:{ed25519_hex}:{x25519_hex}`` identity string the Rust
|
|
side parses (``protocol::parse_identity_broadcast``) and binds via
|
|
``handle_identity_received`` — so a Reticulum-carried identity merges
|
|
into the SAME conversation as the meshcore/Meshtastic/federation twins
|
|
of the same Archy node. The keys are the node's real Archipelago
|
|
pubkeys (passed in by the Rust side) — NOT this daemon's
|
|
internally-HKDF-derived RNS keys, which exist only to make the RNS
|
|
destination hash deterministic. ``None`` when the pubkeys weren't
|
|
supplied (dev/selftest run)."""
|
|
if self.args.archy_ed_pubkey_hex and self.args.archy_x25519_pubkey_hex:
|
|
return (
|
|
f"ARCHY:2:{self.args.archy_ed_pubkey_hex}:"
|
|
f"{self.args.archy_x25519_pubkey_hex}"
|
|
).encode("ascii")
|
|
return None
|
|
|
|
def _announce_app_data(self) -> bytes:
|
|
"""LXMF-standard announce app_data — msgpack ``[display_name,
|
|
stamp_cost, supported_functionality]`` via the router, so Sideband/
|
|
NomadNet/MeshChat (and upgraded archy nodes) all see our real display
|
|
name — with the Archy identity blob appended as an EXTRA list element.
|
|
Stock clients only read the elements they know ([0]/[1]), so the blob
|
|
rides along invisibly instead of replacing the name the way the old
|
|
blob-only app_data did (which left every archy node nameless on RNS).
|
|
"""
|
|
import RNS.vendor.umsgpack as msgpack
|
|
app_data = self.router.get_announce_app_data(self.delivery_destination.hash)
|
|
blob = self._archy_identity_blob()
|
|
if blob is None:
|
|
return app_data
|
|
try:
|
|
peer_data = msgpack.unpackb(app_data)
|
|
if not isinstance(peer_data, list):
|
|
raise ValueError("unexpected announce app_data shape")
|
|
peer_data.append(blob)
|
|
return msgpack.packb(peer_data)
|
|
except Exception:
|
|
# Never let announce formatting kill announcing entirely — fall
|
|
# back to the legacy blob-only format (identity binding > name).
|
|
return blob
|
|
|
|
# ---- RNS-thread callbacks → asyncio ----
|
|
def _on_lxmf_delivery(self, message):
|
|
import LXMF
|
|
try:
|
|
app_data = b""
|
|
src = message.source_hash.hex() if message.source_hash else ""
|
|
event = {
|
|
"event": "recv",
|
|
"source_hash": src,
|
|
"content": message.content_as_string() if hasattr(message, "content_as_string")
|
|
else (message.content.decode("utf-8", "replace") if message.content else ""),
|
|
"title": message.title_as_string() if hasattr(message, "title_as_string") else "",
|
|
"app_data": app_data.hex(),
|
|
"stamp": getattr(message, "timestamp", None),
|
|
}
|
|
# Native LXMF attachment fields (Sideband/NomadNet/stock clients use
|
|
# these, NOT our own typed-envelope wire format) — a stock client's
|
|
# photo/voice-memo/file arrives here, not in `content`, which is why
|
|
# it was previously dropped silently (content was just blank/space).
|
|
# See LXMF field format confirmed against Sideband's own source
|
|
# (sbapp/sideband/core.py): FIELD_IMAGE = [format_str, bytes],
|
|
# FIELD_AUDIO = [mode_byte, bytes], FIELD_FILE_ATTACHMENTS =
|
|
# [[filename, bytes], ...].
|
|
fields = getattr(message, "fields", None) or {}
|
|
if LXMF.FIELD_IMAGE in fields:
|
|
fmt, img_bytes = fields[LXMF.FIELD_IMAGE]
|
|
event["image_format"] = str(fmt)
|
|
event["image_b64"] = base64.b64encode(bytes(img_bytes)).decode("ascii")
|
|
if LXMF.FIELD_FILE_ATTACHMENTS in fields:
|
|
attachments = fields[LXMF.FIELD_FILE_ATTACHMENTS]
|
|
if attachments:
|
|
filename, file_bytes = attachments[0]
|
|
event["attachment_filename"] = str(filename)
|
|
event["attachment_b64"] = base64.b64encode(bytes(file_bytes)).decode("ascii")
|
|
self._emit_threadsafe(event)
|
|
except Exception as e: # never let a callback kill the RNS thread
|
|
self._emit_threadsafe({"event": "error", "where": "delivery", "detail": str(e)})
|
|
|
|
def _emit_threadsafe(self, obj: dict):
|
|
self.loop.call_soon_threadsafe(self._broadcast, obj)
|
|
|
|
def _broadcast(self, obj: dict):
|
|
line = (json.dumps(obj) + "\n").encode("utf-8")
|
|
for w in list(self.clients):
|
|
try:
|
|
w.write(line)
|
|
except Exception:
|
|
self.clients.discard(w)
|
|
|
|
# ---- RPC server ----
|
|
async def _handle_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
|
|
self.clients.add(writer)
|
|
writer.write((json.dumps({
|
|
"event": "ready", "dest_hash": self.dest_hash_hex,
|
|
"display_name": self.args.display_name,
|
|
}) + "\n").encode())
|
|
try:
|
|
async for raw in reader:
|
|
line = raw.decode("utf-8", "replace").strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
req = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
await self._dispatch(req, writer)
|
|
finally:
|
|
self.clients.discard(writer)
|
|
|
|
async def _dispatch(self, req: dict, writer: asyncio.StreamWriter):
|
|
cmd = req.get("cmd")
|
|
if cmd == "send":
|
|
self._send(req)
|
|
elif cmd == "announce":
|
|
self.announce()
|
|
elif cmd == "set_name":
|
|
# Live rename: update the LXMF delivery destination's display name
|
|
# (what get_announce_app_data reads) and re-announce immediately so
|
|
# peers learn the new name without waiting for the next advert tick.
|
|
name = (req.get("name") or "").strip()
|
|
if name and self.delivery_destination is not None:
|
|
self.args.display_name = name
|
|
self.delivery_destination.display_name = name
|
|
self.announce()
|
|
elif cmd == "status":
|
|
self._broadcast(self._status())
|
|
elif cmd == "send_resource":
|
|
self._send_resource(req)
|
|
elif cmd == "shutdown":
|
|
self.loop.stop()
|
|
# unknown verbs are ignored by contract
|
|
|
|
def _status(self) -> dict:
|
|
ifaces = []
|
|
if self.reticulum is not None:
|
|
try:
|
|
ifaces = [str(i) for i in self.reticulum.get_interface_stats().get("interfaces", [])]
|
|
except Exception:
|
|
pass
|
|
return {"event": "status", "connected": self.router is not None,
|
|
"dest_hash": self.dest_hash_hex, "interfaces": ifaces}
|
|
|
|
def _send(self, req: dict):
|
|
import RNS
|
|
import LXMF
|
|
try:
|
|
dest_hash = bytes.fromhex(req["dest_hash"])
|
|
except (KeyError, ValueError):
|
|
return
|
|
recipient_identity = RNS.Identity.recall(dest_hash)
|
|
if recipient_identity is None:
|
|
# No path/identity yet — ask the network and drop this attempt; the Rust
|
|
# side retries once the peer is reachable (mirrors LoRa "unreachable").
|
|
RNS.Transport.request_path(dest_hash)
|
|
self._broadcast({"event": "delivered", "dest_hash": req["dest_hash"],
|
|
"state": "failed", "id": "", "reason": "no_path"})
|
|
return
|
|
dest = RNS.Destination(recipient_identity, RNS.Destination.OUT,
|
|
RNS.Destination.SINGLE, "lxmf", "delivery")
|
|
method = {"direct": LXMF.LXMessage.DIRECT,
|
|
"opportunistic": LXMF.LXMessage.OPPORTUNISTIC,
|
|
"propagated": LXMF.LXMessage.PROPAGATED}.get(
|
|
req.get("method", "direct"), LXMF.LXMessage.DIRECT)
|
|
# Native LXMF FIELD_IMAGE — for a stock Sideband/NomadNet peer, which
|
|
# has no idea how to decode our own typed-envelope wire format. Rust
|
|
# only sets these two keys when the peer isn't an archy contact (see
|
|
# `is_archy_peer` gating in typed_messages.rs); [format, bytes] is the
|
|
# wire shape confirmed against Sideband's own source.
|
|
fields = {}
|
|
if req.get("image_b64") and req.get("image_format"):
|
|
fields[LXMF.FIELD_IMAGE] = [req["image_format"], base64.b64decode(req["image_b64"])]
|
|
msg = LXMF.LXMessage(dest, self.delivery_destination,
|
|
req.get("content", ""), req.get("title", ""),
|
|
desired_method=method, fields=fields or None)
|
|
msg.register_delivery_callback(lambda m: self._emit_threadsafe(
|
|
{"event": "delivered", "dest_hash": req["dest_hash"], "state": "delivered",
|
|
"id": m.hash.hex() if m.hash else ""}))
|
|
msg.register_failed_callback(lambda m: self._emit_threadsafe(
|
|
{"event": "delivered", "dest_hash": req["dest_hash"], "state": "failed",
|
|
"id": m.hash.hex() if m.hash else ""}))
|
|
self.router.handle_outbound(msg)
|
|
|
|
# ---- Resource transfer (large binary payloads over a dedicated Link) ----
|
|
def _resource_dest_hash_for(self, lxmf_dest_hash: bytes):
|
|
"""Derive a peer's *resource* destination hash from their LXMF delivery
|
|
hash. Requires having already recalled their Identity (e.g. via a prior
|
|
`_send`/announce) — returns None if we haven't heard from them yet."""
|
|
import RNS
|
|
identity = RNS.Identity.recall(lxmf_dest_hash)
|
|
if identity is None:
|
|
return None
|
|
return RNS.Destination.hash(identity, "archy", "resource")
|
|
|
|
def _get_or_create_link(self, lxmf_dest_hash: bytes):
|
|
import RNS
|
|
identity = RNS.Identity.recall(lxmf_dest_hash)
|
|
if identity is None:
|
|
RNS.Transport.request_path(lxmf_dest_hash)
|
|
return None, None
|
|
resource_hash = RNS.Destination.hash(identity, "archy", "resource")
|
|
link = self.links.get(resource_hash)
|
|
if link is not None and link.status != RNS.Link.CLOSED:
|
|
return link, resource_hash
|
|
out_dest = RNS.Destination(
|
|
identity, RNS.Destination.OUT, RNS.Destination.SINGLE,
|
|
"archy", "resource",
|
|
)
|
|
# established_callback identifies us to the peer BEFORE any Resource
|
|
# goes out: the receiving daemon attributes an inbound resource via
|
|
# link.get_remote_identity() (see _on_resource_received), which is
|
|
# None unless the initiator calls identify() — without this every
|
|
# transfer arrives with an empty source_hash and the Rust side can't
|
|
# route it to a contact (observed live 2026-07-28: 5KB image sent,
|
|
# concluded COMPLETE, never surfaced on the receiving node).
|
|
link = RNS.Link(
|
|
out_dest,
|
|
established_callback=lambda lk: self._on_out_link_established(
|
|
resource_hash, lk
|
|
),
|
|
)
|
|
link.set_link_closed_callback(
|
|
lambda lk: self.links.pop(resource_hash, None)
|
|
)
|
|
self.links[resource_hash] = link
|
|
return link, resource_hash
|
|
|
|
def _on_out_link_established(self, resource_hash: bytes, link):
|
|
link.identify(self.identity)
|
|
self._flush_pending_resource_sends(resource_hash, link)
|
|
|
|
def _send_resource(self, req: dict):
|
|
import RNS
|
|
req_id = req.get("id", "")
|
|
try:
|
|
lxmf_dest_hash = bytes.fromhex(req["dest_hash"])
|
|
data = base64.b64decode(req["data_b64"])
|
|
except (KeyError, ValueError, TypeError):
|
|
self._emit_threadsafe({"event": "resource_failed", "id": req_id,
|
|
"reason": "bad_request"})
|
|
return
|
|
link, resource_hash = self._get_or_create_link(lxmf_dest_hash)
|
|
if link is None:
|
|
self._emit_threadsafe({"event": "resource_failed", "id": req_id,
|
|
"reason": "no_path"})
|
|
return
|
|
self.pending_resource_sends.setdefault(resource_hash, [])
|
|
if link.status == RNS.Link.ACTIVE:
|
|
self._start_resource(link, data, req_id)
|
|
else:
|
|
# Link is establishing — queue; the creation-time established
|
|
# callback (_on_out_link_established) identifies us then flushes
|
|
# the queue. Re-check afterwards to close the race where the
|
|
# link went ACTIVE between the status check and the append
|
|
# (the callback fires once, on the RNS thread).
|
|
self.pending_resource_sends[resource_hash].append((data, req_id))
|
|
if link.status == RNS.Link.ACTIVE:
|
|
self._flush_pending_resource_sends(resource_hash, link)
|
|
|
|
def _flush_pending_resource_sends(self, resource_hash: bytes, link):
|
|
import RNS
|
|
pending = self.pending_resource_sends.pop(resource_hash, [])
|
|
for data, req_id in pending:
|
|
if link.status == RNS.Link.ACTIVE:
|
|
self._start_resource(link, data, req_id)
|
|
else:
|
|
self._emit_threadsafe({"event": "resource_failed", "id": req_id,
|
|
"reason": "link_failed"})
|
|
|
|
def _start_resource(self, link, data: bytes, req_id: str):
|
|
import RNS
|
|
|
|
def on_progress(resource):
|
|
total = resource.get_data_size() if hasattr(resource, "get_data_size") else len(data)
|
|
transferred = int(total * resource.get_progress()) if hasattr(resource, "get_progress") else 0
|
|
self._emit_threadsafe({"event": "resource_progress", "id": req_id,
|
|
"transferred": transferred, "total": total})
|
|
|
|
def on_concluded(resource):
|
|
if resource.status == RNS.Resource.COMPLETE:
|
|
self._emit_threadsafe({"event": "resource_sent", "id": req_id})
|
|
else:
|
|
self._emit_threadsafe({"event": "resource_failed", "id": req_id,
|
|
"reason": "transfer_failed"})
|
|
|
|
RNS.Resource(data, link, callback=on_concluded, progress_callback=on_progress)
|
|
|
|
def _on_resource_link_established(self, link):
|
|
import RNS
|
|
link.set_resource_strategy(RNS.Link.ACCEPT_ALL)
|
|
link.set_resource_concluded_callback(self._on_resource_received)
|
|
|
|
def _on_resource_received(self, resource):
|
|
import RNS
|
|
try:
|
|
if resource.status != RNS.Resource.COMPLETE:
|
|
return
|
|
identity = resource.link.get_remote_identity()
|
|
# Report the peer's LXMF *delivery* hash (not the raw identity hash,
|
|
# and not the resource-aspect hash) since that's what the Rust side's
|
|
# contact table is keyed on for every other inbound event.
|
|
source_hash = (
|
|
RNS.Destination.hash(identity, "lxmf", "delivery").hex()
|
|
if identity is not None else ""
|
|
)
|
|
# RNS hands a concluded Resource's data as a file-like object
|
|
# (BufferedReader over the assembled stream), not bytes —
|
|
# observed live 2026-07-28: b64encode raised "a bytes-like
|
|
# object is required" and every received transfer was lost.
|
|
data = resource.data
|
|
if hasattr(data, "read"):
|
|
data = data.read()
|
|
self._emit_threadsafe({
|
|
"event": "resource_recv",
|
|
"source_hash": source_hash,
|
|
"data_b64": base64.b64encode(data).decode("ascii"),
|
|
})
|
|
except Exception as e: # never let a callback kill the RNS thread
|
|
self._emit_threadsafe({"event": "error", "where": "resource_recv",
|
|
"detail": str(e)})
|
|
|
|
async def serve(self):
|
|
sock_path = self.args.socket
|
|
if os.path.exists(sock_path):
|
|
os.unlink(sock_path)
|
|
# limit: asyncio's default per-line StreamReader cap is 64KiB, but
|
|
# send_resource requests carry the whole payload base64-encoded on one
|
|
# JSON line — a ~48KB+ attachment overflows the default and the raised
|
|
# LimitOverrunError tears down the RPC connection (the Rust side then
|
|
# sees "reticulum-daemon is gone" and restarts the whole session).
|
|
server = await asyncio.start_unix_server(
|
|
self._handle_client, path=sock_path, limit=16 * 1024 * 1024
|
|
)
|
|
os.chmod(sock_path, 0o600)
|
|
self.announce()
|
|
async with server:
|
|
await server.serve_forever()
|
|
|
|
|
|
class _AnnounceHandler:
|
|
"""Surfaces every heard LXMF delivery announce to the Rust side."""
|
|
aspect_filter = "lxmf.delivery"
|
|
|
|
def __init__(self, daemon: "ReticulumDaemon"):
|
|
self.daemon = daemon
|
|
self.receive_path_responses = True
|
|
|
|
def received_announce(self, destination_hash, announced_identity, app_data):
|
|
# Decode what we can here (both the LXMF-standard display name and our
|
|
# appended ARCHY identity blob — see _announce_app_data) so the Rust
|
|
# side gets clean typed fields instead of re-implementing msgpack.
|
|
display_name = None
|
|
archy_blob = None
|
|
raw = app_data or b""
|
|
try:
|
|
import LXMF
|
|
display_name = LXMF.display_name_from_app_data(raw)
|
|
# A legacy blob-only announce is plain ascii, so LXMF's decoder
|
|
# returns the whole ARCHY identity blob as a "name" — drop it.
|
|
if display_name and display_name.startswith("ARCHY:"):
|
|
display_name = None
|
|
except Exception:
|
|
display_name = None
|
|
try:
|
|
if raw[:1] and ((0x90 <= raw[0] <= 0x9F) or raw[0] == 0xDC):
|
|
import RNS.vendor.umsgpack as msgpack
|
|
peer_data = msgpack.unpackb(raw)
|
|
if isinstance(peer_data, list):
|
|
for el in peer_data[3:]:
|
|
if isinstance(el, bytes) and el.startswith(b"ARCHY:"):
|
|
archy_blob = el.decode("ascii", "ignore")
|
|
break
|
|
elif raw.startswith(b"ARCHY:"):
|
|
# Legacy (pre-upgrade archy node): app_data IS the blob.
|
|
archy_blob = raw.decode("ascii", "ignore")
|
|
except Exception:
|
|
archy_blob = None
|
|
self.daemon._emit_threadsafe({
|
|
"event": "announce",
|
|
"dest_hash": destination_hash.hex(),
|
|
"app_data": raw.hex(),
|
|
"display_name": display_name,
|
|
"archy_blob": archy_blob,
|
|
})
|
|
|
|
|
|
# ─────────────────────────────── entrypoints ─────────────────────────────────
|
|
|
|
def _parse_args(argv):
|
|
p = argparse.ArgumentParser(description="Archipelago Reticulum daemon")
|
|
p.add_argument("--identity-key", required=True, help="path to Archy node_key (32-byte ed25519 seed)")
|
|
p.add_argument("--socket", default="/tmp/archy-reticulum.sock", help="Unix RPC socket path")
|
|
p.add_argument("--rns-config", default=str(Path.home() / ".archy-reticulum"), help="RNS config/storage dir")
|
|
p.add_argument("--serial-port", help="RNode serial device, e.g. /dev/reticulum-radio")
|
|
p.add_argument("--tcp-listen", default=None, metavar="HOST:PORT",
|
|
help="Bind a plain-TCP Reticulum server interface (dev/verification "
|
|
"only; loopback-only, e.g. 127.0.0.1:4242).")
|
|
p.add_argument("--tcp-connect", action="append", default=None, metavar="HOST:PORT",
|
|
help="Dial a plain-TCP Reticulum client interface (repeatable).")
|
|
p.add_argument("--display-name", default="Archy", help="LXMF display name")
|
|
p.add_argument("--archy-ed-pubkey-hex", default=None,
|
|
help="Archy ed25519 pubkey hex (64 chars) — embedded in the announce "
|
|
"app_data as ARCHY:2:... so peers bind this RNS destination onto "
|
|
"the existing Archy contact. Omit for a plain display-name announce.")
|
|
p.add_argument("--archy-x25519-pubkey-hex", default=None,
|
|
help="Archy x25519 pubkey hex (64 chars), paired with --archy-ed-pubkey-hex.")
|
|
# LoRa profile (defaults are EU_868-ish; settled by the Phase-0 spike)
|
|
p.add_argument("--frequency", type=int, default=869525000)
|
|
p.add_argument("--bandwidth", type=int, default=125000)
|
|
p.add_argument("--txpower", type=int, default=17)
|
|
p.add_argument("--spreadingfactor", type=int, default=8)
|
|
p.add_argument("--codingrate", type=int, default=5)
|
|
p.add_argument("--no-radio", action="store_true", help="bring up with no RNode (selftest)")
|
|
p.add_argument("--check", action="store_true", help="print derived dest hash and exit (no RNS)")
|
|
p.add_argument("--selftest", action="store_true", help="bring up RNS+LXMF with no radio, verify, exit")
|
|
return p.parse_args(argv)
|
|
|
|
|
|
def _install_parent_death_signal() -> None:
|
|
"""Die when our parent process does.
|
|
|
|
The daemon ships as a PyInstaller one-file binary: our direct parent is the
|
|
bootloader, and the Rust supervisor (mesh/reticulum.rs) stops us by SIGKILL-
|
|
ing that bootloader. SIGKILL can't be forwarded, so without this the Python
|
|
child is orphaned and keeps holding the RNode serial port — which piles up
|
|
stale daemons that jam the radio (observed: 9 instances on one node). Asking
|
|
the kernel to send us SIGTERM on parent death lets our existing SIGTERM
|
|
handler shut down cleanly and free the port. Linux-only; no-op elsewhere.
|
|
"""
|
|
if sys.platform != "linux":
|
|
return
|
|
try:
|
|
import ctypes
|
|
|
|
PR_SET_PDEATHSIG = 1
|
|
ctypes.CDLL("libc.so.6", use_errno=True).prctl(
|
|
PR_SET_PDEATHSIG, signal.SIGTERM
|
|
)
|
|
except Exception:
|
|
pass # best-effort; never block startup on this
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
_install_parent_death_signal()
|
|
args = _parse_args(argv if argv is not None else sys.argv[1:])
|
|
|
|
if args.serial_port and (args.tcp_listen or args.tcp_connect):
|
|
raise SystemExit(
|
|
"--serial-port is mutually exclusive with --tcp-listen/--tcp-connect "
|
|
"(one daemon process = one interface)"
|
|
)
|
|
|
|
if args.check:
|
|
seed = ReticulumDaemon._read_seed(Path(args.identity_key))
|
|
print(lxmf_destination_hash(seed).hex())
|
|
return 0
|
|
|
|
daemon = ReticulumDaemon(args)
|
|
|
|
if args.selftest:
|
|
args.no_radio = True
|
|
daemon.bring_up()
|
|
# Announce app_data round-trip: the LXMF-standard msgpack name must be
|
|
# decodable by stock clients AND (with archy keys set) the appended
|
|
# identity blob must survive as an extra list element — this is the
|
|
# exact wire contract the Rust announce handler and Sideband both
|
|
# depend on, so verify it here where there's a real router to build it.
|
|
import LXMF as _LXMF
|
|
import RNS.vendor.umsgpack as _msgpack
|
|
args.archy_ed_pubkey_hex = args.archy_ed_pubkey_hex or "ab" * 32
|
|
args.archy_x25519_pubkey_hex = args.archy_x25519_pubkey_hex or "cd" * 32
|
|
app_data = daemon._announce_app_data()
|
|
decoded_name = _LXMF.display_name_from_app_data(app_data)
|
|
assert decoded_name == args.display_name, (
|
|
f"announce name round-trip failed: {decoded_name!r} != {args.display_name!r}"
|
|
)
|
|
blob_elems = [e for e in _msgpack.unpackb(app_data)[3:]
|
|
if isinstance(e, bytes) and e.startswith(b"ARCHY:")]
|
|
assert blob_elems, "identity blob missing from announce app_data"
|
|
# Live rename: set_name must change what the next announce carries.
|
|
daemon.delivery_destination.display_name = "selftest-renamed"
|
|
renamed = _LXMF.display_name_from_app_data(daemon._announce_app_data())
|
|
assert renamed == "selftest-renamed", f"rename round-trip failed: {renamed!r}"
|
|
print(f"selftest ok — dest_hash={daemon.dest_hash_hex} "
|
|
f"display_name={args.display_name!r} lxmf_router=up "
|
|
f"announce_app_data=verified set_name=verified")
|
|
return 0
|
|
|
|
for sig in (signal.SIGINT, signal.SIGTERM):
|
|
daemon.loop.add_signal_handler(sig, daemon.loop.stop)
|
|
try:
|
|
daemon.bring_up()
|
|
daemon.loop.run_until_complete(daemon.serve())
|
|
except RuntimeError:
|
|
pass # loop.stop() during serve_forever
|
|
finally:
|
|
if os.path.exists(args.socket):
|
|
os.unlink(args.socket)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|