Add extensive tests for notification reliability and LXMF field hardening

tests/backend/test_notifications.py

@@ -342,3 +342,239 @@ def test_notification_spike_fuzzing(db, num_notifs):
 
     notifications = db.misc.get_notifications(limit=num_notifs)
     assert len(notifications) == num_notifs
+
+
+# ---------------------------------------------------------------------------
+# Extensive notification reliability tests
+# ---------------------------------------------------------------------------
+
+
+class TestNotificationReliability:
+    """Comprehensive tests to verify notifications are accurate, reliable,
+    and never produce false positives."""
+
+    def test_unread_count_matches_actual_unread(self, db):
+        """Unread count must exactly match unviewed notifications."""
+        for i in range(10):
+            db.misc.add_notification(f"type{i}", f"h{i}", f"t{i}", f"c{i}")
+
+        assert db.misc.get_unread_notification_count() == 10
+
+        ids = [n["id"] for n in db.misc.get_notifications()]
+        db.misc.mark_notifications_as_viewed([ids[0], ids[1], ids[2]])
+        assert db.misc.get_unread_notification_count() == 7
+
+        db.misc.mark_notifications_as_viewed()
+        assert db.misc.get_unread_notification_count() == 0
+
+    def test_marking_empty_list_marks_all(self, db):
+        """Marking with an empty list (falsy) falls through to mark-all behavior."""
+        db.misc.add_notification("t", "h", "title", "content")
+        db.misc.mark_notifications_as_viewed([])
+        assert db.misc.get_unread_notification_count() == 0
+
+    def test_marking_nonexistent_ids_is_safe(self, db):
+        """Marking IDs that don't exist should not crash."""
+        db.misc.mark_notifications_as_viewed([99999, 88888, 77777])
+        assert db.misc.get_unread_notification_count() == 0
+
+    def test_double_mark_viewed_idempotent(self, db):
+        """Marking the same notification as viewed twice is safe."""
+        db.misc.add_notification("t", "h", "title", "content")
+        nid = db.misc.get_notifications()[0]["id"]
+        db.misc.mark_notifications_as_viewed([nid])
+        db.misc.mark_notifications_as_viewed([nid])
+        assert db.misc.get_unread_notification_count() == 0
+
+    def test_no_ghost_notifications_after_clear(self, db):
+        """After marking all as viewed, unread count must be zero and stay zero."""
+        for i in range(5):
+            db.misc.add_notification(f"type{i}", "h", "t", "c")
+        db.misc.mark_notifications_as_viewed()
+        assert db.misc.get_unread_notification_count() == 0
+        assert len(db.misc.get_notifications(filter_unread=True)) == 0
+
+    def test_interleaved_add_and_mark(self, db):
+        """Adding and marking interleaved must produce correct counts."""
+        db.misc.add_notification("a", "h1", "t1", "c1")
+        db.misc.add_notification("b", "h2", "t2", "c2")
+        nid1 = db.misc.get_notifications()[0]["id"]
+        db.misc.mark_notifications_as_viewed([nid1])
+
+        db.misc.add_notification("c", "h3", "t3", "c3")
+        assert db.misc.get_unread_notification_count() == 2
+
+    def test_limit_does_not_affect_unread_count(self, db):
+        """get_unread_notification_count is independent of query limit."""
+        for i in range(20):
+            db.misc.add_notification(f"t{i}", "h", "title", "content")
+        limited = db.misc.get_notifications(limit=5)
+        assert len(limited) == 5
+        assert db.misc.get_unread_notification_count() == 20
+
+    def test_missed_call_creates_exactly_one_notification(self, mock_app):
+        """A single missed call must produce exactly one notification."""
+        mock_app.database.misc.provider.execute("DELETE FROM notifications")
+
+        caller = MagicMock()
+        caller.hash = b"caller_hash_32_bytes_long_012345"
+        mock_app.telephone_manager.call_is_incoming = True
+        mock_app.telephone_manager.call_status_at_end = 4
+        mock_app.telephone_manager.call_start_time = time.time() - 5
+        mock_app.telephone_manager.call_was_established = False
+
+        mock_app.on_telephone_call_ended(caller)
+        notifications = mock_app.database.misc.get_notifications()
+        assert len(notifications) == 1
+        assert notifications[0]["type"] == "telephone_missed_call"
+
+    def test_established_call_creates_no_notification(self, mock_app):
+        """An established (answered) call must not create a missed-call notification."""
+        mock_app.database.misc.provider.execute("DELETE FROM notifications")
+
+        caller = MagicMock()
+        caller.hash = b"caller_hash_32_bytes_long_012345"
+        mock_app.telephone_manager.call_is_incoming = True
+        mock_app.telephone_manager.call_status_at_end = 6
+        mock_app.telephone_manager.call_start_time = time.time() - 60
+        mock_app.telephone_manager.call_was_established = True
+
+        mock_app.on_telephone_call_ended(caller)
+        notifications = mock_app.database.misc.get_notifications()
+        assert len(notifications) == 0
+
+    def test_outgoing_call_creates_no_missed_notification(self, mock_app):
+        """An outgoing call that ends without answer must not produce a missed-call notification."""
+        mock_app.database.misc.provider.execute("DELETE FROM notifications")
+
+        callee = MagicMock()
+        callee.hash = b"callee_hash_32_bytes_long_012345"
+        mock_app.telephone_manager.call_is_incoming = False
+        mock_app.telephone_manager.call_status_at_end = 2
+        mock_app.telephone_manager.call_start_time = time.time() - 10
+        mock_app.telephone_manager.call_was_established = False
+
+        mock_app.on_telephone_call_ended(callee)
+        notifications = mock_app.database.misc.get_notifications()
+        assert len(notifications) == 0
+
+    def test_voicemail_creates_exactly_one_notification(self, mock_app):
+        """A voicemail must create exactly one notification."""
+        mock_app.database.misc.provider.execute("DELETE FROM notifications")
+        mock_app.on_new_voicemail_received("abc123", "User X", 30)
+        notifications = mock_app.database.misc.get_notifications()
+        assert len(notifications) == 1
+        assert notifications[0]["type"] == "telephone_voicemail"
+
+    def test_rapid_missed_calls_unique_notifications(self, mock_app):
+        """Multiple rapid missed calls from different callers produce separate notifications."""
+        mock_app.database.misc.provider.execute("DELETE FROM notifications")
+
+        for i in range(10):
+            caller = MagicMock()
+            caller.hash = f"caller_{i:032d}".encode()[:32]
+            mock_app.telephone_manager.call_is_incoming = True
+            mock_app.telephone_manager.call_status_at_end = 4
+            mock_app.telephone_manager.call_start_time = time.time() - 1
+            mock_app.telephone_manager.call_was_established = False
+            mock_app.on_telephone_call_ended(caller)
+
+        notifications = mock_app.database.misc.get_notifications()
+        assert len(notifications) == 10
+        assert all(n["type"] == "telephone_missed_call" for n in notifications)
+
+    def test_mixed_notification_types_correct_counts(self, mock_app):
+        """Mix of missed calls and voicemails produces correct per-type counts."""
+        mock_app.database.misc.provider.execute("DELETE FROM notifications")
+
+        for _ in range(3):
+            caller = MagicMock()
+            caller.hash = b"caller_hash_32_bytes_long_012345"
+            mock_app.telephone_manager.call_is_incoming = True
+            mock_app.telephone_manager.call_status_at_end = 4
+            mock_app.telephone_manager.call_start_time = time.time()
+            mock_app.telephone_manager.call_was_established = False
+            mock_app.on_telephone_call_ended(caller)
+
+        for _ in range(2):
+            mock_app.on_new_voicemail_received("vm_hash", "VmUser", 10)
+
+        notifications = mock_app.database.misc.get_notifications()
+        missed = [n for n in notifications if n["type"] == "telephone_missed_call"]
+        voicemails = [n for n in notifications if n["type"] == "telephone_voicemail"]
+        assert len(missed) == 3
+        assert len(voicemails) == 2
+        assert mock_app.database.misc.get_unread_notification_count() == 5
+
+    def test_mark_viewed_reduces_count_precisely(self, mock_app):
+        """Marking specific notifications as viewed reduces count by exactly that many."""
+        mock_app.database.misc.provider.execute("DELETE FROM notifications")
+
+        for i in range(5):
+            mock_app.database.misc.add_notification(f"t{i}", f"h{i}", f"t{i}", f"c{i}")
+
+        assert mock_app.database.misc.get_unread_notification_count() == 5
+
+        ids = [n["id"] for n in mock_app.database.misc.get_notifications()]
+        mock_app.database.misc.mark_notifications_as_viewed([ids[0], ids[2]])
+        assert mock_app.database.misc.get_unread_notification_count() == 3
+
+    def test_notification_ordering(self, db):
+        """Notifications should be returned in reverse chronological order (newest first)."""
+        import time as _time
+
+        for i in range(5):
+            db.misc.add_notification(f"type{i}", "h", f"title_{i}", f"content_{i}")
+            _time.sleep(0.01)
+
+        notifications = db.misc.get_notifications()
+        for j in range(len(notifications) - 1):
+            assert notifications[j]["timestamp"] >= notifications[j + 1]["timestamp"]
+
+
+@settings(deadline=None, suppress_health_check=[HealthCheck.function_scoped_fixture])
+@given(
+    num_add=st.integers(min_value=0, max_value=50),
+    num_mark=st.integers(min_value=0, max_value=50),
+)
+def test_add_mark_invariant(db, num_add, num_mark):
+    """Invariant: unread count == total - marked, never negative."""
+    db.provider.execute("DELETE FROM notifications")
+    for i in range(num_add):
+        db.misc.add_notification(f"t{i}", "h", "t", "c")
+
+    all_notifs = db.misc.get_notifications()
+    ids_to_mark = [n["id"] for n in all_notifs[:num_mark]]
+    if ids_to_mark:
+        db.misc.mark_notifications_as_viewed(ids_to_mark)
+
+    expected = max(0, num_add - len(ids_to_mark))
+    assert db.misc.get_unread_notification_count() == expected
+
+
+@settings(deadline=None, suppress_health_check=[HealthCheck.function_scoped_fixture])
+@given(
+    operations=st.lists(
+        st.tuples(
+            st.sampled_from(["add", "mark_one", "mark_all"]),
+            st.text(min_size=1, max_size=20),
+        ),
+        min_size=1,
+        max_size=30,
+    ),
+)
+def test_random_notification_operations(db, operations):
+    """Fuzz random sequences of add/mark operations; count must never go negative."""
+    db.provider.execute("DELETE FROM notifications")
+    for op, payload in operations:
+        if op == "add":
+            db.misc.add_notification("type", payload, "title", "content")
+        elif op == "mark_one":
+            notifs = db.misc.get_notifications(limit=1)
+            if notifs:
+                db.misc.mark_notifications_as_viewed([notifs[0]["id"]])
+        elif op == "mark_all":
+            db.misc.mark_notifications_as_viewed()
+
+    count = db.misc.get_unread_notification_count()
+    assert count >= 0

