fix(appsec): catch correctly BlockingException aggregated in ExceptionGroup (#14250)

## Description

When an Appsec `BlockingException` is raised within a fastapi
middleware, it can be aggregated and nested in a `BaseExceptionGroup`
multiple levels deep. The behaviour occurs when using http middlewares.

## 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)

Author: florentinl

ddtrace/contrib/internal/asgi/middleware.py

@@ -22,6 +22,7 @@
 from ddtrace.ext.net import TARGET_HOST
 from ddtrace.internal import core
 from ddtrace.internal._exceptions import BlockingException
+from ddtrace.internal._exceptions import find_exception
 from ddtrace.internal.compat import is_valid_ip
 from ddtrace.internal.constants import COMPONENT
 from ddtrace.internal.constants import SAMPLING_DECISION_MAKER_INHERITED
@@ -498,11 +499,9 @@ async def wrapped_blocked_send(message: Mapping[str, Any]):
                 raise
             except BaseException as exception:
                 # managing python 3.11+ BaseExceptionGroup with compatible code for 3.10 and below
-                if exception.__class__.__name__ == "BaseExceptionGroup":
-                    for exc in exception.exceptions:
-                        if isinstance(exc, BlockingException):
-                            set_blocked(exc.args[0])
-                            return await _blocked_asgi_app(scope, receive, wrapped_blocked_send)
+                if exc := find_exception(exception, BlockingException):
+                    set_blocked(exc.args[0])
+                    return await _blocked_asgi_app(scope, receive, wrapped_blocked_send)
                 raise
             finally:
                 core.dispatch("web.request.final_tags", (span,))

ddtrace/internal/_exceptions.py

@@ -1,5 +1,29 @@
+from typing import Optional
+from typing import Type
+from typing import TypeVar
+
+
 class BlockingException(BaseException):
     """
     Exception raised when a request is blocked by ASM
     It derives from BaseException to avoid being caught by the general Exception handler
     """
+
+
+E = TypeVar("E", bound=BaseException)
+
+
+def find_exception(
+    exc: BaseException,
+    exception_type: Type[E],
+) -> Optional[E]:
+    """Traverse an exception and its children to find the first occurrence of a specific exception type."""
+    if isinstance(exc, exception_type):
+        return exc
+    # The check matches both native Python3.11+ and `exceptiongroup` compatibility package versions of ExceptionGroup
+    if exc.__class__.__name__ in ("BaseExceptionGroup", "ExceptionGroup") and hasattr(exc, "exceptions"):
+        for sub_exc in exc.exceptions:
+            found = find_exception(sub_exc, exception_type)
+            if found:
+                return found
+    return None

releasenotes/notes/asgi-exception-group-catching-5d5132b367b01f4b.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    AAP: resolves a bug where ASGI middleware would not catch the BlockingException raised by AAP because it was aggregated in an ExceptionGroup

tests/appsec/contrib_appsec/fastapi_app/app.py

@@ -39,6 +39,18 @@ class User(BaseModel):
 def get_app():
     app = FastAPI()
 
+    @app.middleware("http")
+    async def passthrough_middleware(request: Request, call_next):
+        """Middleware to test BlockingException nesting in ExceptionGroups (or BaseExceptionGroups)
+
+        With middlewares, the BlockingException can become nested multiple levels deep inside
+        an ExceptionGroup (or BaseExceptionGroup). The nesting depends the version of FastAPI
+        and AnyIO used, as well as the version of python.
+        By adding this empty middleware, we ensure that the BlockingException is catched
+        no matter how deep the ExceptionGroup is nested or else the contrib tests fail.
+        """
+        return await call_next(request)
+
     @app.get("/")
     @app.post("/")
     @app.options("/")