feat: add support for executable pages in Mesh Server (Linux)

This commit is contained in:
Ivan 2026-08-12 15:23:17 -05:00
parent 3786a67c76
commit 11d0abd6fd
No known key found for this signature in database
15 changed files with 486 additions and 40 deletions

View file

@ -40,6 +40,18 @@ Typical workflow:
3. Start the node and announce it on the mesh.
4. Share your destination hash so others can open `/page/index.mu` on your node.
### Executable (dynamic) pages
On Linux and macOS you can opt in per node to **executable pages**. When enabled:
- Non-executable pages are served as static files.
- Pages marked executable (via the Mesh Server UI or `chmod +x`) run as scripts using their shebang.
- Request `field_*` and `var_*` values are passed as environment variables.
- `link_id` and `remote_identity` are supplied when available.
- Script stdout is returned as the page body. Failures return a controlled error page.
Executable pages are disabled on Windows (static serving only). Editing a page in Mesh Server always shows the file source, never the script output.
API endpoints under `/api/v1/page-nodes/` manage CRUD operations, start and stop, and file listings.
Pages are served at `/page/<name>` and files at `/file/<name>` on the node destination.

Binary file not shown.

View file

@ -10216,7 +10216,14 @@ class ReticulumMeshChat:
self.config.nomad_default_page_path.get() or "/page/index.mu",
)
def _try_serve_local_page_node(self, destination_hash, page_path):
def _try_serve_local_page_node(
self,
destination_hash,
page_path,
request_data=None,
link_id=None,
remote_identity=None,
):
"""Serve a page from disk when the hash matches a local page node.
Returns the page content string, or None.
@ -10228,10 +10235,15 @@ class ReticulumMeshChat:
page_name = page_path.lstrip("/")
page_name = page_name.removeprefix("page/")
page_name = os.path.basename(page_name)
content = node.get_page_content(page_name)
if content is not None:
node._stats["pages_served"] += 1
return content
raw = node.serve_page_content(
page_name,
data=request_data,
link_id=link_id,
remote_identity=remote_identity,
)
if raw is None:
return None
return raw.decode("utf-8", errors="replace")
return None
def _try_serve_local_page_node_file(self, destination_hash, file_path):

View file

@ -148,10 +148,12 @@ def register_page_nodes_routes(routes, app):
return web.json_response({"message": "Name is required"}, status=400)
announce_enabled = bool(data.get("announce_enabled", True))
announce_interval_seconds = data.get("announce_interval_seconds")
executable_pages_enabled = bool(data.get("executable_pages_enabled", False))
node = app.page_node_manager.create_node(
name,
announce_enabled=announce_enabled,
announce_interval_seconds=announce_interval_seconds,
executable_pages_enabled=executable_pages_enabled,
)
return web.json_response(node.get_status())
@ -240,12 +242,25 @@ def register_page_nodes_routes(routes, app):
if "announce_interval_seconds" in data
else None
)
executable_pages_enabled = (
data.get("executable_pages_enabled")
if "executable_pages_enabled" in data
else None
)
try:
node = app.page_node_manager.get_node(node_id)
if node is None:
return web.json_response({"message": "Node not found"}, status=404)
node = app.page_node_manager.set_announce_settings(
node_id,
announce_enabled=announce_enabled,
announce_interval_seconds=announce_interval_seconds,
)
if executable_pages_enabled is not None:
node = app.page_node_manager.set_executable_pages_enabled(
node_id,
bool(executable_pages_enabled),
)
return web.json_response(node.get_status())
except KeyError:
return web.json_response({"message": "Node not found"}, status=404)
@ -273,15 +288,21 @@ def register_page_nodes_routes(routes, app):
)
name = data.get("name", "")
content = data.get("content", "")
executable = data.get("executable") if "executable" in data else None
if not name:
return web.json_response(
{"message": "Page name is required"},
status=400,
)
try:
saved_name = node.add_page(name, content)
saved_name = node.add_page(name, content, executable=executable)
except ValueError as e:
return web.json_response({"message": str(e)}, status=400)
except OSError as e:
return web.json_response(
{"message": f"Failed to set executable flag: {e}"},
status=400,
)
except OSError as e:
return web.json_response(
{"message": f"Failed to write page: {e}"},
@ -292,7 +313,13 @@ def register_page_nodes_routes(routes, app):
{"message": f"Failed to save page: {e}"},
status=500,
)
return web.json_response({"name": saved_name, "message": "Page saved"})
return web.json_response(
{
"name": saved_name,
"executable": node.is_page_executable(saved_name),
"message": "Page saved",
},
)
@routes.get("/api/v1/page-nodes/{node_id}/pages/{page_name}")
async def page_nodes_get_page(request):
@ -304,7 +331,13 @@ def register_page_nodes_routes(routes, app):
content = node.get_page_content(page_name)
if content is None:
return web.json_response({"message": "Page not found"}, status=404)
return web.json_response({"name": page_name, "content": content})
return web.json_response(
{
"name": page_name,
"content": content,
"executable": node.is_page_executable(page_name),
},
)
@routes.delete("/api/v1/page-nodes/{node_id}/pages/{page_name}")
async def page_nodes_delete_page(request):

View file

@ -531,6 +531,7 @@ async def handle_nomadnet_page_download(app, client, data):
local_page = app._try_serve_local_page_node(
destination_hash,
page_path_to_download,
request_data=combined_data,
)
if local_page is not None:
app.archive_page(destination_hash.hex(), page_path, local_page)

View file

@ -17,6 +17,8 @@ Supported page filename extensions are .mu, .md, .txt, and .html.
import contextlib
import json
import os
import stat
import subprocess
import threading
import time
@ -31,6 +33,9 @@ ALLOWED_PAGE_EXTENSIONS = frozenset({".mu", ".md", ".txt", ".html"})
DEFAULT_ANNOUNCE_INTERVAL_SECONDS = 900
MIN_ANNOUNCE_INTERVAL_SECONDS = 60
MAX_ANNOUNCE_INTERVAL_SECONDS = 86400
EXECUTABLE_PAGE_TIMEOUT_SECONDS = 15
PAGE_GENERATION_FAILED_MICRON = ">Page Generation Failed\n\nThe page could not be generated.\n"
def normalize_announce_interval_seconds(
@ -83,6 +88,47 @@ def _safe_mesh_file_basename(name: str) -> str:
return base
def _is_windows_platform() -> bool:
if os.name == "nt":
return True
try:
return bool(RNS.vendor.platformutils.is_windows())
except Exception:
return False
def _page_generation_error_bytes(reason: str) -> bytes:
text = PAGE_GENERATION_FAILED_MICRON
if reason:
text += f"\n{reason}\n"
return text.encode("utf-8")
def _build_executable_page_env(
data,
link_id=None,
remote_identity=None,
) -> dict[str, str]:
env_map: dict[str, str] = {}
if "PATH" in os.environ:
env_map["PATH"] = os.environ["PATH"]
if link_id is not None:
with contextlib.suppress(Exception):
env_map["link_id"] = RNS.hexrep(link_id, delimit=False)
if remote_identity is not None:
with contextlib.suppress(Exception):
remote_hash = getattr(remote_identity, "hash", None)
if remote_hash is not None:
env_map["remote_identity"] = RNS.hexrep(remote_hash, delimit=False)
if data is not None and isinstance(data, dict):
for key, value in data.items():
if isinstance(key, str) and (
key.startswith("field_") or key.startswith("var_")
):
env_map[key] = str(value)
return env_map
def _reject_name_component_too_long(parent_dir: str, component: str) -> None:
"""Raise ValueError if basename exceeds this directory's filename length limit."""
try:
@ -108,6 +154,7 @@ class PageNode:
identity_path=None,
announce_enabled=True,
announce_interval_seconds=DEFAULT_ANNOUNCE_INTERVAL_SECONDS,
executable_pages_enabled=False,
on_announce=None,
):
self.node_id = node_id
@ -131,6 +178,7 @@ class PageNode:
self.announce_interval_seconds = normalize_announce_interval_seconds(
announce_interval_seconds,
)
self.executable_pages_enabled = bool(executable_pages_enabled)
self.last_announced_at = None
self.on_announce = on_announce
self._announce_timer = None
@ -227,6 +275,10 @@ class PageNode:
)
self._sync_announce_timer()
def set_executable_pages_enabled(self, enabled):
"""Enable or disable dynamic executable page serving for this node."""
self.executable_pages_enabled = bool(enabled)
def teardown(self):
"""Deregister handlers and clean up."""
self.running = False
@ -363,22 +415,108 @@ class PageNode:
self.destination.deregister_request_handler(rpath)
self._registered_file_paths.discard(rpath)
def _resolve_page_path(self, name):
try:
safe_name = normalize_page_filename(name)
_reject_name_component_too_long(self.pages_dir, safe_name)
except ValueError:
return None
page_path = os.path.join(self.pages_dir, safe_name)
if not os.path.isfile(page_path):
return None
return page_path
def is_page_executable(self, name):
"""Return whether the page file has the execute permission bit set."""
if _is_windows_platform():
return False
page_path = self._resolve_page_path(name)
if page_path is None:
return False
return os.access(page_path, os.X_OK)
def set_page_executable(self, name, enabled):
"""Set or clear the execute permission bit on a page file."""
if _is_windows_platform():
raise OSError("executable pages are not supported on Windows")
page_path = self._resolve_page_path(name)
if page_path is None:
raise ValueError("page not found")
mode = os.stat(page_path).st_mode
if enabled:
os.chmod(page_path, mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
else:
os.chmod(
page_path,
mode & ~(stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH),
)
def _should_execute_page(self, page_path):
if not self.executable_pages_enabled:
return False
if _is_windows_platform():
return False
return os.access(page_path, os.X_OK)
def _read_static_page_bytes(self, page_path):
with open(page_path, "rb") as f:
return f.read()
def _execute_page_bytes(self, page_path, data=None, link_id=None, remote_identity=None):
env_map = _build_executable_page_env(data, link_id, remote_identity)
try:
generated = subprocess.run(
[page_path],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
env=env_map,
timeout=EXECUTABLE_PAGE_TIMEOUT_SECONDS,
)
return generated.stdout
except subprocess.TimeoutExpired:
return _page_generation_error_bytes("The page script timed out.")
except OSError as e:
return _page_generation_error_bytes(str(e))
except Exception as e:
return _page_generation_error_bytes(str(e))
def serve_page_content(
self,
name,
data=None,
link_id=None,
remote_identity=None,
):
"""Serve a page as static bytes or by executing it when enabled and executable."""
page_path = self._resolve_page_path(name)
if page_path is None:
return None
try:
if self._should_execute_page(page_path):
content = self._execute_page_bytes(
page_path,
data=data,
link_id=link_id,
remote_identity=remote_identity,
)
else:
content = self._read_static_page_bytes(page_path)
self._stats["pages_served"] += 1
return content
except Exception:
return None
def _make_page_responder(self, page_name):
"""Return a closure that serves a specific page."""
def responder(path, data, request_id, link_id, remote_identity, requested_at):
self._note_remote_identity(remote_identity)
safe_name = os.path.basename(page_name)
page_path = os.path.join(self.pages_dir, safe_name)
if not os.path.isfile(page_path):
return None
try:
with open(page_path, "rb") as f:
content = f.read()
self._stats["pages_served"] += 1
return content
except Exception:
return None
return self.serve_page_content(
page_name,
data=data,
link_id=link_id,
remote_identity=remote_identity,
)
return responder
@ -401,7 +539,7 @@ class PageNode:
return responder
def add_page(self, name, content):
def add_page(self, name, content, executable=None):
"""Write a page file and register its request handler."""
os.makedirs(self.pages_dir, exist_ok=True)
name = normalize_page_filename(name)
@ -411,6 +549,8 @@ class PageNode:
content = content.encode("utf-8")
with open(page_path, "wb") as f:
f.write(content)
if executable is not None:
self.set_page_executable(name, bool(executable))
if self.running:
try:
self._register_page_handler(name)
@ -435,15 +575,21 @@ class PageNode:
return False
def list_pages(self):
"""Return a list of page names."""
"""Return page metadata dicts with name and executable state."""
if not os.path.isdir(self.pages_dir):
return []
return sorted(
f
for f in os.listdir(self.pages_dir)
if os.path.isfile(os.path.join(self.pages_dir, f))
and is_allowed_page_filename(f)
)
pages = []
for fname in sorted(os.listdir(self.pages_dir)):
fpath = os.path.join(self.pages_dir, fname)
if not os.path.isfile(fpath) or not is_allowed_page_filename(fname):
continue
pages.append(
{
"name": fname,
"executable": self.is_page_executable(fname),
},
)
return pages
def get_page_content(self, name):
"""Read and return a page's content."""
@ -524,6 +670,7 @@ class PageNode:
"stats": dict(self._stats),
"announce_enabled": self.announce_enabled,
"announce_interval_seconds": self.announce_interval_seconds,
"executable_pages_enabled": self.executable_pages_enabled,
"last_announced_at": self.last_announced_at,
}
@ -534,6 +681,7 @@ class PageNode:
"name": self.name,
"announce_enabled": self.announce_enabled,
"announce_interval_seconds": self.announce_interval_seconds,
"executable_pages_enabled": self.executable_pages_enabled,
}
config_path = os.path.join(self.base_dir, "config.json")
with open(config_path, "w") as f:

