433 lines
14 KiB
Python
433 lines
14 KiB
Python
"""Network browser — discovers NomadNet nodes and fetches pages via Reticulum.
|
|
|
|
Starts an RNS transport on FastAPI startup, listens for NomadNet node
|
|
announces, and exposes discovered nodes + remote page fetching via API.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Query
|
|
from starlette.responses import StreamingResponse
|
|
|
|
router = APIRouter()
|
|
log = logging.getLogger("browse")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Module-level state
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_nodes: dict[str, dict] = {} # hash_hex -> node info
|
|
_own_hash: str | None = None
|
|
_own_name: str = os.environ.get("NOMADNET_NODE_NAME", "Micronomicon")
|
|
_lock = threading.Lock()
|
|
_started = False
|
|
_subscribers: list[asyncio.Queue] = []
|
|
_sub_lock = threading.Lock()
|
|
_loop: asyncio.AbstractEventLoop | None = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RNS announce handler (must be an object with aspect_filter + method)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _push_node(node_data: dict) -> None:
|
|
"""Push a node update to all SSE subscribers (thread-safe)."""
|
|
with _sub_lock:
|
|
for q in list(_subscribers):
|
|
if _loop and _loop.is_running():
|
|
_loop.call_soon_threadsafe(q.put_nowait, node_data)
|
|
else:
|
|
try:
|
|
q.put_nowait(node_data)
|
|
except asyncio.QueueFull:
|
|
pass
|
|
|
|
|
|
class _AnnounceHandler:
|
|
"""RNS-compatible announce handler.
|
|
|
|
RNS.Transport.register_announce_handler() requires an object with:
|
|
- aspect_filter: str attribute
|
|
- received_announce(dest_hash, identity, app_data, ...): callable
|
|
"""
|
|
|
|
aspect_filter = "nomadnetwork.node"
|
|
|
|
def received_announce(
|
|
self,
|
|
destination_hash: bytes,
|
|
announced_identity,
|
|
app_data: bytes | None,
|
|
**kwargs,
|
|
) -> None:
|
|
import RNS
|
|
|
|
hash_hex = RNS.hexrep(destination_hash, delimit=False)
|
|
|
|
name = hash_hex[:12]
|
|
if app_data:
|
|
try:
|
|
name = app_data.decode("utf-8")
|
|
except Exception:
|
|
pass
|
|
|
|
is_self = name == _own_name
|
|
|
|
# Determine which interface this announce arrived on
|
|
iface_name = None
|
|
try:
|
|
path_entry = RNS.Transport.path_table.get(destination_hash)
|
|
if path_entry and path_entry[5]: # IDX_PT_RVCD_IF = 5
|
|
iface_name = getattr(path_entry[5], "name", None)
|
|
except Exception:
|
|
pass
|
|
|
|
with _lock:
|
|
_nodes[hash_hex] = {
|
|
"hash": hash_hex,
|
|
"name": name,
|
|
"last_seen": time.time(),
|
|
"is_self": is_self,
|
|
"type": "node",
|
|
"interface": iface_name,
|
|
}
|
|
if is_self:
|
|
global _own_hash
|
|
_own_hash = hash_hex
|
|
|
|
log.info(
|
|
"Node announce: %s (%s) via %s%s",
|
|
name, hash_hex[:8], iface_name or "?",
|
|
" [self]" if is_self else "",
|
|
)
|
|
_push_node(_nodes[hash_hex])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RNS lifecycle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_reticulum = None
|
|
|
|
|
|
def _collect_interfaces() -> list[dict]:
|
|
"""Read active RNS interfaces and return them as node-like dicts."""
|
|
try:
|
|
import RNS
|
|
except ImportError:
|
|
return []
|
|
|
|
ifaces = []
|
|
for iface in RNS.Transport.interfaces:
|
|
name = getattr(iface, "name", str(iface))
|
|
iface_id = f"iface_{name}"
|
|
target = getattr(iface, "target_ip", None) or getattr(iface, "target_host", None)
|
|
port = getattr(iface, "target_port", None) or getattr(iface, "bind_port", None)
|
|
ifaces.append({
|
|
"hash": iface_id,
|
|
"name": name,
|
|
"last_seen": time.time(),
|
|
"is_self": False,
|
|
"type": "interface",
|
|
"online": getattr(iface, "online", False),
|
|
"target": f"{target}:{port}" if target and port else None,
|
|
"txb": getattr(iface, "txb", 0),
|
|
"rxb": getattr(iface, "rxb", 0),
|
|
"bitrate": getattr(iface, "bitrate", 0),
|
|
"clients": len(getattr(iface, "clients", None) or []) if hasattr(iface, "clients") else None,
|
|
})
|
|
return ifaces
|
|
|
|
|
|
def start_browser() -> None:
|
|
"""Initialize RNS and begin listening for NomadNet node announces."""
|
|
global _started, _loop, _reticulum
|
|
|
|
if _started:
|
|
return
|
|
|
|
try:
|
|
_loop = asyncio.get_event_loop()
|
|
except RuntimeError:
|
|
_loop = None
|
|
|
|
try:
|
|
import RNS
|
|
|
|
configdir = os.environ.get("RNS_CONFIG_DIR", None)
|
|
if configdir:
|
|
Path(configdir).mkdir(parents=True, exist_ok=True)
|
|
|
|
_reticulum = RNS.Reticulum(configdir=configdir)
|
|
|
|
RNS.Transport.register_announce_handler(_AnnounceHandler())
|
|
|
|
_started = True
|
|
log.info("RNS browser started (v%s)", RNS.__version__)
|
|
|
|
except Exception as exc:
|
|
log.warning("Failed to start RNS browser: %s", exc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_CONFIG_PATHS = {
|
|
"reticulum": lambda: Path(os.environ.get("RNS_SERVER_CONFIG_DIR", os.environ.get("RNS_CONFIG_DIR", str(Path.home() / ".reticulum")))) / "config",
|
|
"nomadnet": lambda: Path(os.environ.get("NOMADNET_CONFIG_DIR", str(Path.home() / ".nomadnetwork"))) / "config",
|
|
}
|
|
|
|
|
|
def _config_path(kind: str) -> Path:
|
|
resolver = _CONFIG_PATHS.get(kind)
|
|
if not resolver:
|
|
raise ValueError(f"Unknown config kind: {kind}")
|
|
return resolver()
|
|
|
|
|
|
def _restart_nomadnet() -> bool:
|
|
"""Restart the NomadNet container. Returns True on success."""
|
|
try:
|
|
from docker_utils import NOMADNET_CONTAINER
|
|
import docker
|
|
client = docker.from_env()
|
|
container = client.containers.get(NOMADNET_CONTAINER)
|
|
container.restart()
|
|
log.info("NomadNet container restarted")
|
|
return True
|
|
except Exception as exc:
|
|
log.info("NomadNet container not available: %s", exc)
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# API endpoints
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _build_snapshot() -> list[dict]:
|
|
"""Build a full snapshot: self node + interfaces + discovered nodes."""
|
|
with _lock:
|
|
nodes = list(_nodes.values())
|
|
|
|
# Ensure type field on all nodes
|
|
for n in nodes:
|
|
n.setdefault("type", "node")
|
|
|
|
if not any(n["is_self"] for n in nodes):
|
|
nodes.insert(0, {
|
|
"hash": _own_hash or "self",
|
|
"name": _own_name,
|
|
"last_seen": time.time(),
|
|
"is_self": True,
|
|
"type": "node",
|
|
})
|
|
|
|
# Add interfaces
|
|
nodes.extend(_collect_interfaces())
|
|
return nodes
|
|
|
|
|
|
@router.get("/browse/nodes")
|
|
async def list_nodes():
|
|
"""Return all discovered NomadNet nodes and interfaces."""
|
|
return _build_snapshot()
|
|
|
|
|
|
@router.get("/browse/nodes/stream")
|
|
async def stream_nodes():
|
|
"""SSE stream — pushes full snapshot then live node announces."""
|
|
queue: asyncio.Queue = asyncio.Queue(maxsize=64)
|
|
|
|
with _sub_lock:
|
|
_subscribers.append(queue)
|
|
|
|
async def event_generator():
|
|
try:
|
|
# Send full snapshot (self + interfaces + known nodes)
|
|
for entry in _build_snapshot():
|
|
yield f"data: {json.dumps(entry)}\n\n"
|
|
|
|
# Then stream new announces as they arrive
|
|
while True:
|
|
node = await queue.get()
|
|
yield f"data: {json.dumps(node)}\n\n"
|
|
except asyncio.CancelledError:
|
|
pass
|
|
finally:
|
|
with _sub_lock:
|
|
_subscribers.remove(queue)
|
|
|
|
return StreamingResponse(
|
|
event_generator(),
|
|
media_type="text/event-stream",
|
|
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
|
|
)
|
|
|
|
|
|
@router.get("/browse/page/{hash_hex}")
|
|
async def get_remote_page(hash_hex: str, path: str = Query("index.mu")):
|
|
"""Fetch a Micron page from a node.
|
|
|
|
For the user's own node reads from the local pages directory.
|
|
For remote nodes establishes an RNS link and requests the page.
|
|
"""
|
|
with _lock:
|
|
node = _nodes.get(hash_hex)
|
|
if (node and node.get("is_self")) or hash_hex == "self":
|
|
return _read_local_page(path)
|
|
|
|
content = await _request_remote_page(hash_hex, path)
|
|
if content is None:
|
|
return {"content": None, "error": "Could not reach node"}
|
|
return {"content": content}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Local page reader
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _read_local_page(path: str) -> dict:
|
|
from converter import execute_dynamic_script
|
|
|
|
pages_dir = os.environ.get("PAGES_DIR", str(Path.home() / ".nomadnetwork/storage/pages"))
|
|
filepath = Path(pages_dir) / path
|
|
if not filepath.exists():
|
|
return {"content": None, "error": "Page not found"}
|
|
try:
|
|
if os.access(filepath, os.X_OK):
|
|
script = filepath.read_text(encoding="utf-8")
|
|
return {"content": execute_dynamic_script(script)}
|
|
return {"content": filepath.read_text()}
|
|
except Exception as exc:
|
|
return {"content": None, "error": str(exc)}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Remote page fetcher (RNS link + request/response)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def _request_remote_page(hash_hex: str, path: str) -> str | None:
|
|
"""Establish an RNS link to a remote NomadNet node and request a page."""
|
|
try:
|
|
import RNS
|
|
except ImportError:
|
|
return None
|
|
|
|
loop = asyncio.get_event_loop()
|
|
future: asyncio.Future[str | None] = loop.create_future()
|
|
|
|
def _do_request():
|
|
try:
|
|
dest_hash = bytes.fromhex(hash_hex)
|
|
|
|
if not RNS.Transport.has_path(dest_hash):
|
|
RNS.Transport.request_path(dest_hash)
|
|
deadline = time.time() + 10
|
|
while time.time() < deadline:
|
|
if RNS.Transport.has_path(dest_hash):
|
|
break
|
|
time.sleep(0.1)
|
|
else:
|
|
loop.call_soon_threadsafe(future.set_result, None)
|
|
return
|
|
|
|
identity = RNS.Identity.recall(dest_hash)
|
|
if not identity:
|
|
loop.call_soon_threadsafe(future.set_result, None)
|
|
return
|
|
|
|
dest = RNS.Destination(
|
|
identity,
|
|
RNS.Destination.OUT,
|
|
RNS.Destination.SINGLE,
|
|
"nomadnetwork",
|
|
"node",
|
|
)
|
|
|
|
link = RNS.Link(dest)
|
|
|
|
deadline = time.time() + 15
|
|
while time.time() < deadline:
|
|
if link.status == RNS.Link.ACTIVE:
|
|
break
|
|
time.sleep(0.1)
|
|
else:
|
|
link.teardown()
|
|
loop.call_soon_threadsafe(future.set_result, None)
|
|
return
|
|
|
|
def on_response(request_receipt):
|
|
try:
|
|
resp = request_receipt.response
|
|
if resp is not None:
|
|
content = resp.decode("utf-8") if isinstance(resp, bytes) else str(resp)
|
|
loop.call_soon_threadsafe(future.set_result, content)
|
|
else:
|
|
loop.call_soon_threadsafe(future.set_result, None)
|
|
except Exception:
|
|
loop.call_soon_threadsafe(future.set_result, None)
|
|
finally:
|
|
link.teardown()
|
|
|
|
def on_failed(request_receipt):
|
|
if not future.done():
|
|
loop.call_soon_threadsafe(future.set_result, None)
|
|
link.teardown()
|
|
|
|
link.request(
|
|
"/page/" + path,
|
|
response_callback=on_response,
|
|
failed_callback=on_failed,
|
|
)
|
|
|
|
except Exception as exc:
|
|
log.warning("Remote page request failed: %s", exc)
|
|
if not future.done():
|
|
loop.call_soon_threadsafe(future.set_result, None)
|
|
|
|
threading.Thread(target=_do_request, daemon=True).start()
|
|
|
|
try:
|
|
return await asyncio.wait_for(future, timeout=30.0)
|
|
except asyncio.TimeoutError:
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Reticulum config endpoints
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@router.post("/browse/restart")
|
|
async def restart_services():
|
|
"""Restart NomadNet to apply config changes."""
|
|
restarted = _restart_nomadnet()
|
|
return {"ok": True, "nomadnet_restarted": restarted}
|
|
|
|
|
|
@router.get("/browse/config/{kind}")
|
|
async def get_config(kind: str):
|
|
"""Return config file contents for reticulum or nomadnet."""
|
|
path = _config_path(kind)
|
|
if not path.exists():
|
|
return {"content": ""}
|
|
return {"content": path.read_text(encoding="utf-8")}
|
|
|
|
|
|
@router.post("/browse/config/{kind}")
|
|
async def save_config(kind: str, body: dict):
|
|
"""Write config file for reticulum or nomadnet."""
|
|
content = body.get("content", "")
|
|
path = _config_path(kind)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content, encoding="utf-8")
|
|
return {"ok": True}
|