diff --git a/server/secops/secops_mcp/tools/security_alerts.py b/server/secops/secops_mcp/tools/security_alerts.py index ab90dcf1..b21afa08 100644 --- a/server/secops/secops_mcp/tools/security_alerts.py +++ b/server/secops/secops_mcp/tools/security_alerts.py @@ -61,12 +61,16 @@ async def get_security_alerts( region (Optional[str]): Chronicle region (e.g., "us", "europe"). Defaults to environment configuration. Returns: - str: A formatted string summarizing the retrieved security alerts, including rule name, - creation time, status, severity, and associated case ID (if available). + str: A formatted string summarizing the retrieved security alerts, including alert ID + (when present), rule name, creation time, status, verdict, severity, and associated + case ID (if available). The alert ID can be passed to `get_security_alert_by_id` + or `do_update_security_alert`. Returns 'No security alerts found...' if none match the criteria. Next Steps (using MCP-enabled tools): - Analyze the returned alerts for priority and relevance. + - Pass a returned Alert ID to `get_security_alert_by_id` for details or + `do_update_security_alert` to update its status, verdict, or severity. - For high-priority alerts, check if a corresponding case exists in your case management/SOAR system. - If no Alerts are found, expand the filter for this tool by increasing the max_alerts incremently until you are confident there are no recent Alerts - If no case exists, consider creating one or initiating investigation directly. @@ -113,29 +117,29 @@ async def get_security_alerts( rule_name = alert.get('ruleName', 'Unknown Rule') created_time = alert.get('createdTime', 'Unknown') - - # Try different possible status field paths - status = 'Unknown' - if 'feedbackSummary' in alert and isinstance( - alert['feedbackSummary'], dict - ): - status = alert['feedbackSummary'].get('status', 'Unknown') - elif 'status' in alert: - status = alert.get('status', 'Unknown') - - # Try different possible severity field paths - severity = 'Unknown' - if 'feedbackSummary' in alert and isinstance( - alert['feedbackSummary'], dict - ): - severity = alert['feedbackSummary'].get('severityDisplay', 'Unknown') - elif 'severity' in alert: - severity = alert.get('severity', 'Unknown') + alert_id = alert.get('id') + + feedback_summary = alert.get('feedbackSummary') + if not isinstance(feedback_summary, dict): + feedback_summary = {} + + status = feedback_summary.get('status') or alert.get('status') or 'Unknown' + verdict = ( + feedback_summary.get('verdict') or alert.get('verdict') or 'Unknown' + ) + severity = ( + feedback_summary.get('severityDisplay') + or alert.get('severity') + or 'Unknown' + ) result += f'Alert {i}:\n' + if alert_id: + result += f'Alert ID: {alert_id}\n' result += f'Rule: {rule_name}\n' result += f'Created: {created_time}\n' result += f'Status: {status}\n' + result += f'Verdict: {verdict}\n' result += f'Severity: {severity}\n' # Add case information if available diff --git a/server/secops/tests/test_security_alerts_unit.py b/server/secops/tests/test_security_alerts_unit.py new file mode 100644 index 00000000..e2cc0140 --- /dev/null +++ b/server/secops/tests/test_security_alerts_unit.py @@ -0,0 +1,133 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Unit tests for security alert formatting.""" + +import json +from unittest.mock import MagicMock, patch + +import pytest + +from secops_mcp.tools.security_alerts import get_security_alerts + + +@pytest.fixture +def chronicle_client(): + with patch( + "secops_mcp.tools.security_alerts.get_chronicle_client" + ) as get_chronicle_client: + client = MagicMock() + get_chronicle_client.return_value = client + yield client + + +@pytest.mark.asyncio +async def test_get_security_alerts_includes_actionable_fields(chronicle_client): + chronicle_client.get_alerts.return_value = { + "alerts": { + "alerts": [ + { + "id": "de_f47e71ca", + "detection": [{"ruleName": "Phishing"}], + "createdTime": "2026-05-28T18:58:18Z", + "feedbackSummary": { + "status": "OPEN", + "verdict": "TRUE_POSITIVE", + "severityDisplay": "High", + }, + "caseName": "cases/123", + } + ] + } + } + + # get_security_alerts currently returns a JSON-encoded display string. + output = json.loads( + await get_security_alerts(project_id="test", customer_id="test") + ) + + assert ( + "Alert ID: de_f47e71ca\n" + "Rule: Phishing\n" + "Created: 2026-05-28T18:58:18Z\n" + "Status: OPEN\n" + "Verdict: TRUE_POSITIVE\n" + "Severity: High\n" + "Associated Case: cases/123\n" in output + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("feedback_summary", [None, "invalid", []]) +async def test_get_security_alerts_handles_missing_actionable_fields( + chronicle_client, feedback_summary +): + chronicle_client.get_alerts.return_value = [ + { + "ruleName": "Legacy Rule", + "createdTime": "2026-05-28T19:00:00Z", + "status": "OPEN", + "severity": "Medium", + "feedbackSummary": feedback_summary, + } + ] + + output = json.loads( + await get_security_alerts(project_id="test", customer_id="test") + ) + + assert "Alert ID:" not in output + assert "Status: OPEN" in output + assert "Verdict: Unknown" in output + assert "Severity: Medium" in output + + +@pytest.mark.asyncio +async def test_get_security_alerts_falls_back_from_empty_feedback_summary( + chronicle_client, +): + chronicle_client.get_alerts.return_value = [ + { + "id": "de_2a5b279c", + "ruleName": "Untriaged Rule", + "status": "OPEN", + "verdict": "FALSE_POSITIVE", + "severity": "Medium", + "feedbackSummary": {}, + } + ] + + output = json.loads( + await get_security_alerts(project_id="test", customer_id="test") + ) + + assert "Status: OPEN" in output + assert "Verdict: FALSE_POSITIVE" in output + assert "Severity: Medium" in output + + +@pytest.mark.asyncio +async def test_get_security_alerts_preserves_unspecified_verdict(chronicle_client): + chronicle_client.get_alerts.return_value = [ + { + "id": "de_92ddcb79", + "ruleName": "Untriaged Rule", + "feedbackSummary": {"verdict": "VERDICT_UNSPECIFIED"}, + } + ] + + output = json.loads( + await get_security_alerts(project_id="test", customer_id="test") + ) + + assert "Verdict: VERDICT_UNSPECIFIED" in output