diff --git a/sentry_sdk/data_collection.py b/sentry_sdk/data_collection.py
index 4908b82e9a..bcdf767409 100644
--- a/sentry_sdk/data_collection.py
+++ b/sentry_sdk/data_collection.py
@@ -23,7 +23,9 @@
 """
 
 import warnings
-from typing import TYPE_CHECKING, cast
+from typing import TYPE_CHECKING, List, Mapping, Optional, cast
+
+from sentry_sdk._types import SENSITIVE_DATA_SUBSTITUTE
 
 if TYPE_CHECKING:
     from typing import Any, Dict, Literal
@@ -77,6 +79,62 @@
 ]
 
 
+def _is_sensitive_key(key: str, extra_terms: "Optional[List[str]]" = None) -> bool:
+    """
+    Return whether ``key`` matches the sensitive denylist using a partial,
+    case-insensitive substring match.
+
+    :param extra_terms: additional deny terms (e.g. user-provided) to consider
+        alongside the built-in `_SENSITIVE_DENYLIST`.
+    """
+    lowered = key.lower()
+    for term in _SENSITIVE_DENYLIST:
+        if term in lowered:
+            return True
+    if extra_terms:
+        for term in extra_terms:
+            if term and term.lower() in lowered:
+                return True
+    return False
+
+
+def _apply_key_value_collection_filtering(
+    items: "Mapping[str, Any]",
+    behaviour: "KeyValueCollectionBehaviour",
+    substitute: "Any" = SENSITIVE_DATA_SUBSTITUTE,
+) -> "Dict[str, Any]":
+
+    if behaviour["mode"] == "off":
+        return {}
+
+    result: "Dict[str, Any]" = {}
+
+    if behaviour["mode"] == "allowlist":
+        for key, value in items.items():
+            is_allowed = False
+            if isinstance(key, str):
+                lowered = key.lower()
+                is_allowed = any(
+                    term and term.lower() in lowered
+                    for term in behaviour.get("terms", [])
+                )
+            if is_allowed and not _is_sensitive_key(key):
+                result[key] = value
+            else:
+                result[key] = substitute
+        return result
+
+    # denylist behaviour
+    for key, value in items.items():
+        if isinstance(key, str) and _is_sensitive_key(
+            key=key, extra_terms=behaviour.get("terms", [])
+        ):
+            result[key] = substitute
+        else:
+            result[key] = value
+    return result
+
+
 def _map_from_send_default_pii(
     *,
     send_default_pii: bool,
diff --git a/sentry_sdk/integrations/_asgi_common.py b/sentry_sdk/integrations/_asgi_common.py
index 85f8f11d6d..7ff4657013 100644
--- a/sentry_sdk/integrations/_asgi_common.py
+++ b/sentry_sdk/integrations/_asgi_common.py
@@ -115,9 +115,13 @@ def _get_request_data(
     if ty in ("http", "websocket"):
         request_data["method"] = asgi_scope.get("method")
 
-        request_data["headers"] = headers = _filter_headers(
-            _get_headers(asgi_scope),
+        headers = _get_headers(asgi_scope)
+
+        request_data["headers"] = _filter_headers(
+            headers,
+            use_annotated_value=False,
         )
+
         request_data["query_string"] = _get_query(asgi_scope)
 
         request_data["url"] = _get_url(
@@ -148,8 +152,10 @@ def _get_request_attributes(
         if asgi_scope.get("method"):
             attributes["http.request.method"] = asgi_scope["method"].upper()
 
-        headers = _filter_headers(_get_headers(asgi_scope), use_annotated_value=False)
-        for header, value in headers.items():
+        headers = _get_headers(asgi_scope)
+
+        filtered_headers = _filter_headers(headers, use_annotated_value=False)
+        for header, value in filtered_headers.items():
             attributes[f"http.request.header.{header.lower()}"] = value
 
         if should_send_default_pii():
diff --git a/sentry_sdk/integrations/_wsgi_common.py b/sentry_sdk/integrations/_wsgi_common.py
index cf1a365209..ad0ab7c734 100644
--- a/sentry_sdk/integrations/_wsgi_common.py
+++ b/sentry_sdk/integrations/_wsgi_common.py
@@ -4,8 +4,9 @@
 
 import sentry_sdk
 from sentry_sdk._types import SENSITIVE_DATA_SUBSTITUTE
+from sentry_sdk.data_collection import _apply_key_value_collection_filtering
 from sentry_sdk.scope import should_send_default_pii
-from sentry_sdk.utils import AnnotatedValue, logger
+from sentry_sdk.utils import AnnotatedValue, has_data_collection_enabled, logger
 
 try:
     from django.http.request import RawPostDataException
@@ -213,19 +214,39 @@ def _filter_headers(
     headers: "Mapping[str, str]",
     use_annotated_value: bool = True,
 ) -> "Mapping[str, Union[AnnotatedValue, str]]":
-    if should_send_default_pii():
-        return headers
+    client_options = sentry_sdk.get_client().options
 
-    substitute: "Union[AnnotatedValue, str]" = (
-        SENSITIVE_DATA_SUBSTITUTE
-        if not use_annotated_value
-        else AnnotatedValue.removed_because_over_size_limit()
-    )
+    if has_data_collection_enabled(client_options):
+        data_collection_configuration = client_options["data_collection"]
+
+        filtered = _apply_key_value_collection_filtering(
+            items=headers,
+            behaviour=data_collection_configuration["http_headers"]["request"],
+        )
+
+        for key in filtered:
+            if isinstance(key, str) and key.lower() in ("cookie", "set-cookie"):
+                filtered[key] = SENSITIVE_DATA_SUBSTITUTE
 
-    return {
-        k: (v if k.upper().replace("-", "_") not in SENSITIVE_HEADERS else substitute)
-        for k, v in headers.items()
-    }
+        return filtered
+    else:
+        if should_send_default_pii():
+            return headers
+
+        substitute: "Union[AnnotatedValue, str]" = (
+            SENSITIVE_DATA_SUBSTITUTE
+            if not use_annotated_value
+            else AnnotatedValue.removed_because_over_size_limit()
+        )
+
+        return {
+            k: (
+                v
+                if k.upper().replace("-", "_") not in SENSITIVE_HEADERS
+                else substitute
+            )
+            for k, v in headers.items()
+        }
 
 
 def _in_http_status_code_range(
diff --git a/sentry_sdk/integrations/aiohttp.py b/sentry_sdk/integrations/aiohttp.py
index d22f3a745b..59bae92a60 100644
--- a/sentry_sdk/integrations/aiohttp.py
+++ b/sentry_sdk/integrations/aiohttp.py
@@ -150,7 +150,8 @@ async def sentry_app_handle(
 
                         header_attributes: "dict[str, Any]" = {}
                         for header, header_value in _filter_headers(
-                            headers, use_annotated_value=False
+                            headers,
+                            use_annotated_value=False,
                         ).items():
                             header_attributes[
                                 f"http.request.header.{header.lower()}"
diff --git a/tests/integrations/aiohttp/test_aiohttp.py b/tests/integrations/aiohttp/test_aiohttp.py
index 583e7a50b6..dc485954ec 100644
--- a/tests/integrations/aiohttp/test_aiohttp.py
+++ b/tests/integrations/aiohttp/test_aiohttp.py
@@ -1258,8 +1258,185 @@ async def hello(request):
     )
 
 
+@pytest.mark.parametrize(
+    "options,expected",
+    [
+        pytest.param(
+            {
+                "send_default_pii": True,
+                "data_collection": None,
+            },
+            {
+                "authorization": "[Filtered]",
+                "custom": "foobar",
+                "cookie": "[Filtered]",
+            },
+            id="enabled_send_default_pii_redacts_auth_header_due_to_data_collection_default_settings",
+        ),
+        pytest.param(
+            {
+                "send_default_pii": False,
+                "data_collection": None,
+            },
+            {
+                "authorization": "[Filtered]",
+                "custom": "foobar",
+                "cookie": "[Filtered]",
+            },
+            id="disabled_send_default_pii_redacts_auth_header_due_to_data_collection_default_settings",
+        ),
+        pytest.param(
+            {
+                "send_default_pii": False,
+                "data_collection": {"http_headers": {"request": {"mode": "off"}}},
+            },
+            None,
+            id="data_collection_off_does_not_add_headers",
+        ),
+        pytest.param(
+            {
+                "send_default_pii": False,
+                "data_collection": {"http_headers": {"request": {"mode": "allowlist"}}},
+            },
+            {
+                "authorization": "[Filtered]",
+                "custom": "[Filtered]",
+                "cookie": "[Filtered]",
+            },
+            id="data_collection_allow_list_redacts_terms_that_do_not_appear",
+        ),
+        pytest.param(
+            {
+                "send_default_pii": False,
+                "data_collection": {
+                    "http_headers": {
+                        "request": {"mode": "allowlist", "terms": ["Authorization"]}
+                    }
+                },
+            },
+            {
+                "authorization": "[Filtered]",
+                "custom": "[Filtered]",
+                "cookie": "[Filtered]",
+            },
+            id="data_collection_allow_list_redacts_sensitive_terms_even_when_provided_by_user",
+        ),
+        pytest.param(
+            {
+                "send_default_pii": False,
+                "data_collection": {
+                    "http_headers": {
+                        "request": {"mode": "allowlist", "terms": ["custom"]}
+                    }
+                },
+            },
+            {
+                "authorization": "[Filtered]",
+                "custom": "foobar",
+                "cookie": "[Filtered]",
+            },
+            id="data_collection_allow_list_does_not_redact_provided_term",
+        ),
+        pytest.param(
+            {
+                "send_default_pii": False,
+                "data_collection": {
+                    "http_headers": {
+                        "request": {"mode": "denylist", "terms": ["custom"]}
+                    }
+                },
+            },
+            {
+                "authorization": "[Filtered]",
+                "custom": "[Filtered]",
+                "cookie": "[Filtered]",
+            },
+            id="data_collection_deny_list_redacts_sensitive_terms_when_provided_by_user",
+        ),
+        pytest.param(
+            {
+                "send_default_pii": False,
+                "data_collection": {
+                    "http_headers": {
+                        "request": {"mode": "allowlist", "terms": ["cookie"]}
+                    }
+                },
+            },
+            {
+                "authorization": "[Filtered]",
+                "custom": "[Filtered]",
+                "cookie": "[Filtered]",
+            },
+            id="data_collection_cookie_is_always_redacted_even_when_allow_listed",
+        ),
+    ],
+)
 @pytest.mark.asyncio
 async def test_sensitive_header_passthrough_with_pii_span_streaming(
+    sentry_init, aiohttp_client, capture_items, options, expected, request
+):
+    sentry_init(
+        integrations=[AioHttpIntegration()],
+        traces_sample_rate=1.0,
+        send_default_pii=options["send_default_pii"],
+        _experiments={
+            "trace_lifecycle": "stream",
+            "data_collection": options["data_collection"],
+        },
+    )
+
+    async def hello(request):
+        return web.Response(text="hello")
+
+    app = web.Application()
+    app.router.add_get("/", hello)
+
+    items = capture_items("span")
+
+    client = await aiohttp_client(app)
+    await client.get(
+        "/",
+        headers={
+            "Authorization": "Bearer secret-token",
+            "x-custom-header": "foobar",
+            "Cookie": "sessionid=secret",
+        },
+    )
+
+    sentry_sdk.flush()
+
+    server_span, _client_segment = [item.payload for item in items]
+
+    if request.node.callspec.id.endswith("data_collection_off_does_not_add_headers"):
+        assert "http.request.header.authorization" not in server_span["attributes"]
+        assert "http.request.header.cookie" not in server_span["attributes"]
+    else:
+        assert (
+            server_span["attributes"]["http.request.header.authorization"]
+            == expected["authorization"]
+        )
+        assert (
+            server_span["attributes"]["http.request.header.x-custom-header"]
+            == expected["custom"]
+        )
+        assert (
+            server_span["attributes"]["http.request.header.cookie"]
+            == expected["cookie"]
+        )
+
+    # client.address and user.ip_address is captured under send_default_pii=True.
+    # TODO: This block will eventually need to be removed from this test into a separate
+    # test once data collection gating is introduced on these values
+    if options["send_default_pii"]:
+        assert server_span["attributes"]["client.address"] == "127.0.0.1"
+        assert server_span["attributes"]["user.ip_address"] == "127.0.0.1"
+    else:
+        assert "user.ip_address" not in server_span["attributes"]
+        assert "client.address" not in server_span["attributes"]
+
+
+@pytest.mark.asyncio
+async def test_sensitive_header_passthrough_with_pii_span_streaming_without_data_collection(
     sentry_init, aiohttp_client, capture_items
 ):
     sentry_init(
diff --git a/tests/integrations/asgi/test_asgi.py b/tests/integrations/asgi/test_asgi.py
index 3bce0d1e10..e72cd317c2 100644
--- a/tests/integrations/asgi/test_asgi.py
+++ b/tests/integrations/asgi/test_asgi.py
@@ -5,7 +5,13 @@
 
 import sentry_sdk
 from sentry_sdk import capture_message
-from sentry_sdk.integrations._asgi_common import _get_headers, _get_ip
+from sentry_sdk.integrations._asgi_common import (
+    _get_headers,
+    _get_ip,
+    _get_request_attributes,
+    _get_request_data,
+    _RootPathInPath,
+)
 from sentry_sdk.integrations.asgi import SentryAsgiMiddleware, _looks_like_asgi3
 from sentry_sdk.tracing import TransactionSource
 
@@ -831,6 +837,87 @@ def test_get_headers():
     }
 
 
+def test_get_request_data_url_with_filtered_host(sentry_init):
+    # allowlist mode in data collection that does not allow "host" scrubs the host header value,
+    # but the reported URL must still resolve via rather than embedding the substituted "[Filtered]" value.
+    sentry_init(
+        _experiments={
+            "data_collection": {
+                "http_headers": {"request": {"mode": "allowlist", "terms": []}}
+            }
+        }
+    )
+
+    scope = {
+        "type": "http",
+        "method": "GET",
+        "scheme": "http",
+        "server": ("example.com", 80),
+        "path": "/foo",
+        "query_string": b"",
+        "headers": [(b"host", b"example.com")],
+    }
+
+    request_data = _get_request_data(scope, _RootPathInPath.EXCLUDED)
+
+    assert request_data["headers"]["host"] == "[Filtered]"
+    assert request_data["url"] == "http://example.com/foo"
+
+
+def test_get_request_attributes_url_with_filtered_host(sentry_init):
+    # As with _get_request_data, an allowlist mode that does not allow "host"
+    # scrubs the host header value, but "url.full" must still resolve rather than embedding the substituted value.
+    sentry_init(
+        send_default_pii=True,
+        _experiments={
+            "data_collection": {
+                "http_headers": {"request": {"mode": "allowlist", "terms": []}}
+            }
+        },
+    )
+
+    scope = {
+        "type": "http",
+        "method": "GET",
+        "scheme": "http",
+        "server": ("example.com", 80),
+        "path": "/foo",
+        "query_string": b"somevalue=123",
+        "headers": [(b"host", b"example.com")],
+    }
+
+    attributes = _get_request_attributes(scope, _RootPathInPath.EXCLUDED)
+
+    assert attributes["http.request.header.host"] == "[Filtered]"
+    assert attributes["url.full"] == "http://example.com/foo?somevalue=123"
+
+
+def test_get_request_attributes_url_with_headers_off(sentry_init):
+    # "off" mode in data collection captures no headers at all, but "url.full"
+    # must still resolve via the (uncaptured) host header rather than being dropped.
+    sentry_init(
+        send_default_pii=True,
+        _experiments={
+            "data_collection": {"http_headers": {"request": {"mode": "off"}}}
+        },
+    )
+
+    scope = {
+        "type": "http",
+        "method": "GET",
+        "scheme": "http",
+        "server": ("example.com", 80),
+        "path": "/foo",
+        "query_string": b"somevalue=123",
+        "headers": [(b"host", b"example.com")],
+    }
+
+    attributes = _get_request_attributes(scope, _RootPathInPath.EXCLUDED)
+
+    assert not any(key.startswith("http.request.header.") for key in attributes)
+    assert attributes["url.full"] == "http://example.com/foo?somevalue=123"
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "request_url,transaction_style,expected_transaction_name,expected_transaction_source",
diff --git a/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionAllowlist/.gitignore b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionAllowlist/.gitignore
new file mode 100644
index 0000000000..1c56884372
--- /dev/null
+++ b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionAllowlist/.gitignore
@@ -0,0 +1,11 @@
+# Need to add some ignore rules in this directory, because the unit tests will add the Sentry SDK and its dependencies
+# into this directory to create a Lambda function package that contains everything needed to instrument a Lambda function using Sentry.
+
+# Ignore everything
+*
+
+# But not index.py
+!index.py
+
+# And not .gitignore itself
+!.gitignore
diff --git a/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionAllowlist/.lock b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionAllowlist/.lock
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionAllowlist/index.py b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionAllowlist/index.py
new file mode 100644
index 0000000000..a79e1a03a7
--- /dev/null
+++ b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionAllowlist/index.py
@@ -0,0 +1,27 @@
+import os
+
+import sentry_sdk
+from sentry_sdk.integrations.aws_lambda import AwsLambdaIntegration
+
+sentry_sdk.init(
+    dsn=os.environ.get("SENTRY_DSN"),
+    traces_sample_rate=1.0,
+    integrations=[AwsLambdaIntegration()],
+    _experiments={
+        "data_collection": {
+            "http_headers": {
+                "request": {
+                    "mode": "allowlist",
+                    # "authorization" is allowlisted on purpose to show that an
+                    # allowlist entry cannot override the built-in sensitive
+                    # denylist.
+                    "terms": ["user-agent", "x-allow-me", "authorization"],
+                }
+            }
+        }
+    },
+)
+
+
+def handler(event, context):
+    return {"event": event}
diff --git a/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionDenylist/.gitignore b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionDenylist/.gitignore
new file mode 100644
index 0000000000..1c56884372
--- /dev/null
+++ b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionDenylist/.gitignore
@@ -0,0 +1,11 @@
+# Need to add some ignore rules in this directory, because the unit tests will add the Sentry SDK and its dependencies
+# into this directory to create a Lambda function package that contains everything needed to instrument a Lambda function using Sentry.
+
+# Ignore everything
+*
+
+# But not index.py
+!index.py
+
+# And not .gitignore itself
+!.gitignore
diff --git a/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionDenylist/.lock b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionDenylist/.lock
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionDenylist/index.py b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionDenylist/index.py
new file mode 100644
index 0000000000..4c59de9cb4
--- /dev/null
+++ b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionDenylist/index.py
@@ -0,0 +1,26 @@
+import os
+
+import sentry_sdk
+from sentry_sdk.integrations.aws_lambda import AwsLambdaIntegration
+
+sentry_sdk.init(
+    dsn=os.environ.get("SENTRY_DSN"),
+    traces_sample_rate=1.0,
+    integrations=[AwsLambdaIntegration()],
+    _experiments={
+        "data_collection": {
+            "http_headers": {
+                "request": {
+                    "mode": "denylist",
+                    # Custom terms deny otherwise non-sensitive headers on top
+                    # of the built-in sensitive denylist.
+                    "terms": ["x-forwarded", "user-agent"],
+                }
+            }
+        }
+    },
+)
+
+
+def handler(event, context):
+    return {"event": event}
diff --git a/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionOff/.gitignore b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionOff/.gitignore
new file mode 100644
index 0000000000..1c56884372
--- /dev/null
+++ b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionOff/.gitignore
@@ -0,0 +1,11 @@
+# Need to add some ignore rules in this directory, because the unit tests will add the Sentry SDK and its dependencies
+# into this directory to create a Lambda function package that contains everything needed to instrument a Lambda function using Sentry.
+
+# Ignore everything
+*
+
+# But not index.py
+!index.py
+
+# And not .gitignore itself
+!.gitignore
diff --git a/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionOff/.lock b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionOff/.lock
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionOff/index.py b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionOff/index.py
new file mode 100644
index 0000000000..6f3a7a8126
--- /dev/null
+++ b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkDataCollectionOff/index.py
@@ -0,0 +1,21 @@
+import os
+
+import sentry_sdk
+from sentry_sdk.integrations.aws_lambda import AwsLambdaIntegration
+
+sentry_sdk.init(
+    dsn=os.environ.get("SENTRY_DSN"),
+    traces_sample_rate=1.0,
+    integrations=[AwsLambdaIntegration()],
+    _experiments={
+        "data_collection": {
+            "http_headers": {
+                "request": {"mode": "off"},
+            }
+        }
+    },
+)
+
+
+def handler(event, context):
+    return {"event": event}
diff --git a/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkSendDefaultPii/.gitignore b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkSendDefaultPii/.gitignore
new file mode 100644
index 0000000000..1c56884372
--- /dev/null
+++ b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkSendDefaultPii/.gitignore
@@ -0,0 +1,11 @@
+# Need to add some ignore rules in this directory, because the unit tests will add the Sentry SDK and its dependencies
+# into this directory to create a Lambda function package that contains everything needed to instrument a Lambda function using Sentry.
+
+# Ignore everything
+*
+
+# But not index.py
+!index.py
+
+# And not .gitignore itself
+!.gitignore
diff --git a/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkSendDefaultPii/.lock b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkSendDefaultPii/.lock
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkSendDefaultPii/index.py b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkSendDefaultPii/index.py
new file mode 100644
index 0000000000..502a6ae3ee
--- /dev/null
+++ b/tests/integrations/aws_lambda/lambda_functions_with_embedded_sdk/BasicOkSendDefaultPii/index.py
@@ -0,0 +1,15 @@
+import os
+
+import sentry_sdk
+from sentry_sdk.integrations.aws_lambda import AwsLambdaIntegration
+
+sentry_sdk.init(
+    dsn=os.environ.get("SENTRY_DSN"),
+    traces_sample_rate=1.0,
+    send_default_pii=True,
+    integrations=[AwsLambdaIntegration()],
+)
+
+
+def handler(event, context):
+    return {"event": event}
diff --git a/tests/integrations/aws_lambda/test_aws_lambda.py b/tests/integrations/aws_lambda/test_aws_lambda.py
index ec01284e58..e01001fb6b 100644
--- a/tests/integrations/aws_lambda/test_aws_lambda.py
+++ b/tests/integrations/aws_lambda/test_aws_lambda.py
@@ -354,7 +354,7 @@ def test_non_dict_event(
         assert transaction_event["tags"]["batch_request"] is True
 
 
-def test_request_data(lambda_client, test_environment):
+def test_request_data_with_send_default_pii_false(lambda_client, test_environment):
     payload = b"""
         {
           "resource": "/asd",
@@ -363,7 +363,9 @@ def test_request_data(lambda_client, test_environment):
           "headers": {
             "Host": "iwsz2c7uwi.execute-api.us-east-1.amazonaws.com",
             "User-Agent": "custom",
-            "X-Forwarded-Proto": "https"
+            "X-Forwarded-Proto": "https",
+            "Authorization": "Bearer secret-token",
+            "Cookie": "sessionid=secret"
           },
           "queryStringParameters": {
             "bonkers": "true"
@@ -391,9 +393,186 @@ def test_request_data(lambda_client, test_environment):
 
     assert transaction_event["request"] == {
         "headers": {
+            "Host": "iwsz2c7uwi.execute-api.us-east-1.amazonaws.com",
+            "User-Agent": "custom",
+            # X-Forwarded-Proto is not sensitive and passes through.
+            "X-Forwarded-Proto": "https",
+            # With send_default_pii=False, _filter_headers substitutes the
+            # SENSITIVE_HEADERS (Authorization, Cookie); the EventScrubber
+            # also scrubs them. Both end up as "[Filtered]".
+            "Authorization": "[Filtered]",
+            "Cookie": "[Filtered]",
+        },
+        "method": "GET",
+        "query_string": {"bonkers": "true"},
+        "url": "https://iwsz2c7uwi.execute-api.us-east-1.amazonaws.com/asd",
+    }
+
+
+def test_request_data_with_send_default_pii_true(lambda_client, test_environment):
+    payload = b"""
+        {
+          "resource": "/asd",
+          "path": "/asd",
+          "httpMethod": "GET",
+          "headers": {
+            "Host": "iwsz2c7uwi.execute-api.us-east-1.amazonaws.com",
+            "User-Agent": "custom",
+            "X-Forwarded-Proto": "https",
+            "Authorization": "Bearer secret-token",
+            "Cookie": "sessionid=secret"
+          },
+          "queryStringParameters": {
+            "bonkers": "true"
+          },
+          "pathParameters": null,
+          "stageVariables": null,
+          "requestContext": {
+            "identity": {
+              "sourceIp": "213.47.147.207",
+              "userArn": "42"
+            }
+          },
+          "body": null,
+          "isBase64Encoded": false
+        }
+    """
+
+    lambda_client.invoke(
+        FunctionName="BasicOkSendDefaultPii",
+        Payload=payload,
+    )
+    envelopes = test_environment["server"].envelopes
+
+    (transaction_event,) = envelopes
+
+    assert transaction_event["request"] == {
+        "headers": {
+            "Host": "iwsz2c7uwi.execute-api.us-east-1.amazonaws.com",
+            "User-Agent": "custom",
+            "X-Forwarded-Proto": "https",
+            # With send_default_pii=True (and no data_collection config),
+            # _filter_headers passes headers through untouched. Authorization
+            # and Cookie are still scrubbed to "[Filtered]" by the always-on
+            # EventScrubber (DEFAULT_DENYLIST), independent of PII settings.
+            "Authorization": "[Filtered]",
+            "Cookie": "[Filtered]",
+        },
+        "method": "GET",
+        "query_string": {"bonkers": "true"},
+        "url": "https://iwsz2c7uwi.execute-api.us-east-1.amazonaws.com/asd",
+        "data": None,
+    }
+
+
+def test_request_data_with_data_collection_allowlist(lambda_client, test_environment):
+    payload = b"""
+        {
+          "resource": "/asd",
+          "path": "/asd",
+          "httpMethod": "GET",
+          "headers": {
+            "Host": "iwsz2c7uwi.execute-api.us-east-1.amazonaws.com",
+            "User-Agent": "custom",
+            "X-Forwarded-Proto": "https",
+            "Authorization": "Bearer secret-token",
+            "Cookie": "sessionid=secret",
+            "X-Allow-Me": "yes"
+          },
+          "queryStringParameters": {
+            "bonkers": "true"
+          },
+          "pathParameters": null,
+          "stageVariables": null,
+          "requestContext": {
+            "identity": {
+              "sourceIp": "213.47.147.207",
+              "userArn": "42"
+            }
+          },
+          "body": null,
+          "isBase64Encoded": false
+        }
+    """
+
+    lambda_client.invoke(
+        FunctionName="BasicOkDataCollectionAllowlist",
+        Payload=payload,
+    )
+    envelopes = test_environment["server"].envelopes
+
+    (transaction_event,) = envelopes
+
+    assert transaction_event["request"] == {
+        "headers": {
+            # Allowlisted, non-sensitive headers pass through.
+            "User-Agent": "custom",
+            "X-Allow-Me": "yes",
+            # Not allowlisted -> substituted.
+            "Host": "[Filtered]",
+            "X-Forwarded-Proto": "[Filtered]",
+            # Allowlisted but sensitive -> still filtered; an allowlist entry
+            # cannot override the built-in sensitive denylist.
+            "Authorization": "[Filtered]",
+            # Not allowlisted, and cookies are always substituted.
+            "Cookie": "[Filtered]",
+        },
+        "method": "GET",
+        "query_string": {"bonkers": "true"},
+        "url": "https://iwsz2c7uwi.execute-api.us-east-1.amazonaws.com/asd",
+    }
+
+
+def test_request_data_with_data_collection_denylist(lambda_client, test_environment):
+    payload = b"""
+        {
+          "resource": "/asd",
+          "path": "/asd",
+          "httpMethod": "GET",
+          "headers": {
             "Host": "iwsz2c7uwi.execute-api.us-east-1.amazonaws.com",
             "User-Agent": "custom",
             "X-Forwarded-Proto": "https",
+            "Authorization": "Bearer secret-token",
+            "Cookie": "sessionid=secret",
+            "X-Custom": "keep-me"
+          },
+          "queryStringParameters": {
+            "bonkers": "true"
+          },
+          "pathParameters": null,
+          "stageVariables": null,
+          "requestContext": {
+            "identity": {
+              "sourceIp": "213.47.147.207",
+              "userArn": "42"
+            }
+          },
+          "body": null,
+          "isBase64Encoded": false
+        }
+    """
+
+    lambda_client.invoke(
+        FunctionName="BasicOkDataCollectionDenylist",
+        Payload=payload,
+    )
+    envelopes = test_environment["server"].envelopes
+
+    (transaction_event,) = envelopes
+
+    assert transaction_event["request"] == {
+        "headers": {
+            # Not denied by any term -> pass through.
+            "Host": "iwsz2c7uwi.execute-api.us-east-1.amazonaws.com",
+            "X-Custom": "keep-me",
+            # Denied by custom terms.
+            "User-Agent": "[Filtered]",
+            "X-Forwarded-Proto": "[Filtered]",
+            # Denied by the built-in sensitive denylist.
+            "Authorization": "[Filtered]",
+            # Cookies are always substituted.
+            "Cookie": "[Filtered]",
         },
         "method": "GET",
         "query_string": {"bonkers": "true"},
@@ -401,6 +580,52 @@ def test_request_data(lambda_client, test_environment):
     }
 
 
+def test_request_data_with_data_collection_off(lambda_client, test_environment):
+    payload = b"""
+        {
+          "resource": "/asd",
+          "path": "/asd",
+          "httpMethod": "GET",
+          "headers": {
+            "Host": "iwsz2c7uwi.execute-api.us-east-1.amazonaws.com",
+            "User-Agent": "custom",
+            "X-Forwarded-Proto": "https",
+            "Authorization": "Bearer secret-token",
+            "Cookie": "sessionid=secret"
+          },
+          "queryStringParameters": {
+            "bonkers": "true"
+          },
+          "pathParameters": null,
+          "stageVariables": null,
+          "requestContext": {
+            "identity": {
+              "sourceIp": "213.47.147.207",
+              "userArn": "42"
+            }
+          },
+          "body": null,
+          "isBase64Encoded": false
+        }
+    """
+
+    lambda_client.invoke(
+        FunctionName="BasicOkDataCollectionOff",
+        Payload=payload,
+    )
+    envelopes = test_environment["server"].envelopes
+
+    (transaction_event,) = envelopes
+
+    assert transaction_event["request"] == {
+        # With request headers collection turned off, no headers are collected.
+        "headers": {},
+        "method": "GET",
+        "query_string": {"bonkers": "true"},
+        "url": "https://iwsz2c7uwi.execute-api.us-east-1.amazonaws.com/asd",
+    }
+
+
 def test_trace_continuation(lambda_client, test_environment):
     trace_id = "471a43a4192642f0b136d5159a501701"
     parent_span_id = "6e8f22c393e68f19"
diff --git a/tests/integrations/quart/test_quart.py b/tests/integrations/quart/test_quart.py
index 730b24ef33..e381dedfe0 100644
--- a/tests/integrations/quart/test_quart.py
+++ b/tests/integrations/quart/test_quart.py
@@ -13,8 +13,8 @@
     capture_message,
     set_tag,
 )
+from sentry_sdk._types import SENSITIVE_DATA_SUBSTITUTE
 from sentry_sdk.integrations.logging import LoggingIntegration
-from sentry_sdk.utils import SENSITIVE_DATA_SUBSTITUTE
 
 
 def quart_app_factory():
@@ -861,8 +861,171 @@ async def test_span_streaming_request_attributes_with_pii(sentry_init, capture_i
     assert "user.ip_address" in segment["attributes"]
 
 
+@pytest.mark.parametrize(
+    "options,expected",
+    [
+        pytest.param(
+            {
+                "send_default_pii": True,
+                "data_collection": None,
+            },
+            {
+                "authorization": "[Filtered]",
+                "custom": "passthrough",
+                "cookie": "[Filtered]",
+            },
+            id="enabled_send_default_pii_redacts_auth_header_due_to_data_collection_default_settings",
+        ),
+        pytest.param(
+            {
+                "send_default_pii": False,
+                "data_collection": None,
+            },
+            {
+                "authorization": "[Filtered]",
+                "custom": "passthrough",
+                "cookie": "[Filtered]",
+            },
+            id="disabled_send_default_pii_redacts_auth_header_due_to_data_collection_default_settings",
+        ),
+        pytest.param(
+            {
+                "send_default_pii": False,
+                "data_collection": {"http_headers": {"request": {"mode": "off"}}},
+            },
+            None,
+            id="data_collection_off_does_not_add_headers",
+        ),
+        pytest.param(
+            {
+                "send_default_pii": False,
+                "data_collection": {"http_headers": {"request": {"mode": "allowlist"}}},
+            },
+            {
+                "authorization": "[Filtered]",
+                "custom": "[Filtered]",
+                "cookie": "[Filtered]",
+            },
+            id="data_collection_allow_list_redacts_terms_that_do_not_appear",
+        ),
+        pytest.param(
+            {
+                "send_default_pii": False,
+                "data_collection": {
+                    "http_headers": {
+                        "request": {"mode": "allowlist", "terms": ["Authorization"]}
+                    }
+                },
+            },
+            {
+                "authorization": "[Filtered]",
+                "custom": "[Filtered]",
+                "cookie": "[Filtered]",
+            },
+            id="data_collection_allow_list_redacts_sensitive_terms_even_when_provided_by_user",
+        ),
+        pytest.param(
+            {
+                "send_default_pii": False,
+                "data_collection": {
+                    "http_headers": {
+                        "request": {"mode": "allowlist", "terms": ["custom"]}
+                    }
+                },
+            },
+            {
+                "authorization": "[Filtered]",
+                "custom": "passthrough",
+                "cookie": "[Filtered]",
+            },
+            id="data_collection_allow_list_does_not_redact_provided_term",
+        ),
+        pytest.param(
+            {
+                "send_default_pii": False,
+                "data_collection": {
+                    "http_headers": {
+                        "request": {"mode": "denylist", "terms": ["custom"]}
+                    }
+                },
+            },
+            {
+                "authorization": "[Filtered]",
+                "custom": "[Filtered]",
+                "cookie": "[Filtered]",
+            },
+            id="data_collection_deny_list_redacts_sensitive_terms_when_provided_by_user",
+        ),
+        pytest.param(
+            {
+                "send_default_pii": False,
+                "data_collection": {
+                    "http_headers": {
+                        "request": {"mode": "allowlist", "terms": ["cookie"]}
+                    }
+                },
+            },
+            {
+                "authorization": "[Filtered]",
+                "custom": "[Filtered]",
+                "cookie": "[Filtered]",
+            },
+            id="data_collection_cookie_is_always_redacted_even_when_allow_listed",
+        ),
+    ],
+)
 @pytest.mark.asyncio
-async def test_span_streaming_sensitive_header_scrubbing(sentry_init, capture_items):
+async def test_span_streaming_sensitive_header_scrubbing(
+    sentry_init, capture_items, options, expected, request
+):
+    sentry_init(
+        integrations=[quart_sentry.QuartIntegration()],
+        traces_sample_rate=1.0,
+        send_default_pii=options["send_default_pii"],
+        _experiments={
+            "trace_lifecycle": "stream",
+            "data_collection": options["data_collection"],
+        },
+    )
+    items = capture_items("span")
+
+    app = quart_app_factory()
+    client = app.test_client()
+    response = await client.get(
+        "/message",
+        headers={
+            "Authorization": "Bearer secret-token",
+            "X-Custom-Header": "passthrough",
+            "Cookie": "sessionid=secret",
+        },
+    )
+    assert response.status_code == 200
+
+    sentry_sdk.flush()
+
+    spans = [item.payload for item in items]
+    assert len(spans) == 1
+
+    segment = spans[0]
+    if request.node.callspec.id == "data_collection_off_does_not_add_headers":
+        assert "http.request.header.authorization" not in segment["attributes"]
+        assert "http.request.header.cookie" not in segment["attributes"]
+    else:
+        assert (
+            segment["attributes"]["http.request.header.authorization"]
+            == expected["authorization"]
+        )
+        assert (
+            segment["attributes"]["http.request.header.x-custom-header"]
+            == expected["custom"]
+        )
+        assert segment["attributes"]["http.request.header.cookie"] == expected["cookie"]
+
+
+@pytest.mark.asyncio
+async def test_span_streaming_sensitive_header_without_data_collection(
+    sentry_init, capture_items
+):
     sentry_init(
         integrations=[quart_sentry.QuartIntegration()],
         traces_sample_rate=1.0,
@@ -939,7 +1102,7 @@ async def login():
 
 
 @pytest.mark.asyncio
-async def test_span_streaming_sensitive_header_passthrough_with_pii(
+async def test_span_streaming_sensitive_header_passthrough_with_pii_and_no_data_collection(
     sentry_init, capture_items
 ):
     sentry_init(
diff --git a/tests/integrations/wsgi/test_wsgi.py b/tests/integrations/wsgi/test_wsgi.py
index 3b684adb0d..cceb351ba7 100644
--- a/tests/integrations/wsgi/test_wsgi.py
+++ b/tests/integrations/wsgi/test_wsgi.py
@@ -858,6 +858,256 @@ def test_get_request_url_x_forwarded_proto(environ, use_x_forwarded_for, expecte
     assert get_request_url(environ, use_x_forwarded_for) == expected_url
 
 
+@pytest.mark.parametrize("send_default_pii", [True, False])
+def test_request_headers_data_collection_default_redacts_sensitive(
+    sentry_init, crashing_app, capture_events, send_default_pii
+):
+    """
+    When ``data_collection`` is configured (even as ``None``, i.e. spec
+    defaults), the WSGI event processor routes request headers through the
+    data-collection filtering path. Sensitive headers are redacted regardless
+    of ``send_default_pii`` -- the value of that legacy option must not change
+    the outcome.
+    """
+    sentry_init(
+        send_default_pii=send_default_pii,
+        _experiments={"data_collection": None},
+    )
+    app = SentryWsgiMiddleware(crashing_app)
+    client = Client(app)
+    events = capture_events()
+
+    with pytest.raises(ZeroDivisionError):
+        client.get(
+            "/",
+            headers={
+                "Authorization": "Bearer secret-token",
+                "X-Custom-Header": "passthrough",
+            },
+        )
+
+    (event,) = events
+    headers = event["request"]["headers"]
+
+    assert headers["Authorization"] == "[Filtered]"
+    assert headers["X-Custom-Header"] == "passthrough"
+
+
+def test_request_headers_legacy_no_pii_redacts_sensitive(
+    sentry_init, crashing_app, capture_events
+):
+    """
+    With no ``data_collection`` configured, ``_filter_headers`` falls back to
+    the legacy ``send_default_pii`` behaviour. When PII is disabled, headers in
+    ``SENSITIVE_HEADERS`` are replaced with an ``AnnotatedValue`` (the default
+    ``use_annotated_value=True`` on the event-processor call site), which
+    serializes to an emptied value plus a ``_meta`` annotation. Non-sensitive
+    headers pass through untouched.
+
+    ``X-Forwarded-For`` is used because it is in ``SENSITIVE_HEADERS`` but is
+    not scrubbed by the default ``EventScrubber``, so the substitution we are
+    asserting on can only come from ``_filter_headers``.
+    """
+    sentry_init(send_default_pii=False)
+    app = SentryWsgiMiddleware(crashing_app)
+    client = Client(app)
+    events = capture_events()
+
+    with pytest.raises(ZeroDivisionError):
+        client.get(
+            "/",
+            headers={
+                "X-Forwarded-For": "1.2.3.4",
+                "X-Custom-Header": "passthrough",
+            },
+        )
+
+    (event,) = events
+
+    assert event["request"]["headers"]["X-Forwarded-For"] == ""
+    assert event["request"]["headers"]["X-Custom-Header"] == "passthrough"
+
+    # The emptied value is accompanied by a `_meta` annotation marking it as
+    # removed, confirming the substitution came from the AnnotatedValue path.
+    assert event["_meta"]["request"]["headers"]["X-Forwarded-For"] == {
+        "": {"rem": [["!config", "x"]]}
+    }
+
+
+def test_request_headers_data_collection_off_collects_no_headers(
+    sentry_init, crashing_app, capture_events
+):
+    """
+    With ``http_headers.request`` mode set to ``off``, no request headers are
+    collected at all -- the filtering returns an empty mapping.
+    """
+    sentry_init(
+        _experiments={
+            "data_collection": {"http_headers": {"request": {"mode": "off"}}}
+        },
+    )
+    app = SentryWsgiMiddleware(crashing_app)
+    client = Client(app)
+    events = capture_events()
+
+    with pytest.raises(ZeroDivisionError):
+        client.get(
+            "/",
+            headers={
+                "X-Forwarded-For": "1.2.3.4",
+                "X-Custom-Header": "passthrough",
+            },
+        )
+
+    (event,) = events
+
+    assert event["request"]["headers"] == {}
+
+
+def test_request_headers_data_collection_allowlist_redacts_all_but_allowed_terms(
+    sentry_init, crashing_app, capture_events
+):
+    """
+    An ``allowlist`` allows through only headers matching a configured term
+    (partial, case-insensitive); every other header key is kept but its value
+    is redacted.
+    """
+    sentry_init(
+        _experiments={
+            "data_collection": {
+                "http_headers": {"request": {"mode": "allowlist", "terms": ["custom"]}}
+            }
+        },
+    )
+    app = SentryWsgiMiddleware(crashing_app)
+    client = Client(app)
+    events = capture_events()
+
+    with pytest.raises(ZeroDivisionError):
+        client.get(
+            "/",
+            headers={
+                "X-Forwarded-For": "1.2.3.4",
+                "X-Custom-Header": "passthrough",
+            },
+        )
+
+    (event,) = events
+    headers = event["request"]["headers"]
+
+    assert headers["X-Custom-Header"] == "passthrough"
+    assert headers["X-Forwarded-For"] == "[Filtered]"
+    assert headers["Host"] == "[Filtered]"
+
+
+def test_request_headers_data_collection_denylist_redacts_only_matched_terms(
+    sentry_init, crashing_app, capture_events
+):
+    """
+    A ``denylist`` passes headers through by default, redacting only those
+    matching a configured term (partial, case-insensitive).
+    """
+    sentry_init(
+        _experiments={
+            "data_collection": {
+                "http_headers": {"request": {"mode": "denylist", "terms": ["custom"]}}
+            }
+        },
+    )
+    app = SentryWsgiMiddleware(crashing_app)
+    client = Client(app)
+    events = capture_events()
+
+    with pytest.raises(ZeroDivisionError):
+        client.get(
+            "/",
+            headers={
+                "X-Forwarded-For": "1.2.3.4",
+                "X-Custom-Header": "passthrough",
+            },
+        )
+
+    (event,) = events
+    headers = event["request"]["headers"]
+
+    assert headers["X-Custom-Header"] == "[Filtered]"
+    assert headers["X-Forwarded-For"] == "1.2.3.4"
+    assert headers["Host"] == "localhost"
+
+
+def test_request_headers_data_collection_cookie_always_redacted(
+    sentry_init, crashing_app, capture_events
+):
+    """
+    The ``cookie``/``set-cookie`` headers are always redacted in the
+    data-collection path, even when explicitly allowlisted. A sibling header
+    (``custom``) that is also allowlisted passes through, isolating the
+    cookie override.
+
+    The middleware is driven with an explicit environ because werkzeug's test
+    ``Client`` manages its own cookie jar and strips the ``Cookie`` header.
+    """
+    sentry_init(
+        _experiments={
+            "data_collection": {
+                "http_headers": {
+                    "request": {"mode": "allowlist", "terms": ["cookie", "custom"]}
+                }
+            }
+        },
+    )
+    app = SentryWsgiMiddleware(crashing_app)
+    events = capture_events()
+
+    environ = {
+        "REQUEST_METHOD": "GET",
+        "PATH_INFO": "/",
+        "SERVER_NAME": "localhost",
+        "SERVER_PORT": "80",
+        "wsgi.url_scheme": "http",
+        "HTTP_COOKIE": "sessionid=secret",
+        "HTTP_X_CUSTOM_HEADER": "passthrough",
+    }
+
+    with pytest.raises(ZeroDivisionError):
+        list(app(environ, lambda status, headers: None))
+
+    (event,) = events
+    headers = event["request"]["headers"]
+
+    assert headers["Cookie"] == "[Filtered]"
+    assert headers["X-Custom-Header"] == "passthrough"
+
+
+def test_request_headers_legacy_pii_passes_headers_through(
+    sentry_init, crashing_app, capture_events
+):
+    """
+    With no ``data_collection`` configured and ``send_default_pii`` enabled,
+    the legacy path returns all headers unchanged -- including those in
+    ``SENSITIVE_HEADERS``.
+    """
+    sentry_init(send_default_pii=True)
+    app = SentryWsgiMiddleware(crashing_app)
+    client = Client(app)
+    events = capture_events()
+
+    with pytest.raises(ZeroDivisionError):
+        client.get(
+            "/",
+            headers={
+                "X-Forwarded-For": "1.2.3.4",
+                "X-Custom-Header": "passthrough",
+            },
+        )
+
+    (event,) = events
+    headers = event["request"]["headers"]
+
+    assert headers["X-Forwarded-For"] == "1.2.3.4"
+    assert headers["X-Custom-Header"] == "passthrough"
+
+
 @pytest.mark.parametrize("send_default_pii", [True, False])
 def test_user_ip_address_on_all_spans(sentry_init, capture_items, send_default_pii):
     def dogpark(environ, start_response):