diff --git a/tests/backend/test_archives_api_robustness.py b/tests/backend/test_archives_api_robustness.py
new file mode 100644
index 0000000..38f28c5
--- /dev/null
+++ b/tests/backend/test_archives_api_robustness.py
@@ -0,0 +1,167 @@
+# SPDX-License-Identifier: 0BSD
+
+"""Robustness tests for REST handlers (archives, bundled docs, lxmf send) and related WS paths."""
+
+import json
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from hypothesis import HealthCheck, given, settings
+from hypothesis import strategies as st
+
+
+def _handler(app, method: str, path: str):
+ for route in app.get_routes():
+ if route.method == method and route.path == path:
+ return route.handler
+ return None
+
+
+@pytest.mark.asyncio
+async def test_get_archives_returns_json_for_edge_case_queries(mock_app):
+ handler = _handler(mock_app, "GET", "/api/v1/nomadnet/archives")
+ assert handler is not None
+
+ edge_queries = [
+ {},
+ {"q": "", "page": "1", "limit": "100"},
+ {"q": "%00%3Cscript%3E", "page": "not-a-number", "limit": "not-a-number"},
+ {"page": "-5", "limit": "0"},
+ {"limit": "99999"},
+ ]
+ for query in edge_queries:
+ request = MagicMock()
+ request.query = query
+ response = await handler(request)
+ assert response.status == 200
+ payload = json.loads(response.body)
+ assert isinstance(payload.get("archives"), list)
+ assert "pagination" in payload
+
+
+@settings(
+ max_examples=60,
+ suppress_health_check=[HealthCheck.function_scoped_fixture],
+ deadline=None,
+)
+@given(
+ q=st.text(max_size=800),
+ page=st.one_of(
+ st.integers(min_value=-1000, max_value=10**9),
+ st.text(max_size=40),
+ ),
+ limit=st.one_of(
+ st.integers(min_value=-1000, max_value=10**9),
+ st.text(max_size=40),
+ ),
+)
+@pytest.mark.asyncio
+async def test_get_archives_query_fuzz(mock_app, q, page, limit):
+ handler = _handler(mock_app, "GET", "/api/v1/nomadnet/archives")
+ assert handler is not None
+
+ request = MagicMock()
+ request.query = {
+ "q": q,
+ "page": str(page),
+ "limit": str(limit),
+ }
+ response = await handler(request)
+ assert response.status == 200
+ payload = json.loads(response.body)
+ assert isinstance(payload.get("archives"), list)
+ assert "pagination" in payload
+
+
+@pytest.mark.asyncio
+async def test_delete_archives_requires_nonempty_ids(mock_app):
+ handler = _handler(mock_app, "DELETE", "/api/v1/nomadnet/archives")
+ assert handler is not None
+
+ request = MagicMock()
+ request.json = AsyncMock(return_value={"ids": []})
+ response = await handler(request)
+ assert response.status == 400
+
+
+@pytest.mark.asyncio
+async def test_meshchatx_docs_content_requires_path(mock_app):
+ handler = _handler(mock_app, "GET", "/api/v1/meshchatx-docs/content")
+ assert handler is not None
+ request = MagicMock()
+ request.query = {}
+ response = await handler(request)
+ assert response.status == 400
+
+
+@settings(
+ max_examples=40,
+ suppress_health_check=[HealthCheck.function_scoped_fixture],
+ deadline=None,
+)
+@given(path=st.text(min_size=1, max_size=600))
+@pytest.mark.asyncio
+async def test_meshchatx_docs_content_path_fuzz(mock_app, path):
+ handler = _handler(mock_app, "GET", "/api/v1/meshchatx-docs/content")
+ assert handler is not None
+ request = MagicMock()
+ request.query = {"path": path}
+ response = await handler(request)
+ assert response.status in (200, 404)
+ json.loads(response.body)
+
+
+@pytest.mark.asyncio
+async def test_lxmf_messages_send_requires_lxmf_message(mock_app):
+ handler = _handler(mock_app, "POST", "/api/v1/lxmf-messages/send")
+ assert handler is not None
+ request = MagicMock()
+ request.json = AsyncMock(return_value={})
+ response = await handler(request)
+ assert response.status == 400
+
+
+@pytest.mark.asyncio
+async def test_lxmf_messages_send_requires_destination_and_content(mock_app):
+ handler = _handler(mock_app, "POST", "/api/v1/lxmf-messages/send")
+ assert handler is not None
+ request = MagicMock()
+ request.json = AsyncMock(
+ return_value={"lxmf_message": {"fields": {}}},
+ )
+ response = await handler(request)
+ assert response.status == 400
+
+
+@pytest.mark.asyncio
+async def test_lxmf_messages_send_rejects_bad_image_field(mock_app):
+ handler = _handler(mock_app, "POST", "/api/v1/lxmf-messages/send")
+ assert handler is not None
+ request = MagicMock()
+ request.json = AsyncMock(
+ return_value={
+ "lxmf_message": {
+ "destination_hash": "aa",
+ "content": "x",
+ "fields": {
+ "image": {"image_type": "png", "image_bytes": "@@@not-base64@@@"},
+ },
+ },
+ },
+ )
+ response = await handler(request)
+ assert response.status == 400
+
+
+@pytest.mark.asyncio
+async def test_nomadnet_file_download_invalid_hex_does_not_raise(mock_app):
+ await mock_app.on_websocket_data_received(
+ MagicMock(),
+ {
+ "type": "nomadnet.file.download",
+ "nomadnet_file_download": {
+ "destination_hash": "gg",
+ "file_path": "/file/x.txt",
+ },
+ },
+ )
diff --git a/tests/backend/test_deep_links_security.py b/tests/backend/test_deep_links_security.py
index 756d40d..0cbabcd 100644
--- a/tests/backend/test_deep_links_security.py
+++ b/tests/backend/test_deep_links_security.py
@@ -272,6 +272,169 @@ def test_meshchatx_map_numeric_params_fuzzing(mock_app, lat, lon, z, extra):
assert 0 <= mq["zoom"] <= 22
+@pytest.mark.asyncio
+async def test_lxm_ingest_docs_uri_with_reticulum(mock_app):
+ mock_client = MagicMock()
+ mock_client.send_str = MagicMock(return_value=asyncio.sleep(0))
+ mock_app.message_router.ingest_lxm_uri = MagicMock()
+
+ with patch(
+ "meshchatx.meshchat.AsyncUtils.run_async",
+ side_effect=lambda coro: asyncio.create_task(coro),
+ ):
+ await mock_app.on_websocket_data_received(
+ mock_client,
+ {
+ "type": "lxm.ingest_uri",
+ "uri": "meshchatx://docs?reticulum=manual/interfaces.html%23section",
+ },
+ )
+ await asyncio.sleep(0)
+
+ mock_app.message_router.ingest_lxm_uri.assert_not_called()
+ payload = json.loads(mock_client.send_str.call_args[0][0])
+ assert payload["type"] == "lxm.ingest_uri.result"
+ assert payload["status"] == "success"
+ assert payload["ingest_type"] == "docs_view"
+ assert payload["docs_query"]["reticulum"] == "manual/interfaces.html#section"
+
+
+@pytest.mark.asyncio
+async def test_lxm_ingest_docs_uri_meshchat_alias(mock_app):
+ mock_client = MagicMock()
+ mock_client.send_str = MagicMock(return_value=asyncio.sleep(0))
+ mock_app.message_router.ingest_lxm_uri = MagicMock()
+
+ with patch(
+ "meshchatx.meshchat.AsyncUtils.run_async",
+ side_effect=lambda coro: asyncio.create_task(coro),
+ ):
+ await mock_app.on_websocket_data_received(
+ mock_client,
+ {"type": "lxm.ingest_uri", "uri": "meshchat://docs?path=manual/index.html"},
+ )
+ await asyncio.sleep(0)
+
+ payload = json.loads(mock_client.send_str.call_args[0][0])
+ assert payload["ingest_type"] == "docs_view"
+ assert payload["docs_query"]["reticulum"] == "manual/index.html"
+
+
+@pytest.mark.asyncio
+async def test_lxm_ingest_docs_uri_open_index(mock_app):
+ mock_client = MagicMock()
+ mock_client.send_str = MagicMock(return_value=asyncio.sleep(0))
+ mock_app.message_router.ingest_lxm_uri = MagicMock()
+
+ with patch(
+ "meshchatx.meshchat.AsyncUtils.run_async",
+ side_effect=lambda coro: asyncio.create_task(coro),
+ ):
+ await mock_app.on_websocket_data_received(
+ mock_client,
+ {"type": "lxm.ingest_uri", "uri": "meshchatx://docs"},
+ )
+ await asyncio.sleep(0)
+
+ payload = json.loads(mock_client.send_str.call_args[0][0])
+ assert payload["ingest_type"] == "docs_view"
+ assert "docs_query" not in payload
+
+
+@pytest.mark.asyncio
+async def test_lxm_ingest_docs_hostname_spoof_not_docs_view(mock_app):
+ mock_client = MagicMock()
+ mock_client.send_str = MagicMock(return_value=asyncio.sleep(0))
+ mock_app.message_router.ingest_lxm_uri = MagicMock()
+
+ with patch(
+ "meshchatx.meshchat.AsyncUtils.run_async",
+ side_effect=lambda coro: asyncio.create_task(coro),
+ ):
+ await mock_app.on_websocket_data_received(
+ mock_client,
+ {"type": "lxm.ingest_uri", "uri": "meshchatx://docs-foo?reticulum=evil"},
+ )
+ await asyncio.sleep(0)
+
+ payload = json.loads(mock_client.send_str.call_args[0][0])
+ assert payload.get("ingest_type") != "docs_view"
+ mock_app.message_router.ingest_lxm_uri.assert_called()
+
+
+@pytest.mark.parametrize(
+ "rel",
+ [
+ "",
+ "'; DROP TABLE docs;--",
+ "a" * 8000,
+ ],
+)
+@pytest.mark.asyncio
+async def test_lxm_ingest_docs_opaque_reticulum_roundtrip(mock_app, rel):
+ mock_client = MagicMock()
+ mock_client.send_str = MagicMock(return_value=asyncio.sleep(0))
+ mock_app.message_router.ingest_lxm_uri = MagicMock()
+ q = urlencode({"reticulum": rel}, quote_via=quote)
+ uri = f"meshchatx://docs?{q}"
+
+ with patch(
+ "meshchatx.meshchat.AsyncUtils.run_async",
+ side_effect=lambda coro: asyncio.create_task(coro),
+ ):
+ await mock_app.on_websocket_data_received(
+ mock_client,
+ {"type": "lxm.ingest_uri", "uri": uri},
+ )
+ await asyncio.sleep(0)
+
+ payload = json.loads(mock_client.send_str.call_args[0][0])
+ assert payload["ingest_type"] == "docs_view"
+ assert payload["docs_query"]["reticulum"] == rel
+
+
+@settings(
+ suppress_health_check=[HealthCheck.function_scoped_fixture],
+ deadline=None,
+ max_examples=80,
+)
+@given(tail=st.text(max_size=800))
+def test_meshchatx_docs_query_tail_fuzzing(mock_app, tail):
+ uri = "meshchatx://docs?" + tail
+ mock_client = MagicMock()
+ mock_client.send_str = MagicMock(return_value=asyncio.sleep(0))
+ mock_app.message_router.ingest_lxm_uri = MagicMock()
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+
+ async def _run():
+ with patch(
+ "meshchatx.meshchat.AsyncUtils.run_async",
+ side_effect=lambda coro: asyncio.create_task(coro),
+ ):
+ await mock_app.on_websocket_data_received(
+ mock_client,
+ {"type": "lxm.ingest_uri", "uri": uri},
+ )
+ await asyncio.sleep(0)
+
+ loop.run_until_complete(_run())
+ finally:
+ loop.close()
+
+ mock_client.send_str.assert_called()
+ raw = mock_client.send_str.call_args[0][0]
+ try:
+ payload = json.loads(raw)
+ except json.JSONDecodeError:
+ return
+ assert payload["type"] == "lxm.ingest_uri.result"
+ assert payload["status"] == "success"
+ assert payload["ingest_type"] == "docs_view"
+
+
def test_telemetry_pack_location_xss_like_strings_return_none():
from meshchatx.src.backend.telemetry_utils import Telemeter
diff --git a/tests/backend/test_interface_options.py b/tests/backend/test_interface_options.py
index a03a51f..186e021 100644
--- a/tests/backend/test_interface_options.py
+++ b/tests/backend/test_interface_options.py
@@ -412,6 +412,27 @@ async def test_rnode_persists_flow_control_and_id_callsign(temp_dir):
assert saved["airtime_limit_short"] == 33.0
+@pytest.mark.asyncio
+async def test_rnode_ble_uart_port_persisted(temp_dir):
+ config = ConfigDict({"reticulum": {}, "interfaces": {}})
+
+ async with make_app(temp_dir, config) as handler:
+ payload = {
+ "name": "RadioBLE",
+ "type": "RNodeInterface",
+ "port": "ble://aa:bb:cc:dd:ee:ff",
+ "frequency": 868000000,
+ "bandwidth": 125000,
+ "txpower": 7,
+ "spreadingfactor": 8,
+ "codingrate": 5,
+ }
+ response = await handler(make_request(payload))
+ body = json.loads(response.body)
+ assert response.status == 200, body
+ assert config["interfaces"]["RadioBLE"]["port"] == "ble://aa:bb:cc:dd:ee:ff"
+
+
@pytest.mark.asyncio
async def test_rnode_frequency_mhz_decimal_normalized_to_hz(temp_dir):
config = ConfigDict({"reticulum": {}, "interfaces": {}})
diff --git a/tests/frontend/archivesPage.security.test.js b/tests/frontend/archivesPage.security.test.js
new file mode 100644
index 0000000..6fee4e9
--- /dev/null
+++ b/tests/frontend/archivesPage.security.test.js
@@ -0,0 +1,138 @@
+// SPDX-License-Identifier: 0BSD
+
+import { mount } from "@vue/test-utils";
+import { describe, it, expect, vi } from "vitest";
+import ArchivesPage from "@/components/archives/ArchivesPage.vue";
+
+function mountArchives() {
+ const routerPush = vi.fn();
+ return {
+ wrapper: mount(ArchivesPage, {
+ global: {
+ mocks: {
+ $t: (key) => key,
+ $router: { push: routerPush },
+ },
+ stubs: {
+ MaterialDesignIcon: true,
+ ArchiveSidebar: true,
+ },
+ },
+ }),
+ routerPush,
+ };
+}
+
+function randText(len) {
+ const alphabet = "abc<>\"'`\\/\u0000\n\r`topic_id=";
+ let s = "";
+ for (let i = 0; i < len; i++) {
+ s += alphabet[(Math.random() * alphabet.length) | 0];
+ }
+ return s;
+}
+
+describe("Archives page viewing-archive surface (security / fuzz)", () => {
+ const nastyPaths = [
+ "/page/article.mu`topic_id=40",
+ "/page/article.mu`topic_id=40`extra",
+ "/forum/thread.mu`sort=hot",
+ "/../../../etc/passwd",
+ "javascript:alert(1)",
+ "",
+ "a".repeat(6000),
+ ];
+
+ const nastyContents = [
+ "
",
+ "