chore(tracer): core interface improvement (#7863)

# Motivation

Currently core interface can be use to register and call listener on
specific internal events with

- `core.on` to register an event
- `core.dispatch` to call listener on a specific event

`core.dispatch` returns a tuple of lists containing results or possible
exceptions of all listeners in the order they were registered. This is
difficult to use, as we have to properly check the lists length and
guess the registration order in case of several listeners. Using
`[0][0]` to access result is not meaningful nor without a risk of
errors.

# Proposition of this PR

- `core.on` to register an event with an additional optional name
- `core.dispatch` to call listener on a specific event will return a
EventResultDict that can be used like that:

```python
# some file
core.on("my_event", listener_function, "my_name")

# some other file
core.on("my_event", listener_function_2, "my_second_name")

# some other listener, but you don't care about the return value, so no need to set a name
core.on("my_event", listener_function_3)

# in another file
res = core.dispatch("my_event")
if res.my_name:
  result = res.my_name.value
  ...
# optional error handling
else:
  if res.my_name.response_type == core.ResultType.RESULT_UNDEFINED:
     log.debug("listener my_name was not registered on my_event")
  elif res..my_name.response_type == core.ResultType.RESULT_EXCEPTION:
     log.debug("listener my_name on my_event raised exception %s", res.my_name.exception)

# you can also just simply get the returned value of the listener. 
# In case of an error, this is set to None anyway, so it never breaks.
second_name = res.my_second_name.value
```

## 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: christophe-papazian

ddtrace/appsec/_asm_request_context.py

@@ -447,12 +447,12 @@ def _set_headers_and_response(response, headers, *_):
 
 def _call_waf_first(integration, *_):
     log.debug("%s WAF call for Suspicious Request Blocking on request", integration)
-    call_waf_callback()
+    return call_waf_callback()
 
 
 def _call_waf(integration, *_):
     log.debug("%s WAF call for Suspicious Request Blocking on response", integration)
-    call_waf_callback()
+    return call_waf_callback()
 
 
 def _on_block_decided(callback):
@@ -466,15 +466,15 @@ def _get_headers_if_appsec():
 
 def listen_context_handlers():
     core.on("flask.finalize_request.post", _set_headers_and_response)
-    core.on("flask.wrapped_view", _on_wrapped_view)
+    core.on("flask.wrapped_view", _on_wrapped_view, "callback_and_args")
     core.on("flask._patched_request", _on_pre_tracedrequest)
     core.on("wsgi.block_decided", _on_block_decided)
-    core.on("flask.start_response", _call_waf)
+    core.on("flask.start_response", _call_waf, "waf")
 
     core.on("django.start_response.post", _call_waf)
     core.on("django.finalize_response", _call_waf)
-    core.on("django.after_request_headers", _get_headers_if_appsec)
-    core.on("django.extract_body", _get_headers_if_appsec)
+    core.on("django.after_request_headers", _get_headers_if_appsec, "headers")
+    core.on("django.extract_body", _get_headers_if_appsec, "headers")
     core.on("django.after_request_headers.finalize", _set_headers_and_response)
     core.on("flask.set_request_tags", _on_set_request_tags)
 

ddtrace/appsec/_handlers.py

@@ -293,14 +293,14 @@ def _on_django_patch():
 
 
 def listen():
-    core.on("flask.request_call_modifier", _on_request_span_modifier)
+    core.on("flask.request_call_modifier", _on_request_span_modifier, "request_body")
     core.on("flask.request_init", _on_request_init)
     core.on("flask.blocked_request_callable", _on_flask_blocked_request)
 
 
 core.on("django.func.wrapped", _on_django_func_wrapped)
-core.on("django.wsgi_environ", _on_wsgi_environ)
+core.on("django.wsgi_environ", _on_wsgi_environ, "wrapped_result")
 core.on("django.patch", _on_django_patch)
 core.on("flask.patch", _on_flask_patch)
 
-core.on("asgi.request.parse.body", _on_asgi_request_parse_body)
+core.on("asgi.request.parse.body", _on_asgi_request_parse_body, "await_receive_and_body")

ddtrace/appsec/_metrics.py

@@ -71,7 +71,7 @@ def _set_waf_init_metric(info):
 
 def _set_waf_request_metrics(*args):
     try:
-        list_results, list_result_info, list_is_blocked = _asm_request_context.get_waf_results()
+        list_results, list_result_info, list_is_blocked = _asm_request_context.get_waf_results() or ([], [], [])
         if any((list_results, list_result_info, list_is_blocked)):
             is_blocked = any(list_is_blocked)
             is_triggered = any((result.data for result in list_results))

ddtrace/appsec/_trace_utils.py

@@ -339,4 +339,4 @@ def _on_django_auth(result_user, mode, kwargs, pin, info_retriever):
 
 
 core.on("django.login", _on_django_login)
-core.on("django.auth", _on_django_auth)
+core.on("django.auth", _on_django_auth, "user")

ddtrace/contrib/asgi/middleware.py

@@ -194,9 +194,9 @@ async def __call__(self, scope, receive, send):
             if not self.integration_config.trace_query_string:
                 query_string = None
             body = None
-            parse_body_result = core.dispatch_with_results("asgi.request.parse.body", (receive, headers))[0]
-            if len(parse_body_result) == 1:
-                receive, body = await parse_body_result[0]
+            result = core.dispatch_with_results("asgi.request.parse.body", (receive, headers)).await_receive_and_body
+            if result:
+                receive, body = await result.value
 
             client = scope.get("client")
             if isinstance(client, list) and len(client):
@@ -252,7 +252,11 @@ async def wrapped_send(message):
                         span.finish()
 
             async def wrapped_blocked_send(message):
-                status, headers, content = core.dispatch_with_results("asgi.block.started", (ctx, url))[0][0]
+                result = core.dispatch_with_results("asgi.block.started", (ctx, url)).status_headers_content
+                if result:
+                    status, headers, content = result.value
+                else:
+                    status, headers, content = 403, [], ""
                 if span and message.get("type") == "http.response.start":
                     message["headers"] = headers
                     message["status"] = int(status)

ddtrace/contrib/django/patch.py

@@ -724,9 +724,9 @@ def traced_authenticate(django, pin, func, instance, args, kwargs):
                 pin,
                 _DjangoUserInfoRetriever(result_user),
             ),
