Handle invalid UTF-8 as an OSC parse error

pythonosc/parsing/osc_types.py

@@ -86,6 +86,8 @@ def get_string(dgram: bytes, start_index: int) -> Tuple[str, int]:
 
         data_str = dgram[start_index : start_index + offset]
         return data_str.decode("utf-8"), start_index + total_len
+    except UnicodeDecodeError as error:
+        raise ParseError(f"Could not decode OSC string: {error}") from error
     except IndexError as ie:
         raise ParseError(f"Could not parse datagram {ie}")
     except TypeError as te:

pythonosc/test/parsing/test_osc_types.py

@@ -34,6 +34,20 @@ def test_get_string_raises_on_wrong_dgram(self):
         for case in cases:
             self.assertRaises(osc_types.ParseError, osc_types.get_string, case, 0)
 
+    def test_get_string_rejects_invalid_utf8(self):
+        for value in (b"\xff", b"\xc3", b"\xc0\xaf", b"\xed\xa0\x80"):
+            with self.subTest(value=value):
+                dgram = value + b"\x00" * (4 - len(value) % 4)
+                with self.assertRaises(osc_types.ParseError):
+                    osc_types.get_string(dgram, 0)
+
+    def test_get_string_preserves_valid_utf8_and_offset(self):
+        encoded = osc_types.write_string("café 🎵")
+        self.assertEqual(
+            ("café 🎵", 4 + len(encoded)),
+            osc_types.get_string(b"skip" + encoded + b"tail", 4),
+        )
+
     def test_get_string_raises_when_datagram_too_short(self):
         self.assertRaises(osc_types.ParseError, osc_types.get_string, b"abc\x00", 1)
 

pythonosc/test/test_dispatcher.py

@@ -14,6 +14,27 @@ def sort(lst):
 
         return self.assertSequenceEqual(sort(expected), sort(result))
 
+    def test_invalid_utf8_is_ignored_before_next_valid_packet(self):
+        received = []
+        self.dispatcher.map("/test", lambda address, *args: received.append(args))
+        for packet in (
+            b"/\xff\x00\x00,s\x00\x00ok\x00\x00",
+            b"/test\x00\x00\x00,\xff\x00\x00",
+            b"/test\x00\x00\x00,s\x00\x00\xff\x00\x00\x00",
+        ):
+            with self.subTest(packet=packet):
+                self.assertEqual(
+                    [],
+                    self.dispatcher.call_handlers_for_packet(
+                        packet, ("127.0.0.1", 1234)
+                    ),
+                )
+        self.assertEqual([], received)
+        self.dispatcher.call_handlers_for_packet(
+            b"/test\x00\x00\x00,s\x00\x00ok\x00\x00", ("127.0.0.1", 1234)
+        )
+        self.assertEqual([("ok",)], received)
+
     def test_empty_by_default(self):
         self.sortAndAssertSequenceEqual(
             [], self.dispatcher.handlers_for_address("/test")