dtoro/micronomicon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: graph browser

Browse Source
This commit is contained in:
authentik Default Admin
2026-04-04 16:18:21 +02:00
parent 72ff02dbdf
commit 3132d40391
10 changed files with 735 additions and 443 deletions
Download Patch File Download Diff File Expand all files Collapse all files

218
backend/browse.py
View File

@@ -7,6 +7,7 @@ announces, and exposes discovered nodes + remote page fetching via API.
from __future__ import annotations
import asyncio
import json
import logging
import os
import threading
@@ -14,6 +15,7 @@ import time
from pathlib import Path
from fastapi import APIRouter, Query
from starlette.responses import StreamingResponse
router = APIRouter()
log = logging.getLogger("browse")
@@ -27,19 +29,136 @@ _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, _own_hash
global _started, _loop, _reticulum
if _started:
return
try:
_loop = asyncio.get_event_loop()
except RuntimeError:
_loop = None
try:
import RNS
@@ -47,13 +166,9 @@ def start_browser() -> None:
if configdir:
Path(configdir).mkdir(parents=True, exist_ok=True)
reticulum = RNS.Reticulum(configdir=configdir)
_reticulum = RNS.Reticulum(configdir=configdir)
# Register handler for NomadNet page-serving node announces
RNS.Transport.register_announce_handler(
_on_announce,
aspect_filter="nomadnetwork.node",
)
RNS.Transport.register_announce_handler(_AnnounceHandler())
_started = True
log.info("RNS browser started (v%s)", RNS.__version__)
@@ -62,61 +177,70 @@ def start_browser() -> None:
log.warning("Failed to start RNS browser: %s", exc)
def _on_announce(
destination_hash: bytes,
announced_identity,
app_data: bytes | None,
) -> None:
"""Handle an incoming NomadNet node announce."""
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
with _lock:
_nodes[hash_hex] = {
"hash": hash_hex,
"name": name,
"last_seen": time.time(),
"is_self": is_self,
}
if is_self:
global _own_hash
_own_hash = hash_hex
log.info("Node announce: %s (%s)%s", name, hash_hex[:8], " [self]" if is_self else "")
# ---------------------------------------------------------------------------
# API endpoints
# ---------------------------------------------------------------------------
@router.get("/browse/nodes")
async def list_nodes():
"""Return all discovered NomadNet page-serving nodes."""
def _build_snapshot() -> list[dict]:
"""Build a full snapshot: self node + interfaces + discovered nodes."""
with _lock:
nodes = list(_nodes.values())
# Ensure the user's own node is always present
# 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.
@@ -124,13 +248,11 @@ async def get_remote_page(hash_hex: str, path: str = Query("index.mu")):
For the user's own node reads from the local pages directory.
For remote nodes establishes an RNS link and requests the page.
"""
# Own node → read from disk
with _lock:
node = _nodes.get(hash_hex)
if (node and node.get("is_self")) or hash_hex == "self":
return _read_local_page(path)
# Remote node → RNS request
content = await _request_remote_page(hash_hex, path)
if content is None:
return {"content": None, "error": "Could not reach node"}
@@ -168,7 +290,6 @@ async def _request_remote_page(hash_hex: str, path: str) -> str | None:
try:
dest_hash = bytes.fromhex(hash_hex)
# Ensure path to destination is known
if not RNS.Transport.has_path(dest_hash):
RNS.Transport.request_path(dest_hash)
deadline = time.time() + 10
@@ -195,7 +316,6 @@ async def _request_remote_page(hash_hex: str, path: str) -> str | None:
link = RNS.Link(dest)
# Wait for link to become active
deadline = time.time() + 15
while time.time() < deadline:
if link.status == RNS.Link.ACTIVE:
@@ -206,7 +326,6 @@ async def _request_remote_page(hash_hex: str, path: str) -> str | None:
loop.call_soon_threadsafe(future.set_result, None)
return
# Request the page via NomadNet's protocol
def on_response(request_receipt):
try:
resp = request_receipt.response
@@ -236,7 +355,6 @@ async def _request_remote_page(hash_hex: str, path: str) -> str | None:
if not future.done():
loop.call_soon_threadsafe(future.set_result, None)
# Run blocking RNS operations in a thread
threading.Thread(target=_do_request, daemon=True).start()
try: