Add synthetic detection on the server side.

Author: JacksonWeber

instrumentation/opentelemetry-instrumentation-asgi/src/opentelemetry/instrumentation/asgi/__init__.py

@@ -258,6 +258,9 @@ def client_response_hook(span: Span, scope: Scope, message: dict[str, Any]):
 from opentelemetry.instrumentation.utils import _start_internal_or_server_span
 from opentelemetry.metrics import get_meter
 from opentelemetry.propagators.textmap import Getter, Setter
+from opentelemetry.semconv._incubating.attributes.user_agent_attributes import (
+    USER_AGENT_SYNTHETIC_TYPE,
+)
 from opentelemetry.semconv._incubating.metrics.http_metrics import (
     create_http_server_active_requests,
     create_http_server_request_body_size,
@@ -276,6 +279,7 @@ def client_response_hook(span: Span, scope: Scope, message: dict[str, Any]):
     ExcludeList,
     SanitizeValue,
     _parse_url_query,
+    detect_synthetic_user_agent,
     get_custom_headers,
     normalise_request_header_name,
     normalise_response_header_name,
@@ -397,7 +401,13 @@ def collect_request_attributes(
             )
     http_user_agent = asgi_getter.get(scope, "user-agent")
     if http_user_agent:
-        _set_http_user_agent(result, http_user_agent[0], sem_conv_opt_in_mode)
+        user_agent_value = http_user_agent[0]
+        _set_http_user_agent(result, user_agent_value, sem_conv_opt_in_mode)
+
+        # Check for synthetic user agent type
+        synthetic_type = detect_synthetic_user_agent(user_agent_value)
+        if synthetic_type:
+            result[USER_AGENT_SYNTHETIC_TYPE] = synthetic_type
 
     if "client" in scope and scope["client"] is not None:
         _set_http_peer_ip_server(

instrumentation/opentelemetry-instrumentation-asgi/tests/test_asgi_middleware.py

@@ -42,6 +42,9 @@
     HistogramDataPoint,
     NumberDataPoint,
 )
+from opentelemetry.semconv._incubating.attributes.user_agent_attributes import (
+    USER_AGENT_SYNTHETIC_TYPE,
+)
 from opentelemetry.semconv.attributes.client_attributes import (
     CLIENT_ADDRESS,
     CLIENT_PORT,
@@ -883,6 +886,133 @@ def update_expected_user_agent(expected):
             new_sem_conv=True,
         )
 
+    async def test_user_agent_synthetic_bot_detection(self):
+        """Test that bot user agents are detected as synthetic with type 'bot'"""
+        test_cases = [
+            b"Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)",
+            b"Mozilla/5.0 (compatible; bingbot/2.0; +http://www.bing.com/bingbot.htm)",
+            b"googlebot/1.0",
+            b"bingbot/1.0",
+        ]
+
+        for user_agent in test_cases:
+            with self.subTest(user_agent=user_agent):
+                # Clear headers first
+                self.scope["headers"] = []
+
+                def update_expected_synthetic_bot(expected):
+                    expected[3]["attributes"].update(
+                        {
+                            SpanAttributes.HTTP_USER_AGENT: user_agent.decode(
+                                "utf8"
+                            ),
+                            USER_AGENT_SYNTHETIC_TYPE: "bot",
+                        }
+                    )
+                    return expected
+
+                self.scope["headers"].append([b"user-agent", user_agent])
+                app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
+                self.seed_app(app)
+                await self.send_default_request()
+                outputs = await self.get_all_output()
+                self.validate_outputs(
+                    outputs, modifiers=[update_expected_synthetic_bot]
+                )
+
+    async def test_user_agent_synthetic_test_detection(self):
+        """Test that test user agents are detected as synthetic with type 'test'"""
+        test_cases = [
+            b"alwayson/1.0",
+            b"AlwaysOn/2.0",
+            b"test-alwayson-client",
+        ]
+
+        for user_agent in test_cases:
+            with self.subTest(user_agent=user_agent):
+                # Clear headers first
+                self.scope["headers"] = []
+
+                def update_expected_synthetic_test(expected):
+                    expected[3]["attributes"].update(
+                        {
+                            SpanAttributes.HTTP_USER_AGENT: user_agent.decode(
+                                "utf8"
+                            ),
+                            USER_AGENT_SYNTHETIC_TYPE: "test",
+                        }
+                    )
+                    return expected
+
+                self.scope["headers"].append([b"user-agent", user_agent])
+                app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
+                self.seed_app(app)
+                await self.send_default_request()
+                outputs = await self.get_all_output()
+                self.validate_outputs(
+                    outputs, modifiers=[update_expected_synthetic_test]
+                )
+
+    async def test_user_agent_non_synthetic(self):
+        """Test that normal user agents are not marked as synthetic"""
+        test_cases = [
+            b"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
+            b"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.1 Safari/605.1.15",
+            b"PostmanRuntime/7.28.4",
+            b"curl/7.68.0",
+        ]
+
+        for user_agent in test_cases:
+            with self.subTest(user_agent=user_agent):
+                # Clear headers first
+                self.scope["headers"] = []
+
+                def update_expected_non_synthetic(expected):
+                    # Should only have the user agent, not synthetic type
+                    expected[3]["attributes"].update(
+                        {
+                            SpanAttributes.HTTP_USER_AGENT: user_agent.decode(
+                                "utf8"
+                            ),
+                        }
+                    )
+                    return expected
+
+                self.scope["headers"].append([b"user-agent", user_agent])
+                app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
+                self.seed_app(app)
+                await self.send_default_request()
+                outputs = await self.get_all_output()
+                self.validate_outputs(
+                    outputs, modifiers=[update_expected_non_synthetic]
+                )
+
+    async def test_user_agent_synthetic_new_semconv(self):
+        """Test synthetic user agent detection with new semantic conventions"""
+        user_agent = b"Mozilla/5.0 (compatible; Googlebot/2.1)"
+
+        def update_expected_synthetic_new_semconv(expected):
+            expected[3]["attributes"].update(
+                {
+                    USER_AGENT_ORIGINAL: user_agent.decode("utf8"),
+                    USER_AGENT_SYNTHETIC_TYPE: "bot",
+                }
+            )
+            return expected
+
+        self.scope["headers"] = []
+        self.scope["headers"].append([b"user-agent", user_agent])
+        app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
+        self.seed_app(app)
+        await self.send_default_request()
+        outputs = await self.get_all_output()
+        self.validate_outputs(
+            outputs,
+            modifiers=[update_expected_synthetic_new_semconv],
+            old_sem_conv=False,
+            new_sem_conv=True,
+        )
+
     async def test_traceresponse_header(self):
         """Test a traceresponse header is sent when a global propagator is set."""
 

instrumentation/opentelemetry-instrumentation-wsgi/src/opentelemetry/instrumentation/wsgi/__init__.py

@@ -258,6 +258,9 @@ def response_hook(span: Span, environ: WSGIEnvironment, status: str, response_he
     HTTP_SERVER_NAME,
     HTTP_URL,
 )
+from opentelemetry.semconv._incubating.attributes.user_agent_attributes import (
+    USER_AGENT_SYNTHETIC_TYPE,
+)
 from opentelemetry.semconv.attributes.error_attributes import ERROR_TYPE
 from opentelemetry.semconv.metrics import MetricInstruments
 from opentelemetry.semconv.metrics.http_metrics import (
@@ -271,6 +274,7 @@ def response_hook(span: Span, environ: WSGIEnvironment, status: str, response_he
     OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE,
     SanitizeValue,
     _parse_url_query,
+    detect_synthetic_user_agent,
     get_custom_headers,
     normalise_request_header_name,
     normalise_response_header_name,
@@ -391,6 +395,11 @@ def collect_request_attributes(
     if user_agent is not None and len(user_agent) > 0:
         _set_http_user_agent(result, user_agent, sem_conv_opt_in_mode)
 
+        # Check for synthetic user agent type
+        synthetic_type = detect_synthetic_user_agent(user_agent)
+        if synthetic_type:
+            result[USER_AGENT_SYNTHETIC_TYPE] = synthetic_type
+
     flavor = environ.get("SERVER_PROTOCOL", "")
     if flavor.upper().startswith(_HTTP_VERSION_PREFIX):
         flavor = flavor[len(_HTTP_VERSION_PREFIX) :]

instrumentation/opentelemetry-instrumentation-wsgi/tests/test_wsgi_middleware.py

@@ -52,6 +52,9 @@
     NET_HOST_NAME,
     NET_HOST_PORT,
 )
+from opentelemetry.semconv._incubating.attributes.user_agent_attributes import (
+    USER_AGENT_SYNTHETIC_TYPE,
+)
 from opentelemetry.semconv.attributes.http_attributes import (
     HTTP_REQUEST_METHOD,
     HTTP_RESPONSE_STATUS_CODE,
@@ -791,6 +794,83 @@ def test_http_user_agent_attribute(self):
             expected_new.items(),
         )
 
+    def test_http_user_agent_synthetic_bot_detection(self):
+        """Test that bot user agents are detected as synthetic with type 'bot'"""
+        test_cases = [
+            "Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)",
+            "Mozilla/5.0 (compatible; bingbot/2.0; +http://www.bing.com/bingbot.htm)",
+            "googlebot/1.0",
+            "bingbot/1.0",
+        ]
+
+        for user_agent in test_cases:
+            with self.subTest(user_agent=user_agent):
+                self.environ["HTTP_USER_AGENT"] = user_agent
+                attributes = otel_wsgi.collect_request_attributes(self.environ)
+
+                # Should have both the original user agent and synthetic type
+                self.assertIn(HTTP_USER_AGENT, attributes)
+                self.assertEqual(attributes[HTTP_USER_AGENT], user_agent)
+                self.assertIn(USER_AGENT_SYNTHETIC_TYPE, attributes)
+                self.assertEqual(attributes[USER_AGENT_SYNTHETIC_TYPE], "bot")
+
+    def test_http_user_agent_synthetic_test_detection(self):
+        """Test that test user agents are detected as synthetic with type 'test'"""
+        test_cases = [
+            "alwayson/1.0",
+            "AlwaysOn/2.0",
+            "test-alwayson-client",
+        ]
+
+        for user_agent in test_cases:
+            with self.subTest(user_agent=user_agent):
+                self.environ["HTTP_USER_AGENT"] = user_agent
+                attributes = otel_wsgi.collect_request_attributes(self.environ)
+
+                # Should have both the original user agent and synthetic type
+                self.assertIn(HTTP_USER_AGENT, attributes)
+                self.assertEqual(attributes[HTTP_USER_AGENT], user_agent)
+                self.assertIn(USER_AGENT_SYNTHETIC_TYPE, attributes)
+                self.assertEqual(attributes[USER_AGENT_SYNTHETIC_TYPE], "test")
+
+    def test_http_user_agent_non_synthetic(self):
+        """Test that normal user agents are not marked as synthetic"""
+        test_cases = [
+            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
+            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.1 Safari/605.1.15",
+            "PostmanRuntime/7.28.4",
+            "curl/7.68.0",
+        ]
+
+        for user_agent in test_cases:
+            with self.subTest(user_agent=user_agent):
+                self.environ["HTTP_USER_AGENT"] = user_agent
+                attributes = otel_wsgi.collect_request_attributes(self.environ)
+
+                # Should have the original user agent but no synthetic type
+                self.assertIn(HTTP_USER_AGENT, attributes)
+                self.assertEqual(attributes[HTTP_USER_AGENT], user_agent)
+                self.assertNotIn(USER_AGENT_SYNTHETIC_TYPE, attributes)
+
+    def test_http_user_agent_synthetic_new_semconv(self):
+        """Test synthetic user agent detection with new semantic conventions"""
+        self.environ["HTTP_USER_AGENT"] = (
+            "Mozilla/5.0 (compatible; Googlebot/2.1)"
+        )
+        attributes = otel_wsgi.collect_request_attributes(
+            self.environ,
+            _StabilityMode.HTTP,
+        )
+
+        # Should have both the new semconv user agent and synthetic type
+        self.assertIn(USER_AGENT_ORIGINAL, attributes)
+        self.assertEqual(
+            attributes[USER_AGENT_ORIGINAL],
+            "Mozilla/5.0 (compatible; Googlebot/2.1)",
+        )
+        self.assertIn(USER_AGENT_SYNTHETIC_TYPE, attributes)
+        self.assertEqual(attributes[USER_AGENT_SYNTHETIC_TYPE], "bot")
+
     def test_response_attributes(self):
         otel_wsgi.add_response_attributes(self.span, "404 Not Found", {})
         otel_wsgi.add_response_attributes(