View file

@ -48,6 +48,7 @@ class PageNodeManager:
announce_interval_seconds=normalize_announce_interval_seconds(
config.get("announce_interval_seconds"),
),
executable_pages_enabled=config.get("executable_pages_enabled", False),
on_announce=self.on_announce,
)
self.nodes[node_id] = node
@ -58,6 +59,7 @@ class PageNodeManager:
node_id=None,
announce_enabled=True,
announce_interval_seconds=None,
executable_pages_enabled=False,
):
"""Create a new page node, persist its config, and return it."""
if node_id is None:
@ -74,6 +76,7 @@ class PageNodeManager:
announce_interval_seconds=normalize_announce_interval_seconds(
announce_interval_seconds,
),
executable_pages_enabled=executable_pages_enabled,
on_announce=self.on_announce,
)
node.save_config()
@ -158,6 +161,15 @@ class PageNodeManager:
node.save_config()
return node
def set_executable_pages_enabled(self, node_id, enabled):
"""Enable or disable executable page serving for a node."""
node = self.nodes.get(node_id)
if node is None:
raise KeyError(f"Node {node_id} not found")
node.set_executable_pages_enabled(enabled)
node.save_config()
return node
def rename_node(self, node_id, new_name):
"""Rename a node and persist the change."""
node = self.nodes.get(node_id)

