fix(iast): ast patching side effect (#7638) [backport 2.0] (#7657)

IAST: This fix resolves an issue where, at AST patching to replace code
with IAST aspects, the original function/method was passed as an extra
parameter for accurate patching (i.e. check the function inside the
aspect), and this strategy lead to unintentionally triggers side effects
in methods obtained from an expression (like ``decode`` in
``file.read(n).decode()``), resulting in unexpected multiple calls to
the expression (``file.read(n)`` in the example).

- [x] Change(s) are motivated and described in the PR description.
- [x] Testing strategy is described if automated tests are not included
in the PR.
- [x] Risk is outlined (performance impact, potential for breakage,
maintainability, etc).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] [Library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
are followed. If no release note is required, add label
`changelog/no-changelog`.
- [x] Documentation is included (in-code, generated user docs, [public
corp docs](https://github.com/DataDog/documentation/)).
- [x] Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist

- [ ] Title is accurate.
- [ ] No unnecessary changes are introduced.
- [ ] Description motivates each change.
- [ ] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes unless absolutely necessary.
- [ ] Testing strategy adequately addresses listed risk(s).
- [ ] Change is maintainable (easy to change, telemetry, documentation).
- [ ] Release note makes sense to a user of the library.
- [ ] Reviewer has explicitly 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)
- [ ] If this PR touches code that signs or publishes builds or
packages, or handles credentials of any kind, I've requested a review
from `@DataDog/security-design-and-guidance`.
- [ ] This PR doesn't touch any of that.

---------

Co-authored-by: Alberto Vara <[email protected]>
(cherry picked from commit 601fd6a)

Author: gnufede

ddtrace/appsec/_iast/_ast/visitor.py

@@ -443,8 +443,9 @@ def visit_Call(self, call_node):  # type: (ast.Call) -> Any
                 # Send 1 as flag_added_args value
                 call_node.args.insert(0, self._int_constant(call_node, 1))
 
-                # Insert original method as first parameter (a.b.c.method)
-                call_node.args = self._add_original_function_as_arg(call_node, False)
+                # Insert None as first parameter instead of a.b.c.method
+                # to avoid unexpected side effects such as a.b.read(4).method
+                call_node.args.insert(0, self._none_constant(call_node))
 
                 # Create a new Name node for the replacement and set it as node.func
                 call_node.func = self._attr_node(call_node, aspect)

ddtrace/appsec/_iast/_taint_tracking/aspects.py

@@ -52,13 +52,16 @@ def add_aspect(op1, op2):
 
 
 def str_aspect(orig_function, flag_added_args, *args, **kwargs):
-    # type: (Callable, int, Any, Any) -> str
-    if orig_function != builtin_str:
-        if flag_added_args > 0:
-            args = args[flag_added_args:]
-        return orig_function(*args, **kwargs)
+    # type: (Optional[Callable], int, Any, Any) -> str
+    if orig_function:
+        if orig_function != builtin_str:
+            if flag_added_args > 0:
+                args = args[flag_added_args:]
+            return orig_function(*args, **kwargs)
+        result = builtin_str(*args, **kwargs)
+    else:
+        result = args[0].str(*args[1:], **kwargs)
 
-    result = builtin_str(*args, **kwargs)
     if isinstance(args[0], TEXT_TYPES) and is_pyobject_tainted(args[0]):
         try:
             if isinstance(args[0], (bytes, bytearray)):
@@ -75,13 +78,16 @@ def str_aspect(orig_function, flag_added_args, *args, **kwargs):
 
 
 def bytes_aspect(orig_function, flag_added_args, *args, **kwargs):
-    # type: (Callable, int, Any, Any) -> bytes
-    if orig_function != builtin_bytes:
-        if flag_added_args > 0:
-            args = args[flag_added_args:]
-        return orig_function(*args, **kwargs)
+    # type: (Optional[Callable], int, Any, Any) -> bytes
+    if orig_function:
+        if orig_function != builtin_bytes:
+            if flag_added_args > 0:
+                args = args[flag_added_args:]
+            return orig_function(*args, **kwargs)
+        result = builtin_bytes(*args, **kwargs)
+    else:
+        result = args[0].bytes(*args[1:], **kwargs)
 
-    result = builtin_bytes(*args, **kwargs)
     if isinstance(args[0], TEXT_TYPES) and is_pyobject_tainted(args[0]):
         try:
             taint_pyobject_with_ranges(result, tuple(get_ranges(args[0])))
@@ -91,13 +97,16 @@ def bytes_aspect(orig_function, flag_added_args, *args, **kwargs):
 
 
 def bytearray_aspect(orig_function, flag_added_args, *args, **kwargs):
-    # type: (Callable, int, Any, Any) -> bytearray
-    if orig_function != builtin_bytearray:
-        if flag_added_args > 0:
-            args = args[flag_added_args:]
-        return orig_function(*args, **kwargs)
+    # type: (Optional[Callable], int, Any, Any) -> bytearray
+    if orig_function:
+        if orig_function != builtin_bytearray:
+            if flag_added_args > 0:
+                args = args[flag_added_args:]
+            return orig_function(*args, **kwargs)
+        result = builtin_bytearray(*args, **kwargs)
+    else:
+        result = args[0].bytearray(*args[1:], **kwargs)
 
-    result = builtin_bytearray(*args, **kwargs)
     if isinstance(args[0], TEXT_TYPES) and is_pyobject_tainted(args[0]):
         try:
             taint_pyobject_with_ranges(result, tuple(get_ranges(args[0])))
@@ -107,7 +116,9 @@ def bytearray_aspect(orig_function, flag_added_args, *args, **kwargs):
 
 
 def join_aspect(orig_function, flag_added_args, *args, **kwargs):
-    # type: (Callable, int, Any, Any) -> Any
+    # type: (Optional[Callable], int, Any, Any) -> Any
+    if not orig_function:
+        orig_function = args[0].join
     if not isinstance(orig_function, BuiltinFunctionType):
         if flag_added_args > 0:
             args = args[flag_added_args:]
@@ -125,16 +136,22 @@ def join_aspect(orig_function, flag_added_args, *args, **kwargs):
 
 
 def bytearray_extend_aspect(orig_function, flag_added_args, *args, **kwargs):
-    # type: (Callable, int, Any, Any) -> Any
-    if not isinstance(orig_function, BuiltinFunctionType):
+    # type: (Optional[Callable], int, Any, Any) -> Any
+    if orig_function and not isinstance(orig_function, BuiltinFunctionType):
         if flag_added_args > 0:
             args = args[flag_added_args:]
         return orig_function(*args, **kwargs)
 
+    if len(args) < 2:
+        # If we're not receiving at least 2 arguments, means the call was
+        # ``x.extend()`` and not ``x.extend(y)``
+        # so either not the extend we're looking for, or no changes in taint ranges.
+        return args[0].extend(*args[1:], **kwargs)
+
     op1 = args[0]
     op2 = args[1]
     if not isinstance(op1, bytearray) or not isinstance(op2, (bytearray, bytes)):
-        return op1.extend(op2)
+        return op1.extend(*args[1:], **kwargs)
     try:
         return _extend_aspect(op1, op2)
     except Exception as e:
@@ -184,7 +201,9 @@ def build_string_aspect(*args):  # type: (List[Any]) -> str
 
 
 def ljust_aspect(orig_function, flag_added_args, *args, **kwargs):
-    # type: (Callable, int, Any, Any) -> Union[str, bytes, bytearray]
+    # type: (Optional[Callable], int, Any, Any) -> Union[str, bytes, bytearray]
+    if not orig_function:
+        orig_function = args[0].ljust
     if not isinstance(orig_function, BuiltinFunctionType):
         if flag_added_args > 0:
             args = args[flag_added_args:]
@@ -217,8 +236,8 @@ def ljust_aspect(orig_function, flag_added_args, *args, **kwargs):
 
 
 def zfill_aspect(orig_function, flag_added_args, *args, **kwargs):
-    # type: (Callable, int, Any, Any) -> Any
-    if not isinstance(orig_function, BuiltinFunctionType):
+    # type: (Optional[Callable], int, Any, Any) -> Any
+    if orig_function and not isinstance(orig_function, BuiltinFunctionType):
         if flag_added_args > 0:
             args = args[flag_added_args:]
         return orig_function(*args, **kwargs)
@@ -260,11 +279,14 @@ def zfill_aspect(orig_function, flag_added_args, *args, **kwargs):
 
 
 def format_aspect(
-    orig_function,  # type: Callable
+    orig_function,  # type: Optional[Callable]
     flag_added_args,  # type: int
     *args,  # type: Any
     **kwargs,  # type: Dict[str, Any]
 ):  # type: (...) -> str
+    if not orig_function:
+        orig_function = args[0].format
+
     if not isinstance(orig_function, BuiltinFunctionType):
         if flag_added_args > 0:
             args = args[flag_added_args:]
@@ -310,8 +332,10 @@ def format_aspect(
     return result
 
 
-def format_map_aspect(orig_function, flag_added_args, *args, **kwargs):  # type: (Callable, int, Any, Any) -> str
-    if not isinstance(orig_function, BuiltinFunctionType):
+def format_map_aspect(
+    orig_function, flag_added_args, *args, **kwargs
+):  # type: (Optional[Callable], int, Any, Any) -> str
+    if orig_function and not isinstance(orig_function, BuiltinFunctionType):
         if flag_added_args > 0:
             args = args[flag_added_args:]
         return orig_function(*args, **kwargs)
@@ -455,12 +479,14 @@ def incremental_translation(self, incr_coder, funcode, empty):
 
 
 def decode_aspect(orig_function, flag_added_args, *args, **kwargs):
-    if flag_added_args > 0:
-        self = args[0]
-        args = args[flag_added_args:]
-    else:
+    if orig_function and not flag_added_args:
+        # This patch is unexpected, so we fallback
+        # to executing the original function
         return orig_function(*args, **kwargs)
 
+    self = args[0]
+    args = args[(flag_added_args or 1) :]
+    # Assume we call decode method of the first argument
     result = self.decode(*args, **kwargs)
 
     if not is_pyobject_tainted(self) or not isinstance(self, bytes):
@@ -475,12 +501,14 @@ def decode_aspect(orig_function, flag_added_args, *args, **kwargs):
 
 
 def encode_aspect(orig_function, flag_added_args, *args, **kwargs):
-    if flag_added_args > 0:
-        self = args[0]
-        args = args[flag_added_args:]
-    else:
+    if orig_function and not flag_added_args:
+        # This patch is unexpected, so we fallback
+        # to executing the original function
         return orig_function(*args, **kwargs)
 
+    self = args[0]
+    args = args[(flag_added_args or 1) :]
+    # Assume we call encode method of the first argument
     result = self.encode(*args, **kwargs)
 
     if not is_pyobject_tainted(self) or not isinstance(self, str):
@@ -494,8 +522,10 @@ def encode_aspect(orig_function, flag_added_args, *args, **kwargs):
     return result
 
 
-def upper_aspect(orig_function, flag_added_args, *args, **kwargs):  # type: (Callable, int, Any, Any) -> TEXT_TYPE
-    if not isinstance(orig_function, BuiltinFunctionType):
+def upper_aspect(
+    orig_function, flag_added_args, *args, **kwargs
+):  # type: (Optional[Callable], int, Any, Any) -> TEXT_TYPE
+    if orig_function and not isinstance(orig_function, BuiltinFunctionType):
         if flag_added_args > 0:
             args = args[flag_added_args:]
         return orig_function(*args, **kwargs)
@@ -512,17 +542,17 @@ def upper_aspect(orig_function, flag_added_args, *args, **kwargs):  # type: (Cal
         return candidate_text.upper(*args, **kwargs)
 
 
-def lower_aspect(orig_function, flag_added_args, *args, **kwargs):  # type: (Callable, int, Any, Any) -> TEXT_TYPE
-    if not isinstance(orig_function, BuiltinFunctionType):
+def lower_aspect(
+    orig_function, flag_added_args, *args, **kwargs
+):  # type: (Optional[Callable], int, Any, Any) -> TEXT_TYPE
+    if orig_function and not isinstance(orig_function, BuiltinFunctionType):
         if flag_added_args > 0:
             args = args[flag_added_args:]
         return orig_function(*args, **kwargs)
 
     candidate_text = args[0]
     args = args[flag_added_args:]
     if not isinstance(candidate_text, TEXT_TYPES):
-        if flag_added_args > 0:
-            args = args[flag_added_args:]
         return candidate_text.lower(*args, **kwargs)
 
     try:
@@ -532,8 +562,10 @@ def lower_aspect(orig_function, flag_added_args, *args, **kwargs):  # type: (Cal
         return candidate_text.lower(*args, **kwargs)
 
 
-def swapcase_aspect(orig_function, flag_added_args, *args, **kwargs):  # type: (Callable, int, Any, Any) -> TEXT_TYPE
-    if not isinstance(orig_function, BuiltinFunctionType):
+def swapcase_aspect(
+    orig_function, flag_added_args, *args, **kwargs
+):  # type: (Optional[Callable], int, Any, Any) -> TEXT_TYPE
+    if orig_function and not isinstance(orig_function, BuiltinFunctionType):
         if flag_added_args > 0:
             args = args[flag_added_args:]
         return orig_function(*args, **kwargs)
@@ -549,8 +581,10 @@ def swapcase_aspect(orig_function, flag_added_args, *args, **kwargs):  # type: (
         return candidate_text.swapcase(*args, **kwargs)
 
 
-def title_aspect(orig_function, flag_added_args, *args, **kwargs):  # type: (Callable, int, Any, Any) -> TEXT_TYPE
-    if not isinstance(orig_function, BuiltinFunctionType):
+def title_aspect(
+    orig_function, flag_added_args, *args, **kwargs
+):  # type: (Optional[Callable], int, Any, Any) -> TEXT_TYPE
+    if orig_function and not isinstance(orig_function, BuiltinFunctionType):
         if flag_added_args > 0:
             args = args[flag_added_args:]
         return orig_function(*args, **kwargs)
@@ -566,8 +600,10 @@ def title_aspect(orig_function, flag_added_args, *args, **kwargs):  # type: (Cal
         return candidate_text.title(*args, **kwargs)
 
 
-def capitalize_aspect(orig_function, flag_added_args, *args, **kwargs):  # type: (Callable, int, Any, Any) -> TEXT_TYPE
-    if not isinstance(orig_function, BuiltinFunctionType):
+def capitalize_aspect(
+    orig_function, flag_added_args, *args, **kwargs
+):  # type: (Optional[Callable], int, Any, Any) -> TEXT_TYPE
+    if orig_function and not isinstance(orig_function, BuiltinFunctionType):
         if flag_added_args > 0:
             args = args[flag_added_args:]
         return orig_function(*args, **kwargs)
@@ -584,16 +620,21 @@ def capitalize_aspect(orig_function, flag_added_args, *args, **kwargs):  # type:
         return candidate_text.capitalize(*args, **kwargs)
 
 
-def casefold_aspect(orig_function, flag_added_args, *args, **kwargs):  # type: (Callable, int, Any, Any) -> TEXT_TYPE
-    if not isinstance(orig_function, BuiltinFunctionType):
-        if flag_added_args > 0:
-            args = args[flag_added_args:]
-        return orig_function(*args, **kwargs)
+def casefold_aspect(
+    orig_function, flag_added_args, *args, **kwargs
+):  # type: (Optional[Callable], int, Any, Any) -> TEXT_TYPE
+    if orig_function:
+        if not isinstance(orig_function, BuiltinFunctionType):
+            if flag_added_args > 0:
+                args = args[flag_added_args:]
+            return orig_function(*args, **kwargs)
+    else:
+        orig_function = getattr(args[0], "casefold", None)
 
-    if orig_function.__qualname__ not in ("str.casefold", "bytes.casefold", "bytearray.casefold"):
+    if orig_function and orig_function.__qualname__ not in ("str.casefold", "bytes.casefold", "bytearray.casefold"):
         if flag_added_args > 0:
             args = args[flag_added_args:]
-        return orig_function(args, **kwargs)
+        return orig_function(*args, **kwargs)
 
     candidate_text = args[0]
     args = args[flag_added_args:]
@@ -608,8 +649,10 @@ def casefold_aspect(orig_function, flag_added_args, *args, **kwargs):  # type: (
         return candidate_text.casefold(*args, **kwargs)  # type: ignore[union-attr]
 
 
-def translate_aspect(orig_function, flag_added_args, *args, **kwargs):  # type: (Callable, int, Any, Any) -> TEXT_TYPE
-    if not isinstance(orig_function, BuiltinFunctionType):
+def translate_aspect(
+    orig_function, flag_added_args, *args, **kwargs
+):  # type: (Optional[Callable], int, Any, Any) -> TEXT_TYPE
+    if orig_function and not isinstance(orig_function, BuiltinFunctionType):
         if flag_added_args > 0:
             args = args[flag_added_args:]
         return orig_function(*args, **kwargs)

releasenotes/notes/iast-fix-patching-side-effect-56d0b168e74839fc.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    Vulnerability Management for Code-level (IAST): This fix resolves an issue where, at AST patching to replace code with IAST aspects, passing the original function/method as an extra parameter for accurate patching unintentionally triggers side effects in methods obtained from an expression (like ``decode`` in ``file.read(n).decode()``), resulting in unexpected multiple calls to the expression (``file.read(n)`` in the example).

tests/appsec/iast/_ast/fixtures/misleading_methods.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+
+import tests.appsec.iast.fixtures.aspects.callees as callees_module
+
+
+class FakeStr(str):
+    def __init__(self, *args, **kwargs):
+        for i in [x for x in dir(callees_module) if not x.startswith(("_", "@"))]:
+            setattr(self, i, getattr(callees_module, i))
+
+    def call_ljust(self, *args, **kwargs):
+        return self.ljust(*args, **kwargs)
+
+    def call_join(self, *args, **kwargs):
+        return self.join(*args, **kwargs)
+
+    def call_zfill(self, *args, **kwargs):
+        return self.zfill(*args, **kwargs)
+
+    def call_format(self, *args, **kwargs):
+        return self.format(*args, **kwargs)
+
+    def call_format_map(self, *args, **kwargs):
+        return self.format_map(*args, **kwargs)
+
+    def call_repr(self, *args, **kwargs):
+        return self.repr(*args, **kwargs)
+
+    def call_upper(self, *args, **kwargs):
+        return self.upper(*args, **kwargs)
+
+    def call_lower(self, *args, **kwargs):
+        return self.lower(*args, **kwargs)
+
+    def call_casefold(self, *args, **kwargs):
+        return self.casefold(*args, **kwargs)
+
+    def call_capitalize(self, *args, **kwargs):
+        return self.capitalize(*args, **kwargs)
+
+    def call_title(self, *args, **kwargs):
+        return self.title(*args, **kwargs)
+
+    def call_swapcase(self, *args, **kwargs):
+        return self.swapcase(*args, **kwargs)
+
+    def call_translate(self, *args, **kwargs):
+        return self.translate(*args, **kwargs)
+
+    def call_extend(self, *args, **kwargs):
+        return self.extend(*args, **kwargs)
+
+    def call_encode(self, *args, **kwargs):
+        return self.encode(*args, **kwargs)
+
+    def call_decode(self, *args, **kwargs):
+        return self.decode(*args, **kwargs)
+
+    def call_str(self, *args, **kwargs):
+        return self.str(*args, **kwargs)
+
+    def call_bytes(self, *args, **kwargs):
+        return self.bytes(*args, **kwargs)
+
+    def call_bytearray(self, *args, **kwargs):
+        return self.bytearray(*args, **kwargs)