feat: detect body attacks on flask (#4015)

## Description
Detect attacks on the request body when running in Flask.

## Checklist
- [X] Title must conform to [conventional commit](https://github.com/conventional-changelog/commitlint/tree/master/%40commitlint/config-conventional).
- [ ] Add additional sections for `feat` and `fix` pull requests.
- [ ] Ensure tests are passing for affected code.
- [ ] [Library documentation](https://github.com/DataDog/dd-trace-py/tree/1.x/docs) and/or [Datadog's documentation site](https://github.com/DataDog/documentation/) is updated. Link to doc PR in description.

## Reviewer Checklist
- [ ] Title is accurate.
- [ ] Description motivates each change.
- [ ] No unnecessary changes were introduced in this PR.
- [ ] PR cannot be broken up into smaller PRs.
- [ ] Avoid breaking [API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces) changes unless absolutely necessary.
- [ ] Tests provided or description of manual testing performed is included in the code or PR.
- [ ] Release note has been added for fixes and features, or else `changelog/no-changelog` label added.
- [ ] All relevant GitHub issues are correctly linked.
- [ ] Backports are identified and tagged with Mergifyio.
- [ ] Add to milestone.

Author: juanjux

ddtrace/contrib/flask/patch.py

@@ -1,5 +1,17 @@
+import json
+
 import flask
 import werkzeug
+from werkzeug.exceptions import BadRequest
+
+
+# Not all versions of flask/werkzeug have this mixin
+try:
+    from werkzeug.wrappers.json import JSONMixin
+
+    _HAS_JSON_MIXIN = True
+except ImportError:
+    _HAS_JSON_MIXIN = False
 
 from ddtrace import Pin
 from ddtrace import config
@@ -27,6 +39,7 @@
 FLASK_VIEW_ARGS = "flask.view_args"
 FLASK_URL_RULE = "flask.url_rule"
 FLASK_VERSION = "flask.version"
+_BODY_METHODS = {"POST", "PUT", "DELETE", "PATCH"}
 
 # Configure default configuration
 config._add(
@@ -42,6 +55,15 @@
 )
 
 
+if _HAS_JSON_MIXIN:
+
+    class RequestWithJson(werkzeug.Request, JSONMixin):
+        pass
+
+    _RequestType = RequestWithJson
+else:
+    _RequestType = werkzeug.Request
+
 # Extract flask version into a tuple e.g. (0, 12, 1) or (1, 0, 2)
 # DEV: This makes it so we can do `if flask_version >= (0, 12, 0):`
 # DEV: Example tests:
@@ -297,7 +319,7 @@ def traced_wsgi_app(pin, wrapped, instance, args, kwargs):
 
     # Create a werkzeug request from the `environ` to make interacting with it easier
     # DEV: This executes before a request context is created
-    request = werkzeug.Request(environ)
+    request = _RequestType(environ)
 
     # Configure distributed tracing
     trace_utils.activate_distributed_headers(pin.tracer, int_config=config.flask, request_headers=request.headers)
@@ -323,6 +345,24 @@ def traced_wsgi_app(pin, wrapped, instance, args, kwargs):
 
         start_response = _wrap_start_response(start_response, span, request)
 
+        req_body = None
+        if config._appsec_enabled and request.method in _BODY_METHODS:
+            content_type = request.content_type
+            try:
+                if content_type == "application/json":
+                    if _HAS_JSON_MIXIN and hasattr(request, "json"):
+                        req_body = request.json
+                    else:
+                        req_body = json.loads(request.data.decode("UTF-8"))
+                elif hasattr(request, "values"):
+                    req_body = request.values.to_dict()
+                elif hasattr(request, "args"):
+                    req_body = request.args.to_dict()
+                elif hasattr(request, "form"):
+                    req_body = request.form.to_dict()
+            except (AttributeError, RuntimeError, TypeError, BadRequest):
+                log.warning("Failed to parse werkzeug request body", exc_info=True)
+
         # DEV: We set response status code in `_wrap_start_response`
         # DEV: Use `request.base_url` and not `request.url` to keep from leaking any query string parameters
         trace_utils.set_http_meta(
@@ -335,6 +375,7 @@ def traced_wsgi_app(pin, wrapped, instance, args, kwargs):
             parsed_query=request.args,
             request_headers=request.headers,
             request_cookies=request.cookies,
+            request_body=req_body,
         )
 
         return wrapped(environ, start_response)

releasenotes/notes/detect-attack-on-flask-body-9305779e1c804b24.yaml

@@ -0,0 +1,5 @@
+---
+features:
+  - |
+    ASM: Detect attacks on Flask body.
+

tests/contrib/flask/test_flask_appsec.py

@@ -1,9 +1,11 @@
 import json
 
 from ddtrace.internal import _context
+from ddtrace.internal.compat import urlencode
 from tests.appsec.test_processor import RULES_GOOD_PATH
 from tests.contrib.flask import BaseFlaskTestCase
 from tests.utils import override_env
+from tests.utils import override_global_config
 
 
 class FlaskAppSecTestCase(BaseFlaskTestCase):
@@ -109,3 +111,63 @@ def test_flask_cookie(self):
         assert query == {"testingcookie_key": "testingcookie_value"} or query == {
             "testingcookie_key": ["testingcookie_value"]
         }
+
+    def test_flask_body_urlencoded(self):
+        with override_global_config(dict(_appsec_enabled=True)):
+            self.tracer._appsec_enabled = True
+            # Hack: need to pass an argument to configure so that the processors are recreated
+            self.tracer.configure(api_version="v0.4")
+            payload = urlencode({"mytestingbody_key": "mytestingbody_value"})
+            self.client.post("/", data=payload, content_type="application/x-www-form-urlencoded")
+            root_span = self.pop_spans()[0]
+            query = dict(_context.get_item("http.request.body", span=root_span))
+
+            assert root_span.get_tag("_dd.appsec.json") is None
+            assert query == {"mytestingbody_key": "mytestingbody_value"}
+
+    def test_flask_body_urlencoded_appsec_disabled_then_no_body(self):
+        with override_global_config(dict(_appsec_enabled=False)):
+            self.tracer._appsec_enabled = True
+            # Hack: need to pass an argument to configure so that the processors are recreated
+            self.tracer.configure(api_version="v0.4")
+            payload = urlencode({"mytestingbody_key": "mytestingbody_value"})
+            self.client.post("/", data=payload, content_type="application/x-www-form-urlencoded")
+            root_span = self.pop_spans()[0]
+            assert not _context.get_item("http.request.body", span=root_span)
+
+    def test_flask_request_body_urlencoded_attack(self):
+        with override_global_config(dict(_appsec_enabled=True)):
+            self.tracer._appsec_enabled = True
+            # Hack: need to pass an argument to configure so that the processors are recreated
+            self.tracer.configure(api_version="v0.4")
+            payload = urlencode({"attack": "1' or '1' = '1'"})
+            self.client.post("/", data=payload, content_type="application/x-www-form-urlencoded")
+            root_span = self.pop_spans()[0]
+            query = dict(_context.get_item("http.request.body", span=root_span))
+            assert "triggers" in json.loads(root_span.get_tag("_dd.appsec.json"))
+            assert query == {"attack": "1' or '1' = '1'"}
+
+    def test_flask_body_json(self):
+        with override_global_config(dict(_appsec_enabled=True)):
+            self.tracer._appsec_enabled = True
+            # Hack: need to pass an argument to configure so that the processors are recreated
+            self.tracer.configure(api_version="v0.4")
+            payload = {"mytestingbody_key": "mytestingbody_value"}
+            self.client.post("/", json=payload, content_type="application/json")
+            root_span = self.pop_spans()[0]
+            query = dict(_context.get_item("http.request.body", span=root_span))
+
+            assert root_span.get_tag("_dd.appsec.json") is None
+            assert query == {"mytestingbody_key": "mytestingbody_value"}
+
+    def test_flask_body_json_attack(self):
+        with override_global_config(dict(_appsec_enabled=True)):
+            self.tracer._appsec_enabled = True
+            # Hack: need to pass an argument to configure so that the processors are recreated
+            self.tracer.configure(api_version="v0.4")
+            payload = {"attack": "1' or '1' = '1'"}
+            self.client.post("/", json=payload, content_type="application/json")
+            root_span = self.pop_spans()[0]
+            query = dict(_context.get_item("http.request.body", span=root_span))
+            assert "triggers" in json.loads(root_span.get_tag("_dd.appsec.json"))
+            assert query == {"attack": "1' or '1' = '1'"}