feat(iast): implement the stacktrace leak vulnerability (#12007)

## Description

Implement the stacktrace leak vulnerability detection.

STATUS: Waiting for DataDog/system-tests#3874 to
be merged.


Signed-off-by: Juanjo Alvarez <[email protected]>
## 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)

---------

Signed-off-by: Juanjo Alvarez <[email protected]>
Co-authored-by: Alberto Vara <[email protected]>

ddtrace/appsec/_iast/_handlers.py

@@ -5,7 +5,9 @@
 from wrapt import wrap_function_wrapper as _w
 
 from ddtrace.appsec._iast import _is_iast_enabled
+from ddtrace.appsec._iast._iast_request_context import get_iast_stacktrace_reported
 from ddtrace.appsec._iast._iast_request_context import in_iast_context
+from ddtrace.appsec._iast._iast_request_context import set_iast_stacktrace_reported
 from ddtrace.appsec._iast._metrics import _set_metric_iast_instrumented_source
 from ddtrace.appsec._iast._patch import _iast_instrument_starlette_request
 from ddtrace.appsec._iast._patch import _iast_instrument_starlette_request_body
@@ -445,3 +447,69 @@ def _on_set_request_tags_iast(request, span, flask_config):
             OriginType.PARAMETER,
             override_pyobject_tainted=True,
         )
+
+
+def _on_django_finalize_response_pre(ctx, after_request_tags, request, response):
+    if not response or get_iast_stacktrace_reported() or not _is_iast_enabled() or not is_iast_request_enabled():
+        return
+
+    try:
+        from .taint_sinks.stacktrace_leak import asm_check_stacktrace_leak
+
+        content = response.content.decode("utf-8", errors="ignore")
+        asm_check_stacktrace_leak(content)
+    except Exception:
+        log.debug("Unexpected exception checking for stacktrace leak", exc_info=True)
+
+
+def _on_django_technical_500_response(request, response, exc_type, exc_value, tb):
+    if not _is_iast_enabled() or not is_iast_request_enabled() or not exc_value:
+        return
+
+    try:
+        from .taint_sinks.stacktrace_leak import asm_report_stacktrace_leak_from_django_debug_page
+
+        exc_name = exc_type.__name__
+        module = tb.tb_frame.f_globals.get("__name__", "")
+        asm_report_stacktrace_leak_from_django_debug_page(exc_name, module)
+    except Exception:
+        log.debug("Unexpected exception checking for stacktrace leak on 500 response view", exc_info=True)
+
+
+def _on_flask_finalize_request_post(response, _):
+    if not response or get_iast_stacktrace_reported() or not _is_iast_enabled() or not is_iast_request_enabled():
+        return
+
+    try:
+        from .taint_sinks.stacktrace_leak import asm_check_stacktrace_leak
+
+        content = response[0].decode("utf-8", errors="ignore")
+        asm_check_stacktrace_leak(content)
+    except Exception:
+        log.debug("Unexpected exception checking for stacktrace leak", exc_info=True)
+
+
+def _on_asgi_finalize_response(body, _):
+    if not _is_iast_enabled() or not is_iast_request_enabled() or not body:
+        return
+
+    try:
+        from .taint_sinks.stacktrace_leak import asm_check_stacktrace_leak
+
+        content = body.decode("utf-8", errors="ignore")
+        asm_check_stacktrace_leak(content)
+    except Exception:
+        log.debug("Unexpected exception checking for stacktrace leak", exc_info=True)
+
+
+def _on_werkzeug_render_debugger_html(html):
+    if not _is_iast_enabled() or not is_iast_request_enabled() or not html:
+        return
+
+    try:
+        from .taint_sinks.stacktrace_leak import asm_check_stacktrace_leak
+
+        asm_check_stacktrace_leak(html)
+        set_iast_stacktrace_reported(True)
+    except Exception:
+        log.debug("Unexpected exception checking for stacktrace leak", exc_info=True)

ddtrace/appsec/_iast/_iast_request_context.py

@@ -46,6 +46,7 @@ def __init__(self, span: Optional[Span] = None):
         self.iast_reporter: Optional[IastSpanReporter] = None
         self.iast_span_metrics: Dict[str, int] = {}
         self.iast_stack_trace_id: int = 0
+        self.iast_stack_trace_reported: bool = False
 
 
 def _get_iast_context() -> Optional[IASTEnvironment]:
@@ -88,6 +89,19 @@ def get_iast_reporter() -> Optional[IastSpanReporter]:
     return None
 
 
