chore(asm): add support for headers always sent when ASM enabled (#8916)

Makes 3 request headers always to be sent if ASM is enabled:

- accept
- content-type
- user-agent


APPSEC-52404

## 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] Risks are described (performance impact, potential for breakage,
maintainability)
- [x] Change is maintainable (easy to change, telemetry, documentation)
- [x] [Library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
are followed or label `changelog/no-changelog` is set
- [x] Documentation is included (in-code, generated user docs, [public
corp docs](https://github.com/DataDog/documentation/))
- [x] Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))
- [x] If this PR changes the public interface, I've notified
`@DataDog/apm-tees`.

## Reviewer Checklist

- [x] Title is accurate
- [x] All changes are related to the pull request's stated goal
- [x] Description motivates each change
- [x] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- [x] Testing strategy adequately addresses listed risks
- [x] Change is maintainable (easy to change, telemetry, documentation)
- [x] Release note makes sense to a user of the library
- [x] Author has acknowledged and discussed the performance implications
of this PR as reported in the benchmarks PR comment
- [x] Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

ddtrace/appsec/_processor.py

@@ -101,16 +101,22 @@ def get_appsec_obfuscation_parameter_value_regexp() -> bytes:
     "x-real-ip",
 }
 
+_COLLECTED_REQUEST_HEADERS_ASM_ENABLED = {
+    "accept",
+    "content-type",
+    "user-agent",
+}
+
 
-def _set_headers(span: Span, headers: Any, kind: str) -> None:
+def _set_headers(span: Span, headers: Any, kind: str, only_asm_enabled: bool = False) -> None:
     from ddtrace.contrib.trace_utils import _normalize_tag_name
 
     for k in headers:
         if isinstance(k, tuple):
             key, value = k
         else:
             key, value = k, headers[k]
-        if key.lower() in _COLLECTED_REQUEST_HEADERS:
+        if key.lower() in (_COLLECTED_REQUEST_HEADERS_ASM_ENABLED if only_asm_enabled else _COLLECTED_REQUEST_HEADERS):
             # since the header value can be a list, use `set_tag()` to ensure it is converted to a string
             span.set_tag(_normalize_tag_name(kind, key), value)
 
@@ -411,9 +417,13 @@ def on_span_finish(self, span: Span) -> None:
         try:
             if span.span_type == SpanTypes.WEB:
                 # Force to set respond headers at the end
-                headers_req = core.get_item(SPAN_DATA_NAMES.RESPONSE_HEADERS_NO_COOKIES, span=span)
+                headers_res = core.get_item(SPAN_DATA_NAMES.RESPONSE_HEADERS_NO_COOKIES, span=span)
+                if headers_res:
+                    _set_headers(span, headers_res, kind="response")
+
+                headers_req = core.get_item(SPAN_DATA_NAMES.REQUEST_HEADERS_NO_COOKIES, span=span)
                 if headers_req:
-                    _set_headers(span, headers_req, kind="response")
+                    _set_headers(span, headers_req, kind="request", only_asm_enabled=True)
 
                 # this call is only necessary for tests or frameworks that are not using blocking
                 if not has_triggers(span) and _asm_request_context.in_context():

tests/appsec/contrib_appsec/utils.py

@@ -1049,6 +1049,25 @@ def test_multiple_service_name(self, interface):
             else:
                 raise AssertionError("extra service not found")
 
+    @pytest.mark.parametrize("asm_enabled", [True, False])
+    def test_asm_enabled_headers(self, asm_enabled, interface, get_tag, root_span):
+        with override_global_config(dict(_asm_enabled=asm_enabled)):
+            self.update_tracer(interface)
+            response = interface.client.get(
+                "/",
+                headers={"accept": "testheaders/a1b2c3", "user-agent": "UnitTestAgent", "content-type": "test/x0y9z8"},
+            )
+            assert response.status_code == 200
+            assert self.status(response) == 200
+            if asm_enabled:
+                assert get_tag("http.request.headers.accept") == "testheaders/a1b2c3"
+                assert get_tag("http.request.headers.user-agent") == "UnitTestAgent"
+                assert get_tag("http.request.headers.content-type") == "test/x0y9z8"
+            else:
+                assert get_tag("http.request.headers.accept") is None
+                assert get_tag("http.request.headers.user-agent") is None
+                assert get_tag("http.request.headers.content-type") is None
+
     def test_global_callback_list_length(self, interface):
         from ddtrace.appsec import _asm_request_context
 

tests/snapshots/tests.contrib.django.test_django_appsec_snapshots.test_appsec_enabled.json

@@ -23,6 +23,8 @@
       "django.view": "tests.contrib.django.views.index",
       "http.client_ip": "127.0.0.1",
       "http.method": "GET",
+      "http.request.headers.accept": "*/*",
+      "http.request.headers.user-agent": "python-requests/2.31.0",
       "http.response.headers.content-length": "16",
       "http.response.headers.content-type": "text/html; charset=utf-8",
       "http.route": "^$",

tests/snapshots/tests.contrib.django.test_django_appsec_snapshots.test_request_ipblock_nomatch_200.json

@@ -23,6 +23,8 @@
       "django.view": "tests.contrib.django.views.index",
       "http.client_ip": "1.1.1.1",
       "http.method": "GET",
+      "http.request.headers.accept": "*/*",
+      "http.request.headers.user-agent": "python-requests/2.31.0",
       "http.response.headers.content-length": "16",
       "http.response.headers.content-type": "text/html; charset=utf-8",
       "http.route": "^$",