fix(iast): improve traceback message from telemetry logs [Backport 1.20] (#7522)

Backport #7510 to 1.20.

## 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/_metrics.py

@@ -1,4 +1,6 @@
 import os
+import sys
+import traceback
 
 from ddtrace.appsec._constants import IAST
 from ddtrace.appsec._deduplications import deduplication
@@ -50,9 +52,20 @@ def wrapper(f):
 
 @metric_verbosity(TELEMETRY_MANDATORY_VERBOSITY)
 @deduplication
-def _set_iast_error_metric(msg, stack_trace):
-    # type: (str, str) -> None
+def _set_iast_error_metric(msg):
+    # type: (str) -> None
+    # Due to format_exc and format_exception returns the error and the last frame
     try:
+        exception_type, exception_instance, _traceback_list = sys.exc_info()
+        res = []
+        # first 3 frames are this function, the exception in aspects and the error line
+        res.extend(traceback.format_stack(limit=10)[:-3])
+
+        # get the frame with the error and the error message
+        result = traceback.format_exception(exception_type, exception_instance, _traceback_list)
+        res.extend(result[1:])
+
+        stack_trace = "".join(res)
         tags = {
             "lib_language": "python",
         }

ddtrace/appsec/iast/_taint_tracking/aspects.py

@@ -2,7 +2,6 @@
 from builtins import bytes as builtin_bytes
 from builtins import str as builtin_str
 import codecs
-import traceback
 from typing import TYPE_CHECKING
 
 from ddtrace.internal.compat import iteritems
@@ -63,7 +62,7 @@ def str_aspect(*args, **kwargs):
             if new_ranges:
                 taint_pyobject_with_ranges(result, tuple(new_ranges))
         except Exception as e:
-            _set_iast_error_metric("IAST propagation error. str_aspect. {}".format(e), traceback.format_exc())
+            _set_iast_error_metric("IAST propagation error. str_aspect. {}".format(e))
     return result
 
 
@@ -74,7 +73,7 @@ def bytes_aspect(*args, **kwargs):
         try:
             taint_pyobject_with_ranges(result, tuple(get_ranges(args[0])))
         except Exception as e:
-            _set_iast_error_metric("IAST propagation error. bytes_aspect. {}".format(e), traceback.format_exc())
+            _set_iast_error_metric("IAST propagation error. bytes_aspect. {}".format(e))
     return result
 
 
@@ -85,7 +84,7 @@ def bytearray_aspect(*args, **kwargs):
         try:
             taint_pyobject_with_ranges(result, tuple(get_ranges(args[0])))
         except Exception as e:
-            _set_iast_error_metric("IAST propagation error. bytearray_aspect. {}".format(e), traceback.format_exc())
+            _set_iast_error_metric("IAST propagation error. bytearray_aspect. {}".format(e))
     return result
 
 
@@ -96,7 +95,7 @@ def join_aspect(joiner, *args, **kwargs):
     try:
         return _join_aspect(joiner, *args, **kwargs)
     except Exception as e:
-        _set_iast_error_metric("IAST propagation error. join_aspect. {}".format(e), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. join_aspect. {}".format(e))
         return joiner.join(*args, **kwargs)
 
 
@@ -107,7 +106,7 @@ def bytearray_extend_aspect(op1, op2):
     try:
         return _extend_aspect(op1, op2)
     except Exception as e:
-        _set_iast_error_metric("IAST propagation error. extend_aspect. {}".format(e), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. extend_aspect. {}".format(e))
         return op1.extend(op2)
 
 
@@ -144,7 +143,7 @@ def modulo_aspect(candidate_text, candidate_tuple):
             ranges_orig=ranges_orig,
         )
     except Exception as e:
-        _set_iast_error_metric("IAST propagation error. modulo_aspect. {}".format(e), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. modulo_aspect. {}".format(e))
         return candidate_text % candidate_tuple
 
 
@@ -171,7 +170,7 @@ def ljust_aspect(candidate_text, *args, **kwargs):
         taint_pyobject_with_ranges(res, ranges_new)
         return res
     except Exception as e:
-        _set_iast_error_metric("IAST propagation error. ljust_aspect. {}".format(e), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. ljust_aspect. {}".format(e))
         return candidate_text.ljust(*args, **kwargs)
 
 
@@ -204,7 +203,7 @@ def zfill_aspect(candidate_text, *args, **kwargs):
         taint_pyobject_with_ranges(res, tuple(ranges_new))
         return res
     except Exception as e:
-        _set_iast_error_metric("IAST propagation error. format_aspect. {}".format(e), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. format_aspect. {}".format(e))
         return candidate_text.zfill(*args, **kwargs)
 
 
@@ -244,7 +243,7 @@ def format_aspect(
             )
         return result
     except Exception as e:
-        _set_iast_error_metric("IAST propagation error. format_aspect. {}".format(e), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. format_aspect. {}".format(e))
         return candidate_text.format(*args, **kwargs)
 
 
@@ -275,7 +274,7 @@ def format_map_aspect(candidate_text, *args, **kwargs):  # type: (str, Any, Any)
             ranges_orig=ranges_orig,
         )
     except Exception as e:
-        _set_iast_error_metric("IAST propagation error. format_map_aspect. {}".format(e), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. format_map_aspect. {}".format(e))
         return candidate_text.format_map(*args, **kwargs)
 
 
@@ -293,7 +292,7 @@ def repr_aspect(*args, **kwargs):
             if new_ranges:
                 taint_pyobject_with_ranges(result, tuple(new_ranges))
         except Exception as e:
-            _set_iast_error_metric("IAST propagation error. repr_aspect. {}".format(e), traceback.format_exc())
+            _set_iast_error_metric("IAST propagation error. repr_aspect. {}".format(e))
     return result
 
 
@@ -335,7 +334,7 @@ def format_value_aspect(
         else:
             return str_aspect(new_text)
     except Exception as e:
-        _set_iast_error_metric("IAST propagation error. format_value_aspect. {}".format(e), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. format_value_aspect. {}".format(e))
         return new_text
 
 
@@ -387,7 +386,7 @@ def decode_aspect(self, *args, **kwargs):
         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), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. decode_aspect. {}".format(e))
         return self.decode(*args, **kwargs)
 
 
@@ -399,7 +398,7 @@ def encode_aspect(self, *args, **kwargs):
         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), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. encode_aspect. {}".format(e))
         return self.encode(*args, **kwargs)
 
 
@@ -410,7 +409,7 @@ def upper_aspect(candidate_text, *args, **kwargs):  # type: (Any, Any, Any) -> T
     try:
         return common_replace("upper", candidate_text, *args, **kwargs)
     except Exception as e:
-        _set_iast_error_metric("IAST propagation error. upper_aspect. {}".format(e), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. upper_aspect. {}".format(e))
         return candidate_text.upper(*args, **kwargs)
 
 
@@ -421,7 +420,7 @@ def lower_aspect(candidate_text, *args, **kwargs):  # type: (Any, Any, Any) -> T
     try:
         return common_replace("lower", candidate_text, *args, **kwargs)
     except Exception as e:
-        _set_iast_error_metric("IAST propagation error. lower_aspect. {}".format(e), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. lower_aspect. {}".format(e))
         return candidate_text.lower(*args, **kwargs)
 
 
@@ -431,7 +430,7 @@ def swapcase_aspect(candidate_text, *args, **kwargs):  # type: (Any, Any, Any) -
     try:
         return common_replace("swapcase", candidate_text, *args, **kwargs)
     except Exception as e:
-        _set_iast_error_metric("IAST propagation error. swapcase_aspect. {}".format(e), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. swapcase_aspect. {}".format(e))
         return candidate_text.swapcase(*args, **kwargs)
 
 
@@ -441,7 +440,7 @@ def title_aspect(candidate_text, *args, **kwargs):  # type: (Any, Any, Any) -> T
     try:
         return common_replace("title", candidate_text, *args, **kwargs)
     except Exception as e:
-        _set_iast_error_metric("IAST propagation error. title_aspect. {}".format(e), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. title_aspect. {}".format(e))
         return candidate_text.title(*args, **kwargs)
 
 
@@ -452,7 +451,7 @@ def capitalize_aspect(candidate_text, *args, **kwargs):  # type: (Any, Any, Any)
     try:
         return common_replace("capitalize", candidate_text, *args, **kwargs)
     except Exception as e:
-        _set_iast_error_metric("IAST propagation error. capitalize_aspect. {}".format(e), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. capitalize_aspect. {}".format(e))
         return candidate_text.capitalize(*args, **kwargs)
 
 
@@ -462,7 +461,7 @@ def casefold_aspect(candidate_text, *args, **kwargs):  # type: (Any, Any, Any) -
     try:
         return common_replace("casefold", candidate_text, *args, **kwargs)
     except Exception as e:
-        _set_iast_error_metric("IAST propagation error. casefold_aspect. {}".format(e), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. casefold_aspect. {}".format(e))
         return candidate_text.casefold(*args, **kwargs)  # type: ignore[union-attr]
 
 
@@ -472,7 +471,7 @@ def translate_aspect(candidate_text, *args, **kwargs):  # type: (Any, Any, Any)
     try:
         return common_replace("translate", candidate_text, *args, **kwargs)
     except Exception as e:
-        _set_iast_error_metric("IAST propagation error. translate_aspect. {}".format(e), traceback.format_exc())
+        _set_iast_error_metric("IAST propagation error. translate_aspect. {}".format(e))
         return candidate_text.translate(*args, **kwargs)
 
 

tests/appsec/iast/test_telemetry.py

@@ -5,6 +5,7 @@
     from ddtrace.appsec.iast._metrics import TELEMETRY_DEBUG_VERBOSITY
     from ddtrace.appsec.iast._metrics import TELEMETRY_INFORMATION_VERBOSITY
     from ddtrace.appsec.iast._metrics import TELEMETRY_MANDATORY_VERBOSITY
+    from ddtrace.appsec.iast._metrics import _set_iast_error_metric
     from ddtrace.appsec.iast._metrics import metric_verbosity
     from ddtrace.appsec.iast._patch_modules import patch_iast
     from ddtrace.appsec.iast._taint_tracking import OriginType
@@ -105,3 +106,12 @@ def test_metric_request_tainted(mock_telemetry_lifecycle_writer):
     generate_metrics = metrics_result[TELEMETRY_TYPE_GENERATE_METRICS][TELEMETRY_NAMESPACE_TAG_IAST]
     assert len(generate_metrics) == 2, "Expected 1 generate_metrics"
     assert [metric.name for metric in generate_metrics.values()] == ["executed.source", "request.tainted"]
+
+
+def test_log_metric(mock_telemetry_lifecycle_writer):
+    _set_iast_error_metric("test_format_key_error_and_no_log_metric raises")
+
+    list_metrics_logs = list(mock_telemetry_lifecycle_writer._logs)
+    assert len(list_metrics_logs) == 1
+    assert list_metrics_logs[0]["message"] == "test_format_key_error_and_no_log_metric raises"
+    assert str(list_metrics_logs[0]["stack_trace"]).startswith('  File "/')