"""Platform-specific paths and tool resolution for meshbay-node.""" import asyncio import logging import os import shutil import subprocess import sys import threading import time from pathlib import Path log = logging.getLogger(__name__) # ── Console ────────────────────────────────────────────────────────────────── def force_utf8_stdio() -> None: """ Make stdout/stderr UTF-8. A Windows console is cp1252 by default, so any ``print()`` carrying a character outside it — the ``->`` arrows and em dashes the CLI help and messages are full of — raises UnicodeEncodeError and takes the command down with it. No effect where the streams are already UTF-8 or cannot be reconfigured. """ for stream in (sys.stdout, sys.stderr): try: stream.reconfigure(encoding="utf-8") except (AttributeError, ValueError, OSError): pass # ── Event loop ─────────────────────────────────────────────────────────────── def configure_event_loop() -> None: """ The daemon runs on Windows' default ProactorEventLoop: verified end to end (a live browser peer connecting, an index sync, a file download and an ffmpeg-transcoded video stream). aiortc only ever hangs on it in the *same-process loopback* the tests use, which the test suite handles on its own (repo-root conftest). Escape hatch, opt-in only: MESHBAY_NODE_EVENT_LOOP=selector switches to the SelectorEventLoop. That fixes aiortc-in-one-process but breaks ffmpeg (SelectorEventLoop cannot spawn subprocesses on Windows), so it is not the default and probably never should be. """ if sys.platform != "win32": return if os.environ.get("MESHBAY_NODE_EVENT_LOOP", "").lower() == "selector": asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) # ── Directories ────────────────────────────────────────────────────────────── def config_dir() -> Path: if sys.platform == "win32": return Path(os.environ.get("LOCALAPPDATA") or Path.home()) / "meshbay" return Path.home() / ".config" / "meshbay" def data_dir() -> Path: if sys.platform == "win32": return Path(os.environ.get("LOCALAPPDATA") or Path.home()) / "meshbay" / "data" return Path.home() / ".local" / "share" / "meshbay" def state_dir() -> Path: if sys.platform == "win32": return Path(os.environ.get("LOCALAPPDATA") or Path.home()) / "meshbay" / "state" return Path.home() / ".local" / "state" / "meshbay" def log_file() -> Path | None: """Where the daemon logs on Windows; None elsewhere, where journald has it. A daemon started by the service task or the Startup launcher has no console: its stderr goes nowhere, and a node that fails there fails in silence. """ if sys.platform == "win32": return state_dir() / "node.log" return None # ── Packaged defaults ──────────────────────────────────────────────────────── def packaged_default_env() -> Path | None: """ The `default.env` shipped with the package: build-time defaults, currently the shared read-only TMDB token. The daemon reads it in place, beneath /node.env, so it reaches every node however that node was set up -- `meshbay-node init` and the desktop client's onboarding alike -- and an operator's own node.env still wins. Frozen (PyInstaller/Windows): beside the executable, where build-node-runtime.ps1 puts it -- the same placement it uses for ffmpeg. Packaged (Linux): /opt/meshbay-node/share/default.env, from build-node.sh. None in a source checkout, where no package wrote one. """ candidates = [] if getattr(sys, "frozen", False): candidates.append(Path(sys.executable).parent / "default.env") candidates.append(Path("/opt/meshbay-node/share/default.env")) for path in candidates: try: if path.is_file(): return path except OSError: continue return None def _load_env_file(path: Path) -> int: try: text = path.read_text(encoding="utf-8") except (OSError, UnicodeDecodeError): return 0 count = 0 for line in text.splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue name, _, value = line.partition("=") name = name.strip() value = value.strip().strip('"').strip("'") if not name or name in os.environ: continue os.environ[name] = value count += 1 return count def load_node_env(source_dir: Path) -> int: """ Read /node.env, then the packaged default.env, into os.environ, returning how many names were set. systemd does the first through `EnvironmentFile=`, but the Windows autostart is a Startup-folder .vbs with no equivalent, so the daemon reads the file itself and both platforms behave the same. An existing environment variable always wins, and node.env wins over the packaged default -- an operator exporting a value, or systemd having already loaded the same file, overrides the packaged default rather than being overridden by it. The packaged default used to reach a node only as a copy made by `meshbay-node init`; a node onboarded by the desktop client never ran it and ran with no TMDB token at all. """ count = _load_env_file(source_dir / "node.env") packaged = packaged_default_env() if packaged is not None: count += _load_env_file(packaged) return count # ── File permissions ───────────────────────────────────────────────────────── def chmod_private(path: Path, *, mode: int = 0o600) -> None: """Set restrictive permissions on a file. No-op on Windows (NTFS ignores mode bits).""" if sys.platform != "win32": path.chmod(mode) # ── Media tools ────────────────────────────────────────────────────────────── _ffmpeg_path: str = "ffmpeg" _ffprobe_path: str = "ffprobe" def check_media_tools( ffmpeg: str = "ffmpeg", ffprobe: str = "ffprobe", ) -> None: """Resolve ffmpeg/ffprobe at daemon startup. Raises RuntimeError if not found.""" global _ffmpeg_path, _ffprobe_path resolved = shutil.which(ffmpeg) if not resolved: raise RuntimeError( f"{ffmpeg!r} not found in PATH. " "Install ffmpeg or set [node] ffmpeg_path in node.toml." ) _ffmpeg_path = resolved resolved = shutil.which(ffprobe) if not resolved: raise RuntimeError( f"{ffprobe!r} not found in PATH. " "Install ffmpeg or set [node] ffprobe_path in node.toml." ) _ffprobe_path = resolved def ffmpeg_cmd() -> str: return _ffmpeg_path def ffprobe_cmd() -> str: return _ffprobe_path # ── Autostart (Windows) ────────────────────────────────────────────────────── # # The Windows stand-in for the Linux `systemctl --user` unit. Task Scheduler # would be nicer (retry semantics), but a logon-triggered task needs elevation # to create — and this must work for an ordinary user with no admin rights. # So: a `.vbs` launcher in the per-user Startup folder. wscript runs it hidden # (Run(..., 0, ...)) at every sign-in; no console window, no admin, no # third-party dependency. See the Service mode section below for the # boot-capable, admin-once alternative built on top of Task Scheduler instead. def autostart_supported() -> bool: return sys.platform == "win32" def _startup_vbs() -> Path: base = os.environ.get("APPDATA") or str(Path.home() / "AppData" / "Roaming") return (Path(base) / "Microsoft" / "Windows" / "Start Menu" / "Programs" / "Startup" / "MeshBay Node.vbs") def _pid_alive(pid: int) -> bool: """Whether a process with this pid exists, in any session (tasklist lists session 0 too, where a service-mode daemon runs; opening it would not).""" r = subprocess.run(["tasklist", "/FI", f"PID eq {pid}", "/NH", "/FO", "CSV"], capture_output=True, text=True) return f'"{pid}"' in r.stdout def _node_exe() -> str | None: """The meshbay-node to launch: this very one when frozen, then the launcher next to the interpreter (a venv's Scripts/), then argv[0], then PATH. PATH was first, and a machine with two copies on it -- an MSIX and an NSIS install, an older one left behind -- had `autostart start` from one launch the other, a different version, found by testing a real install. """ if getattr(sys, "frozen", False): return sys.executable for cand in (Path(sys.executable).parent / "meshbay-node.exe", Path(sys.argv[0])): if cand.name.lower().startswith("meshbay-node") and cand.exists(): return str(cand.resolve()) return shutil.which("meshbay-node") def request_graceful_stop(data_dir: Path, ui_port: int, timeout: float = 15.0) -> bool: """Ask the running daemon to stop through its control API, and wait until its process is gone. True once it is; False when nothing answered or it did not exit in time -- the caller then forces it. The one stop that reaches a daemon in any session with no elevation, and runs its _shutdown() (WebRTC sessions closed, transcodes stopped). CTRL_BREAK did neither across sessions, and aimed at a process on another console it reached every process on the caller's own -- the CLI killed itself. """ import json import urllib.request try: token = (data_dir / "ui-token").read_text(encoding="utf-8").strip() except OSError: return False base = f"http://127.0.0.1:{ui_port}" try: with urllib.request.urlopen(f"{base}/api/status?t={token}", timeout=3) as r: pid = json.loads(r.read()).get("pid") req = urllib.request.Request(f"{base}/api/shutdown?t={token}", method="POST", data=b"") with urllib.request.urlopen(req, timeout=3): pass except (OSError, ValueError): return False deadline = time.monotonic() + timeout while time.monotonic() < deadline: if pid is None: try: urllib.request.urlopen(f"{base}/api/status?t={token}", timeout=1) except OSError: return True elif not _pid_alive(int(pid)): return True time.sleep(0.3) log.warning("the node did not exit within %.0fs of being asked to stop", timeout) return False def autostart_status() -> dict: """{'installed': bool, 'state': str}. 'state' is left empty — there is no Task Scheduler to ask 'is it running'; the Node page probes the daemon.""" if not autostart_supported(): return {"installed": False, "state": ""} return {"installed": _startup_vbs().exists(), "state": ""} def autostart_install(exe: str | None = None) -> None: """Write the Startup-folder launcher. Raises RuntimeError on failure.""" if not autostart_supported(): raise RuntimeError("autostart is Windows-only") exe = exe or _node_exe() if not exe: raise RuntimeError( "cannot locate the meshbay-node launcher — pass its path, or run " "this from where meshbay-node is on PATH") vbs = _startup_vbs() vbs.parent.mkdir(parents=True, exist_ok=True) # Chr(34) is a literal " — wraps the path so a space in it doesn't split the # command. 0 = hidden window, False = don't wait. (A Windows path cannot # itself contain ", so no further escaping is needed.) vbs.write_text( f'CreateObject("WScript.Shell").Run Chr(34) & "{exe}" & Chr(34), 0, False\n', encoding="utf-8", newline="\r\n") def autostart_remove() -> None: """Delete the Startup-folder launcher if present.""" if autostart_supported(): _startup_vbs().unlink(missing_ok=True) def autostart_run() -> None: """Start the daemon now, windowless. Raises RuntimeError if the launcher cannot be located.""" if not autostart_supported(): raise RuntimeError("autostart is Windows-only") exe = _node_exe() if not exe: raise RuntimeError("cannot locate the meshbay-node launcher") # CREATE_NO_WINDOW: a hidden console of its own, which still delivers # CTRL_LOGOFF/SHUTDOWN to install_console_close_handler, so signing out # stops the node properly. close_fds: nothing of the caller's is inherited # -- the desktop app starts its node through here for that reason, since a # child of Electron inherited Electron's sockets and held them after it quit. subprocess.Popen([exe], creationflags=0x00000200 | 0x08000000, close_fds=True) NODE_IMAGE = "meshbay-node.exe" def autostart_end() -> None: """ Force-stop a daemon running in this session: the fallback after request_graceful_stop(), which callers try first. It used to send CTRL_BREAK_EVENT to the pid autostart_run() recorded. That process has a console of its own (CREATE_NO_WINDOW), so it is not a group on the caller's console -- and GenerateConsoleCtrlEvent aimed at such a pid reaches every process on the caller's console instead: `autostart stop` and `restart-daemon` killed themselves, the latter before restarting anything (found by running both against a real install). `taskkill /F` is TerminateProcess: cannot be caught, cannot reach session 0. Never this process or its parent: the CLI is meshbay-node.exe too (and a venv's launcher is its parent), and `taskkill /IM meshbay-node.exe` killed the very command that ran it -- `autostart stop` and `restart-daemon` exited 1 in silence, the latter with the node down. `/T` because a venv's meshbay-node.exe is a launcher whose python.exe child is the daemon, and a frozen daemon's children are its ffmpeg transcodes. """ if not autostart_supported(): return argv = ["taskkill", "/F", "/T", "/IM", NODE_IMAGE] for pid in {os.getpid(), os.getppid()}: argv += ["/FI", f"PID ne {pid}"] subprocess.run(argv, capture_output=True) # ── Service mode (Windows, opt-in at install time) ─────────────────────────── # # The Startup-folder .vbs above only ever runs after *this* user signs in. A # real Windows Service would run before anyone signs in, but under # LocalSystem/NetworkService — accounts with no normal user profile, so # %LOCALAPPDATA%\meshbay\ (config, keystore, data) would not exist for it. # Relocating storage to make that work is real surgery (Phase 2, deliberately # not this). # # The middle ground: a Scheduled Task, created once with admin rights, that # runs *as this user* at system boot without needing them to sign in first. # That requires an S4U (Service For User) logon: no password stored anywhere, # and unlike LocalSystem it loads this account's own profile, so # config_dir()/data_dir() need no special-casing at all. The cost: S4U # carries no *network* credential (no reaching a domain share as this user), # which the node never needed anyway — everything it touches is local disk # plus outbound internet. # # Getting S4U out of raw `schtasks.exe /create` means inferring it from # whether `/rp` is present and what it holds -- undocumented, and it went # wrong twice on this exact machine's blank-password account (common on a # personal PC — confirmed via `net user`, "Password required: No"): # `/rp ""` routes through credential validation, which Windows' default # policy blocks for a blank password ("WARNING: When the run-as password is # empty..." then "ERROR: The user name or password is incorrect.", 2026-09-05 # repro); omitting `/rp` entirely does get past that, but registers # `Logon Mode: Interactive only` instead of S4U — confirmed live the same # day: the task never ran at boot, and manually running it while signed in # still failed (`Last Result: -2147024894`, no process ever launched). # `Register-ScheduledTask` from the `ScheduledTasks` PowerShell module takes # `-LogonType S4U` as a named, explicit value — no inference, no ambiguity — # so that is what actually creates the task, shelling out to `powershell.exe` # instead of `schtasks.exe` for this one call. Still unverified end-to-end # past "the syntax runs and denies access when not elevated" (2026-09-05) — # needs a real elevated install + reboot to confirm S4U registers as such and # the task actually launches the process. # # Creating the task needs admin (a boot-trigger touches system-wide scheduler # state, the same reason /sc onlogon did — see the autostart section above). # Querying, running and ending an *already-created* task, as the same user it # was registered for, does not — Task Scheduler grants the owner that much by # default, which is what lets the Node page drive it with no further prompts. TASK_NAME = "MeshBay Node" # the Scheduled Task's own name # Task Scheduler's defaults end a task after 72 hours, never start it on # battery and stop it when the power cable comes out — a node on a laptop was # down for any of the three. Kept identical to packaging/win/service.ps1. SERVICE_TASK_SETTINGS = ( "New-ScheduledTaskSettingsSet -ExecutionTimeLimit ([TimeSpan]::Zero) " "-AllowStartIfOnBatteries -DontStopIfGoingOnBatteries " "-MultipleInstances IgnoreNew -StartWhenAvailable" ) def service_supported() -> bool: return sys.platform == "win32" def _current_user() -> str: domain = os.environ.get("USERDOMAIN") or os.environ.get("COMPUTERNAME") or "." user = os.environ.get("USERNAME") or "" return f"{domain}\\{user}" if user else "" def _schtasks(*args: str) -> subprocess.CompletedProcess: return subprocess.run(["schtasks", *args], capture_output=True, text=True) def service_status() -> dict: """{'installed': bool, 'state': str}. 'state' is Task Scheduler's own word ('Ready', 'Running', 'Disabled', ...), '' when not installed.""" if not service_supported(): return {"installed": False, "state": ""} r = _schtasks("/query", "/tn", TASK_NAME, "/fo", "list") if r.returncode != 0: return {"installed": False, "state": ""} state = "" for line in r.stdout.splitlines(): if line.lower().startswith("status:"): state = line.split(":", 1)[1].strip() break return {"installed": True, "state": state} def service_state() -> str: """Task Scheduler's word for the task ('Running', 'Ready', ...), '' if absent.""" return service_status()["state"] def service_install(exe: str | None = None) -> None: """ Register the boot-time Scheduled Task with an S4U logon. Needs admin — raises RuntimeError with Register-ScheduledTask's own message on failure, which is "Access is denied." when not elevated. Removes the per-user Startup launcher first, if present: the two mechanisms are mutually exclusive by design (both installed would start the daemon twice, once at boot and again at sign-in), and this is a separate front door from the Node page's own startup-mode selector (which enforces the same thing on its side) -- the CLI (`meshbay-node service install`) must not be able to leave that invariant broken. """ if not service_supported(): raise RuntimeError("service mode is Windows-only") autostart_remove() exe = exe or _node_exe() if not exe: raise RuntimeError( "cannot locate the meshbay-node launcher — pass its path, or run " "this from where meshbay-node is on PATH") user = _current_user() if not user: raise RuntimeError("could not determine the current user (USERNAME unset)") # Passed via the environment, not interpolated into the -Command string, # so a path or username containing a quote or $ can't break the script. env = {**os.environ, "MESHBAY_SVC_EXE": exe, "MESHBAY_SVC_USER": user, "MESHBAY_SVC_TASK": TASK_NAME} ps_script = ( "$ErrorActionPreference = 'Stop'; " "$a = New-ScheduledTaskAction -Execute $env:MESHBAY_SVC_EXE; " "$t = New-ScheduledTaskTrigger -AtStartup; " "$p = New-ScheduledTaskPrincipal -UserId $env:MESHBAY_SVC_USER " "-LogonType S4U -RunLevel Limited; " f"$s = {SERVICE_TASK_SETTINGS}; " "Register-ScheduledTask -TaskName $env:MESHBAY_SVC_TASK " "-Action $a -Trigger $t -Principal $p -Settings $s -Force | Out-Null" ) r = subprocess.run( ["powershell", "-NoProfile", "-NonInteractive", "-Command", ps_script], capture_output=True, text=True, env=env) if r.returncode != 0: raise RuntimeError( f"Register-ScheduledTask failed: {r.stderr.strip() or r.stdout.strip()}") def service_remove() -> None: """Delete the Scheduled Task if present. Needs admin; silent otherwise (mirrors autostart_remove — nothing to report if it was never installed).""" if service_supported(): _schtasks("/end", "/tn", TASK_NAME) # deleting does not end the instance _schtasks("/delete", "/tn", TASK_NAME, "/f") def service_run() -> None: """Start the task now. No admin needed for an already-registered task.""" if service_supported(): _schtasks("/run", "/tn", TASK_NAME) def service_end() -> None: """Stop the running instance, if any. No admin needed.""" if service_supported(): _schtasks("/end", "/tn", TASK_NAME) # ── Console close / logoff / shutdown handler (Windows) ────────────────────── # # CPython's own console handler claims CTRL_C_EVENT and CTRL_BREAK_EVENT -- # delivered as SIGINT/SIGBREAK, handled in daemon.py's win32 signal block -- # but returns "not handled" for CTRL_CLOSE_EVENT, CTRL_LOGOFF_EVENT and # CTRL_SHUTDOWN_EVENT: there is no Python signal for any of the three. Without # a handler of our own, Windows just ends the process for these -- no # _shutdown(), no closed WebRTC sessions, no killed ffmpeg. Covers: closing # the console window of an interactively-run `meshbay-node run`, user logoff, # system shutdown. Does NOT cover `taskkill /F` -- TerminateProcess is # uncatchable on any OS, the same as SIGKILL; request_graceful_stop() is how # every Stop avoids relying on it. _CONSOLE_HANDLER_REFS: list = [] # ctypes callbacks must be kept referenced or they may be freed CTRL_CLOSE_EVENT = 2 CTRL_LOGOFF_EVENT = 5 CTRL_SHUTDOWN_EVENT = 6 def install_console_close_handler( loop: asyncio.AbstractEventLoop, stop_event: asyncio.Event, ) -> "threading.Event | None": """ Register the handler. Returns a threading.Event the caller must set once its own graceful shutdown has actually finished -- daemon.py does this right after `await self._shutdown()` -- or None off-Windows, or if registration itself failed (logged, not raised: losing this is a regression, refusing to start the daemon over it would not be). MSDN: for these three events the process is ended "after the process returns from the handler function, or after 5 seconds, whichever occurs first" -- so the handler, which Windows runs on a thread of its own and never the main one, blocks here instead of returning immediately, and nudges the asyncio loop the thread-safe way since it is not the loop's own thread. The wait is capped just under that ceiling so the process still exits by itself if cleanup runs long, rather than the OS treating an unresponsive handler as a hang. """ if sys.platform != "win32": return None import ctypes from ctypes import wintypes handled = {CTRL_CLOSE_EVENT, CTRL_LOGOFF_EVENT, CTRL_SHUTDOWN_EVENT} shutdown_done = threading.Event() handler_type = ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.DWORD) def _handler(ctrl_type: int) -> bool: if ctrl_type not in handled: return False # not ours -- let Python's own handler or the default action take it log.info("Console control event %d (close/logoff/shutdown) -- shutting down", ctrl_type) loop.call_soon_threadsafe(stop_event.set) shutdown_done.wait(4.5) return True handler_ref = handler_type(_handler) if not ctypes.windll.kernel32.SetConsoleCtrlHandler(handler_ref, True): log.warning("SetConsoleCtrlHandler failed (%s) -- closing the console window, " "logging off or shutting down will not run a clean shutdown; " "SIGINT/SIGTERM/SIGBREAK are unaffected", ctypes.WinError()) return None _CONSOLE_HANDLER_REFS.append(handler_ref) return shutdown_done