+def get_iast_stacktrace_reported() -> bool:
+    env = _get_iast_context()
+    if env:
+        return env.iast_stack_trace_reported
+    return False
+
+
+def set_iast_stacktrace_reported(reported: bool) -> None:
+    env = _get_iast_context()
+    if env:
+        env.iast_stack_trace_reported = reported
+
+
 def get_iast_stacktrace_id() -> int:
     env = _get_iast_context()
     if env:

ddtrace/appsec/_iast/_listener.py

@@ -1,11 +1,16 @@
+from ddtrace.appsec._iast._handlers import _on_asgi_finalize_response
+from ddtrace.appsec._iast._handlers import _on_django_finalize_response_pre
 from ddtrace.appsec._iast._handlers import _on_django_func_wrapped
 from ddtrace.appsec._iast._handlers import _on_django_patch
+from ddtrace.appsec._iast._handlers import _on_django_technical_500_response
+from ddtrace.appsec._iast._handlers import _on_flask_finalize_request_post
 from ddtrace.appsec._iast._handlers import _on_flask_patch
 from ddtrace.appsec._iast._handlers import _on_grpc_response
 from ddtrace.appsec._iast._handlers import _on_pre_tracedrequest_iast
 from ddtrace.appsec._iast._handlers import _on_request_init
 from ddtrace.appsec._iast._handlers import _on_set_http_meta_iast
 from ddtrace.appsec._iast._handlers import _on_set_request_tags_iast
+from ddtrace.appsec._iast._handlers import _on_werkzeug_render_debugger_html
 from ddtrace.appsec._iast._handlers import _on_wsgi_environ
 from ddtrace.appsec._iast._iast_request_context import _iast_end_request
 from ddtrace.internal import core
@@ -18,11 +23,16 @@ def iast_listen():
     core.on("set_http_meta_for_asm", _on_set_http_meta_iast)
     core.on("django.patch", _on_django_patch)
     core.on("django.wsgi_environ", _on_wsgi_environ, "wrapped_result")
+    core.on("django.finalize_response.pre", _on_django_finalize_response_pre)
     core.on("django.func.wrapped", _on_django_func_wrapped)
+    core.on("django.technical_500_response", _on_django_technical_500_response)
     core.on("flask.patch", _on_flask_patch)
     core.on("flask.request_init", _on_request_init)
     core.on("flask._patched_request", _on_pre_tracedrequest_iast)
     core.on("flask.set_request_tags", _on_set_request_tags_iast)
+    core.on("flask.finalize_request.post", _on_flask_finalize_request_post)
+    core.on("asgi.finalize_response", _on_asgi_finalize_response)
+    core.on("werkzeug.render_debugger_html", _on_werkzeug_render_debugger_html)
 
     core.on("context.ended.wsgi.__call__", _iast_end_request)
     core.on("context.ended.asgi.__call__", _iast_end_request)

ddtrace/appsec/_iast/constants.py

@@ -1,3 +1,4 @@
+import re
 from typing import Any
 from typing import Dict
 
@@ -14,6 +15,7 @@
 VULN_HEADER_INJECTION = "HEADER_INJECTION"
 VULN_CODE_INJECTION = "CODE_INJECTION"
 VULN_SSRF = "SSRF"
+VULN_STACKTRACE_LEAK = "STACKTRACE_LEAK"
 
 VULNERABILITY_TOKEN_TYPE = Dict[int, Dict[str, Any]]
 
@@ -27,6 +29,12 @@
 RC2_DEF = "rc2"
 RC4_DEF = "rc4"
 IDEA_DEF = "idea"
+STACKTRACE_RE_DETECT = re.compile(r"Traceback \(most recent call last\):")
+HTML_TAGS_REMOVE = re.compile(r"<!--[\s\S]*?-->|<[^>]*>|&#\w+;")
+STACKTRACE_FILE_LINE = re.compile(r"File (.*?), line (\d+), in (.+)")
+STACKTRACE_EXCEPTION_REGEX = re.compile(
+    r"^(?P<exc>[A-Za-z_]\w*(?:Error|Exception|Interrupt|Fault|Warning))" r"(?:\s*:\s*(?P<msg>.*))?$"
+)
 
 DEFAULT_WEAK_HASH_ALGORITHMS = {MD5_DEF, SHA1_DEF}
 

ddtrace/appsec/_iast/taint_sinks/stacktrace_leak.py

