fix(iast): eval wrapping error [backport 3.10] (#13997)

Backport fe421d2 from #13976 to 3.10.

Python treats eval() differently depending on whether the 'locals'
argument is provided.

If only 'globals' is given, any new functions or variables are stored in
the globals.

If both 'globals' and 'locals' are given, new definitions go into locals
instead.

To ensure we can access any functions created by eval(), we copy them
from locals to globals when needed.

## Checklist
- [x] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [x] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

Co-authored-by: Alberto Vara <[email protected]>

Author: github-actions[bot]

ddtrace/appsec/_iast/taint_sinks/code_injection.py

@@ -4,6 +4,7 @@
 from ddtrace.appsec._constants import IAST
 from ddtrace.appsec._constants import IAST_SPAN_TAGS
 from ddtrace.appsec._iast._logs import iast_error
+from ddtrace.appsec._iast._logs import iast_propagation_sink_point_debug_log
 from ddtrace.appsec._iast._metrics import _set_metric_iast_executed_sink
 from ddtrace.appsec._iast._metrics import _set_metric_iast_instrumented_sink
 from ddtrace.appsec._iast._patch_modules import WrapFunctonsForIAST
@@ -32,7 +33,6 @@ def patch():
     # TODO: wrap exec functions is very dangerous because it needs and modifies locals and globals from the original
     #  function
     # iast_funcs.wrap_function("builtins", "exec", _iast_coi)
-
     iast_funcs.patch()
 
     _set_metric_iast_instrumented_sink(VULN_CODE_INJECTION)
@@ -47,27 +47,49 @@ def _iast_coi(wrapped, instance, args, kwargs):
     if len(args) >= 1:
         _iast_report_code_injection(args[0])
 
-    caller_frame = None
-    if len(args) > 1:
-        func_globals = args[1]
-    elif kwargs.get("globals"):
-        func_globals = kwargs.get("globals")
-    else:
-        frames = inspect.currentframe()
-        caller_frame = frames.f_back
-        func_globals = caller_frame.f_globals
-
-    if len(args) > 2:
-        func_locals = args[2]
-    elif kwargs.get("locals"):
-        func_locals = kwargs.get("locals")
-    else:
-        if caller_frame is None:
+    try:
+        caller_frame = None
+        if len(args) > 1:
+            func_globals = args[1]
+        elif kwargs.get("globals"):
+            func_globals = kwargs.get("globals")
+        else:
             frames = inspect.currentframe()
             caller_frame = frames.f_back
-        func_locals = caller_frame.f_locals
+            func_globals = caller_frame.f_globals
+
+        func_locals_copy_to_check = None
+        if len(args) > 2:
+            func_locals = args[2]
+        elif kwargs.get("locals"):
+            func_locals = kwargs.get("locals")
+        else:
+            if caller_frame is None:
+                frames = inspect.currentframe()
+                caller_frame = frames.f_back
+            func_locals = caller_frame.f_locals
+            func_locals_copy_to_check = func_locals.copy() if func_locals else None
+    except Exception as e:
+        iast_propagation_sink_point_debug_log(f"Error in _iast_code_injection. {e}")
+        return wrapped(*args, **kwargs)
+
+    res = wrapped(args[0], func_globals, func_locals)
+
+    # We need to perform this `func_locals_copy_to_check` check because of how Python handles `eval()` depending
+    # on whether `locals` is provided. If we have code like `def evaluate(n): return n` and we call
+    # `eval(code, my_globals)`, the new function will be stored in `my_globals["evaluate"]`. However, if we call
+    # `eval(code, my_globals, my_locals)`, then the function will be stored in `my_locals["evaluate"]`. So, if `eval()`
+    # is called without a `locals` argument, we need to transfer the newly created code from the local
+    # context to the global one.
+    try:
+        if func_locals_copy_to_check is not None:
+            diff_keys = set(func_locals) - set(func_locals_copy_to_check)
+            for key in diff_keys:
+                func_globals[key] = func_locals[key]
+    except Exception as e:
+        iast_propagation_sink_point_debug_log(f"Error in _iast_code_injection. {e}")
 
-    return wrapped(args[0], func_globals, func_locals)
+    return res
 
 
 def _iast_report_code_injection(code_string: Text):

hatch.toml

@@ -316,6 +316,7 @@ dependencies = [
     "pytest-xdist",
     "hypothesis",
     "requests",
+    "babel",
     "SQLAlchemy",
     "psycopg2-binary~=2.9.9",
     "pymysql",

releasenotes/notes/iast-fix-code-injection-eval-827d51948f29b52a.yaml

@@ -0,0 +1,8 @@
+---
+fixes:
+  - |
+    Code Security (IAST): Improved compatibility with eval() when used with custom globals and locals. When 
+    instrumenting eval(), Python behaves differently depending on whether locals is passed. If both globals and 
+    locals are provided, new functions are stored in the locals dictionary. This fix ensures any dynamically 
+    defined functions (e.g., via eval(code, globals, locals)) are accessible by copying them from locals to 
+    globals when necessary. This resolves issues with third-party libraries (like babel) that rely on this behavior.

tests/appsec/app.py

@@ -22,6 +22,7 @@
 from tests.appsec.iast_packages.packages.pkg_annotated_types import pkg_annotated_types
 from tests.appsec.iast_packages.packages.pkg_asn1crypto import pkg_asn1crypto
 from tests.appsec.iast_packages.packages.pkg_attrs import pkg_attrs
+from tests.appsec.iast_packages.packages.pkg_babel import pkg_babel
 from tests.appsec.iast_packages.packages.pkg_beautifulsoup4 import pkg_beautifulsoup4
 from tests.appsec.iast_packages.packages.pkg_cachetools import pkg_cachetools
 from tests.appsec.iast_packages.packages.pkg_certifi import pkg_certifi
@@ -101,6 +102,7 @@
 app.register_blueprint(pkg_annotated_types)
 app.register_blueprint(pkg_asn1crypto)
 app.register_blueprint(pkg_attrs)
+app.register_blueprint(pkg_babel)
 app.register_blueprint(pkg_beautifulsoup4)
 app.register_blueprint(pkg_cachetools)
 app.register_blueprint(pkg_certifi)

tests/appsec/iast/fixtures/taint_sinks/code_injection.py

@@ -38,6 +38,11 @@ def pt_eval_lambda_globals(origin_string):
     return r
 
 
+def pt_eval_add_data_to_global(origin_string, globals_namespace, locals_namespace):
+    code = compile(origin_string, "<rule>", "exec")
+    eval(code, globals_namespace, locals_namespace)
+
+
 def pt_literal_eval(origin_string):
     r = literal_eval(origin_string)
     return r

tests/appsec/iast/taint_sinks/test_code_injection.py

@@ -113,7 +113,6 @@ def test_code_injection_eval_globals_locals_override(iast_context_defaults):
 
 def test_code_injection_eval_lambda(iast_context_defaults):
     """Validate globals and locals of the function"""
-    mod = _iast_patched_module("tests.appsec.iast.fixtures.taint_sinks.code_injection")
 
     def pt_eval_lambda_no_tainted(fun):
         return eval("lambda v,fun=fun:not fun(v)")
@@ -126,7 +125,6 @@ def is_true_no_tainted(value):
 
 def test_code_injection_eval_globals_kwargs_lambda(iast_context_defaults):
     """Validate globals and locals of the function"""
-
     code_string = "square(5)"
 
     tainted_string = taint_pyobject(
@@ -151,7 +149,6 @@ def test_code_injection_eval_globals_kwargs_lambda(iast_context_defaults):
 
 
 def test_code_injection_literal_eval(iast_context_defaults):
-    mod = _iast_patched_module("tests.appsec.iast.fixtures.taint_sinks.code_injection")
     code_string = "[1, 2, 3]"
 
     tainted_string = taint_pyobject(
@@ -162,3 +159,43 @@ def test_code_injection_literal_eval(iast_context_defaults):
     data = get_iast_reporter()
 
     assert data is None
+
+
+def test_code_injection_eval_add_data_to_global(iast_context_defaults):
+    mod = _iast_patched_module("tests.appsec.iast.fixtures.taint_sinks.code_injection")
+    code_string = """
+def evaluate(n):
+    return func(n)
+"""
+
+    tainted_string = taint_pyobject(
+        code_string, source_name="path", source_value=code_string, source_origin=OriginType.PATH
+    )
+    namespace_globals = {}
+    namespace_locals = None
+    mod.pt_eval_add_data_to_global(tainted_string, namespace_globals, namespace_locals)
+    assert namespace_globals["evaluate"]
+    data = get_iast_reporter()
+
+    assert data is None
+
+
+def test_code_injection_eval_add_data_to_local(iast_context_defaults):
+    mod = _iast_patched_module("tests.appsec.iast.fixtures.taint_sinks.code_injection")
+    code_string = """
+def evaluate(n):
+    return func(n)
+"""
+
+    tainted_string = taint_pyobject(
+        code_string, source_name="path", source_value=code_string, source_origin=OriginType.PATH
+    )
+    namespace_globals = {}
+    namespace_locals = {"var1": "value1", "var2": "value2"}
+    mod.pt_eval_add_data_to_global(tainted_string, namespace_globals, namespace_locals)
+    assert namespace_locals["evaluate"]
+    with pytest.raises(KeyError):
+        _ = namespace_globals["evaluate"]
+    data = get_iast_reporter()
+
+    assert data is None

tests/appsec/iast_packages/packages/pkg_babel.py

@@ -0,0 +1,40 @@
+"""
+idna==3.6
+
+https://pypi.org/project/idna/
+"""
+
+from flask import Blueprint
+from flask import request
+
+from .utils import ResultResponse
+
+
+pkg_babel = Blueprint("package_babel", __name__)
+
+
+@pkg_babel.route("/babel")
+def pkg_babel_view():
+    from babel.plural import to_python
+
+    response = ResultResponse(request.args.get("package_param"))
+    func = to_python({"one": "n is 1", "few": "n in 2..4"})
+
+    response.result1 = func(int(request.args.get("package_param")))
+    return response.json()
+
+
+@pkg_babel.route("/babel_propagation")
+def pkg_babel_propagation_view():
+    from babel import Locale
+
+    from ddtrace.appsec._iast._taint_tracking._taint_objects_base import is_pyobject_tainted
+
+    response = ResultResponse(request.args.get("package_param"))
+    if not is_pyobject_tainted(response.package_param):
+        response.result1 = "Error: package_param is not tainted"
+        return response.json()
+
+    response.result1 = Locale("en", "US").currency_formats["standard"]
+    response.result1 = "OK" if is_pyobject_tainted(response.package_param) else "Error: result is not tainted"
+    return response.json()

tests/appsec/iast_packages/test_packages.py

@@ -918,6 +918,15 @@ def create_venv(self):
         import_name="annotated_types",
         skip_python_version=[(3, 8)],
     ),
+    PackageForTesting(
+        "babel",
+        "2.17.0",
+        "15",
+        "other",
+        "",
+        test_import=False,
+        test_propagation=True,
+    ),
 ]
 
 # Sort by name so it's easier to infer the progress of the tests

tests/appsec/integrations/fixtures/patch_babel.py

@@ -0,0 +1,28 @@
+import types
+
+
+code1 = """
+def evaluate_locale():
+    from babel import Locale
+    response = Locale('en', 'US').currency_formats['standard']
+    return response
+"""
+
+
+def babel_locale():
+    module_name = "test_babel"
+    compiled_code = compile(code1, "tests/appsec/integrations/packages_tests/", mode="exec")
+    module_changed = types.ModuleType(module_name)
+    exec(compiled_code, module_changed.__dict__)  # define evaluate
+    result = eval("evaluate_locale()", module_changed.__dict__)  # llama evaluate
+    return f"OK:{result}"
+
+
+def babel_to_python(n):
+    def evaluate_to_python(n):
+        from babel.plural import to_python
+
+        func = to_python({"one": "n is 1", "few": "n in 2..4"})
+        return func(n)
+
+    return f"OK:{evaluate_to_python(n)}"

tests/appsec/integrations/packages_tests/conftest.py

@@ -1,5 +1,6 @@
 import pytest
 
+from ddtrace.appsec._iast.taint_sinks.code_injection import patch as code_injection_patch
 from ddtrace.contrib.internal.psycopg.patch import patch as psycopg_patch
 from ddtrace.contrib.internal.psycopg.patch import unpatch as psycopg_unpatch
 from ddtrace.contrib.internal.sqlalchemy.patch import patch as sqlalchemy_patch
@@ -19,6 +20,7 @@ def iast_create_context():
         sqlalchemy_patch()
         psycopg_patch()
         sqli_sqlite_patch()
+        code_injection_patch()
         _start_iast_context_and_oce()
         yield
         _end_iast_context_and_oce()