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

sentry_sdk/integrations/_asgi_common.py

@@ -88,7 +88,7 @@ def _get_query(asgi_scope: "Any") -> "Optional[str]":
     return urllib.parse.unquote(qs.decode("latin-1"))
 
 
-def _get_ip(asgi_scope: "Any") -> str:
+def _get_ip(asgi_scope: "Any") -> "Optional[str]":
     """
     Extract IP Address from the ASGI scope based on request headers with fallback to scope client.
     """
@@ -103,7 +103,10 @@ def _get_ip(asgi_scope: "Any") -> str:
     except KeyError:
         pass
 
-    return asgi_scope.get("client")[0]
+    if asgi_scope.get("client"):
+        return asgi_scope["client"][0]
+
+    return None
 
 
 def _get_request_data(

sentry_sdk/integrations/asgi.py

@@ -45,6 +45,7 @@
     HAS_REAL_CONTEXTVARS,
     ContextVar,
     _get_installed_modules,
+    _is_localhost,
     capture_internal_exceptions,
     event_from_exception,
     has_data_collection_enabled,
@@ -232,6 +233,19 @@ async def _run_app(
                     processor = partial(self.event_processor, asgi_scope=scope)
                     sentry_scope.add_event_processor(processor)
 
+                    headers = _get_headers(scope)
+                    sentry_scope.set_attribute(
+                        SPANDATA.SENTRY_IS_LOCALHOST,
+                        _is_localhost(
+                            client_ip=_get_ip(scope),
+                            url_host=scope.get("server")[0]
+                            if scope.get("server") and not headers.get("host")
+                            else None,
+                            host_header=headers.get("host"),
+                            forwarded_host_header=headers.get("x-forwarded-host"),
+                        ),
+                    )
+
                     ty = scope["type"]
                     (
                         transaction_name,
@@ -254,17 +268,16 @@ async def _run_app(
                             "network.protocol.name": ty,
                         }
 
-                        if scope.get("client"):
+                        ip = _get_ip(scope)
+                        if ip:
                             client_options = sentry_sdk.get_client().options
                             if has_data_collection_enabled(client_options):
                                 if client_options["data_collection"]["user_info"]:
                                     sentry_scope.set_attribute(
-                                        SPANDATA.USER_IP_ADDRESS, _get_ip(scope)
+                                        SPANDATA.USER_IP_ADDRESS, ip
                                     )
                             elif should_send_default_pii():
-                                sentry_scope.set_attribute(
-                                    SPANDATA.USER_IP_ADDRESS, _get_ip(scope)
-                                )
+                                sentry_scope.set_attribute(SPANDATA.USER_IP_ADDRESS, ip)
 
                         if ty in ("http", "websocket"):
                             if (

tests/integrations/asgi/test_asgi.py

@@ -1490,3 +1490,76 @@ async def wrapped_app(scope, receive, send):
     else:
         assert "user.ip_address" not in server_span["attributes"]
         assert "user.ip_address" not in child_span["attributes"]
+
+
+@pytest.mark.parametrize(
+    "client_ip, server, host_header, is_localhost",
+    [
+        # Loopback IP
+        ("127.0.0.1", ("example.com", 80), b"example.com", True),
+        # IPv6 loopback
+        ("::1", ("example.com", 80), b"example.com", True),
+        # Localhost host header with non-local IP
+        ("203.0.113.50", ("example.com", 80), b"localhost:8000", True),
+        # Server bound to localhost, but host header is public (reverse proxy)
+        ("203.0.113.50", ("localhost", 8000), b"example.com", False),
+        # Server bound to localhost, no host header (fallback to server)
+        ("203.0.113.50", ("localhost", 8000), None, True),
+        # Non-local everything
+        ("203.0.113.50", ("example.com", 80), b"example.com", False),
+    ],
+)
+@pytest.mark.asyncio
+async def test_is_localhost_attribute(
+    sentry_init,
+    capture_items,
+    client_ip,
+    server,
+    host_header,
+    is_localhost,
+):
+    async def app(scope, receive, send):
+        if scope["type"] == "lifespan":
+            while True:
+                message = await receive()
+                if message["type"] == "lifespan.startup":
+                    await send({"type": "lifespan.startup.complete"})
+                elif message["type"] == "lifespan.shutdown":
+                    await send({"type": "lifespan.shutdown.complete"})
+                    return
+
+        with sentry_sdk.traces.start_span(name="child-span"):
+            pass
+
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [[b"content-type", b"text/plain"]],
+            }
+        )
+        await send({"type": "http.response.body", "body": b"Hello, world!"})
+
+    sentry_init(traces_sample_rate=1.0, trace_lifecycle="stream")
+    sentry_app = SentryAsgiMiddleware(app)
+
+    async def wrapped_app(scope, receive, send):
+        if scope["type"] != "lifespan":
+            scope["client"] = (client_ip, 0)
+            scope["server"] = server
+            scope["headers"] = [(k, v) for k, v in scope["headers"] if k != b"host"]
+            if host_header is not None:
+                scope["headers"].append((b"host", host_header))
+
+        await sentry_app(scope, receive, send)
+
+    async with TestClient(wrapped_app) as client:
+        items = capture_items("span")
+        await client.get("/some_url")
+
+    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