diff --git a/server/secops/secops_mcp/tools/security_rules.py b/server/secops/secops_mcp/tools/security_rules.py index c3ffe237..f410b644 100644 --- a/server/secops/secops_mcp/tools/security_rules.py +++ b/server/secops/secops_mcp/tools/security_rules.py @@ -233,6 +233,9 @@ async def get_rule_detections( project_id: Optional[str] = None, customer_id: Optional[str] = None, region: Optional[str] = None, + start_time: Optional[str] = None, + end_time: Optional[str] = None, + list_basis: Optional[str] = None, ) -> Dict[str, Any]: """Retrieves historical detections generated by a specific Chronicle SIEM rule. @@ -266,6 +269,10 @@ async def get_rule_detections( project_id (Optional[str]): Google Cloud project ID. Defaults to environment configuration. customer_id (Optional[str]): Chronicle customer ID. Defaults to environment configuration. region (Optional[str]): Chronicle region (e.g., "us", "europe"). Defaults to environment configuration. + start_time (Optional[str]): Start of the detection time range in ISO 8601 format. + end_time (Optional[str]): End of the detection time range in ISO 8601 format. + list_basis (Optional[str]): Basis for listing detections. Valid values: + "LIST_BASIS_UNSPECIFIED", "CREATED_TIME", "DETECTION_TIME". Returns: Dict[str, Any]: A dictionary containing the list of detections (under a 'detections' key, typically) @@ -283,6 +290,8 @@ async def get_rule_detections( - **Visualize Detections:** Export detection data and use data visualization tools to identify trends or patterns over time. """ try: + from datetime import datetime + chronicle = get_chronicle_client(project_id, customer_id, region) if ( @@ -308,14 +317,31 @@ async def get_rule_detections( f"alert_state must be one of {valid_alert_states}, got {alert_state}" ) + start_dt = ( + datetime.fromisoformat(start_time.replace("Z", "+00:00")) + if start_time + else None + ) + end_dt = ( + datetime.fromisoformat(end_time.replace("Z", "+00:00")) + if end_time + else None + ) + detections_response = chronicle.list_detections( - rule_id, alert_state, page_size, page_token + rule_id, + start_time=start_dt, + end_time=end_dt, + list_basis=list_basis, + alert_state=alert_state, + page_size=page_size, + page_token=page_token, ) return detections_response except ( ValueError - ) as ve: # Catch specific ValueError from alert_state validation + ) as ve: logger.error( f"Validation error getting rule detections for rule {rule_id}: {str(ve)}", exc_info=True, diff --git a/server/secops/tests/test_secops_tools_unit.py b/server/secops/tests/test_secops_tools_unit.py index 8add9f11..68027e9e 100644 --- a/server/secops/tests/test_secops_tools_unit.py +++ b/server/secops/tests/test_secops_tools_unit.py @@ -45,6 +45,7 @@ def wrapper(func): from secops_mcp.tools.search import search_udm from secops_mcp.tools.udm_search import export_udm_search_csv from secops_mcp.tools.security_events import search_security_events +from secops_mcp.tools.security_rules import get_rule_detections @pytest.fixture def mock_chronicle_client(): @@ -59,9 +60,65 @@ def mock_chronicle_client(): def mock_get_client(mock_chronicle_client): with patch('secops_mcp.tools.search.get_chronicle_client', return_value=mock_chronicle_client) as m1, \ patch('secops_mcp.tools.udm_search.get_chronicle_client', return_value=mock_chronicle_client) as m2, \ - patch('secops_mcp.tools.security_events.get_chronicle_client', return_value=mock_chronicle_client) as m3: + patch('secops_mcp.tools.security_events.get_chronicle_client', return_value=mock_chronicle_client) as m3, \ + patch('secops_mcp.tools.security_rules.get_chronicle_client', return_value=mock_chronicle_client): yield mock_chronicle_client + +@pytest.mark.asyncio +async def test_get_rule_detections_forwards_filters(mock_get_client): + """Test detection filters use the Chronicle SDK parameter names.""" + expected = {"detections": []} + + def list_detections( + rule_id, + start_time=None, + end_time=None, + list_basis=None, + alert_state=None, + page_size=None, + page_token=None, + ): + if start_time: + start_time.strftime("%Y-%m-%dT%H:%M:%SZ") + return expected + + mock_get_client.list_detections.side_effect = list_detections + + result = await get_rule_detections( + "ru_test", + alert_state="ALERTING", + page_size=25, + page_token="next-page", + ) + + assert result == expected + call_args = mock_get_client.list_detections.call_args + assert call_args.args == ("ru_test",) + assert call_args.kwargs["alert_state"] == "ALERTING" + assert call_args.kwargs["page_size"] == 25 + assert call_args.kwargs["page_token"] == "next-page" + + +@pytest.mark.asyncio +async def test_get_rule_detections_forwards_time_range(mock_get_client): + """Test detection time range parameters are converted and forwarded.""" + await get_rule_detections( + "ru_test", + start_time="2025-01-20T00:00:00Z", + end_time="2025-01-27T23:59:59Z", + list_basis="DETECTION_TIME", + ) + + call_args = mock_get_client.list_detections.call_args + assert call_args.kwargs["start_time"] == datetime( + 2025, 1, 20, tzinfo=timezone.utc + ) + assert call_args.kwargs["end_time"] == datetime( + 2025, 1, 27, 23, 59, 59, tzinfo=timezone.utc + ) + assert call_args.kwargs["list_basis"] == "DETECTION_TIME" + @pytest.mark.asyncio async def test_search_udm_with_start_time(mock_get_client): """Test search_udm with explicit start_time."""