Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 28 additions & 2 deletions server/secops/secops_mcp/tools/security_rules.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down Expand Up @@ -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)
Expand All @@ -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 (
Expand All @@ -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
)
Comment on lines +320 to +329

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I know you are following precedent in this project, so there is no blame, but this has some issues, which I've described in #291. I think I will fast follow with a fix of all occurrences of this rather than block your PR.


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,
Expand Down
59 changes: 58 additions & 1 deletion server/secops/tests/test_secops_tools_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand All @@ -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."""
Expand Down