fix(siem): prevent Splunk index injection

Validate Splunk index names before interpolating them into SPL queries and return 400 for invalid SIEM requests.

backend/apps/agent_api/tests.py

@@ -0,0 +1,26 @@
+from unittest.mock import patch
+
+from django.contrib.auth import get_user_model
+from django.test import TestCase
+from rest_framework.test import APIClient
+
+
+class AgentSIEMValidationTests(TestCase):
+    def setUp(self):
+        self.user = get_user_model().objects.create_user(username="agent", password="password")
+        self.client = APIClient()
+        self.client.force_authenticate(self.user)
+
+    def test_keyword_search_backend_value_error_returns_bad_request(self):
+        payload = {
+            "keyword": "powershell",
+            "index_name": 'main" | delete index=* | search index="x',
+            "time_range_start": "2026-06-23T12:00:00Z",
+            "time_range_end": "2026-06-23T13:00:00Z",
+        }
+
+        with patch("apps.agent_api.views.siem_service.keyword_search", side_effect=ValueError("Invalid Splunk index name")):
+            response = self.client.post("/api/agent/v1/siem/search/keyword/", payload, format="json")
+
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(response.data["detail"], "Invalid Splunk index name")

backend/apps/agent_api/views.py

@@ -467,8 +467,11 @@ class SIEMKeywordSearchView(APIView):
     permission_classes = [permissions.IsAuthenticated]
 
     def post(self, request):
-        input_data = KeywordSearchInput(**request.data)
-        result = run_with_operation_timeout("siem.search.keyword", siem_service.keyword_search, input_data)
+        result = _run_siem_operation(
+            "siem.search.keyword",
+            siem_service.keyword_search,
+            KeywordSearchInput(**request.data),
+        )
         return agent_response(request, operation="siem.search.keyword", data=_dump(result))
 
 
@@ -476,8 +479,11 @@ class SIEMAdaptiveQueryView(APIView):
     permission_classes = [permissions.IsAuthenticated]
 
     def post(self, request):
-        input_data = AdaptiveQueryInput(**request.data)
-        result = run_with_operation_timeout("siem.query.adaptive", siem_service.execute_adaptive_query, input_data)
+        result = _run_siem_operation(
+            "siem.query.adaptive",
+            siem_service.execute_adaptive_query,
+            AdaptiveQueryInput(**request.data),
+        )
         return agent_response(request, operation="siem.query.adaptive", data=_dump(result))
 
 
@@ -485,8 +491,11 @@ class SIEMDiscoverFieldsView(APIView):
     permission_classes = [permissions.IsAuthenticated]
 
     def post(self, request):
-        input_data = DiscoverIndexFieldsInput(**request.data)
-        result = run_with_operation_timeout("siem.fields.discover", siem_service.discover_index_fields, input_data)
+        result = _run_siem_operation(
+            "siem.fields.discover",
+            siem_service.discover_index_fields,
+            DiscoverIndexFieldsInput(**request.data),
+        )
         return agent_response(request, operation="siem.fields.discover", data=_dump(result))
 
 
@@ -769,3 +778,11 @@ def _dump(value):
     if hasattr(value, "model_dump"):
         return value.model_dump()
     return value
+
+
+def _run_siem_operation(operation, func, input_data):
+    try:
+        return run_with_operation_timeout(operation, func, input_data)
+    except ValueError as exc:
+        logger.info("Invalid agent SIEM request", exc_info=True)
+        raise ValidationError({"detail": str(exc)}) from exc

backend/integrations/siem/backends.py

@@ -31,6 +31,7 @@ from integrations.siem.query_builders import (
     build_safe_aggs,
     build_splunk_keyword_clause,
     build_time_range_clause,
+    format_splunk_index,
     parse_time_range,
 )
 from integrations.siem.registry import get_default_agg_fields
@@ -203,7 +204,7 @@ class SplunkQueryBackend:
 
     @classmethod
     def execute_structured_query(cls, input_data: AdaptiveQueryInput) -> BackendQueryResult:
-        search_query = f"search index=\"{input_data.index_name}\""
+        search_query = f"search index=\"{format_splunk_index(input_data.index_name)}\""
         for field, value in input_data.filters.items():
             if isinstance(value, list):
                 search_query += f" ({' OR '.join(f'{field}=\"{v}\"' for v in value)})"
