feat: Add a new eval helper function

Author: sfayer

src/DIRAC/Core/Utilities/SaferEval.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python3
+
+import ast
+
+
+def saferEval(obj_str, max_len=2048):
+    """This function adds an extra length check around literal_eval.
+    On python3.11 and above (which has a recursion guard), this should
+    be safe enough for use on general authenticated user input.
+
+    Note: This doesn't handle all of the cases of eval, such as
+          datetime as those are technically executing code to
+          instantiate the non-base objects.
+    """
+    # Ensure input is a string
+    obj_str = str(obj_str)
+    if len(obj_str) > max_len:
+        raise ValueError(f"Object string is too long (>{max_len} bytes)")
+    try:
+        return ast.literal_eval(obj_str)
+    except (ValueError, TypeError, SyntaxError):
+        # This covers all of the cases where the string is wrong (unclosed brackets...)
+        # or contains disallowed items like function calls or non-expression.
+        raise ValueError("Syntax error processing object expression")
+    except (MemoryError, RecursionError):
+        # This is encountered if the object is nested too deeply and other structures
+        # that are probably malicious.
+        raise ValueError("Object expression too large")
+    except Exception:
+        # There are no other possible exceptions at the time of writing,
+        # this is to catch any added in future python versions.
+        raise ValueError("Unknown error processing object expression")

src/DIRAC/Core/Utilities/test/Test_SaferEval.py

