fix: Don't add rosapi services to glob patterns if we pass empty services_glob parameter (backport #1115) (#1117)

* Support empty lists for globs

* Parse empty globs to None instead of empty list

* Default to empty globs instead of [*]

* fix: prevent appending to services_glob if it is None

Author: bjsowa

rosbridge_library/src/rosbridge_library/capabilities/advertise.py

@@ -123,7 +123,7 @@ def advertise(self, message: dict[str, Any]) -> None:
         latch: bool = message.get("latch", False)
         queue_size: int = message.get("queue_size", 100)
 
-        if self.topics_glob:
+        if self.topics_glob is not None:
             self.protocol.log("debug", "Topic security glob enabled, checking topic: " + topic)
             match = False
             for glob in self.topics_glob:

rosbridge_library/src/rosbridge_library/capabilities/call_service.py

@@ -91,7 +91,7 @@ def call_service(self, message: dict[str, Any]) -> None:
         args: list | dict[str, Any] = message.get("args", [])
         timeout: float = message.get("timeout", self.default_timeout)
 
-        if self.services_glob:
+        if self.services_glob is not None:
             self.protocol.log(
                 "debug", "Service security glob enabled, checking service: " + service
             )

rosbridge_library/src/rosbridge_library/capabilities/publish.py

@@ -67,7 +67,7 @@ def publish(self, message: dict[str, Any]) -> None:
         latch: bool = message.get("latch", False)
         queue_size: int = message.get("queue_size", 100)
 
-        if self.topics_glob:
+        if self.topics_glob is not None:
             self.protocol.log("debug", "Topic security glob enabled, checking topic: " + topic)
             match = False
             for glob in self.topics_glob:

rosbridge_library/src/rosbridge_library/capabilities/send_action_goal.py

@@ -103,7 +103,7 @@ def send_action_goal(self, message: dict) -> None:
         compression: str = message.get("compression", "none")
         args: list | dict[str, Any] = message.get("args", [])
 
-        if self.actions_glob:
+        if self.actions_glob is not None:
             self.protocol.log("debug", f"Action security glob enabled, checking action: {action}")
             match = False
             for glob in self.actions_glob:

rosbridge_library/src/rosbridge_library/capabilities/subscribe.py

@@ -272,7 +272,7 @@ def subscribe(self, msg: dict[str, Any]) -> None:
         # Make the subscription
         topic: str = msg["topic"]
 
-        if self.topics_glob is not None and self.topics_glob:
+        if self.topics_glob is not None:
             self.protocol.log("debug", "Topic security glob enabled, checking topic: " + topic)
             match = False
             for glob in self.topics_glob:

rosbridge_library/src/rosbridge_library/capabilities/unadvertise_action.py

@@ -61,7 +61,7 @@ def unadvertise_action(self, message: dict[str, Any]) -> None:
         # parse the message
         action_name: str = message["action"]
 
-        if self.actions_glob:
+        if self.actions_glob is not None:
             self.protocol.log(
                 "debug",
                 f"Action security glob enabled, checking action: {action_name}",

rosbridge_library/src/rosbridge_library/capabilities/unadvertise_service.py

@@ -29,7 +29,7 @@ def unadvertise_service(self, message: dict[str, Any]) -> None:
         # parse the message
         service_name: str = message["service"]
 
-        if self.services_glob:
+        if self.services_glob is not None:
             self.protocol.log(
                 "debug",
                 "Service security glob enabled, checking service: " + service_name,

rosbridge_server/scripts/rosbridge_websocket.py

@@ -94,9 +94,9 @@ def shutdown_hook() -> None:
         "Ignored if bson_only_mode is True.",
     ),
     ("bson_only_mode", bool, False, "Use BSON only mode for messages."),
-    ("topics_glob", str, "[*]", "Glob patterns for topics publish/subscribe."),
-    ("services_glob", str, "[*]", "Glob patterns for services call/advertise."),
-    ("actions_glob", str, "[*]", "Glob patterns for actions send/advertise."),
+    ("topics_glob", str, "", "Glob patterns for topics publish/subscribe."),
+    ("services_glob", str, "", "Glob patterns for services call/advertise."),
+    ("actions_glob", str, "", "Glob patterns for actions send/advertise."),
     ("call_services_in_new_thread", bool, False, "Call services in a new threads."),
     ("default_call_service_timeout", float, 0.0, "Default timeout for service calls."),
     ("send_action_goals_in_new_thread", bool, False, "Send action goals in a new threads."),
@@ -112,13 +112,15 @@ def parse_args() -> argparse.Namespace:
     return parser.parse_args(args)
 
 
-def parse_glob_string(glob_string: str) -> list[str]:
+def parse_glob_string(glob_string: str) -> list[str] | None:
     """
     Parse a glob string into a list of patterns.
 
     The glob string is expected to be in the format: "['pattern1', 'pattern2']"
     """
-    if not glob_string or glob_string == "[]":
+    if not glob_string:
+        return None
+    if glob_string == "[]":
         return []
     # Remove the surrounding brackets and split by comma
     return [s.strip().strip("'") for s in glob_string[1:-1].split(",") if s.strip()]
@@ -136,7 +138,8 @@ def __init__(self) -> None:
 
         # To be able to access the list of topics and services,
         # you must be able to access the rosapi services.
-        self.protocol_parameters["services_glob"].append("/rosapi/*")
+        if self.protocol_parameters["services_glob"] is not None:
+            self.protocol_parameters["services_glob"].append("/rosapi/*")
 
         RosbridgeWebSocket.protocol_parameters = self.protocol_parameters
         RosbridgeWebSocket.use_compression = self.use_compression