fixed regexp parsing, enforced ',' on begining of datagrams as per spec and also fixed time parsing (when > 24h)

Author: attwad

pythonosc/dispatcher.py

@@ -182,28 +182,62 @@ def handlers_for_address(
     ) -> Generator[Handler, None, None]:
         """Yields handlers matching an address
 
-
         Args:
             address_pattern: Address to match
 
         Returns:
             Generator yielding Handlers matching address_pattern
         """
-        # First convert the address_pattern into a matchable regexp.
-        # '?' in the OSC Address Pattern matches any single character.
-        # Let's consider numbers and _ "characters" too here, it's not said
-        # explicitly in the specification but it sounds good.
-        escaped_address_pattern = re.escape(address_pattern)
-        pattern = escaped_address_pattern.replace("\\?", "\\w?")
-        # '*' in the OSC Address Pattern matches any sequence of zero or more
-        # characters.
-        pattern = pattern.replace("\\*", "[\\w|\\+]*")
-        # The rest of the syntax in the specification is like the re module so
-        # we're fine.
-        pattern = f"{pattern}$"
-        patterncompiled = re.compile(pattern)
-        matched = False
+        # Convert OSC Address Pattern to a Python regular expression.
+        # Spec: https://opensoundcontrol.stanford.edu/spec-1_0.html#osc-address-patterns
+
+        pattern = "^"
+        i = 0
+        while i < len(address_pattern):
+            c = address_pattern[i]
+            if c == "*":
+                pattern += "[^/]*"
+            elif c == "?":
+                pattern += "[^/]"
+            elif c == "[":
+                pattern += "["
+                i += 1
+                if i < len(address_pattern) and address_pattern[i] == "!":
+                    pattern += "^"
+                    i += 1
+                while i < len(address_pattern) and address_pattern[i] != "]":
+                    if address_pattern[i] in r"\^$.|()+*?":
+                        pattern += "\\"
+                    pattern += address_pattern[i]
+                    i += 1
+                pattern += "]"
+            elif c == "{":
+                pattern += "("
+                i += 1
+                while i < len(address_pattern) and address_pattern[i] != "}":
+                    char = address_pattern[i]
+                    if char == ",":
+                        pattern += "|"
+                    elif char in r"\^$.|()[]+*?":
+                        pattern += "\\" + char
+                    else:
+                        pattern += char
+                    i += 1
+                pattern += ")"
+            elif c in r"\^$.|()[]+?":
+                pattern += "\\" + c
+            else:
+                pattern += c
+            i += 1
+        pattern += "$"
 
+        try:
+            patterncompiled = re.compile(pattern)
+        except re.error:
+            # If the pattern is invalid, it won't match anything.
+            return
+
+        matched = False
         for addr, handlers in self._map.items():
             if patterncompiled.match(addr) or (
                 ("*" in addr)

pythonosc/osc_message.py

@@ -34,8 +34,12 @@ def _parse_datagram(self) -> None:
 
             # Get the parameters types.
             type_tag, index = osc_types.get_string(self._dgram, index)
-            if type_tag.startswith(","):
-                type_tag = type_tag[1:]
+            if not type_tag.startswith(","):
+                raise ParseError(
+                    f"OSC Type Tag String must start with a comma, got: {type_tag}"
+                )
+
+            type_tag = type_tag[1:]
 
             params = []  # type: List[Any]
             param_stack = [params]

pythonosc/osc_message_builder.py

@@ -121,7 +121,7 @@ def _get_arg_type(self, arg_value: ArgValue) -> Union[str, Any]:
         elif arg_value is False:
             arg_type = self.ARG_TYPE_FALSE
         elif isinstance(arg_value, int):
-            if arg_value.bit_length() > 32:
+            if arg_value.bit_length() > 31:
                 arg_type = self.ARG_TYPE_INT64
             else:
                 arg_type = self.ARG_TYPE_INT

pythonosc/parsing/osc_types.py

@@ -71,24 +71,21 @@ def get_string(dgram: bytes, start_index: int) -> Tuple[str, int]:
         raise ParseError("start_index < 0")
     offset = 0
     try:
-        if (
-            len(dgram) > start_index + _STRING_DGRAM_PAD
-            and dgram[start_index + _STRING_DGRAM_PAD] == _EMPTY_STR_DGRAM
-        ):
-            return "", start_index + _STRING_DGRAM_PAD
         while dgram[start_index + offset] != 0:
             offset += 1
-        # Align to a byte word.
-        if (offset) % _STRING_DGRAM_PAD == 0:
-            offset += _STRING_DGRAM_PAD
-        else:
-            offset += -offset % _STRING_DGRAM_PAD
-        # Python slices do not raise an IndexError past the last index,
-        # do it ourselves.
-        if offset > len(dgram[start_index:]):
+
+        # OSC spec: "followed by a null, followed by 0-3 additional null characters
+        # to make the total number of bits a multiple of 32"
+        # This means the total length (including the first null) must be a multiple of 4.
+        total_len = offset + 1
+        if total_len % 4 != 0:
+            total_len += 4 - (total_len % 4)
+
+        if start_index + total_len > len(dgram):
             raise ParseError("Datagram is too short")
+
         data_str = dgram[start_index : start_index + offset]
-        return data_str.replace(b"\x00", b"").decode("utf-8"), start_index + offset
+        return data_str.decode("utf-8"), start_index + total_len
     except IndexError as ie:
         raise ParseError(f"Could not parse datagram {ie}")
     except TypeError as te:
@@ -214,11 +211,8 @@ def get_timetag(dgram: bytes, start_index: int) -> Tuple[Tuple[datetime, int], i
         timetag, _ = get_uint64(dgram, start_index)
         seconds, fraction = ntp.parse_timestamp(timetag)
 
-        hours, seconds = seconds // 3600, seconds % 3600
-        minutes, seconds = seconds // 60, seconds % 60
-
         utc = datetime.combine(ntp._NTP_EPOCH, datetime.min.time()) + timedelta(
-            hours=hours, minutes=minutes, seconds=seconds
+            seconds=seconds
         )
 
         return (utc, fraction), start_index + _TIMETAG_DGRAM_LEN

pythonosc/test/test_osc_message.py

@@ -22,15 +22,7 @@
     b"\x00\x00\x00\x08stuff\x00\x00\x00"
 )  # b"stuff\x00\x00\x00"
 
-_DGRAM_ALL_NON_STANDARD_TYPES_OF_PARAMS = (
-    b"/SYNC\x00\x00\x00"
-    b"T"  # True
-    b"F"  # False
-    b"N"  # Nil
-    b"[]th\x00"  # Empty array
-    b"\x00\x00\x00\x00\x00\x00\x00\x00"
-    b"\x00\x00\x00\xe8\xd4\xa5\x10\x00"  # 1000000000000
-)
+_DGRAM_ALL_NON_STANDARD_TYPES_OF_PARAMS = b"/SYNC\x00\x00\x00,TFN[]th\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xe8\xd4\xa5\x10\x00"
 
 _DGRAM_COMPLEX_ARRAY_PARAMS = (
     b"/SYNC\x00\x00\x00"

pythonosc/udp_client.py

@@ -58,7 +58,9 @@ def __init__(
     def __enter__(self) -> "UDPClient":
         return self
 
-    def __exit__(self, exc_type: type | None, exc_val: Exception | None, exc_tb: object | None) -> None:
+    def __exit__(
+        self, exc_type: type | None, exc_val: Exception | None, exc_tb: object | None
+    ) -> None:
         self.close()
 
     def close(self) -> None: