feat(wsgi): Add sentry.is_localhost for inbound filters

sentry_sdk/consts.py

@@ -1164,6 +1164,14 @@ class SPANDATA:
     Example: "prod"
     """
 
+    SENTRY_IS_LOCALHOST = "sentry.is_localhost"
+    """
+    Indicates whether a telemetry item was sent on a host, device or browser on
+    a localhost URL or IP address.
+
+    Used in inbound filters.
+    """
+
     SENTRY_RELEASE = "sentry.release"
     """
     The Sentry release.

sentry_sdk/integrations/wsgi.py

@@ -18,6 +18,7 @@
 from sentry_sdk.tracing_utils import has_span_streaming_enabled
 from sentry_sdk.utils import (
     ContextVar,
+    _is_localhost,
     capture_internal_exceptions,
     event_from_exception,
     has_data_collection_enabled,
@@ -126,6 +127,18 @@
                             )
                         )
 
+                    scope.set_attribute(
+                        SPANDATA.SENTRY_IS_LOCALHOST,
+                        _is_localhost(
+                            client_ip=get_client_ip(environ),
+                            url_host=environ.get("SERVER_NAME")
+                            if not environ.get("HTTP_HOST")
+                            else None,
+                            host_header=environ.get("HTTP_HOST"),
+                            forwarded_host_header=environ.get("HTTP_X_FORWARDED_HOST"),
+                        ),
+                    )
+
                     method = environ.get("REQUEST_METHOD", "").upper()
 
                     span_ctx: "Optional[ContextManager[Union[Span, StreamedSpan, None]]]" = None

sentry_sdk/utils.py

@@ -100,6 +100,9 @@
 be affected by this limit if they have a custom recursion limit.
 """
 
+_LOCAL_IPS = frozenset({"127.0.0.1", "::1"})
+_LOCAL_DOMAINS = frozenset({"localhost", "127.0.0.1"})
+
 
 def env_to_bool(value: "Any", *, strict: "Optional[bool]" = False) -> "bool | None":
     """Casts an ENV variable value to boolean using the constants defined above.
@@ -2205,6 +2208,43 @@
     return {"value": safe_repr(val), "type": "string"}
 
 
+def _host_matches_local_domain(host: str) -> bool:
+    """Check if host matches a local domain, including subdomains like foo.localhost."""
+    if host in _LOCAL_DOMAINS:
+        return True
+    for domain in _LOCAL_DOMAINS:
+        if host.endswith("." + domain):
+            return True
+    return False
+
+
+def _is_localhost(
+    client_ip: "Optional[str]" = None,
+    url_host: "Optional[str]" = None,
+    host_header: "Optional[str]" = None,
+    forwarded_host_header: "Optional[str]" = None,
+) -> bool:
+    """
+    Determine if a request originates from localhost.
+
+    Based on the logic in relay's localhost filter.
+    """
+    if client_ip is not None and client_ip in _LOCAL_IPS:
+        return True
+
+    # URL host uses subdomain-aware matching
+    if url_host is not None and _host_matches_local_domain(url_host):
+        return True
+
+    for header_val in (host_header, forwarded_host_header):
+        if header_val is not None:
+            domain = header_val.split(":")[0]
+            if _host_matches_local_domain(domain):
+                return True
+
+    return False
+
+
 # This noop context manager can be replaced with "from contextlib import nullcontext" when we drop Python 3.6 support
 @contextmanager
 def nullcontext() -> "Iterator[None]":

tests/integrations/wsgi/test_wsgi.py

@@ -1333,3 +1333,54 @@
     (event,) = events
 
     assert "ip_address" not in event.get("user", {})
