We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
UnicodeDecodeError for invalid headers in ASGI middleware
When the ASGI middleware receives headers with invalid encoding, for example as the result of somebody fuzzing an endpoint as is the case here, it introduces a Unicode decode error into the HTTP pipeline. It shouldn't be doing this.

### Which version of dd-trace-py are you using?

0.46.0

### Which version of the libraries are you using?

What I believe are the potentially relevant libraries are listed below.

```
datadog==0.40.1
...
ddtrace==0.46.0
...
fastapi==0.63.0
...
httpcore==0.12.3
httptools==0.1.1
...
starlette==0.13.6
...
uvicorn==0.13.4
uvloop==0.15.2
```

### How can we reproduce your problem?

```python
from ddtrace.contrib.asgi.middleware import _extract_headers


def test_ddtrace_invalid_header_name_encoding():
    assert _extract_headers({"headers": [((bytes.fromhex("c0")), "test")]}) == {}


def test_ddtrace_invalid_header_value_encoding():
    assert _extract_headers({"headers": [("test", (bytes.fromhex("c0")))]}) == {}
```

### What is the result that you get?

The middleware blows up with a `UnicodeDecodeError: 'utf-8' codec can't decode byte 0xc0 in position 0: invalid start byte` exception.

### What is the result that you expected?

This is trace middleware so it should not introduce exceptions into request processing. In the case of a header it can't decode then I'd expect it to either:

1. Omit the invalid header from the result and continue attempting to decode the remainder, or
2. Substitute a sentinel value for the unprocessable header to indicate an error