@@ -0,0 +1,102 @@
+import os
+import re
+
+from ..._constants import IAST_SPAN_TAGS
+from .. import oce
+from .._iast_request_context import set_iast_stacktrace_reported
+from .._metrics import _set_metric_iast_executed_sink
+from .._metrics import increment_iast_span_metric
+from .._taint_tracking._errors import iast_taint_log_error
+from ..constants import HTML_TAGS_REMOVE
+from ..constants import STACKTRACE_EXCEPTION_REGEX
+from ..constants import STACKTRACE_FILE_LINE
+from ..constants import VULN_STACKTRACE_LEAK
+from ..taint_sinks._base import VulnerabilityBase
+
+
+@oce.register
+class StacktraceLeak(VulnerabilityBase):
+    vulnerability_type = VULN_STACKTRACE_LEAK
+    skip_location = True
+
+
+def asm_report_stacktrace_leak_from_django_debug_page(exc_name, module):
+    increment_iast_span_metric(IAST_SPAN_TAGS.TELEMETRY_EXECUTED_SINK, StacktraceLeak.vulnerability_type)
+    _set_metric_iast_executed_sink(StacktraceLeak.vulnerability_type)
+    evidence = "Module: %s\nException: %s" % (module, exc_name)
+    StacktraceLeak.report(evidence_value=evidence)
+    set_iast_stacktrace_reported(True)
+
+
+def asm_check_stacktrace_leak(content: str) -> None:
+    if not content:
+        return
+
+    try:
+        # Quick check to avoid the slower operations if on stacktrace
+        if "Traceback (most recent call last):" not in content:
+            return
+
+        text = HTML_TAGS_REMOVE.sub("", content)
+        lines = [line.strip() for line in text.splitlines() if line.strip()]
+
+        file_lines = []
+        exception_line = ""
+
+        for i, line in enumerate(lines):
+            if line.startswith("Traceback (most recent call last):"):
+                # from here until we find an exception line
+                continue
+
+            # See if this line is a "File ..." line
+            m_file = STACKTRACE_FILE_LINE.match(line)
+            if m_file:
+                file_lines.append(m_file.groups())
+                continue
+
+            # See if this line might be the exception line
+            m_exc = STACKTRACE_EXCEPTION_REGEX.match(line)
+            if m_exc:
+                # We consider it as the "final" exception line. Keep it.
+                exception_line = m_exc.group("exc")
+                # We won't break immediately because sometimes Django
+                # HTML stack traces can have repeated exception lines, etc.
+                # But typically the last match is the real final exception
+                # We'll keep updating exception_line if we see multiple
+                continue
+
+        if not file_lines and not exception_line:
+            return
+
+        module_path = None
+        if file_lines:
+            # file_lines looks like [ ("/path/to/file.py", "line_no", "funcname"), ... ]
+            last_file_entry = file_lines[-1]
+            module_path = last_file_entry[0]  # the path in quotes
+
+        # Attempt to convert a path like "/myproj/foo/bar.py" into "foo.bar"
+        # or "myproj.foo.bar" depending on your directory structure.
+        # This is a *best effort* approach (it can be environment-specific).
+        module_name = ""
+        if module_path:
+            mod_no_ext = re.sub(r"\.py$", "", module_path)
+            parts: list[str] = []
+            while True:
+                head, tail = os.path.split(mod_no_ext)
+                if tail:
+                    parts.insert(0, tail)
+                    mod_no_ext = head
+                else:
+                    # might still have a leftover 'head' if it’s not just root
+                    break
+
+            module_name = ".".join(parts)
+            if not module_name:
+                module_name = module_path  # fallback: just the path
+
+        increment_iast_span_metric(IAST_SPAN_TAGS.TELEMETRY_EXECUTED_SINK, StacktraceLeak.vulnerability_type)
+        _set_metric_iast_executed_sink(StacktraceLeak.vulnerability_type)
+        evidence = "Module: %s\nException: %s" % (module_name.strip(), exception_line.strip())
+        StacktraceLeak.report(evidence_value=evidence)
+    except Exception as e:
+        iast_taint_log_error("[IAST] error in check stacktrace leak. {}".format(e))

ddtrace/contrib/internal/django/patch.py

@@ -696,6 +696,23 @@ def traced_as_view(django, pin, func, instance, args, kwargs):
     return wrapt.FunctionWrapper(view, traced_func(django, "django.view", resource=func_name(instance)))
 
 