+
+
+@pytest.mark.parametrize(
+    "client_kwargs, is_localhost",
+    [
+        # Default werkzeug client: REMOTE_ADDR=127.0.0.1, Host=localhost
+        ({"path": "/dogs/"}, True),
+        # IPv6 loopback
+        ({"path": "/dogs/", "environ_base": {"REMOTE_ADDR": "::1"}}, True),
+        # Localhost Host header with non-local IP
+        (
+            {
+                "path": "http://localhost:8000/dogs/",
+                "environ_overrides": {"REMOTE_ADDR": "1.2.3.4"},
+            },
+            True,
+        ),
+        # Non-local IP and non-local host
+        (
+            {
+                "path": "/dogs/",
+                "base_url": "http://example.com",
+                "environ_overrides": {"REMOTE_ADDR": "1.2.3.4"},
+            },
+            False,
+        ),
+    ],
+)
+def test_is_localhost_span_attribute(
+    sentry_init, capture_items, client_kwargs, is_localhost
+):
+    def dogpark(environ, start_response):
+        with sentry_sdk.traces.start_span(name="child-span"):
+            pass
+        start_response("200 OK", [])
+        return ["woof"]
+
+    sentry_init(traces_sample_rate=1.0, trace_lifecycle="stream")
+    app = SentryWsgiMiddleware(dogpark)
+    client = Client(app)
+
+    items = capture_items("span")
+
+    client.get(**client_kwargs)
+
+    sentry_sdk.flush()
+
+    child_span, server_span = [item.payload for item in items]
+
+    assert server_span["attributes"]["sentry.is_localhost"] is is_localhost
+    assert child_span["attributes"]["sentry.is_localhost"] is is_localhost

tests/test_utils.py

@@ -13,6 +13,7 @@
     Components,
     Dsn,
     _get_installed_modules,
+    _is_localhost,
     datetime_from_isoformat,
     ensure_integration_enabled,
     env_to_bool,
@@ -1409,3 +1410,53 @@
 
 def test_package_version_is_none():
     assert package_version("non_existent_package") is None
+
+
+@pytest.mark.parametrize(
+    "kwargs, expected",
+    [
+        # Client IP matches
+        ({"client_ip": "127.0.0.1"}, True),
+        ({"client_ip": "::1"}, True),
+        ({"client_ip": "192.168.1.1"}, False),
+        ({"client_ip": "10.0.0.1"}, False),
+        ({"client_ip": "8.8.8.8"}, False),
+        ({"client_ip": "::2"}, False),
+        # URL host (subdomain-aware matching)
+        ({"url_host": "localhost"}, True),
+        ({"url_host": "127.0.0.1"}, True),
+        ({"url_host": "foo.localhost"}, True),
+        ({"url_host": "bar.foo.localhost"}, True),
+        ({"url_host": "foolocalhost"}, False),
+        ({"url_host": "notlocalhost"}, False),
+        ({"url_host": "example.com"}, False),
+        # Host header (subdomain-aware matching after port strip)
+        ({"host_header": "localhost"}, True),
+        ({"host_header": "localhost:3000"}, True),
+        ({"host_header": "127.0.0.1:8080"}, True),
+        ({"host_header": "foo.localhost"}, True),
+        ({"host_header": "foo.localhost:8080"}, True),
+        ({"host_header": "example.com"}, False),
+        ({"host_header": "localhost.example.com"}, False),
+        # X-Forwarded-Host header
+        ({"forwarded_host_header": "localhost"}, True),
+        ({"forwarded_host_header": "localhost:3000"}, True),
+        ({"forwarded_host_header": "127.0.0.1:8080"}, True),
+        ({"forwarded_host_header": "foo.localhost"}, True),
+        ({"forwarded_host_header": "example.com"}, False),
+        # No args
+        ({}, False),
+        # All non-local
+        (
+            {
+                "client_ip": "8.8.8.8",
+                "url_host": "example.com",
+                "host_header": "example.com",
+                "forwarded_host_header": "example.com",
+            },
+            False,
+        ),
+    ],
+)
+def test_is_localhost(kwargs, expected):
+    assert _is_localhost(**kwargs) is expected