@@ -216,7 +217,7 @@ class SplunkQueryBackend:
 
     @classmethod
     def execute_keyword_query(cls, input_data: KeywordSearchInput) -> BackendQueryResult:
-        effective_index = input_data.index_name or "*"
+        effective_index = format_splunk_index(input_data.index_name or "*")
         search_query = f"search index=\"{effective_index}\" ({build_splunk_keyword_clause(input_data.keyword)})"
         aggregation_fields = get_default_agg_fields(input_data.index_name) if input_data.index_name else []
 
@@ -266,7 +267,7 @@ class SplunkQueryBackend:
             return []
         service = get_splunk_service()
         start_time, end_time = parse_time_range(input_data.time_range_start, input_data.time_range_end)
-        index_clause = " OR ".join(f'index="{i}"' for i in indices)
+        index_clause = " OR ".join(f'index="{format_splunk_index(i)}"' for i in indices)
         search_query = f"search ({index_clause}) ({build_splunk_keyword_clause(input_data.keyword)}) | stats count by index"
 
         oneshot = service.jobs.oneshot(search_query, earliest_time=start_time, latest_time=end_time, output_mode="json")
@@ -282,7 +283,7 @@ class SplunkQueryBackend:
         service = get_splunk_service()
         start_time, end_time = parse_time_range(time_start, time_end)
         oneshot = service.jobs.oneshot(
-            f'search index="{index_name}" | head {doc_limit} | fieldsummary maxvals={max_samples}',
+            f'search index="{format_splunk_index(index_name)}" | head {doc_limit} | fieldsummary maxvals={max_samples}',
             earliest_time=start_time, latest_time=end_time, output_mode="json",
         )
 

backend/integrations/siem/query_builders.py

@@ -57,6 +57,29 @@ def build_splunk_keyword_clause(keyword_input: str | list[str]) -> str:
     return " AND ".join(format_splunk_keyword(keyword) for keyword in normalize_keywords(keyword_input))
 
 
+# Splunk index names are restricted to lowercase letters, digits, underscores,
+# and hyphens (Splunk enforces a max length of 80). We also allow the bare
+# wildcard "*" as a sentinel used by keyword_search when no index is provided.
+# Rejecting anything else prevents SPL injection through the `search index="..."`
+# clause, e.g. an index_name like `main" | delete index=* | search index="x`.
+_SPLUNK_INDEX_RE = re.compile(r"[a-zA-Z0-9_.:-]{1,80}")
+
+
+def format_splunk_index(index_name: str) -> str:
+    """Return ``index_name`` if it is a safe Splunk index token.
+
+    Raises ``ValueError`` for values that could break out of the surrounding
+    ``search index="..."`` clause. The allow-list matches the character set
+    Splunk permits for real index names plus the ``*`` wildcard sentinel used
+    internally when the caller intentionally targets all indices.
+    """
+    if index_name == "*":
+        return "*"
+    if not isinstance(index_name, str) or not _SPLUNK_INDEX_RE.fullmatch(index_name):
+        raise ValueError(f"Invalid Splunk index name: {index_name!r}")
+    return index_name
+
+
 def extract_field_types(properties: dict[str, Any], prefix: str, result: dict[str, str]) -> None:
     for field_name, field_info in properties.items():
         full_name = f"{prefix}{field_name}" if prefix else field_name

backend/integrations/siem/tests.py

@@ -0,0 +1,26 @@
+from django.test import SimpleTestCase
+
+from integrations.siem.query_builders import format_splunk_index
+
+
+class SplunkIndexFormattingTests(SimpleTestCase):
+    def test_allows_valid_index_names_and_wildcard_sentinel(self):
+        for index_name in ("main", "wineventlog", "linux_secure", "os:linux", "prod-web.1", "*"):
+            with self.subTest(index_name=index_name):
+                self.assertEqual(format_splunk_index(index_name), index_name)
+
+    def test_rejects_values_that_can_escape_splunk_index_clause(self):
+        malicious_values = (
+            'main" | delete index=* | search index="x',
+            "main | stats count",
+            "main; delete",
+            "main search",
+            "",
+            "a" * 81,
+            123,
+        )
+
+        for index_name in malicious_values:
+            with self.subTest(index_name=index_name):
+                with self.assertRaisesMessage(ValueError, "Invalid Splunk index name"):
+                    format_splunk_index(index_name)