fix(apm): fix unicode decode errors in pylons (#5029)

## 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/contributing.html#Release-Note-Guidelines)
are followed.
- [x] Documentation is included (in-code, generated user docs, [public
corp docs](https://github.com/DataDog/documentation/)).

This PR is related to a critical exception detected two days ago on
https://app.datadoghq.com/logs

<img width="1352" alt="Pylons Unicode Error"
src="https://user-images.githubusercontent.com/114495376/216595049-a0fa6bb0-5c2b-4e5f-ac97-e83000c225eb.png">

Silently drop non translatable UTF-8 characters instead of raising
critical exception in Pylons framework.

## Reviewer Checklist

- [ ] Title is accurate.
- [ ] No unnecessary changes are introduced.
- [ ] Description motivates each change.
- [ ] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes unless absolutely necessary.
- [ ] Testing strategy adequately addresses listed risk(s).
- [ ] Change is maintainable (easy to change, telemetry, documentation).
- [ ] Release note makes sense to a user of the library.

Author: christophe-papazian

ddtrace/contrib/pylons/middleware.py

@@ -107,11 +107,11 @@ def __call__(self, environ, start_response):
                         if hasattr(request, "json"):
                             req_body = request.json
                         else:
-                            req_body = json.loads(request.body.decode("UTF-8"))
+                            req_body = json.loads(request.body.decode(request.charset or "utf-8", errors="ignore"))
                     elif content_type in ("application/xml", "text/xml"):
-                        req_body = xmltodict.parse(request.body.decode("UTF-8"))
+                        req_body = xmltodict.parse(request.body.decode(request.charset or "utf-8", errors="ignore"))
                     else:  # text/plain, xml, others: take them as strings
-                        req_body = request.body.decode("UTF-8")
+                        req_body = request.body.decode(request.charset or "utf-8", errors="ignore")
 
                 except (
                     AttributeError,

releasenotes/notes/fix-unicode-decode-errors-pylons-52d28b22f82e184a.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    pylons: This fix resolves an issue where ``str.decode`` could cause critical unicode decode errors when ASM is enabled. ASM is disabled by default.

tests/contrib/pylons/test_pylons.py

@@ -1,7 +1,9 @@
+# -*- coding: utf-8 -*-
 import json
 import logging
 import os
 
+import mock
 from paste import fixture
 from paste.deploy import loadapp
 import pylons
@@ -674,11 +676,60 @@ def test_pylons_body_xml(self):
             assert span
             assert span["mytestingbody_key"] == "mytestingbody_value"
 
-    def test_pylons_body_xml_attack(self):
+    def test_pylons_body_json_unicode_decode_error_charset(self):
+        # Regression test, if request.charset returns None, a TypeError is raised
+        with self.override_global_config(dict(_appsec_enabled=True)):
+
+            with override_env(dict(DD_APPSEC_RULES=RULES_GOOD_PATH)), mock.patch(
+                "ddtrace.contrib.pylons.middleware.Request.charset", new_callable=mock.PropertyMock
+            ) as mock_charset:
+                mock_charset.return_value = None
+                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")
+                self.app.post(
+                    url_for(controller="root", action="index"),
+                    params=b"\x80",
+                    extra_environ={"CONTENT_TYPE": "application/json"},
+                )
+
+                spans = self.pop_spans()
+                assert spans
+
+                root_span = spans[0]
+                appsec_json = root_span.get_tag("_dd.appsec.json")
+                assert appsec_json is None
+
+                assert "UnicodeDecodeError" not in self._caplog.text
+                assert _context.get_item("http.request.body", span=root_span) is None
+
+    def test_pylons_body_json_unicode_decode_error(self):
+        with self.override_global_config(dict(_appsec_enabled=True)):
+            with override_env(dict(DD_APPSEC_RULES=RULES_GOOD_PATH)):
+                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")
+                self.app.post(
+                    url_for(controller="root", action="index"),
+                    params=b"\x80",
+                    extra_environ={"CONTENT_TYPE": "application/json"},
+                )
+
+                spans = self.pop_spans()
+                assert spans
+
+                root_span = spans[0]
+                appsec_json = root_span.get_tag("_dd.appsec.json")
+                assert appsec_json is None
+
+                assert "UnicodeDecodeError" not in self._caplog.text
+                assert _context.get_item("http.request.body", span=root_span) is None
+
+    def test_pylons_body_xml_attack_and_unicode_decode_error(self):
         with override_global_config(dict(_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'</attack>"
+            payload = b"\x80<attack>1' or '1' = '1'</attack>"
             self.app.post(
                 url_for(controller="root", action="index"),
                 params=payload,
@@ -690,6 +741,8 @@ def test_pylons_body_xml_attack(self):
 
             root_span = spans[0]
             assert root_span
+
+            assert "UnicodeDecodeError" not in self._caplog.text
             assert root_span.get_tag("_dd.appsec.json") is None
 
             span = dict(_context.get_item("http.request.body", span=root_span))