-        )[0]
-        if result and result[0][0]:
-            return result[0][1]
+        ).user
+        if result and result.value[0]:
+            return result.value[1]
 
     except Exception:
         log.debug("Error while trying to trace Django authenticate", exc_info=True)
@@ -822,7 +822,7 @@ def _(m):
 
 
 def wrap_wsgi_environ(wrapped, _instance, args, kwargs):
-    return core.dispatch_with_results("django.wsgi_environ", (wrapped, _instance, args, kwargs))[0][0]
+    return core.dispatch_with_results("django.wsgi_environ", (wrapped, _instance, args, kwargs)).wrapped_result.value
 
 
 def patch():

ddtrace/contrib/django/utils.py

@@ -269,8 +269,7 @@ def _extract_body(request):
     if request.method in _BODY_METHODS:
         req_body = None
         content_type = request.content_type if hasattr(request, "content_type") else request.META.get("CONTENT_TYPE")
-        results = core.dispatch_with_results("django.extract_body", ())[0]
-        headers = results[0] if results else None
+        headers = core.dispatch_with_results("django.extract_body").headers.value
         try:
             if content_type == "application/x-www-form-urlencoded":
                 req_body = parse_form_params(request.body.decode("UTF-8", errors="ignore"))
@@ -366,8 +365,7 @@ def _after_request_tags(pin, span: Span, request, response):
 
             url = get_request_uri(request)
 
-            results = core.dispatch_with_results("django.after_request_headers", ())[0]
-            request_headers = results[0] if results else None
+            request_headers = core.dispatch_with_results("django.after_request_headers").headers.value
             if not request_headers:
                 request_headers = _get_request_headers(request)
 

ddtrace/contrib/flask/patch.py