tests/backend/test_security_fuzzing.py

@@ -2196,3 +2196,306 @@ def test_lxmf_display_name_parsing_regression():
 
     # None input
     assert parse_lxmf_display_name(None, default_value="Fallback") == "Fallback"
+
+
+class TestLxmfFieldHardening:
+    """Tests for LXMF message field type safety and edge cases."""
+
+    def test_file_attachment_non_list(self, mock_app):
+        """File attachments field as non-list should not crash."""
+        mock_message = MagicMock()
+        mock_message.source_hash = os.urandom(16)
+        mock_message.destination_hash = os.urandom(16)
+        mock_message.hash = os.urandom(16)
+        mock_message.content = b"test"
+        mock_message.title = b""
+        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.delivery_attempts = 1
+        mock_message.next_delivery_attempt = None
+        mock_message.get_fields.return_value = {
+            LXMF.FIELD_FILE_ATTACHMENTS: "not-a-list",
+        }
+        from meshchatx.src.backend.lxmf_utils import convert_lxmf_message_to_dict
+
+        result = convert_lxmf_message_to_dict(mock_message)
+        assert isinstance(result, dict)
+
+    def test_file_attachment_item_wrong_type(self, mock_app):
+        """File attachment items that are not [name, data] pairs should not crash."""
+        mock_message = MagicMock()
+        mock_message.source_hash = os.urandom(16)
+        mock_message.destination_hash = os.urandom(16)
+        mock_message.hash = os.urandom(16)
+        mock_message.content = b"test"
+        mock_message.title = b""
+        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.delivery_attempts = 1
+        mock_message.next_delivery_attempt = None
+        mock_message.get_fields.return_value = {
+            LXMF.FIELD_FILE_ATTACHMENTS: [42, None, "string", {"a": "b"}],
+        }
+        from meshchatx.src.backend.lxmf_utils import convert_lxmf_message_to_dict
+
+        try:
+            convert_lxmf_message_to_dict(mock_message)
+        except (TypeError, IndexError, KeyError):
+            pass
+
+    def test_image_field_non_list(self, mock_app):
+        """Image field as unexpected type should not crash."""
+        mock_message = MagicMock()
+        mock_message.source_hash = os.urandom(16)
+        mock_message.destination_hash = os.urandom(16)
+        mock_message.hash = os.urandom(16)
+        mock_message.content = b"test"
+        mock_message.title = b""
+        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.delivery_attempts = 1
+        mock_message.next_delivery_attempt = None
+        mock_message.get_fields.return_value = {
+            LXMF.FIELD_IMAGE: 42,
+        }
+        from meshchatx.src.backend.lxmf_utils import convert_lxmf_message_to_dict
+
+        try:
+            convert_lxmf_message_to_dict(mock_message)
+        except (TypeError, IndexError, KeyError):
+            pass
+
+    @settings(
+        suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None
+    )
+    @given(
+        fields_data=st.dictionaries(
+            keys=st.integers(min_value=0, max_value=20),
+            values=st.one_of(
+                st.none(),
+                st.booleans(),
+                st.integers(),
+                st.text(max_size=200),
+                st.binary(max_size=200),
+                st.lists(st.text(max_size=50), max_size=5),
+            ),
+            max_size=10,
+        ),
+    )
+    def test_arbitrary_fields_do_not_crash(self, mock_app, fields_data):
+        """Fuzz arbitrary LXMF fields dict to ensure no crashes."""
+        mock_message = MagicMock()
+        mock_message.source_hash = os.urandom(16)
+        mock_message.destination_hash = os.urandom(16)
+        mock_message.hash = os.urandom(16)
+        mock_message.content = b"ok"
+        mock_message.title = b""
+        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.delivery_attempts = 1
+        mock_message.next_delivery_attempt = None
+        mock_message.get_fields.return_value = fields_data
+        from meshchatx.src.backend.lxmf_utils import convert_lxmf_message_to_dict
+
+        try:
+            convert_lxmf_message_to_dict(mock_message)
+        except (TypeError, IndexError, KeyError, ValueError):
+            pass
+
+
+class TestStrangerAttachmentBlocking:
+    """Tests for stranger attachment stripping logic."""
+
+    def _make_mock_message(self, source_hash=None, with_attachments=True):
+        mock_msg = MagicMock()
+        mock_msg.source_hash = source_hash or os.urandom(16)
+        mock_msg.destination_hash = os.urandom(16)
+        mock_msg.hash = os.urandom(16)
+        mock_msg.content = b"hello from stranger"
+        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 = {}
+        if with_attachments:
+            fields[LXMF.FIELD_FILE_ATTACHMENTS] = [
+                [b"malware.exe", b"\x00" * 100],
+            ]
+        mock_msg.get_fields.return_value = fields
+        mock_msg.fields = fields
+        return mock_msg
+
+    def test_stranger_attachments_stripped_when_enabled(self, mock_app):
+        """Attachments from non-contacts should be stripped when setting is on."""
+        source_hash = os.urandom(16)
+        mock_msg = self._make_mock_message(source_hash=source_hash)
+
+        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
+        )
+
+    def test_contact_attachments_not_stripped(self, mock_app):
+        """Attachments from contacts should NOT be stripped."""
+        source_hash = os.urandom(16)
+        mock_msg = self._make_mock_message(source_hash=source_hash)
+
+        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)
+        mock_app.db_upsert_lxmf_message.assert_called_once()
+        fields = mock_msg.get_fields()
+        assert LXMF.FIELD_FILE_ATTACHMENTS in fields
+
+    def test_setting_disabled_allows_stranger_attachments(self, mock_app):
+        """When setting is off, stranger attachments pass through."""
+        source_hash = os.urandom(16)
+        mock_msg = self._make_mock_message(source_hash=source_hash)
+
+        mock_app.config.block_attachments_from_strangers.get.return_value = False
+        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()
+        fields = mock_msg.get_fields()
+        assert LXMF.FIELD_FILE_ATTACHMENTS in fields
+
+    def test_text_message_from_stranger_still_delivered(self, mock_app):
+        """Text-only messages from strangers are delivered normally."""
+        source_hash = os.urandom(16)
+        mock_msg = self._make_mock_message(
+            source_hash=source_hash, with_attachments=False
+        )
+
+        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()
+
+
+class TestBlockAllFromStrangers:
+    """Tests for the block-everything-from-strangers feature."""
+
+    def _make_mock_message(self, source_hash=None, with_attachments=False):
+        mock_msg = MagicMock()
+        mock_msg.source_hash = source_hash or os.urandom(16)
+        mock_msg.destination_hash = os.urandom(16)
+        mock_msg.hash = os.urandom(16)
+        mock_msg.content = b"hello from stranger"
+        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 = {}
+        if with_attachments:
+            fields[LXMF.FIELD_FILE_ATTACHMENTS] = [[b"file.txt", b"data"]]
+        mock_msg.get_fields.return_value = fields
+        mock_msg.fields = fields
+        return mock_msg
+
+    def test_stranger_message_dropped_when_enabled(self, mock_app):
+        """Text message from a stranger is silently dropped when block_all is on."""
+        mock_msg = self._make_mock_message()
+        mock_app.config.block_all_from_strangers.get.return_value = True
+        mock_app.config.block_attachments_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_not_called()
+
+    def test_contact_message_delivered_when_block_all_enabled(self, mock_app):
+        """Messages from contacts pass through even when block_all is on."""
+        mock_msg = self._make_mock_message()
+        mock_app.config.block_all_from_strangers.get.return_value = True
+        mock_app.config.block_attachments_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)
+        mock_app.db_upsert_lxmf_message.assert_called_once()
+
+    def test_stranger_with_attachments_dropped(self, mock_app):
+        """Message with attachments from stranger is dropped entirely, not just stripped."""
+        mock_msg = self._make_mock_message(with_attachments=True)
+        mock_app.config.block_all_from_strangers.get.return_value = True
+        mock_app.config.block_attachments_from_strangers.get.return_value = True
+        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_not_called()
+
+    def test_disabled_allows_stranger_messages(self, mock_app):
+        """When block_all is off, stranger messages are delivered."""
+        mock_msg = self._make_mock_message()
+        mock_app.config.block_all_from_strangers.get.return_value = False
+        mock_app.config.block_attachments_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()