From 0fbcd855266271154c17c273270d05682d7d8500 Mon Sep 17 00:00:00 2001 From: Ivan Date: Mon, 13 Apr 2026 20:56:15 -0500 Subject: [PATCH] feat(tests): enhance coverage for LXMF message handling, including image attachment stripping for strangers and sticker validation tests --- tests/backend/test_meshchat_coverage.py | 174 +++++++++++++++++++++-- tests/backend/test_security_fuzzing.py | 74 ++++++++++ tests/backend/test_sticker_utils.py | 180 ++++++++++++++++++++++++ tests/backend/test_stickers_dao.py | 154 ++++++++++++++++++++ 4 files changed, 574 insertions(+), 8 deletions(-) create mode 100644 tests/backend/test_sticker_utils.py create mode 100644 tests/backend/test_stickers_dao.py diff --git a/tests/backend/test_meshchat_coverage.py b/tests/backend/test_meshchat_coverage.py index 5f01025..123477d 100644 --- a/tests/backend/test_meshchat_coverage.py +++ b/tests/backend/test_meshchat_coverage.py @@ -1,9 +1,12 @@ import pytest -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import asyncio import json import os +import subprocess +import LXMF from meshchatx.meshchat import ReticulumMeshChat +from meshchatx.src.backend.lxmf_message_fields import LxmfAudioField @pytest.fixture @@ -167,6 +170,7 @@ def test_get_config_dict_basic(mock_app): "message_outbound_bubble_color", "message_inbound_bubble_color", "message_failed_bubble_color", + "message_waiting_bubble_color", ]: getattr(mock_config, attr).get.return_value = None @@ -289,18 +293,34 @@ async def test_send_config_to_websocket_clients(mock_app): @pytest.mark.asyncio async def test_on_lxmf_sending_state_updated(mock_app): mock_msg = MagicMock() - mock_app.db_upsert_lxmf_message = MagicMock() + mock_msg.progress = 0.75 + mock_msg.rssi = None + mock_msg.snr = None + mock_msg.q = None + mock_msg.hash.hex.return_value = "ab" * 16 + mock_msg.delivery_attempts = 2 + + ctx = mock_app.current_context mock_app.websocket_broadcast = MagicMock(return_value=asyncio.Future()) mock_app.websocket_broadcast.return_value.set_result(None) - with patch( - "meshchatx.meshchat.convert_lxmf_message_to_dict", return_value={"h": "v"} + with ( + patch( + "meshchatx.meshchat.convert_lxmf_message_to_dict", return_value={"h": "v"} + ), + patch( + "meshchatx.meshchat.convert_lxmf_state_to_string", return_value="delivered" + ), + patch("meshchatx.meshchat.AsyncUtils.run_async") as mock_run_async, ): - # Pass context explicitly to match expectation or fix expectation - ctx = mock_app.current_context mock_app.on_lxmf_sending_state_updated(mock_msg, context=ctx) - mock_app.db_upsert_lxmf_message.assert_called_once_with(mock_msg, context=ctx) - mock_app.websocket_broadcast.assert_called_once() + + ctx.database.messages.update_lxmf_message_state.assert_called_once() + call_kwargs = ctx.database.messages.update_lxmf_message_state.call_args + assert call_kwargs.kwargs["message_hash"] == "ab" * 16 + assert call_kwargs.kwargs["progress"] == 75.0 + assert call_kwargs.kwargs["state"] == "delivered" + mock_run_async.assert_called_once() @pytest.mark.asyncio @@ -344,3 +364,141 @@ def test_on_lxmf_sending_failed_no_propagation(mock_app): mock_app.on_lxmf_sending_state_updated.assert_called_once_with( mock_msg, context=None ) + + +def test_convert_webm_opus_to_ogg_already_ogg(mock_app): + ogg_data = b"OggS" + b"\x00" * 100 + result = mock_app._convert_webm_opus_to_ogg(ogg_data) + assert result is ogg_data + + +def test_convert_webm_opus_to_ogg_no_ffmpeg(mock_app): + webm_data = b"\x1a\x45\xdf\xa3" + b"\x00" * 100 + with patch("shutil.which", return_value=None): + result = mock_app._convert_webm_opus_to_ogg(webm_data) + assert result is webm_data + + +def test_convert_webm_opus_to_ogg_success(mock_app): + webm_data = b"\x1a\x45\xdf\xa3" + b"\x00" * 100 + fake_ogg = b"OggS" + b"\xff" * 50 + mock_result = MagicMock() + mock_result.returncode = 0 + mock_result.stdout = fake_ogg + + with patch("shutil.which", return_value="/usr/bin/ffmpeg"): + with patch("subprocess.run", return_value=mock_result) as mock_run: + result = mock_app._convert_webm_opus_to_ogg(webm_data) + + assert result == fake_ogg + mock_run.assert_called_once() + call_args = mock_run.call_args + assert call_args.kwargs["input"] == webm_data + assert "-f" in call_args[0][0] + assert "ogg" in call_args[0][0] + + +def test_convert_webm_opus_to_ogg_ffmpeg_fails(mock_app): + webm_data = b"\x1a\x45\xdf\xa3" + b"\x00" * 100 + mock_result = MagicMock() + mock_result.returncode = 1 + mock_result.stdout = b"" + + with patch("shutil.which", return_value="/usr/bin/ffmpeg"): + with patch("subprocess.run", return_value=mock_result): + result = mock_app._convert_webm_opus_to_ogg(webm_data) + + assert result is webm_data + + +def test_convert_webm_opus_to_ogg_exception(mock_app): + webm_data = b"\x1a\x45\xdf\xa3" + b"\x00" * 100 + + with patch("shutil.which", return_value="/usr/bin/ffmpeg"): + with patch( + "subprocess.run", + side_effect=subprocess.TimeoutExpired(cmd="ffmpeg", timeout=30), + ): + result = mock_app._convert_webm_opus_to_ogg(webm_data) + + assert result is webm_data + + +@pytest.fixture +def sendable_app(mock_app): + """mock_app wired so send_message can run to completion.""" + ctx = mock_app.current_context + ctx.config.auto_send_failed_messages_to_propagation_node.get.return_value = False + ctx.message_router.delivery_link_available.return_value = True + ctx.local_lxmf_destination = MagicMock() + ctx.forwarding_manager = None + + mock_app._await_transport_path = AsyncMock() + mock_app.get_current_icon_hash = MagicMock(return_value=None) + mock_app.db_upsert_lxmf_message = MagicMock() + mock_app.websocket_broadcast = AsyncMock() + mock_app.handle_lxmf_message_progress = AsyncMock() + mock_app._convert_webm_opus_to_ogg = MagicMock(side_effect=lambda b: b) + + return mock_app + + +async def _run_send(app, destination_hash="aa" * 16, **kwargs): + fake_identity = MagicMock() + fake_destination = MagicMock() + fake_lxm = MagicMock(spec=LXMF.LXMessage) + fake_lxm.fields = {} + fake_lxm.include_ticket = False + + with ( + patch("meshchatx.meshchat.RNS.Identity.recall", return_value=fake_identity), + patch("meshchatx.meshchat.RNS.Destination", return_value=fake_destination), + patch("meshchatx.meshchat.LXMF.LXMessage", return_value=fake_lxm), + patch( + "meshchatx.meshchat.convert_lxmf_message_to_dict", + return_value={"hash": "x"}, + ), + patch("meshchatx.meshchat.AsyncUtils.run_async"), + ): + await app.send_message( + destination_hash=destination_hash, content="hi", **kwargs + ) + + return fake_lxm + + +@pytest.mark.asyncio +async def test_send_message_sets_renderer_markdown(sendable_app): + lxm = await _run_send(sendable_app) + assert LXMF.FIELD_RENDERER in lxm.fields + assert lxm.fields[LXMF.FIELD_RENDERER] == LXMF.RENDERER_MARKDOWN + + +@pytest.mark.asyncio +async def test_send_message_include_ticket_for_contact(sendable_app): + sendable_app._is_contact = MagicMock(return_value=True) + lxm = await _run_send(sendable_app) + assert lxm.include_ticket is True + + +@pytest.mark.asyncio +async def test_send_message_no_ticket_for_stranger(sendable_app): + sendable_app._is_contact = MagicMock(return_value=False) + lxm = await _run_send(sendable_app) + assert lxm.include_ticket is False + + +@pytest.mark.asyncio +async def test_send_message_opus_audio_triggers_conversion(sendable_app): + audio = LxmfAudioField(audio_mode=LXMF.AM_OPUS_OGG, audio_bytes=b"\x1a\x45") + lxm = await _run_send(sendable_app, audio_field=audio) + sendable_app._convert_webm_opus_to_ogg.assert_called_once_with(b"\x1a\x45") + assert LXMF.FIELD_AUDIO in lxm.fields + + +@pytest.mark.asyncio +async def test_send_message_codec2_audio_skips_conversion(sendable_app): + audio = LxmfAudioField(audio_mode=LXMF.AM_CODEC2_1200, audio_bytes=b"\xcc") + lxm = await _run_send(sendable_app, audio_field=audio) + sendable_app._convert_webm_opus_to_ogg.assert_not_called() + assert lxm.fields[LXMF.FIELD_AUDIO] == [LXMF.AM_CODEC2_1200, b"\xcc"] diff --git a/tests/backend/test_security_fuzzing.py b/tests/backend/test_security_fuzzing.py index c1ad9ee..fd04494 100644 --- a/tests/backend/test_security_fuzzing.py +++ b/tests/backend/test_security_fuzzing.py @@ -2424,6 +2424,80 @@ class TestStrangerAttachmentBlocking: mock_app.on_lxmf_delivery(mock_msg) mock_app.db_upsert_lxmf_message.assert_called_once() + def test_stranger_lxmf_image_stripped_when_enabled(self, mock_app): + """Image attachments (including sticker-shaped PNGs) use FIELD_IMAGE; strip like files.""" + source_hash = os.urandom(16) + mock_msg = MagicMock() + mock_msg.source_hash = source_hash + mock_msg.destination_hash = os.urandom(16) + mock_msg.hash = os.urandom(16) + mock_msg.content = b"sticker msg" + mock_msg.title = b"" + mock_msg.incoming = True + mock_msg.state = LXMF.LXMessage.DELIVERED + mock_msg.method = LXMF.LXMessage.DIRECT + mock_msg.progress = 1.0 + mock_msg.timestamp = 123456789.0 + mock_msg.rssi = -50 + mock_msg.snr = 10 + mock_msg.q = 100 + mock_msg.delivery_attempts = 1 + mock_msg.next_delivery_attempt = None + + fields = { + LXMF.FIELD_IMAGE: [b"png", b"\x89PNG\r\n\x1a\n" + b"\x00" * 64], + } + mock_msg.get_fields.return_value = fields + mock_msg.fields = fields + + mock_app.config.block_attachments_from_strangers.get.return_value = True + mock_app.config.block_all_from_strangers.get.return_value = False + mock_app._is_contact = MagicMock(return_value=False) + mock_app.is_destination_blocked = MagicMock(return_value=False) + mock_app.check_spam_keywords = MagicMock(return_value=False) + + mock_app.on_lxmf_delivery(mock_msg) + + mock_app.db_upsert_lxmf_message.assert_called_once() + call_kwargs = mock_app.db_upsert_lxmf_message.call_args + assert call_kwargs[1].get("attachments_stripped") is True or ( + len(call_kwargs[0]) > 2 and call_kwargs[0][2] is True + ) + assert LXMF.FIELD_IMAGE not in mock_msg.get_fields() + + def test_contact_lxmf_image_not_stripped(self, mock_app): + source_hash = os.urandom(16) + mock_msg = MagicMock() + mock_msg.source_hash = source_hash + mock_msg.destination_hash = os.urandom(16) + mock_msg.hash = os.urandom(16) + mock_msg.content = b"img" + mock_msg.title = b"" + mock_msg.incoming = True + mock_msg.state = LXMF.LXMessage.DELIVERED + mock_msg.method = LXMF.LXMessage.DIRECT + mock_msg.progress = 1.0 + mock_msg.timestamp = 123456789.0 + mock_msg.rssi = -50 + mock_msg.snr = 10 + mock_msg.q = 100 + mock_msg.delivery_attempts = 1 + mock_msg.next_delivery_attempt = None + + fields = {LXMF.FIELD_IMAGE: [b"png", b"\x00" * 32]} + mock_msg.get_fields.return_value = fields + mock_msg.fields = fields + + mock_app.config.block_attachments_from_strangers.get.return_value = True + mock_app.config.block_all_from_strangers.get.return_value = False + mock_app._is_contact = MagicMock(return_value=True) + mock_app.is_destination_blocked = MagicMock(return_value=False) + mock_app.check_spam_keywords = MagicMock(return_value=False) + + mock_app.on_lxmf_delivery(mock_msg) + + assert LXMF.FIELD_IMAGE in mock_msg.get_fields() + class TestBlockAllFromStrangers: """Tests for the block-everything-from-strangers feature.""" diff --git a/tests/backend/test_sticker_utils.py b/tests/backend/test_sticker_utils.py new file mode 100644 index 0000000..f76d950 --- /dev/null +++ b/tests/backend/test_sticker_utils.py @@ -0,0 +1,180 @@ +"""Unit tests for sticker validation and export/import document parsing.""" + +import base64 + +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st + +from meshchatx.src.backend import sticker_utils + + +def test_normalize_image_type(): + assert sticker_utils.normalize_image_type("PNG") == "png" + assert sticker_utils.normalize_image_type("image/jpeg") == "jpeg" + assert sticker_utils.normalize_image_type("jpg") == "jpeg" + assert sticker_utils.normalize_image_type("webp") == "webp" + assert sticker_utils.normalize_image_type("svg") is None + assert sticker_utils.normalize_image_type("") is None + assert sticker_utils.normalize_image_type(None) is None + + +def test_validate_sticker_payload_ok(): + raw = b"\x89PNG\r\n\x1a\n" + b"\x00" * 32 + nt, h = sticker_utils.validate_sticker_payload(raw, "png") + assert nt == "png" + assert len(h) == 64 + + +def test_validate_sticker_payload_too_large(): + raw = b"x" * (sticker_utils.MAX_STICKER_BYTES + 1) + with pytest.raises(ValueError, match="image_too_large"): + sticker_utils.validate_sticker_payload(raw, "png") + + +def test_validate_sticker_payload_empty(): + with pytest.raises(ValueError, match="empty_image"): + sticker_utils.validate_sticker_payload(b"", "png") + + +def test_validate_sticker_payload_bad_type(): + with pytest.raises(ValueError, match="invalid_image_type"): + sticker_utils.validate_sticker_payload(b"abc", "svg") + + +def test_validate_sticker_payload_bad_magic(): + with pytest.raises(ValueError, match="invalid_image_signature"): + sticker_utils.validate_sticker_payload(b"not-an-image-bytes", "png") + + +def test_validate_sticker_payload_magic_type_mismatch(): + png_bytes = b"\x89PNG\r\n\x1a\n" + b"\x00" * 16 + with pytest.raises(ValueError, match="magic_type_mismatch"): + sticker_utils.validate_sticker_payload(png_bytes, "jpeg") + + +def test_detect_image_format_from_magic(): + assert sticker_utils.detect_image_format_from_magic(b"\x89PNG\r\n\x1a\n") == "png" + assert ( + sticker_utils.detect_image_format_from_magic(b"\xff\xd8\xff\xe0\x00\x10") + == "jpeg" + ) + assert ( + sticker_utils.detect_image_format_from_magic(b"GIF89a" + b"\x00" * 4) == "gif" + ) + assert sticker_utils.detect_image_format_from_magic(b"BM" + b"\x00" * 20) == "bmp" + webp = b"RIFF\x00\x00\x00\x00WEBP" + b"\x00" * 8 + assert sticker_utils.detect_image_format_from_magic(webp) == "webp" + assert sticker_utils.detect_image_format_from_magic(b"") is None + assert sticker_utils.detect_image_format_from_magic(b"short") is None + + +def test_validate_sticker_payload_jpg_alias_matches_jpeg_magic(): + raw = b"\xff\xd8\xff\xe0" + b"\x00" * 64 + nt, _ = sticker_utils.validate_sticker_payload(raw, "jpg") + assert nt == "jpeg" + + +def test_sanitize_sticker_name(): + assert sticker_utils.sanitize_sticker_name(" hello ") == "hello" + assert sticker_utils.sanitize_sticker_name("") is None + assert sticker_utils.sanitize_sticker_name("x" * 200) is not None + assert len(sticker_utils.sanitize_sticker_name("x" * 200) or "") == 128 + + +def test_validate_export_document_ok(): + png = b"\x89PNG\r\n\x1a\n" + b"\x00" * 32 + b64 = base64.b64encode(png).decode("ascii") + doc = { + "format": "meshchatx-stickers", + "version": 1, + "stickers": [ + {"name": "a", "image_type": "png", "image_bytes": b64}, + ], + } + items = sticker_utils.validate_export_document(doc) + assert len(items) == 1 + assert items[0]["image_bytes_b64"] == b64 + + +def test_validate_export_document_wrong_format(): + with pytest.raises(ValueError, match="invalid_format"): + sticker_utils.validate_export_document( + {"format": "other", "version": 1, "stickers": []} + ) + + +def test_validate_export_document_bad_version(): + with pytest.raises(ValueError, match="unsupported_version"): + sticker_utils.validate_export_document( + {"format": "meshchatx-stickers", "version": 99, "stickers": []}, + ) + + +def test_validate_export_document_missing_version(): + with pytest.raises(ValueError, match="unsupported_version"): + sticker_utils.validate_export_document( + {"format": "meshchatx-stickers", "stickers": []}, + ) + + +def test_validate_export_document_not_dict(): + with pytest.raises(ValueError, match="invalid_document"): + sticker_utils.validate_export_document([]) + + +def test_mime_for_image_type(): + assert "image/" in sticker_utils.mime_for_image_type("png") + assert sticker_utils.mime_for_image_type("unknown") == "application/octet-stream" + + +@settings(max_examples=200, deadline=None) +@given( + raw=st.binary(min_size=0, max_size=sticker_utils.MAX_STICKER_BYTES + 1), + typ=st.one_of( + st.none(), + st.text(max_size=40), + st.sampled_from( + ["png", "jpeg", "jpg", "webp", "gif", "bmp", "svg", "image/png", ""] + ), + ), +) +def test_validate_sticker_payload_fuzz_never_raises_unexpected(raw, typ): + """Fuzz: validation either succeeds or raises ValueError with known reasons.""" + try: + sticker_utils.validate_sticker_payload(raw, typ) + except ValueError: + pass + + +@settings(max_examples=500, deadline=None) +@given(raw=st.binary(min_size=0, max_size=4096)) +def test_detect_image_format_from_magic_fuzz_never_raises(raw): + out = sticker_utils.detect_image_format_from_magic(raw) + assert out is None or out in {"png", "jpeg", "gif", "webp", "bmp"} + + +@settings(max_examples=100, deadline=None) +@given( + doc=st.dictionaries( + keys=st.text(max_size=16), + values=st.recursive( + st.none() + | st.booleans() + | st.floats(allow_nan=False) + | st.text(max_size=40) + | st.binary(max_size=48), + lambda children: ( + st.lists(children, max_size=3) + | st.dictionaries(st.text(max_size=6), children, max_size=3) + ), + max_leaves=12, + ), + max_size=8, + ), +) +def test_validate_export_document_fuzz_never_raises_unexpected(doc): + try: + sticker_utils.validate_export_document(doc) + except ValueError: + pass diff --git a/tests/backend/test_stickers_dao.py b/tests/backend/test_stickers_dao.py new file mode 100644 index 0000000..916bac9 --- /dev/null +++ b/tests/backend/test_stickers_dao.py @@ -0,0 +1,154 @@ +"""Tests for UserStickersDAO and schema migration for user_stickers.""" + +import base64 + +import pytest + +from meshchatx.src.backend.database import Database +from meshchatx.src.backend.database.schema import DatabaseSchema + + +@pytest.fixture +def db(tmp_path): + path = tmp_path / "t.db" + database = Database(str(path)) + database.initialize() + assert DatabaseSchema.LATEST_VERSION >= 44 + return database + + +def _tiny_png(): + return bytes([0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A]) + b"\x00" * 32 + + +def test_insert_and_list(db): + identity = "ab" * 16 + raw = _tiny_png() + row = db.stickers.insert(identity, "one", "png", raw, None) + assert row is not None + assert row["id"] >= 1 + listed = db.stickers.list_for_identity(identity) + assert len(listed) == 1 + assert listed[0]["image_size"] == len(raw) + + +def test_duplicate_returns_none(db): + identity = "cd" * 16 + raw = _tiny_png() + r1 = db.stickers.insert(identity, None, "png", raw, None) + assert r1 is not None + r2 = db.stickers.insert(identity, None, "png", raw, None) + assert r2 is None + assert db.stickers.count_for_identity(identity) == 1 + + +def test_delete_and_delete_all(db): + identity = "ef" * 16 + db.stickers.insert(identity, "a", "png", _tiny_png(), None) + row = db.stickers.list_for_identity(identity)[0] + assert db.stickers.delete(row["id"], identity) is True + assert len(db.stickers.list_for_identity(identity)) == 0 + + db.stickers.insert(identity, "b", "png", _tiny_png(), None) + db.stickers.insert( + identity, + "c", + "png", + _tiny_png() + b"x", + None, + ) + n = db.stickers.delete_all_for_identity(identity) + assert n == 2 + assert db.stickers.count_for_identity(identity) == 0 + + +def test_wrong_identity_delete(db): + identity = "01" * 16 + other = "02" * 16 + row = db.stickers.insert(identity, None, "png", _tiny_png(), None) + assert db.stickers.delete(row["id"], other) is False + + +def test_get_row_image_blob(db): + identity = "aa" * 16 + raw = _tiny_png() + ins = db.stickers.insert(identity, "n", "png", raw, "msg1") + full = db.stickers.get_row(ins["id"], identity) + assert full["image_blob"] == raw + assert full["source_message_hash"] == "msg1" + + +def test_update_name(db): + identity = "bb" * 16 + ins = db.stickers.insert(identity, "old", "png", _tiny_png(), None) + assert db.stickers.update_name(ins["id"], identity, "new") is True + row = db.stickers.get_row(ins["id"], identity) + assert row["name"] == "new" + + +def test_export_and_import_roundtrip(db): + identity = "cc" * 16 + raw = _tiny_png() + db.stickers.insert(identity, "a", "png", raw, None) + payloads = db.stickers.export_payloads_for_identity(identity) + assert len(payloads) == 1 + assert base64.b64decode(payloads[0]["image_bytes"]) == raw + + other = "dd" * 16 + items = [ + { + "name": "x", + "image_type": "png", + "image_bytes_b64": payloads[0]["image_bytes"], + "source_message_hash": None, + }, + ] + r = db.stickers.import_payloads(other, items, replace_duplicates=False) + assert r["imported"] == 1 + assert db.stickers.count_for_identity(other) == 1 + r2 = db.stickers.import_payloads(other, items, replace_duplicates=False) + assert r2["skipped_duplicates"] == 1 + + +def test_import_replace_duplicate(db): + identity = "ee" * 16 + raw = _tiny_png() + db.stickers.insert(identity, "first", "png", raw, None) + items = [ + { + "name": "second", + "image_type": "png", + "image_bytes_b64": base64.b64encode(raw).decode("ascii"), + "source_message_hash": None, + }, + ] + r = db.stickers.import_payloads(identity, items, replace_duplicates=True) + assert r["imported"] == 1 + rows = db.stickers.list_for_identity(identity) + assert len(rows) == 1 + assert rows[0]["name"] == "second" + + +def test_import_invalid_base64_skipped(db): + identity = "ff" * 16 + items = [ + { + "name": "x", + "image_type": "png", + "image_bytes_b64": "!!!not-base64!!!", + "source_message_hash": None, + }, + ] + r = db.stickers.import_payloads(identity, items, replace_duplicates=False) + assert r["skipped_invalid"] >= 1 + + +def test_sticker_limit(db, monkeypatch): + from meshchatx.src.backend import sticker_utils + + monkeypatch.setattr(sticker_utils, "MAX_STICKERS_PER_IDENTITY", 2) + identity = "11" * 16 + db.stickers.insert(identity, None, "png", _tiny_png(), None) + db.stickers.insert(identity, None, "png", _tiny_png() + b"y", None) + with pytest.raises(ValueError, match="sticker_limit"): + db.stickers.insert(identity, None, "png", _tiny_png() + b"z", None)