From 3a9f6eb31667cd0b8e4d1d672bf1c84a85bc31fb Mon Sep 17 00:00:00 2001 From: Sudo-Ivan Date: Thu, 5 Mar 2026 15:59:47 -0600 Subject: [PATCH] Add tests for database backup logic and interface discovery configuration --- tests/backend/test_database_snapshots.py | 191 +++++++++- tests/backend/test_https_wss_side_sniffing.py | 168 +++++++++ tests/backend/test_interface_discovery.py | 143 +++++++ tests/backend/test_property_based.py | 77 ++++ tests/backend/test_security_fuzzing.py | 355 ++++++++++++++++++ tests/backend/test_startup_advanced.py | 63 +++- .../InterfacesDiscoveryActions.test.js | 172 +++++++++ 7 files changed, 1167 insertions(+), 2 deletions(-) create mode 100644 tests/backend/test_https_wss_side_sniffing.py create mode 100644 tests/frontend/InterfacesDiscoveryActions.test.js diff --git a/tests/backend/test_database_snapshots.py b/tests/backend/test_database_snapshots.py index bdcd0fd..13eca3c 100644 --- a/tests/backend/test_database_snapshots.py +++ b/tests/backend/test_database_snapshots.py @@ -1,3 +1,4 @@ +import json import os import shutil import tempfile @@ -75,4 +76,192 @@ def test_database_auto_backup_logic(temp_dir): assert os.path.exists(result["path"]) backup_dir = os.path.join(temp_dir, "database-backups") - assert len(os.listdir(backup_dir)) == 1 + zips = [f for f in os.listdir(backup_dir) if f.endswith(".zip")] + assert len(zips) == 1 + + +def test_backup_baseline_created_on_first_backup(temp_dir): + db_path = os.path.join(temp_dir, "test.db") + db = Database(db_path) + db.initialize() + db.backup_database(temp_dir) + baseline_path = os.path.join(temp_dir, "database-backups", "backup-baseline.json") + assert os.path.exists(baseline_path) + with open(baseline_path) as f: + data = json.load(f) + assert "message_count" in data + assert "total_bytes" in data + assert "timestamp" in data + + +def test_backup_suspicious_when_messages_gone_skips_cleanup_and_baseline(temp_dir): + db_path = os.path.join(temp_dir, "test.db") + db = Database(db_path) + db.initialize() + db.messages.upsert_lxmf_message({ + "hash": "h1", "source_hash": "s", "destination_hash": "d", "peer_hash": "p", + "state": "delivered", "progress": 1.0, "is_incoming": 1, "method": "direct", + "delivery_attempts": 0, "next_delivery_attempt_at": None, "title": "t", "content": "c", + "fields": "{}", "timestamp": 0, "rssi": None, "snr": None, "quality": None, + "is_spam": 0, "reply_to_hash": None, + }) + result1 = db.backup_database(temp_dir, max_count=3) + assert result1.get("suspicious") is not True + backup_dir = os.path.join(temp_dir, "database-backups") + zip_count_after_first = sum(1 for f in os.listdir(backup_dir) if f.endswith(".zip")) + assert zip_count_after_first == 1 + with open(os.path.join(backup_dir, "backup-baseline.json")) as f: + baseline_after_first = json.load(f) + assert baseline_after_first["message_count"] == 1 + + db.messages.delete_all_lxmf_messages() + assert db.messages.count_lxmf_messages() == 0 + result2 = db.backup_database(temp_dir, max_count=3) + assert result2.get("suspicious") is True + assert "baseline" in result2 + assert result2["baseline"]["message_count"] == 1 + assert result2["current_stats"]["message_count"] == 0 + zip_count_after_suspicious = sum(1 for f in os.listdir(backup_dir) if f.endswith(".zip")) + assert zip_count_after_suspicious == 2 + assert any("SUSPICIOUS" in f for f in os.listdir(backup_dir) if f.endswith(".zip")) + with open(os.path.join(backup_dir, "backup-baseline.json")) as f: + baseline_after_suspicious = json.load(f) + assert baseline_after_suspicious["message_count"] == 1 + + +def test_backup_suspicious_when_size_collapsed_skips_cleanup(temp_dir): + db_path = os.path.join(temp_dir, "test.db") + db = Database(db_path) + db.initialize() + for i in range(200): + db.messages.upsert_lxmf_message({ + "hash": f"h{i}", "source_hash": "s", "destination_hash": "d", "peer_hash": "p", + "state": "delivered", "progress": 1.0, "is_incoming": 1, "method": "direct", + "delivery_attempts": 0, "next_delivery_attempt_at": None, "title": "t", "content": "x" * 500, + "fields": "{}", "timestamp": float(i), "rssi": None, "snr": None, "quality": None, + "is_spam": 0, "reply_to_hash": None, + }) + db.backup_database(temp_dir, max_count=3) + baseline_path = os.path.join(temp_dir, "database-backups", "backup-baseline.json") + with open(baseline_path) as f: + baseline = json.load(f) + assert baseline["total_bytes"] > 100_000 + db.messages.delete_all_lxmf_messages() + db.execute_sql("VACUUM") + result = db.backup_database(temp_dir, max_count=3) + assert result.get("suspicious") is True + backup_dir = os.path.join(temp_dir, "database-backups") + zips = [f for f in os.listdir(backup_dir) if f.endswith(".zip")] + assert len(zips) >= 2 + + +def test_backup_normal_rotation_and_baseline_update(temp_dir): + import time + db_path = os.path.join(temp_dir, "test.db") + db = Database(db_path) + db.initialize() + db.backup_database(temp_dir, max_count=2) + time.sleep(1.1) + db.backup_database(temp_dir, max_count=2) + time.sleep(1.1) + db.backup_database(temp_dir, max_count=2) + backup_dir = os.path.join(temp_dir, "database-backups") + zips = sorted([f for f in os.listdir(backup_dir) if f.endswith(".zip")]) + assert len(zips) == 2 + assert os.path.exists(os.path.join(backup_dir, "backup-baseline.json")) + + +def test_backup_failure_does_not_remove_existing_backups(temp_dir): + from unittest.mock import patch + db_path = os.path.join(temp_dir, "test.db") + db = Database(db_path) + db.initialize() + db.backup_database(temp_dir) + backup_dir = os.path.join(temp_dir, "database-backups") + existing = [f for f in os.listdir(backup_dir) if f.endswith(".zip")] + assert len(existing) == 1 + with patch.object(db, "_backup_to_zip", side_effect=OSError("disk full")): + with pytest.raises(OSError): + db.backup_database(temp_dir, max_count=1) + still_there = [f for f in os.listdir(backup_dir) if f.endswith(".zip")] + assert len(still_there) == 1 + + +def test_check_db_health_at_open_no_baseline_ok(temp_dir): + db_path = os.path.join(temp_dir, "test.db") + db = Database(db_path) + db.initialize() + issues = db.check_db_health_at_open(temp_dir) + assert issues == [] + + +def test_check_db_health_at_open_baseline_suspicious_content(temp_dir): + db_path = os.path.join(temp_dir, "test.db") + db = Database(db_path) + db.initialize() + db.messages.upsert_lxmf_message({ + "hash": "h1", "source_hash": "s", "destination_hash": "d", "peer_hash": "p", + "state": "delivered", "progress": 1.0, "is_incoming": 1, "method": "direct", + "delivery_attempts": 0, "next_delivery_attempt_at": None, "title": "t", "content": "c", + "fields": "{}", "timestamp": 0, "rssi": None, "snr": None, "quality": None, + "is_spam": 0, "reply_to_hash": None, + }) + db.backup_database(temp_dir) + db.messages.delete_all_lxmf_messages() + issues = db.check_db_health_at_open(temp_dir) + assert len(issues) >= 1 + assert any("anomaly" in i.lower() or "messages" in i for i in issues) + + +def test_check_db_health_at_open_integrity_fail(temp_dir): + from unittest.mock import patch + db_path = os.path.join(temp_dir, "test.db") + db = Database(db_path) + db.initialize() + with patch.object(db.provider, "integrity_check", return_value=[("corrupt",)]): + issues = db.check_db_health_at_open(temp_dir) + assert len(issues) >= 1 + assert any("integrity" in i.lower() for i in issues) + + +def test_check_db_health_at_close_no_issues(temp_dir): + db_path = os.path.join(temp_dir, "test.db") + db = Database(db_path) + db.initialize() + issues = db.check_db_health_at_close(temp_dir) + assert issues == [] + + +def test_check_db_health_at_close_integrity_fail(temp_dir): + from unittest.mock import patch + db_path = os.path.join(temp_dir, "test.db") + db = Database(db_path) + db.initialize() + with patch.object(db.provider, "integrity_check", return_value=[("corrupt",)]): + issues = db.check_db_health_at_close(temp_dir) + assert len(issues) >= 1 + assert any("integrity" in i.lower() for i in issues) + + +def test_is_backup_suspicious_does_not_mistrigger_empty_baseline(): + from meshchatx.src.backend.database import Database + db = Database(":memory:") + db.initialize() + assert db._is_backup_suspicious({"message_count": 0, "total_bytes": 0}, None) is False + assert db._is_backup_suspicious({"message_count": 10, "total_bytes": 1000}, None) is False + + +def test_is_backup_suspicious_does_not_mistrigger_legitimate_empty(): + from meshchatx.src.backend.database import Database + db = Database(":memory:") + db.initialize() + baseline = {"message_count": 0, "total_bytes": 5000} + assert db._is_backup_suspicious({"message_count": 0, "total_bytes": 5000}, baseline) is False + + +def test_is_backup_suspicious_does_not_mistrigger_small_db(): + from meshchatx.src.backend.database import Database + db = Database(":memory:") + db.initialize() + baseline = {"message_count": 5, "total_bytes": 50_000} + assert db._is_backup_suspicious({"message_count": 5, "total_bytes": 55_000}, baseline) is False diff --git a/tests/backend/test_https_wss_side_sniffing.py b/tests/backend/test_https_wss_side_sniffing.py new file mode 100644 index 0000000..6c8d2c4 --- /dev/null +++ b/tests/backend/test_https_wss_side_sniffing.py @@ -0,0 +1,168 @@ +""" +Tests that HTTPS/WSS is used so local traffic cannot be sniffed by other apps. +Server must speak TLS only on the API/WS port; plain HTTP must not be accepted. +""" + +import asyncio +import os +import socket +import ssl +import tempfile +from datetime import UTC, datetime, timedelta + +import aiohttp +import pytest +from aiohttp import web +from cryptography import x509 +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.x509.oid import NameOID + + +def _make_self_signed_cert_and_key(cert_path: str, key_path: str) -> None: + private_key = rsa.generate_private_key( + public_exponent=65537, + key_size=2048, + backend=default_backend(), + ) + subject = issuer = x509.Name( + [ + x509.NameAttribute(NameOID.COUNTRY_NAME, "US"), + x509.NameAttribute(NameOID.COMMON_NAME, "localhost"), + ], + ) + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(private_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(datetime.now(UTC)) + .not_valid_after(datetime.now(UTC) + timedelta(days=365)) + .sign(private_key, hashes.SHA256(), default_backend()) + ) + os.makedirs(os.path.dirname(cert_path) or ".", exist_ok=True) + with open(cert_path, "wb") as f: + f.write(cert.public_bytes(serialization.Encoding.PEM)) + with open(key_path, "wb") as f: + f.write( + private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ), + ) + + +@pytest.fixture +def temp_storage(): + with tempfile.TemporaryDirectory() as d: + yield d + + +@pytest.fixture +def ssl_context_and_cert(temp_storage): + cert_dir = temp_storage + cert_path = os.path.join(cert_dir, "cert.pem") + key_path = os.path.join(cert_dir, "key.pem") + _make_self_signed_cert_and_key(cert_path, key_path) + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + ctx.load_cert_chain(cert_path, key_path) + return ctx, cert_path, key_path + + +@pytest.mark.asyncio +async def test_https_serves_over_tls_only_plain_http_gets_no_http_response( + ssl_context_and_cert, +): + """ + Server started with ssl_context must not respond to plain HTTP. + A local sniffer connecting with raw TCP and sending HTTP would get + TLS handshake bytes or connection closure, not plaintext HTTP response. + """ + ssl_context, _, _ = ssl_context_and_cert + app = web.Application() + + async def root_handler(_request): + return web.Response(text="ok") + + app.router.add_get("/", root_handler) + runner = web.AppRunner(app, keepalive_timeout=0) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0, ssl_context=ssl_context) + await site.start() + try: + port = site._server.sockets[0].getsockname()[1] + client_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + client_ctx.check_hostname = False + client_ctx.verify_mode = ssl.CERT_NONE + + async with aiohttp.ClientSession() as session: + resp = await session.get( + f"https://127.0.0.1:{port}/", + ssl=client_ctx, + ) + assert resp.status == 200 + assert (await resp.text()) == "ok" + + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(1.0) + try: + sock.connect(("127.0.0.1", port)) + sock.sendall(b"GET / HTTP/1.0\r\n\r\n") + try: + raw = sock.recv(1024) + except socket.timeout: + raw = b"" + finally: + sock.close() + assert not raw.startswith(b"HTTP/"), ( + "Server must not respond with plain HTTP when TLS is enabled; " + "plaintext would allow local side-sniffing" + ) + finally: + await runner.cleanup() + + +@pytest.mark.asyncio +async def test_wss_over_same_tls_port(ssl_context_and_cert): + """ + WebSocket upgrade over the same TLS port uses WSS (encrypted). + Verifies that a WS endpoint is reachable only via TLS. + """ + ssl_context, _, _ = ssl_context_and_cert + app = web.Application() + + async def ws_handler(request): + ws = web.WebSocketResponse() + await ws.prepare(request) + await ws.close() + return ws + + app.router.add_get("/ws", ws_handler) + runner = web.AppRunner(app, keepalive_timeout=0) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0, ssl_context=ssl_context) + await site.start() + try: + port = site._server.sockets[0].getsockname()[1] + client_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + client_ctx.check_hostname = False + client_ctx.verify_mode = ssl.CERT_NONE + + async with aiohttp.ClientSession() as session: + async with session.ws_connect( + f"wss://127.0.0.1:{port}/ws", + ssl=client_ctx, + ) as ws: + msg = await ws.receive() + assert msg.type in ( + aiohttp.WSMsgType.CLOSE, + aiohttp.WSMsgType.CLOSING, + aiohttp.WSMsgType.CLOSED, + ) + finally: + await runner.cleanup() + + diff --git a/tests/backend/test_interface_discovery.py b/tests/backend/test_interface_discovery.py index ab919a7..a96ca8b 100644 --- a/tests/backend/test_interface_discovery.py +++ b/tests/backend/test_interface_discovery.py @@ -50,6 +50,8 @@ async def test_reticulum_discovery_get_and_patch(temp_dir): "reticulum": { "discover_interfaces": "true", "interface_discovery_sources": "abc,def", + "interface_discovery_whitelist": "tcp-*,10.0.*", + "interface_discovery_blacklist": "tcp-bad,*:9999", "required_discovery_value": "16", "autoconnect_discovered_interfaces": "2", "network_identity": "/tmp/net_id", @@ -93,6 +95,8 @@ async def test_reticulum_discovery_get_and_patch(temp_dir): get_data = json.loads(get_response.body) assert get_data["discovery"]["discover_interfaces"] == "true" assert get_data["discovery"]["interface_discovery_sources"] == "abc,def" + assert get_data["discovery"]["interface_discovery_whitelist"] == "tcp-*,10.0.*" + assert get_data["discovery"]["interface_discovery_blacklist"] == "tcp-bad,*:9999" assert get_data["discovery"]["required_discovery_value"] == "16" assert get_data["discovery"]["autoconnect_discovered_interfaces"] == "2" assert get_data["discovery"]["network_identity"] == "/tmp/net_id" @@ -101,6 +105,8 @@ async def test_reticulum_discovery_get_and_patch(temp_dir): new_config = { "discover_interfaces": False, "interface_discovery_sources": "", + "interface_discovery_whitelist": "peer-*,172.16.*", + "interface_discovery_blacklist": "", "required_discovery_value": 18, "autoconnect_discovered_interfaces": 5, "network_identity": "/tmp/other_id", @@ -115,17 +121,154 @@ async def test_reticulum_discovery_get_and_patch(temp_dir): patch_data = json.loads(patch_response.body) assert patch_data["discovery"]["discover_interfaces"] is False assert patch_data["discovery"]["interface_discovery_sources"] is None + assert patch_data["discovery"]["interface_discovery_whitelist"] == "peer-*,172.16.*" + assert patch_data["discovery"]["interface_discovery_blacklist"] is None assert patch_data["discovery"]["required_discovery_value"] == 18 assert patch_data["discovery"]["autoconnect_discovered_interfaces"] == 5 assert patch_data["discovery"]["network_identity"] == "/tmp/other_id" assert config["reticulum"]["discover_interfaces"] is False assert "interface_discovery_sources" not in config["reticulum"] + assert config["reticulum"]["interface_discovery_whitelist"] == "peer-*,172.16.*" + assert "interface_discovery_blacklist" not in config["reticulum"] assert config["reticulum"]["required_discovery_value"] == 18 assert config["reticulum"]["autoconnect_discovered_interfaces"] == 5 assert config["reticulum"]["network_identity"] == "/tmp/other_id" assert config.write_called +@pytest.mark.asyncio +async def test_discovered_interfaces_respect_whitelist_and_blacklist(temp_dir): + config = ConfigDict( + { + "reticulum": { + "interface_discovery_whitelist": "peer-*,10.0.*", + "interface_discovery_blacklist": "*:9999,bad-*", + }, + "interfaces": {}, + }, + ) + + with ( + patch("meshchatx.meshchat.generate_ssl_certificate"), + patch("RNS.Reticulum") as mock_rns, + patch("RNS.Transport"), + patch("LXMF.LXMRouter"), + patch("meshchatx.meshchat.InterfaceDiscovery") as mock_discovery_cls, + ): + mock_reticulum = mock_rns.return_value + mock_reticulum.config = config + mock_reticulum.configpath = "/tmp/mock_config" + mock_reticulum.is_connected_to_shared_instance = False + mock_reticulum.transport_enabled.return_value = True + mock_reticulum.get_interface_stats.return_value = {"interfaces": []} + + mock_discovery = mock_discovery_cls.return_value + mock_discovery.list_discovered_interfaces.return_value = [ + { + "name": "peer-good-1", + "type": "TCPClientInterface", + "reachable_on": "10.0.0.7", + "port": 4242, + }, + { + "name": "peer-blocked-port", + "type": "TCPClientInterface", + "reachable_on": "10.0.0.8", + "port": 9999, + }, + { + "name": "bad-node", + "type": "TCPClientInterface", + "reachable_on": "10.0.0.9", + "port": 4242, + }, + { + "name": "other-network", + "type": "TCPClientInterface", + "reachable_on": "192.168.1.10", + "port": 4242, + }, + ] + + app_instance = ReticulumMeshChat( + identity=build_identity(), + storage_dir=temp_dir, + reticulum_config_dir=temp_dir, + ) + + handler = await find_route_handler( + app_instance, + "/api/v1/reticulum/discovered-interfaces", + "GET", + ) + assert handler + + response = await handler(MagicMock()) + data = json.loads(response.body) + interfaces = data["interfaces"] + + assert len(interfaces) == 1 + assert interfaces[0]["name"] == "peer-good-1" + + +@pytest.mark.asyncio +async def test_discovery_patch_sanitizes_whitelist_blacklist_values(temp_dir): + config = ConfigDict({"reticulum": {}, "interfaces": {}}) + + with ( + patch("meshchatx.meshchat.generate_ssl_certificate"), + patch("RNS.Reticulum") as mock_rns, + patch("RNS.Transport"), + patch("LXMF.LXMRouter"), + ): + mock_reticulum = mock_rns.return_value + mock_reticulum.config = config + mock_reticulum.configpath = "/tmp/mock_config" + mock_reticulum.is_connected_to_shared_instance = False + mock_reticulum.transport_enabled.return_value = True + + app_instance = ReticulumMeshChat( + identity=build_identity(), + storage_dir=temp_dir, + reticulum_config_dir=temp_dir, + ) + + patch_handler = await find_route_handler( + app_instance, + "/api/v1/reticulum/discovery", + "PATCH", + ) + assert patch_handler + + payload = { + "interface_discovery_whitelist": "peer-1,\npeer-1,host:4242,\r\n,\n", + "interface_discovery_blacklist": ["bad-node", "bad-node", "evil,\nentry"], + } + + class PatchRequest: + @staticmethod + async def json(): + return payload + + response = await patch_handler(PatchRequest()) + data = json.loads(response.body) + + assert data["discovery"]["interface_discovery_whitelist"] == "peer-1,host:4242" + assert data["discovery"]["interface_discovery_blacklist"] == "bad-node,evilentry" + assert config["reticulum"]["interface_discovery_whitelist"] == "peer-1,host:4242" + assert config["reticulum"]["interface_discovery_blacklist"] == "bad-node,evilentry" + assert config.write_called + + +def test_filter_discovered_interfaces_handles_non_list_inputs(): + result = ReticulumMeshChat.filter_discovered_interfaces( + {"unexpected": "shape"}, + "peer-*", + "bad-*", + ) + assert result == {"unexpected": "shape"} + + @pytest.mark.asyncio async def test_interface_add_includes_discovery_fields(temp_dir): config = ConfigDict({"reticulum": {}, "interfaces": {}}) diff --git a/tests/backend/test_property_based.py b/tests/backend/test_property_based.py index 3882796..be9fc96 100644 --- a/tests/backend/test_property_based.py +++ b/tests/backend/test_property_based.py @@ -8,6 +8,7 @@ import pytest from hypothesis import HealthCheck, given, settings from hypothesis import strategies as st +from meshchatx.meshchat import ReticulumMeshChat from meshchatx.src.backend.colour_utils import ColourUtils from meshchatx.src.backend.identity_manager import IdentityManager from meshchatx.src.backend.interface_config_parser import InterfaceConfigParser @@ -317,6 +318,46 @@ def test_interface_config_parser_structured(names, keys, values): pytest.fail(f"InterfaceConfigParser.parse failed on structured input: {e}") +@given( + interfaces=st.lists( + st.dictionaries( + keys=st.sampled_from( + [ + "name", + "type", + "reachable_on", + "target_host", + "remote", + "listen_ip", + "port", + "target_port", + "listen_port", + "discovery_hash", + "transport_id", + "network_id", + ] + ), + values=st.one_of(st.text(), st.integers(), st.none()), + max_size=12, + ), + max_size=40, + ), + whitelist=st.one_of(st.text(), st.lists(st.text(), max_size=20), st.none()), + blacklist=st.one_of(st.text(), st.lists(st.text(), max_size=20), st.none()), +) +def test_discovery_filter_robustness(interfaces, whitelist, blacklist): + try: + filtered = ReticulumMeshChat.filter_discovered_interfaces( + interfaces, + whitelist, + blacklist, + ) + except Exception as e: + pytest.fail(f"Discovery filtering crashed: {e}") + assert isinstance(filtered, list) + assert len(filtered) <= len(interfaces) + + # Strategy for a database message row st_db_message = st.dictionaries( keys=st.sampled_from( @@ -870,3 +911,39 @@ def test_parse_lxmf_display_name_extended(aspect, data): assert isinstance(result, str) except Exception: pass + + +MIN_SIZE_RATIO = 0.2 + + +def _is_backup_suspicious_reference(current_stats, baseline): + if not baseline: + return False + prev_count = baseline.get("message_count", 0) + prev_bytes = baseline.get("total_bytes", 0) + curr_count = current_stats.get("message_count", 0) + curr_bytes = current_stats.get("total_bytes", 0) + if prev_count > 0 and curr_count == 0: + return True + if prev_bytes > 100_000 and curr_bytes < prev_bytes * MIN_SIZE_RATIO: + return True + return False + + +@given( + prev_count=st.integers(min_value=0, max_value=1_000_000), + prev_bytes=st.integers(min_value=0, max_value=10_000_000), + curr_count=st.integers(min_value=0, max_value=1_000_000), + curr_bytes=st.integers(min_value=0, max_value=10_000_000), +) +@settings(suppress_health_check=[HealthCheck.too_slow]) +def test_is_backup_suspicious_property(prev_count, prev_bytes, curr_count, curr_bytes): + from meshchatx.src.backend.database import Database + + baseline = {"message_count": prev_count, "total_bytes": prev_bytes} + current_stats = {"message_count": curr_count, "total_bytes": curr_bytes} + db = Database(":memory:") + db.initialize() + result = db._is_backup_suspicious(current_stats, baseline) + expected = _is_backup_suspicious_reference(current_stats, baseline) + assert result == expected diff --git a/tests/backend/test_security_fuzzing.py b/tests/backend/test_security_fuzzing.py index 6dedd90..0a4b952 100644 --- a/tests/backend/test_security_fuzzing.py +++ b/tests/backend/test_security_fuzzing.py @@ -13,6 +13,21 @@ from hypothesis import strategies as st from meshchatx.meshchat import ReticulumMeshChat +@given(value=st.one_of(st.text(), st.lists(st.text(), min_size=0, max_size=20))) +@settings(deadline=None) +def test_discovery_pattern_sanitization_never_emits_unsafe_tokens(value): + patterns = ReticulumMeshChat.sanitize_discovery_patterns(value) + assert isinstance(patterns, list) + assert len(patterns) <= 128 + for token in patterns: + assert token + assert len(token) <= 128 + assert "," not in token + assert "\n" not in token + assert "\r" not in token + assert token.strip() == token + + @pytest.fixture def mock_app(): # Save real Identity class to use as base for our mock class @@ -1280,6 +1295,346 @@ def test_nomadnet_page_content_fuzzing(mock_app, page_content): convert_nomadnet_field_data_to_map(page_content) +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given(path_data=st.one_of(st.text(min_size=0, max_size=10000), st.none())) +def test_nomadnet_string_data_to_map_fuzzing(mock_app, path_data): + """Fuzz NomadNet path variable string parsing (e.g. after backtick in page path).""" + from meshchatx.src.backend.nomadnet_utils import convert_nomadnet_string_data_to_map + + result = convert_nomadnet_string_data_to_map(path_data) + assert isinstance(result, dict) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + download_id=st.one_of( + st.integers(), + st.text(min_size=0, max_size=200), + st.none(), + ), +) +def test_nomadnet_download_cancel_fuzzing(mock_app, download_id): + """Fuzz nomadnet.download.cancel WebSocket handler.""" + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + loop.run_until_complete( + mock_app.on_websocket_data_received( + MagicMock(), + {"type": "nomadnet.download.cancel", "download_id": download_id}, + ), + ) + finally: + loop.close() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + destination_hash=st.text(min_size=0, max_size=200), + page_path=st.text(min_size=0, max_size=2000), +) +def test_nomadnet_page_archives_get_fuzzing(mock_app, destination_hash, page_path): + """Fuzz nomadnet.page.archives.get WebSocket handler.""" + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + loop.run_until_complete( + mock_app.on_websocket_data_received( + MagicMock(), + { + "type": "nomadnet.page.archives.get", + "destination_hash": destination_hash, + "page_path": page_path, + }, + ), + ) + finally: + loop.close() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + archive_id=st.one_of( + st.integers(), + st.text(min_size=0, max_size=100), + st.floats(allow_nan=False, allow_infinity=False), + st.none(), + ), +) +def test_nomadnet_page_archive_load_fuzzing(mock_app, archive_id): + """Fuzz nomadnet.page.archive.load and get_archived_page_by_id.""" + mock_app.database.misc.get_archived_page_by_id(archive_id) + import asyncio + + mock_app.database.misc.get_archived_page_by_id.return_value = None + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + loop.run_until_complete( + mock_app.on_websocket_data_received( + MagicMock(), + {"type": "nomadnet.page.archive.load", "archive_id": archive_id}, + ), + ) + finally: + loop.close() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + destination_hash=st.text(min_size=0, max_size=200), + page_path=st.text(min_size=0, max_size=2000), + content=st.one_of( + st.text(min_size=0, max_size=50000), + st.binary(min_size=0, max_size=10000), + ), +) +def test_nomadnet_page_archive_add_fuzzing( + mock_app, destination_hash, page_path, content +): + """Fuzz nomadnet.page.archive.add WebSocket handler and archive_page.""" + import asyncio + + content_str = ( + content.decode("utf-8", errors="replace") + if isinstance(content, bytes) + else content + ) + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + loop.run_until_complete( + mock_app.on_websocket_data_received( + MagicMock(), + { + "type": "nomadnet.page.archive.add", + "destination_hash": destination_hash, + "page_path": page_path, + "content": content_str, + }, + ), + ) + finally: + loop.close() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + destination_hash=st.text(min_size=0, max_size=200), + file_path=st.text(min_size=0, max_size=2000), +) +def test_nomadnet_file_download_fuzzing(mock_app, destination_hash, file_path): + """Fuzz nomadnet.file.download WebSocket handler (path traversal, malformed hash).""" + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + payload = { + "type": "nomadnet.file.download", + "nomadnet_file_download": { + "destination_hash": destination_hash, + "file_path": file_path, + }, + } + loop.run_until_complete( + mock_app.on_websocket_data_received(MagicMock(), payload), + ) + except (ValueError, TypeError): + pass + finally: + loop.close() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + destination_hash=st.text(min_size=0, max_size=200), + page_path=st.text(min_size=0, max_size=2000), + field_data=st.one_of( + st.dictionaries( + keys=st.text(), + values=st.one_of(st.text(), st.integers(), st.booleans(), st.none()), + ), + st.text(min_size=0, max_size=5000), + st.lists(st.text(), min_size=0, max_size=20), + st.none(), + ), +) +def test_nomadnet_page_download_fuzzing( + mock_app, destination_hash, page_path, field_data +): + """Fuzz nomadnet.page.download WebSocket handler (page_path with backtick, field_data).""" + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + payload = { + "type": "nomadnet.page.download", + "nomadnet_page_download": { + "destination_hash": destination_hash, + "page_path": page_path, + "field_data": field_data, + }, + } + loop.run_until_complete( + mock_app.on_websocket_data_received(MagicMock(), payload), + ) + except (ValueError, TypeError): + pass + finally: + loop.close() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given(archive_id=st.one_of(st.integers(), st.text(), st.floats(), st.none())) +def test_get_archived_page_by_id_fuzzing(mock_app, archive_id): + """Fuzz archived page lookup by id (SQL injection, type confusion).""" + mock_app.database.misc.get_archived_page_by_id(archive_id) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + forward_to_hash=st.text(min_size=0, max_size=100), + identity_hash=st.one_of(st.text(min_size=0, max_size=100), st.none()), + source_filter_hash=st.one_of(st.text(min_size=0, max_size=100), st.none()), + is_active=st.one_of(st.booleans(), st.integers(), st.none()), + name=st.one_of(st.text(min_size=0, max_size=200), st.none()), +) +def test_lxmf_forwarding_rule_add_fuzzing( + mock_app, + forward_to_hash, + identity_hash, + source_filter_hash, + is_active, + name, +): + """Fuzz lxmf.forwarding.rule.add WebSocket handler and create_forwarding_rule.""" + import asyncio + + rule = {"forward_to_hash": forward_to_hash} + if identity_hash is not None: + rule["identity_hash"] = identity_hash + if source_filter_hash is not None: + rule["source_filter_hash"] = source_filter_hash + if is_active is not None: + rule["is_active"] = is_active + if name is not None: + rule["name"] = name + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + loop.run_until_complete( + mock_app.on_websocket_data_received( + MagicMock(), + {"type": "lxmf.forwarding.rule.add", "rule": rule}, + ), + ) + finally: + loop.close() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + rule_id=st.one_of(st.integers(), st.text(min_size=0, max_size=50), st.none()), +) +def test_lxmf_forwarding_rule_delete_fuzzing(mock_app, rule_id): + """Fuzz lxmf.forwarding.rule.delete WebSocket handler.""" + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + loop.run_until_complete( + mock_app.on_websocket_data_received( + MagicMock(), + {"type": "lxmf.forwarding.rule.delete", "id": rule_id}, + ), + ) + finally: + loop.close() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + rule_id=st.one_of(st.integers(), st.text(min_size=0, max_size=50), st.none()), +) +def test_lxmf_forwarding_rule_toggle_fuzzing(mock_app, rule_id): + """Fuzz lxmf.forwarding.rule.toggle WebSocket handler.""" + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + loop.run_until_complete( + mock_app.on_websocket_data_received( + MagicMock(), + {"type": "lxmf.forwarding.rule.toggle", "id": rule_id}, + ), + ) + finally: + loop.close() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + action=st.text(min_size=0, max_size=200), + keys=st.one_of( + st.lists(st.text(), min_size=0, max_size=20), + st.text(min_size=0, max_size=500), + st.integers(), + ), +) +def test_keyboard_shortcuts_set_fuzzing(mock_app, action, keys): + """Fuzz keyboard_shortcuts.set WebSocket handler (action and keys).""" + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + loop.run_until_complete( + mock_app.on_websocket_data_received( + MagicMock(), + {"type": "keyboard_shortcuts.set", "action": action, "keys": keys}, + ), + ) + except (KeyError, TypeError, ValueError): + pass + finally: + loop.close() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given(message_hash=st.text(min_size=0, max_size=200)) +def test_messages_get_by_hash_fuzzing(mock_app, message_hash): + """Fuzz message lookup by hash (SQL, type confusion).""" + mock_app.database.messages.get_lxmf_message_by_hash(message_hash) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given(message_hash=st.text(min_size=0, max_size=200)) +def test_messages_delete_by_hash_fuzzing(mock_app, message_hash): + """Fuzz single message delete by hash.""" + try: + mock_app.database.messages.delete_lxmf_message_by_hash(message_hash) + except Exception: + pass + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given(message_hashes=st.lists(st.text(min_size=0, max_size=100), min_size=0, max_size=50)) +def test_messages_delete_by_hashes_fuzzing(mock_app, message_hashes): + """Fuzz bulk message delete by hashes.""" + try: + mock_app.database.messages.delete_lxmf_messages_by_hashes(message_hashes) + except Exception: + pass + + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( table_name=st.text(min_size=0, max_size=100), diff --git a/tests/backend/test_startup_advanced.py b/tests/backend/test_startup_advanced.py index 045ae8a..d61f96e 100644 --- a/tests/backend/test_startup_advanced.py +++ b/tests/backend/test_startup_advanced.py @@ -160,7 +160,11 @@ def test_database_integrity_recovery(mock_rns, temp_dir): patch("meshchatx.src.backend.identity_context.RNProbeHandler"), patch("meshchatx.src.backend.identity_context.TranslatorHandler"), patch("meshchatx.src.backend.identity_context.CommunityInterfacesManager"), + patch( + "meshchatx.src.backend.identity_context.IntegrityManager", + ) as mock_im_class, ): + mock_im_class.return_value.check_integrity.return_value = (True, []) mock_db_instance = mock_db_class.return_value # Fail the first initialize call mock_db_instance.initialize.side_effect = [ @@ -228,7 +232,64 @@ def test_identity_loading_fallback(mock_rns, temp_dir): mock_gen_id.get_private_key.assert_called() -# 4. Test flags/envs +# 4. Database health issues set on setup and exposed to app +def test_database_health_issues_set_on_setup(mock_rns, temp_dir): + with ( + patch("meshchatx.src.backend.identity_context.Database") as mock_db_class, + patch( + "meshchatx.src.backend.identity_context.ConfigManager", + ) as mock_config_class, + patch("meshchatx.src.backend.identity_context.MessageHandler"), + patch("meshchatx.src.backend.identity_context.AnnounceManager"), + patch("meshchatx.src.backend.identity_context.ArchiverManager"), + patch("meshchatx.src.backend.identity_context.MapManager"), + patch("meshchatx.src.backend.identity_context.DocsManager"), + patch("meshchatx.src.backend.identity_context.NomadNetworkManager"), + patch("meshchatx.src.backend.identity_context.TelephoneManager"), + patch("meshchatx.src.backend.identity_context.VoicemailManager"), + patch("meshchatx.src.backend.identity_context.RingtoneManager"), + patch("meshchatx.src.backend.identity_context.RNCPHandler"), + patch("meshchatx.src.backend.identity_context.RNStatusHandler"), + patch("meshchatx.src.backend.identity_context.RNProbeHandler"), + patch("meshchatx.src.backend.identity_context.TranslatorHandler"), + patch("meshchatx.src.backend.identity_context.CommunityInterfacesManager"), + patch( + "meshchatx.src.backend.identity_context.IntegrityManager", + ) as mock_int_class, + patch("aiohttp.web.run_app"), + ): + mock_int_class.return_value.check_integrity.return_value = (True, []) + mock_db_instance = mock_db_class.return_value + mock_db_instance.check_db_health_at_open.return_value = [ + "Database content anomaly: test." + ] + mock_config = mock_config_class.return_value + mock_config.auth_session_secret.get.return_value = base64.urlsafe_b64encode( + secrets.token_bytes(32), + ).decode() + mock_config.display_name.get.return_value = "Test" + mock_config.lxmf_propagation_node_stamp_cost.get.return_value = 0 + mock_config.lxmf_delivery_transfer_limit_in_bytes.get.return_value = 1000000 + mock_config.lxmf_inbound_stamp_cost.get.return_value = 0 + mock_config.lxmf_preferred_propagation_node_destination_hash.get.return_value = None + mock_config.lxmf_local_propagation_node_enabled.get.return_value = False + mock_config.libretranslate_url.get.return_value = "http://localhost:5000" + mock_config.translator_enabled.get.return_value = False + mock_config.initial_docs_download_attempted.get.return_value = True + + app = ReticulumMeshChat( + identity=mock_rns["id_instance"], + storage_dir=temp_dir, + reticulum_config_dir=temp_dir, + ) + app.run(host="127.0.0.1", port=8000, launch_browser=False, enable_https=False) + assert getattr(app, "database_health_issues", []) == [ + "Database content anomaly: test." + ] + app.teardown_identity() + + +# 5. Test flags/envs def test_cli_flags_and_envs(mock_rns, temp_dir): with ( patch("meshchatx.meshchat.ReticulumMeshChat") as mock_app_class, diff --git a/tests/frontend/InterfacesDiscoveryActions.test.js b/tests/frontend/InterfacesDiscoveryActions.test.js new file mode 100644 index 0000000..48fd8c6 --- /dev/null +++ b/tests/frontend/InterfacesDiscoveryActions.test.js @@ -0,0 +1,172 @@ +import { mount } from "@vue/test-utils"; +import { describe, it, expect, vi, beforeEach } from "vitest"; +import InterfacesPage from "../../meshchatx/src/frontend/components/interfaces/InterfacesPage.vue"; + +vi.mock("../../meshchatx/src/frontend/js/GlobalState", () => ({ + default: { + config: { theme: "light" }, + hasPendingInterfaceChanges: false, + modifiedInterfaceNames: new Set(), + }, +})); + +vi.mock("../../meshchatx/src/frontend/js/Utils", () => ({ + default: { + formatBytes: (b) => `${b} B`, + isInterfaceEnabled: () => true, + }, +})); + +vi.mock("../../meshchatx/src/frontend/js/ToastUtils", () => ({ + default: { + success: vi.fn(), + error: vi.fn(), + }, +})); + +vi.mock("../../meshchatx/src/frontend/js/ElectronUtils", () => ({ + default: { + relaunch: vi.fn(), + isElectron: () => false, + }, +})); + +const mockAxios = { + get: vi.fn((url) => { + if (url.includes("/api/v1/reticulum/interfaces")) { + return Promise.resolve({ data: { interfaces: {} } }); + } + if (url.includes("/api/v1/app/info")) { + return Promise.resolve({ data: { app_info: { is_reticulum_running: true } } }); + } + if (url.includes("/api/v1/interface-stats")) { + return Promise.resolve({ data: { interface_stats: { interfaces: [] } } }); + } + if (url.includes("/api/v1/reticulum/discovery")) { + return Promise.resolve({ + data: { + discovery: { + discover_interfaces: true, + interface_discovery_whitelist: "", + interface_discovery_blacklist: "", + }, + }, + }); + } + if (url.includes("/api/v1/reticulum/discovered-interfaces")) { + return Promise.resolve({ data: { interfaces: [], active: [] } }); + } + return Promise.resolve({ data: {} }); + }), + post: vi.fn(() => Promise.resolve({ data: {} })), + patch: vi.fn(() => Promise.resolve({ data: {} })), +}; +window.axios = mockAxios; + +describe("InterfacesPage discovery actions", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("shows discovery action menu and allows announce", async () => { + const wrapper = mount(InterfacesPage, { + global: { + stubs: { + RouterLink: true, + MaterialDesignIcon: true, + Toggle: true, + ImportInterfacesModal: true, + Interface: true, + }, + mocks: { + $t: (key) => key, + $router: { push: vi.fn() }, + }, + }, + }); + + await wrapper.setData({ + discoveredInterfaces: [ + { + name: "Peer Node", + type: "TCPClientInterface", + reachable_on: "10.0.0.8", + port: 4242, + discovery_hash: "hash-1", + }, + ], + discoveryConfig: { + discover_interfaces: true, + interface_discovery_sources: "", + interface_discovery_whitelist: "", + interface_discovery_blacklist: "old-entry", + required_discovery_value: null, + autoconnect_discovered_interfaces: 0, + network_identity: "", + }, + }); + + const actionButton = wrapper.find('button[title="Discovery actions"]'); + expect(actionButton.exists()).toBe(true); + await actionButton.trigger("click"); + expect(wrapper.text()).toContain("Allow this announce"); + + const allowButton = wrapper.findAll("button").find((btn) => btn.text().includes("Allow this announce")); + expect(allowButton).toBeTruthy(); + await allowButton.trigger("click"); + + expect(mockAxios.patch).toHaveBeenCalledWith("/api/v1/reticulum/discovery", { + interface_discovery_whitelist: "10.0.0.8:4242", + interface_discovery_blacklist: null, + }); + }); + + it("blacklists announce and removes same token from whitelist", async () => { + const wrapper = mount(InterfacesPage, { + global: { + stubs: { + RouterLink: true, + MaterialDesignIcon: true, + Toggle: true, + ImportInterfacesModal: true, + Interface: true, + }, + mocks: { + $t: (key) => key, + $router: { push: vi.fn() }, + }, + }, + }); + + await wrapper.setData({ + discoveredInterfaces: [ + { + name: "Peer Node", + type: "TCPClientInterface", + reachable_on: "10.0.0.8", + port: 4242, + discovery_hash: "hash-2", + }, + ], + discoveryConfig: { + discover_interfaces: true, + interface_discovery_sources: "", + interface_discovery_whitelist: "10.0.0.8:4242,other", + interface_discovery_blacklist: "", + required_discovery_value: null, + autoconnect_discovered_interfaces: 0, + network_identity: "", + }, + }); + + await wrapper.find('button[title="Discovery actions"]').trigger("click"); + const blockButton = wrapper.findAll("button").find((btn) => btn.text().includes("Blacklist this announce")); + expect(blockButton).toBeTruthy(); + await blockButton.trigger("click"); + + expect(mockAxios.patch).toHaveBeenCalledWith("/api/v1/reticulum/discovery", { + interface_discovery_whitelist: null, + interface_discovery_blacklist: "10.0.0.8:4242", + }); + }); +});