chore(iast): fastapi body

ddtrace/appsec/_iast/_patch.py

@@ -172,6 +172,7 @@ def _on_iast_fastapi_patch():
 
     # Body source
     try_wrap_function_wrapper("starlette.requests", "Request.__init__", _iast_instrument_starlette_request)
+    try_wrap_function_wrapper("starlette.requests", "Request.body", _iast_instrument_starlette_request_body)
     _set_metric_iast_instrumented_source(OriginType.BODY)
 
     # Instrumented on _iast_starlette_scope_taint
@@ -212,6 +213,18 @@ async def wrapped_property_call():
     instance.__class__.receive = property(receive)
 
 
+async def _iast_instrument_starlette_request_body(wrapped, instance, args, kwargs):
+    from ddtrace.appsec._iast._taint_tracking import OriginType
+    from ddtrace.appsec._iast._taint_tracking import origin_to_str
+    from ddtrace.appsec._iast._taint_tracking import taint_pyobject
+
+    result = await wrapped(*args, **kwargs)
+
+    return taint_pyobject(
+        result, source_name=origin_to_str(OriginType.PATH), source_value=result, source_origin=OriginType.BODY
+    )
+
+
 def _iast_instrument_starlette_scope(scope):
     from ddtrace.appsec._iast._taint_tracking import OriginType
     from ddtrace.appsec._iast._taint_tracking import taint_pyobject

tests/contrib/fastapi/test_fastapi_appsec_iast.py

@@ -301,7 +301,7 @@ async def test_route(request: Request):
 
 
 @pytest.mark.usefixtures("setup_core_ok_after_test")
-def test_path_body_source(fastapi_application, client, tracer, test_spans):
+def test_path_body_receive_source(fastapi_application, client, tracer, test_spans):
     @fastapi_application.post("/index.html")
     async def test_route(request: Request):
         from ddtrace.appsec._iast._taint_tracking import get_tainted_ranges
@@ -341,6 +341,46 @@ async def test_route(request: Request):
         assert result["ranges_origin"] == "http.request.body"
 
 
+@pytest.mark.usefixtures("setup_core_ok_after_test")
+def test_path_body_body_source(fastapi_application, client, tracer, test_spans):
+    @fastapi_application.post("/index.html")
+    async def test_route(request: Request):
+        from ddtrace.appsec._iast._taint_tracking import get_tainted_ranges
+        from ddtrace.appsec._iast._taint_tracking import origin_to_str
+
+        body = await request.body()
+        ranges_result = get_tainted_ranges(body)
+
+        return JSONResponse(
+            {
+                "result": str(body, encoding="utf-8"),
+                "is_tainted": len(ranges_result),
+                "ranges_start": ranges_result[0].start,
+                "ranges_length": ranges_result[0].length,
+                "ranges_origin": origin_to_str(ranges_result[0].source.origin),
+            }
+        )
+
+    # test if asgi middleware is ok without any callback registered
+    # core.reset_listeners(event_id="asgi.request.parse.body")
+
+    with override_global_config(dict(_iast_enabled=True)), override_env(IAST_ENV):
+        # disable callback
+        _aux_appsec_prepare_tracer(tracer)
+        resp = client.post(
+            "/index.html",
+            data='{"name": "yqrweytqwreasldhkuqwgervflnmlnli"}',
+            headers={"Content-Type": "application/json"},
+        )
+        assert resp.status_code == 200
+        result = json.loads(get_response_body(resp))
+        assert result["result"] == '{"name": "yqrweytqwreasldhkuqwgervflnmlnli"}'
+        assert result["is_tainted"] == 1
+        assert result["ranges_start"] == 0
+        assert result["ranges_length"] == 44
+        assert result["ranges_origin"] == "http.request.body"
+
+
 @pytest.mark.skip(reason="Pydantic not supported yet APPSEC-52941")
 def test_path_body_source_pydantic(fastapi_application, client, tracer, test_spans):
     from pydantic import BaseModel