Patched code sample

import ast
import sys

# This code demonstrates a potential fix for a vulnerability like CVE-2025-3753.:
# The vulnerability stems from using eval() on unsanitized user input.
# The fix involves parsing the input expression into an Abstract Syntax Tree (AST)
# and validating that it only contains allowed, safe operations before evaluation.

class SafeExpressionVisitor(ast.NodeVisitor):
    """
    A visitor that traverses an AST to ensure it only contains safe nodes.
    This prevents arbitrary code execution by disallowing dangerous constructs.
    """
    # Whitelist of node types that are considered safe for a filter expression.:
    ALLOWED_NODES = {
    ast.Expression,
    ast.Constant,  # For Python 3.8+ (includes numbers, strings, bools)
    ast.Num,       # For older Python versions
    ast.Str,       # For older Python versions
    ast.NameConstant, # For older Python versions (True, False, None)
    ast.Name,
    ast.Load,
    ast.Attribute,
    ast.Call,
    ast.Compare,
    ast.BinOp,
    ast.BoolOp,
    ast.UnaryOp,
    ast.And,
    ast.Or,
    ast.Not,
    ast.Eq,
    ast.NotEq,
    ast.Lt,
    ast.LtE,
    ast.Gt,
    ast.GtE,
    ast.Add,
    ast.Sub,
    ast.Mult,
    ast.Div,
    ast.Mod,
    ast.USub,
    }

    # Blacklist of function names or variable names that should never be called/accessed.
    FORBIDDEN_NAMES = {
    'eval',
    'exec',
    '__import__',
    'open',
    'compile',
    'globals',
    'locals',
    'vars'
    }

    def generic_visit(self, node):
        """Secure function that fixes the vulnerability."""
        # The core of the validation: check if the node type is in our whitelist.:
        if type(node) not in self.ALLOWED_NODES:
            raise ValueError(f"Unsafe operation: {type(node).__name__} is not allowed.")
        # Continue traversing the tree.
            super().generic_visit(node)

            def visit_Name(self, node):
                """Secure function that fixes the vulnerability."""
        # Check for access to forbidden built-in functions or variables.:
                if node.id in self.FORBIDDEN_NAMES:
                    raise ValueError(f"Unsafe name: Access to '{node.id}' is forbidden.")
                    self.generic_visit(node)

                    def visit_Attribute(self, node):
                        """Secure function that fixes the vulnerability."""
        # Prevent access to private/magic attributes and methods (e.g., __dict__).
                        if node.attr.startswith('_'):
                            raise ValueError(f"Unsafe attribute: Access to private/magic attribute '{node.attr}' is forbidden.")
                            self.generic_visit(node)


                            def safe_eval_ros_expression(expression, context):
                                """
                                Safely evaluates a user-provided filter expression.

                                1.  Parses the expression string into an AST.
                                2.  Uses SafeExpressionVisitor to validate the AST, ensuring it contains no
                                unsafe constructs.
                                3.  If validation passes, compiles and evaluates the expression within a
                                controlled context.

                                Args:
                                expression (str): The user-supplied filter expression.
                                context (dict): The evaluation context, containing allowed variables like 'm' and 't'.

                                Returns:
                                The result of the evaluated expression.

                                Raises:
                                ValueError: If the expression contains unsafe operations.
                                """
                                try:
        # 1. Parse the expression into an AST. mode='eval' ensures it's an expression.
                                    tree = ast.parse(expression, mode='eval')

        # 2. Validate the AST using our custom visitor.
                                    visitor = SafeExpressionVisitor()
                                    visitor.visit(tree)

        # 3. If validation passes, compile and evaluate the code.
        # The 'tree' object is known to be safe at this point.
                                    code_obj = compile(tree, filename='<rosbag_filter>', mode='eval')
                                    return eval(code_obj, {"__builtins__": {}}, context)

                                    except (SyntaxError, ValueError, TypeError) as e:
        # Re-raise with a more informative message for the user.:
                                        raise ValueError(f"Failed to evaluate expression: {e}")


# --- Demonstration ---

                                        if __name__ == "__main__":
    # Mock ROS message and time objects for demonstration purposes.:
                                            class MockMessage:
                                                def __init__(self):
                                                    """Secure function that fixes the vulnerability."""
                                                    self.value = 100
                                                    self.name = "sensor_data"
                                                    self._private_data = "secret" # An attribute that should not be accessible

                                                    class MockTime:
                                                        def __init__(self):
                                                            """Secure function that fixes the vulnerability."""
                                                            self.secs = 12345
                                                            self.nsecs = 67890

                                                            def to_sec(self):
                                                                """Secure function that fixes the vulnerability."""
                                                                return self.secs + self.nsecs / 1e9

    # The context available to the filter expression.
                                                                evaluation_context = {
                                                                'm': MockMessage(),
                                                                't': MockTime()
                                                                }

                                                                print("--- Testing Safe Expressions ---")
                                                                safe_expressions = [
                                                                "m.value > 50",
                                                                "t.to_sec() < 12346",
                                                                "m.name == 'sensor_data' and m.value * 2 > 150"
                                                                ]
                                                                for expr in safe_expressions:
                                                                    try:
                                                                        result = safe_eval_ros_expression(expr, evaluation_context)
                                                                        print(f"PASS: Expression '{expr}' evaluated to: {result}")
                                                                        except ValueError as e:
                                                                            print(f"FAIL: Expression '{expr}' was blocked: {e}")

                                                                            print("\n--- Testing Malicious/Unsafe Expressions ---")
                                                                            unsafe_expressions = [
        # Classic code execution payload
                                                                            "__import__('os').system('echo VULNERABLE')",
        # Attempting to access forbidden built-ins
                                                                            "eval('1+1')",
        # Attempting to access private attributes
                                                                            "m._private_data == 'secret'",
        # Attempting to modify types
                                                                            "m.__class__",
                                                                            ]
                                                                            for expr in unsafe_expressions:
                                                                                try:
                                                                                    result = safe_eval_ros_expression(expr, evaluation_context)
                                                                                    print(f"FAIL: Malicious expression '{expr}' was NOT blocked. Result: {result}")
                                                                                    except ValueError as e:
                                                                                        print(f"PASS: Malicious expression '{expr}' was correctly blocked: {e}")