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

Author: sentrivana

sentry_sdk/integrations/asgi.py

@@ -48,6 +48,7 @@
     capture_internal_exceptions,
     event_from_exception,
     has_data_collection_enabled,
+    is_localhost,
     logger,
     nullcontext,
     qualname_from_function,
@@ -232,6 +233,18 @@ async def _run_app(
                     processor = partial(self.event_processor, asgi_scope=scope)
                     sentry_scope.add_event_processor(processor)
 
+                    headers = _get_headers(scope)
+                    server = scope.get("server")
+                    sentry_scope.set_attribute(
+                        SPANDATA.SENTRY_IS_LOCALHOST,
+                        is_localhost(
+                            client_ip=_get_ip(scope) if scope.get("client") else None,
+                            url_host=server[0] if server else None,
+                            host_header=headers.get("host"),
+                            forwarded_host_header=headers.get("x-forwarded-host"),
+                        ),
+                    )
+
                     ty = scope["type"]
                     (
                         transaction_name,

sentry_sdk/integrations/wsgi.py

@@ -131,7 +131,9 @@ def __call__(
                         SPANDATA.SENTRY_IS_LOCALHOST,
                         is_localhost(
                             client_ip=get_client_ip(environ),
-                            url_host=environ.get("SERVER_NAME"),
+                            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"),
                         ),

tests/integrations/asgi/test_asgi.py

@@ -1490,3 +1490,73 @@ 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
+        ("203.0.113.50", ("localhost", 8000), b"example.com", 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"]
+            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