Validate Splunk realm before building ingest URLs

src/splunk_otel/distro.py

@@ -13,6 +13,7 @@
 #  limitations under the License.
 
 import logging
+import re
 
 from opentelemetry.instrumentation.distro import BaseDistro
 from opentelemetry.instrumentation.environment_variables import OTEL_PYTHON_DISABLED_INSTRUMENTATIONS
@@ -53,6 +54,7 @@
 _X_SF_TOKEN = "x-sf-token"  # noqa S105
 _DISABLED_INSTRUMENTATIONS_WILDCARD = "*"
 _LOGGING_INSTRUMENTATION_NAME = "logging"
+_REALM_RE = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?$")
 
 _pylogger = logging.getLogger(__name__)
 
@@ -99,8 +101,12 @@ def set_resource_attributes(self):
         self.env.list_append(OTEL_RESOURCE_ATTRIBUTES, f"telemetry.distro.version={version}")
 
     def handle_realm(self):
-        realm = self.env.getval(SPLUNK_REALM)
+        realm = self.env.getval(SPLUNK_REALM).strip()
         if len(realm):
+            if not _REALM_RE.fullmatch(realm):
+                _pylogger.warning("Ignoring invalid SPLUNK_REALM value: %r", realm)
+                return
+
             ingest_url = f"https://ingest.{realm}.observability.splunkcloud.com"
             self.env.setdefault(
                 OTEL_EXPORTER_OTLP_TRACES_ENDPOINT,

tests/test_distro.py

@@ -130,6 +130,37 @@ def test_realm():
     assert env_store["OTEL_EXPORTER_OTLP_PROTOCOL"] == "http/protobuf"
 
 
+def test_realm_strips_whitespace():
+    env_store = {"SPLUNK_REALM": " us2 "}
+    configure_distro(env_store)
+    assert (
+        env_store["OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"]
+        == "https://ingest.us2.observability.splunkcloud.com/v2/trace/otlp"
+    )
+
+
+@pytest.mark.parametrize(
+    "realm",
+    [
+        "attacker.test",
+        "attacker.test/anything",
+        "attacker.test:443/foo",
+        "us2/anything",
+        "us2:443",
+        "-us2",
+        "us2-",
+    ],
+)
+def test_invalid_realm_is_ignored(caplog, realm):
+    env_store = {"SPLUNK_REALM": realm}
+    with caplog.at_level(logging.WARNING):
+        configure_distro(env_store)
+    assert "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT" not in env_store
+    assert "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT" not in env_store
+    assert "OTEL_EXPORTER_OTLP_PROTOCOL" not in env_store
+    assert "Ignoring invalid SPLUNK_REALM value" in caplog.text
+
+
 def test_callgraphs_propagator_disabled_by_default():
     env_store = {}
     configure_distro(env_store)