Revert "Patch CVE-2024-42002" (#1000)

This reverts commit 644310d.

Signed-off-by: Michael Carroll <[email protected]>

Author: mjcarroll

ros2topic/package.xml

@@ -12,7 +12,6 @@
 
   <license>Apache License 2.0</license>
   <license>BSD-3-Clause</license>  <!-- ros2topic/verb/delay.py|hz.py|bw.py are BSD -->
-  <license>MIT License</license> <!-- ros2topic/eval uses MIT License -->
 
   <author email="[email protected]">Aditya Pande</author>
   <author email="[email protected]">Dirk Thomas</author>

ros2topic/ros2topic/eval/__init__.py

@@ -48,7 +48,6 @@
 from ros2topic.api import get_msg_class
 from ros2topic.api import positive_int
 from ros2topic.api import TopicNameCompleter
-from ros2topic.eval import base_eval_model, Expr
 from ros2topic.verb import VerbExtension
 
 DEFAULT_WINDOW_SIZE = 10000
@@ -92,69 +91,18 @@ def main(self, *, args):
         return main(args)
 
 
-def _get_nested_messages(msg_class):
-    all_attributes = list(msg_class.__slots__)
-    for attr in msg_class.__slots__:
-        value = getattr(msg_class, attr)
-        if hasattr(value, '__slots__'):
-            nested_messages = _get_nested_messages(value)
-            all_attributes.extend(nested_messages)
-    return all_attributes
-
-
-def _setup_base_safe_eval():
-    safe_eval_model = base_eval_model.clone()
-
-    # extend base_eval_model
-    safe_eval_model.nodes.extend(['Call', 'Attribute', 'List', 'Tuple', 'Dict', 'Set',
-                                  'ListComp', 'DictComp', 'SetComp', 'comprehension',
-                                  'Mult', 'Pow', 'boolop', 'mod', 'Invert',
-                                  'Is', 'IsNot', 'FloorDiv', 'If', 'For'])
-
-    # allow-list safe Python built-in functions
-    safe_builtins = [
-        'abs', 'all', 'any', 'bin', 'bool', 'chr', 'cmp', 'divmod', 'enumerate',
-        'float', 'format', 'hex', 'id', 'int', 'isinstance', 'issubclass',
-        'len', 'list', 'long', 'max', 'min', 'ord', 'pow', 'range', 'reversed',
-        'round', 'slice', 'sorted', 'str', 'sum', 'tuple', 'type', 'unichr',
-        'unicode', 'xrange', 'zip', 'filter', 'dict', 'set', 'next'
-    ]
-
-    safe_eval_model.allowed_functions.extend(safe_builtins)
-    return safe_eval_model
-
-def _setup_safe_eval(safe_eval_model, msg_class, topic):
-    # allow-list topic builtins, msg attributes
-    topic_builtins = [i for i in dir(topic) if not i.startswith('_')]
-    safe_eval_model.attributes.extend(topic_builtins)
-    # recursively get all nested message attributes
-    msg_attributes = _get_nested_messages(msg_class)
-    safe_eval_model.attributes.extend(msg_attributes)
-    return safe_eval_model
-
-
 def main(args):
-    with DirectNode(args) as node:
-        topics = args.topic_name
+    topics = args.topic_name
+    if args.filter_expr:
+        def expr_eval(expr):
+            def eval_fn(m):
+                return eval(expr)
+            return eval_fn
+        filter_expr = expr_eval(args.filter_expr)
+    else:
         filter_expr = None
-        # set up custom safe eval model for filter expression
-        if args.filter_expr:
-            safe_eval_model = _setup_base_safe_eval()
-            for topic in topics:
-                msg_class = get_msg_class(
-                    node, topic, blocking=True, include_hidden_topics=True)
-                if msg_class is None:
-                    continue
-
-                safe_eval_model = _setup_safe_eval(safe_eval_model, msg_class, topic)
-
-            def expr_eval(expr):
-                def eval_fn(m):
-                    safe_expression = Expr(expr, model=safe_eval_model)
-                    return eval(safe_expression.code)
-                return eval_fn
-            filter_expr = expr_eval(args.filter_expr)
 
+    with DirectNode(args) as node:
         _rostopic_hz(node.node, topics, qos_args=args, window_size=args.window_size,
                      filter_expr=filter_expr, use_wtime=args.use_wtime)
 

ros2topic/ros2topic/verb/hz.py

@@ -938,19 +938,6 @@ def test_filtered_topic_hz(self):
         average_rate = float(average_rate_line_pattern.match(head_line).group(1))
         assert math.isclose(average_rate, 0.5, rel_tol=1e-2)
 
-        # check that use of eval() on hz verb cannot be exploited
-        try:
-            self.launch_topic_command(
-                arguments=[
-                    'hz',
-                    '--filter',
-                    '__import__("os").system("cat /etc/passwd")',
-                    '/chatter'
-                ]
-            )
-        except ValueError as e:
-            self.assertIn('Attribute system is not allowed', str(e))
-
     @launch_testing.markers.retry_on_failure(times=5, delay=1)
     def test_topic_bw(self):
         with self.launch_topic_command(arguments=['bw', '/defaults']) as topic_command: