diff --git a/backend/apps/agent_api/tests.py b/backend/apps/agent_api/tests.py new file mode 100644 index 0000000..05035f8 --- /dev/null +++ b/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") diff --git a/backend/apps/agent_api/views.py b/backend/apps/agent_api/views.py index 886ebb9..85654d6 100644 --- a/backend/apps/agent_api/views.py +++ b/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 diff --git a/backend/integrations/siem/backends.py b/backend/integrations/siem/backends.py index b15b172..daffd1c 100644 --- a/backend/integrations/siem/backends.py +++ b/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", ) diff --git a/backend/integrations/siem/query_builders.py b/backend/integrations/siem/query_builders.py index b38267e..a23e936 100644 --- a/backend/integrations/siem/query_builders.py +++ b/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 diff --git a/backend/integrations/siem/tests.py b/backend/integrations/siem/tests.py new file mode 100644 index 0000000..4bb1089 --- /dev/null +++ b/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)