fix(iast): remove logs metrics noise [backport 1.20] (#7550)

Backport 5dae3e3 from #7546 to 1.20.

IAST Aspects report a log metric each time an exception is raised. We
noticed an application that wants to raise and control exceptions, such
as in this dummy example:

```python
my_data = {"element1": 1}
is_element2_in_my_data = False
try:
    my_data["element1"]
    is_element2_in_my_data = True
except KeyError:
    pass
```

Those operations should not log a metric.

## Checklist

- [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

- [x] Title is accurate.
- [x] No unnecessary changes are introduced.
- [x] Description motivates each change.
- [x] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes unless absolutely necessary.
- [x] Testing strategy adequately addresses listed risk(s).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] Release note makes sense to a user of the library.
- [x] Reviewer has explicitly acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment.
- [x] 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)
- [x] 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`.
- [x] This PR doesn't touch any of that.

Author: avara1986

ddtrace/appsec/iast/_taint_tracking/aspects.py

@@ -51,6 +51,7 @@ def add_aspect(op1, op2):
 def str_aspect(*args, **kwargs):
     # type: (Any, Any) -> str
     result = builtin_str(*args, **kwargs)
+
     if isinstance(args[0], TEXT_TYPES) and is_pyobject_tainted(args[0]):
         try:
             if isinstance(args[0], (bytes, bytearray)):
@@ -153,39 +154,46 @@ def build_string_aspect(*args):  # type: (List[Any]) -> str
 
 def ljust_aspect(candidate_text, *args, **kwargs):
     # type: (Any, Any, Any) -> Union[str, bytes, bytearray]
+
+    result = candidate_text.ljust(*args, **kwargs)
+
     if not isinstance(candidate_text, TEXT_TYPES):
-        return candidate_text.ljust(*args, **kwargs)
+        return result
+
     try:
         ranges_new = get_ranges(candidate_text)
         fillchar = parse_params(1, "fillchar", " ", *args, **kwargs)
         fillchar_ranges = get_ranges(fillchar)
         if ranges_new is None or (not ranges_new and not fillchar_ranges):
-            return candidate_text.ljust(*args, **kwargs)
+            return result
 
         if fillchar_ranges:
             # Can only be one char, so we create one range to cover from the start to the end
             ranges_new = ranges_new + [shift_taint_range(fillchar_ranges[0], len(candidate_text))]
 
-        res = candidate_text.ljust(parse_params(0, "width", None, *args, **kwargs), fillchar)
-        taint_pyobject_with_ranges(res, ranges_new)
-        return res
+        new_result = candidate_text.ljust(parse_params(0, "width", None, *args, **kwargs), fillchar)
+        taint_pyobject_with_ranges(new_result, ranges_new)
+        return new_result
     except Exception as e:
         _set_iast_error_metric("IAST propagation error. ljust_aspect. {}".format(e))
-        return candidate_text.ljust(*args, **kwargs)
+    return result
 
 
 def zfill_aspect(candidate_text, *args, **kwargs):
     # type: (Any, Any, Any) -> Any
+
+    result = candidate_text.zfill(*args, **kwargs)
+
     if not isinstance(candidate_text, TEXT_TYPES):
-        return candidate_text.zfill(*args, **kwargs)
+        return result
+
     try:
         ranges_orig = get_ranges(candidate_text)
         if not ranges_orig:
-            return candidate_text.zfill(*args, **kwargs)
+            return result
         prefix = candidate_text[0] in ("-", "+")
-        res = candidate_text.zfill(*args, **kwargs)
 
-        difflen = len(res) - len(candidate_text)
+        difflen = len(result) - len(candidate_text)
         ranges_new = []  # type: List[TaintRange]
         ranges_new_append = ranges_new.append
         ranges_new_extend = ranges_new.extend
@@ -200,25 +208,27 @@ def zfill_aspect(candidate_text, *args, **kwargs):
                         TaintRange(start=r.start + difflen + 1, length=r.length - 1, source=r.source),
                     ]
                 )
-        taint_pyobject_with_ranges(res, tuple(ranges_new))
-        return res
+        taint_pyobject_with_ranges(result, tuple(ranges_new))
     except Exception as e:
         _set_iast_error_metric("IAST propagation error. format_aspect. {}".format(e))
-        return candidate_text.zfill(*args, **kwargs)
+    return result
 
 
 def format_aspect(
     candidate_text,  # type: str
     *args,  # type: List[Any]
     **kwargs  # type: Dict[str, Any]
 ):  # type: (...) -> str
+    result = candidate_text.format(*args, **kwargs)
+
     if not isinstance(candidate_text, TEXT_TYPES):
-        return candidate_text.format(*args, **kwargs)
+        return result
+
     try:
         params = tuple(args) + tuple(kwargs.values())
         ranges_orig, candidate_text_ranges = are_all_text_all_ranges(candidate_text, params)
         if not ranges_orig:
-            return candidate_text.format(*args, **kwargs)
+            return result
 
         new_template = as_formatted_evidence(
             candidate_text, candidate_text_ranges, tag_mapping_function=TagMappingMode.Mapper
@@ -232,24 +242,25 @@ def format_aspect(
         new_args = list(map(fun, args))
 
         new_kwargs = {key: fun(value) for key, value in iteritems(kwargs)}
-        result = _convert_escaped_text_to_tainted_text(
+        new_result = _convert_escaped_text_to_tainted_text(
             new_template.format(*new_args, **new_kwargs),
             ranges_orig=ranges_orig,
         )
-        if result != candidate_text.format(*args):
+        if new_result != result:
             raise Exception(
                 "format_aspect result %s is different to candidate_text.format %s"
                 % (result, candidate_text.format(*args))
             )
-        return result
+        return new_result
     except Exception as e:
         _set_iast_error_metric("IAST propagation error. format_aspect. {}".format(e))
-        return candidate_text.format(*args, **kwargs)
+    return result
 
 
 def format_map_aspect(candidate_text, *args, **kwargs):  # type: (str, Any, Any) -> str
     if not isinstance(candidate_text, TEXT_TYPES):
         return candidate_text.format_map(*args, **kwargs)
+
     try:
         mapping = parse_params(0, "mapping", None, *args, **kwargs)
         mapping_tuple = tuple(mapping if not isinstance(mapping, dict) else mapping.values())
@@ -281,6 +292,7 @@ def format_map_aspect(candidate_text, *args, **kwargs):  # type: (str, Any, Any)
 def repr_aspect(*args, **kwargs):
     # type: (Any, Any) -> Any
     result = repr(*args, **kwargs)
+
     if isinstance(args[0], TEXT_TYPES) and is_pyobject_tainted(args[0]):
         try:
             if isinstance(args[0], (bytes, bytearray)):
@@ -379,27 +391,33 @@ def incremental_translation(self, incr_coder, funcode, empty):
 
 
 def decode_aspect(self, *args, **kwargs):
+    result = self.decode(*args, **kwargs)
+
     if not is_pyobject_tainted(self) or not isinstance(self, bytes):
-        return self.decode(*args, **kwargs)
+        return result
+
     try:
         codec = args[0] if args else "utf-8"
         inc_dec = codecs.getincrementaldecoder(codec)(**kwargs)
         return incremental_translation(self, inc_dec, inc_dec.decode, "")
     except Exception as e:
         _set_iast_error_metric("IAST propagation error. decode_aspect. {}".format(e))
-        return self.decode(*args, **kwargs)
+    return result
 
 
 def encode_aspect(self, *args, **kwargs):
+    result = self.encode(*args, **kwargs)
+
     if not is_pyobject_tainted(self) or not isinstance(self, str):
-        return self.encode(*args, **kwargs)
+        return result
+
     try:
         codec = args[0] if args else "utf-8"
         inc_enc = codecs.getincrementalencoder(codec)(**kwargs)
         return incremental_translation(self, inc_enc, inc_enc.encode, b"")
     except Exception as e:
         _set_iast_error_metric("IAST propagation error. encode_aspect. {}".format(e))
-        return self.encode(*args, **kwargs)
+    return result
 
 
 def upper_aspect(candidate_text, *args, **kwargs):  # type: (Any, Any, Any) -> TEXT_TYPE

tests/appsec/iast/aspects/test_encode_decode_aspect.py

@@ -3,14 +3,20 @@
 import pytest
 
 from ddtrace.appsec.iast._utils import _is_python_version_supported as python_supported_by_iast
+from tests.appsec.iast.aspects.conftest import _iast_patched_module
 
 
 try:
     from ddtrace.appsec.iast._taint_tracking import OriginType
+    from ddtrace.appsec.iast._taint_tracking import taint_pyobject
 except (ImportError, AttributeError):
     pytest.skip("IAST not supported for this Python version", allow_module_level=True)
 
 
+if python_supported_by_iast():
+    mod = _iast_patched_module("tests.appsec.iast.fixtures.aspects.str_methods")
+
+
 def catch_all(fun, args, kwargs):
     try:
         return True, fun(*args, **kwargs)
@@ -109,3 +115,33 @@ def test_encode_and_add_aspect(infix, args, kwargs, should_be_tainted, prefix, s
         assert list_ranges[0].start == len(prefix.encode(*args, **kwargs))
         len_infix = len(infix.encode(*args, **kwargs))
         assert list_ranges[0].length == len_infix
+
+
+@pytest.mark.skipif(not python_supported_by_iast(), reason="Python version not supported by IAST")
+def test_encode_error_and_no_log_metric(mock_telemetry_lifecycle_writer):
+    string_input = taint_pyobject(
+        pyobject="abcde",
+        source_name="test_add_aspect_tainting_left_hand",
+        source_value="abcde",
+        source_origin=OriginType.PARAMETER,
+    )
+    with pytest.raises(LookupError):
+        mod.do_encode(string_input, "encoding-not-exists")
+
+    list_metrics_logs = list(mock_telemetry_lifecycle_writer._logs)
+    assert len(list_metrics_logs) == 0
+
+
+@pytest.mark.skipif(not python_supported_by_iast(), reason="Python version not supported by IAST")
+def test_dencode_error_and_no_log_metric(mock_telemetry_lifecycle_writer):
+    string_input = taint_pyobject(
+        pyobject=b"abcde",
+        source_name="test_add_aspect_tainting_left_hand",
+        source_value="abcde",
+        source_origin=OriginType.PARAMETER,
+    )
+    with pytest.raises(LookupError):
+        mod.do_decode(string_input, "decoding-not-exists")
+
+    list_metrics_logs = list(mock_telemetry_lifecycle_writer._logs)
+    assert len(list_metrics_logs) == 0

tests/appsec/iast/aspects/test_format_aspect_fixtures.py

@@ -224,3 +224,10 @@ def test_format_when_tainted_template_range_special_template_then_tainted_result
         #     escaped_expected_result="a:+-<input2>aaaa<input2>-+:a parameter",
         # )
         pass
+
+    def test_format_key_error_and_no_log_metric(self, mock_telemetry_lifecycle_writer):
+        with pytest.raises(KeyError):
+            mod.do_format_key_error("test1")
+
+        list_metrics_logs = list(mock_telemetry_lifecycle_writer._logs)
+        assert len(list_metrics_logs) == 0

tests/appsec/iast/aspects/test_str_aspect.py

@@ -112,6 +112,14 @@ def test_aspect_ljust_str_tainted(self):
         ljusted = mod.do_ljust(string_input, 4)  # pylint: disable=no-member
         assert as_formatted_evidence(ljusted) == ":+-foo-+: "
 
+    def test_aspect_ljust_error_and_no_log_metric(self, mock_telemetry_lifecycle_writer):
+        string_input = create_taint_range_with_format(":+-foo-+:")
+        with pytest.raises(TypeError):
+            mod.do_ljust(string_input, "aaaaa")
+
+        list_metrics_logs = list(mock_telemetry_lifecycle_writer._logs)
+        assert len(list_metrics_logs) == 0
+
     def test_zfill(self):
         # Not tainted
         string_input = "-1234"
@@ -131,6 +139,14 @@ def test_zfill(self):
         res = mod.do_zfill(string_input, 7)  # pylint: disable=no-member
         assert as_formatted_evidence(res) == "00:+-012-+:34"
 
+    def test_aspect_zfill_error_and_no_log_metric(self, mock_telemetry_lifecycle_writer):
+        string_input = create_taint_range_with_format(":+-foo-+:")
+        with pytest.raises(TypeError):
+            mod.do_zfill(string_input, "aaaaa")
+
+        list_metrics_logs = list(mock_telemetry_lifecycle_writer._logs)
+        assert len(list_metrics_logs) == 0
+
     def test_format(self):
         # type: () -> None
         string_input = "foo"

tests/appsec/iast/fixtures/aspects/str_methods.py

@@ -845,6 +845,10 @@ def do_format_map(template, mapping):  # type: (str, Dict[str, Any]) -> str
     return template.format_map(mapping)
 
 
+def do_format_key_error(param1):  # type: (str, Dict[str, Any]) -> str
+    return "Test {param1}, {param2}".format(param1=param1)  # noqa
+
+
 def do_join(s, iterable):
     # type: (str, Iterable) -> str
     return s.join(iterable)

tests/appsec/test_psycopg2.py

@@ -5,11 +5,11 @@
 
 
 try:
-    from ddtrace.appsec._iast import oce
-    from ddtrace.appsec._iast._taint_tracking import OriginType
-    from ddtrace.appsec._iast._taint_tracking import create_context
-    from ddtrace.appsec._iast._taint_tracking import is_pyobject_tainted
-    from ddtrace.appsec._iast._taint_utils import LazyTaintList
+    from ddtrace.appsec.iast import oce
+    from ddtrace.appsec.iast._taint_tracking import OriginType
+    from ddtrace.appsec.iast._taint_tracking import create_context
+    from ddtrace.appsec.iast._taint_tracking import is_pyobject_tainted
+    from ddtrace.appsec.iast._taint_utils import LazyTaintList
 except (ImportError, AttributeError):
     pytest.skip("IAST not supported for this Python version", allow_module_level=True)