View file

@ -185,6 +185,10 @@ import ToolsPageHeader from "../tools/ToolsPageHeader.vue";
const NOMAD_DESTINATION_HASH = /^[a-fA-F0-9]{32}$/;
const PAGE_EXTENSIONS = [".mu", ".html", ".md", ".txt"];
function pageNamesFromList(pages) {
return (pages || []).map((entry) => (typeof entry === "string" ? entry : entry?.name)).filter(Boolean);
}
export default {
name: "MicronEditorPage",
components: {
@ -1052,7 +1056,8 @@ ${b}=
return /^\d+$/.test(trimmed.slice(numberedPrefix.length));
},
async resolvePublishPageBase(tab, existingPages, serverName) {
const hasIndex = existingPages.includes("index.mu");
const pageNames = pageNamesFromList(existingPages);
const hasIndex = pageNames.includes("index.mu");
if (!hasIndex) {
return "index";
}
@ -1127,7 +1132,8 @@ ${b}=
});
const savedName = response.data?.name;
if (savedName) {
existingPages = [...new Set([...existingPages, savedName])];
const pageNames = pageNamesFromList(existingPages);
existingPages = [...new Set([...pageNames, savedName])];
}
published++;
} catch {

View file

@ -205,6 +205,16 @@
<div class="text-xs font-bold uppercase tracking-wider text-gray-500 dark:text-gray-400">
{{ $t("tools.mesh_server.announce_settings") }}
</div>
<div class="flex items-center justify-between gap-3">
<Toggle
id="mesh-server-executable-pages-enabled"
v-model="announceSettingsForm.executable_pages_enabled"
:label="$t('tools.mesh_server.executable_pages_enabled_label')"
/>
</div>
<div class="text-xs text-gray-500 dark:text-gray-400">
{{ $t("tools.mesh_server.executable_pages_warning") }}
</div>
<div class="flex items-center justify-between gap-3">
<Toggle
id="mesh-server-announce-enabled"
@ -299,20 +309,26 @@
<div
v-for="page in selectedNode.pages"
:key="page"
:key="page.name"
class="flex items-center justify-between p-3 rounded-lg bg-slate-50 dark:bg-zinc-800/50 border border-gray-200 dark:border-zinc-700"
>
<div class="flex items-center gap-2">
<MaterialDesignIcon icon-name="file-document-outline" class="w-4 h-4 text-teal-500" />
<span class="text-sm font-mono text-gray-900 dark:text-white">{{ page }}</span>
<span class="text-sm font-mono text-gray-900 dark:text-white">{{ page.name }}</span>
<span
v-if="page.executable"
class="text-[10px] uppercase tracking-wide px-1.5 py-0.5 rounded bg-violet-100 dark:bg-violet-900/40 text-violet-700 dark:text-violet-300"
>
{{ $t("tools.mesh_server.executable_badge") }}
</span>
</div>
<div class="flex items-center gap-2">
<button class="secondary-chip py-0.5! px-2! text-xs!" @click="editPage(page)">
<button class="secondary-chip py-0.5! px-2! text-xs!" @click="editPage(page.name)">
{{ $t("common.edit") }}
</button>
<button
class="secondary-chip py-0.5! px-2! text-xs! text-red-500!"
@click="deletePage(page)"
@click="deletePage(page.name)"
>
<MaterialDesignIcon icon-name="delete" class="w-3 h-3" />
</button>
@ -338,6 +354,13 @@
</button>
</div>
</div>
<div class="flex items-center gap-2">
<Toggle
id="mesh-server-page-executable"
v-model="editingPageExecutable"
:label="$t('tools.mesh_server.page_executable_label')"
/>
</div>
<textarea
v-model="editingPageContent"
class="w-full h-64 bg-white dark:bg-zinc-900 text-gray-900 dark:text-white p-3 font-mono text-sm rounded-lg border border-gray-200 dark:border-zinc-700 resize-y focus:outline-hidden focus:ring-2 focus:ring-blue-500/50"
@ -493,9 +516,11 @@ export default {
newPageName: "",
editingPage: null,
editingPageContent: "",
editingPageExecutable: false,
announceSettingsForm: {
announce_enabled: true,
announce_interval_seconds: DEFAULT_ANNOUNCE_INTERVAL_SECONDS,
executable_pages_enabled: false,
},
};
},
@ -555,6 +580,7 @@ export default {
this.announceSettingsForm = {
announce_enabled: node.announce_enabled !== false,
announce_interval_seconds: resolveAnnounceIntervalSeconds(node.announce_interval_seconds),
executable_pages_enabled: node.executable_pages_enabled === true,
};
},
async createNode() {
@ -621,6 +647,7 @@ export default {
{
announce_enabled: this.announceSettingsForm.announce_enabled,
announce_interval_seconds: this.announceSettingsForm.announce_interval_seconds,
executable_pages_enabled: this.announceSettingsForm.executable_pages_enabled,
},
);
this.selectedNode = response.data;
@ -681,6 +708,7 @@ export default {
}
this.editingPage = pageName;
this.editingPageContent = body?.content ?? "";
this.editingPageExecutable = body?.executable === true;
} catch {
ToastUtils.error(this.$t("tools.mesh_server.failed_page_load"));
}
@ -691,9 +719,11 @@ export default {
await window.api.post(`/api/v1/page-nodes/${this.selectedNode.node_id}/pages`, {
name: this.editingPage,
content: this.editingPageContent,
executable: this.editingPageExecutable,
});
this.editingPage = null;
this.editingPageContent = "";
this.editingPageExecutable = false;
ToastUtils.success(this.$t("tools.mesh_server.page_saved"));
await this.loadNodes();
} catch {

View file

@ -2955,6 +2955,10 @@
"never_announced": "Never announced",
"announce_settings_saved": "Announce settings saved",
"announce_settings_failed": "Failed to save announce settings",
"executable_pages_enabled_label": "Enable executable pages",
"executable_pages_warning": "When enabled, pages marked executable run as scripts on this machine. Unix only. Off by default.",
"page_executable_label": "Executable page",
"executable_badge": "Executable",
"failed_load": "Failed to load nodes",
"created": "Server created",
"failed_create": "Failed to create server",

View file

@ -405,6 +405,47 @@ def test_landlock_rnsh_module_spawn_when_installed(tmp_path):
assert_probe_ok(result)
@requires_landlock_integration
def test_landlock_executable_page_script_spawn(tmp_path):
storage = tmp_path / "storage"
storage.mkdir()
result = run_python_under_landlock(
"""
import os
import stat
import subprocess
import sys
pages_dir = os.path.join(storage, "pages")
os.makedirs(pages_dir, exist_ok=True)
script_path = os.path.join(pages_dir, "dyn.mu")
with open(script_path, "w", encoding="utf-8") as handle:
handle.write("#!/usr/bin/env python3\\nprint('landlock-page-ok', flush=True)\\n")
os.chmod(
script_path,
os.stat(script_path).st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH,
)
env = {}
if "PATH" in os.environ:
env["PATH"] = os.environ["PATH"]
child = subprocess.run(
[script_path],
capture_output=True,
text=True,
timeout=15,
env=env,
)
if "landlock-page-ok" not in (child.stdout or ""):
print("BAD_OUT", child.stdout, child.stderr, child.returncode)
sys.exit(4)
print("OK")
sys.exit(0)
""",
storage=storage,
)
assert_probe_ok(result)
@pytest.mark.parametrize(
("path_name", "collector"),
[

View file

@ -203,6 +203,7 @@ def test_try_serve_local_helpers_strip_traversal():
node.pages_dir = "/tmp/mesh_pages"
node._stats = {"files_served": 0, "pages_served": 0}
node.get_page_content = MagicMock(return_value="page ok")
node.serve_page_content = MagicMock(return_value=b"page ok")
app.page_node_manager = MagicMock()
app.page_node_manager.nodes = {"1": node}
@ -218,8 +219,10 @@ def test_try_serve_local_helpers_strip_traversal():
app,
dh,
"/page/../../../x.mu",
request_data={"var_x": "1"},
)
node.get_page_content.assert_called_once()
called = node.get_page_content.call_args[0][0]
assert called == "x.mu"
node.serve_page_content.assert_called_once()
called = node.serve_page_content.call_args
assert called.args[0] == "x.mu"
assert called.kwargs.get("data") == {"var_x": "1"}
assert page_out == "page ok"

View file

@ -34,6 +34,9 @@ def mock_rns():
mock.Destination.SINGLE = 0
mock.Destination.ALLOW_ALL = 0xFF
mock.Transport = MagicMock()
mock.vendor = MagicMock()
mock.vendor.platformutils = MagicMock()
mock.vendor.platformutils.is_windows.return_value = False
yield mock, mock_identity, mock_destination
@ -458,7 +461,10 @@ class TestPageNodePages:
node.add_page("b.mu", "b")
node.add_page("a.mu", "a")
pages = node.list_pages()
assert pages == ["a.mu", "b.mu"]
assert pages == [
{"name": "a.mu", "executable": False},
{"name": "b.mu", "executable": False},
]
def test_get_page_content(self, node_dir, mock_rns):
node = _make_node(node_dir, mock_rns)
@ -618,6 +624,16 @@ class TestPageNodeConfig:
assert config["announce_enabled"] is True
assert config["announce_interval_seconds"] == DEFAULT_ANNOUNCE_INTERVAL_SECONDS
def test_save_config_persists_executable_pages_enabled(self, node_dir, mock_rns):
from meshchatx.src.backend.page_node import PageNode
node = _make_node(node_dir, mock_rns)
node.set_executable_pages_enabled(True)
node.save_config()
config = PageNode.load_config(node_dir)
assert config["executable_pages_enabled"] is True
class TestPageNodeStatus:
def test_get_status(self, node_dir, mock_rns):
@ -629,7 +645,7 @@ class TestPageNodeStatus:
assert status["name"] == "Test Node"
assert status["running"] is True
assert status["destination_hash"] is not None
assert "index.mu" in status["pages"]
assert "index.mu" in [p["name"] for p in status["pages"]]
assert isinstance(status["stats"], dict)
assert status["unique_connections"] == 0
assert status["uptime_seconds"] >= 0
@ -749,3 +765,92 @@ class TestPageNodeEdgeCases:
name = node.add_page("../../etc/passwd", "malicious")
assert name == "passwd.mu"
assert not os.path.exists(os.path.join(node_dir, "..", "etc"))
class TestPageNodeExecutablePages:
def test_executable_disabled_serves_static_even_when_chmod_x(self, node_dir, mock_rns):
node = _make_node(node_dir, mock_rns)
node.setup()
script = "#!/usr/bin/env python3\nprint('dynamic')\n"
node.add_page("dyn.mu", script, executable=True)
assert node.serve_page_content("dyn.mu") == script.encode("utf-8")
def test_executable_enabled_runs_script(self, node_dir, mock_rns):
node = _make_node(node_dir, mock_rns)
node.setup()
node.set_executable_pages_enabled(True)
script = "#!/usr/bin/env python3\nprint('dynamic-output')\n"
node.add_page("dyn.mu", script, executable=True)
result = node.serve_page_content("dyn.mu")
assert result == b"dynamic-output\n"
def test_executable_passes_env_vars(self, node_dir, mock_rns):
node = _make_node(node_dir, mock_rns)
node.setup()
node.set_executable_pages_enabled(True)
script = (
"#!/usr/bin/env python3\n"
"import os\n"
"print(os.environ.get('var_name', ''), end='')\n"
"print('|', os.environ.get('field_form', ''), end='')\n"
)
node.add_page("env.mu", script, executable=True)
result = node.serve_page_content(
"env.mu",
data={"var_name": "alice", "field_form": "submit", "ignored": "x"},
link_id=b"\x01" * 16,
remote_identity=MagicMock(hash=b"\x02" * 16),
)
assert b"alice" in result
assert b"submit" in result
def test_executable_timeout_returns_error_page(self, node_dir, mock_rns):
node = _make_node(node_dir, mock_rns)
node.setup()
node.set_executable_pages_enabled(True)
script = "#!/usr/bin/env python3\nimport time\ntime.sleep(30)\n"
node.add_page("slow.mu", script, executable=True)
result = node.serve_page_content("slow.mu")
assert b">Page Generation Failed" in result
assert b"timed out" in result.lower()
def test_get_page_content_never_executes(self, node_dir, mock_rns):
node = _make_node(node_dir, mock_rns)
node.setup()
node.set_executable_pages_enabled(True)
script = "#!/usr/bin/env python3\nprint('dynamic')\n"
node.add_page("dyn.mu", script, executable=True)
assert node.get_page_content("dyn.mu") == script
def test_windows_platform_skips_execution(self, node_dir, mock_rns):
node = _make_node(node_dir, mock_rns)
node.setup()
node.set_executable_pages_enabled(True)
script = "#!/usr/bin/env python3\nprint('dynamic')\n"
node.add_page("dyn.mu", script, executable=True)
with patch(
"meshchatx.src.backend.page_node._is_windows_platform",
return_value=True,
):
result = node.serve_page_content("dyn.mu")
assert result == script.encode("utf-8")
def test_set_page_executable_round_trip(self, node_dir, mock_rns):
node = _make_node(node_dir, mock_rns)
node.setup()
node.add_page("toggle.mu", "static")
assert node.is_page_executable("toggle.mu") is False
node.set_page_executable("toggle.mu", True)
assert node.is_page_executable("toggle.mu") is True
node.set_page_executable("toggle.mu", False)
assert node.is_page_executable("toggle.mu") is False
def test_page_responder_uses_serve_page_content(self, node_dir, mock_rns):
node = _make_node(node_dir, mock_rns)
node.setup()
node.set_executable_pages_enabled(True)
script = "#!/usr/bin/env python3\nprint('from-responder')\n"
node.add_page("dyn.mu", script, executable=True)
responder = node._make_page_responder("dyn.mu")
result = responder("/page/dyn.mu", {"var_x": "1"}, "req1", b"\xab" * 16, None, 0)
assert result == b"from-responder\n"

View file

@ -32,6 +32,9 @@ def mock_rns():
mock.Destination.SINGLE = 0
mock.Destination.ALLOW_ALL = 0xFF
mock.Transport = MagicMock()
mock.vendor = MagicMock()
mock.vendor.platformutils = MagicMock()
mock.vendor.platformutils.is_windows.return_value = False
yield mock, mock_identity, mock_destination
@ -55,7 +58,7 @@ class TestPageNodeManagerCreate:
node = mgr.create_node("Publish Target")
saved_name = node.add_page("index", "hello mesh")
assert saved_name == "index.mu"
assert node.list_pages() == ["index.mu"]
assert node.list_pages() == [{"name": "index.mu", "executable": False}]
assert node.running is False
def test_create_node_with_custom_id(self, storage_dir, mock_rns):
@ -181,6 +184,20 @@ class TestPageNodeManagerAnnounceSettings:
node = mgr.create_node("Defaults")
assert node.announce_enabled is True
def test_create_node_with_executable_pages_enabled(self, storage_dir, mock_rns):
mgr = _make_manager(storage_dir)
node = mgr.create_node("Dynamic", executable_pages_enabled=True)
assert node.executable_pages_enabled is True
def test_set_executable_pages_enabled_persists(self, storage_dir, mock_rns):
from meshchatx.src.backend.page_node import PageNode
mgr = _make_manager(storage_dir)
node = mgr.create_node("Dynamic")
mgr.set_executable_pages_enabled(node.node_id, True)
config = PageNode.load_config(node.base_dir)
assert config["executable_pages_enabled"] is True
def test_set_announce_settings_updates_node(self, storage_dir, mock_rns):
mgr = _make_manager(storage_dir)
node = mgr.create_node("Settable")

View file

@ -27,11 +27,12 @@ function makeNode(overrides = {}) {
active_links: 0,
unique_connections: 0,
uptime_seconds: 60,
pages: ["index.mu"],
pages: [{ name: "index.mu", executable: false }],
files: [],
stats: { pages_served: 0, files_served: 0, links_established: 0 },
announce_enabled: true,
announce_interval_seconds: 900,
executable_pages_enabled: false,
last_announced_at: null,
...overrides,
};
@ -96,6 +97,7 @@ describe("PageNodesPage.vue", () => {
expect(wrapper.vm.announceSettingsForm).toEqual({
announce_enabled: true,
announce_interval_seconds: 900,
executable_pages_enabled: false,
});
});
@ -109,6 +111,7 @@ describe("PageNodesPage.vue", () => {
expect(wrapper.vm.announceSettingsForm).toEqual({
announce_enabled: false,
announce_interval_seconds: 300,
executable_pages_enabled: false,
});
});
@ -145,12 +148,14 @@ describe("PageNodesPage.vue", () => {
wrapper.vm.announceSettingsForm = {
announce_enabled: false,
announce_interval_seconds: 300,
executable_pages_enabled: true,
};
await wrapper.vm.saveAnnounceSettings();
expect(axiosMock.patch).toHaveBeenCalledWith(`/api/v1/page-nodes/${NODE_ID}/announce-settings`, {
announce_enabled: false,
announce_interval_seconds: 300,
executable_pages_enabled: true,
});
});
@ -185,4 +190,21 @@ describe("PageNodesPage.vue", () => {
expect(axiosMock.post).toHaveBeenCalledWith("/api/v1/page-nodes", { name: "Fresh Node" });
});
it("saves page executable flag via POST API", async () => {
const wrapper = mountPage();
await vi.waitFor(() => expect(wrapper.vm.nodes.length).toBe(1));
wrapper.vm.selectNode(wrapper.vm.nodes[0]);
wrapper.vm.editingPage = "index.mu";
wrapper.vm.editingPageContent = "#!/usr/bin/env python3\nprint('x')\n";
wrapper.vm.editingPageExecutable = true;
await wrapper.vm.savePage();
expect(axiosMock.post).toHaveBeenCalledWith(`/api/v1/page-nodes/${NODE_ID}/pages`, {
name: "index.mu",
content: "#!/usr/bin/env python3\nprint('x')\n",
executable: true,
});
});
});