@@ -0,0 +1,269 @@
+"""Test saferEval function from DIRAC.Core.Utilities.saferEval"""
+
+
+import time
+import unittest
+
+from DIRAC.Core.Utilities.SaferEval import saferEval
+
+
+class Test_saferEval(unittest.TestCase):
+    # --- Normal Python literal inputs ---
+
+    def test_none(self):
+        self.assertEqual(saferEval("None"), None)
+
+    def test_bool_true(self):
+        self.assertEqual(saferEval("True"), True)
+
+    def test_bool_false(self):
+        self.assertEqual(saferEval("False"), False)
+
+    def test_int_zero(self):
+        self.assertEqual(saferEval("0"), 0)
+
+    def test_int_positive(self):
+        self.assertEqual(saferEval("42"), 42)
+
+    def test_int_negative(self):
+        self.assertEqual(saferEval("-17"), -17)
+
+    def test_int_hex(self):
+        self.assertEqual(saferEval("0xFF"), 0xFF)
+
+    def test_int_octal(self):
+        self.assertEqual(saferEval("0o77"), 0o77)
+
+    def test_int_binary(self):
+        self.assertEqual(saferEval("0b1010"), 0b1010)
+
+    def test_float(self):
+        self.assertEqual(saferEval("3.14"), 3.14)
+
+    def test_float_scientific(self):
+        self.assertEqual(saferEval("1e10"), 1e10)
+
+    def test_complex(self):
+        self.assertEqual(saferEval("1j"), 1j)
+
+    def test_str_double_quoted(self):
+        self.assertEqual(saferEval('"hello"'), "hello")
+
+    def test_str_single_quoted(self):
+        self.assertEqual(saferEval("'hello'"), "hello")
+
+    def test_str_escaped(self):
+        self.assertEqual(saferEval("'a\\nb'"), "a\nb")
+
+    def test_str_raw(self):
+        self.assertEqual(saferEval(r"r'\n'"), r"\n")
+
+    def test_str_unicode(self):
+        self.assertEqual(saferEval('"hello 🌍"'), "hello 🌍")
+
+    def test_bytes(self):
+        self.assertEqual(saferEval("b'bytes'"), b"bytes")
+
+    def test_bytes_escape(self):
+        self.assertEqual(saferEval("b'\\xff'"), b"\xff")
+
+    def test_empty_list(self):
+        self.assertEqual(saferEval("[]"), [])
+
+    def test_list_mixed(self):
+        self.assertEqual(saferEval("[1, 'two', True, None]"), [1, "two", True, None])
+
+    def test_empty_tuple(self):
+        self.assertEqual(saferEval("()"), ())
+
+    def test_singleton_tuple(self):
+        self.assertEqual(saferEval("(1,)"), (1,))
+
+    def test_tuple(self):
+        self.assertEqual(saferEval("(1, 2, 3)"), (1, 2, 3))
+
+    def test_empty_dict(self):
+        self.assertEqual(saferEval("{}"), {})
+
+    def test_dict(self):
+        self.assertEqual(saferEval("{'a': 1, 'b': 2}"), {"a": 1, "b": 2})
+
+    def test_set(self):
+        self.assertEqual(saferEval("{1, 2, 3}"), {1, 2, 3})
+
+    def test_nested_list(self):
+        self.assertEqual(saferEval("[[1, 2], [3, 4]]"), [[1, 2], [3, 4]])
+
+    def test_nested_dict(self):
+        result = saferEval("{'a': {'b': {'c': [1, 2]}}}")
+        self.assertEqual(result, {"a": {"b": {"c": [1, 2]}}})
+
+    # --- Invalid inputs (should raise ValueError) ---
+
+    def test_code_execution(self):
+        self.assertRaises(ValueError, saferEval, "__import__('os').system('id')")
+
+    def test_code_execution_open(self):
+        self.assertRaises(ValueError, saferEval, "open('/etc/passwd').read()")
+
+    def test_function_call(self):
+        self.assertRaises(ValueError, saferEval, "list()")
+
+    def test_variable_reference(self):
+        self.assertRaises(ValueError, saferEval, "foo")
+
+    def test_class_instantiation(self):
+        self.assertRaises(ValueError, saferEval, "datetime.datetime.now()")
+
+    def test_lambda(self):
+        self.assertRaises(ValueError, saferEval, "lambda x: x")
+
+    def test_dict_comprehension(self):
+        self.assertRaises(ValueError, saferEval, "{k: v for k, v in []}")
+
+    def test_generator_expression(self):
+        self.assertRaises(ValueError, saferEval, "(x for x in [])")
+
+    def test_name_comparison(self):
+        self.assertRaises(ValueError, saferEval, "x == y")
+
+    def test_binary_ops(self):
+        self.assertRaises(ValueError, saferEval, "1 + 2")
+
+    def test_attribute_access(self):
+        self.assertRaises(ValueError, saferEval, "().__class__")
+
+    def test_subscript(self):
+        self.assertRaises(ValueError, saferEval, "x[0]")
+
+    def test_slice(self):
+        self.assertRaises(ValueError, saferEval, "[1,2][1:]")
+
+    def test_starred(self):
+        self.assertRaises(ValueError, saferEval, "*1")
+
+    # --- Security edge cases ---
+
+    def test_builtin_name(self):
+        self.assertRaises(ValueError, saferEval, "builtins.open")
+
+    def test_class_constructor(self):
+        self.assertRaises(ValueError, saferEval, "object()")
+
+    def test_custom_class(self):
+        self.assertRaises(ValueError, saferEval, "MyList()")
+
+    def test_fstring(self):
+        self.assertRaises(ValueError, saferEval, "f'{1+2}'")
+
+    def test_decorator(self):
+        self.assertRaises(ValueError, saferEval, "@decorator")
+
+    def test_assert_statement(self):
+        self.assertRaises(ValueError, saferEval, "assert True")
+
+    def test_return_statement(self):
+        self.assertRaises(ValueError, saferEval, "return 42")
+
+    def test_augmented_assignment(self):
+        self.assertRaises(ValueError, saferEval, "x += 1")
+
+    def test_with_statement(self):
+        self.assertRaises(ValueError, saferEval, "with open('x') as f: pass")
+
+    def test_for_loop(self):
+        self.assertRaises(ValueError, saferEval, "for x in []: pass")
+
+    def test_try_except(self):
+        self.assertRaises(ValueError, saferEval, "try: pass\nexcept: pass")
+
+    def test_import(self):
+        self.assertRaises(ValueError, saferEval, "import os")
+
+    def test_from_import(self):
+        self.assertRaises(ValueError, saferEval, "from os import path")
+
+    def test_del_statement(self):
+        self.assertRaises(ValueError, saferEval, "del x")
+
+    def test_raise_statement(self):
+        self.assertRaises(ValueError, saferEval, "raise ValueError('x')")
+
+    def test_yield(self):
+        self.assertRaises(ValueError, saferEval, "yield 1")
+
+    def test_await(self):
+        self.assertRaises(ValueError, saferEval, "await something")
+
+    def test_named_expression(self):
+        self.assertRaises(ValueError, saferEval, "(x := 1)")
+
+    def test_positional_only_lambda(self):
+        self.assertRaises(ValueError, saferEval, "(lambda x, /: x)(1)")
+
+    def test_power_arithmetic(self):
+        # 10**200 is an expression, not a literal
+        self.assertRaises(ValueError, saferEval, "10**200")
+
+    # --- Recursion limit ---
+
+    def test_deep_nesting(self):
+        depth = 2000
+        s = "[" * depth + "1" + "]" * depth
+        try:
+            result = saferEval(s)
+            self.assertEqual(result, 1)
+        except (ValueError, RecursionError):
+            # Either ast.literal_eval's recursion guard (3.11+) or Python's
+            # built-in recursion limit catches it — both are safe.
+            pass
+
+    def test_deep_nesting_within_max_len(self):
+        depth = 500
+        s = "[" * depth + "1" + "]" * depth
+        try:
+            saferEval(s)
+        except (ValueError, RecursionError):
+            pass
+
+    # --- Max length ---
+
+    def test_max_len_boundary(self):
+        self.assertEqual(saferEval("42"), 42)
+
+    def test_max_len_exceeded(self):
+        s = "1" * 2049
+        self.assertRaises(ValueError, saferEval, s, 2048)
+
+    def test_max_len_custom(self):
+        self.assertRaises(ValueError, saferEval, "[1, 2, 3]", 5)
+
+    def test_max_len_under_custom(self):
+        self.assertEqual(saferEval("[1, 2, 3]", 10), [1, 2, 3])
+
+    # --- Memory explosion prevention ---
+
+    def test_large_string_literal(self):
+        s = "'" + "a" * 3000 + "'"
+        self.assertRaises(ValueError, saferEval, s, 2048)
+
+    def test_large_list(self):
+        self.assertRaises(ValueError, saferEval, str([1] * 3000), 2048)
+
+    # --- Performance ---
+
+    def test_normal_dict_performance(self):
+        parts = ['"key": 0'] + ['"k{0}": {0}'.format(i) for i in range(50)]
+        s = "{" + ", ".join(parts) + "}"
+        start = time.time()
+        saferEval(s, 2048)
+        self.assertLess(time.time() - start, 0.1)
+
+    def test_normal_list_performance(self):
+        s = "[" + ",".join(str(i) for i in range(50)) + "]"
+        start = time.time()
+        saferEval(s, 2048)
+        self.assertLess(time.time() - start, 0.1)
+
+    if __name__ == "__main__":
+        unittest.main()