I've suggested (1) in the repro above, as that seems like it'd be the least likely to cause any problems.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `ddtrace/contrib/asgi/middleware.py`
Content:
```
1 import sys
2 
3 import ddtrace
4 from ddtrace import config
5 from ddtrace.constants import ANALYTICS_SAMPLE_RATE_KEY
6 from ddtrace.ext import SpanTypes
7 from ddtrace.ext import http
8 
9 from .. import trace_utils
10 from ...internal.compat import reraise
11 from ...internal.logger import get_logger
12 from .utils import guarantee_single_callable
13 
14 
15 log = get_logger(__name__)
16 
17 config._add(
18     "asgi",
19     dict(service_name=config._get_service(default="asgi"), request_span_name="asgi.request", distributed_tracing=True),
20 )
21 
22 ASGI_VERSION = "asgi.version"
23 ASGI_SPEC_VERSION = "asgi.spec_version"
24 
25 
26 def bytes_to_str(str_or_bytes):
27     return str_or_bytes.decode() if isinstance(str_or_bytes, bytes) else str_or_bytes
28 
29 
30 def _extract_versions_from_scope(scope, integration_config):
31     tags = {}
32 
33     http_version = scope.get("http_version")
34     if http_version:
35         tags[http.VERSION] = http_version
36 
37     scope_asgi = scope.get("asgi")
38 
39     if scope_asgi and "version" in scope_asgi:
40         tags[ASGI_VERSION] = scope_asgi["version"]
41 
42     if scope_asgi and "spec_version" in scope_asgi:
43         tags[ASGI_SPEC_VERSION] = scope_asgi["spec_version"]
44 
45     return tags
46 
47 
48 def _extract_headers(scope):
49     headers = scope.get("headers")
50     if headers:
51         # headers: (Iterable[[byte string, byte string]])
52         return dict((bytes_to_str(k), bytes_to_str(v)) for (k, v) in headers)
53     return {}
54 
55 
56 def _default_handle_exception_span(exc, span):
57     """Default handler for exception for span"""
58     span.set_tag(http.STATUS_CODE, 500)
59 
60 
61 class TraceMiddleware:
62     """
63     ASGI application middleware that traces the requests.
64 
65     Args:
66         app: The ASGI application.
67         tracer: Custom tracer. Defaults to the global tracer.
68     """
69 
70     def __init__(
71         self,
72         app,
73         tracer=None,
74         integration_config=config.asgi,
75         handle_exception_span=_default_handle_exception_span,
76         span_modifier=None,
77     ):
78         self.app = guarantee_single_callable(app)
79         self.tracer = tracer or ddtrace.tracer
80         self.integration_config = integration_config
81         self.handle_exception_span = handle_exception_span
82         self.span_modifier = span_modifier
83 
84     async def __call__(self, scope, receive, send):
85         if scope["type"] != "http":
86             return await self.app(scope, receive, send)
87 
88         headers = _extract_headers(scope)
89 
90         trace_utils.activate_distributed_headers(
91             self.tracer, int_config=self.integration_config, request_headers=headers
92         )
93 
94         resource = "{} {}".format(scope["method"], scope["path"])
95 
96         span = self.tracer.trace(
97             name=self.integration_config.get("request_span_name", "asgi.request"),
98             service=trace_utils.int_service(None, self.integration_config),
99             resource=resource,
100             span_type=SpanTypes.WEB,
101         )
102 
103         if self.span_modifier:
104             self.span_modifier(span, scope)
105 
106         sample_rate = self.integration_config.get_analytics_sample_rate(use_global_config=True)
107         if sample_rate is not None:
108             span.set_tag(ANALYTICS_SAMPLE_RATE_KEY, sample_rate)
109 
110         method = scope.get("method")
111         server = scope.get("server")
112         if server and len(server) == 2:
113             port = server[1]
114             server_host = server[0] + (":" + str(port) if port is not None and port != 80 else "")
115             full_path = scope.get("root_path", "") + scope.get("path", "")
116             url = scope.get("scheme", "http") + "://" + server_host + full_path
117         else:
118             url = None
119 
120         if self.integration_config.trace_query_string:
121             query_string = scope.get("query_string")
122             if len(query_string) > 0:
123                 query_string = bytes_to_str(query_string)
124         else:
125             query_string = None
126 
127         trace_utils.set_http_meta(
128             span, self.integration_config, method=method, url=url, query=query_string, request_headers=headers
129         )
130 
131         tags = _extract_versions_from_scope(scope, self.integration_config)
132         span.set_tags(tags)
133 
134         async def wrapped_send(message):
135             if span and message.get("type") == "http.response.start" and "status" in message:
136                 status_code = message["status"]
137             else:
138                 status_code = None
139 
140             if "headers" in message:
141                 response_headers = message["headers"]
142             else:
143                 response_headers = None
144 
145             trace_utils.set_http_meta(
146                 span, self.integration_config, status_code=status_code, response_headers=response_headers
147             )
148 
149             return await send(message)
150 
151         try:
152             return await self.app(scope, receive, wrapped_send)
153         except Exception as exc:
154             (exc_type, exc_val, exc_tb) = sys.exc_info()
155             span.set_exc_info(exc_type, exc_val, exc_tb)
156             self.handle_exception_span(exc, span)
157             reraise(exc_type, exc_val, exc_tb)
158         finally:
159             span.finish()
160
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/ddtrace/contrib/asgi/middleware.py b/ddtrace/contrib/asgi/middleware.py
--- a/ddtrace/contrib/asgi/middleware.py
+++ b/ddtrace/contrib/asgi/middleware.py
@@ -85,11 +85,15 @@
         if scope["type"] != "http":
             return await self.app(scope, receive, send)
 
-        headers = _extract_headers(scope)
-
-        trace_utils.activate_distributed_headers(
-            self.tracer, int_config=self.integration_config, request_headers=headers
-        )
+        try:
+            headers = _extract_headers(scope)
+        except Exception:
+            log.warning("failed to decode headers for distributed tracing", exc_info=True)
+            headers = {}
+        else:
+            trace_utils.activate_distributed_headers(
+                self.tracer, int_config=self.integration_config, request_headers=headers
+            )
 
         resource = "{} {}".format(scope["method"], scope["path"])

{"golden_diff": "diff --git a/ddtrace/contrib/asgi/middleware.py b/ddtrace/contrib/asgi/middleware.py\n--- a/ddtrace/contrib/asgi/middleware.py\n+++ b/ddtrace/contrib/asgi/middleware.py\n@@ -85,11 +85,15 @@\n         if scope[\"type\"] != \"http\":\n             return await self.app(scope, receive, send)\n \n-        headers = _extract_headers(scope)\n-\n-        trace_utils.activate_distributed_headers(\n-            self.tracer, int_config=self.integration_config, request_headers=headers\n-        )\n+        try:\n+            headers = _extract_headers(scope)\n+        except Exception:\n+            log.warning(\"failed to decode headers for distributed tracing\", exc_info=True)\n+            headers = {}\n+        else:\n+            trace_utils.activate_distributed_headers(\n+                self.tracer, int_config=self.integration_config, request_headers=headers\n+            )\n \n         resource = \"{} {}\".format(scope[\"method\"], scope[\"path\"])\n", "issue": "UnicodeDecodeError for invalid headers in ASGI middleware\nWhen the ASGI middleware receives headers with invalid encoding, for example as the result of somebody fuzzing an endpoint as is the case here, it introduces a Unicode decode error into the HTTP pipeline. It shouldn't be doing this.\r\n\r\n### Which version of dd-trace-py are you using?\r\n\r\n0.46.0\r\n\r\n### Which version of the libraries are you using?\r\n\r\nWhat I believe are the potentially relevant libraries are listed below.\r\n\r\n```\r\ndatadog==0.40.1\r\n...\r\nddtrace==0.46.0\r\n...\r\nfastapi==0.63.0\r\n...\r\nhttpcore==0.12.3\r\nhttptools==0.1.1\r\n...\r\nstarlette==0.13.6\r\n...\r\nuvicorn==0.13.4\r\nuvloop==0.15.2\r\n```\r\n\r\n### How can we reproduce your problem?\r\n\r\n```python\r\nfrom ddtrace.contrib.asgi.middleware import _extract_headers\r\n\r\n\r\ndef test_ddtrace_invalid_header_name_encoding():\r\n    assert _extract_headers({\"headers\": [((bytes.fromhex(\"c0\")), \"test\")]}) == {}\r\n\r\n\r\ndef test_ddtrace_invalid_header_value_encoding():\r\n    assert _extract_headers({\"headers\": [(\"test\", (bytes.fromhex(\"c0\")))]}) == {}\r\n```\r\n\r\n### What is the result that you get?\r\n\r\nThe middleware blows up with a `UnicodeDecodeError: 'utf-8' codec can't decode byte 0xc0 in position 0: invalid start byte` exception.\r\n\r\n### What is the result that you expected?\r\n\r\nThis is trace middleware so it should not introduce exceptions into request processing. In the case of a header it can't decode then I'd expect it to either:\r\n\r\n1. Omit the invalid header from the result and continue attempting to decode the remainder, or\r\n2. Substitute a sentinel value for the unprocessable header to indicate an error\r\n\r\nI've suggested (1) in the repro above, as that seems like it'd be the least likely to cause any problems.\n", "before_files": [{"content": "import sys\n\nimport ddtrace\nfrom ddtrace import config\nfrom ddtrace.constants import ANALYTICS_SAMPLE_RATE_KEY\nfrom ddtrace.ext import SpanTypes\nfrom ddtrace.ext import http\n\nfrom .. import trace_utils\nfrom ...internal.compat import reraise\nfrom ...internal.logger import get_logger\nfrom .utils import guarantee_single_callable\n\n\nlog = get_logger(__name__)\n\nconfig._add(\n    \"asgi\",\n    dict(service_name=config._get_service(default=\"asgi\"), request_span_name=\"asgi.request\", distributed_tracing=True),\n)\n\nASGI_VERSION = \"asgi.version\"\nASGI_SPEC_VERSION = \"asgi.spec_version\"\n\n\ndef bytes_to_str(str_or_bytes):\n    return str_or_bytes.decode() if isinstance(str_or_bytes, bytes) else str_or_bytes\n\n\ndef _extract_versions_from_scope(scope, integration_config):\n    tags = {}\n\n    http_version = scope.get(\"http_version\")\n    if http_version:\n        tags[http.VERSION] = http_version\n\n    scope_asgi = scope.get(\"asgi\")\n\n    if scope_asgi and \"version\" in scope_asgi:\n        tags[ASGI_VERSION] = scope_asgi[\"version\"]\n\n    if scope_asgi and \"spec_version\" in scope_asgi:\n        tags[ASGI_SPEC_VERSION] = scope_asgi[\"spec_version\"]\n\n    return tags\n\n\ndef _extract_headers(scope):\n    headers = scope.get(\"headers\")\n    if headers:\n        # headers: (Iterable[[byte string, byte string]])\n        return dict((bytes_to_str(k), bytes_to_str(v)) for (k, v) in headers)\n    return {}\n\n\ndef _default_handle_exception_span(exc, span):\n    \"\"\"Default handler for exception for span\"\"\"\n    span.set_tag(http.STATUS_CODE, 500)\n\n\nclass TraceMiddleware:\n    \"\"\"\n    ASGI application middleware that traces the requests.\n\n    Args:\n        app: The ASGI application.\n        tracer: Custom tracer. Defaults to the global tracer.\n    \"\"\"\n\n    def __init__(\n        self,\n        app,\n        tracer=None,\n        integration_config=config.asgi,\n        handle_exception_span=_default_handle_exception_span,\n        span_modifier=None,\n    ):\n        self.app = guarantee_single_callable(app)\n        self.tracer = tracer or ddtrace.tracer\n        self.integration_config = integration_config\n        self.handle_exception_span = handle_exception_span\n        self.span_modifier = span_modifier\n\n    async def __call__(self, scope, receive, send):\n        if scope[\"type\"] != \"http\":\n            return await self.app(scope, receive, send)\n\n        headers = _extract_headers(scope)\n\n        trace_utils.activate_distributed_headers(\n            self.tracer, int_config=self.integration_config, request_headers=headers\n        )\n\n        resource = \"{} {}\".format(scope[\"method\"], scope[\"path\"])\n\n        span = self.tracer.trace(\n            name=self.integration_config.get(\"request_span_name\", \"asgi.request\"),\n            service=trace_utils.int_service(None, self.integration_config),\n            resource=resource,\n            span_type=SpanTypes.WEB,\n        )\n\n        if self.span_modifier:\n            self.span_modifier(span, scope)\n\n        sample_rate = self.integration_config.get_analytics_sample_rate(use_global_config=True)\n        if sample_rate is not None:\n            span.set_tag(ANALYTICS_SAMPLE_RATE_KEY, sample_rate)\n\n        method = scope.get(\"method\")\n        server = scope.get(\"server\")\n        if server and len(server) == 2:\n            port = server[1]\n            server_host = server[0] + (\":\" + str(port) if port is not None and port != 80 else \"\")\n            full_path = scope.get(\"root_path\", \"\") + scope.get(\"path\", \"\")\n            url = scope.get(\"scheme\", \"http\") + \"://\" + server_host + full_path\n        else:\n            url = None\n\n        if self.integration_config.trace_query_string:\n            query_string = scope.get(\"query_string\")\n            if len(query_string) > 0:\n                query_string = bytes_to_str(query_string)\n        else:\n            query_string = None\n\n        trace_utils.set_http_meta(\n            span, self.integration_config, method=method, url=url, query=query_string, request_headers=headers\n        )\n\n        tags = _extract_versions_from_scope(scope, self.integration_config)\n        span.set_tags(tags)\n\n        async def wrapped_send(message):\n            if span and message.get(\"type\") == \"http.response.start\" and \"status\" in message:\n                status_code = message[\"status\"]\n            else:\n                status_code = None\n\n            if \"headers\" in message:\n                response_headers = message[\"headers\"]\n            else:\n                response_headers = None\n\n            trace_utils.set_http_meta(\n                span, self.integration_config, status_code=status_code, response_headers=response_headers\n            )\n\n            return await send(message)\n\n        try:\n            return await self.app(scope, receive, wrapped_send)\n        except Exception as exc:\n            (exc_type, exc_val, exc_tb) = sys.exc_info()\n            span.set_exc_info(exc_type, exc_val, exc_tb)\n            self.handle_exception_span(exc, span)\n            reraise(exc_type, exc_val, exc_tb)\n        finally:\n            span.finish()\n", "path": "ddtrace/contrib/asgi/middleware.py"}], "after_files": [{"content": "import sys\n\nimport ddtrace\nfrom ddtrace import config\nfrom ddtrace.constants import ANALYTICS_SAMPLE_RATE_KEY\nfrom ddtrace.ext import SpanTypes\nfrom ddtrace.ext import http\n\nfrom .. import trace_utils\nfrom ...internal.compat import reraise\nfrom ...internal.logger import get_logger\nfrom .utils import guarantee_single_callable\n\n\nlog = get_logger(__name__)\n\nconfig._add(\n    \"asgi\",\n    dict(service_name=config._get_service(default=\"asgi\"), request_span_name=\"asgi.request\", distributed_tracing=True),\n)\n\nASGI_VERSION = \"asgi.version\"\nASGI_SPEC_VERSION = \"asgi.spec_version\"\n\n\ndef bytes_to_str(str_or_bytes):\n    return str_or_bytes.decode() if isinstance(str_or_bytes, bytes) else str_or_bytes\n\n\ndef _extract_versions_from_scope(scope, integration_config):\n    tags = {}\n\n    http_version = scope.get(\"http_version\")\n    if http_version:\n        tags[http.VERSION] = http_version\n\n    scope_asgi = scope.get(\"asgi\")\n\n    if scope_asgi and \"version\" in scope_asgi:\n        tags[ASGI_VERSION] = scope_asgi[\"version\"]\n\n    if scope_asgi and \"spec_version\" in scope_asgi:\n        tags[ASGI_SPEC_VERSION] = scope_asgi[\"spec_version\"]\n\n    return tags\n\n\ndef _extract_headers(scope):\n    headers = scope.get(\"headers\")\n    if headers:\n        # headers: (Iterable[[byte string, byte string]])\n        return dict((bytes_to_str(k), bytes_to_str(v)) for (k, v) in headers)\n    return {}\n\n\ndef _default_handle_exception_span(exc, span):\n    \"\"\"Default handler for exception for span\"\"\"\n    span.set_tag(http.STATUS_CODE, 500)\n\n\nclass TraceMiddleware:\n    \"\"\"\n    ASGI application middleware that traces the requests.\n\n    Args:\n        app: The ASGI application.\n        tracer: Custom tracer. Defaults to the global tracer.\n    \"\"\"\n\n    def __init__(\n        self,\n        app,\n        tracer=None,\n        integration_config=config.asgi,\n        handle_exception_span=_default_handle_exception_span,\n        span_modifier=None,\n    ):\n        self.app = guarantee_single_callable(app)\n        self.tracer = tracer or ddtrace.tracer\n        self.integration_config = integration_config\n        self.handle_exception_span = handle_exception_span\n        self.span_modifier = span_modifier\n\n    async def __call__(self, scope, receive, send):\n        if scope[\"type\"] != \"http\":\n            return await self.app(scope, receive, send)\n\n        try:\n            headers = _extract_headers(scope)\n        except Exception:\n            log.warning(\"failed to decode headers for distributed tracing\", exc_info=True)\n            headers = {}\n        else:\n            trace_utils.activate_distributed_headers(\n                self.tracer, int_config=self.integration_config, request_headers=headers\n            )\n\n        resource = \"{} {}\".format(scope[\"method\"], scope[\"path\"])\n\n        span = self.tracer.trace(\n            name=self.integration_config.get(\"request_span_name\", \"asgi.request\"),\n            service=trace_utils.int_service(None, self.integration_config),\n            resource=resource,\n            span_type=SpanTypes.WEB,\n        )\n\n        if self.span_modifier:\n            self.span_modifier(span, scope)\n\n        sample_rate = self.integration_config.get_analytics_sample_rate(use_global_config=True)\n        if sample_rate is not None:\n            span.set_tag(ANALYTICS_SAMPLE_RATE_KEY, sample_rate)\n\n        method = scope.get(\"method\")\n        server = scope.get(\"server\")\n        if server and len(server) == 2:\n            port = server[1]\n            server_host = server[0] + (\":\" + str(port) if port is not None and port != 80 else \"\")\n            full_path = scope.get(\"root_path\", \"\") + scope.get(\"path\", \"\")\n            url = scope.get(\"scheme\", \"http\") + \"://\" + server_host + full_path\n        else:\n            url = None\n\n        if self.integration_config.trace_query_string:\n            query_string = scope.get(\"query_string\")\n            if len(query_string) > 0:\n                query_string = bytes_to_str(query_string)\n        else:\n            query_string = None\n\n        trace_utils.set_http_meta(\n            span, self.integration_config, method=method, url=url, query=query_string, request_headers=headers\n        )\n\n        tags = _extract_versions_from_scope(scope, self.integration_config)\n        span.set_tags(tags)\n\n        async def wrapped_send(message):\n            if span and message.get(\"type\") == \"http.response.start\" and \"status\" in message:\n                status_code = message[\"status\"]\n            else:\n                status_code = None\n\n            if \"headers\" in message:\n                response_headers = message[\"headers\"]\n            else:\n                response_headers = None\n\n            trace_utils.set_http_meta(\n                span, self.integration_config, status_code=status_code, response_headers=response_headers\n            )\n\n            return await send(message)\n\n        try:\n            return await self.app(scope, receive, wrapped_send)\n        except Exception as exc:\n            (exc_type, exc_val, exc_tb) = sys.exc_info()\n            span.set_exc_info(exc_type, exc_val, exc_tb)\n            self.handle_exception_span(exc, span)\n            reraise(exc_type, exc_val, exc_tb)\n        finally:\n            span.finish()\n", "path": "ddtrace/contrib/asgi/middleware.py"}]}