""" MeshBay Node daemon — main process. Startup sequence: 1. Load config (~/.config/meshbay/node.toml) 2. Load or create keystore (Argon2id unlock) 3. Connect to hub: register → login → announce node 4. Fetch GEK bundle from hub (if group configured) 5. Start directory indexer (watchdog) 6. Create chat stores (one SQLite DB per group) 7. Create WebRTC transport (browser + native clients via DataChannel) 8. Start QUIC chunk server (LAN / port-forwarded / hub-less direct access) 9. (Phase 11.5: the unauthenticated HTTP file API and the TCP+TLS server were removed) 10. Start hub WebSocket (signaling, revocations, WebRTC offers) 11. Start local control API on node.ui_port (loopback only, token-gated) 12. Run until SIGINT/SIGTERM Usage: meshbay-node # interactive password prompt meshbay-node --config /path # custom config meshbay-node status # node state + public key (works while stopped) meshbay-node gek-init # initialise the group key (no browser needed) meshbay-node init # write example config + create keystore meshbay-node --calibrate-argon2 # benchmark Argon2id, suggest parameters """ import asyncio import base64 import logging import os import signal import sys import time from dataclasses import asdict, replace from pathlib import Path import uvicorn from meshbay_common import MNP_VERSION from meshbay_common.background import spawn from meshbay_common.paths import fold from meshbay_common.protocol import MNP from meshbay_node import uploads as uploads_mod from meshbay_node.audit import RETENTION_DAYS as AUDIT_RETENTION_DAYS from meshbay_node.audit import AuditStore from meshbay_node.bundle_store import BundleStore from meshbay_node.chat.store import ChatStore from meshbay_node.config import DEFAULT_CONFIG_PATH, Config, load_config from meshbay_node.hub_client import HubClient, HubConfig from meshbay_node.indexer import DirectoryIndexer, GroupIndex, IndexCache from meshbay_node.indexer.enrich import Enricher from meshbay_node.indexer.enrich_audio import AudioEnricher from meshbay_node.indexer.enrich_photo import PhotoEnricher from meshbay_node.keystore import create_keystore, load_keystore, load_or_create_keystore from meshbay_node.media_cache import MediaCache from meshbay_node.musicbrainz import MusicBrainzClient from meshbay_node.platform import chmod_private, config_dir, data_dir, state_dir from meshbay_node.roots import RootError, RootSet, entry_abs_path, off_disk from meshbay_node.roster import Roster from meshbay_node.tmdb import TmdbClient from meshbay_node.transport import ( QUIC_AVAILABLE, WEBRTC_AVAILABLE, Denylist, ) from meshbay_node.transport.wire import index_delta_message, index_sync_message if QUIC_AVAILABLE: from meshbay_node.transport import QuicChunkServer if WEBRTC_AVAILABLE: from meshbay_node.transport import WebRTCTransport log = logging.getLogger(__name__) def _under_any_directory(path: str, directories: list[str]) -> bool: """ Whether an entry's folder is one of an app's directories, or inside one. Mirrors `underAnyDirectory` in the SPA's app modules. One helper for every app since they all take a list: Videos and Music used to take a single folder and had a function each saying the same thing, which is how the two came to differ in what they did with a trailing slash. """ path = path or "" return any(path == d or path.startswith(d + "/") for d in directories) def _owning_directory(path: str, directories: list[str]) -> str | None: """ Which of an app's directories an entry belongs to — the deepest match. Deepest, because directories may nest: with both `Media` and `Media/Albums` configured, a file under the second belongs to the second. Taking the first match instead would measure it against a boundary one level too shallow, which for Music is the difference between reading a folder as an artist and reading it as a release. """ path = path or "" best: str | None = None for d in directories: if path == d or path.startswith(d + "/"): if best is None or len(d) > len(best): best = d return best # ── Argon2id calibration ────────────────────────────────────────────────────── def calibrate_argon2(target_ms: int = 500) -> None: """Benchmark Argon2id and suggest parameters targeting ~target_ms.""" import os import time print(f"Calibrating Argon2id (target: {target_ms}ms) ...") salt = os.urandom(16) for mem in [65536, 131072, 262144, 524288]: from cryptography.hazmat.primitives.kdf.argon2 import Argon2id t0 = time.perf_counter() Argon2id(salt=salt, length=32, iterations=3, lanes=4, memory_cost=mem).derive(b"benchmark") elapsed_ms = (time.perf_counter() - t0) * 1000 print(f" memory_cost={mem:>7} ({mem//1024:>4}MB): {elapsed_ms:.0f}ms", end="") if abs(elapsed_ms - target_ms) < target_ms * 0.3: print(" ← recommended") else: print() print("Set memory_cost in meshbay_common/crypto.py: ARGON2_MEMORY_COST") # ── Hub WS sender bridge ───────────────────────────────────────────────────── class _WsSender: """Thin bridge so WebRTC context can call hub_ws.send() for chat_notify.""" def __init__(self, hub_client: HubClient): self._hub = hub_client async def send(self, data: str) -> None: await self._hub.send_ws(data) # ── Daemon ──────────────────────────────────────────────────────────────────── def _root_shape(roots) -> set[tuple]: """What has to match for a group's roots to count as unchanged on reload.""" return {(r.name, str(r.path), r.writable, r.removable) for r in roots} class NodeDaemon: def __init__(self, config: Config, config_path: Path = DEFAULT_CONFIG_PATH): self._config = config self._config_path = config_path self._state: dict = { "status": "starting", "hub_url": config.hub.url, "username": config.hub.username, "groups": [g.name for g in config.groups], "quic_port": config.node.quic_port, "endpoint_hint": None, "indexes": {}, "indexers": {}, } self._quic_server = None self._webrtc = None # Persisted so a restart does not silently un-revoke everyone (H4) self._denylist = ( Denylist(path=config.data_dir / "denylist.json") if Denylist else None) self._chat_stores: dict[str, ChatStore] = {} # One instance, shared by every group's DirectoryIndexer — see # indexer/cache.py's docstring for why this stopped being per-group. self._index_cache: IndexCache | None = None # Coalesces a burst of index changes (one per debounced watchdog # event) into a single broadcast — see _on_index_change. 0.5s is # short enough nobody notices the wait, long enough that dropping a # few hundred files into a watched folder produces one push instead # of one per file. self._broadcast_coalesce_secs = 0.5 self._pending_broadcasts: dict[str, asyncio.TimerHandle] = {} # group_id -> (version, {id: entry}) as of the last thing actually # broadcast — the comparison point for the next delta. self._last_broadcast_snapshot: dict[str, tuple] = {} self._audit_store: AuditStore | None = None self._bundle_store: BundleStore | None = None self._media_cache: MediaCache | None = None self._enricher: Enricher | None = None self._tmdb_client: TmdbClient | None = None self._audio_enricher: AudioEnricher | None = None self._musicbrainz_client: MusicBrainzClient | None = None self._photo_enricher: PhotoEnricher | None = None # A file id attempted at most once per daemon run, success or # failure — a persistently unprobeable file (corrupt, still being # written) does not get re-queued on every coalesced broadcast. A # restart retries everything, matching the "disposable, rebuildable" # stance the rest of this cache takes (docs/MESHBAY_DESIGN.md §6.5). # Shared across the video and audio enrichment paths — content- # addressed ids never collide between the two. Keyed by # (group_id, entry.id), not entry.id alone: the id is a content # hash, so the same physical file shared into two different groups # (found live — overlapping test libraries across several demo # groups) produces the same id in both. A bare-id set marked the # second group's copy "already attempted" the moment the first # group's enrichment ran, even though nothing had ever populated # *that* group's own index — every file in the second group stayed # at duration 0 with no artist/album, permanently, since nothing # ever revisits an id already in this set. self._enriched_attempted: set[tuple[str, str]] = set() self._roster: Roster | None = None self._indexers: list[DirectoryIndexer] = [] self._tasks: list[asyncio.Task] = [] self._hub: HubClient | None = None self._reload_lock = asyncio.Lock() async def run(self) -> None: log.info("MeshBay Node starting up") # 1. Keystore keys = load_or_create_keystore( path=self._config.keystore.path, unlock_file=self._config.keystore.unlock_file, ) log.info("Keys loaded: %s", keys.pk_ed25519_b64[:16]) # 2. Start the local control API early (so the operator can read the # node key before hub login). It is JSON-only, loopback-only, and both # the CLI and the desktop client's Node page are its clients. self._state["pk_node_ed25519"] = keys.pk_ed25519_b64 self._state["config"] = self._config # Where it came from, so `group add` appends to the file this process # actually read rather than guessing at the default. self._state["config_path"] = str(self._config_path) # Per-run token for the control API (11.5.3). Not a password: it keeps # other local processes and rebound browser pages out of an API that can # re-initialise group keys. ui_token = base64.urlsafe_b64encode(os.urandom(18)).decode().rstrip("=") self._state["ui_token"] = ui_token # Persisted so the CLI and the desktop client can read it — nobody # should ever copy a token out of a log or a terminal. self._config.data_dir.mkdir(parents=True, exist_ok=True) self._ui_token_file = self._config.data_dir / "ui-token" self._ui_token_file.write_text(ui_token, encoding="utf-8", newline="\n") chmod_private(self._ui_token_file) from meshbay_node.ui import create_ui_app ui_app = create_ui_app(self._state) ui_cfg = uvicorn.Config( ui_app, host="127.0.0.1", port=self._config.node.ui_port, log_level="warning", ) ui_server = uvicorn.Server(ui_cfg) self._tasks.append(asyncio.create_task(ui_server.serve())) log.info("Control API on 127.0.0.1:%d", self._config.node.ui_port) self._tasks.append(asyncio.create_task(self._reap_partial_uploads())) # 3. Hub connection (Ed25519 auth — retries until node key is linked) hub_cfg = HubConfig( hub_url=self._config.hub.url, username=self._config.hub.username, ) async with HubClient(hub_cfg, keys) as hub: self._hub = hub session = await self._login_with_retry(hub) self._state["endpoint_hint"] = session.node_id try: session.email = await hub.fetch_owner_email() except Exception as e: log.warning("Could not fetch owner email: %s", e) # 4. Bundle store (P2P GEK bundles) data_dir = self._config.data_dir data_dir.mkdir(parents=True, exist_ok=True) self._bundle_store = BundleStore(db_path=data_dir / "bundles.db") await self._bundle_store.open() log.info("Bundle store opened: %s", data_dir / "bundles.db") # 4a. Path->hash cache, node-wide — opened once, shared by every # group's DirectoryIndexer below (indexer/cache.py). self._index_cache = IndexCache(db_path=data_dir / "index_cache.db") await self._index_cache.open() self._state["index_cache"] = self._index_cache log.info("Index cache opened: %s", data_dir / "index_cache.db") # 4b. Roster — who this node recognises and which keys are theirs. # Node authority is established here, locally, and never learned from # the hub: a hub that could name the operator's key could install # itself as node administrator. self._roster = Roster(db_path=data_dir / "roster.db") await self._roster.open() await self._roster.purge_expired() # Apply any roster overrides to node config (panel-edited values # take precedence over node.toml defaults). from meshbay_node.config import node_settings_defaults nd = self._config.node effective = await self._roster.node_settings( node_settings_defaults(nd)) for k, v in effective.items(): setattr(nd, k, v) # X25519 key material for GEK unwrapping from cryptography.hazmat.primitives import serialization sk_x_raw = keys.sk_x25519.private_bytes( serialization.Encoding.Raw, serialization.PrivateFormat.Raw, serialization.NoEncryption()) pk_x_raw = base64.b64decode(keys.pk_x25519_b64) # 4. Build per-group contexts groups_ctx: dict[str, dict] = {} for group_cfg in self._config.groups: if not group_cfg.id or not group_cfg.roots: log.warning("Group %r has no id or no shared directory — " "skipping", group_cfg.name) continue try: roots = await self._build_roots(group_cfg) except RootError as e: # Configuration the operator has to fix; guessing would put # a member's file on the wrong disk or index one twice. log.error("Group %r: %s — skipping", group_cfg.name, e) continue await off_disk(roots, roots.refresh_availability) if not any(r.available for r in roots): # Not skipped for being empty: a group whose only drive is # unplugged still exists, and its index is frozen rather # than lost. But there is nothing to serve until it returns. log.warning( "Group %r: none of its %d root(s) are readable right now " "(%s) — serving nothing until one returns", group_cfg.name, len(roots), ", ".join(str(r.path) for r in roots)) gek = None gek = await self._load_gek( group_cfg.id, session.user_id, sk_x_raw, pk_x_raw) if gek: log.info("GEK loaded for group %s", group_cfg.id[:8]) else: log.info("No GEK yet for group %s — will accept first setup", group_cfg.name) # Read once at load, like enabled_apps below — # kept current in place afterwards by set_scan_settings # (ops.py), which updates both this indexer object directly # and roster.db, so a restart picks up the same values. scan_settings = ( await self._roster.scan_settings(group_cfg.id) if self._roster else { "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS, "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS, }) indexer = DirectoryIndexer( roots=roots, group_id=group_cfg.id, sk_node=keys.sk_ed25519, gek=gek, on_change=self._on_index_change, on_root_ejected=self._eject_persister(group_cfg.id), cache=self._index_cache, reconcile_secs=scan_settings["reconcile_interval_secs"], debounce_secs=scan_settings["debounce_secs"], ) await indexer.start(defer_scan=True) self._indexers.append(indexer) self._state["indexes"][group_cfg.id] = indexer.index self._state["indexers"][group_cfg.id] = indexer log.info("Group %s configured: %s (scan deferred)", group_cfg.name, ", ".join(f"{r.name}={r.path}" for r in roots)) groups_ctx[group_cfg.id] = { "gek": gek, "roots": roots, "index": indexer.index, # Live reference, mutated in place by the indexer itself # (see IndexProgress in indexer.py) — read, never copied, # by the handshake ack and the periodic progress pusher. "progress": indexer.progress, # Bound method, called when a peer completes the # handshake — resets reconcile's backoff (indexer.py # _reconcile_loop) so the backstop is prompt again now # that someone is actually looking. "note_activity": indexer.note_activity, # Bound method, called when an upload finishes. The entry # it belongs to does not exist yet (see # webrtc_server._register_uploader), so the indexer keeps # the record and stamps the entry when it creates it. "record_upload": indexer.record_upload, # Shown to the operator in Settings, and kept current in # place by set_scan_settings (ops.py) — same reasoning as # enabled_apps below. "reconcile_interval_secs": scan_settings["reconcile_interval_secs"], "debounce_secs": scan_settings["debounce_secs"], "visibility": group_cfg.visibility, # Admission policy comes from node.toml, never from the hub: # a hub that could declare a group open would be handed its key. "join_policy": group_cfg.join_policy, # Read once at load, kept current in place by the signed # operation that changes it — the upload handler is # synchronous and a database round trip per chunk would be # absurd. (Whether a member may upload is not here any # more: it is `writable` on the root being written to, # which the RootSet above already carries.) "enabled_apps": await self._roster.enabled_apps( group_cfg.id) if self._roster else list(Roster.DEFAULT_APPS), # How many transfers one member may run at once here. Empty # means the operator has not said, and the node's default # applies — never "unlimited" (transfers.member_cap). "transfer_limits": ( await self._roster.transfer_limits(group_cfg.id) if self._roster else {}), # Which folder(s) inside the shared roots each app works # over. One shape for every app (roster.py's # app_directories) — an empty list means nothing has been # chosen, which every app reads as "show nothing yet", # never "the whole group index". **(await self._app_directories_ctx(group_cfg.id)), # Whether the node unfurls links members post here. "chat_link_preview": await self._roster.chat_link_preview( group_cfg.id) if self._roster else True, # Whether members' cross-group Search lists this group. "search_listed": await self._roster.search_listed( group_cfg.id) if self._roster else True, # Which chat epoch key is current. Opened here if the # group has none, because chat is always encrypted (MNP # 2.0) and a group with no epoch is a group nobody can # speak in — the node cannot wait for an operator to notice. "chat_epoch": await self._ensure_chat_epoch(group_cfg.id), # Whether TMDB lookups run for this group at all — # per-group (2026-08-24, used to be node-wide), same # "read once, kept current in place by the signed op" # shape as the app directories above. "tmdb_enabled": await self._roster.tmdb_enabled( group_cfg.id) if self._roster else True, # Music app equivalent of tmdb_enabled — per-group from # the start (docs/MESHBAY_DESIGN.md §9.8). "musicbrainz_enabled": await self._roster.musicbrainz_enabled( group_cfg.id) if self._roster else True, } if not groups_ctx: log.warning("No groups configured yet — the control API and hub " "connection stay up; attach a group to go live") # 5. Chat stores (one SQLite DB per group) for gid in groups_ctx: chat_db = data_dir / gid[:16] / "chat.db" store = ChatStore(db_path=chat_db) await store.open() self._chat_stores[gid] = store groups_ctx[gid]["chat_store"] = store log.info("Chat stores opened: %d groups", len(self._chat_stores)) # 6. Audit store (legal compliance — IP + action logging) audit_db = data_dir / "audit.db" self._audit_store = AuditStore(db_path=audit_db) await self._audit_store.open() log.info("Audit store opened: %s", audit_db) self._tasks.append(asyncio.create_task(self._purge_audit_log())) # 6b. Media cache (Videos app — TMDB metadata + thumbnails). # Node-wide like audit.db, not per-group: a thumbnail is the same # bytes regardless of which group happens to share the file # (docs/MESHBAY_DESIGN.md §6.5, §9.7). media_cache_db = data_dir / "media_cache.db" self._media_cache = MediaCache(db_path=media_cache_db) await self._media_cache.open() self._enricher = Enricher(self._media_cache) self._tmdb_client = TmdbClient(roster=self._roster) # Token/language only — read once at load, kept current in place # by ops.set_tmdb_config (the signed op), exposed to every # group's handshake ack via `daemon_state` (already wired to # self._webrtc._ctx below) since these stay node-wide, one # shared credential/cache. Whether TMDB is used at all is now # per-group instead — see each group's own "tmdb_enabled" in # groups_ctx above. tmdb_token, tmdb_language = await self._roster.tmdb_config() self._state["tmdb_token_customized"] = bool(tmdb_token) self._state["tmdb_language"] = tmdb_language or "" # 6c. Music app (docs/MESHBAY_DESIGN.md §9.8) — same # media_cache.db, its own enricher (mutagen, not ffmpeg) and its # own MusicBrainz client. # The User-Agent contact is the owner's hub email, resolved at # login — no roster setting or env var needed any more. self._audio_enricher = AudioEnricher(self._media_cache) self._musicbrainz_client = MusicBrainzClient(owner_email=session.email) self._state["musicbrainz_contact_configured"] = bool(session.email) # 6d. Photos app (docs/MESHBAY_DESIGN.md §9.9) — same # media_cache.db, its own enricher (Pillow, not ffmpeg/mutagen). # No credential, no # third-party client to construct: EXIF is read locally. self._photo_enricher = PhotoEnricher(self._media_cache) self._state["media_cache"] = self._media_cache log.info("Media cache opened: %s", media_cache_db) # 5. Denylist denylist = self._denylist # 6. WebRTC transport (browser clients) # # The operator's answer about the GPU, before the first stream asks # the question. The probe itself is lazy — it costs a test encode, # and a node that never serves a video should never pay it. from meshbay_node import hwaccel hwaccel.set_enabled(self._config.node.hardware_video_encode) from meshbay_node.transport.ice_filter import install as install_ice_filter install_ice_filter( self._config.node.ice_interfaces or None, ) # aiortc keeps only the FIRST entry of RTCConfiguration.iceServers, so # the multi-STUN fallback only exists on the node if aioice itself # fans out — see transport/stun_multi. from meshbay_node.transport.stun_multi import install as install_stun_multi install_stun_multi(self._config.node.stun_servers or None) first = next(iter(groups_ctx.values()), None) if WEBRTC_AVAILABLE: self._webrtc = WebRTCTransport( sk_node=keys.sk_ed25519, hub_pk_pem=session.hub_pk_pem, gek=first["gek"] if first else None, roots=first["roots"] if first else None, index=first["index"] if first else None, groups=groups_ctx, denylist=denylist, max_concurrent_streams=self._config.node.max_concurrent_streams, max_concurrent_downloads=self._config.node.max_concurrent_downloads, max_concurrent_uploads=self._config.node.max_concurrent_uploads, max_upload_gb=self._config.node.max_upload_gb, transcode_incompatible_video=self._config.node.transcode_incompatible_video, stun_servers=self._config.node.stun_servers or None, ) # No global chat_store here: each group's store lives in # groups_ctx[gid]["chat_store"] and is resolved per session via # _group_ctx(). Assigning the first group's store transport-wide # sent every group's chat to one database and served it back to # members of every other group (finding H1). self._webrtc._ctx["groups"] = groups_ctx self._webrtc._ctx["hub_ws"] = _WsSender(hub) self._webrtc._ctx["node_user_id"] = session.user_id self._webrtc._ctx["audit_store"] = self._audit_store self._webrtc._ctx["bundle_store"] = self._bundle_store self._webrtc._ctx["media_cache"] = self._media_cache self._webrtc._ctx["tmdb_client"] = self._tmdb_client self._webrtc._ctx["musicbrainz_client"] = self._musicbrainz_client self._webrtc._ctx["sk_x25519_raw"] = sk_x_raw self._webrtc._ctx["pk_x25519_raw"] = pk_x_raw self._webrtc._ctx["pk_x25519_b64"] = keys.pk_x25519_b64 self._webrtc._ctx["roster"] = self._roster # The MNP adapter calls the same operations as the loopback API # (meshbay_node.ops), and those take the daemon's state. Handing # the transport a second set of lookups is how two paths to one # operation start disagreeing — the shape of C1 and C6. self._webrtc._ctx["daemon_state"] = self._state self._webrtc._ctx["invite_ttl"] = ( self._config.node.invite_ttl_hours * 3600) self._webrtc._ctx["device_request_ttl"] = ( self._config.node.device_request_ttl_minutes * 60) paired = await self._roster.has_operator() if self._roster else False self._webrtc._ctx["has_admin_authority"] = paired if paired: log.info("Node authority: paired operator") else: log.warning( "No operator paired — invites and file deletion are " "refused. Run: meshbay-node operator pair") log.info("WebRTC transport ready") else: log.warning("WebRTC not available (aiortc not installed)") # 7. QUIC chunk server (LAN / port-forwarded / hub-less direct access) # # Off unless `[node] quic_enabled = true`: no shipping client speaks # QUIC (browser and desktop use WebRTC; the `group://` sidecar is # unbuilt), so starting it by default only exposes a UDP port. if QUIC_AVAILABLE and self._config.node.quic_enabled: self._quic_server = QuicChunkServer( sk_node=keys.sk_ed25519, hub_pk_pem=session.hub_pk_pem, gek=first["gek"] if first else None, roots=first["roots"] if first else None, index=first["index"] if first else None, host="::", port=self._config.node.quic_port, groups=groups_ctx, denylist=denylist, ) self._quic_server._ctx["groups"] = groups_ctx await self._quic_server.start() log.info("QUIC server on port %d (%d groups)", self._config.node.quic_port, len(groups_ctx)) elif QUIC_AVAILABLE: log.info("QUIC server disabled ([node] quic_enabled = false)") # 8. Hub WebSocket (signaling + revocations + WebRTC offers) async def on_webrtc_offer(sdp, peer_id, ice_candidates): if not self._webrtc: return None try: answer_sdp, answer_ice = await self._webrtc.handle_offer( sdp, peer_id) log.info("WebRTC answer for peer=%s (%d peers)", peer_id, self._webrtc.active_peers) return (answer_sdp, answer_ice) except Exception as e: log.error("WebRTC offer failed: %s", e) return None async def on_incoming(peer_ip, peer_port): if self._quic_server: self._quic_server.punch_nat(peer_ip, peer_port) def on_revocation(token): if denylist and token: import jwt as _jwt try: payload = _jwt.decode( token, session.hub_pk_pem, algorithms=["EdDSA"], options={"verify_exp": False}) target = payload.get("target") tid = payload.get("target_id", "") if target == "user": denylist.deny_user(tid) elif target == "group": # H4: previously dropped on the floor, so "suspend a # group" was a hub-only gesture that no node enforced. denylist.deny_group(tid) self._drop_group_sessions(tid) elif target == "jti": denylist.deny_jti(tid) else: log.warning("Unknown revocation target: %r", target) except Exception as e: log.warning("Invalid revocation token: %s", e) ws_task = asyncio.create_task(hub.maintain_ws( on_incoming=on_incoming, on_revocation=on_revocation, on_webrtc_offer=on_webrtc_offer, group_ids=lambda: list((self._state.get("groups_ctx") or {}).keys()), )) self._tasks.append(ws_task) log.info("Hub WS task started") # 9. (removed in Phase 11.5) The per-group HTTP file API used to start here. # It served the Mesh Group Index and raw plaintext files on 0.0.0.0 with no # authentication, for private groups too — finding C1. Every client path now # goes through the MNP handshake (JWT + group claim + GEK proof). # 10. Update control API state (already running from step 2) self._state["groups_ctx"] = groups_ctx self._state["audit_store"] = self._audit_store self._state["bundle_store"] = self._bundle_store self._state["roster"] = self._roster self._state["node_user_id"] = session.user_id # The node's own Ed25519 key. Needed by `encrypt_chat_history`, # which seals migrated messages under a synthetic device of the # node's rather than pretending to hold a member's signing key. self._state["sk_node"] = keys.sk_ed25519 self._state["webrtc"] = self._webrtc self._state["quic_server"] = self._quic_server self._state["hub"] = hub self._state["reload_fn"] = self._reload_config # Keyed by app, so `ops.set_app_directories` finds the right # sweep without knowing which apps exist — an app with nothing to # enrich simply has no entry. self._state["enrich_app_dirs_fns"] = { "video": self._enrich_video_root_now, "music": self._enrich_audio_root_now, "photo": self._enrich_photo_roots_now, } # Rotating a key has to reach every transport holding a copy of it, # and clearing the denylist has to reach the one the handshake # consults — so both are published rather than reachable only # through the object that happens to own them. self._state["denylist"] = self._denylist self._state["pk_x25519_raw"] = pk_x_raw self._state["sk_x25519_raw"] = sk_x_raw self._state["sk_ed25519"] = keys.sk_ed25519 self._state["status"] = "running" log.info("Node ready — %d groups, WebRTC=%s, QUIC=%s", len(groups_ctx), "yes" if self._webrtc else "no", "yes" if self._quic_server else "no") # 11. Background initial scan — files appear progressively. async def _bg_scan(indexer, name, gctx): await indexer.initial_scan() log.info("Background scan complete for %s: %d files", name, indexer.index.count) # Swarm registration for public groups (after files are known). if gctx.get("visibility") == "public": endpoint = f"webrtc:{self._config.node.quic_port}" hashes = [e.id for e in gctx["index"].entries] if hashes: await self._register_swarm(hashes, endpoint) # initial_scan() itself never calls on_change (it predates # the concept — every existing caller only cared about the # scan finishing, not about notifying anyone) — but Videos # app enrichment (duration/thumb_hash/display_title/...) # hangs entirely off that callback (_broadcast_index_change). # Without this, every file already on disk at startup — the # common case, an existing library — would never get # enriched at all; only a file added later, while the node # is already running, would trigger it via the watchdog. await self._on_index_change(indexer) for idx, group_cfg in zip(self._indexers, self._config.groups): gctx = groups_ctx.get(group_cfg.id) if gctx: self._tasks.append(asyncio.create_task( _bg_scan(idx, group_cfg.name, gctx))) self._tasks.append(asyncio.create_task( self._progress_pusher(idx))) # 12. Wait for shutdown stop_event = asyncio.Event() loop = asyncio.get_event_loop() console_shutdown_done = None if sys.platform == "win32": # SIGBREAK: CTRL_BREAK_EVENT, how platform.autostart_end() asks # a per-user-mode daemon to stop gracefully instead of only # ever taskkill /F. for sig in (signal.SIGINT, signal.SIGTERM, signal.SIGBREAK): signal.signal(sig, lambda *_: stop_event.set()) # CTRL_CLOSE/LOGOFF/SHUTDOWN reach no Python signal at all -- # see platform.install_console_close_handler for why this is # a separate mechanism rather than another signal.signal() line. from meshbay_node.platform import install_console_close_handler console_shutdown_done = install_console_close_handler(loop, stop_event) else: for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, stop_event.set) # Milestone 14.8: re-read node.toml without dropping connections. try: loop.add_signal_handler( signal.SIGHUP, lambda: spawn(self._reload_config())) except (NotImplementedError, AttributeError): pass # no SIGHUP on Windows; `reload` says so there await stop_event.wait() await self._shutdown() if console_shutdown_done is not None: # Releases the console-control handler's blocking wait (see # platform.install_console_close_handler) so it can return and # let Windows actually end the process for CTRL_CLOSE/LOGOFF/ # SHUTDOWN, now that cleanup is genuinely done rather than just # started. console_shutdown_done.set() async def _reload_config(self) -> None: """ Re-read node.toml and reconcile groups. Handles root changes on existing groups, hot-loads new groups, and tears down removed groups. Existing connections are untouched: a member watching a film keeps watching it. Serialised by _reload_lock: fire-and-forget reloads from config-mutating endpoints can overlap with the wizard's explicit /api/reload call, and two concurrent hot-loads of the same group corrupt the runtime state. The reload belongs to the node, never to whoever asked for it. A root added from a browser reaches here through the operator's WebRTC session, whose tasks are all cancelled when that session closes — and on 2026-09-14 one closed 47 s into the scan of a 900 GB root. The reload died without a line in the log, the new root was in node.toml and in the indexer but never in the group's context, and nothing ever tried again: the lock was free, and nobody was waiting on it. So the work runs in a task of its own, and a caller that goes away only stops waiting. """ await asyncio.shield(spawn(self._reload_config_locked(), what="config reload")) async def _reload_config_locked(self) -> None: try: async with self._reload_lock: await self._reload_config_inner() except asyncio.CancelledError: log.warning("Config reload cancelled before it finished — the node may " "be serving part of the previous configuration until the " "next reload") raise async def _reload_config_inner(self) -> None: log.info("Reloading config from %s", self._config_path) try: fresh = load_config(self._config_path) except Exception as e: log.error("Reload failed, keeping the running config: %s", e) return groups_ctx = self._state.get("groups_ctx", {}) hosted = set(groups_ctx) incoming = {g.id for g in fresh.groups if g.id} # ── Root changes on existing groups ────────────────────────────── changed = 0 for group_cfg in fresh.groups: ctx = groups_ctx.get(group_cfg.id) if not ctx: continue try: roots = await self._build_roots(group_cfg) except RootError as e: log.error("Group %r: %s — keeping the roots already loaded", group_cfg.name, e) continue # `writable` and `removable` are in the comparison because an # operator editing node.toml by hand and reloading is a supported # way to change them, and a set compared on name and path alone # reports "nothing changed" for exactly that edit. if _root_shape(ctx["roots"]) == _root_shape(roots): continue await off_disk(roots, roots.refresh_availability) indexer = next((i for i in self._indexers if i.group_id == group_cfg.id), None) if indexer is None: continue log.info("Group %r roots changed: %s", group_cfg.name, ", ".join(f"{r.name}={r.path}" for r in roots)) # The new set is what the node serves from this moment, and the # scan of an added root is not waited for. Awaiting it here held # `_reload_lock` and the old set for as long as the scan ran — # hours for a large drive — so every file request under the new # root found no root to resolve against, and any op answering with # the live table (a writable/removable toggle) showed the directory # gone from the operator's settings. ctx["roots"] = roots await indexer.retarget(roots, wait=False) changed += 1 # ── Hot-load new groups ────────────────────────────────────────── added_names = [] sk_ed = self._state.get("sk_ed25519") sk_x_raw = self._state.get("sk_x25519_raw") pk_x_raw = self._state.get("pk_x25519_raw") node_user_id = self._state.get("node_user_id") data_dir = fresh.data_dir for group_cfg in fresh.groups: if group_cfg.id in hosted: continue if not group_cfg.id or not group_cfg.roots: log.warning("New group %r has no id or roots — skipping", group_cfg.name) continue if not sk_ed: log.warning("Cannot hot-load %r — signing key not available", group_cfg.name) continue try: roots = await self._build_roots(group_cfg) except RootError as e: log.error("New group %r: %s — skipping", group_cfg.name, e) continue await off_disk(roots, roots.refresh_availability) gek = None if sk_x_raw and pk_x_raw: gek = await self._load_gek( group_cfg.id, node_user_id, sk_x_raw, pk_x_raw) if gek: log.info("GEK loaded for new group %s", group_cfg.id[:8]) scan_settings = ( await self._roster.scan_settings(group_cfg.id) if self._roster else { "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS, "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS, }) indexer = DirectoryIndexer( roots=roots, group_id=group_cfg.id, sk_node=sk_ed, gek=gek, on_change=self._on_index_change, on_root_ejected=self._eject_persister(group_cfg.id), cache=self._index_cache, reconcile_secs=scan_settings["reconcile_interval_secs"], debounce_secs=scan_settings["debounce_secs"], ) # Registered *before* start() runs its (blocking, possibly very # long — see the StarWars benchmark) initial scan, specifically # so /api/groups/{id}/index-status can see indexer.progress # while a brand-new group is still scanning — this is the one # group state that must stay visible during the very window the # group is not yet authorized for member connections (below). self._indexers.append(indexer) self._state["indexes"][group_cfg.id] = indexer.index self._state["indexers"][group_cfg.id] = indexer await indexer.start() data_dir.mkdir(parents=True, exist_ok=True) chat_db = data_dir / group_cfg.id[:16] / "chat.db" store = ChatStore(db_path=chat_db) await store.open() self._chat_stores[group_cfg.id] = store new_ctx = { "gek": gek, "roots": roots, "index": indexer.index, "progress": indexer.progress, "note_activity": indexer.note_activity, "record_upload": indexer.record_upload, "reconcile_interval_secs": scan_settings["reconcile_interval_secs"], "debounce_secs": scan_settings["debounce_secs"], "visibility": group_cfg.visibility, "join_policy": group_cfg.join_policy, "enabled_apps": ( await self._roster.enabled_apps(group_cfg.id) if self._roster else list(Roster.DEFAULT_APPS)), "transfer_limits": ( await self._roster.transfer_limits(group_cfg.id) if self._roster else {}), **(await self._app_directories_ctx(group_cfg.id)), "chat_link_preview": ( await self._roster.chat_link_preview(group_cfg.id) if self._roster else True), "search_listed": ( await self._roster.search_listed(group_cfg.id) if self._roster else True), "chat_epoch": await self._ensure_chat_epoch(group_cfg.id), "tmdb_enabled": ( await self._roster.tmdb_enabled(group_cfg.id) if self._roster else True), "musicbrainz_enabled": ( await self._roster.musicbrainz_enabled(group_cfg.id) if self._roster else True), "chat_store": store, } groups_ctx[group_cfg.id] = new_ctx if self._webrtc: self._webrtc._ctx["groups"][group_cfg.id] = new_ctx log.info("Hot-loaded group %s (%s, %d roots)", group_cfg.name, group_cfg.id[:8], len(roots)) added_names.append(group_cfg.name) # The initial scan above already ran to completion (indexer.start() # is not deferred here), so this only matters for whatever scans # this group as time goes on — a root added later, reconcile # picking one back up. self._tasks.append(asyncio.create_task(self._progress_pusher(indexer))) # ── Tear down removed groups ───────────────────────────────────── removed_names = [] for gid in hosted - incoming: indexer = next((i for i in self._indexers if i.group_id == gid), None) if indexer: try: await indexer.stop() except Exception: pass self._indexers.remove(indexer) store = self._chat_stores.pop(gid, None) if store: try: await store.close() except Exception: pass # No per-group index cache to close here (2026-08-25): the # (path, size, mtime) -> hash cache is now one shared instance, # open for the life of the daemon, since another group may still # reference the same physical folder — see indexer/cache.py. pending = self._pending_broadcasts.pop(gid, None) if pending: pending.cancel() self._last_broadcast_snapshot.pop(gid, None) self._state["indexes"].pop(gid, None) self._state["indexers"].pop(gid, None) old_name = gid[:8] for g_cfg in self._config.groups: if g_cfg.id == gid: old_name = g_cfg.name break groups_ctx.pop(gid, None) if self._webrtc and self._webrtc._ctx.get("groups") is not groups_ctx: self._webrtc._ctx["groups"].pop(gid, None) log.info("Unloaded group %s (%s)", old_name, gid[:8]) removed_names.append(old_name) self._config = fresh self._state["config"] = fresh self._state["groups"] = [g.name for g in fresh.groups] log.info("Reload complete — %d re-rooted, %d added, %d removed", changed, len(added_names), len(removed_names)) if (added_names or removed_names) and self._hub: gids = list((self._state.get("groups_ctx") or {}).keys()) await self._hub.update_ws_groups(gids) async def _login_with_retry(self, hub: HubClient): """Login to hub, retrying if the node key hasn't been linked yet.""" import httpx as _httpx while True: try: return await hub.startup(endpoint_hint=None) except _httpx.HTTPStatusError as e: body = e.response.text if hasattr(e.response, 'text') else '' # Any 401 here needs a human at a browser, and the operator needs # this daemon alive to read its public key (via `meshbay-node # status` or the desktop client, both of which query the control # API). Exiting would strand them — which is exactly what # happened when a node was started before its owner had # registered. if e.response.status_code == 401: if "No node key" in body: self._state["status"] = "waiting_for_node_key" log.warning( "Node key not linked. Get it from `meshbay-node " "status` and paste it in Settings > Link Node on %s " "(the desktop client links it automatically). " "Retrying in 5s...", self._config.hub.url, ) else: self._state["status"] = "waiting_for_account" log.warning( "Hub rejected the node credentials for user %r. " "Register that account on %s first, then link this " "node's key. Retrying in 5s...", self._config.hub.username, self._config.hub.url, ) await asyncio.sleep(5) else: raise except Exception as e: log.warning("Hub login failed: %s — retrying in 10s", e) await asyncio.sleep(10) async def _load_gek( self, group_id: str, node_user_id: str, sk_x_raw: bytes, pk_x_raw: bytes, ) -> bytes | None: """Load GEK from local bundle store (node-only, hub never touches crypto).""" from meshbay_common.crypto import unwrap_gek_aes if not self._bundle_store: return None # Try node-specific bundle first (stored by init_gek for daemon reload), # then fall back to operator's user bundle (legacy / pre-dual-key) for user_key in [f"_node_{node_user_id}", node_user_id]: bundle = await self._bundle_store.fetch(group_id, user_key) if not bundle: continue try: gek = unwrap_gek_aes(bundle, sk_x_raw, pk_x_raw) log.info("GEK loaded from local bundle store for group %s (key=%s)", group_id[:8], user_key[:16]) return gek except Exception as e: log.debug("Failed to unwrap GEK bundle (key=%s): %s", user_key[:16], e) log.warning("No unwrappable GEK bundle found for group %s", group_id[:8]) return None async def _ensure_chat_epoch(self, group_id: str) -> int: """ The group's current chat epoch, opening the first one if it has none. Chat is always encrypted, so a group with no epoch key is a group in which nobody can say anything. Attaching one is the node's job and happens here rather than on the first message: a failure at start-up is in the log the operator is already reading, and a failure on someone's first message is a chat that mysteriously refuses them. Never fatal. A group whose epoch cannot be opened keeps every other function — files, video, the index — and only its chat is unusable, which is strictly better than refusing to host the group at all. """ if not self._bundle_store: return 0 try: epoch = await self._bundle_store.latest_chat_epoch(group_id) if epoch: return epoch from meshbay_node import ops return (await ops.open_chat_epoch(self._state, group_id))["epoch"] except Exception as e: log.error("chat: no epoch key for group %s (%s) — chat is " "unusable in this group until this is fixed", group_id[:8], e) return 0 async def _reap_partial_uploads(self, interval: float = 3600.0, first_delay: float = 60.0) -> None: """ Delete `.part` files that no upload will ever finish. An upload interrupted for good leaves its partial file behind, and nothing else ever looks at it: `.part` is not an index entry, so it is invisible to every group member and to the operator's own file list. One abandoned film is a gigabyte of their disk, kept for ever. Two conditions, both required, and `uploads.orphaned_parts` is where they are stated and tested. What this adds is the walk and the deletion, and one rule of its own: it runs a minute after start rather than at once, so a client reconnecting to finish an upload that outlived a node restart is not raced by the janitor that would have deleted it — the age threshold makes that impossible in practice, and doing it anyway costs a minute. `interval` and `first_delay` are parameters so a test can drive this without waiting an hour. """ await asyncio.sleep(first_delay) while True: try: self._reap_once() except Exception as exc: # never let the janitor kill the node log.warning("Reaping partial uploads failed: %s", exc) await asyncio.sleep(interval) async def _purge_audit_log(self, interval: float = 86400.0) -> None: """ Delete audit entries past the retention period, at start then daily. `AuditStore.cleanup` leaves scheduling to its caller; until this loop nobody called it and a node kept every entry for ever. `interval` is a parameter so a test can drive this without waiting a day. """ while True: try: if self._audit_store: deleted = await self._audit_store.cleanup() if deleted: log.info("Purged %d audit entries older than %d days", deleted, AUDIT_RETENTION_DAYS) except Exception as exc: # never let the janitor kill the node log.warning("Purging the audit log failed: %s", exc) await asyncio.sleep(interval) def _reap_once(self, now: float | None = None) -> int: """One pass over every group. Returns how many files were deleted.""" groups = (self._webrtc._ctx.get("groups") or {}) if self._webrtc else {} when = time.time() if now is None else now deleted = 0 for gid, ctx in groups.items(): roots = ctx.get("roots") if roots is None: continue store = ctx.get("partial_uploads") live = store.live_paths() if store is not None else set() for path in uploads_mod.orphaned_parts( uploads_mod.find_parts(roots.roots), live, when): try: size = path.stat().st_size path.unlink() except OSError as exc: log.warning("Could not remove abandoned upload %s: %s", path.name, exc) continue deleted += 1 log.info("Removed abandoned upload %s (%d bytes, group %s)", path.name, size, gid[:8]) return deleted async def _progress_pusher(self, indexer: DirectoryIndexer, interval: float = 2.0) -> None: """ Watches indexer.progress and pushes a light INDEX_PROGRESS message to this group's connected peers — never the index itself, that stays _on_index_change's job. Runs for the node's whole lifetime: a scan can start from several places (initial scan, a root added later, reconcile picking a root back up), and this only needs to notice the flag, not why it changed. The final push at the False transition is what lets a presence dot reliably turn back off on an already-connected client — the handshake ack only covers the moment of connecting. `interval` is a parameter (not a bare constant) only so a test can drive this loop without waiting on the real 2s cadence. """ was_busy = False while True: await asyncio.sleep(interval) progress = indexer.progress # A root waiting for the scan lock is work the operator is owed a # sight of, even in the gap where nothing is walking yet. busy = progress.scanning or bool(progress.queued) if busy or was_busy: self._push_index_progress(indexer.group_id, progress) was_busy = busy def _push_index_progress(self, group_id: str, progress) -> None: if not self._webrtc: return # Deliberately NOT sealed, unlike index_sync/index_delta (decision D3). # Counters only — never a path, never a filename, see IndexProgress in # indexer.py — pushed every couple of seconds for the whole length of a # scan. Sealing it would buy an attacker's rough estimate of a library's # size and cost a key derivation and a decrypt per push. If a field that # names anything is ever added here, that trade is void and this message # joins the other two. `kind` is one of four fixed words; the root under # way is a position in the roots table the member already opened from # the sealed index, and the roots waiting are a count, never names. msg = { "type": MNP.INDEX_PROGRESS, "v": MNP_VERSION, "group_id": group_id, "scanning": progress.scanning, "scanned_bytes": progress.scanned_bytes, "total_bytes": progress.total_bytes, "files_done": progress.files_done, "files_total": progress.files_total, "kind": progress.kind, "root_pos": progress.root_pos, "queued": len(progress.queued), } pushed = 0 for session in list(self._webrtc._sessions.values()): if session._group_id == group_id: try: session._send(msg) pushed += 1 except Exception: pass if pushed: log.debug("Index progress pushed to %d peer(s) for group %s", pushed, group_id[:8]) async def _build_roots(self, group_cfg) -> RootSet: """ Build a group's RootSet from node.toml, with the ejected state restored. node.toml carries configuration (`writable`, `removable`); the roster carries the runtime answer to "is this drive ejected right now". They are merged here, in the one place every caller goes through, because a root that quietly comes back available across a restart is exactly the surprise unplug that eject exists to survive. """ specs = [asdict(r) for r in group_cfg.roots] if self._roster: ejected = await self._roster.ejected_roots(group_cfg.id) if ejected: for spec in specs: name = spec.get("name") or Path(spec.get("path", "")).name if fold(name) in ejected: spec["ejected"] = True return RootSet.build(specs) # Every application that keeps directories. This is the one list, and it # lives here because the daemon is what wires a group's context: `roster.py`, # `ops.py` and the rest must name no application at all — that is the # property the reference app exists to demonstrate # (`test_helloworld_proves_the_plugin_claim.py`). # # Not derived from `enabled_apps`: the context is read once at load, and an # application enabled later must not find its own setting missing. # # The handshake ack does **not** get a copy of this. It emits whatever # `_directories` the context holds, so the two cannot drift — a copy # lived in `webrtc_server.py` until 2026-09-10 and had already lost # `helloworld`, which made the app that proves a new one needs no # special-casing the single app whose directories never reached a client. APP_DIR_KEYS = ("video", "music", "photo", "chat", "helloworld") async def _app_directories_ctx(self, group_id: str) -> dict: """ Each app's configured directories, plus the second name an app is also published under where something reads one (`chat_directory`). Derived here rather than stored, so the two can never disagree. """ dirs = {} for app in self.APP_DIR_KEYS: dirs[f"{app}_directories"] = ( await self._roster.app_directories(group_id, app) if self._roster else []) aliases = {} for app in self.APP_DIR_KEYS: alias = Roster.ctx_alias(app, dirs[f"{app}_directories"]) if alias: aliases[alias[0]] = alias[1] return {**dirs, **aliases} def _eject_persister(self, group_id: str): """`on_root_ejected` bound to one group, for that group's indexer.""" async def persist(root_name: str, ejected: bool) -> None: if self._roster: await self._roster.set_root_ejected( group_id, root_name, ejected, set_by=self._state.get("node_user_id", "")) return persist async def _on_index_change(self, indexer: DirectoryIndexer) -> None: """ Called when a DirectoryIndexer detects file changes — once per debounced watchdog event, so dropping N files into a watched folder calls this N times in quick succession. Coalesces those into one broadcast (_broadcast_index_change) rather than one push per file: the timer is reset on every call and only fires once calls stop arriving for _broadcast_coalesce_secs. """ group_id = indexer.group_id loop = asyncio.get_event_loop() pending = self._pending_broadcasts.pop(group_id, None) if pending: pending.cancel() def fire() -> None: self._pending_broadcasts.pop(group_id, None) spawn(self._broadcast_index_change(indexer)) self._pending_broadcasts[group_id] = loop.call_later( self._broadcast_coalesce_secs, fire) async def _broadcast_index_change(self, indexer: DirectoryIndexer) -> None: """ The actual push, run once per coalesced burst. Sends a full INDEX_SYNC the first time a group is ever broadcast (no previous snapshot to diff against — the client's own first fetchIndex() call already covers that case) and an INDEX_DELTA every time after, computed against the last thing this method actually sent. """ group_id = indexer.group_id idx = indexer.index log.info("Index changed for group %s: %d files (v%d)", group_id[:8], idx.count, idx.version) prev = self._last_broadcast_snapshot.get(group_id) delta = None previous = None if prev is not None: prev_version, prev_entries = prev previous = GroupIndex._snapshot( idx.group_id, idx.sk_node, idx.gek, prev_version, prev_entries) delta = idx.diff(previous) self._last_broadcast_snapshot[group_id] = (idx.version, idx.entries_by_id()) # Videos app (docs/MESHBAY_DESIGN.md §6.5): schedule async technical # probe + title parse + thumbnail generation for every newly-seen # video entry under the group's configured video_root. Never blocks # this broadcast — enrichment fields arrive later as their own # INDEX_DELTA update (_on_enriched below). new_entries = delta.additions if delta is not None else list(idx.entries) # A root that was ejected and plugged back in, or that fell off and # re-mounted, has had its entries thrown away and rebuilt from disk # (`indexer._drop_root_entries`). The rebuilt entry has the same # content-hash id and none of the enrichment fields, so the diff above # reports neither an addition nor a deletion — and `_enriched_attempted` # still says "done" for a file whose album and cover no longer exist. # Found live: a Music library came back with its files and without its # albums, and stayed that way, because only a restart (which starts # with no snapshot, making every entry an addition) could clear either # gate. Treated here as what it is — those entries are new again. rebuilt_ids = indexer.drain_rescanned_ids() if rebuilt_ids: rebuilt = [e for e in idx.entries if e.id in rebuilt_ids] for entry in rebuilt: self._enriched_attempted.discard((group_id, entry.id)) seen = {e.id for e in new_entries} new_entries = new_entries + [e for e in rebuilt if e.id not in seen] spawn(self._enrich_new_video_entries(indexer, new_entries)) # Music app (docs/MESHBAY_DESIGN.md §9.8): same shape, gated on # audio_root exactly like video_root above (added later — the original # "no root, whole shared tree" call didn't hold up). spawn(self._enrich_new_audio_entries(indexer, new_entries)) # Photos app (docs/MESHBAY_DESIGN.md §9.9): same shape, gated on # photo_roots (a list, not a single string). spawn(self._enrich_new_photo_entries(indexer, new_entries)) # A rename/move changes the very filename (or season folder) that # docs/MESHBAY_DESIGN.md §9.7's title-parse read # display_title/season/episode from, but leaves the file's content — # and so its id and everything ffprobe/thumbnailing already found — # untouched. Only entries # whose name or path actually differ from the last broadcast get a # fresh pass; an update that is enrichment's own field-fill # (duration/thumb_hash/... landing via _on_enriched below) leaves # name/path alone and must not re-trigger itself forever. if delta is not None and delta.updates and previous is not None: spawn( self._reenrich_renamed_video_entries(indexer, delta.updates, previous)) spawn( self._reenrich_renamed_audio_entries(indexer, delta.updates, previous)) spawn( self._reenrich_renamed_photo_entries(indexer, delta.updates, previous)) # Videos/Music/Photos apps: a file that leaves the index also loses # its thumbnail/cover and file->tmdb/file->mbid mapping — the "real # deletion obligation" docs/MESHBAY_DESIGN.md §6.5 calls out # explicitly rather than leaving implicit (docs/MESHBAY_DESIGN.md §9.8 # follows the same rule). tmdb_meta/mbid_meta rows are left alone # (shared across files). # # Found live (docs/MESHBAY_DESIGN.md §9.9): a root removed and a new # one added for the identical content (an operator renaming/relocating a # shared folder) pruned the thumbnail here — correctly, the content # is gone from *this* root — but left the hash in # `_enriched_attempted`, which is never otherwise cleared. The same # bytes reappearing under the new root's path were then permanently # skipped: "already attempted" was true forever, for a thumbnail # that no longer existed. Discarding the attempt alongside the # cache entry is what makes pruning actually reversible — the next # sweep re-enriches it exactly as if it were new, which content # that is content-addressed and simply moved effectively is. if delta is not None and delta.deletions and self._media_cache: for file_id in delta.deletions: self._enriched_attempted.discard((indexer.group_id, file_id)) # media_cache.db is node-wide, keyed by content hash — a file # shared into two groups is one row there, same reasoning as # `_enriched_attempted`'s own docstring above. This group's # copy is genuinely gone (that's what a deletion delta is), # but another group may still hold the same content: only # prune once *no* group's index has this file_id any more, # or the surviving group pays for a redundant re-fetch/ # re-probe/re-thumbnail for content it never actually lost. still_referenced = any( i.index.get_entry(file_id) is not None for i in self._indexers) if not still_referenced: spawn(self._media_cache.prune_file(file_id)) # 11.5 — Push to connected WebRTC peers in this group if self._webrtc: peers = [s for s in list(self._webrtc._sessions.values()) if s._group_id == group_id] # Both messages are sealed under a GEK-derived subkey, so building one # needs a key. A group without one has no peers to push to either — the # node refuses every handshake while the GEK is None (NS8) — so this is # "nobody is listening", not a case to send in clear for. if peers and idx.gek: msg = (index_delta_message(idx, delta, indexer.roots) if delta is not None else index_sync_message(idx, indexer.roots)) pushed = 0 for session in peers: try: session._send(msg) pushed += 1 except Exception: pass if pushed: log.info("Index %s pushed to %d WebRTC peers", "delta" if delta is not None else "sync", pushed) # 11.9 — Register file hashes with hub swarm table (public groups only, H7) group_cfg = next( (g for g in self._config.groups if g.id == group_id), None) if (self._hub and self._state.get("endpoint_hint") and group_cfg and group_cfg.visibility == "public"): # Only the newly added hashes once there is a delta to know them # from — registering the whole library again on every change is # the same O(changes x library size) cost the delta above exists # to avoid. hashes = ([e.id for e in delta.additions] if delta is not None else [e.id for e in idx.entries]) if hashes: endpoint = f"webrtc:{self._config.node.quic_port}" spawn(self._register_swarm(hashes, endpoint)) async def _enrich_new_video_entries(self, indexer: DirectoryIndexer, entries: list) -> None: """ Videos app: fire (never await further) enrichment for unattempted video entries under the group's configured video_root. A group with no video_root set yet does not enrich anything — TMDB lookups and ffmpeg thumbnailing are real, ongoing per-file cost (docs/MESHBAY_DESIGN.md §6.5), and running them over an operator's whole shared index before they have chosen which folder is actually their media library would burn both TMDB's rate limit and the node's CPU on files that were never meant to be in the Videos app at all. Once a root is set, `_enrich_video_root_now` (called when it changes) separately sweeps whatever it already contains — this path alone only ever sees entries new since the last broadcast. """ if not self._enricher or not self._roster: return video_dirs = await self._roster.app_directories(indexer.group_id, "video") if not video_dirs: return for entry in entries: if entry.type != "video" or (indexer.group_id, entry.id) in self._enriched_attempted: continue if not _under_any_directory(entry.path, video_dirs): continue file_path = entry_abs_path(indexer.roots, entry) if not file_path or not file_path.exists(): continue self._enriched_attempted.add((indexer.group_id, entry.id)) async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None: await self._on_enriched(_indexer, file_id, fields) self._enricher.spawn(entry, file_path, on_done) async def _enrich_video_root_now(self, group_id: str) -> None: """ Videos app: sweep a group's existing index for enrichment right after its video root is set or changed. The ordinary path above only ever looks at entries new since the last broadcast, so a folder that already had files sitting in it before it became the video_root would otherwise never get enriched at all — nothing else re-visits already-indexed entries once they have been broadcast once. """ indexer = self._state.get("indexers", {}).get(group_id) if not indexer: return await self._enrich_new_video_entries(indexer, list(indexer.index.entries)) async def _reenrich_renamed_video_entries( self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex, ) -> None: """ Videos app: found live — a French-named episode file, renamed by the operator to match its English-named siblings, kept showing as its own separate poster-grid card (and its own row in Flat list) indefinitely, because `_enriched_attempted` — there specifically to stop enrichment's own field-fill from re-triggering itself forever (see the caller) — also silently blocked the *new* filename from ever being title-parsed at all. `entry.id in self._enriched_attempted` is the same content, so simply discarding it here and re-running the ordinary enrichment path is enough: a fresh ffprobe/thumbnail for an unchanged file is redundant work, not a correctness issue, and renames are rare enough that the redundancy is not worth a separate "title-parse only" code path. """ for entry in updates: if entry.type != "video": continue old = previous.get_entry(entry.id) if old is None or (old.name == entry.name and old.path == entry.path): continue self._enriched_attempted.discard((indexer.group_id, entry.id)) # The rename re-derives the title (the whole point of this # method), which can change the correct TMDB match — but # file_tmdb is keyed by content hash, unchanged by a rename, so # nothing else would ever dislodge the old name's match. A # manual "Fix match" correction is kept (clear_file_tmdb skips # anything in media_cache.tmdb_override). if self._media_cache is not None: await self._media_cache.clear_file_tmdb(entry.id) await self._enrich_new_video_entries(indexer, updates) async def _enrich_new_audio_entries(self, indexer: DirectoryIndexer, entries: list) -> None: """ Music app (docs/MESHBAY_DESIGN.md §9.8): fire (never await further) tag/cover enrichment for unattempted audio entries under the group's configured audio_root — same gate as `_enrich_new_video_entries` above (the original "no root, whole shared tree" call turned out wrong against a real messy library: everything under every shared folder got mixed together with no way to scope it down). `_enriched_attempted` is shared with the video path — content-addressed ids never collide across the two. """ if not self._audio_enricher or not self._roster: return audio_dirs = await self._roster.app_directories(indexer.group_id, "music") if not audio_dirs: return # Resolved once per directory, not once per file: a library is # thousands of entries and this is a filesystem call each time. boundaries = {d: indexer.roots.resolve(d, require_available=False) for d in audio_dirs} for entry in entries: if entry.type != "audio" or (indexer.group_id, entry.id) in self._enriched_attempted: continue owner = _owning_directory(entry.path, audio_dirs) if owner is None: continue file_path = entry_abs_path(indexer.roots, entry) if not file_path or not file_path.exists(): continue self._enriched_attempted.add((indexer.group_id, entry.id)) async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None: await self._on_enriched(_indexer, file_id, fields) # The boundary is *the configured directory this file is under*, # not the shared root it lives in — so the ancestor walk # (enrich_audio._artist_album_from_ancestors) treats a flat # top-level folder right under the configured Music directory as # ambiguous (artist-or-release, docs/MESHBAY_DESIGN.md §9.8), # rather than one level too shallow when that directory is # itself a subfolder. With several configured, each file is # measured against its own: # a single shared boundary would be wrong for all but one of them. self._audio_enricher.spawn(entry, file_path, on_done, boundaries.get(owner)) async def _enrich_audio_root_now(self, group_id: str) -> None: """ Music app: sweep a group's existing index right after its audio root is set or changed. Mirrors `_enrich_video_root_now` exactly — the ordinary path above only ever looks at entries new since the last broadcast, so a folder that already had files in it before it became the audio_root would otherwise never get enriched at all. """ indexer = self._state.get("indexers", {}).get(group_id) if not indexer: return await self._enrich_new_audio_entries(indexer, list(indexer.index.entries)) async def _reenrich_renamed_audio_entries( self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex, ) -> None: """ Music app equivalent of `_reenrich_renamed_video_entries` — a rename can change the filename-parse fallback (title/track_no) even though embedded tags, when present, are unaffected. Re-running the whole pass on a rename is redundant work for a tagged file and a real fix for an untagged one, and renames are rare enough not to need a cheaper, tags-only special case. """ for entry in updates: if entry.type != "audio": continue old = previous.get_entry(entry.id) if old is None or (old.name == entry.name and old.path == entry.path): continue self._enriched_attempted.discard((indexer.group_id, entry.id)) await self._enrich_new_audio_entries(indexer, updates) async def _enrich_new_photo_entries(self, indexer: DirectoryIndexer, entries: list) -> None: """ Photos app (docs/MESHBAY_DESIGN.md §9.9): fire (never await further) thumbnail/EXIF enrichment for unattempted image entries under any of the group's configured photo_roots. Same gate shape as `_enrich_new_video_entries`/`_enrich_new_audio_entries` — no root configured yet means no work, since thumbnailing every image in a whole shared tree before the operator has chosen which folders are actually photo albums would burn CPU on files never meant to be in the Photos app at all. `_enriched_attempted` is shared with the video/audio paths — content-addressed ids never collide across them. """ if not self._photo_enricher or not self._roster: return photo_dirs = await self._roster.app_directories(indexer.group_id, "photo") if not photo_dirs: return for entry in entries: if entry.type != "image" or (indexer.group_id, entry.id) in self._enriched_attempted: continue if not _under_any_directory(entry.path, photo_dirs): continue file_path = entry_abs_path(indexer.roots, entry) if not file_path or not file_path.exists(): continue self._enriched_attempted.add((indexer.group_id, entry.id)) async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None: await self._on_enriched(_indexer, file_id, fields) self._photo_enricher.spawn(entry, file_path, on_done) async def _enrich_photo_roots_now(self, group_id: str) -> None: """ Photos app: sweep a group's existing index right after its photo roots change. Mirrors `_enrich_video_root_now`/`_enrich_audio_root_now` — the ordinary path above only ever looks at entries new since the last broadcast, so a folder that already had photos in it before it was added to photo_roots would otherwise never get enriched at all. Also covers a root being *removed*: nothing un-enriches on removal (the cache entry is harmless, just unused — the media cache is disposable and tied to the index, docs/MESHBAY_DESIGN.md §6.5), so re-sweeping the new set is enough. """ indexer = self._state.get("indexers", {}).get(group_id) if not indexer: return await self._enrich_new_photo_entries(indexer, list(indexer.index.entries)) async def _reenrich_renamed_photo_entries( self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex, ) -> None: """ Photos app equivalent of `_reenrich_renamed_video_entries` — a rename changes nothing about the image's own bytes (thumbnail, EXIF fields are content-derived, not name-derived), so this exists only for consistency/symmetry with Videos/Music and to catch the case of a file moving *into* a newly-covered photo_roots subtree via a rename rather than a fresh add. Re-running enrichment on an unchanged file is redundant work, not a correctness issue. """ for entry in updates: if entry.type != "image": continue old = previous.get_entry(entry.id) if old is None or (old.name == entry.name and old.path == entry.path): continue self._enriched_attempted.discard((indexer.group_id, entry.id)) await self._enrich_new_photo_entries(indexer, updates) async def _on_enriched(self, indexer: DirectoryIndexer, file_id: str, fields: dict) -> None: """ Merge enrichment fields into the live index and re-trigger a broadcast so they reach connected clients as an INDEX_DELTA update (GroupIndex.diff's `updates`, not `additions` — same id, new fields). Builds a *new* IndexEntry via dataclasses.replace rather than mutating the existing one in place: the diff mechanism compares against a shallow snapshot of entry *references*, so an in-place mutation would silently also change what "previous" looks like, and the change would never show up as a diff (see group_index.py's diff() docstring). """ idx = indexer.index entry = idx.get_entry(file_id) if entry is None: return # removed from the index while enrichment was in flight idx.add_entry(replace(entry, **fields)) await self._on_index_change(indexer) def _drop_group_sessions(self, group_id: str) -> None: """Close live sessions for a revoked group (H4).""" if not self._webrtc or not group_id: return for session in list(self._webrtc._sessions.values()): if session._group_id == group_id: spawn(session.close()) log.info("Dropped session for revoked group %s", group_id[:8]) async def _register_swarm(self, hashes: list[str], endpoint: str) -> None: try: n = await self._hub.register_swarm(hashes, endpoint) log.info("Swarm: registered %d/%d hashes", n, len(hashes)) except Exception as e: log.warning("Swarm registration failed: %s", e) async def _shutdown(self) -> None: log.info("Shutting down...") self._state["status"] = "stopping" for handle in self._pending_broadcasts.values(): handle.cancel() self._pending_broadcasts.clear() for task in self._tasks: task.cancel() for task in self._tasks: try: await task except (asyncio.CancelledError, Exception): pass if self._webrtc: await self._webrtc.close_all() if self._audit_store: await self._audit_store.close() if self._bundle_store: await self._bundle_store.close() if self._tmdb_client: await self._tmdb_client.close() if self._musicbrainz_client: await self._musicbrainz_client.close() if self._media_cache: await self._media_cache.close() if self._roster: await self._roster.close() for store in self._chat_stores.values(): await store.close() if self._index_cache: await self._index_cache.close() for indexer in self._indexers: await indexer.stop() # The thread each group's roots are read from (roots.py `io_executor`). # Created on the first read, so a group nobody downloaded from never # started one and this is a no-op for it. groups = self._webrtc._ctx.get("groups") if self._webrtc else None for group in (groups or {}).values(): roots = group.get("roots") if roots is not None: roots.close_io() if self._quic_server: await self._quic_server.stop() token_file = getattr(self, "_ui_token_file", None) if token_file is not None: token_file.unlink(missing_ok=True) log.info("Node stopped") # ── CLI helpers ─────────────────────────────────────────────────────────────── def _daemon_api(cfg: Config, path: str, method: str = "GET", timeout: int = 30, body: dict | None = None) -> dict: """ Call the daemon's loopback API. The daemon owns the roster, the hub session and the live group contexts, so the CLI asks it to act rather than opening its databases behind its back. It also means every operator action goes through the control API's per-run session token (11.5.3), the same gate the desktop client's Node page passes. """ import json as _json import urllib.error import urllib.parse import urllib.request token_file = cfg.data_dir / "ui-token" if not token_file.exists(): print("Node is not running — start it with: meshbay-node") sys.exit(1) sep = "&" if "?" in path else "?" url = (f"http://127.0.0.1:{cfg.node.ui_port}{path}" f"{sep}t={token_file.read_text(encoding="utf-8").strip()}") try: data = _json.dumps(body).encode() if body is not None else None req = urllib.request.Request( url, method=method, data=data, headers={"Content-Type": "application/json"} if data else {}) with urllib.request.urlopen(req, timeout=timeout) as r: return _json.loads(r.read()) except urllib.error.HTTPError as e: raw = e.read().decode()[:600] try: parsed = _json.loads(raw) detail = parsed.get("error", raw) # Endpoints that refuse a name offer the ones that would work; a bare # "no such group" leaves the operator guessing at a UUID. for row in parsed.get("available", []): detail += f"\n {row.get('name', ''):<24} {row.get('id', '')}" except Exception: detail = raw print(f"failed: {detail}") sys.exit(1) except Exception as e: print(f"failed: {e}") sys.exit(1) def _resolve_group(cfg: Config, group: str | None) -> str: """ The group argument as an id, or the only configured one. Accepts a name as well, because node.toml already gives every group one and nobody remembers a UUID. A name that matches nothing configured says so, and lists what is — silently passing it through produced a 404 from the daemon that read like the group did not exist on the hub. """ if group: by_id = [g for g in cfg.groups if g.id == group] if by_id: return by_id[0].id by_name = [g for g in cfg.groups if g.name == group and g.id] if len(by_name) == 1: return by_name[0].id if len(by_name) > 1: print(f"several groups in node.toml are named {group!r} — use the id") sys.exit(1) # An id this node does not host is still worth passing on: the daemon # gives the better error, naming the group it does host. if "-" in group and len(group) == 36: return group print(f"no group named {group!r} in {DEFAULT_CONFIG_PATH}") if cfg.groups: print("configured groups:") for g in cfg.groups: print(f" {g.name or '(unnamed)':<24} {g.id or '(no id yet)'}") sys.exit(1) configured = [g.id for g in cfg.groups if g.id] if len(configured) == 1: return configured[0] print("--group is required (several groups configured)" if configured else "no group configured in node.toml") sys.exit(1) def _systemctl_user(verb: str, unit: str, *, not_running_hint: str, success: str, watch: str | None) -> None: """ Run `systemctl --user ` and report the result. The lifecycle authority is the unit, not this process: systemd already knows which PID it started, restarts it on failure (`Restart=on-failure` in the unit) and reloads it correctly (`ExecReload=`). Anything this CLI did instead — finding a process by pattern-matching its command line, signalling it, respawning it — is a second, worse implementation of what systemd is already doing, and pattern-matching a process list has already hit a real developer's real running node by accident. Reloads the user manager's view of unit files first. The package installers (deb postinst, rpm %post) run as root and can only reload the *system* manager — a different process from any signed-in user's *user* manager, which is the one that actually owns this unit — so a package upgrade leaves that manager still holding the old unit file and prints a warning naming the exact fix. Doing it here runs it under the right privilege, the user's own, right before the command that would otherwise act on a stale definition. Best-effort and unchecked: a reload the manager did not need must never block what the operator actually asked for, and a genuine problem still surfaces from the verb below. """ import subprocess subprocess.run(["systemctl", "--user", "daemon-reload"], capture_output=True, text=True) result = subprocess.run(["systemctl", "--user", verb, unit], capture_output=True, text=True) if result.returncode != 0: detail = (result.stderr or result.stdout).strip() print(detail or not_running_hint) sys.exit(1) print(success) if watch: print(watch) # ── Entry point ─────────────────────────────────────────────────────────────── def main() -> None: import argparse from meshbay_node.platform import configure_event_loop, force_utf8_stdio, load_node_env force_utf8_stdio() configure_event_loop() # Before anything reads the environment. On Linux systemd has usually loaded # the same file already via EnvironmentFile=; this is what makes a Windows # run (Startup-folder .vbs, no systemd) and a bare `meshbay-node` behave the # same. Already-set variables are left alone, so it cannot undo either. load_node_env(config_dir()) parser = argparse.ArgumentParser(description="MeshBay Node daemon") parser.add_argument("command", nargs="?", choices=["init", "reset", "status", "gek-init", "gek", "operator", "member", "group", "root", "file", "video", "chat", "denylist", "stun", "transfers", "reload", "restart-daemon", "autostart", "service", "calibrate-argon2"], help="init: provision config + keystore | reset: erase all " "node state | status: node state and keys " "| operator pair: pair a " "browser with this node | member list|invite|revoke|unpin " "| group list|add|remove " "| root list|add|remove|set|eject|plug " "| gek init|rotate | file list|rm " "| video rematch: re-resolve TMDB matches for a group's " "videos " "| chat status|rotate|encrypt-history|prune " "| denylist show|clear " "| stun list|add|remove|reset " "| transfers show|set|max-size|per-member: live " "transfer slots, the node-wide caps, the largest " "single upload, and how many one member may run " "at once in a group " "| reload: re-read node.toml (hot; systemd or the " "loopback API) | restart-daemon: restart the node " "(systemd unit, the Windows autostart launcher, or the " "service task, whichever applies) " "| autostart install|remove|start|stop|status " "(Windows: run meshbay-node at each sign-in, no admin) " "| service install|remove|start|stop|status " "(Windows: run at boot, before sign-in, needs admin " "once to install) " "| calibrate-argon2: benchmark") parser.add_argument("subcommand", nargs="?", help="'pair' for operator; list|invite|revoke|unpin for " "member; list|add|remove for group; " "list|add|remove|set|eject|plug for root; " "init|rotate for gek; " "list|rm for file; rematch for video; show|clear for " "denylist; list|add|remove|reset for stun; " "show|set|max-size|per-member for transfers; " "install|remove|start|stop|status for autostart and " "for service") parser.add_argument("target", nargs="?", help="username for member invite|revoke|unpin; group name " "for group add; file id for file rm; identifier for " "denylist clear; download cap for transfers set; " "size in GB for transfers max-size") parser.add_argument("value", nargs="?", help="the second value where a verb takes two: the " "upload cap for transfers set") parser.add_argument("--hub-url", default=None, help="hub URL, for init (e.g. https://meshbay.org)") parser.add_argument("--username", default=None, help="hub username, for init") parser.add_argument("--dir", default=None, help="shared directory, for group add") parser.add_argument("--yes", action="store_true", help="skip the confirmation for destructive commands") parser.add_argument("--config", type=Path, default=None, help="Config file path") parser.add_argument("--group", default=None, help="group id (optional if only one is configured)") parser.add_argument("--writable", action="store_true", default=None, dest="writable", help="root accepts member uploads (root add/set)") parser.add_argument("--no-writable", action="store_false", dest="writable", help="root is read-only (root add/set, group add)") parser.add_argument("--removable", action="store_true", default=None, dest="removable", help="mark root as removable (root set/add)") parser.add_argument("--no-removable", action="store_false", dest="removable", help="mark root as not removable (root set)") parser.add_argument("--name", default=None, help="root name (root add; defaults to directory basename)") parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"]) args = parser.parse_args() # Query commands print a report; library logging would interleave with it. quiet = args.command in ("status", "gek-init", "gek", "operator", "member", "group", "root", "file", "video", "chat", "denylist", "stun", "reload", "restart-daemon", "reset") logging.basicConfig( level=logging.ERROR if quiet else getattr(logging, args.log_level), format="%(asctime)s %(levelname)-8s %(name)s: %(message)s", ) if args.command == "init": cfg_path = args.config or DEFAULT_CONFIG_PATH cfg_dir = cfg_path.parent cfg_dir.mkdir(parents=True, exist_ok=True) from meshbay_node.platform import install_node_env env_written = install_node_env(cfg_dir) if env_written: print(f"Wrote {env_written} (packaged defaults).") hub_url = args.hub_url username = args.username if not hub_url: hub_url = input("Hub URL [https://meshbay.org]: ").strip() or "https://meshbay.org" if not username: username = input("Hub username: ").strip() if not username: print("Username is required.") sys.exit(1) if cfg_path.exists(): existing = cfg_path.read_text(encoding="utf-8") import re as _re m = _re.search(r'username\s*=\s*"([^"]*)"', existing) existing_user = m.group(1) if m else "" if existing_user and existing_user != "myusername" and existing_user != username: print(f"Config already exists with username {existing_user!r}.") print("This node belongs to another operator. Use 'meshbay-node reset' first.") sys.exit(1) if existing_user in ("", "myusername"): updated = _re.sub( r'(username\s*=\s*)"[^"]*"', rf'\1"{username}"', existing) updated = _re.sub( r'(url\s*=\s*)"[^"]*"', rf'\1"{hub_url}"', updated, count=1) cfg_path.write_text(updated, encoding="utf-8", newline="\n") print(f"Config updated: username={username}, hub={hub_url}") else: print(f"Config already exists: {cfg_path}") else: unlock_file = cfg_dir / "unlock.key" toml_lines = [ "[hub]", f'url = "{hub_url}"', f'username = "{username}"', "", "[node]", "quic_enabled = false # QUIC direct path; no client uses it yet", "quic_port = 19010", "ui_port = 18000", "", "[keystore]", # Forward slashes: a Windows path in a TOML basic string is a # parse error (`\U`, `\a`, ... are escape sequences). f'unlock_file = "{unlock_file.as_posix()}"', "", ] cfg_path.write_text("\n".join(toml_lines) + "\n", encoding="utf-8", newline="\n") chmod_private(cfg_path) print(f"Config written to {cfg_path}") unlock_file = cfg_dir / "unlock.key" if not unlock_file.exists(): import secrets key = secrets.token_urlsafe(32) unlock_file.write_text(key + "\n", encoding="utf-8", newline="\n") chmod_private(unlock_file) print(f"Unlock key created: {unlock_file}") cfg = load_config(cfg_path) if cfg.keystore.path.exists(): print(f"Keystore already exists: {cfg.keystore.path}") keys = load_keystore( path=cfg.keystore.path, unlock_file=cfg.keystore.unlock_file) else: keys = create_keystore( path=cfg.keystore.path, unlock_file=cfg.keystore.unlock_file) print(f"Keystore created: {cfg.keystore.path}") print(f"Node key: {keys.pk_ed25519_b64}") print() print("Next steps:") print(f" 1. Link this node key on {hub_url} → Settings → Link Node") if sys.platform == "win32": print(" 2. meshbay-node autostart install (run at each sign-in)") print(" — or just: meshbay-node (start it now, this session)") else: print(" 2. systemctl --user enable --now meshbay-node") print(" 3. meshbay-node group add --dir /path/to/files") print(" 4. meshbay-node gek init") print(" 5. meshbay-node operator pair") return if args.command == "reset": import shutil config_dir_ = config_dir() data_dir_ = data_dir() state_dir_ = state_dir() items = [] for d in (config_dir_, data_dir_): if d.exists(): for child in sorted(d.iterdir()): items.append(child) if not items: print("Nothing to reset — no node state found.") return print("This will permanently erase all node state:") for p in items: print(f" {p}") print() print("WARNING: a new keystore means a new identity. All group") print("memberships, operator pairings, and invitations are lost.") if not args.yes: answer = input("\nProceed? [y/N] ").strip().lower() if answer != "y": print("Aborted.") return import json as _json import subprocess as _sp import urllib.error import urllib.request token_file = data_dir_ / "ui-token" if token_file.exists(): try: cfg = Config(config_dir_ / "node.toml") tok = token_file.read_text(encoding="utf-8").strip() url = (f"http://127.0.0.1:{cfg.node.ui_port}" f"/api/unlink?t={tok}") req = urllib.request.Request(url, method="DELETE") with urllib.request.urlopen(req, timeout=5) as r: _json.loads(r.read()) print("Unlinked node key from hub.") except Exception: print("Could not unlink from hub (daemon not reachable).") if sys.platform == "win32": from meshbay_node.platform import autostart_remove, service_remove autostart_remove() service_remove() # no-op, silently, if not elevated or not installed else: _sp.run(["systemctl", "--user", "disable", "--now", "meshbay-node"], capture_output=True) for d in (config_dir_, data_dir_): if d.exists(): shutil.rmtree(d) print(f"Removed {d}") if state_dir_.is_dir(): shutil.rmtree(state_dir_) print(f"Removed {state_dir_}") print("Node state erased. Run 'meshbay-node init' to start over.") return if args.command == "calibrate-argon2": calibrate_argon2() return if args.command == "status": import json as _json import urllib.request cfg = load_config(args.config or DEFAULT_CONFIG_PATH) print(f"hub {cfg.hub.url} (user {cfg.hub.username or '—'})") try: keys = load_keystore( path=cfg.keystore.path, unlock_file=cfg.keystore.unlock_file) print(f"node key {keys.pk_ed25519_b64}") except FileNotFoundError: print("node key ") except Exception as e: print(f"node key ") token_file = cfg.data_dir / "ui-token" live = None if token_file.exists(): try: url = (f"http://127.0.0.1:{cfg.node.ui_port}" f"/api/status?t={token_file.read_text(encoding="utf-8").strip()}") with urllib.request.urlopen(url, timeout=3) as r: live = _json.loads(r.read()) except Exception: live = None if live: print(f"daemon running — {live.get('status')}") print(f"node_id {live.get('endpoint_hint') or '—'}") print(f"groups {live.get('group_count', 0)}" f" files {live.get('total_files', 0)}" f" peers {live.get('webrtc_peers', 0)}") needs = live.get("needs", []) if needs: _GUIDANCE = { "node_key_link": ( "Link node key", f"Copy the node key above and paste it in Settings → Link Node on {cfg.hub.url}"), "group_add": ( "Add a group", "meshbay-node group add --dir /path/to/files"), "operator_pair": ( "Pair as operator", "meshbay-node operator pair"), } print() print("action needed:") for need in needs: if need.startswith("gek_init:"): name = need.split(":", 1)[1] print(f" → Initialize group key for {name}") print(f" meshbay-node gek init --group \"{name}\"") elif need in _GUIDANCE: label, hint = _GUIDANCE[need] print(f" → {label}") print(f" {hint}") else: print(f" → {need}") else: print("daemon not running") print(f"config {DEFAULT_CONFIG_PATH}") if not cfg.groups: print("groups none configured — create a group on the hub, then add") print(" a [[groups]] entry with its id and a directory") else: for g in cfg.groups: print(f" group {g.name} [{g.visibility}] {g.id or ''}") if not g.roots: print(" ") for r in g.roots: label = r.name or Path(r.path).name flags = [] if getattr(r, 'writable', False) or getattr(r, 'upload', False): flags.append("rw") else: flags.append("ro") if getattr(r, 'removable', False): flags.append("removable") flag_str = f" ({', '.join(flags)})" if flags else "" live = "" if Path(r.path).expanduser().is_dir() else " [UNAVAILABLE]" print(f" {label} → {r.path}{flag_str}{live}") # Node authority: the roster is the source of truth, node.toml the legacy # form. Read the DB directly so this reports correctly while the daemon is # stopped — the state an operator is most often in when checking. import asyncio as _asyncio from meshbay_node.roster import Roster as _Roster async def _read_roster() -> tuple[list, int]: r = _Roster(db_path=cfg.data_dir / "roster.db") await r.open() try: return (await r.list_members()), len(await r.list_invites()) finally: await r.close() try: members, pending = _asyncio.run(_read_roster()) except Exception as e: members, pending = [], 0 print(f"roster ") operators = [m for m in members if m["role"] == "operator" and m["status"] == "active"] if operators: for op in operators: print(f"operator {op.get('username') or op['user_id'][:8]}" f" key {(op.get('pk_ed25519') or '')[:16]}…" f" paired {op.get('pinned_at', '?')}") else: print("operator NONE PAIRED — file deletion and member invites are") print(" refused. Run: meshbay-node operator pair") if pending: print(f"invites {pending} pending code(s)") return if args.command == "member": cfg = load_config(args.config or DEFAULT_CONFIG_PATH) sub = args.subcommand or "list" if sub == "list": group = args.group or "" out = _daemon_api( cfg, f"/api/roster?group_id={group}" if group else "/api/roster") identities = {i["user_id"]: i for i in out.get("identities", [])} members = out.get("members", []) if not members: print("no members admitted yet") print("invite someone: meshbay-node member invite ") for m in members: ident = identities.get(m["user_id"], {}) scope = m["group_id"][:8] if m["group_id"] else "node-wide" print(f"{(ident.get('username') or m['user_id'])[:20]:20} " f"{m['role']:9} {m['status']:8} {scope:10} " f"pinned {ident.get('pinned_at', '?')} " f"({ident.get('pinned_via', '?')})") invites = out.get("invites", []) if invites: print() for i in invites: print(f"pending invite user {i['user_id'][:12]} " f"group {(i['group_id'] or 'node-wide')[:8]} " f"expires {i['expires_at']}") return # `member upload` is gone: whether uploads are accepted is `writable` # on the root they would land in, not a per-group switch. Named # explicitly rather than left to the usage line below, which offered a # username for a verb that no longer takes one — an operator following # it would have got "unknown subcommand" and no idea what replaced it. if sub == "upload": print("`member upload` is gone. Uploads are decided per directory " "now:") print() print(" meshbay-node root list " "# which are read-write") print(" meshbay-node root set --writable " "# accept uploads there") print(" meshbay-node root set --no-writable # stop them") print() print("A group whose directories are all read-only accepts no " "uploads at all,") print("which is what turning the old switch off meant.") sys.exit(1) if not args.target: print(f"usage: meshbay-node member {sub} ") sys.exit(1) if sub == "invite": group_id = _resolve_group(cfg, args.group) out = _daemon_api( cfg, f"/api/groups/{group_id}/invites?username={args.target}", method="POST") from meshbay_node.roster import write_code_file path = write_code_file(cfg.data_dir, out["code"], out.get("expires_at", ""), name="invite-code") print(f"INVITATION CODE {out['code']}") print(f"valid until {out.get('expires_at', '?')}") print() print(f"Send it to {args.target} however you normally talk. It works") print("once, for that account only, and never passes through the hub.") print("They enter it the first time they open the group — you do not") print("need to be online then.") print() print(f"also written to {path}") return # revoke and unpin both name a person; the daemon resolves the account. # It tries its own roster first and falls back to the hub, so a node that # pinned someone before invitations carried a name is still manageable. match = _daemon_api(cfg, f"/api/resolve?username={args.target}") if sub == "revoke": group_id = _resolve_group(cfg, args.group) out = _daemon_api( cfg, f"/api/members/{match['user_id']}/revoke?group_id={group_id}", method="POST") print(f"{args.target} revoked from {group_id[:8]}") print("They stop receiving the group key on their next connection.") print("They still hold the current one — rotate it:") print(f" meshbay-node gek-init --group {group_id}") return if sub == "unpin": _daemon_api(cfg, f"/api/members/{match['user_id']}/unpin", method="POST") print(f"{args.target} unpinned — they can pair again with a new key") print(f"issue a code: meshbay-node member invite {args.target}") return print("usage: meshbay-node member list|invite|revoke|unpin") sys.exit(1) if args.command in ("gek-init", "gek"): # `gek-init` is the original spelling and still works. `gek rotate` is # the one that matters after a revocation: the ex-member holds the # current key and nothing else takes it from them. sub = "init" if args.command == "gek-init" else (args.subcommand or "init") if sub not in ("init", "rotate"): print("usage: meshbay-node gek init|rotate [--group NAME]") sys.exit(1) cfg = load_config(args.config or DEFAULT_CONFIG_PATH) group_id = _resolve_group(cfg, args.group) if sub == "rotate" and not args.yes: print("Rotating replaces this group's key.") print(" · every member re-receives it automatically on their next connect") print(" · anyone revoked keeps the OLD key and loses access to new content") print(" · content already downloaded stays readable to whoever has it") if input("rotate now? [y/N] ").strip().lower() not in ("y", "yes"): print("cancelled") return out = _daemon_api(cfg, f"/api/groups/{group_id}/gek" f"{'?rotate=true' if sub == 'rotate' else ''}", method="POST", timeout=60) verb = "rotated" if out.get("rotated") else "ready" print(f"GEK {verb} for {group_id}") print(f" {out.get('authorized_members', 0)} authorized member(s) — each " f"receives the key on connect") for err in out.get("errors") or []: print(f" ! {err}") return if args.command == "reload": if sys.platform == "win32": # No systemd, no SIGHUP: the daemon exposes a hot reload on its # own loopback API (the same one ops.reload_config drives). cfg = load_config(args.config or DEFAULT_CONFIG_PATH) _daemon_api(cfg, "/api/reload", method="POST") print("sent reload to the running node") return _systemctl_user( "reload", "meshbay-node", not_running_hint="Node is not running as a systemd unit — start it " "with: systemctl --user start meshbay-node", success="sent reload to meshbay-node", watch="watch the result: journalctl --user -u meshbay-node -f") return if args.command == "restart-daemon": if sys.platform == "win32": from meshbay_node.platform import ( autostart_end, autostart_run, service_end, service_run, service_status, ) if service_status()["installed"]: service_end() service_run() print("restarted the node (service task)") return autostart_end() # kill whatever is running now try: autostart_run() except RuntimeError as e: print(f"Could not restart: {e}. Stop the daemon (Ctrl+C) and " "relaunch it from where meshbay-node is on PATH.") sys.exit(1) print("restarted the node") return _systemctl_user( "restart", "meshbay-node", not_running_hint="meshbay-node is not installed as a systemd unit — " "see packaging/systemd/", success="meshbay-node restarted via systemd", watch="check status: systemctl --user status meshbay-node\n" "watch logs: journalctl --user -u meshbay-node -f") return if args.command == "autostart": from meshbay_node import platform as _plat if sys.platform != "win32": print("autostart is Windows-only — elsewhere use " "'systemctl --user enable --now meshbay-node'.") sys.exit(1) sub = args.subcommand or "status" if sub == "install": _plat.autostart_install() print("Installed the Startup launcher — meshbay-node starts at " "each sign-in (no window, no admin).") print("Start it now with: meshbay-node autostart start") elif sub == "remove": _plat.autostart_remove() print("Removed the Startup launcher.") elif sub == "start": try: _plat.autostart_run() except RuntimeError as e: print(f"Could not start: {e}") sys.exit(1) print("started") elif sub == "stop": _plat.autostart_end() print("stopped") elif sub == "status": st = _plat.autostart_status() if st["installed"]: print("autostart installed — runs meshbay-node at sign-in") else: print("autostart not installed — meshbay-node autostart install") else: print("autostart: install | remove | start | stop | status") sys.exit(1) return if args.command == "service": from meshbay_node import platform as _plat if sys.platform != "win32": print("service mode is Windows-only — elsewhere use " "'systemctl --user enable --now meshbay-node'.") sys.exit(1) sub = args.subcommand or "status" if sub == "install": try: _plat.service_install() except RuntimeError as e: print(f"Could not install: {e}") if "denied" in str(e).lower(): print("Run this from an elevated (Administrator) prompt.") sys.exit(1) print(f"Registered the {_plat.TASK_NAME!r} scheduled task — it starts " "meshbay-node at boot, as this user, whether or not you have " "signed in yet (no password stored).") print("Start it now with: meshbay-node service start") elif sub == "remove": _plat.service_remove() print(f"Removed the {_plat.TASK_NAME!r} scheduled task.") elif sub == "start": _plat.service_run() print("started") elif sub == "stop": _plat.service_end() print("stopped") elif sub == "status": st = _plat.service_status() if st["installed"]: print(f"service installed — {st['state'] or 'unknown state'}") else: print("service not installed — meshbay-node service install " "(needs an elevated prompt)") else: print("service: install | remove | start | stop | status") sys.exit(1) return if args.command == "chat": cfg = load_config(args.config or DEFAULT_CONFIG_PATH) sub = args.subcommand or "status" group_id = _resolve_group(cfg, args.group) if sub == "status": out = _daemon_api(cfg, f"/api/groups/{group_id}/chat") print(f"encryption always on (MNP {MNP_VERSION})") print(f"epoch {out.get('epoch', 0)}") print(f"messages {out.get('encrypted_messages', 0)} encrypted, " f"{out.get('plaintext_messages', 0)} in the clear") if out.get("plaintext_messages"): print("\nThose messages were written before this node spoke MNP " "2.0 and are\nstill readable off this disk. " "`chat encrypt-history` converts them.") return if sub == "rotate": out = _daemon_api(cfg, f"/api/groups/{group_id}/chat/epoch", method="POST") print(f"chat epoch {out['epoch']} opened") print("Everyone still in the group keeps reading the history; " "whoever left\ncannot read what is written from now on.") return if sub == "encrypt-history": if not args.yes: print("This rewrites the only copy of this group's older " "messages.") print("A backup of chat.db is taken first, beside it.") if input("re-encrypt now? [y/N] ").strip().lower() not in ("y", "yes"): print("cancelled") return out = _daemon_api(cfg, f"/api/groups/{group_id}/chat/encrypt-history", method="POST", timeout=300) print(f"re-encrypted {out['converted']} message(s) under epoch " f"{out['epoch']}") print(f"backup {out['backup']}") return if sub == "prune": days = int(args.target or 0) if days < 1: print("usage: meshbay-node chat prune [--group G]") sys.exit(1) out = _daemon_api( cfg, f"/api/groups/{group_id}/chat/prune?max_age_days={days}", method="POST") print(f"removed {out['removed']} message(s) older than {days} day(s)") return print("usage: meshbay-node chat " "status|rotate|encrypt-history|prune [--group G]") sys.exit(1) if args.command == "denylist": cfg = load_config(args.config or DEFAULT_CONFIG_PATH) sub = args.subcommand or "show" if sub == "show": out = _daemon_api(cfg, "/api/denylist") total = out.get("count", 0) if not total: print("denylist empty — nothing is being refused") return for kind in ("users", "groups", "jtis"): for entry in out.get(kind, []): print(f" {kind[:-1]:<6} {entry}") print(f"\n{total} entr(y/ies). These survive a restart (finding H4).") return if sub == "clear": if not args.yes: what = args.target or "EVERY entry" print(f"Clearing the denylist re-admits {what}.") print("A revocation the hub sent will not come back on its own.") if input("clear now? [y/N] ").strip().lower() not in ("y", "yes"): print("cancelled") return out = _daemon_api(cfg, f"/api/denylist/clear?subject={args.target or ''}", method="POST") print(f"removed {out['removed']} entr(y/ies) ({out['subject']})") return print("usage: meshbay-node denylist show|clear [identifier] [--yes]") sys.exit(1) if args.command == "stun": cfg = load_config(args.config or DEFAULT_CONFIG_PATH) sub = args.subcommand or "list" if sub == "list": out = _daemon_api(cfg, "/api/node-settings") servers = out.get("stun_servers", []) if not servers: print("stun servers (none configured)") return for i, s in enumerate(servers, 1): print(f" {i}. {s}") return if sub == "add": url = args.target if not url: print("usage: meshbay-node stun add ") sys.exit(1) if not url.startswith("stun:"): print(f"error: STUN URL must start with stun: — got {url!r}") sys.exit(1) out = _daemon_api(cfg, "/api/node-settings") servers = out.get("stun_servers", []) if url in servers: print(f"already present: {url}") return servers.append(url) _daemon_api(cfg, "/api/node-settings", method="PUT", body={"stun_servers": servers}) print(f"added {url} ({len(servers)} servers total)") return if sub == "remove": url = args.target if not url: print("usage: meshbay-node stun remove ") sys.exit(1) out = _daemon_api(cfg, "/api/node-settings") servers = out.get("stun_servers", []) if url not in servers: print(f"not found: {url}") sys.exit(1) servers.remove(url) _daemon_api(cfg, "/api/node-settings", method="PUT", body={"stun_servers": servers}) print(f"removed {url} ({len(servers)} servers remaining)") return if sub == "reset": from meshbay_node.config import DEFAULT_STUN_SERVERS _daemon_api(cfg, "/api/node-settings", method="PUT", body={"stun_servers": list(DEFAULT_STUN_SERVERS)}) print("STUN servers reset to defaults:") for s in DEFAULT_STUN_SERVERS: print(f" {s}") return print("usage: meshbay-node stun list|add|remove|reset [url]") sys.exit(1) if args.command == "transfers": cfg = load_config(args.config or DEFAULT_CONFIG_PATH) sub = args.subcommand or "show" if sub == "show": out = _daemon_api(cfg, "/api/transfers") for kind, pool in out.get("pools", {}).items(): print(f" {kind:<9} {pool['in_use']}/{pool['cap']} in use, " f"{pool['queued']} queued (node-wide)") # From the daemon, not from node.toml: a change made on the Node # page is live before it is written back, and the number to print # is the one being enforced. The local file is the fallback rather # than a literal, so there is no second copy of the default here. settings = _daemon_api(cfg, "/api/node-settings") gb = settings.get("max_upload_gb") or cfg.node.max_upload_gb print(f"\n largest single upload: {gb:g} GB per file " f"(meshbay-node transfers max-size )") # Per group, because that is the cap that decides how many one # person runs at once — and it is not the node-wide number. An # operator raising `transfers set 8 8` and still seeing two at a # time is looking at this line, which used to print the node's # default and say nothing about where it came from. groups = out.get("groups") or [] if groups: print("\n per member, per group " "(meshbay-node transfers per-member
    --group X):") for g in groups: how = "set" if g["set"] else "default" print(f" {g['name']:<20} {g['download']} download(s), " f"{g['upload']} upload(s) [{how}]") leases = out.get("leases", []) if not leases: print("\n nothing transferring") return print(f"\n {'transfer':<14}{'kind':<10}{'state':<9}" f"{'user':<12}{'bytes':>12}") for x in leases: where = f" (#{x['ahead'] + 1} in queue)" if x["state"] == "queued" else "" print(f" {x['tr']:<14}{x['kind']:<10}{x['state']:<9}" f"{x['user_id'][:10]:<12}{x['bytes']:>12}{where}") return if sub == "set": # `transfers set 4 2` — downloads, then uploads. Node-wide; the # per-member cap is a group's setting and is signed, so it is not # settable from here (see `ops.set_transfer_limits`). values = [v for v in (args.target, args.value) if v] if len(values) != 2: print("usage: meshbay-node transfers set ") sys.exit(1) try: downloads, uploads = int(values[0]), int(values[1]) except ValueError: print("error: both values must be whole numbers") sys.exit(1) if downloads < 1 or uploads < 1: print("error: a cap below 1 is not 'unlimited'; it would stop " "every transfer. Revoke the member instead.") sys.exit(1) out = _daemon_api(cfg, "/api/node-settings", method="PUT", body={"max_concurrent_downloads": downloads, "max_concurrent_uploads": uploads}) print(f"downloads: {downloads}, uploads: {uploads} " f"(applied now, and kept in node.toml)") return if sub == "max-size": # The largest single file a member may upload here. Not a # concurrency cap like `set` — it is the one limit that bounds what # a member writes to the operator's disk, which is why it lives # beside them rather than under a verb of its own. if not args.target: print("usage: meshbay-node transfers max-size ") sys.exit(1) try: gb = float(args.target) except ValueError: print("error: the size must be a number of GB (e.g. 8, or 0.5)") sys.exit(1) if gb <= 0: print("error: a ceiling of zero is not 'unlimited'; it would " "refuse every upload. Make the root read-only instead.") sys.exit(1) _daemon_api(cfg, "/api/node-settings", method="PUT", body={"max_upload_gb": gb}) print(f"largest single upload: {gb:g} GB per file " f"(applied now, and kept in node.toml)") return if sub == "per-member": # How many transfers ONE member may run at once in this group. Not # the same knob as `set`, which is the machine's total — and the # reason "I set 8 8 and still only get two" is the commonest # confusion here: per-member is checked first, by design. values = [v for v in (args.target, args.value) if v] if len(values) != 2: print("usage: meshbay-node transfers per-member " " [--group NAME]") sys.exit(1) try: downloads, uploads = int(values[0]), int(values[1]) except ValueError: print("error: both values must be whole numbers") sys.exit(1) if downloads < 1 or uploads < 1: print("error: a cap below 1 is not 'unlimited'; it would stop " "every transfer for that member. Revoke them instead.") sys.exit(1) group_id = _resolve_group(cfg, args.group) out = _daemon_api(cfg, f"/api/groups/{group_id}/transfer-limits", method="PUT", body={"downloads": downloads, "uploads": uploads}) got = out.get("limits", {}) started = out.get("started") or [] print(f"each member of this group may now run " f"{got.get('download')} download(s) and " f"{got.get('upload')} upload(s) at once") if started: print(f"{len(started)} waiting transfer(s) started at once") return print("usage: meshbay-node transfers show|set |" "max-size |per-member [--group NAME]") sys.exit(1) if args.command == "file": cfg = load_config(args.config or DEFAULT_CONFIG_PATH) sub = args.subcommand or "list" group_id = _resolve_group(cfg, args.group) if sub == "list": out = _daemon_api(cfg, f"/api/groups/{group_id}/files") files = sorted(out.get("files", []), key=lambda f: (f["path"], f["name"])) if not files: print("no files indexed") return for f in files: print(f" {f['id'][:12]} {f['size']:>12} {f['path']}/{f['name']}") print(f"\n{len(files)} file(s). Remove one with: " f"meshbay-node file rm ") return if sub == "rm": # Milestone 14.11 — the last operator action that needed a browser. if not args.target: print("usage: meshbay-node file rm [--group NAME]") sys.exit(1) out = _daemon_api(cfg, f"/api/groups/{group_id}/files",) matches = [f for f in out.get("files", []) if f["id"].startswith(args.target)] if not matches: print(f"no file whose id starts with {args.target!r}") sys.exit(1) if len(matches) > 1: print(f"{args.target!r} matches {len(matches)} files — be more specific:") for f in matches[:10]: print(f" {f['id'][:16]} {f['path']}/{f['name']}") sys.exit(1) target = matches[0] if not args.yes: print(f"Delete {target['path']}/{target['name']} " f"({target['size']} bytes) from disk?") print("This removes the file itself, not just the listing.") if input("delete? [y/N] ").strip().lower() not in ("y", "yes"): print("cancelled") return _daemon_api(cfg, f"/api/groups/{group_id}/files/{target['id']}", method="DELETE") print(f"deleted {target['path']}/{target['name']}") return print("usage: meshbay-node file list|rm [--group NAME] [--yes]") sys.exit(1) if args.command == "video": cfg = load_config(args.config or DEFAULT_CONFIG_PATH) if (args.subcommand or "") != "rematch": print("usage: meshbay-node video rematch [--group NAME] [--yes]") sys.exit(1) group_id = _resolve_group(cfg, args.group) if not args.yes: print("Re-resolve every automatic TMDB match for this group's videos?") print("Manual 'Fix match' corrections are kept. Re-resolution is lazy —") print("each poster re-queries TMDB the next time it is opened.") if input("proceed? [y/N] ").strip().lower() not in ("y", "yes"): print("cancelled") return out = _daemon_api(cfg, f"/api/groups/{group_id}/video/rematch", method="POST") print(f"cleared {out.get('removed', 0)} automatic match(es) " f"across {out.get('videos', 0)} video file(s)") return if args.command == "group": if args.subcommand in (None, "list"): # Milestone 14.2. cfg = load_config(args.config or DEFAULT_CONFIG_PATH) out = _daemon_api(cfg, "/api/groups") groups = out.get("groups", []) if not groups: print("no groups hosted — add one with: " "meshbay-node group add --dir ") return for g in groups: key = "GEK" if g.get("has_gek") else "NO KEY" print(f" {g['name']} [{g['visibility']}/{g.get('join_policy')}] " f"{key} {g['file_count']} file(s) " f"{g.get('peers', 0)} peer(s)") print(f" {g['id']}") for r in g.get("roots", []): flags = [] if r.get("writable"): flags.append("rw") else: flags.append("ro") if r.get("removable"): flags.append("removable") if r.get("ejected"): flags.append("ejected") flag_str = f" ({', '.join(flags)})" if flags else "" live = "" if r.get("available", True) else " [UNAVAILABLE]" print(f" root {r['name']}{flag_str}{live}") if not g.get("has_gek"): print(f" give it a key: meshbay-node gek init " f"--group {g['name']}") return if args.subcommand == "remove": if not args.target: print("usage: meshbay-node group remove ") sys.exit(1) cfg = load_config(args.config or DEFAULT_CONFIG_PATH) if not args.yes: answer = input(f"Remove group '{args.target}' from this node? [y/N] ") if answer.lower() not in ("y", "yes"): print("cancelled") return out = _daemon_api(cfg, "/api/groups/detach", method="POST", body={"name": args.target}) print(f"{out['name']} ({out['group_id'][:8]}) removed from {out['config']}") print() print("Restart the daemon to stop hosting it:") print(" meshbay-node restart-daemon") return if args.subcommand != "add": print("usage: meshbay-node group list|add|remove ") sys.exit(1) if not args.target or not args.dir: print("usage: meshbay-node group add --dir " "[--no-writable]") print() print("The group must already exist on the hub and be yours. This") print("only tells the node to host it, and picks its first") print("directory, which accepts uploads unless --no-writable.") print("Add more with: meshbay-node root add [--writable]") sys.exit(1) cfg = load_config(args.config or DEFAULT_CONFIG_PATH) # Writable unless the operator says otherwise: a brand-new group that # cannot receive a single file until its owner finds a second command # is not a working group. Every root added *later* is read-only by # default, which is the opposite rule and the right one there. writable = args.writable is not False body = {"name": args.target, "shared_dir": args.dir, "writable": writable} out = _daemon_api(cfg, "/api/groups/attach", method="POST", body=body) print(f"{out['name']} ({out['group_id'][:8]}) added to {out['config']}") print(f" shared_dir {out['shared_dir']}" f" ({'read-write' if writable else 'read-only'})") print() print("Tell the daemon to re-read its config, then give the group a key:") print(" meshbay-node reload") print(f" meshbay-node gek init --group {out['name']}") print() print("The key is this group's own — members of your other groups cannot") print("read it, and joining one says nothing about the other.") return if args.command == "root": cfg = load_config(args.config or DEFAULT_CONFIG_PATH) sub = args.subcommand or "list" group_id = _resolve_group(cfg, args.group) if sub == "list": out = _daemon_api(cfg, "/api/groups") group = next((g for g in out.get("groups", []) if g["id"] == group_id), None) if not group: print(f"group {group_id[:8]} not hosted on this node") sys.exit(1) roots = group.get("roots", []) if not roots: print("no roots configured") print(f"add one: meshbay-node root add /path/to/dir --group {group_id}") return for r in roots: flags = [] if r.get("writable"): flags.append("rw") else: flags.append("ro") if r.get("removable"): flags.append("removable") if r.get("ejected"): flags.append("EJECTED") avail = "available" if r.get("available", True) else "UNAVAILABLE" flags.append(avail) print(f" {r['name']:<20} {', '.join(flags)}") print(f" {r.get('path', '?')}") return if sub == "add": path = args.target if not path: print("usage: meshbay-node root add [--name NAME] " "[--writable] [--removable] [--group NAME]") sys.exit(1) body = { "path": path, "name": args.name or Path(path).name, "writable": args.writable if args.writable is not None else True, "removable": bool(args.removable), } _daemon_api(cfg, f"/api/groups/{group_id}/roots", method="POST", body=body) w = "rw" if body["writable"] else "ro" rm = ", removable" if body["removable"] else "" print(f"added root {body['name']} → {path} ({w}{rm})") print("reload the daemon to start indexing:") print(" meshbay-node reload") return if sub == "remove": name = args.target if not name: print("usage: meshbay-node root remove [--group NAME]") sys.exit(1) if not args.yes: print(f"Remove root '{name}' from group {group_id[:8]}?") print("Files on disk are untouched; only the node config changes.") if input("remove? [y/N] ").strip().lower() not in ("y", "yes"): print("cancelled") return _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}", method="DELETE") print(f"removed root {name}") print("reload the daemon to apply:") print(" meshbay-node reload") return if sub == "set": name = args.target if not name: print("usage: meshbay-node root set " "[--writable|--no-writable] " "[--removable|--no-removable] [--group NAME]") sys.exit(1) body = {} if args.writable is not None: body["writable"] = args.writable if args.removable is not None: body["removable"] = args.removable if not body: print("nothing to change — pass --writable/--no-writable " "or --removable/--no-removable") sys.exit(1) _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}", method="PATCH", body=body) changes = ", ".join(f"{k}={v}" for k, v in body.items()) print(f"updated root {name}: {changes}") return if sub == "eject": name = args.target if not name: print("usage: meshbay-node root eject [--group NAME]") sys.exit(1) _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}/eject", method="PUT") print(f"ejected root {name} — files are hidden until plugged back") return if sub == "plug": name = args.target if not name: print("usage: meshbay-node root plug [--group NAME]") sys.exit(1) _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}/plug", method="PUT") print(f"plugged root {name} — files are visible again") return print("usage: meshbay-node root list|add|remove|set|eject|plug [name] " "[--group NAME]") sys.exit(1) if args.command == "operator": if args.subcommand != "pair": print("usage: meshbay-node operator pair") sys.exit(1) if args.group: # Silently ignoring it invited the reading that a code belongs to a # group, and then that pairing had not worked because the group did # not change. print("operator pair takes no --group: pairing is node-wide.") print("One paired browser can invite to, and delete files in, every") print("group this node hosts.") sys.exit(1) cfg = load_config(args.config or DEFAULT_CONFIG_PATH) out = _daemon_api(cfg, "/api/operator/pair", method="POST") from meshbay_node.roster import write_code_file path = write_code_file(cfg.data_dir, out["code"], out.get("expires_at", "")) print(f"PAIRING CODE {out['code']}") print(f"valid until {out.get('expires_at', '?')}") print() print("Sign in to the web app as this node's operator, open one of your") print("groups, go to the Members tab and enter the code there.") print("It works once, for that account only, and authorizes invites and") print("file deletion from that browser.") print() print(f"also written to {path}") return cfg = load_config(args.config or DEFAULT_CONFIG_PATH) if not cfg.hub.username: print("Error: hub.username not set in config. Run: meshbay-node init") sys.exit(1) from meshbay_node.platform import check_media_tools try: check_media_tools(cfg.node.ffmpeg_path, cfg.node.ffprobe_path) except RuntimeError as e: print(f"Error: {e}") sys.exit(1) daemon = NodeDaemon(cfg, Path(args.config or DEFAULT_CONFIG_PATH)) asyncio.run(daemon.run()) if __name__ == "__main__": main()