feat(AAP): add support for sub applications in FastAPI (#17570)

APPSEC-62352

## Description

Add Application & API Protection (AAP) support for FastAPI and Starlette applications using mounted sub-applications (`app.mount()`).

Previously, when endpoints were defined on sub-applications, several issues occurred:
- **Duplicate WAF invocations**: The WAF ran on both the intermediate `Mount` handler and the final `Route` handler, producing duplicate triggers.
- **Duplicate tracing**: Each sub-app's `TraceMiddleware` created redundant processing (body parsing, distributed tracing activation, WAF dispatch).
- **Incomplete path parameters**: `scope["path_params"]` was only partially populated when ASM processed the `Mount` handler.
- **Incorrect endpoint discovery**: Routes in sub-apps were registered with their local paths (e.g., `/get`) instead of the full mounted path (e.g., `/api/v2/get`).

### Changes

**`ddtrace/contrib/internal/asgi/middleware.py`**
- Detect sub-app middleware via `is_subapp = "datadog" in scope`. Sub-app middlewares still create spans (for trace visibility), but skip body parsing, distributed tracing activation, and WAF dispatch to avoid duplicates.
- Trigger endpoint discovery via route tree walk on first request (root middleware only).

**`ddtrace/contrib/internal/starlette/patch.py`**
- Gate ASM/WAF processing behind `isinstance(instance, starlette.routing.Route)` so it only runs on the final `Route` handler, not intermediate `Mount` handlers.
- Replace init-time endpoint registration with `_collect_routes_from_app()`, a recursive route tree walker that accumulates mount prefixes for correct full-path registration.
- Handle `Host`-based routing in the tree walker.
- Per-route error handling in the walker to prevent one bad route from blocking discovery.

**`ddtrace/appsec/_asm_request_context.py`**
- Guard `start_context()` to skip creating a new ASM environment when `is_subapp=True` and an ASM context already exists from the parent. This preserves the parent's request data (body, headers, etc.) for the WAF instead of creating an empty child context that would shadow it.

**Tests**
- Parametrize the FastAPI appsec test suite with both flat (`get_app`) and sub-app (`get_app_with_subapps`) app variants via the `interface` fixture.
- Add `app_subapps.py` with endpoints grouped into mounted sub-applications.
- Reset `endpoint_collection` singleton between parametrized variants to prevent state leakage.

## Testing

All test suites pass with both flat and sub-app variants:
- `appsec_threats_fastapi_no_iast`: 2006 passed
- `appsec_threats_fastapi_iast`: 2006 passed
- `appsec_threats_fastapi_rc`: 10 passed
- `contrib::fastapi`: 72 passed (snapshot tests unchanged)
- `contrib::starlette`: 52 passed (snapshot tests unchanged)
- `contrib::asgi`: 183 passed

## Risks

- **Endpoint discovery timing**: Endpoint registration is now deferred to first request (was at `Route.__init__` time). This is necessary because mount prefixes are unknown at init time. The app tree is guaranteed complete at first request for all standard ASGI server setups.
- **Trace shape**: Sub-app spans are preserved (child `starlette.request`/`fastapi.request` spans still appear), but they no longer carry body/query data or trigger WAF independently. Existing snapshot tests pass without changes.
- **`is_subapp` propagation**: The flag is passed via `core.context_with_data` and checked in `start_context()`. Only affects mounted sub-apps where `scope["datadog"]` is already set by the parent middleware.

## Additional Notes

- The `is_subapp` approach was chosen over a full middleware skip to preserve sub-app child spans in traces, maintaining backward compatibility with existing snapshot tests and regular monitoring behavior.




Co-authored-by: christophe.papazian <christophe.papazian@datadoghq.com>

Author: christophe-papazian

ddtrace/appsec/_asm_request_context.py

@@ -576,7 +576,12 @@ def store_waf_results_data(data: "list[WafEvent]") -> None:
 
 def start_context(waf_callable: Optional[WafCallable], span: Span, rc_products: str) -> None:
     if asm_config._asm_enabled:
-        # it should only be called at start of a core context, when ASM_Env is not set yet
+        # Skip creating a new ASM context if one already exists in a parent context
+        # AND this is a sub-app span (e.g., mounted FastAPI/Starlette sub-application).
+        # The parent's ASM context already has the request data (body, headers, etc.)
+        # and is accessible via core.find_item thanks to context tree traversal.
+        if in_asm_context() and core.find_item("is_subapp"):
+            return
         core.set_item(
             _ASM_CONTEXT,
             ASM_Environment(

ddtrace/contrib/internal/asgi/middleware.py

@@ -195,6 +195,28 @@ async def __call__(self, scope: Mapping[str, Any], receive: Callable, send: Call
             - HTTP: asgi.request spans with HTTP metadata
             - websocket: websocket.receive, websocket.send, websocket.close spans
         """
+        # Track whether this is a sub-app middleware (parent already set up tracing).
+        # Sub-app middlewares still create spans for visibility, but skip body parsing,
+        # distributed tracing activation, HTTP meta, and WAF dispatch to avoid duplicates.
+        is_subapp = "datadog" in scope
+
+        # On the first request to the root app, walk the route tree to register all
+        # endpoints (including those in mounted sub-apps) for API endpoint discovery.
+        # Only do this for the root app's middleware, not sub-app middlewares.
+        # scope["app"] is set by Starlette before its middleware stack runs.
+        if not is_subapp:
+            root_app = scope.get("app")
+            if root_app is not None and not getattr(root_app, "_datadog_endpoints_collected", False):
+                # Set the flag before attempting collection to avoid retrying on every request
+                # if the import or walk fails (e.g., starlette not patched, pure ASGI app).
+                root_app._datadog_endpoints_collected = True
+                try:
+                    from ddtrace.contrib.internal.starlette.patch import _collect_routes_from_app
+
+                    _collect_routes_from_app(root_app)
+                except Exception:
+                    log.debug("failed to collect routes from app for endpoint discovery", exc_info=True)
+
         if scope["type"] == "http":
             method = scope["method"]
         elif scope["type"] == "websocket" and self.integration_config.trace_asgi_websocket_messages:
@@ -207,9 +229,10 @@ async def __call__(self, scope: Mapping[str, Any], receive: Callable, send: Call
             log.warning("failed to decode headers for distributed tracing", exc_info=True)
             headers = {}
         else:
-            trace_utils.activate_distributed_headers(
-                tracer, int_config=self.integration_config, request_headers=headers
-            )
+            if not is_subapp:
+                trace_utils.activate_distributed_headers(
+                    tracer, int_config=self.integration_config, request_headers=headers
+                )
         resource = " ".join([method, scope["path"]])
 
         # in the case of websockets we don't currently schematize the operation names
@@ -233,6 +256,7 @@ async def __call__(self, scope: Mapping[str, Any], receive: Callable, send: Call
                 activate_distributed_headers=True,
                 scope=scope,
                 integration_config=self.integration_config,
+                is_subapp=is_subapp,
             ) as ctx,
             ctx.span as span,
         ):
@@ -281,30 +305,34 @@ async def __call__(self, scope: Mapping[str, Any], receive: Callable, send: Call
                     raw_url = f"{raw_url}?{query_string}"
             if not self.integration_config.trace_query_string:
                 query_string = None
+            # Sub-app middlewares skip body parsing since it's already handled
+            # by the parent app's middleware. HTTP meta and version tags are still
+            # set on the child span for visibility.
             body = None
-            result = core.dispatch_with_results(  # ast-grep-ignore: 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")
-            # Both list and tuple must be supported for scope["client"].
-            # In Startlette's ASGI implementation, it is a 2-item tuple (host, port). Other implementations
-            # may use a list, and Starlette's own testing code often uses a 2-item list here.
-            if isinstance(client, (list, tuple)) and len(client) and is_valid_ip(client[0]):
-                peer_ip = client[0]
-            else:
-                peer_ip = None
+            peer_ip = None
+            if not is_subapp:
+                result = core.dispatch_with_results(  # ast-grep-ignore: 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")
+                # Both list and tuple must be supported for scope["client"].
+                # In Startlette's ASGI implementation, it is a 2-item tuple (host, port). Other implementations
+                # may use a list, and Starlette's own testing code often uses a 2-item list here.
+                if isinstance(client, (list, tuple)) and len(client) and is_valid_ip(client[0]):
+                    peer_ip = client[0]
+
             trace_utils.set_http_meta(
                 span,
                 self.integration_config,
                 method=method,
                 url=url,
                 query=query_string,
                 request_headers=headers,
-                raw_uri=raw_url,
-                parsed_query=parsed_query,
+                raw_uri=raw_url if not is_subapp else None,
+                parsed_query=parsed_query if not is_subapp else None,
                 request_body=body,
                 peer_ip=peer_ip,
                 headers_are_case_sensitive=True,
@@ -474,7 +502,8 @@ async def wrapped_blocked_send(message: Mapping[str, Any]):
 
             wrapped_recv = wrapped_receive if scope["type"] == "websocket" else receive
             try:
-                core.dispatch("asgi.start_request", ("asgi",))
+                if not is_subapp:
+                    core.dispatch("asgi.start_request", ("asgi",))
                 # Do not block right here. Wait for route to be resolved in starlette/patch.py
                 return await self.app(scope, wrapped_recv, wrapped_send)
             except BlockingException as e:

ddtrace/contrib/internal/starlette/patch.py

@@ -77,25 +77,53 @@ def traced_init(wrapped, instance, args, kwargs):
 
 
 def traced_route_init(wrapped, _instance, args, kwargs):
-    route = args[0] if args else None
-    if route is not None:
-        response_body_type = getattr(kwargs.get("response_class", None), "media_type", None)
-        response_body_type = [response_body_type] if isinstance(response_body_type, str) else []
-        response_code = kwargs.get("status_code", None)
-        response_code = [response_code] if isinstance(response_code, int) else []
-        for m in kwargs.get("methods", None) or []:
-            endpoint_collection.add_endpoint(
-                m,
-                route,
-                operation_name="fastapi.request",
-                response_body_type=response_body_type,
-                response_code=response_code,
-            )
+    # Endpoint registration for the endpoint_collection is NOT done here because at
+    # Route.__init__ time, we don't know the mount prefix for sub-app routes.
+    # Instead, _collect_routes_from_app walks the full route tree on first request
+    # and registers endpoints with their complete paths (mount prefix + local route).
     handler = get_argument_value(args, kwargs, 1, "endpoint")
     core.dispatch("service_entrypoint.patch", (inspect.unwrap(handler),))
     return wrapped(*args, **kwargs)
 
 
+def _collect_routes_from_app(app, prefix=""):
+    """Walk an ASGI app's route tree and register all endpoints with their full paths.
+
+    Called once on first request via the ASGI TraceMiddleware. At that point the app
+    is fully constructed (all mounts done). Endpoint registration cannot happen at
+    Route.__init__ time because the mount prefix is unknown then (sub-apps are
+    created before being mounted).
+    """
+    routes = getattr(app, "routes", None)
+    if not routes:
+        return
+    for route in routes:
+        try:
+            if isinstance(route, starlette.routing.Mount):
+                mount_path = prefix + route.path
+                _collect_routes_from_app(route, prefix=mount_path)
+            elif hasattr(starlette.routing, "Host") and isinstance(route, starlette.routing.Host):
+                # Host-based routing: recurse into the host's app without adding a path prefix
+                _collect_routes_from_app(route, prefix=prefix)
+            elif isinstance(route, starlette.routing.Route):
+                full_path = prefix + route.path
+                response_class = getattr(route, "response_class", None)
+                media_type = getattr(response_class, "media_type", None)
+                response_body_type = [media_type] if isinstance(media_type, str) else []
+                response_code = getattr(route, "status_code", None)
+                response_code = [response_code] if isinstance(response_code, int) else []
+                for m in getattr(route, "methods", None) or []:
+                    endpoint_collection.add_endpoint(
+                        m,
+                        full_path,
+                        operation_name="fastapi.request",
+                        response_body_type=response_body_type,
+                        response_code=response_code,
+                    )
+        except Exception:
+            log.debug("failed to collect endpoint for route %r", route, exc_info=True)
+
+
 def patch():
     if getattr(starlette, "_datadog_patch", False):
         return
@@ -189,29 +217,33 @@ def traced_handler(wrapped, instance, args, kwargs):
             request_spans,
             resource_paths,
         )
-    request_cookies = ""
-    for name, value in scope.get("headers", []):
-        if name == b"cookie":
-            request_cookies = value.decode("utf-8", errors="ignore")
-            break
-
-    if request_spans:
-        if asm_config._iast_enabled:
-            from ddtrace.appsec._iast._handlers import _iast_instrument_starlette_scope
-
-            _iast_instrument_starlette_scope(scope, request_spans[0].get_tag(http.ROUTE))
-
-        trace_utils.set_http_meta(
-            request_spans[0],
-            "starlette",
-            request_path_params=scope.get("path_params"),
-            request_cookies=starlette_requests.cookie_parser(request_cookies),
-            route=request_spans[0].get_tag(http.ROUTE),
-        )
-    core.dispatch("asgi.start_request", ("starlette",))
-    blocked = get_blocked()
-    if blocked:
-        raise BlockingException(blocked)
+    # Only run ASM/WAF processing on the final Route handler, not on intermediate Mount handlers.
+    # With sub-applications, traced_handler is called for each routing layer (Mount then Route).
+    # Running ASM on Mount would cause duplicate WAF triggers and incomplete path_params.
+    if isinstance(instance, starlette.routing.Route):
+        request_cookies = ""
+        for name, value in scope.get("headers", []):
+            if name == b"cookie":
+                request_cookies = value.decode("utf-8", errors="ignore")
+                break
+
+        if request_spans:
+            if asm_config._iast_enabled:
+                from ddtrace.appsec._iast._handlers import _iast_instrument_starlette_scope
+
+                _iast_instrument_starlette_scope(scope, request_spans[0].get_tag(http.ROUTE))
+
+            trace_utils.set_http_meta(
+                request_spans[0],
+                "starlette",
+                request_path_params=scope.get("path_params"),
+                request_cookies=starlette_requests.cookie_parser(request_cookies),
+                route=request_spans[0].get_tag(http.ROUTE),
+            )
+        core.dispatch("asgi.start_request", ("starlette",))
+        blocked = get_blocked()
+        if blocked:
+            raise BlockingException(blocked)
 
     # https://github.com/encode/starlette/issues/1336
     if _STARLETTE_VERSION_LTE_0_33_0 and len(request_spans) > 1:

releasenotes/notes/fastapi-starlette-subapp-support-78e7acbf358bafc3.yaml

@@ -0,0 +1,7 @@
+---
+features:
+  - |
+    AAP: This adds Application Security support for FastAPI and Starlette applications using
+    mounted sub-applications (via ``app.mount()``). WAF evaluation, path parameter extraction,
+    API endpoint discovery, and ``http.route`` reporting now correctly account for mount prefixes
+    in sub-application routing.