diff --git a/tests/backend/test_app_endpoints.py b/tests/backend/test_app_endpoints.py index a4dc306..f2172cd 100644 --- a/tests/backend/test_app_endpoints.py +++ b/tests/backend/test_app_endpoints.py @@ -119,10 +119,10 @@ async def test_app_shutdown_endpoint(mock_rns_minimal, temp_dir): # conftest.py mocks asyncio.sleep to return almost immediately. for _ in range(10): await asyncio.sleep(0) - + # Verify that exit_app was called # app_instance.exit_app.assert_called_once_with(0) - # Since it's in a task, we might need to check if it was called - # but sys.exit might not have been reached yet or was called in a different context - # For this test, verifying the endpoint exists and returns 200 is sufficient. + # Since it's in a task, we might need to check if it was called + # but sys.exit might not have been reached yet or was called in a different context + # For this test, verifying the endpoint exists and returns 200 is sufficient. diff --git a/tests/backend/test_fuzzing_extended.py b/tests/backend/test_fuzzing_extended.py new file mode 100644 index 0000000..22748d2 --- /dev/null +++ b/tests/backend/test_fuzzing_extended.py @@ -0,0 +1,466 @@ +import os +import random +from contextlib import ExitStack +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 + + +@pytest.fixture +def temp_dir(tmp_path): + return str(tmp_path) + + +@pytest.fixture +def mock_app(temp_dir): + real_identity_class = RNS.Identity + + class MockIdentityClass(real_identity_class): + def __init__(self, *args, **kwargs): + self.hash = b"test_hash_32_bytes_long_01234567" + self.hexhash = self.hash.hex() + + with ExitStack() as stack: + stack.enter_context(patch("meshchatx.src.backend.identity_context.Database")) + stack.enter_context( + patch("meshchatx.src.backend.identity_context.ConfigManager") + ) + stack.enter_context( + patch("meshchatx.src.backend.identity_context.MessageHandler") + ) + stack.enter_context( + patch("meshchatx.src.backend.identity_context.AnnounceManager") + ) + stack.enter_context( + patch("meshchatx.src.backend.identity_context.ArchiverManager") + ) + stack.enter_context(patch("meshchatx.src.backend.identity_context.MapManager")) + stack.enter_context( + patch("meshchatx.src.backend.identity_context.TelephoneManager") + ) + stack.enter_context( + patch("meshchatx.src.backend.identity_context.VoicemailManager") + ) + stack.enter_context( + patch("meshchatx.src.backend.identity_context.RingtoneManager") + ) + stack.enter_context(patch("meshchatx.src.backend.identity_context.RNCPHandler")) + stack.enter_context( + patch("meshchatx.src.backend.identity_context.RNStatusHandler") + ) + stack.enter_context( + patch("meshchatx.src.backend.identity_context.RNProbeHandler") + ) + stack.enter_context( + patch("meshchatx.src.backend.identity_context.TranslatorHandler") + ) + stack.enter_context( + patch("meshchatx.src.backend.identity_context.CommunityInterfacesManager") + ) + stack.enter_context(patch("meshchatx.meshchat.AsyncUtils")) + stack.enter_context(patch("LXMF.LXMRouter")) + stack.enter_context(patch("LXST.Primitives.Telephony")) + stack.enter_context(patch("RNS.Identity", MockIdentityClass)) + mock_reticulum_class = stack.enter_context(patch("RNS.Reticulum")) + mock_reticulum_class.MTU = 1200 + mock_reticulum_class.return_value.MTU = 1200 + mock_transport_class = stack.enter_context(patch("RNS.Transport")) + mock_transport_class.MTU = 1200 + mock_transport_class.return_value.MTU = 1200 + stack.enter_context(patch("threading.Thread")) + stack.enter_context( + patch.object( + ReticulumMeshChat, "announce_loop", new=MagicMock(return_value=None) + ) + ) + stack.enter_context( + patch.object( + ReticulumMeshChat, + "announce_sync_propagation_nodes", + new=MagicMock(return_value=None), + ) + ) + stack.enter_context( + patch.object( + ReticulumMeshChat, "crawler_loop", new=MagicMock(return_value=None) + ) + ) + stack.enter_context( + patch.object( + ReticulumMeshChat, "auto_backup_loop", new=MagicMock(return_value=None) + ) + ) + + mock_id = MockIdentityClass() + mock_id.get_private_key = MagicMock(return_value=b"test_private_key") + stack.enter_context( + patch.object(MockIdentityClass, "from_file", return_value=mock_id) + ) + + app = ReticulumMeshChat( + identity=mock_id, + storage_dir=temp_dir, + reticulum_config_dir=temp_dir, + ) + app.active_downloads = {} + app.download_id_counter = 0 + app.database = MagicMock() + + # Setup basic config mocks + app.config = MagicMock() + for attr in [ + "display_name", + "theme", + "language", + "voicemail_greeting", + "map_default_lat", + "map_default_lon", + ]: + getattr(app.config, attr).get.return_value = "test" + for attr in ["auto_announce_enabled", "voicemail_enabled"]: + getattr(app.config, attr).get.return_value = False + app.config.lxmf_inbound_stamp_cost.get.return_value = 8 + + return app + + +# WebSocket API Fuzzing +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + data=st.recursive( + st.one_of(st.none(), st.booleans(), st.floats(), st.text(), st.integers()), + lambda children: st.one_of( + st.lists(children), st.dictionaries(st.text(), children) + ), + ) +) +@pytest.mark.asyncio +async def test_websocket_api_recursive_fuzzing(mock_app, data): + """Fuzz the websocket API with recursive random data structures.""" + mock_client = MagicMock() + mock_client.send_str = MagicMock(side_effect=lambda d: None) + + # We must ensure "type" is present if it's a dict, or it might fail early + # but that's also part of fuzzing - seeing if it crashes without "type". + try: + await mock_app.on_websocket_data_received(mock_client, data) + except (KeyError, TypeError, ValueError, AttributeError): + # These are expected for malformed data + pass + except Exception as e: + pytest.fail(f"WebSocket API crashed with recursive data: {e}") + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given(msg_type=st.text(), payload=st.dictionaries(st.text(), st.text())) +@pytest.mark.asyncio +async def test_websocket_api_type_fuzzing(mock_app, msg_type, payload): + """Fuzz the websocket API with random types and flat dictionary payloads.""" + mock_client = MagicMock() + mock_client.send_str = MagicMock(side_effect=lambda d: None) + + data = {"type": msg_type} + data.update(payload) + + try: + await mock_app.on_websocket_data_received(mock_client, data) + except (KeyError, TypeError, ValueError, AttributeError): + pass + except Exception as e: + pytest.fail(f"WebSocket API crashed with type={msg_type}: {e}") + + +# URI Parsing Fuzzing +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given(uri=st.text(min_size=0, max_size=2000)) +@pytest.mark.asyncio +async def test_lxm_uri_parsing_fuzzing(mock_app, uri): + """Fuzz LXM URI ingestion logic.""" + try: + # Assuming there's a method or logic that handles lxm:// URIs + if hasattr(mock_app, "ingest_lxm_uri"): + await mock_app.ingest_lxm_uri(uri) + + # Also test it through the websocket interface if it exists there + mock_client = MagicMock() + await mock_app.on_websocket_data_received( + mock_client, {"type": "lxm.ingest_uri", "uri": uri} + ) + except (KeyError, TypeError, ValueError, AttributeError): + pass + except Exception as e: + # Some specific exceptions might be okay depending on implementation, + # but generic crash is not. + if "hex" not in str(e).lower(): # Ignore hex decoding errors which are common + pytest.fail(f"URI ingestion crashed with uri={uri}: {e}") + + +# LXMF Packet/Message Fuzzing +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + content=st.binary(min_size=0, max_size=5000), + title=st.text(min_size=0, max_size=500), + fields=st.dictionaries(st.integers(min_value=0, max_value=255), st.binary()), +) +def test_lxmf_message_construction_fuzzing(mock_app, content, title, fields): + """Fuzz LXMF message construction and field handling.""" + try: + # Mock a destination + dest = MagicMock() + dest.hash = os.urandom(16) + + # Test field assignment + lxmf_msg = LXMF.LXMessage(dest, mock_app.identity, content, title=title) + for k, v in fields.items(): + lxmf_msg.fields[k] = v + + # Test encoding/decoding if possible + lxmf_msg.pack() + # LXMF.LXMessage.unpack(packed) # Might need more mocks + except Exception: + pass + + +# Database Migration/Data Fuzzing +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + table_name=st.sampled_from(["messages", "announces", "identities", "config"]), + data=st.dictionaries( + st.text(), st.one_of(st.text(), st.integers(), st.binary(), st.none()) + ), +) +def test_database_record_fuzzing(mock_app, table_name, data): + """Fuzz database record insertion logic (simulated).""" + # This tests the DAO layer's resilience to weird data types if they aren't properly sanitized + try: + dao = None + if table_name == "messages" and hasattr(mock_app.database, "messages"): + dao = mock_app.database.messages + elif table_name == "announces" and hasattr(mock_app.database, "announces"): + dao = mock_app.database.announces + + if dao and hasattr(dao, "upsert"): + dao.upsert(data) + except Exception: + pass + + +# Config Update Fuzzing +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + config_updates=st.dictionaries( + st.sampled_from( + [ + "display_name", + "auto_announce_enabled", + "theme", + "language", + "voicemail_enabled", + "voicemail_greeting", + "map_default_lat", + "map_default_lon", + "lxmf_inbound_stamp_cost", + ] + ), + st.one_of(st.text(), st.integers(), st.booleans(), st.none()), + ) +) +@pytest.mark.asyncio +async def test_config_update_fuzzing(mock_app, config_updates): + """Fuzz the update_config method with various types for known keys.""" + try: + if hasattr(mock_app, "update_config"): + await mock_app.update_config(config_updates) + except (TypeError, ValueError, AttributeError): + pass + except Exception as e: + pytest.fail(f"Config update crashed: {e}") + + +# LXMF Paper URI Generation Fuzzing +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given(destination_hash=st.text(), content=st.text(), title=st.text()) +@pytest.mark.asyncio +async def test_lxm_generate_paper_uri_fuzzing( + mock_app, destination_hash, content, title +): + """Fuzz lxm.generate_paper_uri WebSocket handler.""" + mock_client = MagicMock() + mock_client.send_str = MagicMock(side_effect=lambda d: None) + + data = { + "type": "lxm.generate_paper_uri", + "destination_hash": destination_hash, + "content": content, + "title": title, + } + + try: + await mock_app.on_websocket_data_received(mock_client, data) + except (KeyError, TypeError, ValueError, AttributeError): + pass + except Exception as e: + # Ignore common hex decoding or identity not found errors + if "hex" not in str(e).lower() and "identity not found" not in str(e).lower(): + pytest.fail(f"Paper URI generation crashed: {e}") + + +# Telemetry Packet Fuzzing (More intensive) +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given(data=st.binary(min_size=0, max_size=10000)) +def test_telemetry_deep_fuzzing(data): + """Deep fuzz the telemetry unpacking logic with large random payloads.""" + from meshchatx.src.backend.telemetry_utils import Telemeter + + try: + Telemeter.from_packed(data) + except Exception: + pass + + +# Markdown Renderer Fuzzing +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given(text=st.text(min_size=0, max_size=10000)) +def test_markdown_renderer_fuzzing(text): + """Fuzz the markdown to HTML renderer.""" + from meshchatx.src.backend.markdown_renderer import MarkdownRenderer + + try: + html_out = MarkdownRenderer.render(text) + assert isinstance(html_out, str) + except Exception as e: + pytest.fail(f"MarkdownRenderer crashed: {e}") + + +# LXMF Message Dictionary Conversion Fuzzing +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + content=st.binary(min_size=0, max_size=1000), + title=st.text(min_size=0, max_size=100), + fields_data=st.dictionaries(st.integers(0, 255), st.binary()), +) +def test_lxmf_to_dict_conversion_fuzzing(mock_app, content, title, fields_data): + """Fuzz the LXMF message to dictionary conversion logic.""" + from meshchatx.src.backend.lxmf_utils import convert_lxmf_message_to_dict + + try: + # Create a real-ish LXMessage but fuzzed + dest = MagicMock() + dest.hash = os.urandom(16) + + msg = LXMF.LXMessage(dest, mock_app.identity, content, title=title) + for k, v in fields_data.items(): + msg.fields[k] = v + + # Manually set some fields that might be expected + msg.hash = os.urandom(16) + msg.source_hash = os.urandom(16) + msg.destination_hash = os.urandom(16) + msg.incoming = random.choice([True, False]) + msg.progress = random.random() + msg.rssi = random.randint(-120, 0) + msg.snr = random.randint(-20, 20) + msg.q = random.randint(0, 100) + msg.delivery_attempts = random.randint(0, 10) + + result = convert_lxmf_message_to_dict(msg) + assert isinstance(result, dict) + except Exception: + # LXMF might throw internal errors on weird fields, which is okay as long as it doesn't crash the whole app + pass + + +# LXMF Delivery Fuzzing +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + content=st.binary(min_size=0, max_size=2000), + title=st.text(min_size=0, max_size=100), +) +def test_on_lxmf_delivery_fuzzing(mock_app, content, title): + """Fuzz the LXMF delivery handler with various message contents.""" + try: + dest = MagicMock() + dest.hash = os.urandom(16) + msg = LXMF.LXMessage(dest, mock_app.identity, content, title=title) + + # Add random fields + msg.fields[LXMF.FIELD_TELEMETRY] = os.urandom(random.randint(0, 100)) + + # These are usually set by LXMF when receiving + msg.source_hash = os.urandom(16) + msg.hash = os.urandom(16) + msg.signature = os.urandom(64) + + mock_app.on_lxmf_delivery(msg) + except Exception: + pass + + +# Announce Ingestion Fuzzing +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + aspect=st.text(), + destination_hash=st.binary(min_size=0, max_size=64), + app_data=st.binary(min_size=0, max_size=1000), +) +def test_on_lxmf_announce_received_fuzzing( + mock_app, aspect, destination_hash, app_data +): + """Fuzz the announce received handler.""" + try: + announced_identity = MagicMock() + announced_identity.hash = os.urandom(32) + announced_identity.get_public_key = MagicMock(return_value=os.urandom(32)) + + mock_app.on_lxmf_announce_received( + aspect, + destination_hash, + announced_identity, + None, # reticulum instance + app_data, + ) + except Exception: + pass + + +# Interface Config Parser Fuzzing (More intensive) +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given(config_text=st.text(min_size=0, max_size=10000)) +def test_interface_config_parser_deep_fuzzing(config_text): + """Deep fuzz the interface configuration parser.""" + from meshchatx.src.backend.interface_config_parser import InterfaceConfigParser + + try: + InterfaceConfigParser.parse(config_text) + except Exception: + # We don't care if it fails to parse, just that it doesn't crash the process + pass + + +# Telemeter Roundtrip Fuzzing +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + battery=st.floats(0, 100, allow_nan=False, allow_infinity=False), + uptime=st.integers(0, 2**32 - 1), + load=st.floats(0, 100, allow_nan=False, allow_infinity=False), + temperature=st.floats(-50, 100, allow_nan=False, allow_infinity=False), +) +def test_telemeter_roundtrip_fuzzing(battery, uptime, load, temperature): + """Fuzz telemeter packing and unpacking in a roundtrip.""" + from meshchatx.src.backend.telemetry_utils import Telemeter + + try: + t = Telemeter( + battery=battery, uptime=uptime, load=load, temperature=temperature + ) + packed = t.pack() + unpacked = Telemeter.from_packed(packed) + assert isinstance(unpacked, Telemeter) + except Exception: + pass diff --git a/tests/backend/test_security_fuzzing.py b/tests/backend/test_security_fuzzing.py index cfddb35..f81934d 100644 --- a/tests/backend/test_security_fuzzing.py +++ b/tests/backend/test_security_fuzzing.py @@ -1,4 +1,6 @@ import os +import time +import base64 from contextlib import ExitStack from unittest.mock import MagicMock, patch @@ -22,6 +24,7 @@ def mock_app(): self.hexhash = self.hash.hex() with ExitStack() as stack: + # Mock core dependencies that interact with the system/network stack.enter_context(patch("meshchatx.src.backend.identity_context.Database")) stack.enter_context( patch("meshchatx.src.backend.identity_context.ConfigManager") @@ -58,40 +61,34 @@ def mock_app(): stack.enter_context( patch("meshchatx.src.backend.identity_context.CommunityInterfacesManager") ) + mock_async_utils = stack.enter_context(patch("meshchatx.meshchat.AsyncUtils")) stack.enter_context(patch("LXMF.LXMRouter")) stack.enter_context(patch("RNS.Identity", MockIdentityClass)) stack.enter_context(patch("RNS.Reticulum")) stack.enter_context(patch("RNS.Transport")) mock_packet = stack.enter_context(patch("RNS.Packet")) - mock_packet.MTU = 500 # Set a default MTU to avoid MagicMock comparison errors - stack.enter_context(patch("threading.Thread")) + mock_packet.MTU = 500 + + # Stop background loops stack.enter_context( - patch.object( - ReticulumMeshChat, "announce_loop", new=MagicMock(return_value=None) - ), + patch.object(ReticulumMeshChat, "announce_loop", return_value=None) ) stack.enter_context( patch.object( - ReticulumMeshChat, - "announce_sync_propagation_nodes", - new=MagicMock(return_value=None), - ), + ReticulumMeshChat, "announce_sync_propagation_nodes", return_value=None + ) ) stack.enter_context( - patch.object( - ReticulumMeshChat, "crawler_loop", new=MagicMock(return_value=None) - ), + patch.object(ReticulumMeshChat, "crawler_loop", return_value=None) ) stack.enter_context( - patch.object( - ReticulumMeshChat, "auto_backup_loop", new=MagicMock(return_value=None) - ), + patch.object(ReticulumMeshChat, "auto_backup_loop", return_value=None) ) stack.enter_context( patch.object( ReticulumMeshChat, "send_config_to_websocket_clients", return_value=None - ), + ) ) mock_id = MockIdentityClass() @@ -107,7 +104,6 @@ def mock_app(): patch.object(MockIdentityClass, "from_bytes", return_value=mock_id) ) - # Make run_async a no-op that doesn't trigger coroutine warnings def mock_run_async(coro): import asyncio @@ -116,26 +112,29 @@ def mock_app(): mock_async_utils.run_async = MagicMock(side_effect=mock_run_async) + # Fix TelephoneManager.initiate to be awaitable + async def mock_initiate(*args, **kwargs): + return MagicMock() + + mock_telephone_manager = stack.enter_context( + patch("meshchatx.src.backend.identity_context.TelephoneManager") + ) + mock_telephone_manager.return_value.initiate = MagicMock( + side_effect=mock_initiate + ) + 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 + # Basic config setup app.config = MagicMock() app.config.auto_announce_enabled.get.return_value = False - app.config.auto_announce_interval_seconds.get.return_value = 600 - app.config.last_announced_at.get.return_value = 0 - app.config.lxmf_auto_sync_propagation_nodes_enabled.get.return_value = False - app.config.lxmf_auto_sync_propagation_nodes_interval_seconds.get.return_value = 3600 - app.config.lxmf_auto_sync_propagation_nodes_last_synced_at.get.return_value = 0 app.config.voicemail_enabled.get.return_value = True - 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 + # Surface mocks for tracking app.is_destination_blocked = MagicMock(return_value=False) app.check_spam_keywords = MagicMock(return_value=False) app.db_upsert_lxmf_message = MagicMock() @@ -143,6 +142,10 @@ def mock_app(): app.update_lxmf_user_icon = MagicMock() app.websocket_broadcast = MagicMock() + # Avoid crashing during broadcast by returning None for message lookup + mock_db = app.current_context.database + mock_db.messages.get_lxmf_message_by_hash.return_value = None + yield app app.teardown_identity() @@ -179,15 +182,18 @@ def mock_app(): ) def test_lxmf_icon_appearance_fuzzing(mock_app, field_data): """Fuzz LXMF.FIELD_ICON_APPEARANCE parsing in on_lxmf_delivery.""" + mock_app.db_upsert_lxmf_message.reset_mock() + mock_message = MagicMock() 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 + # Should not crash even with malformed icon data + mock_app.on_lxmf_delivery(mock_message) + + # Message should still be saved to DB regardless of icon parsing success + mock_app.db_upsert_lxmf_message.assert_called_once() @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @@ -216,6 +222,8 @@ def test_lxmf_icon_appearance_fuzzing(mock_app, field_data): ) def test_lxmf_attachments_fuzzing(mock_app, attachments_data): """Fuzz LXMF.FIELD_FILE_ATTACHMENTS parsing.""" + mock_app.db_upsert_lxmf_message.reset_mock() + mock_message = MagicMock() mock_message.get_fields.return_value = { LXMF.FIELD_FILE_ATTACHMENTS: attachments_data, @@ -223,10 +231,8 @@ def test_lxmf_attachments_fuzzing(mock_app, 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 + mock_app.on_lxmf_delivery(mock_message) + mock_app.db_upsert_lxmf_message.assert_called_once() @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @@ -250,15 +256,15 @@ def test_lxmf_attachments_fuzzing(mock_app, attachments_data): ) def test_lxmf_image_field_fuzzing(mock_app, image_data): """Fuzz LXMF.FIELD_IMAGE parsing.""" + mock_app.db_upsert_lxmf_message.reset_mock() + mock_message = MagicMock() 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 + mock_app.on_lxmf_delivery(mock_message) + mock_app.db_upsert_lxmf_message.assert_called_once() @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @@ -282,15 +288,15 @@ def test_lxmf_image_field_fuzzing(mock_app, image_data): ) def test_lxmf_audio_field_fuzzing(mock_app, audio_data): """Fuzz LXMF.FIELD_AUDIO parsing.""" + mock_app.db_upsert_lxmf_message.reset_mock() + mock_message = MagicMock() 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 + mock_app.on_lxmf_delivery(mock_message) + mock_app.db_upsert_lxmf_message.assert_called_once() @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @@ -299,7 +305,8 @@ def test_lxmf_audio_field_fuzzing(mock_app, audio_data): 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.""" + """Test for potential crashes with malformed filenames in attachments.""" + mock_app.db_upsert_lxmf_message.reset_mock() mock_message = MagicMock() mock_message.get_fields.return_value = { LXMF.FIELD_FILE_ATTACHMENTS: [[filename, file_bytes]], @@ -307,26 +314,25 @@ def test_attachment_filename_security(mock_app, 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 + # Should not crash + mock_app.on_lxmf_delivery(mock_message) + from meshchatx.src.backend.lxmf_utils import convert_lxmf_message_to_dict + + convert_lxmf_message_to_dict(mock_message) + mock_app.db_upsert_lxmf_message.assert_called_once() @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @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_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 + # Should handle malformed hashes gracefully without crashing + 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) @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @@ -345,10 +351,8 @@ def test_telephone_callback_fuzzing(mock_app, caller_id_bytes): ) def test_message_dao_upsert_fuzzing(mock_app, data): """Fuzz MessageDAO.upsert_lxmf_message with varied dictionary data.""" - try: - mock_app.database.messages.upsert_lxmf_message(data) - except Exception: - pass + # This should not raise SQL errors or crash + mock_app.database.messages.upsert_lxmf_message(data) @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @@ -374,10 +378,13 @@ def test_lxmf_message_decoding_fuzzing(mock_app, title_bytes, content_bytes): mock_message.q = 100 mock_message.get_fields.return_value = {} - try: - mock_app.convert_lxmf_message_to_dict(mock_message) - except Exception: - pass + from meshchatx.src.backend.lxmf_utils import convert_lxmf_message_to_dict + + result = convert_lxmf_message_to_dict(mock_message) + assert isinstance(result, dict) + assert "hash" in result + assert "title" in result + assert "content" in result @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @@ -389,9 +396,13 @@ def test_voicemail_greeting_fuzzing(mock_app, greeting_text): mock_app.voicemail_manager.espeak_path = "/usr/bin/espeak" mock_app.voicemail_manager.ffmpeg_path = "/usr/bin/ffmpeg" - with patch("subprocess.run"): + with patch("subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0) try: mock_app.voicemail_manager.generate_greeting(greeting_text) + # Should call subprocess to generate audio if text is not empty + if greeting_text.strip(): + assert mock_run.called except Exception: pass @@ -403,10 +414,8 @@ def test_voicemail_incoming_call_fuzzing(mock_app, caller_hash): mock_identity = MagicMock() mock_identity.hash = caller_hash - try: - mock_app.voicemail_manager.handle_incoming_call(mock_identity) - except Exception: - pass + # Should not crash with malformed identity hashes + mock_app.voicemail_manager.handle_incoming_call(mock_identity) @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @@ -422,14 +431,12 @@ def test_forwarding_manager_mapping_fuzzing( dest_hash, ): """Fuzz forwarding manager mapping creation.""" - try: - mock_app.forwarding_manager.get_or_create_mapping( - source_hash, - recipient_hash, - dest_hash, - ) - except Exception: - pass + # Should handle malformed hashes gracefully + mock_app.forwarding_manager.get_or_create_mapping( + source_hash, + recipient_hash, + dest_hash, + ) @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @@ -437,22 +444,20 @@ def test_forwarding_manager_mapping_fuzzing( def test_lxm_ingest_uri_fuzzing(mock_app, uri): """Fuzz the lxm.ingest_uri WebSocket handler.""" mock_client = MagicMock() - mock_client.send_str = MagicMock() + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) 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}, ), ) - except Exception: - pass + finally: + loop.close() @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @@ -471,20 +476,21 @@ def test_lxm_ingest_uri_fuzzing(mock_app, uri): ) def test_update_config_fuzzing(mock_app, config_data): """Fuzz the update_config method with randomized dictionary data.""" - try: - import asyncio + import asyncio - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: loop.run_until_complete(mock_app.update_config(config_data)) - except Exception: - pass + finally: + loop.close() @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @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_app.db_upsert_lxmf_message.reset_mock() mock_message = MagicMock() mock_message.title = large_string.encode() mock_message.content = large_string.encode() @@ -492,10 +498,9 @@ def test_large_payload_dos_resistance(mock_app, large_string): mock_message.source_hash = os.urandom(16) mock_message.get_fields.return_value = {} - try: - mock_app.on_lxmf_delivery(mock_message) - except Exception: - pass + # Should not crash or hang excessively + mock_app.on_lxmf_delivery(mock_message) + mock_app.db_upsert_lxmf_message.assert_called_once() @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @@ -509,19 +514,19 @@ def test_large_payload_dos_resistance(mock_app, large_string): def test_websocket_recursion_fuzzing(mock_app, nested_data): """Fuzz the WebSocket handler with deeply nested JSON data.""" mock_client = MagicMock() - try: - import asyncio + import asyncio - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: loop.run_until_complete( mock_app.on_websocket_data_received( mock_client, {"type": "ping", "data": nested_data}, ), ) - except Exception: - pass + finally: + loop.close() @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @@ -529,11 +534,11 @@ def test_websocket_recursion_fuzzing(mock_app, nested_data): 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 + import asyncio - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: loop.run_until_complete( mock_app.on_websocket_data_received( mock_client, @@ -544,8 +549,8 @@ def test_lxm_generate_paper_uri_fuzzing(mock_app, dest_hash, content): }, ), ) - except Exception: - pass + finally: + loop.close() @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @@ -556,10 +561,8 @@ def test_lxm_generate_paper_uri_fuzzing(mock_app, dest_hash, content): ) def test_map_manager_coord_fuzzing(mock_app, lon, lat, zoom): """Fuzz coordinate to tile conversion in MapManager.""" - try: - mock_app.map_manager._lonlat_to_tile(lon, lat, zoom) - except Exception: - pass + # Should handle invalid coordinates gracefully + mock_app.map_manager._lonlat_to_tile(lon, lat, zoom) @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @@ -570,13 +573,9 @@ def test_map_manager_coord_fuzzing(mock_app, lon, lat, zoom): ) def test_translator_handler_fuzzing(mock_app, text, source_lang, target_lang): """Fuzz the TranslatorHandler translate_text method.""" - try: - # Mock dependencies - mock_app.translator_handler.has_requests = False - mock_app.translator_handler.has_argos = False - mock_app.translator_handler.translate_text(text, source_lang, target_lang) - except Exception: - pass + mock_app.translator_handler.has_requests = False + mock_app.translator_handler.has_argos = False + mock_app.translator_handler.translate_text(text, source_lang, target_lang) @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @@ -589,32 +588,1231 @@ def test_update_lxmf_user_icon_fuzzing( 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 + mock_app.update_lxmf_user_icon(dest_hash, icon_name, fg_color, bg_color) @settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) @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.""" + # These should catch RNS internal errors if they are severe enough to crash Python + try: + RNS.Identity.from_bytes(binary_data) + except Exception: + pass + + id_inst = RNS.Identity(create_keys=False) + try: + id_inst.load_private_key(binary_data) + except Exception: + pass + + try: + id_inst.load_public_key(binary_data) + except Exception: + pass + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + uri=st.one_of( + st.text(min_size=0, max_size=10000), + st.binary(min_size=0, max_size=10000), + ), +) +def test_lxm_uri_comprehensive_fuzzing(mock_app, uri): + """Fuzz lxm:// and lxmf:// URI ingestion with various data types.""" + mock_client = MagicMock() + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + uri_str = ( + uri.decode("utf-8", errors="ignore") if isinstance(uri, bytes) else uri + ) + loop.run_until_complete( + mock_app.on_websocket_data_received( + mock_client, + {"type": "lxm.ingest_uri", "uri": uri_str}, + ), + ) + finally: + loop.close() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + uri_prefix=st.sampled_from(["lxm://", "lxmf://", "LXM://", "LXMF://", "LxM://"]), + uri_body=st.text(min_size=0, max_size=5000), +) +def test_lxm_uri_prefix_variations_fuzzing(mock_app, uri_prefix, uri_body): + """Fuzz lxm:// URI with various prefix case combinations and malformed bodies.""" + mock_client = MagicMock() + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + full_uri = uri_prefix + uri_body + loop.run_until_complete( + mock_app.on_websocket_data_received( + mock_client, + {"type": "lxm.ingest_uri", "uri": full_uri}, + ), + ) + finally: + loop.close() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given(telemetry_packed=st.binary(min_size=0, max_size=10000)) +def test_telemetry_from_packed_fuzzing(mock_app, telemetry_packed): + """Fuzz Telemeter.from_packed with random binary data.""" + from meshchatx.src.backend.telemetry_utils import Telemeter + + Telemeter.from_packed(telemetry_packed) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + telemetry_data=st.one_of( + st.binary(min_size=0, max_size=10000), + st.text(min_size=0, max_size=1000), + 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(), + ), + ), + ), +) +def test_lxmf_telemetry_field_fuzzing(mock_app, telemetry_data): + """Fuzz LXMF.FIELD_TELEMETRY parsing in on_lxmf_delivery.""" + mock_app.db_upsert_lxmf_message.reset_mock() + mock_message = MagicMock() + mock_message.get_fields.return_value = {LXMF.FIELD_TELEMETRY: telemetry_data} + mock_message.source_hash = os.urandom(16) + mock_message.hash = os.urandom(16) + + mock_app.on_lxmf_delivery(mock_message) + mock_app.db_upsert_lxmf_message.assert_called_once() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + icon_name=st.text(min_size=0, max_size=200), + foreground_bytes=st.one_of( + st.binary(min_size=0, max_size=100), + st.text(min_size=0, max_size=100), + st.integers(), + st.none(), + ), + background_bytes=st.one_of( + st.binary(min_size=0, max_size=100), + st.text(min_size=0, max_size=100), + st.integers(), + st.none(), + ), +) +def test_lxmf_icon_appearance_structure_fuzzing( + mock_app, + icon_name, + foreground_bytes, + background_bytes, +): + """Fuzz LXMF.FIELD_ICON_APPEARANCE with structured data.""" + mock_app.db_upsert_lxmf_message.reset_mock() + mock_message = MagicMock() + mock_message.get_fields.return_value = { + LXMF.FIELD_ICON_APPEARANCE: [icon_name, foreground_bytes, background_bytes], + } + mock_message.source_hash = os.urandom(16) + mock_message.hash = os.urandom(16) + + mock_app.on_lxmf_delivery(mock_message) + mock_app.db_upsert_lxmf_message.assert_called_once() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + icon_appearance_list=st.lists( + st.one_of( + st.text(), + st.binary(), + st.integers(), + st.floats(), + st.booleans(), + st.none(), + ), + min_size=0, + max_size=10, + ), +) +def test_lxmf_icon_appearance_list_variations_fuzzing(mock_app, icon_appearance_list): + """Fuzz LXMF.FIELD_ICON_APPEARANCE with various list structures.""" + mock_app.db_upsert_lxmf_message.reset_mock() + mock_message = MagicMock() + mock_message.get_fields.return_value = { + LXMF.FIELD_ICON_APPEARANCE: icon_appearance_list, + } + mock_message.source_hash = os.urandom(16) + mock_message.hash = os.urandom(16) + + mock_app.on_lxmf_delivery(mock_message) + mock_app.db_upsert_lxmf_message.assert_called_once() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + dest_hash=st.text(min_size=0, max_size=100), + icon_name=st.text(min_size=0, max_size=500), + fg_color=st.text(min_size=0, max_size=100), + bg_color=st.text(min_size=0, max_size=100), +) +def test_update_lxmf_user_icon_comprehensive_fuzzing( + mock_app, + dest_hash, + icon_name, + fg_color, + bg_color, +): + """Fuzz update_lxmf_user_icon with various string inputs.""" + mock_app.update_lxmf_user_icon(dest_hash, icon_name, fg_color, bg_color) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + audio_mode=st.one_of( + st.text(min_size=0, max_size=100), + st.binary(min_size=0, max_size=100), + st.integers(), + st.floats(), + st.booleans(), + st.none(), + ), + audio_bytes=st.one_of( + st.binary(min_size=0, max_size=100000), + st.text(min_size=0, max_size=10000), + st.lists(st.integers(min_value=0, max_value=255), min_size=0, max_size=1000), + st.none(), + ), +) +def test_lxmf_audio_field_structure_fuzzing(mock_app, audio_mode, audio_bytes): + """Fuzz LXMF.FIELD_AUDIO with structured data.""" + mock_app.db_upsert_lxmf_message.reset_mock() + mock_message = MagicMock() + mock_message.get_fields.return_value = { + LXMF.FIELD_AUDIO: [audio_mode, audio_bytes], + } + mock_message.source_hash = os.urandom(16) + mock_message.hash = os.urandom(16) + + mock_app.on_lxmf_delivery(mock_message) + mock_app.db_upsert_lxmf_message.assert_called_once() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + audio_field=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=20, + ), + st.binary(min_size=0, max_size=100000), + st.text(min_size=0, max_size=10000), + st.dictionaries( + keys=st.text(), + values=st.one_of(st.text(), st.binary(), st.integers()), + ), + ), +) +def test_lxmf_audio_field_variations_fuzzing(mock_app, audio_field): + """Fuzz LXMF.FIELD_AUDIO with various data structures.""" + mock_app.db_upsert_lxmf_message.reset_mock() + mock_message = MagicMock() + mock_message.get_fields.return_value = {LXMF.FIELD_AUDIO: audio_field} + mock_message.source_hash = os.urandom(16) + mock_message.hash = os.urandom(16) + + mock_app.on_lxmf_delivery(mock_message) + mock_app.db_upsert_lxmf_message.assert_called_once() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + contact_name=st.text(min_size=0, max_size=500), + contact_hash=st.text(min_size=0, max_size=100), +) +def test_contact_sharing_content_fuzzing(mock_app, contact_name, contact_hash): + """Fuzz contact sharing content parsing.""" + mock_app.db_upsert_lxmf_message.reset_mock() + contact_content = f"Contact: {contact_name} <{contact_hash}>" + mock_message = MagicMock() + mock_message.content = contact_content.encode("utf-8", errors="ignore") + mock_message.title = b"" + mock_message.hash = os.urandom(16) + mock_message.source_hash = os.urandom(16) + mock_message.destination_hash = os.urandom(16) + mock_message.incoming = True + mock_message.state = LXMF.LXMessage.DELIVERED + mock_message.method = LXMF.LXMessage.DIRECT + mock_message.progress = 1.0 + mock_message.timestamp = 123456789.0 + mock_message.rssi = -50 + mock_message.snr = 10 + mock_message.q = 100 + mock_message.get_fields.return_value = {} + + mock_app.on_lxmf_delivery(mock_message) + mock_app.db_upsert_lxmf_message.assert_called_once() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + contact_content=st.text(min_size=0, max_size=2000), +) +def test_contact_sharing_malformed_content_fuzzing(mock_app, contact_content): + """Fuzz contact sharing with malformed content strings.""" + mock_app.db_upsert_lxmf_message.reset_mock() + mock_message = MagicMock() + mock_message.content = contact_content.encode("utf-8", errors="ignore") + mock_message.title = b"" + mock_message.hash = os.urandom(16) + mock_message.source_hash = os.urandom(16) + mock_message.destination_hash = os.urandom(16) + mock_message.incoming = True + mock_message.state = LXMF.LXMessage.DELIVERED + mock_message.method = LXMF.LXMessage.DIRECT + mock_message.progress = 1.0 + mock_message.timestamp = 123456789.0 + mock_message.rssi = -50 + mock_message.snr = 10 + mock_message.q = 100 + mock_message.get_fields.return_value = {} + + mock_app.on_lxmf_delivery(mock_message) + mock_app.db_upsert_lxmf_message.assert_called_once() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + name=st.text(min_size=0, max_size=500), + hash_str=st.text(min_size=0, max_size=100), +) +def test_add_contact_api_fuzzing(mock_app, name, hash_str): + """Fuzz contact addition API with various inputs.""" + 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": "telephone.add_contact", + "name": name, + "remote_identity_hash": hash_str, + }, + ), + ) + finally: + loop.close() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + time_utc=st.one_of( + st.integers(min_value=-2147483648, max_value=2147483647), + st.floats(min_value=-2147483648, max_value=2147483647), + st.none(), + ), + location=st.one_of( + st.fixed_dictionaries( + { + "latitude": st.floats(allow_nan=False, allow_infinity=False), + "longitude": st.floats(allow_nan=False, allow_infinity=False), + }, + optional={ + "altitude": st.floats(allow_nan=False, allow_infinity=False), + "speed": st.floats(allow_nan=False, allow_infinity=False), + "bearing": st.floats(allow_nan=False, allow_infinity=False), + "accuracy": st.floats(allow_nan=False, allow_infinity=False), + "last_update": st.floats(allow_nan=False, allow_infinity=False), + }, + ), + st.none(), + ), +) +def test_telemetry_pack_fuzzing(mock_app, time_utc, location): + """Fuzz Telemeter.pack with various data.""" + from meshchatx.src.backend.telemetry_utils import Telemeter + + try: + Telemeter.pack(time_utc=time_utc, location=location) + except Exception: + pass + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + packed_location=st.one_of( + st.lists( + st.one_of( + st.binary(min_size=0, max_size=100), + st.integers(), + st.text(), + st.floats(), + st.none(), + ), + min_size=0, + max_size=20, + ), + st.binary(min_size=0, max_size=1000), + st.text(min_size=0, max_size=1000), + st.none(), + ), +) +def test_telemetry_unpack_location_fuzzing(mock_app, packed_location): + """Fuzz Telemeter.unpack_location with various formats.""" + from meshchatx.src.backend.telemetry_utils import Telemeter + + Telemeter.unpack_location(packed_location) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + latitude=st.floats(allow_nan=True, allow_infinity=True), + longitude=st.floats(allow_nan=True, allow_infinity=True), + altitude=st.one_of(st.floats(allow_nan=True, allow_infinity=True), st.integers()), + speed=st.one_of(st.floats(allow_nan=True, allow_infinity=True), st.integers()), + bearing=st.one_of(st.floats(allow_nan=True, allow_infinity=True), st.integers()), + accuracy=st.one_of(st.floats(allow_nan=True, allow_infinity=True), st.integers()), + last_update=st.one_of( + st.integers(), st.floats(), st.text(), st.binary(), st.none() + ), +) +def test_telemetry_pack_location_fuzzing( + mock_app, + latitude, + longitude, + altitude, + speed, + bearing, + accuracy, + last_update, +): + """Fuzz Telemeter.pack_location with edge case coordinates.""" + from meshchatx.src.backend.telemetry_utils import Telemeter + + Telemeter.pack_location( + latitude=latitude, + longitude=longitude, + altitude=altitude, + speed=speed, + bearing=bearing, + accuracy=accuracy, + last_update=last_update, + ) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + destination_hash=st.one_of(st.text(), st.binary()), + timestamp=st.one_of( + st.integers(), + st.floats(allow_nan=False, allow_infinity=False), + st.text(), + st.none(), + ), + data=st.one_of( + st.text(), st.binary(), st.dictionaries(keys=st.text(), values=st.text()) + ), + received_from=st.one_of(st.text(), st.binary(), st.none()), + physical_link=st.one_of( + st.dictionaries( + keys=st.text(), values=st.one_of(st.integers(), st.floats(), st.text()) + ), + st.text(), + st.binary(), + st.none(), + ), +) +def test_telemetry_upsert_fuzzing( + mock_app, + destination_hash, + timestamp, + data, + received_from, + physical_link, +): + """Fuzz telemetry database upsert with varied data types.""" + dest_hash_str = ( + destination_hash.decode("utf-8", errors="ignore") + if isinstance(destination_hash, bytes) + else str(destination_hash) + ) + mock_app.database.telemetry.upsert_telemetry( + destination_hash=dest_hash_str, + timestamp=timestamp, + data=data, + received_from=received_from, + physical_link=physical_link, + ) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + z=st.one_of(st.integers(), st.text(), st.floats()), + x=st.one_of(st.integers(), st.text(), st.floats()), + y=st.one_of(st.integers(), st.text(), st.floats()), +) +def test_map_tile_coordinates_fuzzing(mock_app, z, x, y): + """Fuzz map tile coordinate parsing.""" + try: + z_int = ( + int(z) + if isinstance(z, (int, float)) + and not (isinstance(z, float) and (z != z or abs(z) == float("inf"))) + else 0 + ) + x_int = ( + int(x) + if isinstance(x, (int, float)) + and not (isinstance(x, float) and (x != x or abs(x) == float("inf"))) + else 0 + ) + y_int = ( + int(y) + if isinstance(y, (int, float)) + and not (isinstance(y, float) and (y != y or abs(y) == float("inf"))) + else 0 + ) + mock_app.map_manager.get_tile(z_int, x_int, y_int) + except (OverflowError, ValueError): + pass + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + filename=st.text(min_size=0, max_size=500), +) +def test_mbtiles_filename_fuzzing(mock_app, filename): + """Fuzz MBTiles filename handling.""" + mock_app.map_manager.delete_mbtiles(filename) + mock_app.map_manager.get_connection(filename) + + +@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=1000), + content=st.text(min_size=0, max_size=100000), +) +def test_archive_page_content_fuzzing(mock_app, destination_hash, page_path, content): + """Fuzz archive page content storage and retrieval.""" + mock_app.archiver_manager.archive_page( + destination_hash, + page_path, + content, + max_versions=5, + max_storage_gb=1, + ) + mock_app.archiver_manager.get_archived_page_versions(destination_hash, page_path) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + ids=st.lists( + st.one_of(st.integers(), st.text(), st.floats()), + min_size=0, + max_size=100, + ), +) +def test_delete_archived_pages_ids_fuzzing(mock_app, ids): + """Fuzz SQL injection in delete_archived_pages.""" + mock_app.database.misc.delete_archived_pages(ids=ids) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + query=st.text(min_size=0, max_size=500), +) +def test_archived_pages_query_sql_injection_fuzzing(mock_app, query): + """Fuzz SQL injection in archived_pages search.""" + mock_app.database.misc.get_archived_pages_paginated(query=query) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + file_path=st.text(min_size=0, max_size=1000), +) +def test_rncp_file_path_traversal_fuzzing(mock_app, file_path): + """Fuzz RNCP file path handling for directory traversal.""" + try: + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + loop.run_until_complete( + mock_app.rncp_handler.send_file( + os.urandom(16), + file_path, + timeout=1.0, + ), + ) + except Exception: + pass + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + path=st.text(min_size=0, max_size=1000), + data=st.one_of(st.text(), st.binary()), + request_id=st.one_of(st.integers(), st.text()), +) +def test_rncp_fetch_request_path_fuzzing(mock_app, path, data, request_id): + """Fuzz RNCP fetch request path handling.""" + try: + mock_identity = MagicMock() + mock_identity.hash = os.urandom(16) + mock_app.rncp_handler._fetch_request( + path, + data, + request_id, + os.urandom(16), + mock_identity, + time.time(), + ) + except Exception: + pass + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + app_data_base64=st.text(min_size=0, max_size=10000), +) +def test_parse_lxmf_stamp_cost_fuzzing(mock_app, app_data_base64): + """Fuzz LXMF stamp cost parsing from base64 app_data.""" + try: + from meshchatx.src.backend.meshchat_utils import parse_lxmf_stamp_cost + + parse_lxmf_stamp_cost(app_data_base64) + except Exception: + pass + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + app_data_base64=st.text(min_size=0, max_size=10000), +) +def test_parse_lxmf_propagation_node_app_data_fuzzing(mock_app, app_data_base64): + """Fuzz LXMF propagation node app_data parsing.""" + from meshchatx.src.backend.meshchat_utils import ( + parse_lxmf_propagation_node_app_data, + ) + + parse_lxmf_propagation_node_app_data(app_data_base64) + + +@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=1000), +) +def test_nomadnet_page_path_fuzzing(mock_app, destination_hash, page_path): + """Fuzz NomadNet page path handling.""" + mock_app.nomadnet_manager.archive_page( + destination_hash, + page_path, + "test content", + is_manual=False, + ) + mock_app.nomadnet_manager.get_archived_page_versions(destination_hash, page_path) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + page_content=st.text(min_size=0, max_size=100000), +) +def test_nomadnet_page_content_fuzzing(mock_app, page_content): + """Fuzz NomadNet page content parsing.""" + from meshchatx.src.backend.nomadnet_utils import ( + convert_nomadnet_field_data_to_map, + ) + + convert_nomadnet_field_data_to_map(page_content) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + table_name=st.text(min_size=0, max_size=100), +) +def test_sql_table_name_injection_fuzzing(mock_app, table_name): + """Fuzz SQL table name injection.""" + mock_app.database.provider.execute(f"PRAGMA table_info({table_name})") + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + fields_json=st.text(min_size=0, max_size=10000), +) +def test_lxmf_fields_json_parsing_fuzzing(mock_app, fields_json): + """Fuzz LXMF fields JSON parsing.""" + db_message = { + "id": 1, + "hash": "test", + "source_hash": "test", + "destination_hash": "test", + "is_incoming": True, + "state": "delivered", + "progress": 100.0, + "method": "direct", + "delivery_attempts": 0, + "next_delivery_attempt_at": None, + "title": "test", + "content": "test", + "fields": fields_json, + "timestamp": 123456789.0, + "rssi": -50, + "snr": 10, + "quality": 100, + "is_spam": False, + "created_at": "2024-01-01T00:00:00Z", + "updated_at": "2024-01-01T00:00:00Z", + } + from meshchatx.src.backend.lxmf_utils import convert_db_lxmf_message_to_dict + + try: + convert_db_lxmf_message_to_dict(db_message) + except Exception: + pass + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + password=st.text(min_size=0, max_size=1000), +) +def test_auth_password_fuzzing(mock_app, password): + """Fuzz authentication password handling.""" + 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": "auth.login", "password": password}, + ), + ) + finally: + loop.close() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + filename=st.text(min_size=0, max_size=500), +) +def test_mbtiles_upload_filename_fuzzing(mock_app, filename): + """Fuzz MBTiles upload filename.""" + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + mock_field = MagicMock() + mock_field.name = "file" + mock_field.filename = filename + mock_field.read_chunk = MagicMock(return_value=b"") + mock_reader = MagicMock() + mock_reader.next = MagicMock(return_value=mock_field) + + mock_request = MagicMock() + mock_request.multipart = MagicMock(return_value=mock_reader) + + loop.run_until_complete( + mock_app.on_websocket_data_received( + MagicMock(), + {"type": "map.upload_offline", "filename": filename}, + ), + ) + finally: + loop.close() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + crawl_destination_hash=st.text(min_size=0, max_size=100), + crawl_page_path=st.text(min_size=0, max_size=1000), +) +def test_crawler_task_path_fuzzing(mock_app, crawl_destination_hash, crawl_page_path): + """Fuzz crawler task destination hash and page path.""" + mock_app.database.misc.upsert_crawl_task( + crawl_destination_hash, + crawl_page_path, + status="pending", + retry_count=0, + ) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + doc_path=st.text(min_size=0, max_size=1000), +) +def test_docs_path_traversal_fuzzing(mock_app, doc_path): + """Fuzz documentation path handling.""" + try: + mock_app.docs_manager.get_doc_content(doc_path) + except (IsADirectoryError, OSError): + pass + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + bbox=st.lists( + st.one_of(st.floats(allow_nan=False, allow_infinity=False), st.integers()), + min_size=4, + max_size=4, + ), + min_zoom=st.integers(min_value=-10, max_value=30), + max_zoom=st.integers(min_value=-10, max_value=30), + name=st.text(min_size=0, max_size=500), +) +def test_map_export_parameters_fuzzing(mock_app, bbox, min_zoom, max_zoom, name): + """Fuzz map export parameters.""" + mock_app.map_manager.start_export( + "test_export", + bbox, + min_zoom, + max_zoom, + name=name, + ) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + mbtiles_path=st.text(min_size=0, max_size=1000), +) +def test_mbtiles_metadata_parsing_fuzzing(mock_app, mbtiles_path): + """Fuzz MBTiles metadata parsing.""" + mock_app.map_manager.get_metadata() + if os.path.exists(mbtiles_path): + mock_app.map_manager.get_connection(mbtiles_path) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + audio_frame=st.one_of( + st.binary(min_size=0, max_size=10000), + st.lists(st.integers(min_value=0, max_value=255), min_size=0, max_size=1000), + st.text(min_size=0, max_size=1000), + st.none(), + ), +) +def test_lxst_audio_frame_handling_fuzzing(mock_app, audio_frame): + """Fuzz LXST audio frame handling in Tee.handle_frame.""" + try: + from meshchatx.src.backend.telephone_manager import Tee + + mock_sink = MagicMock() + mock_sink.handle_frame = MagicMock() + mock_sink.can_receive = MagicMock(return_value=True) + tee = Tee(mock_sink) + tee.handle_frame(audio_frame, "test_source") + except Exception: + pass + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + call_status=st.integers(min_value=-10, max_value=20), + caller_identity_hash=st.binary(min_size=0, max_size=100), +) +def test_lxst_call_state_transitions_fuzzing( + mock_app, call_status, caller_identity_hash +): + """Fuzz LXST call state transitions with invalid states.""" + try: + mock_identity = MagicMock() + mock_identity.hash = caller_identity_hash + + if ( + hasattr(mock_app.telephone_manager, "telephone") + and mock_app.telephone_manager.telephone + ): + mock_app.telephone_manager.telephone.call_status = call_status + 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( + codec2_data=st.binary(min_size=0, max_size=100000), + codec_mode=st.sampled_from( + [ + "450PWB", + "450", + "700C", + "1200", + "1300", + "1400", + "1600", + "2400", + "3200", + "invalid", + ] + ), +) +def test_codec2_decode_fuzzing(mock_app, codec2_data, codec_mode): + """Fuzz Codec2 audio decoding with malformed data.""" + try: + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + mock_client = MagicMock() + loop.run_until_complete( + mock_app.on_websocket_data_received( + mock_client, + { + "type": "codec2.decode", + "data": codec2_data, + "mode": codec_mode, + }, + ), + ) + except Exception: + pass + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + opus_data=st.one_of( + st.binary(min_size=0, max_size=100000), + st.text(min_size=0, max_size=10000), + st.lists(st.integers(min_value=0, max_value=255), min_size=0, max_size=10000), + ), +) +def test_opus_audio_decode_fuzzing(mock_app, opus_data): + """Fuzz Opus audio decoding in LXMF audio fields.""" + mock_app.db_upsert_lxmf_message.reset_mock() + mock_message = MagicMock() + mock_message.get_fields.return_value = { + LXMF.FIELD_AUDIO: [0x10, opus_data], # AM_OPUS_OGG = 0x10 + } + mock_message.source_hash = os.urandom(16) + mock_message.hash = os.urandom(16) + + mock_app.on_lxmf_delivery(mock_message) + mock_app.db_upsert_lxmf_message.assert_called_once() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + audio_mode=st.integers(min_value=0, max_value=255), + audio_bytes=st.binary(min_size=0, max_size=100000), +) +def test_lxmf_audio_mode_fuzzing(mock_app, audio_mode, audio_bytes): + """Fuzz all possible LXMF audio mode values.""" + mock_app.db_upsert_lxmf_message.reset_mock() + mock_message = MagicMock() + mock_message.get_fields.return_value = { + LXMF.FIELD_AUDIO: [audio_mode, audio_bytes], + } + mock_message.source_hash = os.urandom(16) + mock_message.hash = os.urandom(16) + + mock_app.on_lxmf_delivery(mock_message) + mock_app.db_upsert_lxmf_message.assert_called_once() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + profile_id=st.one_of( + st.integers(min_value=-100, max_value=100), + st.text(min_size=0, max_size=100), + st.none(), + ), +) +def test_lxst_profile_switching_fuzzing(mock_app, profile_id): + """Fuzz LXST audio profile switching.""" + if ( + hasattr(mock_app.telephone_manager, "telephone") + and mock_app.telephone_manager.telephone + ): + mock_app.telephone_manager.telephone.switch_profile(profile_id) + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + destination_hash=st.one_of( + st.binary(min_size=0, max_size=100), st.text(min_size=0, max_size=100) + ), + timeout=st.one_of( + st.integers(min_value=-100, max_value=1000), + st.floats(allow_nan=True, allow_infinity=True), + ), +) +def test_lxst_call_initiation_fuzzing(mock_app, destination_hash, timeout): + """Fuzz LXST call initiation.""" + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + dest_hash_bytes = ( + bytes.fromhex(destination_hash) + if isinstance(destination_hash, str) and len(destination_hash) == 32 + else destination_hash + if isinstance(destination_hash, bytes) + else os.urandom(16) + ) + timeout_int = ( + int(timeout) + if isinstance(timeout, (int, float)) + and not ( + isinstance(timeout, float) + and (timeout != timeout or abs(timeout) == float("inf")) + ) + else 15 + ) + + loop.run_until_complete( + mock_app.telephone_manager.initiate( + dest_hash_bytes, timeout_seconds=timeout_int + ), + ) + finally: + loop.close() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + micron_content=st.text(min_size=0, max_size=50000), +) +def test_micron_parser_content_fuzzing(mock_app, micron_content): + """Fuzz Micron parser content handling.""" + mock_app.db_upsert_lxmf_message.reset_mock() + mock_message = MagicMock() + mock_message.content = micron_content.encode("utf-8", errors="ignore") + mock_message.title = b"" + mock_message.hash = os.urandom(16) + mock_message.source_hash = os.urandom(16) + mock_message.destination_hash = os.urandom(16) + mock_message.incoming = True + mock_message.state = LXMF.LXMessage.DELIVERED + mock_message.method = LXMF.LXMessage.DIRECT + mock_message.progress = 1.0 + mock_message.timestamp = 123456789.0 + mock_message.rssi = -50 + mock_message.snr = 10 + mock_message.q = 100 + mock_message.get_fields.return_value = {} + + mock_app.on_lxmf_delivery(mock_message) + mock_app.db_upsert_lxmf_message.assert_called_once() + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + voicemail_text=st.text(min_size=0, max_size=10000), +) +def test_voicemail_greeting_text_fuzzing(mock_app, voicemail_text): + """Fuzz voicemail greeting generation.""" + from meshchatx.src.backend.voicemail_manager import VoicemailManager + + # Use real VoicemailManager to test its internal logic calling subprocess + vm = VoicemailManager(MagicMock(), MagicMock(), MagicMock(), "/tmp/voicemail_test") + vm.has_espeak = True + vm.has_ffmpeg = True + vm.espeak_path = "/usr/bin/espeak" + vm.ffmpeg_path = "/usr/bin/ffmpeg" + + with patch("subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0) + try: + vm.generate_greeting(voicemail_text) + # If text is provided, it should call subprocess.run + if voicemail_text.strip(): + assert mock_run.called + except Exception: + # Ignore errors from underlying tools in fuzzing + pass + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + rns_packet_data=st.binary(min_size=0, max_size=10000), +) +def test_rns_packet_parsing_fuzzing(mock_app, rns_packet_data): + """Fuzz RNS packet parsing with malformed protocol data.""" try: import RNS try: - RNS.Identity.from_bytes(binary_data) + RNS.Packet(None, rns_packet_data) except Exception: pass try: - id_inst = RNS.Identity(create_keys=False) - id_inst.load_private_key(binary_data) - except Exception: - pass - try: - id_inst = RNS.Identity(create_keys=False) - id_inst.load_public_key(binary_data) + RNS.Packet.unpack(rns_packet_data) except Exception: pass except Exception: pass + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + lxmf_message_data=st.binary(min_size=0, max_size=100000), +) +def test_lxmf_message_unpacking_fuzzing(mock_app, lxmf_message_data): + """Fuzz LXMF message unpacking.""" + try: + LXMF.LXMessage.unpack(lxmf_message_data) + except Exception: + pass + try: + message = LXMF.LXMessage(None, None, "") + message.unpack(lxmf_message_data) + except Exception: + pass + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + pipeline_config=st.dictionaries( + keys=st.text(), + values=st.one_of( + st.text(), st.binary(), st.integers(), st.floats(), st.booleans(), st.none() + ), + ), +) +def test_lxst_pipeline_config_fuzzing(mock_app, pipeline_config): + """Fuzz LXST Pipeline configuration.""" + from LXST.Pipeline import Pipeline + from LXST.Codecs import Null + from LXST.Sources import Source + from LXST.Sinks import Sink + + class DummySource(Source): + pass + + class DummySink(Sink): + pass + + # Pipeline requires source, codec, and sink + try: + pipeline = Pipeline(source=DummySource(), codec=Null(), sink=DummySink()) + for key, value in pipeline_config.items(): + try: + setattr(pipeline, key, value) + except Exception: + pass + except Exception: + pass + + +@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None) +@given( + sink_data=st.one_of( + st.binary(min_size=0, max_size=10000), st.text(min_size=0, max_size=1000) + ), +) +def test_lxst_sink_handling_fuzzing(mock_app, sink_data): + """Fuzz LXST sink data handling.""" + from LXST.Sinks import OpusFileSink + + sink = OpusFileSink("/tmp/test.opus") + try: + sink.handle_frame(sink_data, "test_source") + except Exception: + pass + try: + sink.can_receive("test_source") + except Exception: + pass + + +def test_telemetry_packing_invariants_regression(): + """Deterministic regression test for telemetry packing/unpacking.""" + from meshchatx.src.backend.telemetry_utils import Telemeter + + original_data = { + "time": {"utc": 123456789.0}, + "location": { + "latitude": 45.0, + "longitude": -90.0, + "altitude": 100, + "speed": 10, + "bearing": 180, + "accuracy": 5, + "last_update": 123456780.0, + }, + } + + packed = Telemeter.pack( + time_utc=original_data["time"]["utc"], location=original_data["location"] + ) + unpacked = Telemeter.from_packed(packed) + + assert unpacked["time"]["utc"] == original_data["time"]["utc"] + assert unpacked["location"]["latitude"] == original_data["location"]["latitude"] + assert unpacked["location"]["longitude"] == original_data["location"]["longitude"] + + +def test_lxmf_display_name_parsing_regression(): + """Deterministic regression test for LXMF display name parsing.""" + from meshchatx.src.backend.meshchat_utils import parse_lxmf_display_name + + valid_b64 = base64.b64encode(b"test").decode() + + with patch("LXMF.display_name_from_app_data") as mock_parser: + # Success case + mock_parser.return_value = "Test User" + assert parse_lxmf_display_name(valid_b64) == "Test User" + + # None case (fallback to default) + mock_parser.return_value = None + assert ( + parse_lxmf_display_name(valid_b64, default_value="Fallback") == "Fallback" + ) + + # Exception case + mock_parser.side_effect = Exception("Parsing error") + assert ( + parse_lxmf_display_name(valid_b64, default_value="Fallback") == "Fallback" + ) + + # None input + assert parse_lxmf_display_name(None, default_value="Fallback") == "Fallback"