+@trace_utils.with_traced_module
+def traced_technical_500_response(django, pin, func, instance, args, kwargs):
+    """
+    Wrapper for django's views.debug.technical_500_response
+    """
+    response = func(*args, **kwargs)
+    try:
+        request = get_argument_value(args, kwargs, 0, "request")
+        exc_type = get_argument_value(args, kwargs, 1, "exc_type")
+        exc_value = get_argument_value(args, kwargs, 2, "exc_value")
+        tb = get_argument_value(args, kwargs, 3, "tb")
+        core.dispatch("django.technical_500_response", (request, response, exc_type, exc_value, tb))
+    except Exception:
+        log.debug("Error while trying to trace Django technical 500 response", exc_info=True)
+    return response
+
+
 @trace_utils.with_traced_module
 def traced_get_asgi_application(django, pin, func, instance, args, kwargs):
     from ddtrace.contrib.asgi import TraceMiddleware
@@ -891,6 +908,9 @@ def _(m):
             trace_utils.wrap(m, "re_path", traced_urls_path(django))
 
     when_imported("django.views.generic.base")(lambda m: trace_utils.wrap(m, "View.as_view", traced_as_view(django)))
+    when_imported("django.views.debug")(
+        lambda m: trace_utils.wrap(m, "technical_500_response", traced_technical_500_response(django))
+    )
 
     @when_imported("channels.routing")
     def _(m):
@@ -935,6 +955,7 @@ def _unpatch(django):
     trace_utils.unwrap(django.conf.urls, "url")
     trace_utils.unwrap(django.contrib.auth.login, "login")
     trace_utils.unwrap(django.contrib.auth.authenticate, "authenticate")
+    trace_utils.unwrap(django.view.debug.technical_500_response, "technical_500_response")
     if django.VERSION >= (2, 0, 0):
         trace_utils.unwrap(django.urls, "path")
         trace_utils.unwrap(django.urls, "re_path")

ddtrace/contrib/internal/flask/patch.py

@@ -224,6 +224,10 @@ def patch():
     _w("flask.templating", "_render", patched_render)
     _w("flask", "render_template", _build_render_template_wrapper("render_template"))
     _w("flask", "render_template_string", _build_render_template_wrapper("render_template_string"))
+    try:
+        _w("werkzeug.debug.tbtools", "DebugTraceback.render_debugger_html", patched_render_debugger_html)
+    except AttributeError:
+        log.debug("Failed to patch DebugTraceback.render_debugger_html, not supported by this werkzeug version")
 
     bp_hooks = [
         "after_app_request",
@@ -380,12 +384,8 @@ def patched_finalize_request(wrapped, instance, args, kwargs):
     Wrapper for flask.app.Flask.finalize_request
     """
     rv = wrapped(*args, **kwargs)
-    response = None
-    headers = None
     if getattr(rv, "is_sequence", False):
-        response = rv.response
-        headers = rv.headers
-    core.dispatch("flask.finalize_request.post", (response, headers))
+        core.dispatch("flask.finalize_request.post", (rv.response, rv.headers))
     return rv
 
 
@@ -419,6 +419,12 @@ def _wrap(rule, endpoint=None, view_func=None, **kwargs):
     return _wrap(*args, **kwargs)
 
 
+def patched_render_debugger_html(wrapped, instance, args, kwargs):
+    res = wrapped(*args, **kwargs)
+    core.dispatch("werkzeug.render_debugger_html", (res,))
+    return res
+
+
 def patched_add_url_rule(wrapped, instance, args, kwargs):
     """Wrapper for flask.app.Flask.add_url_rule to wrap all views attached to this app"""
 

hatch.toml

@@ -57,7 +57,7 @@ checks = [
     "suitespec-check",
 ]
 spelling = [
-    "codespell -I docs/spelling_wordlist.txt --skip='ddwaf.h,*cassettes*,tests/tracer/fixtures/urls.txt' {args:ddtrace/ tests/ releasenotes/ docs/}",
+    "codespell -I docs/spelling_wordlist.txt --skip='ddwaf.h,*cassettes*,tests/tracer/fixtures/urls.txt,tests/appsec/iast/fixtures/*' {args:ddtrace/ tests/ releasenotes/ docs/}",
 ]
 typing = [
     "mypy {args}",

releasenotes/notes/stacktrace-leak-d6840ab48b29af99.yaml

@@ -0,0 +1,5 @@
+---
+features:
+  - |
+    Code Security: Implement the detection of the Stacktrace-Leak vulnerability for 
+    Django, Flask and FastAPI.