feat(rnsh): update rnsh command resolution to prefer module launcher and add error handling for non-executable paths

This commit is contained in:
Ivan 2026-06-24 15:52:47 -05:00
parent 306530110e
commit daf255deaf
No known key found for this signature in database
3 changed files with 88 additions and 5 deletions

View file

@ -9,6 +9,7 @@ while the app is active, and persists session definitions plus output history.
from __future__ import annotations
import contextlib
import importlib.util
import json
import os
import re
@ -41,6 +42,8 @@ DEFAULT_TERMINAL_COLS = 120
# "rnsh listening for commands on <a1b2c3...>" or "Listening on : <...>".
_LISTEN_ADDRESS_RE = re.compile(r"<([0-9a-fA-F]{16,})>")
_RNSH_MODULE = "RNS.Utilities.rnsh.rnsh"
class RNSHSession:
"""Runtime state for a single rnsh session."""
@ -229,12 +232,35 @@ class RNSHSession:
if match:
self.listen_address = match.group(1).lower()
def _build_command(self):
@staticmethod
def _rnsh_module_available():
try:
return importlib.util.find_spec(_RNSH_MODULE) is not None
except (ImportError, ModuleNotFoundError, ValueError, AttributeError):
return False
@staticmethod
def _resolve_rnsh_launcher():
"""Return argv prefix for launching rnsh.
Prefer the Python module entry point so sessions work when the PATH
console-script wrapper is not executable (common with pip --user
installs) or when Landlock denies executing paths outside allowed
read roots (for example ``~/.local/bin/rnsh``).
"""
if RNSHSession._rnsh_module_available():
return [sys.executable, "-m", _RNSH_MODULE]
executable = shutil.which("rnsh")
if executable and os.access(executable, os.X_OK):
return [executable]
if executable:
command = [executable]
else:
command = [sys.executable, "-m", "RNS.Utilities.rnsh.rnsh"]
raise PermissionError(f"Permission denied: '{executable}'")
raise FileNotFoundError(
"rnsh is not available; install the rns package or ensure rnsh is on PATH",
)
def _build_command(self):
command = list(self._resolve_rnsh_launcher())
# Attach rnsh to the same Reticulum config directory (and therefore the
# same shared instance and rpc_key) as the MeshChatX app. Without this

View file

@ -189,6 +189,64 @@ def test_rnsh_resize_updates_geometry_without_process():
assert result["cols"] == 100
def test_rnsh_prefers_module_launcher_when_rns_installed(monkeypatch):
from meshchatx.src.backend import rnsh_manager as rnsh_mod
monkeypatch.setattr(rnsh_mod.RNSHSession, "_rnsh_module_available", lambda: True)
monkeypatch.setattr(
rnsh_mod.shutil, "which", lambda _name: "/home/user/.local/bin/rnsh"
)
manager = MagicMock()
session = rnsh_mod.RNSHSession(
manager,
"s1",
{"mode": "connect", "destination": "deadbeef"},
)
command = session._build_command()
assert command[:3] == [rnsh_mod.sys.executable, "-m", rnsh_mod._RNSH_MODULE]
assert command[-1] == "deadbeef"
def test_rnsh_falls_back_to_path_binary_when_module_missing(monkeypatch):
from meshchatx.src.backend import rnsh_manager as rnsh_mod
monkeypatch.setattr(rnsh_mod.RNSHSession, "_rnsh_module_available", lambda: False)
monkeypatch.setattr(rnsh_mod.shutil, "which", lambda _name: "/usr/bin/rnsh")
monkeypatch.setattr(rnsh_mod.os, "access", lambda _path, _mode: True)
manager = MagicMock()
session = rnsh_mod.RNSHSession(
manager,
"s1",
{"mode": "connect", "destination": "deadbeef"},
)
command = session._build_command()
assert command[0] == "/usr/bin/rnsh"
assert command[-1] == "deadbeef"
def test_rnsh_non_executable_path_wrapper_raises_permission_error(monkeypatch):
from meshchatx.src.backend import rnsh_manager as rnsh_mod
monkeypatch.setattr(rnsh_mod.RNSHSession, "_rnsh_module_available", lambda: False)
monkeypatch.setattr(
rnsh_mod.shutil,
"which",
lambda _name: "/home/user/.local/bin/rnsh",
)
monkeypatch.setattr(rnsh_mod.os, "access", lambda _path, _mode: False)
manager = MagicMock()
session = rnsh_mod.RNSHSession(
manager,
"s1",
{"mode": "connect", "destination": "deadbeef"},
)
with pytest.raises(PermissionError, match="Permission denied"):
session._build_command()
@pytest.mark.asyncio
async def test_rnsh_session_not_found_returns_404(mock_app):
mock_app.rnsh_manager = _DummyManager()

View file

@ -39,4 +39,3 @@ def test_telemeter_unpack_location_robustness():
assert Telemeter.unpack_location([b"lat", b"lon"]) is None
# Test with invalid types
assert Telemeter.unpack_location(["not_bytes"] * 7) is None