"""Node-wide settings: denylist, node.toml values, transfer caps, scan timing, reload.""" from __future__ import annotations import logging from pathlib import Path from meshbay_common.background import spawn from meshbay_node.config import DEFAULT_CONFIG_PATH from meshbay_node.ops.core import OpError, _config, _group_ctx, _roster from meshbay_node.ops.node_toml import _update_node_toml from meshbay_node.roster import Roster log = logging.getLogger("meshbay_node.ops") # ── Revocation denylist ────────────────────────────────────────────────────── async def read_denylist(state: dict) -> dict: """Milestone 14.10 — what the node is currently refusing.""" denylist = state.get("denylist") if not denylist: return {"users": [], "groups": [], "jtis": [], "count": 0} entries = denylist.entries() return {**entries, "count": sum(len(v) for v in entries.values())} async def clear_denylist(state: dict, *, subject: str = "") -> dict: """ Drop denylist entries — all of them, or one identifier. Deliberately not silent: a cleared denylist re-admits whoever it was keeping out, and the count is what tells the operator whether they undid one revocation or all of them. """ denylist = state.get("denylist") if not denylist: raise OpError("No denylist in this process", status=503) removed = denylist.clear(subject) log.warning("Denylist cleared (%s): %d entr(y/ies) removed", subject or "all", removed) return {"status": "cleared", "removed": removed, "subject": subject or "all"} # ── Node settings ──────────────────────────────────────────────────────────── # What `set_node_settings` accepts, and how each value is validated. A module # constant so a test can hold its key set against `Roster.node_setting_keys()`: # this is the third list of the same settings, and the first two had already # drifted apart once — the reader's defaults covered fewer settings than the # resolver answered for, which is how node.toml's transfer pools came to be # parsed and then ignored. The kinds here are the *writer's* validation and # deliberately not the resolver's coercions. NODE_SETTING_WRITERS: dict[str, tuple[str, str]] = { "invite_ttl_hours": ("int", Roster.SETTING_INVITE_TTL), "pair_ttl_hours": ("int", Roster.SETTING_PAIR_TTL), "device_request_ttl_minutes": ("int", Roster.SETTING_DEVICE_TTL), "max_concurrent_streams": ("int", Roster.SETTING_MAX_STREAMS), "max_concurrent_downloads": ("int", Roster.SETTING_MAX_DOWNLOADS), "max_concurrent_uploads": ("int", Roster.SETTING_MAX_UPLOADS), "max_upload_gb": ("size", Roster.SETTING_MAX_UPLOAD_GB), "transcode_incompatible_video": ("bool", Roster.SETTING_TRANSCODE), "stun_servers": ("stun_list", Roster.SETTING_STUN_SERVERS), "ice_interfaces": ("list", Roster.SETTING_ICE_INTERFACES), } async def get_node_settings(state: dict) -> dict: """Return current effective node settings.""" from meshbay_node.config import node_settings_defaults roster = _roster(state) config = _config(state) defaults = node_settings_defaults(config.node) if roster: return await roster.node_settings(defaults) return defaults async def set_node_settings(state: dict, settings: dict) -> dict: """Update node-level daemon settings. Writes to both roster.db and node.toml.""" roster = _roster(state) config = _config(state) nd = config.node conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH) allowed_keys = NODE_SETTING_WRITERS set_by = state.get("node_user_id", "") updated = {} for key, value in settings.items(): if key not in allowed_keys: continue kind, setting_key = allowed_keys[key] if kind == "int": try: v = int(value) except (TypeError, ValueError): raise OpError(f"{key} must be an integer") if v < 1: raise OpError(f"{key} must be positive") setattr(nd, key, v) await roster.set_node_setting(setting_key, str(v), set_by) updated[key] = v elif kind == "size": # A quantity, not a count: half a gigabyte is a legitimate ceiling # on a small disk, so this one is not run through the `int` branch # above, whose floor of 1 would round it to "refuse everything". # bool before float, as config.py does it: `true` is not 1 GB. if isinstance(value, bool): raise OpError(f"{key} must be a number") try: fv = float(value) except (TypeError, ValueError): raise OpError(f"{key} must be a number") if fv <= 0: raise OpError(f"{key} must be greater than zero") setattr(nd, key, fv) await roster.set_node_setting(setting_key, repr(fv), set_by) updated[key] = fv elif kind == "bool": v = bool(value) setattr(nd, key, v) await roster.set_node_setting(setting_key, "1" if v else "0", set_by) updated[key] = v elif kind in ("list", "stun_list"): import json as _json if not isinstance(value, list): raise OpError(f"{key} must be a list") v = [str(s) for s in value] if kind == "stun_list": for s in v: if not s.startswith("stun:"): raise OpError(f"Invalid STUN server: {s} (must start with stun:)") setattr(nd, key, v) await roster.set_node_setting(setting_key, _json.dumps(v), set_by) updated[key] = v if updated: _update_node_toml(conf_path, updated) if "max_concurrent_streams" in updated: webrtc = state.get("webrtc") # `webrtc._stream_sem` was assigned here for months. That attribute # has never existed -- the pool is `ctx["_transcode_sem"]` -- so the # `hasattr` guard was always False and the setting only ever took # effect on a restart, which docs/MESHBAY_DESIGN.md §6.8 says it # does not need. if webrtc is not None: webrtc.set_capacity( max_concurrent_streams=updated["max_concurrent_streams"]) if ("max_concurrent_downloads" in updated or "max_concurrent_uploads" in updated): webrtc = state.get("webrtc") if webrtc is not None: webrtc.set_capacity( max_concurrent_downloads=updated.get( "max_concurrent_downloads"), max_concurrent_uploads=updated.get( "max_concurrent_uploads")) if "max_upload_gb" in updated: webrtc = state.get("webrtc") if webrtc is not None: webrtc.set_capacity(max_upload_gb=updated["max_upload_gb"]) if "stun_servers" in updated: webrtc = state.get("webrtc") if webrtc and hasattr(webrtc, '_stun'): webrtc._stun = updated["stun_servers"] from meshbay_node.transport.stun_multi import set_servers as _set_stun _set_stun(updated["stun_servers"]) if "ice_interfaces" in updated: from meshbay_node.transport.ice_filter import install as install_ice_filter install_ice_filter(updated["ice_interfaces"] or None) log.info("Node settings updated: %s", updated) return {"updated": updated} # ── Transfers ──────────────────────────────────────────────────────────────── async def set_transfer_limits(state: dict, group_id: str, downloads: int, uploads: int) -> dict: """How many transfers one member may run at once in this group. Same shape as every other operator setting: lives on the node (roster.db, not the hub and not node.toml, for the reason change 5 gives — a hub that decided this would have authority over someone else's machine), signed (webrtc_server checks the caller's admin authority before this runs), and live, so the pools are updated in place rather than at the next restart. """ roster = _roster(state) ctx = _group_ctx(state, group_id) limits = await roster.set_transfer_limits( group_id, {"download": downloads, "upload": uploads}, set_by=state.get("node_user_id", "")) ctx["transfer_limits"] = limits webrtc = state.get("webrtc") slots = getattr(webrtc, "_ctx", {}).get("_transfer_slots") if webrtc else None granted = slots.set_group_limits(group_id, limits) if slots else [] # And **tell them**. The node-wide path (`WebRTCTransport.set_capacity`) # does this and this one did not: the leases were granted in the pool and # the peers waiting on them were never told, so a cap raised from 2 to 4 # left both transfers sitting at "waiting" until the client's own watchdog # re-asked a minute later. That is §5.2's first row — "node granted a slot, # the push was lost" — reached by writing the grant and forgetting the send, # which is the same omission as the missing `touch()` one layer up. for lease in granted: webrtc._notify_granted(lease) log.info("Transfer limits for group %s: %s (%d started at once)", group_id[:8], limits, len(granted)) return {"group_id": group_id, "limits": limits, "started": [x.tr for x in granted]} async def list_transfers(state: dict) -> dict: """Live transfer leases and queue depth. The operator's window into "is anything actually holding a slot". When somebody reports a transfer stuck at waiting, this is the only thing that says whether the node ever had them in a queue — the alternative is reading a log for a line that, by definition, is not being printed. Carries no filename and no path: a lease holds neither, and this is exactly where it would be tempting to add one. """ webrtc = state.get("webrtc") ctx = getattr(webrtc, "_ctx", {}) if webrtc else {} slots = ctx.get("_transfer_slots") if slots is None: from meshbay_node.transfers import DEFAULT_MAX_CONCURRENT, DEFAULT_MAX_PER_MEMBER, KINDS # No pool built means nothing has transferred since the daemon started, # which is a real answer and not an error. # # The caps still have to be the operator's own. Reporting the module # defaults here was worse than reporting nothing: `transfers set 2 2` # answered "applied now", and `transfers show` immediately said 0/8 — # a setting written, acknowledged and displayed wrong, which reads # exactly like the hot-swap that did nothing for months. Found by # running it, not by a test: the test asserted the defaults and so # agreed with the bug. return {"pools": { k: {"in_use": 0, "cap": int(ctx.get(f"max_concurrent_{k}s") or DEFAULT_MAX_CONCURRENT), "per_member": DEFAULT_MAX_PER_MEMBER, "queued": 0} for k in KINDS}, "leases": [], "groups": _group_limits(state)} out = slots.snapshot() out["groups"] = _group_limits(state) return out def _group_limits(state: dict) -> list[dict]: """Each group's per-member caps, as the operator set them. Reported because `transfers show` used to print only the node's default and an operator reading "2 per member" had no way to tell whether that was this group's setting or the fallback — and no way to change it either, since the signed op had no door but MNP. Both were the same bug wearing two faces. """ from meshbay_node.transfers import DEFAULT_MAX_PER_MEMBER config = state.get("config") groups_ctx = state.get("groups_ctx") or {} out = [] for group in (getattr(config, "groups", None) or []): limits = (groups_ctx.get(group.id) or {}).get("transfer_limits") or {} out.append({ "group_id": group.id, "name": group.name, "download": int(limits.get("download") or DEFAULT_MAX_PER_MEMBER), "upload": int(limits.get("upload") or DEFAULT_MAX_PER_MEMBER), "set": bool(limits), }) return out # ── Scan settings ──────────────────────────────────────────────────────────── async def set_scan_settings(state: dict, group_id: str, reconcile_interval_secs: float, debounce_secs: float) -> dict: """ How often the indexer's reconciliation backstop runs, and how long a changed file is left alone before being hashed (indexer.py DirectoryIndexer). Persisted like set_enabled_apps — but there is also a *live* DirectoryIndexer object to update, since it reads these once at construction and runs its own background loop with them rather than consulting groups_ctx on every use. """ roster = _roster(state) await roster.set_scan_settings(group_id, reconcile_interval_secs, debounce_secs, set_by=state.get("node_user_id", "")) indexer = state.get("indexers", {}).get(group_id) if indexer: indexer.reconcile_secs = reconcile_interval_secs indexer.debounce_secs = debounce_secs # Apply the new interval now rather than after whatever backoff had # already stretched the wait to. indexer.note_activity() # Optional, unlike _group_ctx(): a group can be persisted here before it # is hot-loaded (or in a test that only cares about the roster/indexer # side), and that must not turn a successful write into a 404. ctx = state.get("groups_ctx", {}).get(group_id) if ctx is not None: ctx["reconcile_interval_secs"] = reconcile_interval_secs ctx["debounce_secs"] = debounce_secs log.info("Scan settings for group %s: reconcile=%.0fs debounce=%.0fs", group_id[:8], reconcile_interval_secs, debounce_secs) return {"reconcile_interval_secs": reconcile_interval_secs, "debounce_secs": debounce_secs, "group_id": group_id} # ── Reload ────────────────────────────────────────────────────────────────── async def reload_config(state: dict) -> dict: """Hot-reload node.toml without dropping connections. Blocks until the reload actually finishes — see start_reload for why the loopback route uses that instead.""" reload_fn = state.get("reload_fn") if not reload_fn: raise OpError("Reload not available", status=503) await reload_fn() return {"status": "reloaded"} async def start_reload(state: dict) -> dict: """ Same as reload_config, but does not wait for the reload to finish. The loopback route uses this one: the Electron bridge caps every call at a fixed 30s (main.js node:call), and hot-loading a brand-new group runs its full initial scan synchronously inside _reload_config_inner() (daemon.py) before that coroutine returns — minutes, not seconds, on a real library (found against a 45 GB group on the same slow disk the StarWars benchmark used). The reload keeps running on the daemon's own event loop either way; add_root/remove_root below already fire it the same way for exactly this reason. """ reload_fn = state.get("reload_fn") if not reload_fn: raise OpError("Reload not available", status=503) spawn(reload_fn()) return {"status": "reloading"}