Fix domain override validation to accept valid URLs with protocols

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/exporters/agent365_exporter.py

@@ -10,6 +10,7 @@
 import time
 from collections.abc import Callable, Sequence
 from typing import Any, final
+from urllib.parse import urlparse
 
 import requests
 from microsoft_agents_a365.runtime.power_platform_api_discovery import PowerPlatformApiDiscovery
@@ -94,12 +95,24 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
                 else:
                     discovery = PowerPlatformApiDiscovery(self._cluster_category)
                     endpoint = discovery.get_tenant_island_cluster_endpoint(tenant_id)
+
                 endpoint_path = (
                     f"/maven/agent365/service/agents/{agent_id}/traces"
                     if self._use_s2s_endpoint
                     else f"/maven/agent365/agents/{agent_id}/traces"
                 )
-                url = f"https://{endpoint}{endpoint_path}?api-version=1"
+
+                # Construct URL - if endpoint has a scheme (http:// or https://), use it as-is
+                # Otherwise, prepend https://
+                # Note: Check for "://" to distinguish between real protocols and domain:port format
+                # (urlparse treats "example.com:8080" as having scheme="example.com")
+                parsed = urlparse(endpoint)
+                if parsed.scheme and "://" in endpoint:
+                    # Endpoint is a full URL, append path
+                    url = f"{endpoint}{endpoint_path}?api-version=1"
+                else:
+                    # Endpoint is just a domain (possibly with port), prepend https://
+                    url = f"https://{endpoint}{endpoint_path}?api-version=1"
 
                 # Debug: Log endpoint being used
                 logger.info(

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/exporters/utils.py

@@ -5,6 +5,7 @@
 import os
 from collections.abc import Sequence
 from typing import Any
+from urllib.parse import urlparse
 
 from opentelemetry.sdk.trace import ReadableSpan
 from opentelemetry.trace import SpanKind, StatusCode
@@ -153,12 +154,43 @@ def get_validated_domain_override() -> str | None:
     if not domain_override:
         return None
 
-    # Basic validation: ensure domain doesn't contain protocol or path separators
-    if "://" in domain_override or "/" in domain_override:
-        logger.warning(
-            f"Invalid domain override '{domain_override}': "
-            "domain should not contain protocol (://) or path separators (/)"
-        )
+    # Validate that it's a valid URL
+    try:
+        parsed = urlparse(domain_override)
+
+        # If scheme is present and looks like a protocol (contains //)
+        # Note: We check for "://" because urlparse treats "example.com:8080" as having
+        # scheme="example.com", but this is actually a domain with port, not a protocol.
+        if parsed.scheme and "://" in domain_override:
+            # Validate it's http or https
+            if parsed.scheme not in ("http", "https"):
+                logger.warning(
+                    f"Invalid domain override '{domain_override}': "
+                    f"scheme must be http or https, got '{parsed.scheme}'"
+                )
+                return None
+            # Must have a netloc (hostname) when scheme is present
+            if not parsed.netloc:
+                logger.warning(f"Invalid domain override '{domain_override}': missing hostname")
+                return None
+        else:
+            # If no scheme with ://, it should be a domain with optional port (no path)
+            # Note: domain can contain : for port (e.g., example.com:8080)
+            # Reject malformed URLs like "http:8080" that look like protocols but aren't
+            if domain_override.startswith(("http:", "https:")) and "://" not in domain_override:
+                logger.warning(
+                    f"Invalid domain override '{domain_override}': "
+                    "malformed URL - protocol requires '://'"
+                )
+                return None
+            if "/" in domain_override:
+                logger.warning(
+                    f"Invalid domain override '{domain_override}': "
+                    "domain without protocol should not contain path separators (/)"
+                )
+                return None
+    except Exception as e:
+        logger.warning(f"Invalid domain override '{domain_override}': {e}")
         return None
 
     return domain_override

tests/observability/core/exporters/test_utils.py

@@ -1,9 +1,11 @@
-# Copyright (c) Microsoft Corporation.
-# Licensed under the MIT License.
+# Copyright (c) Microsoft. All rights reserved.
 
+import os
 import unittest
+from unittest.mock import patch
 
 from microsoft_agents_a365.observability.core.exporters.utils import (
+    get_validated_domain_override,
     truncate_span,
 )
 
@@ -64,5 +66,89 @@ def test_truncate_span_if_needed(self):
             self.assertEqual(result["attributes"][key], "TRUNCATED")
 
 
+class TestGetValidatedDomainOverride(unittest.TestCase):
+    """Unit tests for get_validated_domain_override function."""
+
+    def test_returns_none_when_env_var_not_set(self):
+        """Test that function returns None when environment variable is not set."""
+        with patch.dict(os.environ, {}, clear=True):
+            result = get_validated_domain_override()
+            self.assertIsNone(result)
+
+    def test_returns_none_when_env_var_is_empty(self):
+        """Test that function returns None when environment variable is empty."""
+        with patch.dict(os.environ, {"A365_OBSERVABILITY_DOMAIN_OVERRIDE": ""}):
+            result = get_validated_domain_override()
+            self.assertIsNone(result)
+
+    def test_returns_none_when_env_var_is_whitespace(self):
+        """Test that function returns None when environment variable is only whitespace."""
+        with patch.dict(os.environ, {"A365_OBSERVABILITY_DOMAIN_OVERRIDE": "   "}):
+            result = get_validated_domain_override()
+            self.assertIsNone(result)
+
+    def test_accepts_valid_domain(self):
+        """Test that function accepts a valid domain without protocol."""
+        with patch.dict(os.environ, {"A365_OBSERVABILITY_DOMAIN_OVERRIDE": "example.com"}):
+            result = get_validated_domain_override()
+            self.assertEqual(result, "example.com")
+
+    def test_accepts_valid_domain_with_port(self):
+        """Test that function accepts a valid domain with port."""
+        with patch.dict(os.environ, {"A365_OBSERVABILITY_DOMAIN_OVERRIDE": "example.com:8080"}):
+            result = get_validated_domain_override()
+            self.assertEqual(result, "example.com:8080")
+
+    def test_accepts_valid_https_url(self):
+        """Test that function accepts a valid URL with https protocol."""
+        with patch.dict(os.environ, {"A365_OBSERVABILITY_DOMAIN_OVERRIDE": "https://example.com"}):
+            result = get_validated_domain_override()
+            self.assertEqual(result, "https://example.com")
+
+    def test_accepts_valid_http_url(self):
+        """Test that function accepts a valid URL with http protocol."""
+        with patch.dict(os.environ, {"A365_OBSERVABILITY_DOMAIN_OVERRIDE": "http://example.com"}):
+            result = get_validated_domain_override()
+            self.assertEqual(result, "http://example.com")
+
+    def test_accepts_valid_http_url_with_port(self):
+        """Test that function accepts a valid URL with http protocol and port."""
+        with patch.dict(
+            os.environ, {"A365_OBSERVABILITY_DOMAIN_OVERRIDE": "http://localhost:8080"}
+        ):
+            result = get_validated_domain_override()
+            self.assertEqual(result, "http://localhost:8080")
+
+    def test_rejects_invalid_protocol(self):
+        """Test that function rejects URLs with invalid protocols (not http/https)."""
+        with patch.dict(os.environ, {"A365_OBSERVABILITY_DOMAIN_OVERRIDE": "ftp://example.com"}):
+            result = get_validated_domain_override()
+            self.assertIsNone(result)
+
+    def test_rejects_domain_with_path(self):
+        """Test that function rejects domain-only format with path separator."""
+        with patch.dict(os.environ, {"A365_OBSERVABILITY_DOMAIN_OVERRIDE": "example.com/path"}):
+            result = get_validated_domain_override()
+            self.assertIsNone(result)
+
+    def test_rejects_protocol_without_hostname(self):
+        """Test that function rejects URLs with protocol but no hostname."""
+        with patch.dict(os.environ, {"A365_OBSERVABILITY_DOMAIN_OVERRIDE": "https://"}):
+            result = get_validated_domain_override()
+            self.assertIsNone(result)
+
+    def test_rejects_malformed_url_http_colon(self):
+        """Test that function rejects malformed URLs like 'http:8080' (missing slashes)."""
+        with patch.dict(os.environ, {"A365_OBSERVABILITY_DOMAIN_OVERRIDE": "http:8080"}):
+            result = get_validated_domain_override()
+            self.assertIsNone(result)
+
+    def test_rejects_malformed_url_https_colon(self):
+        """Test that function rejects malformed URLs like 'https:443' (missing slashes)."""
+        with patch.dict(os.environ, {"A365_OBSERVABILITY_DOMAIN_OVERRIDE": "https:443"}):
+            result = get_validated_domain_override()
+            self.assertIsNone(result)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/observability/core/test_agent365_exporter.py

@@ -507,10 +507,115 @@ def test_export_ignores_empty_domain_override(self):
                 # Verify PowerPlatformApiDiscovery was called (override was ignored)
                 mock_discovery_class.assert_called_once_with("test")
 
+    def test_export_uses_valid_url_override_with_https(self):
+        """Test that domain override with https:// protocol is accepted and used correctly."""
+        # Arrange
+        os.environ["A365_OBSERVABILITY_DOMAIN_OVERRIDE"] = "https://override.example.com"
+
+        # Create exporter after setting environment variable
+        exporter = _Agent365Exporter(
+            token_resolver=self.mock_token_resolver, cluster_category="test"
+        )
+
+        spans = [self._create_mock_span("test_span")]
+
+        # Mock the PowerPlatformApiDiscovery class (should NOT be called since override is valid)
+        with patch(
+            "microsoft_agents_a365.observability.core.exporters.agent365_exporter.PowerPlatformApiDiscovery"
+        ) as mock_discovery_class:
+            # Mock the _post_with_retries method
+            with patch.object(exporter, "_post_with_retries", return_value=True) as mock_post:
+                # Act
+                result = exporter.export(spans)
+
+                # Assert
+                self.assertEqual(result, SpanExportResult.SUCCESS)
+                mock_post.assert_called_once()
+
+                # Verify the call arguments - should use override URL without duplicating protocol
+                args, kwargs = mock_post.call_args
+                url, body, headers = args
+
+                expected_url = "https://override.example.com/maven/agent365/agents/test-agent-456/traces?api-version=1"
+                self.assertEqual(url, expected_url)
+
+                # Verify PowerPlatformApiDiscovery was not called
+                mock_discovery_class.assert_not_called()
+
+    def test_export_uses_valid_url_override_with_http(self):
+        """Test that domain override with http:// protocol is accepted and used correctly."""
+        # Arrange
+        os.environ["A365_OBSERVABILITY_DOMAIN_OVERRIDE"] = "http://localhost:8080"
+
+        # Create exporter after setting environment variable
+        exporter = _Agent365Exporter(
+            token_resolver=self.mock_token_resolver, cluster_category="test"
+        )
+
+        spans = [self._create_mock_span("test_span")]
+
+        # Mock the PowerPlatformApiDiscovery class (should NOT be called since override is valid)
+        with patch(
+            "microsoft_agents_a365.observability.core.exporters.agent365_exporter.PowerPlatformApiDiscovery"
+        ) as mock_discovery_class:
+            # Mock the _post_with_retries method
+            with patch.object(exporter, "_post_with_retries", return_value=True) as mock_post:
+                # Act
+                result = exporter.export(spans)
+
+                # Assert
+                self.assertEqual(result, SpanExportResult.SUCCESS)
+                mock_post.assert_called_once()
+
+                # Verify the call arguments - should use override URL with http protocol
+                args, kwargs = mock_post.call_args
+                url, body, headers = args
+
+                expected_url = "http://localhost:8080/maven/agent365/agents/test-agent-456/traces?api-version=1"
+                self.assertEqual(url, expected_url)
+
+                # Verify PowerPlatformApiDiscovery was not called
+                mock_discovery_class.assert_not_called()
+
+    def test_export_uses_valid_domain_override_with_port(self):
+        """Test that domain override with port (no protocol) is accepted and https:// is prepended."""
+        # Arrange
+        os.environ["A365_OBSERVABILITY_DOMAIN_OVERRIDE"] = "example.com:8080"
+
+        # Create exporter after setting environment variable
+        exporter = _Agent365Exporter(
+            token_resolver=self.mock_token_resolver, cluster_category="test"
+        )
+
+        spans = [self._create_mock_span("test_span")]
+
+        # Mock the PowerPlatformApiDiscovery class (should NOT be called since override is valid)
+        with patch(
+            "microsoft_agents_a365.observability.core.exporters.agent365_exporter.PowerPlatformApiDiscovery"
+        ) as mock_discovery_class:
+            # Mock the _post_with_retries method
+            with patch.object(exporter, "_post_with_retries", return_value=True) as mock_post:
+                # Act
+                result = exporter.export(spans)
+
+                # Assert
+                self.assertEqual(result, SpanExportResult.SUCCESS)
+                mock_post.assert_called_once()
+
+                # Verify the call arguments - should prepend https:// to domain with port
+                args, kwargs = mock_post.call_args
+                url, body, headers = args
+
+                expected_url = "https://example.com:8080/maven/agent365/agents/test-agent-456/traces?api-version=1"
+                self.assertEqual(url, expected_url)
+
+                # Verify PowerPlatformApiDiscovery was not called
+                mock_discovery_class.assert_not_called()
+
     def test_export_ignores_invalid_domain_with_protocol(self):
-        """Test that domain override containing protocol is ignored."""
+        """Test that domain override with invalid protocol is ignored."""
         # Arrange
-        os.environ["A365_OBSERVABILITY_DOMAIN_OVERRIDE"] = "https://invalid.example.com"
+        os.environ["A365_OBSERVABILITY_DOMAIN_OVERRIDE"] = "ftp://invalid.example.com"
 
         # Create exporter after setting environment variable
         exporter = _Agent365Exporter(