@@ -108,9 +108,9 @@ def _wrapped_start_response(self, start_response, ctx, status_code, headers, exc
         core.dispatch("flask.start_response.pre", (flask.request, ctx, config.flask, status_code, headers))
         if not core.get_item(HTTP_REQUEST_BLOCKED):
             headers_from_context = ""
-            results, exceptions = core.dispatch_with_results("flask.start_response", ("Flask",))
-            if not any(exceptions) and results and results[0]:
-                headers_from_context = results[0]
+            result = core.dispatch_with_results("flask.start_response", ("Flask",)).waf
+            if result:
+                headers_from_context = result.value
             if core.get_item(HTTP_REQUEST_BLOCKED):
                 # response code must be set here, or it will be too late
                 block_config = core.get_item(HTTP_REQUEST_BLOCKED)
@@ -139,7 +139,7 @@ def _request_call_modifier(self, ctx, parsed_headers=None):
         request = _RequestType(environ)
 
         req_body = None
-        results, exceptions = core.dispatch_with_results(
+        result = core.dispatch_with_results(
             "flask.request_call_modifier",
             (
                 ctx,
@@ -151,12 +151,9 @@ def _request_call_modifier(self, ctx, parsed_headers=None):
                 flask_version_str,
                 BadRequest,
             ),
-        )
-        if not any(exceptions) and results and any(results):
-            for result in results:
-                if result is not None:
-                    req_body = result
-                    break
+        ).request_body
+        if result:
+            req_body = result.value
         core.dispatch("flask.request_call_modifier.post", (ctx, config.flask, request, req_body))
 
 

ddtrace/contrib/flask/wrappers.py

@@ -47,9 +47,9 @@ def _wrap_call(
         call_key="flask_call",
     ) as ctx, ctx.get_item("flask_call"):
         if do_dispatch:
-            results, exceptions = core.dispatch_with_results("flask.wrapped_view", (kwargs,))
-            if results and results[0]:
-                callback_block, _kwargs = results[0]
+            result = core.dispatch_with_results("flask.wrapped_view", (kwargs,)).callback_and_args
+            if result:
+                callback_block, _kwargs = result.value
                 if callback_block:
                     return callback_block()
                 if _kwargs:

ddtrace/contrib/wsgi/wsgi.py

@@ -109,13 +109,21 @@ def __call__(self, environ: Iterable, start_response: Callable) -> wrapt.ObjectP
             call_key="req_span",
         ) as ctx:
             if core.get_item(HTTP_REQUEST_BLOCKED):
-                status, headers, content = core.dispatch_with_results("wsgi.block.started", (ctx, construct_url))[0][0]
+                result = core.dispatch_with_results("wsgi.block.started", (ctx, construct_url)).status_headers_content
+                if result:
+                    status, headers, content = result.value
+                else:
+                    status, headers, content = 403, [], ""
                 start_response(str(status), headers)
                 closing_iterable = [content]
                 not_blocked = False
 
             def blocked_view():
-                status, headers, content = core.dispatch_with_results("wsgi.block.started", (ctx, construct_url))[0][0]
+                result = core.dispatch_with_results("wsgi.block.started", (ctx, construct_url)).status_headers_content
+                if result:
+                    status, headers, content = result.value
+                else:
+                    status, headers, content = 403, [], ""
                 return content, status, headers
 
             core.dispatch("wsgi.block_decided", (blocked_view,))
@@ -130,12 +138,15 @@ def blocked_view():
                 else:
                     core.dispatch("wsgi.app.success", (ctx, closing_iterable))
                 if core.get_item(HTTP_REQUEST_BLOCKED):
-                    _, _, content = core.dispatch_with_results("wsgi.block.started", (ctx, construct_url))[0][0]
+                    _, _, content = core.dispatch_with_results(
+                        "wsgi.block.started", (ctx, construct_url)
+                    ).status_headers_content.value or (None, None, "")
                     closing_iterable = [content]
 
-            return core.dispatch_with_results("wsgi.request.complete", (ctx, closing_iterable, self.app_is_iterator))[
-                0
-            ][0]
+            result = core.dispatch_with_results(
+                "wsgi.request.complete", (ctx, closing_iterable, self.app_is_iterator)
+            ).traced_iterable
+            return result.value if result else []
 
     def _traced_start_response(self, start_response, request_span, app_span, status, environ, exc_info=None):
         # type: (Callable, Span, Span, str, Dict, Any) -> None