From beb86880e009b760ec29b4b4ef9fb04e8d67e226 Mon Sep 17 00:00:00 2001 From: Sudo-Ivan Date: Fri, 2 Jan 2026 20:53:21 -0600 Subject: [PATCH] refactor(tests): format --- tests/backend/test_fuzzing.py | 260 +++++++++++++---------- tests/backend/test_meshchat_utils.py | 8 +- tests/backend/test_rns_lifecycle.py | 12 +- tests/backend/test_security_fuzzing.py | 282 +++++++++++++++++-------- 4 files changed, 359 insertions(+), 203 deletions(-) diff --git a/tests/backend/test_fuzzing.py b/tests/backend/test_fuzzing.py index 2faa48f..099be1b 100644 --- a/tests/backend/test_fuzzing.py +++ b/tests/backend/test_fuzzing.py @@ -1,21 +1,27 @@ -import pytest import os -import time import random -from unittest.mock import MagicMock, patch -from hypothesis import given, strategies as st, settings, HealthCheck -from meshchatx.meshchat import ReticulumMeshChat -import RNS -import LXMF +import time from contextlib import ExitStack -from meshchatx.src.backend.telemetry_utils import Telemeter +from unittest.mock import MagicMock, patch + +import LXMF +import pytest +import RNS +from hypothesis import HealthCheck, given, settings +from hypothesis import strategies as st + +from meshchatx.meshchat import ReticulumMeshChat from meshchatx.src.backend.interface_config_parser import InterfaceConfigParser -from meshchatx.src.backend.lxmf_message_fields import LxmfAudioField, LxmfImageField, LxmfFileAttachment +from meshchatx.src.backend.lxmf_message_fields import ( + LxmfAudioField, + LxmfFileAttachment, + LxmfImageField, +) +from meshchatx.src.backend.telemetry_utils import Telemeter + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) -@given( - data=st.binary(min_size=0, max_size=1000) -) +@given(data=st.binary(min_size=0, max_size=1000)) def test_telemetry_unpack_fuzzing(data): """Fuzz the telemetry unpacking logic with random binary data.""" try: @@ -25,10 +31,9 @@ def test_telemetry_unpack_fuzzing(data): # We expect some failures for invalid packed data, but no crashes pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) -@given( - config_text=st.text(min_size=0, max_size=5000) -) +@given(config_text=st.text(min_size=0, max_size=5000)) def test_interface_config_parsing_fuzzing(config_text): """Fuzz the interface configuration parser with random text.""" try: @@ -36,45 +41,49 @@ def test_interface_config_parsing_fuzzing(config_text): except Exception as e: pytest.fail(f"InterfaceConfigParser crashed with input: {e}") + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) -@given( - identity_bytes=st.binary(min_size=0, max_size=2048) -) +@given(identity_bytes=st.binary(min_size=0, max_size=2048)) def test_identity_parsing_fuzzing(identity_bytes): """Fuzz RNS.Identity loading with random bytes.""" try: RNS.Identity.from_bytes(identity_bytes) except Exception: - # RNS.Identity.from_bytes is expected to fail on random bytes, + # RNS.Identity.from_bytes is expected to fail on random bytes, # but it should not cause an unhandled crash of the process. pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) -@given( - path_data=st.one_of(st.none(), st.text(min_size=0, max_size=1000)) -) +@given(path_data=st.one_of(st.none(), st.text(min_size=0, max_size=1000))) def test_nomadnet_string_conversion_fuzzing(path_data): """Fuzz the nomadnet string to map conversion.""" try: ReticulumMeshChat.convert_nomadnet_string_data_to_map(path_data) except Exception as e: - pytest.fail(f"convert_nomadnet_string_data_to_map crashed with data {path_data}: {e}") + pytest.fail( + f"convert_nomadnet_string_data_to_map crashed with data {path_data}: {e}", + ) + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( - field_data=st.one_of(st.none(), st.dictionaries(keys=st.text(), values=st.text()), st.text()) + field_data=st.one_of( + st.none(), st.dictionaries(keys=st.text(), values=st.text()), st.text(), + ), ) def test_nomadnet_field_conversion_fuzzing(field_data): """Fuzz the nomadnet field data to map conversion.""" try: ReticulumMeshChat.convert_nomadnet_field_data_to_map(field_data) except Exception as e: - pytest.fail(f"convert_nomadnet_field_data_to_map crashed with data {field_data}: {e}") + pytest.fail( + f"convert_nomadnet_field_data_to_map crashed with data {field_data}: {e}", + ) + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) -@given( - app_data_base64=st.one_of(st.none(), st.text(min_size=0, max_size=1000)) -) +@given(app_data_base64=st.one_of(st.none(), st.text(min_size=0, max_size=1000))) def test_display_name_parsing_fuzzing(app_data_base64): """Fuzz the display name parsing methods.""" try: @@ -83,10 +92,12 @@ def test_display_name_parsing_fuzzing(app_data_base64): except Exception as e: pytest.fail(f"Display name parsing crashed with data {app_data_base64}: {e}") + @pytest.fixture def temp_dir(tmp_path): return str(tmp_path) + @pytest.fixture def mock_app(temp_dir): with ExitStack() as stack: @@ -110,9 +121,17 @@ def mock_app(temp_dir): stack.enter_context(patch("RNS.Reticulum")) stack.enter_context(patch("RNS.Transport")) stack.enter_context(patch("threading.Thread")) - stack.enter_context(patch.object(ReticulumMeshChat, "announce_loop", return_value=None)) - stack.enter_context(patch.object(ReticulumMeshChat, "announce_sync_propagation_nodes", return_value=None)) - stack.enter_context(patch.object(ReticulumMeshChat, "crawler_loop", return_value=None)) + stack.enter_context( + patch.object(ReticulumMeshChat, "announce_loop", return_value=None), + ) + stack.enter_context( + patch.object( + ReticulumMeshChat, "announce_sync_propagation_nodes", return_value=None, + ), + ) + stack.enter_context( + patch.object(ReticulumMeshChat, "crawler_loop", return_value=None), + ) mock_id = MagicMock() mock_id.hash = b"test_hash_32_bytes_long_01234567" @@ -127,7 +146,7 @@ def mock_app(temp_dir): storage_dir=temp_dir, reticulum_config_dir=temp_dir, ) - + # Setup config mock to return real values to avoid JSON serialization issues app.config = MagicMock() app.config.display_name.get.return_value = "Test User" @@ -141,7 +160,9 @@ def mock_app(temp_dir): app.config.auto_send_failed_messages_to_propagation_node.get.return_value = True app.config.show_suggested_community_interfaces.get.return_value = True app.config.lxmf_local_propagation_node_enabled.get.return_value = False - app.config.lxmf_preferred_propagation_node_destination_hash.get.return_value = None + app.config.lxmf_preferred_propagation_node_destination_hash.get.return_value = ( + None + ) app.config.lxmf_preferred_propagation_node_auto_sync_interval_seconds.get.return_value = 3600 app.config.lxmf_preferred_propagation_node_last_synced_at.get.return_value = 0 app.config.lxmf_user_icon_name.get.return_value = "user" @@ -153,10 +174,16 @@ def mock_app(temp_dir): app.config.lxmf_auto_sync_propagation_nodes_min_hops.get.return_value = 1 app.config.lxmf_auto_sync_propagation_nodes_max_hops.get.return_value = 5 app.config.lxmf_auto_sync_propagation_nodes_max_count.get.return_value = 10 - app.config.lxmf_auto_sync_propagation_nodes_max_age_seconds.get.return_value = 86400 - app.config.lxmf_auto_sync_propagation_nodes_max_size_bytes.get.return_value = 1000000 + app.config.lxmf_auto_sync_propagation_nodes_max_age_seconds.get.return_value = ( + 86400 + ) + app.config.lxmf_auto_sync_propagation_nodes_max_size_bytes.get.return_value = ( + 1000000 + ) app.config.lxmf_auto_sync_propagation_nodes_max_total_size_bytes.get.return_value = 10000000 - app.config.lxmf_auto_sync_propagation_nodes_max_total_count.get.return_value = 100 + app.config.lxmf_auto_sync_propagation_nodes_max_total_count.get.return_value = ( + 100 + ) app.config.lxmf_auto_sync_propagation_nodes_max_total_age_seconds.get.return_value = 864000 app.config.lxmf_auto_sync_propagation_nodes_max_total_size_bytes_per_node.get.return_value = 1000000 app.config.lxmf_auto_sync_propagation_nodes_max_total_count_per_node.get.return_value = 100 @@ -169,10 +196,13 @@ def mock_app(temp_dir): app.handle_forwarding = MagicMock() app.convert_db_announce_to_dict = MagicMock(return_value={}) app.get_config_dict = MagicMock(return_value={"test_config": "test_value"}) - app.resend_failed_messages_for_destination = MagicMock(side_effect=lambda dest: None) - + app.resend_failed_messages_for_destination = MagicMock( + side_effect=lambda dest: None, + ) + yield app + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( num_announces=st.integers(min_value=10, max_value=100), @@ -181,34 +211,31 @@ def test_announce_overload(mock_app, num_announces): """Test handling of multiple announces in rapid succession.""" mock_app.announce_manager.upsert_announce.reset_mock() mock_app.websocket_broadcast.reset_mock() - + aspect = "lxmf.delivery" app_data = b"test_app_data" - + # Mock database to return a valid announce dict mock_app.database.announces.get_announce_by_hash.return_value = { "aspect": "lxmf.delivery", "destination_hash": "some_hash", - "display_name": "Test Peer" + "display_name": "Test Peer", } - + for i in range(num_announces): destination_hash = os.urandom(16) announced_identity = MagicMock() announced_identity.hash = os.urandom(32) announce_packet_hash = os.urandom(16) - + mock_app.on_lxmf_announce_received( - aspect, - destination_hash, - announced_identity, - app_data, - announce_packet_hash + aspect, destination_hash, announced_identity, app_data, announce_packet_hash, ) - + # Verify that the database was called for each announce assert mock_app.announce_manager.upsert_announce.call_count == num_announces + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( num_messages=st.integers(min_value=10, max_value=100), @@ -216,19 +243,20 @@ def test_announce_overload(mock_app, num_announces): def test_message_spamming(mock_app, num_messages): """Test handling of many LXMF messages in rapid succession.""" mock_app.db_upsert_lxmf_message.reset_mock() - + for i in range(num_messages): mock_message = MagicMock() mock_message.source_hash = os.urandom(16) mock_message.hash = os.urandom(16) - mock_message.get_fields.return_value = {} # No telemetry field + mock_message.get_fields.return_value = {} # No telemetry field mock_message.title = f"Spam Title {i}" mock_message.content = f"Spam Content {i}" - + mock_app.on_lxmf_delivery(mock_message) - + assert mock_app.db_upsert_lxmf_message.call_count == num_messages + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( num_messages=st.integers(min_value=10, max_value=50), @@ -237,7 +265,7 @@ def test_message_spamming(mock_app, num_messages): def test_message_spamming_large_payloads(mock_app, num_messages, payload_size): """Test handling of many LXMF messages with large payloads.""" mock_app.db_upsert_lxmf_message.reset_mock() - + for i in range(num_messages): mock_message = MagicMock() mock_message.source_hash = os.urandom(16) @@ -245,23 +273,39 @@ def test_message_spamming_large_payloads(mock_app, num_messages, payload_size): mock_message.get_fields.return_value = {} mock_message.title = f"Spam Title {i}" mock_message.content = "A" * payload_size - + mock_app.on_lxmf_delivery(mock_message) - + assert mock_app.db_upsert_lxmf_message.call_count == num_messages + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( msg=st.dictionaries( keys=st.text(), - values=st.one_of(st.text(), st.integers(), st.booleans(), st.dictionaries(keys=st.text(), values=st.text())), - min_size=1, max_size=10 - ).flatmap(lambda d: st.sampled_from([ - "ping", "config.set", "nomadnet.download.cancel", - "nomadnet.page.archives.get", "lxmf.forwarding.rule.add", - "lxmf.forwarding.rule.delete", "lxm.ingest_uri", - "lxm.generate_paper_uri", "keyboard_shortcuts.get" - ]).map(lambda t: {**d, "type": t})) + values=st.one_of( + st.text(), + st.integers(), + st.booleans(), + st.dictionaries(keys=st.text(), values=st.text()), + ), + min_size=1, + max_size=10, + ).flatmap( + lambda d: st.sampled_from( + [ + "ping", + "config.set", + "nomadnet.download.cancel", + "nomadnet.page.archives.get", + "lxmf.forwarding.rule.add", + "lxmf.forwarding.rule.delete", + "lxm.ingest_uri", + "lxm.generate_paper_uri", + "keyboard_shortcuts.get", + ], + ).map(lambda t: {**d, "type": t}), + ), ) @pytest.mark.asyncio async def test_websocket_api_hypothesis(mock_app, msg): @@ -274,33 +318,43 @@ async def test_websocket_api_hypothesis(mock_app, msg): except Exception: pass + @pytest.mark.asyncio async def test_websocket_api_fuzzing(mock_app): """Fuzz the websocket API with various message types and payloads.""" # Use MagicMock instead of AsyncMock to avoid coroutine warnings mock_client = MagicMock() mock_client.send_str = MagicMock(side_effect=lambda data: None) - + # Test cases with different message types and malformed/unexpected data fuzz_messages = [ {"type": "ping"}, {"type": "config.set", "config": {"invalid_key": "invalid_value"}}, {"type": "config.set", "config": "not_a_dict"}, {"type": "nomadnet.download.cancel", "download_id": "non_existent_id"}, - {"type": "nomadnet.page.archives.get", "destination_hash": "invalid_hash", "page_path": "/invalid"}, + { + "type": "nomadnet.page.archives.get", + "destination_hash": "invalid_hash", + "page_path": "/invalid", + }, {"type": "lxmf.forwarding.rule.add", "rule": {}}, {"type": "lxmf.forwarding.rule.delete", "id": -1}, {"type": "lxm.ingest_uri", "uri": "invalid_uri"}, - {"type": "lxm.generate_paper_uri", "destination_hash": "00" * 16, "content": "test"}, + { + "type": "lxm.generate_paper_uri", + "destination_hash": "00" * 16, + "content": "test", + }, {"type": "non_existent_type", "data": "random_data"}, ] - + for msg in fuzz_messages: try: await mock_app.on_websocket_data_received(mock_client, msg) except Exception: pass + @pytest.mark.asyncio async def test_config_fuzzing(mock_app): """Fuzz the config update logic with various values.""" @@ -311,7 +365,7 @@ async def test_config_fuzzing(mock_app): {"unknown_config_option": 123}, {}, ] - + for config in fuzz_configs: try: if hasattr(mock_app, "update_config"): @@ -319,44 +373,37 @@ async def test_config_fuzzing(mock_app): except Exception: pass + def test_malformed_announce_data(mock_app): """Test handling of malformed or unexpected data in announces.""" aspect = "lxmf.delivery" - destination_hash = b"too_short" # Malformed hash - + destination_hash = b"too_short" # Malformed hash + # Test with None identity - mock_app.on_lxmf_announce_received( - aspect, - destination_hash, - None, - None, - b"" - ) - + mock_app.on_lxmf_announce_received(aspect, destination_hash, None, None, b"") + # Test with identity having None hash announced_identity = MagicMock() announced_identity.hash = None mock_app.on_lxmf_announce_received( - aspect, - destination_hash, - announced_identity, - None, - b"" + aspect, destination_hash, announced_identity, None, b"", ) + def test_malformed_message_data(mock_app): """Test handling of malformed LXMF messages.""" mock_message = MagicMock() # Simulate missing attributes or methods del mock_message.source_hash - + # This should be caught by the try-except in on_lxmf_delivery mock_app.on_lxmf_delivery(mock_message) + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( weird_string=st.text(min_size=0, max_size=1000), - large_binary=st.binary(min_size=0, max_size=10000) + large_binary=st.binary(min_size=0, max_size=10000), ) def test_database_dao_fuzzing(mock_app, weird_string, large_binary): """Fuzz the database DAOs with weird strings and large binary data.""" @@ -369,7 +416,7 @@ def test_database_dao_fuzzing(mock_app, weird_string, large_binary): "app_data": large_binary, "rssi": random.randint(-120, 0), "snr": random.uniform(-20, 20), - "quality": random.uniform(0, 100) + "quality": random.uniform(0, 100), } try: mock_app.database.announces.upsert_announce(announce_data) @@ -387,31 +434,34 @@ def test_database_dao_fuzzing(mock_app, weird_string, large_binary): "fields": {"weird": weird_string}, "timestamp": time.time(), "is_incoming": random.choice([0, 1]), - "is_spam": random.choice([0, 1]) + "is_spam": random.choice([0, 1]), } try: mock_app.database.messages.upsert_lxmf_message(message_data) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( audio_bytes=st.binary(min_size=0, max_size=5000), - image_bytes=st.binary(min_size=0, max_size=10000) + image_bytes=st.binary(min_size=0, max_size=10000), ) def test_lxmf_field_fuzzing(audio_bytes, image_bytes): """Fuzz the LXMF field helper classes.""" try: LxmfAudioField(audio_mode=random.randint(0, 10), audio_bytes=audio_bytes) - LxmfImageField(image_type=random.choice(["png", "jpg", "webp", "invalid"]), image_bytes=image_bytes) + LxmfImageField( + image_type=random.choice(["png", "jpg", "webp", "invalid"]), + image_bytes=image_bytes, + ) LxmfFileAttachment(file_name="test.txt", file_bytes=audio_bytes) except Exception as e: pytest.fail(f"LXMF field classes crashed: {e}") + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) -@given( - command_bytes=st.binary(min_size=1, max_size=100) -) +@given(command_bytes=st.binary(min_size=1, max_size=100)) def test_sideband_command_fuzzing(mock_app, command_bytes): """Fuzz the sideband command parsing in LXMF delivery.""" mock_message = MagicMock() @@ -419,17 +469,18 @@ def test_sideband_command_fuzzing(mock_app, command_bytes): mock_message.hash = os.urandom(16) # 0x01 is SidebandCommands.TELEMETRY_REQUEST mock_message.get_fields.return_value = {LXMF.FIELD_COMMANDS: [command_bytes]} - + try: mock_app.on_lxmf_delivery(mock_message) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( destination_hash=st.text(min_size=0, max_size=100), page_path=st.text(min_size=0, max_size=500), - content=st.text(min_size=0, max_size=10000) + content=st.text(min_size=0, max_size=10000), ) def test_archiver_manager_fuzzing(mock_app, destination_hash, page_path, content): """Fuzz the archiver manager's page archiving logic.""" @@ -438,10 +489,9 @@ def test_archiver_manager_fuzzing(mock_app, destination_hash, page_path, content except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) -@given( - state=st.integers(min_value=-10, max_value=30) -) +@given(state=st.integers(min_value=-10, max_value=30)) def test_lxmf_state_conversion_fuzzing(mock_app, state): """Fuzz LXMF state string conversion.""" mock_message = MagicMock() @@ -451,10 +501,9 @@ def test_lxmf_state_conversion_fuzzing(mock_app, state): except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) -@given( - method=st.integers(min_value=-10, max_value=30) -) +@given(method=st.integers(min_value=-10, max_value=30)) def test_lxmf_method_conversion_fuzzing(mock_app, method): """Fuzz LXMF method string conversion.""" mock_message = MagicMock() @@ -464,6 +513,7 @@ def test_lxmf_method_conversion_fuzzing(mock_app, method): except Exception: pass + def test_telephone_announce_fuzzing(mock_app): """Fuzz telephone announce reception.""" aspect = "telephone.call" @@ -472,14 +522,10 @@ def test_telephone_announce_fuzzing(mock_app): announced_identity.hash = os.urandom(32) app_data = b"test_app_data" announce_packet_hash = os.urandom(16) - + try: mock_app.on_telephone_announce_received( - aspect, - destination_hash, - announced_identity, - app_data, - announce_packet_hash + aspect, destination_hash, announced_identity, app_data, announce_packet_hash, ) except Exception: pass diff --git a/tests/backend/test_meshchat_utils.py b/tests/backend/test_meshchat_utils.py index 97fb3d6..380b653 100644 --- a/tests/backend/test_meshchat_utils.py +++ b/tests/backend/test_meshchat_utils.py @@ -38,7 +38,9 @@ def mock_app(temp_dir): patch("RNS.Transport"), patch("threading.Thread"), patch.object( - ReticulumMeshChat, "announce_loop", new=MagicMock(return_value=None), + ReticulumMeshChat, + "announce_loop", + new=MagicMock(return_value=None), ), patch.object( ReticulumMeshChat, @@ -46,7 +48,9 @@ def mock_app(temp_dir): new=MagicMock(return_value=None), ), patch.object( - ReticulumMeshChat, "crawler_loop", new=MagicMock(return_value=None), + ReticulumMeshChat, + "crawler_loop", + new=MagicMock(return_value=None), ), ): mock_id = MagicMock() diff --git a/tests/backend/test_rns_lifecycle.py b/tests/backend/test_rns_lifecycle.py index afdb18e..39bcee2 100644 --- a/tests/backend/test_rns_lifecycle.py +++ b/tests/backend/test_rns_lifecycle.py @@ -16,7 +16,9 @@ def mock_rns(): patch("RNS.Identity") as mock_identity, patch("threading.Thread"), patch.object( - ReticulumMeshChat, "announce_loop", new=MagicMock(return_value=None), + ReticulumMeshChat, + "announce_loop", + new=MagicMock(return_value=None), ), patch.object( ReticulumMeshChat, @@ -24,7 +26,9 @@ def mock_rns(): new=MagicMock(return_value=None), ), patch.object( - ReticulumMeshChat, "crawler_loop", new=MagicMock(return_value=None), + ReticulumMeshChat, + "crawler_loop", + new=MagicMock(return_value=None), ), ): # Setup mock identity @@ -215,7 +219,9 @@ async def test_reload_reticulum_failure_recovery(mock_rns, temp_dir): # We need to make something else fail to reach the except block # or just mock a method inside the try block to raise. with patch.object( - app, "teardown_identity", side_effect=Exception("Reload failed"), + app, + "teardown_identity", + side_effect=Exception("Reload failed"), ): result = await app.reload_reticulum() diff --git a/tests/backend/test_security_fuzzing.py b/tests/backend/test_security_fuzzing.py index 2e2d5c5..ea4e23a 100644 --- a/tests/backend/test_security_fuzzing.py +++ b/tests/backend/test_security_fuzzing.py @@ -1,10 +1,14 @@ -import pytest import os -from unittest.mock import MagicMock, patch -from hypothesis import given, strategies as st, settings, HealthCheck -import LXMF -from meshchatx.meshchat import ReticulumMeshChat from contextlib import ExitStack +from unittest.mock import MagicMock, patch + +import LXMF +import pytest +from hypothesis import HealthCheck, given, settings +from hypothesis import strategies as st + +from meshchatx.meshchat import ReticulumMeshChat + @pytest.fixture def mock_app(): @@ -28,24 +32,32 @@ def mock_app(): stack.enter_context(patch("RNS.Reticulum")) stack.enter_context(patch("RNS.Transport")) stack.enter_context(patch("threading.Thread")) - stack.enter_context(patch.object(ReticulumMeshChat, "announce_loop", return_value=None)) - stack.enter_context(patch.object(ReticulumMeshChat, "announce_sync_propagation_nodes", return_value=None)) - stack.enter_context(patch.object(ReticulumMeshChat, "crawler_loop", return_value=None)) - + stack.enter_context( + patch.object(ReticulumMeshChat, "announce_loop", return_value=None), + ) + stack.enter_context( + patch.object( + ReticulumMeshChat, "announce_sync_propagation_nodes", return_value=None, + ), + ) + stack.enter_context( + patch.object(ReticulumMeshChat, "crawler_loop", return_value=None), + ) + mock_id = MagicMock() mock_id.hash = b"test_hash_32_bytes_long_01234567" mock_id.get_private_key.return_value = b"test_private_key" mock_identity_class.return_value = mock_id - + # Make run_async a no-op that doesn't trigger coroutine warnings mock_async_utils.run_async = MagicMock(side_effect=lambda coroutine: None) - + app = ReticulumMeshChat( identity=mock_id, storage_dir="/tmp/meshchat_test", reticulum_config_dir="/tmp/meshchat_test", ) - + # Setup config mock to return real values to avoid background thread issues app.config = MagicMock() app.config.auto_announce_enabled.get.return_value = False @@ -58,7 +70,7 @@ def mock_app(): app.config.voicemail_auto_answer_delay_seconds.get.return_value = 0 app.config.voicemail_greeting.get.return_value = "Hello" app.config.voicemail_max_recording_seconds.get.return_value = 10 - + # Other required mocks for on_lxmf_delivery app.is_destination_blocked = MagicMock(return_value=False) app.check_spam_keywords = MagicMock(return_value=False) @@ -66,17 +78,39 @@ def mock_app(): app.handle_forwarding = MagicMock() app.update_lxmf_user_icon = MagicMock() app.websocket_broadcast = MagicMock() - + yield app + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( field_data=st.one_of( - st.lists(st.one_of(st.text(), st.binary(), st.integers(), st.floats(), st.booleans(), st.none()), min_size=0, max_size=10), - st.dictionaries(keys=st.text(), values=st.one_of(st.text(), st.binary(), st.integers(), st.floats(), st.booleans(), st.none())), + st.lists( + st.one_of( + st.text(), + st.binary(), + st.integers(), + st.floats(), + st.booleans(), + st.none(), + ), + min_size=0, + max_size=10, + ), + st.dictionaries( + keys=st.text(), + values=st.one_of( + st.text(), + st.binary(), + st.integers(), + st.floats(), + st.booleans(), + st.none(), + ), + ), st.binary(), - st.text() - ) + st.text(), + ), ) def test_lxmf_icon_appearance_fuzzing(mock_app, field_data): """Fuzz LXMF.FIELD_ICON_APPEARANCE parsing in on_lxmf_delivery.""" @@ -84,43 +118,70 @@ def test_lxmf_icon_appearance_fuzzing(mock_app, field_data): mock_message.get_fields.return_value = {LXMF.FIELD_ICON_APPEARANCE: field_data} mock_message.source_hash = os.urandom(16) mock_message.hash = os.urandom(16) - + try: mock_app.on_lxmf_delivery(mock_message) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( attachments_data=st.lists( st.one_of( - st.lists(st.one_of(st.text(), st.binary(), st.integers(), st.floats(), st.booleans(), st.none()), min_size=0, max_size=5), + st.lists( + st.one_of( + st.text(), + st.binary(), + st.integers(), + st.floats(), + st.booleans(), + st.none(), + ), + min_size=0, + max_size=5, + ), st.text(), st.binary(), - st.none() + st.none(), ), - min_size=0, max_size=10 - ) + min_size=0, + max_size=10, + ), ) def test_lxmf_attachments_fuzzing(mock_app, attachments_data): """Fuzz LXMF.FIELD_FILE_ATTACHMENTS parsing.""" mock_message = MagicMock() - mock_message.get_fields.return_value = {LXMF.FIELD_FILE_ATTACHMENTS: attachments_data} + mock_message.get_fields.return_value = { + LXMF.FIELD_FILE_ATTACHMENTS: attachments_data, + } mock_message.source_hash = os.urandom(16) mock_message.hash = os.urandom(16) - + try: mock_app.on_lxmf_delivery(mock_message) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( image_data=st.one_of( - st.lists(st.one_of(st.text(), st.binary(), st.integers(), st.floats(), st.booleans(), st.none()), min_size=0, max_size=5), + st.lists( + st.one_of( + st.text(), + st.binary(), + st.integers(), + st.floats(), + st.booleans(), + st.none(), + ), + min_size=0, + max_size=5, + ), st.binary(), - st.none() - ) + st.none(), + ), ) def test_lxmf_image_field_fuzzing(mock_app, image_data): """Fuzz LXMF.FIELD_IMAGE parsing.""" @@ -128,19 +189,31 @@ def test_lxmf_image_field_fuzzing(mock_app, image_data): mock_message.get_fields.return_value = {LXMF.FIELD_IMAGE: image_data} mock_message.source_hash = os.urandom(16) mock_message.hash = os.urandom(16) - + try: mock_app.on_lxmf_delivery(mock_message) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( audio_data=st.one_of( - st.lists(st.one_of(st.text(), st.binary(), st.integers(), st.floats(), st.booleans(), st.none()), min_size=0, max_size=5), + st.lists( + st.one_of( + st.text(), + st.binary(), + st.integers(), + st.floats(), + st.booleans(), + st.none(), + ), + min_size=0, + max_size=5, + ), st.binary(), - st.none() - ) + st.none(), + ), ) def test_lxmf_audio_field_fuzzing(mock_app, audio_data): """Fuzz LXMF.FIELD_AUDIO parsing.""" @@ -148,48 +221,49 @@ def test_lxmf_audio_field_fuzzing(mock_app, audio_data): mock_message.get_fields.return_value = {LXMF.FIELD_AUDIO: audio_data} mock_message.source_hash = os.urandom(16) mock_message.hash = os.urandom(16) - + try: mock_app.on_lxmf_delivery(mock_message) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( filename=st.text(min_size=0, max_size=1000), - file_bytes=st.binary(min_size=0, max_size=10000) + file_bytes=st.binary(min_size=0, max_size=10000), ) def test_attachment_filename_security(mock_app, filename, file_bytes): """Test for potential directory traversal or malicious filenames in attachments.""" mock_message = MagicMock() mock_message.get_fields.return_value = { - LXMF.FIELD_FILE_ATTACHMENTS: [[filename, file_bytes]] + LXMF.FIELD_FILE_ATTACHMENTS: [[filename, file_bytes]], } mock_message.source_hash = os.urandom(16) mock_message.hash = os.urandom(16) - + try: mock_app.on_lxmf_delivery(mock_message) mock_app.convert_lxmf_message_to_dict(mock_message) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) -@given( - caller_id_bytes=st.binary(min_size=0, max_size=1000) -) +@given(caller_id_bytes=st.binary(min_size=0, max_size=1000)) def test_telephone_callback_fuzzing(mock_app, caller_id_bytes): """Fuzz telephone manager callbacks with malformed identity bytes.""" try: mock_identity = MagicMock() mock_identity.hash = caller_id_bytes - + mock_app.telephone_manager.on_telephone_ringing(mock_identity) mock_app.telephone_manager.on_telephone_call_established(mock_identity) mock_app.telephone_manager.on_telephone_call_ended(mock_identity) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( data=st.dictionaries( @@ -200,9 +274,9 @@ def test_telephone_callback_fuzzing(mock_app, caller_id_bytes): st.integers(), st.floats(), st.lists(st.text()), - st.dictionaries(keys=st.text(), values=st.text()) - ) - ) + st.dictionaries(keys=st.text(), values=st.text()), + ), + ), ) def test_message_dao_upsert_fuzzing(mock_app, data): """Fuzz MessageDAO.upsert_lxmf_message with varied dictionary data.""" @@ -211,10 +285,11 @@ def test_message_dao_upsert_fuzzing(mock_app, data): except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( title_bytes=st.binary(min_size=0, max_size=1000), - content_bytes=st.binary(min_size=0, max_size=5000) + content_bytes=st.binary(min_size=0, max_size=5000), ) def test_lxmf_message_decoding_fuzzing(mock_app, title_bytes, content_bytes): """Fuzz LXMF message title and content decoding.""" @@ -233,95 +308,110 @@ def test_lxmf_message_decoding_fuzzing(mock_app, title_bytes, content_bytes): mock_message.snr = 10 mock_message.q = 100 mock_message.get_fields.return_value = {} - + try: mock_app.convert_lxmf_message_to_dict(mock_message) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) -@given( - greeting_text=st.text(min_size=0, max_size=1000) -) +@given(greeting_text=st.text(min_size=0, max_size=1000)) def test_voicemail_greeting_fuzzing(mock_app, greeting_text): """Fuzz voicemail greeting generation with varied text.""" mock_app.voicemail_manager.has_espeak = True mock_app.voicemail_manager.has_ffmpeg = True mock_app.voicemail_manager.espeak_path = "/usr/bin/espeak" mock_app.voicemail_manager.ffmpeg_path = "/usr/bin/ffmpeg" - + with patch("subprocess.run") as mock_run: try: mock_app.voicemail_manager.generate_greeting(greeting_text) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) -@given( - caller_hash=st.binary(min_size=0, max_size=32) -) +@given(caller_hash=st.binary(min_size=0, max_size=32)) def test_voicemail_incoming_call_fuzzing(mock_app, caller_hash): """Fuzz voicemail incoming call handling.""" mock_identity = MagicMock() mock_identity.hash = caller_hash - + try: mock_app.voicemail_manager.handle_incoming_call(mock_identity) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( source_hash=st.text(min_size=0, max_size=64), recipient_hash=st.text(min_size=0, max_size=64), - dest_hash=st.text(min_size=0, max_size=64) + dest_hash=st.text(min_size=0, max_size=64), ) -def test_forwarding_manager_mapping_fuzzing(mock_app, source_hash, recipient_hash, dest_hash): +def test_forwarding_manager_mapping_fuzzing( + mock_app, source_hash, recipient_hash, dest_hash, +): """Fuzz forwarding manager mapping creation.""" try: - mock_app.forwarding_manager.get_or_create_mapping(source_hash, recipient_hash, dest_hash) + mock_app.forwarding_manager.get_or_create_mapping( + source_hash, recipient_hash, dest_hash, + ) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) -@given( - uri=st.text(min_size=0, max_size=5000) -) +@given(uri=st.text(min_size=0, max_size=5000)) def test_lxm_ingest_uri_fuzzing(mock_app, uri): """Fuzz the lxm.ingest_uri WebSocket handler.""" mock_client = MagicMock() mock_client.send_str = MagicMock() - + try: # We need to wrap it in a task since it's async import asyncio + loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) - loop.run_until_complete(mock_app.on_websocket_data_received(mock_client, {"type": "lxm.ingest_uri", "uri": uri})) + loop.run_until_complete( + mock_app.on_websocket_data_received( + mock_client, {"type": "lxm.ingest_uri", "uri": uri}, + ), + ) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( config_data=st.dictionaries( keys=st.text(), - values=st.one_of(st.text(), st.integers(), st.booleans(), st.none(), st.lists(st.text()), st.dictionaries(keys=st.text(), values=st.text())) - ) + values=st.one_of( + st.text(), + st.integers(), + st.booleans(), + st.none(), + st.lists(st.text()), + st.dictionaries(keys=st.text(), values=st.text()), + ), + ), ) def test_update_config_fuzzing(mock_app, config_data): """Fuzz the update_config method with randomized dictionary data.""" try: import asyncio + loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_until_complete(mock_app.update_config(config_data)) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) -@given( - large_string=st.text(min_size=1000, max_size=10000) -) +@given(large_string=st.text(min_size=1000, max_size=10000)) def test_large_payload_dos_resistance(mock_app, large_string): """Check resistance to DoS via large strings in various fields.""" mock_message = MagicMock() @@ -330,56 +420,67 @@ def test_large_payload_dos_resistance(mock_app, large_string): mock_message.hash = os.urandom(16) mock_message.source_hash = os.urandom(16) mock_message.get_fields.return_value = {} - + try: mock_app.on_lxmf_delivery(mock_message) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( nested_data=st.recursive( st.one_of(st.text(), st.integers()), lambda children: st.dictionaries(st.text(), children) | st.lists(children), - max_leaves=100 - ) + max_leaves=100, + ), ) def test_websocket_recursion_fuzzing(mock_app, nested_data): """Fuzz the WebSocket handler with deeply nested JSON data.""" mock_client = MagicMock() try: import asyncio + loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) - loop.run_until_complete(mock_app.on_websocket_data_received(mock_client, {"type": "ping", "data": nested_data})) + loop.run_until_complete( + mock_app.on_websocket_data_received( + mock_client, {"type": "ping", "data": nested_data}, + ), + ) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) -@given( - dest_hash=st.text(), - content=st.text() -) +@given(dest_hash=st.text(), content=st.text()) def test_lxm_generate_paper_uri_fuzzing(mock_app, dest_hash, content): """Fuzz paper URI generation with randomized inputs.""" mock_client = MagicMock() try: import asyncio + loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) - loop.run_until_complete(mock_app.on_websocket_data_received(mock_client, { - "type": "lxm.generate_paper_uri", - "destination_hash": dest_hash, - "content": content - })) + loop.run_until_complete( + mock_app.on_websocket_data_received( + mock_client, + { + "type": "lxm.generate_paper_uri", + "destination_hash": dest_hash, + "content": content, + }, + ), + ) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( lon=st.floats(allow_nan=False, allow_infinity=False), lat=st.floats(allow_nan=False, allow_infinity=False), - zoom=st.integers(min_value=-100, max_value=100) + zoom=st.integers(min_value=-100, max_value=100), ) def test_map_manager_coord_fuzzing(mock_app, lon, lat, zoom): """Fuzz coordinate to tile conversion in MapManager.""" @@ -388,11 +489,12 @@ def test_map_manager_coord_fuzzing(mock_app, lon, lat, zoom): except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @given( text=st.text(), source_lang=st.text(min_size=0, max_size=10), - target_lang=st.text(min_size=0, max_size=10) + target_lang=st.text(min_size=0, max_size=10), ) def test_translator_handler_fuzzing(mock_app, text, source_lang, target_lang): """Fuzz the TranslatorHandler translate_text method.""" @@ -404,28 +506,26 @@ def test_translator_handler_fuzzing(mock_app, text, source_lang, target_lang): except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) -@given( - dest_hash=st.text(), - icon_name=st.text(), - fg_color=st.text(), - bg_color=st.text() -) -def test_update_lxmf_user_icon_fuzzing(mock_app, dest_hash, icon_name, fg_color, bg_color): +@given(dest_hash=st.text(), icon_name=st.text(), fg_color=st.text(), bg_color=st.text()) +def test_update_lxmf_user_icon_fuzzing( + mock_app, dest_hash, icon_name, fg_color, bg_color, +): """Fuzz user icon update logic with malformed strings.""" try: mock_app.update_lxmf_user_icon(dest_hash, icon_name, fg_color, bg_color) except Exception: pass + @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) -@given( - binary_data=st.binary(min_size=0, max_size=10000) -) +@given(binary_data=st.binary(min_size=0, max_size=10000)) def test_rns_identity_load_fuzzing(mock_app, binary_data): """Fuzz RNS.Identity loading with random binary data.""" try: import RNS + try: RNS.Identity.from_bytes(binary_data) except Exception: