We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
`remove_url_credentials` drops brackets from IPv6 hostnames
**Describe your environment** 

Python 3.11.9, `opentelemetry-instrumentation` is auto-injected via the OpenTelemetry operator, `opentelemetry_util_http` is `0.44b0`

**Steps to reproduce**
```python

from opentelemetry.util.http import remove_url_credentials

literal_ipv6_url = "https://[::1]/somepath?query=foo"
remove_url_credentials(literal_ipv6_url)
# 'https://::1/somepath?query=foo' -- should be 'https://[::1]/somepath?query=foo'

literal_ipv6_url_with_port = "https://[::1]:12345/somepath?query=foo"
remove_url_credentials(literal_ipv6_url_with_port)
# 'https://::1:12345/somepath?query=foo -- should be 'https://[::1]:12345/somepath?query=foo'

literal_ipv6_url_with_auth = "https://someuser:somepass@[::1]:12345/somepath?query=foo"
remove_url_credentials(literal_ipv6_url_with_auth)
# 'https://::1:12345/somepath?query=foo' -- should be https://[::1]:12345/somepath?query=foo

```

**What is the expected behavior?**
The ipv6 host should remain inside `[]` 

**What is the actual behavior?**
`[]` are stripped from the host

**Additional context**
https://github.com/open-telemetry/opentelemetry-python-contrib/blob/main/util/opentelemetry-util-http/src/opentelemetry/util/http/__init__.py#L169 is the causing line. The `hostname` result on `urlparse` does not contain the brackets


```python
from urllib.parse import urlparse

parsed = urlparse(literal_ipv6_url_with_auth)

parsed
# ParseResult(scheme='https', netloc='someuser:somepass@[::1]:12345', path='/somepath', params='', query='query=foo', fragment='')

parsed.hostname
# '::1'
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `util/opentelemetry-util-http/src/opentelemetry/util/http/__init__.py`
Content:
```
1 # Copyright The OpenTelemetry Authors
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 from __future__ import annotations
16 
17 from os import environ
18 from re import IGNORECASE as RE_IGNORECASE
19 from re import compile as re_compile
20 from re import search
21 from typing import Callable, Iterable, Optional
22 from urllib.parse import urlparse, urlunparse
23 
24 from opentelemetry.semconv.trace import SpanAttributes
25 
26 OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SANITIZE_FIELDS = (
27     "OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SANITIZE_FIELDS"
28 )
29 OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST = (
30     "OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST"
31 )
32 OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE = (
33     "OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE"
34 )
35 
36 OTEL_PYTHON_INSTRUMENTATION_HTTP_CAPTURE_ALL_METHODS = (
37     "OTEL_PYTHON_INSTRUMENTATION_HTTP_CAPTURE_ALL_METHODS"
38 )
39 
40 # List of recommended metrics attributes
41 _duration_attrs = {
42     SpanAttributes.HTTP_METHOD,
43     SpanAttributes.HTTP_HOST,
44     SpanAttributes.HTTP_SCHEME,
45     SpanAttributes.HTTP_STATUS_CODE,
46     SpanAttributes.HTTP_FLAVOR,
47     SpanAttributes.HTTP_SERVER_NAME,
48     SpanAttributes.NET_HOST_NAME,
49     SpanAttributes.NET_HOST_PORT,
50 }
51 
52 _active_requests_count_attrs = {
53     SpanAttributes.HTTP_METHOD,
54     SpanAttributes.HTTP_HOST,
55     SpanAttributes.HTTP_SCHEME,
56     SpanAttributes.HTTP_FLAVOR,
57     SpanAttributes.HTTP_SERVER_NAME,
58 }
59 
60 
61 class ExcludeList:
62     """Class to exclude certain paths (given as a list of regexes) from tracing requests"""
63 
64     def __init__(self, excluded_urls: Iterable[str]):
65         self._excluded_urls = excluded_urls
66         if self._excluded_urls:
67             self._regex = re_compile("|".join(excluded_urls))
68 
69     def url_disabled(self, url: str) -> bool:
70         return bool(self._excluded_urls and search(self._regex, url))
71 
72 
73 class SanitizeValue:
74     """Class to sanitize (remove sensitive data from) certain headers (given as a list of regexes)"""
75 
76     def __init__(self, sanitized_fields: Iterable[str]):
77         self._sanitized_fields = sanitized_fields
78         if self._sanitized_fields:
79             self._regex = re_compile("|".join(sanitized_fields), RE_IGNORECASE)
80 
81     def sanitize_header_value(self, header: str, value: str) -> str:
82         return (
83             "[REDACTED]"
84             if (self._sanitized_fields and search(self._regex, header))
85             else value
86         )
87 
88     def sanitize_header_values(
89         self,
90         headers: dict[str, str],
91         header_regexes: list[str],
92         normalize_function: Callable[[str], str],
93     ) -> dict[str, str]:
94         values: dict[str, str] = {}
95 
96         if header_regexes:
97             header_regexes_compiled = re_compile(
98                 "|".join("^" + i + "$" for i in header_regexes),
99                 RE_IGNORECASE,
100             )
101 
102             for header_name in list(
103                 filter(
104                     header_regexes_compiled.match,
105                     headers.keys(),
106                 )
107             ):
108                 header_values = headers.get(header_name)
109                 if header_values:
110                     key = normalize_function(header_name.lower())
111                     values[key] = [
112                         self.sanitize_header_value(
113                             header=header_name, value=header_values
114                         )
115                     ]
116 
117         return values
118 
119 
120 _root = r"OTEL_PYTHON_{}"
121 
122 
123 def get_traced_request_attrs(instrumentation):
124     traced_request_attrs = environ.get(
125         _root.format(f"{instrumentation}_TRACED_REQUEST_ATTRS"), []
126     )
127 
128     if traced_request_attrs:
129         traced_request_attrs = [
130             traced_request_attr.strip()
131             for traced_request_attr in traced_request_attrs.split(",")
132         ]
133 
134     return traced_request_attrs
135 
136 
137 def get_excluded_urls(instrumentation: str) -> ExcludeList:
138     # Get instrumentation-specific excluded URLs. If not set, retrieve them
139     # from generic variable.
140     excluded_urls = environ.get(
141         _root.format(f"{instrumentation}_EXCLUDED_URLS"),
142         environ.get(_root.format("EXCLUDED_URLS"), ""),
143     )
144 
145     return parse_excluded_urls(excluded_urls)
146 
147 
148 def parse_excluded_urls(excluded_urls: str) -> ExcludeList:
149     """
150     Small helper to put an arbitrary url list inside an ExcludeList
151     """
152     if excluded_urls:
153         excluded_url_list = [
154             excluded_url.strip() for excluded_url in excluded_urls.split(",")
155         ]
156     else:
157         excluded_url_list = []
158 
159     return ExcludeList(excluded_url_list)
160 
161 
162 def remove_url_credentials(url: str) -> str:
163     """Given a string url, remove the username and password only if it is a valid url"""
164 
165     try:
166         parsed = urlparse(url)
167         if all([parsed.scheme, parsed.netloc]):  # checks for valid url
168             parsed_url = urlparse(url)
169             netloc = (
170                 (":".join(((parsed_url.hostname or ""), str(parsed_url.port))))
171                 if parsed_url.port
172                 else (parsed_url.hostname or "")
173             )
174             return urlunparse(
175                 (
176                     parsed_url.scheme,
177                     netloc,
178                     parsed_url.path,
179                     parsed_url.params,
180                     parsed_url.query,
181                     parsed_url.fragment,
182                 )
183             )
184     except ValueError:  # an unparsable url was passed
185         pass
186     return url
187 
188 
189 def normalise_request_header_name(header: str) -> str:
190     key = header.lower().replace("-", "_")
191     return f"http.request.header.{key}"
192 
193 
194 def normalise_response_header_name(header: str) -> str:
195     key = header.lower().replace("-", "_")
196     return f"http.response.header.{key}"
197 
198 
199 def sanitize_method(method: Optional[str]) -> Optional[str]:
200     if method is None:
201         return None
202     method = method.upper()
203     if (
204         environ.get(OTEL_PYTHON_INSTRUMENTATION_HTTP_CAPTURE_ALL_METHODS)
205         or
206         # Based on https://www.rfc-editor.org/rfc/rfc7231#section-4.1 and https://www.rfc-editor.org/rfc/rfc5789#section-2.
207         method
208         in [
209             "GET",
210             "HEAD",
211             "POST",
212             "PUT",
213             "DELETE",
214             "CONNECT",
215             "OPTIONS",
216             "TRACE",
217             "PATCH",
218         ]
219     ):
220         return method
221     return "_OTHER"
222 
223 
224 def get_custom_headers(env_var: str) -> list[str]:
225     custom_headers = environ.get(env_var, None)
226     if custom_headers:
227         return [
228             custom_headers.strip()
229             for custom_headers in custom_headers.split(",")
230         ]
231     return []
232 
233 
234 def _parse_active_request_count_attrs(req_attrs):
235     active_requests_count_attrs = {
236         key: req_attrs[key]
237         for key in _active_requests_count_attrs.intersection(req_attrs.keys())
238     }
239     return active_requests_count_attrs
240 
241 
242 def _parse_duration_attrs(req_attrs):
243     duration_attrs = {
244         key: req_attrs[key]
245         for key in _duration_attrs.intersection(req_attrs.keys())
246     }
247     return duration_attrs
248 
249 
250 def _parse_url_query(url: str):
251     parsed_url = urlparse(url)
252     path = parsed_url.path
253     query_params = parsed_url.query
254     return path, query_params
255
```
--- 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/util/opentelemetry-util-http/src/opentelemetry/util/http/__init__.py b/util/opentelemetry-util-http/src/opentelemetry/util/http/__init__.py
--- a/util/opentelemetry-util-http/src/opentelemetry/util/http/__init__.py
+++ b/util/opentelemetry-util-http/src/opentelemetry/util/http/__init__.py
@@ -166,11 +166,7 @@
         parsed = urlparse(url)
         if all([parsed.scheme, parsed.netloc]):  # checks for valid url
             parsed_url = urlparse(url)
-            netloc = (
-                (":".join(((parsed_url.hostname or ""), str(parsed_url.port))))
-                if parsed_url.port
-                else (parsed_url.hostname or "")
-            )
+            _, _, netloc = parsed.netloc.rpartition("@")
             return urlunparse(
                 (
                     parsed_url.scheme,

{"golden_diff": "diff --git a/util/opentelemetry-util-http/src/opentelemetry/util/http/__init__.py b/util/opentelemetry-util-http/src/opentelemetry/util/http/__init__.py\n--- a/util/opentelemetry-util-http/src/opentelemetry/util/http/__init__.py\n+++ b/util/opentelemetry-util-http/src/opentelemetry/util/http/__init__.py\n@@ -166,11 +166,7 @@\n         parsed = urlparse(url)\n         if all([parsed.scheme, parsed.netloc]):  # checks for valid url\n             parsed_url = urlparse(url)\n-            netloc = (\n-                (\":\".join(((parsed_url.hostname or \"\"), str(parsed_url.port))))\n-                if parsed_url.port\n-                else (parsed_url.hostname or \"\")\n-            )\n+            _, _, netloc = parsed.netloc.rpartition(\"@\")\n             return urlunparse(\n                 (\n                     parsed_url.scheme,\n", "issue": "`remove_url_credentials` drops brackets from IPv6 hostnames\n**Describe your environment** \r\n\r\nPython 3.11.9, `opentelemetry-instrumentation` is auto-injected via the OpenTelemetry operator, `opentelemetry_util_http` is `0.44b0`\r\n\r\n**Steps to reproduce**\r\n```python\r\n\r\nfrom opentelemetry.util.http import remove_url_credentials\r\n\r\nliteral_ipv6_url = \"https://[::1]/somepath?query=foo\"\r\nremove_url_credentials(literal_ipv6_url)\r\n# 'https://::1/somepath?query=foo' -- should be 'https://[::1]/somepath?query=foo'\r\n\r\nliteral_ipv6_url_with_port = \"https://[::1]:12345/somepath?query=foo\"\r\nremove_url_credentials(literal_ipv6_url_with_port)\r\n# 'https://::1:12345/somepath?query=foo -- should be 'https://[::1]:12345/somepath?query=foo'\r\n\r\nliteral_ipv6_url_with_auth = \"https://someuser:somepass@[::1]:12345/somepath?query=foo\"\r\nremove_url_credentials(literal_ipv6_url_with_auth)\r\n# 'https://::1:12345/somepath?query=foo' -- should be https://[::1]:12345/somepath?query=foo\r\n\r\n```\r\n\r\n**What is the expected behavior?**\r\nThe ipv6 host should remain inside `[]` \r\n\r\n**What is the actual behavior?**\r\n`[]` are stripped from the host\r\n\r\n**Additional context**\r\nhttps://github.com/open-telemetry/opentelemetry-python-contrib/blob/main/util/opentelemetry-util-http/src/opentelemetry/util/http/__init__.py#L169 is the causing line. The `hostname` result on `urlparse` does not contain the brackets\r\n\r\n\r\n```python\r\nfrom urllib.parse import urlparse\r\n\r\nparsed = urlparse(literal_ipv6_url_with_auth)\r\n\r\nparsed\r\n# ParseResult(scheme='https', netloc='someuser:somepass@[::1]:12345', path='/somepath', params='', query='query=foo', fragment='')\r\n\r\nparsed.hostname\r\n# '::1'\r\n```\r\n\n", "before_files": [{"content": "# Copyright The OpenTelemetry Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom __future__ import annotations\n\nfrom os import environ\nfrom re import IGNORECASE as RE_IGNORECASE\nfrom re import compile as re_compile\nfrom re import search\nfrom typing import Callable, Iterable, Optional\nfrom urllib.parse import urlparse, urlunparse\n\nfrom opentelemetry.semconv.trace import SpanAttributes\n\nOTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SANITIZE_FIELDS = (\n    \"OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SANITIZE_FIELDS\"\n)\nOTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST = (\n    \"OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST\"\n)\nOTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE = (\n    \"OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE\"\n)\n\nOTEL_PYTHON_INSTRUMENTATION_HTTP_CAPTURE_ALL_METHODS = (\n    \"OTEL_PYTHON_INSTRUMENTATION_HTTP_CAPTURE_ALL_METHODS\"\n)\n\n# List of recommended metrics attributes\n_duration_attrs = {\n    SpanAttributes.HTTP_METHOD,\n    SpanAttributes.HTTP_HOST,\n    SpanAttributes.HTTP_SCHEME,\n    SpanAttributes.HTTP_STATUS_CODE,\n    SpanAttributes.HTTP_FLAVOR,\n    SpanAttributes.HTTP_SERVER_NAME,\n    SpanAttributes.NET_HOST_NAME,\n    SpanAttributes.NET_HOST_PORT,\n}\n\n_active_requests_count_attrs = {\n    SpanAttributes.HTTP_METHOD,\n    SpanAttributes.HTTP_HOST,\n    SpanAttributes.HTTP_SCHEME,\n    SpanAttributes.HTTP_FLAVOR,\n    SpanAttributes.HTTP_SERVER_NAME,\n}\n\n\nclass ExcludeList:\n    \"\"\"Class to exclude certain paths (given as a list of regexes) from tracing requests\"\"\"\n\n    def __init__(self, excluded_urls: Iterable[str]):\n        self._excluded_urls = excluded_urls\n        if self._excluded_urls:\n            self._regex = re_compile(\"|\".join(excluded_urls))\n\n    def url_disabled(self, url: str) -> bool:\n        return bool(self._excluded_urls and search(self._regex, url))\n\n\nclass SanitizeValue:\n    \"\"\"Class to sanitize (remove sensitive data from) certain headers (given as a list of regexes)\"\"\"\n\n    def __init__(self, sanitized_fields: Iterable[str]):\n        self._sanitized_fields = sanitized_fields\n        if self._sanitized_fields:\n            self._regex = re_compile(\"|\".join(sanitized_fields), RE_IGNORECASE)\n\n    def sanitize_header_value(self, header: str, value: str) -> str:\n        return (\n            \"[REDACTED]\"\n            if (self._sanitized_fields and search(self._regex, header))\n            else value\n        )\n\n    def sanitize_header_values(\n        self,\n        headers: dict[str, str],\n        header_regexes: list[str],\n        normalize_function: Callable[[str], str],\n    ) -> dict[str, str]:\n        values: dict[str, str] = {}\n\n        if header_regexes:\n            header_regexes_compiled = re_compile(\n                \"|\".join(\"^\" + i + \"$\" for i in header_regexes),\n                RE_IGNORECASE,\n            )\n\n            for header_name in list(\n                filter(\n                    header_regexes_compiled.match,\n                    headers.keys(),\n                )\n            ):\n                header_values = headers.get(header_name)\n                if header_values:\n                    key = normalize_function(header_name.lower())\n                    values[key] = [\n                        self.sanitize_header_value(\n                            header=header_name, value=header_values\n                        )\n                    ]\n\n        return values\n\n\n_root = r\"OTEL_PYTHON_{}\"\n\n\ndef get_traced_request_attrs(instrumentation):\n    traced_request_attrs = environ.get(\n        _root.format(f\"{instrumentation}_TRACED_REQUEST_ATTRS\"), []\n    )\n\n    if traced_request_attrs:\n        traced_request_attrs = [\n            traced_request_attr.strip()\n            for traced_request_attr in traced_request_attrs.split(\",\")\n        ]\n\n    return traced_request_attrs\n\n\ndef get_excluded_urls(instrumentation: str) -> ExcludeList:\n    # Get instrumentation-specific excluded URLs. If not set, retrieve them\n    # from generic variable.\n    excluded_urls = environ.get(\n        _root.format(f\"{instrumentation}_EXCLUDED_URLS\"),\n        environ.get(_root.format(\"EXCLUDED_URLS\"), \"\"),\n    )\n\n    return parse_excluded_urls(excluded_urls)\n\n\ndef parse_excluded_urls(excluded_urls: str) -> ExcludeList:\n    \"\"\"\n    Small helper to put an arbitrary url list inside an ExcludeList\n    \"\"\"\n    if excluded_urls:\n        excluded_url_list = [\n            excluded_url.strip() for excluded_url in excluded_urls.split(\",\")\n        ]\n    else:\n        excluded_url_list = []\n\n    return ExcludeList(excluded_url_list)\n\n\ndef remove_url_credentials(url: str) -> str:\n    \"\"\"Given a string url, remove the username and password only if it is a valid url\"\"\"\n\n    try:\n        parsed = urlparse(url)\n        if all([parsed.scheme, parsed.netloc]):  # checks for valid url\n            parsed_url = urlparse(url)\n            netloc = (\n                (\":\".join(((parsed_url.hostname or \"\"), str(parsed_url.port))))\n                if parsed_url.port\n                else (parsed_url.hostname or \"\")\n            )\n            return urlunparse(\n                (\n                    parsed_url.scheme,\n                    netloc,\n                    parsed_url.path,\n                    parsed_url.params,\n                    parsed_url.query,\n                    parsed_url.fragment,\n                )\n            )\n    except ValueError:  # an unparsable url was passed\n        pass\n    return url\n\n\ndef normalise_request_header_name(header: str) -> str:\n    key = header.lower().replace(\"-\", \"_\")\n    return f\"http.request.header.{key}\"\n\n\ndef normalise_response_header_name(header: str) -> str:\n    key = header.lower().replace(\"-\", \"_\")\n    return f\"http.response.header.{key}\"\n\n\ndef sanitize_method(method: Optional[str]) -> Optional[str]:\n    if method is None:\n        return None\n    method = method.upper()\n    if (\n        environ.get(OTEL_PYTHON_INSTRUMENTATION_HTTP_CAPTURE_ALL_METHODS)\n        or\n        # Based on https://www.rfc-editor.org/rfc/rfc7231#section-4.1 and https://www.rfc-editor.org/rfc/rfc5789#section-2.\n        method\n        in [\n            \"GET\",\n            \"HEAD\",\n            \"POST\",\n            \"PUT\",\n            \"DELETE\",\n            \"CONNECT\",\n            \"OPTIONS\",\n            \"TRACE\",\n            \"PATCH\",\n        ]\n    ):\n        return method\n    return \"_OTHER\"\n\n\ndef get_custom_headers(env_var: str) -> list[str]:\n    custom_headers = environ.get(env_var, None)\n    if custom_headers:\n        return [\n            custom_headers.strip()\n            for custom_headers in custom_headers.split(\",\")\n        ]\n    return []\n\n\ndef _parse_active_request_count_attrs(req_attrs):\n    active_requests_count_attrs = {\n        key: req_attrs[key]\n        for key in _active_requests_count_attrs.intersection(req_attrs.keys())\n    }\n    return active_requests_count_attrs\n\n\ndef _parse_duration_attrs(req_attrs):\n    duration_attrs = {\n        key: req_attrs[key]\n        for key in _duration_attrs.intersection(req_attrs.keys())\n    }\n    return duration_attrs\n\n\ndef _parse_url_query(url: str):\n    parsed_url = urlparse(url)\n    path = parsed_url.path\n    query_params = parsed_url.query\n    return path, query_params\n", "path": "util/opentelemetry-util-http/src/opentelemetry/util/http/__init__.py"}], "after_files": [{"content": "# Copyright The OpenTelemetry Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom __future__ import annotations\n\nfrom os import environ\nfrom re import IGNORECASE as RE_IGNORECASE\nfrom re import compile as re_compile\nfrom re import search\nfrom typing import Callable, Iterable, Optional\nfrom urllib.parse import urlparse, urlunparse\n\nfrom opentelemetry.semconv.trace import SpanAttributes\n\nOTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SANITIZE_FIELDS = (\n    \"OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SANITIZE_FIELDS\"\n)\nOTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST = (\n    \"OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST\"\n)\nOTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE = (\n    \"OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE\"\n)\n\nOTEL_PYTHON_INSTRUMENTATION_HTTP_CAPTURE_ALL_METHODS = (\n    \"OTEL_PYTHON_INSTRUMENTATION_HTTP_CAPTURE_ALL_METHODS\"\n)\n\n# List of recommended metrics attributes\n_duration_attrs = {\n    SpanAttributes.HTTP_METHOD,\n    SpanAttributes.HTTP_HOST,\n    SpanAttributes.HTTP_SCHEME,\n    SpanAttributes.HTTP_STATUS_CODE,\n    SpanAttributes.HTTP_FLAVOR,\n    SpanAttributes.HTTP_SERVER_NAME,\n    SpanAttributes.NET_HOST_NAME,\n    SpanAttributes.NET_HOST_PORT,\n}\n\n_active_requests_count_attrs = {\n    SpanAttributes.HTTP_METHOD,\n    SpanAttributes.HTTP_HOST,\n    SpanAttributes.HTTP_SCHEME,\n    SpanAttributes.HTTP_FLAVOR,\n    SpanAttributes.HTTP_SERVER_NAME,\n}\n\n\nclass ExcludeList:\n    \"\"\"Class to exclude certain paths (given as a list of regexes) from tracing requests\"\"\"\n\n    def __init__(self, excluded_urls: Iterable[str]):\n        self._excluded_urls = excluded_urls\n        if self._excluded_urls:\n            self._regex = re_compile(\"|\".join(excluded_urls))\n\n    def url_disabled(self, url: str) -> bool:\n        return bool(self._excluded_urls and search(self._regex, url))\n\n\nclass SanitizeValue:\n    \"\"\"Class to sanitize (remove sensitive data from) certain headers (given as a list of regexes)\"\"\"\n\n    def __init__(self, sanitized_fields: Iterable[str]):\n        self._sanitized_fields = sanitized_fields\n        if self._sanitized_fields:\n            self._regex = re_compile(\"|\".join(sanitized_fields), RE_IGNORECASE)\n\n    def sanitize_header_value(self, header: str, value: str) -> str:\n        return (\n            \"[REDACTED]\"\n            if (self._sanitized_fields and search(self._regex, header))\n            else value\n        )\n\n    def sanitize_header_values(\n        self,\n        headers: dict[str, str],\n        header_regexes: list[str],\n        normalize_function: Callable[[str], str],\n    ) -> dict[str, str]:\n        values: dict[str, str] = {}\n\n        if header_regexes:\n            header_regexes_compiled = re_compile(\n                \"|\".join(\"^\" + i + \"$\" for i in header_regexes),\n                RE_IGNORECASE,\n            )\n\n            for header_name in list(\n                filter(\n                    header_regexes_compiled.match,\n                    headers.keys(),\n                )\n            ):\n                header_values = headers.get(header_name)\n                if header_values:\n                    key = normalize_function(header_name.lower())\n                    values[key] = [\n                        self.sanitize_header_value(\n                            header=header_name, value=header_values\n                        )\n                    ]\n\n        return values\n\n\n_root = r\"OTEL_PYTHON_{}\"\n\n\ndef get_traced_request_attrs(instrumentation):\n    traced_request_attrs = environ.get(\n        _root.format(f\"{instrumentation}_TRACED_REQUEST_ATTRS\"), []\n    )\n\n    if traced_request_attrs:\n        traced_request_attrs = [\n            traced_request_attr.strip()\n            for traced_request_attr in traced_request_attrs.split(\",\")\n        ]\n\n    return traced_request_attrs\n\n\ndef get_excluded_urls(instrumentation: str) -> ExcludeList:\n    # Get instrumentation-specific excluded URLs. If not set, retrieve them\n    # from generic variable.\n    excluded_urls = environ.get(\n        _root.format(f\"{instrumentation}_EXCLUDED_URLS\"),\n        environ.get(_root.format(\"EXCLUDED_URLS\"), \"\"),\n    )\n\n    return parse_excluded_urls(excluded_urls)\n\n\ndef parse_excluded_urls(excluded_urls: str) -> ExcludeList:\n    \"\"\"\n    Small helper to put an arbitrary url list inside an ExcludeList\n    \"\"\"\n    if excluded_urls:\n        excluded_url_list = [\n            excluded_url.strip() for excluded_url in excluded_urls.split(\",\")\n        ]\n    else:\n        excluded_url_list = []\n\n    return ExcludeList(excluded_url_list)\n\n\ndef remove_url_credentials(url: str) -> str:\n    \"\"\"Given a string url, remove the username and password only if it is a valid url\"\"\"\n\n    try:\n        parsed = urlparse(url)\n        if all([parsed.scheme, parsed.netloc]):  # checks for valid url\n            parsed_url = urlparse(url)\n            _, _, netloc = parsed.netloc.rpartition(\"@\")\n            return urlunparse(\n                (\n                    parsed_url.scheme,\n                    netloc,\n                    parsed_url.path,\n                    parsed_url.params,\n                    parsed_url.query,\n                    parsed_url.fragment,\n                )\n            )\n    except ValueError:  # an unparsable url was passed\n        pass\n    return url\n\n\ndef normalise_request_header_name(header: str) -> str:\n    key = header.lower().replace(\"-\", \"_\")\n    return f\"http.request.header.{key}\"\n\n\ndef normalise_response_header_name(header: str) -> str:\n    key = header.lower().replace(\"-\", \"_\")\n    return f\"http.response.header.{key}\"\n\n\ndef sanitize_method(method: Optional[str]) -> Optional[str]:\n    if method is None:\n        return None\n    method = method.upper()\n    if (\n        environ.get(OTEL_PYTHON_INSTRUMENTATION_HTTP_CAPTURE_ALL_METHODS)\n        or\n        # Based on https://www.rfc-editor.org/rfc/rfc7231#section-4.1 and https://www.rfc-editor.org/rfc/rfc5789#section-2.\n        method\n        in [\n            \"GET\",\n            \"HEAD\",\n            \"POST\",\n            \"PUT\",\n            \"DELETE\",\n            \"CONNECT\",\n            \"OPTIONS\",\n            \"TRACE\",\n            \"PATCH\",\n        ]\n    ):\n        return method\n    return \"_OTHER\"\n\n\ndef get_custom_headers(env_var: str) -> list[str]:\n    custom_headers = environ.get(env_var, None)\n    if custom_headers:\n        return [\n            custom_headers.strip()\n            for custom_headers in custom_headers.split(\",\")\n        ]\n    return []\n\n\ndef _parse_active_request_count_attrs(req_attrs):\n    active_requests_count_attrs = {\n        key: req_attrs[key]\n        for key in _active_requests_count_attrs.intersection(req_attrs.keys())\n    }\n    return active_requests_count_attrs\n\n\ndef _parse_duration_attrs(req_attrs):\n    duration_attrs = {\n        key: req_attrs[key]\n        for key in _duration_attrs.intersection(req_attrs.keys())\n    }\n    return duration_attrs\n\n\ndef _parse_url_query(url: str):\n    parsed_url = urlparse(url)\n    path = parsed_url.path\n    query_params = parsed_url.query\n    return path, query_params\n", "path": "util/opentelemetry-util-http/src/opentelemetry/util/http/__init__.py"}]}