diff --git a/docs/servers/secops_mcp.md b/docs/servers/secops_mcp.md index df99f877..7577987c 100644 --- a/docs/servers/secops_mcp.md +++ b/docs/servers/secops_mcp.md @@ -829,6 +829,39 @@ The service account or user credentials need the following Chronicle roles: - `region` (optional): Chronicle region (defaults to environment config or 'us'). - **Returns:** Dictionary containing investigation associations grouped by detection ID, with verdict and confidence information. +### Case Management + +- **`list_case_close_definitions(page_size=50, page_token=None, filter=None, order_by=None, project_id=None, customer_id=None, region=None)`** + - **Description:** Retrieves configured case close definitions which pair root causes with valid close reasons (e.g., `MALICIOUS`, `NOT_MALICIOUS`, `MAINTENANCE`, `INCONCLUSIVE`). Essential for discovering valid root causes and reasons required to close a case or alert. + - **Parameters:** + - `page_size` (optional): Number of definitions to return per page (default: 50). + - `page_token` (optional): Token for pagination from previous response. + - `filter` (optional): CEL or standard filter expression to restrict definitions (e.g., `close_reason='MALICIOUS'`). + - `order_by` (optional): Field expression to order results (e.g., `root_cause desc`). + - `project_id` (optional): Google Cloud project ID (defaults to environment config). + - `customer_id` (optional): Chronicle customer ID (defaults to environment config). + - `region` (optional): Chronicle region (defaults to environment config or 'us'). + - **Returns:** Dictionary containing list of `caseCloseDefinitions` and pagination tokens, or an `error` message upon failure. + - **Return Example:** + ```json + { + "caseCloseDefinitions": [ + { + "name": "projects/123/locations/us/instances/456/caseCloseDefinitions/def-1", + "closeReason": "MALICIOUS", + "rootCause": "Phishing credential harvest" + }, + { + "name": "projects/123/locations/us/instances/456/caseCloseDefinitions/def-2", + "closeReason": "NOT_MALICIOUS", + "rootCause": "Authorized Security Test" + } + ], + "nextPageToken": "", + "totalSize": 2 + } + ``` + ## Usage Examples ### Example 1: Natural Language Security Event Search diff --git a/docs/servers/secops_soar_mcp.md b/docs/servers/secops_soar_mcp.md index 113a16d0..23247883 100644 --- a/docs/servers/secops_soar_mcp.md +++ b/docs/servers/secops_soar_mcp.md @@ -309,6 +309,28 @@ These tools are always available. } ``` +- **`list_case_close_root_causes()`** + - **Description:** Lists configured case close root causes and their associated close reasons from the SOAR platform (configured under Settings > Case Close Root Causes). Call this tool prior to `close_case` to determine valid (reason, root_cause) pairings accepted by the SOAR tenant. + - **Parameters:** None. + - **Returns:** A dictionary containing a list of `root_causes` with `id`, `root_cause`, and `close_reason`. + - **Return Example:** + ```json + { + "root_causes": [ + { + "id": 1, + "root_cause": "Phishing email with credential harvester", + "close_reason": "Malicious" + }, + { + "id": 2, + "root_cause": "Authorized penetration testing", + "close_reason": "Maintenance" + } + ] + } + ``` + - **`close_case(case_id, root_cause, comment, reason, tags=None)`** - **Description:** Closes a specific case by setting its root cause, close reason, and a closing comment. Marks the end of the investigation lifecycle. - **Parameters:** diff --git a/server/secops-soar/secops_soar_mcp/case_management.py b/server/secops-soar/secops_soar_mcp/case_management.py index cb7444c6..c24b1bd8 100644 --- a/server/secops-soar/secops_soar_mcp/case_management.py +++ b/server/secops-soar/secops_soar_mcp/case_management.py @@ -445,6 +445,61 @@ async def update_case_description( req={"CaseId": case_id, "Description": description}, ) + @mcp.tool() + async def list_case_close_root_causes() -> dict: + """List configured case close root causes and reasons from the SOAR platform. + + Retrieves the tenant's configured case close root causes and their associated + close reasons (e.g., Malicious, NotMalicious, Maintenance, Inconclusive). + This tool should be called prior to `close_case` to determine valid + (reason, root_cause) pairings accepted by the SOAR instance. + + Returns: + dict: A dictionary containing 'root_causes', a list of objects with: + - id: The unique identifier of the root cause record + - root_cause: The configured root cause name/string + - close_reason: The mapped close reason ('Malicious', 'NotMalicious', + 'Maintenance', or 'Inconclusive') + """ + response = await bindings.http_client.get( + Endpoints.GET_ROOT_CAUSE_CLOSE_RECORDS + ) + if response is None: + return {"error": "Failed to retrieve case close root causes from SOAR API."} + + if not isinstance(response, list): + if isinstance(response, dict) and "error" in response: + return response + return { + "error": ( + "Failed to retrieve case close root causes from SOAR API: " + f"unexpected response {response}" + ) + } + + reason_map = { + 0: "Malicious", + 1: "NotMalicious", + 2: "Maintenance", + 3: "Inconclusive", + } + + root_causes = [] + for record in response: + close_reason_num = record.get("forCloseReason") + close_reason_str = reason_map.get( + close_reason_num, str(close_reason_num) + ) + root_causes.append( + { + "id": record.get("id"), + "root_cause": record.get("rootCause"), + "close_reason": close_reason_str, + } + ) + + return {"root_causes": root_causes} + @mcp.tool() async def close_case( case_id: Annotated[str, Field(..., description="The ID of the case.")], diff --git a/server/secops-soar/secops_soar_mcp/utils/consts.py b/server/secops-soar/secops_soar_mcp/utils/consts.py index ae0df312..42ec0187 100644 --- a/server/secops-soar/secops_soar_mcp/utils/consts.py +++ b/server/secops-soar/secops_soar_mcp/utils/consts.py @@ -33,6 +33,7 @@ class Endpoints: FETCH_FULL_UNIQUE_ENTITY = "/api/external/v1/entities/GetEntityData" SEARCH_ENTITY = "/api/external/v1.0/entity-search/entities" GET_SCOPES = "/api/external/v1/settings/GetScopes" + GET_ROOT_CAUSE_CLOSE_RECORDS = "/api/external/v1/settings/GetRootCauseCloseRecords" GET_ALERT_GROUP_IDENTIFIERS_ENTITIES = ( "/api/external/v1/case-overview/GetAlertsEntities" ) diff --git a/server/secops-soar/tests/unit/test_case_management.py b/server/secops-soar/tests/unit/test_case_management.py new file mode 100644 index 00000000..ecdc8459 --- /dev/null +++ b/server/secops-soar/tests/unit/test_case_management.py @@ -0,0 +1,110 @@ +# 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 +# +# https://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. + +from unittest.mock import AsyncMock, patch + +import pytest +from mcp.server.fastmcp import FastMCP +from secops_soar_mcp import bindings +from secops_soar_mcp.case_management import register_tools +from secops_soar_mcp.utils.consts import Endpoints + + +@pytest.fixture +def mock_mcp(): + mcp = FastMCP("test-soar") + register_tools(mcp) + return mcp + + +@pytest.mark.asyncio +async def test_list_case_close_root_causes_success(mock_mcp): + """Test list_case_close_root_causes correctly calls endpoint and formats results.""" + tool = mock_mcp._tool_manager.get_tool("list_case_close_root_causes") + assert tool is not None, "list_case_close_root_causes tool should be registered" + + mock_records = [ + {"id": 1, "rootCause": "Phishing email", "forCloseReason": 0}, + {"id": 2, "rootCause": "False Positive - Scanner", "forCloseReason": 1}, + {"id": 3, "rootCause": "Scheduled Drill", "forCloseReason": 2}, + {"id": 4, "rootCause": "Insufficient Logs", "forCloseReason": 3}, + ] + + mock_client = AsyncMock() + mock_client.get.return_value = mock_records + with patch.object(bindings, "http_client", mock_client): + result = await tool.fn() + + mock_client.get.assert_awaited_once_with(Endpoints.GET_ROOT_CAUSE_CLOSE_RECORDS) + assert isinstance(result, dict) + assert "root_causes" in result + records = result["root_causes"] + assert len(records) == 4 + assert records[0] == { + "id": 1, + "root_cause": "Phishing email", + "close_reason": "Malicious", + } + assert records[1] == { + "id": 2, + "root_cause": "False Positive - Scanner", + "close_reason": "NotMalicious", + } + assert records[2] == { + "id": 3, + "root_cause": "Scheduled Drill", + "close_reason": "Maintenance", + } + assert records[3] == { + "id": 4, + "root_cause": "Insufficient Logs", + "close_reason": "Inconclusive", + } + + +@pytest.mark.asyncio +async def test_list_case_close_root_causes_handles_none(mock_mcp): + """Test list_case_close_root_causes handles None/error from http_client.""" + tool = mock_mcp._tool_manager.get_tool("list_case_close_root_causes") + assert tool is not None + + mock_client = AsyncMock() + mock_client.get.return_value = None + with patch.object(bindings, "http_client", mock_client): + result = await tool.fn() + + assert isinstance(result, dict) + assert "error" in result + + +@pytest.mark.asyncio +async def test_list_case_close_root_causes_handles_non_list_response(mock_mcp): + """Test list_case_close_root_causes handles non-list/error dictionary from http_client.""" + tool = mock_mcp._tool_manager.get_tool("list_case_close_root_causes") + assert tool is not None + + mock_client = AsyncMock() + mock_client.get.return_value = {"error": "Unauthorized access"} + with patch.object(bindings, "http_client", mock_client): + result = await tool.fn() + + assert isinstance(result, dict) + assert result == {"error": "Unauthorized access"} + + mock_client.get.return_value = {"message": "Unexpected error format"} + with patch.object(bindings, "http_client", mock_client): + result = await tool.fn() + + assert isinstance(result, dict) + assert "error" in result diff --git a/server/secops/secops_mcp/tools/__init__.py b/server/secops/secops_mcp/tools/__init__.py index 1b16e316..c7098673 100644 --- a/server/secops/secops_mcp/tools/__init__.py +++ b/server/secops/secops_mcp/tools/__init__.py @@ -13,6 +13,7 @@ # limitations under the License. """Security Operations MCP tools package.""" +from .case_close_definitions import * from .curated_rules_management import * from .data_table_management import * from .entity_lookup import * diff --git a/server/secops/secops_mcp/tools/case_close_definitions.py b/server/secops/secops_mcp/tools/case_close_definitions.py new file mode 100644 index 00000000..6ec6bff4 --- /dev/null +++ b/server/secops/secops_mcp/tools/case_close_definitions.py @@ -0,0 +1,94 @@ +# 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. +"""Security Operations MCP tools for case close definitions.""" + +import logging +from typing import Any, Dict, Optional + +from secops.chronicle.case import APIVersion, chronicle_paginated_request + +from secops_mcp.server import get_chronicle_client, server + +logger = logging.getLogger("secops-mcp") + + +@server.tool() +async def list_case_close_definitions( + page_size: int = 50, + page_token: Optional[str] = None, + filter: Optional[str] = None, + order_by: Optional[str] = None, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """List case close definitions (root causes and close reasons) in Chronicle. + + Retrieves configured case close definitions which pair root causes with + valid close reasons (e.g. MALICIOUS, NOT_MALICIOUS, MAINTENANCE, INCONCLUSIVE). + This tool allows security analysts and automated workflows to discover the valid + root causes required to close a case or alert. + + **Workflow Integration:** + - Use prior to closing a case or alert to discover allowed root causes and close reasons + - Discover tenant-specific root cause classifications and definitions + - Filter definitions by reason or query string + + **Use Cases:** + - "What are the allowed root causes for closing a case as MALICIOUS?" + - "List all case close definitions" + - "Find case close root causes for false positives" + + Args: + page_size (int): Number of definitions to return per page. Defaults to 50. + page_token (Optional[str]): Token for pagination. + filter (Optional[str]): CEL or standard filter string to restrict definitions. + order_by (Optional[str]): Field expression to order results. + project_id (Optional[str]): Google Cloud project ID. Defaults to environment config. + customer_id (Optional[str]): Chronicle customer ID. Defaults to environment config. + region (Optional[str]): Chronicle region (e.g., "us", "europe"). Defaults to environment config. + + Returns: + Dict[str, Any]: Dictionary containing list of `caseCloseDefinitions` and pagination tokens, + or an `error` message upon failure. + """ + try: + chronicle = get_chronicle_client(project_id, customer_id, region) + logger.info("Listing case close definitions (page_size=%s)...", page_size) + + extra_params: Dict[str, Any] = {} + if filter: + extra_params["filter"] = filter + if order_by: + extra_params["orderBy"] = order_by + + result = chronicle_paginated_request( + chronicle, + path="caseCloseDefinitions", + items_key="caseCloseDefinitions", + api_version=APIVersion.V1, + page_size=page_size, + page_token=page_token, + extra_params=extra_params if extra_params else None, + as_list=False, + ) + + if isinstance(result, list): + return {"caseCloseDefinitions": result} + return result + + except Exception as e: + error_msg = f"Error listing case close definitions: {e}" + logger.error("Error listing case close definitions: %s", e) + return {"error": error_msg} diff --git a/server/secops/tests/test_case_close_definitions_unit.py b/server/secops/tests/test_case_close_definitions_unit.py new file mode 100644 index 00000000..6194a111 --- /dev/null +++ b/server/secops/tests/test_case_close_definitions_unit.py @@ -0,0 +1,134 @@ +# 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 Chronicle Case Close Definitions MCP tools.""" + +import importlib.util +import os +import sys +from unittest.mock import MagicMock, patch + +import pytest + +# Ensure server/secops is in path +current_dir = os.path.dirname(os.path.abspath(__file__)) +server_secops_dir = os.path.dirname(current_dir) +if server_secops_dir not in sys.path: + sys.path.append(server_secops_dir) + +# Mock mcp if not installed +if importlib.util.find_spec("mcp") is None: + mock_mcp = MagicMock() + sys.modules["mcp"] = mock_mcp + sys.modules["mcp.server"] = MagicMock() + sys.modules["mcp.server.fastmcp"] = MagicMock() + + def tool_decorator(*args, **kwargs): + def wrapper(func): + return func + return wrapper + + mock_fastmcp_instance = MagicMock() + mock_fastmcp_instance.tool.side_effect = tool_decorator + sys.modules["mcp.server.fastmcp"].FastMCP.return_value = mock_fastmcp_instance + +from secops_mcp.tools.case_close_definitions import list_case_close_definitions + + +@pytest.fixture +def mock_chronicle_client(): + client = MagicMock() + return client + + +@pytest.mark.asyncio +async def test_list_case_close_definitions_success(mock_chronicle_client): + """Test listing case close definitions successfully with default parameters.""" + expected_response = { + "caseCloseDefinitions": [ + { + "name": "projects/p/locations/us/instances/i/caseCloseDefinitions/def1", + "closeReason": "MALICIOUS", + "rootCause": "Phishing credential harvest", + }, + { + "name": "projects/p/locations/us/instances/i/caseCloseDefinitions/def2", + "closeReason": "NOT_MALICIOUS", + "rootCause": "Authorized Security Test", + }, + ], + "nextPageToken": "", + "totalSize": 2, + } + + with patch("secops_mcp.tools.case_close_definitions.get_chronicle_client", return_value=mock_chronicle_client), \ + patch("secops_mcp.tools.case_close_definitions.chronicle_paginated_request", return_value=expected_response) as mock_request: + result = await list_case_close_definitions() + + mock_request.assert_called_once() + call_kwargs = mock_request.call_args[1] + assert call_kwargs["path"] == "caseCloseDefinitions" + assert call_kwargs["page_size"] == 50 + assert call_kwargs["page_token"] is None + assert result == expected_response + + +@pytest.mark.asyncio +async def test_list_case_close_definitions_with_filters(mock_chronicle_client): + """Test listing case close definitions with filter, order_by, and pagination.""" + expected_response = { + "caseCloseDefinitions": [ + { + "name": "projects/p/locations/us/instances/i/caseCloseDefinitions/def1", + "closeReason": "MALICIOUS", + "rootCause": "Phishing credential harvest", + } + ], + "nextPageToken": "token123", + "totalSize": 1, + } + + with patch("secops_mcp.tools.case_close_definitions.get_chronicle_client", return_value=mock_chronicle_client) as mock_get_client, \ + patch("secops_mcp.tools.case_close_definitions.chronicle_paginated_request", return_value=expected_response) as mock_request: + result = await list_case_close_definitions( + page_size=10, + page_token="tok_abc", + filter="close_reason='MALICIOUS'", + order_by="root_cause desc", + project_id="my-proj", + customer_id="my-cust", + region="europe", + ) + + mock_get_client.assert_called_once_with("my-proj", "my-cust", "europe") + call_kwargs = mock_request.call_args[1] + assert call_kwargs["page_size"] == 10 + assert call_kwargs["page_token"] == "tok_abc" + assert call_kwargs["extra_params"] == { + "filter": "close_reason='MALICIOUS'", + "orderBy": "root_cause desc", + } + assert result == expected_response + + +@pytest.mark.asyncio +async def test_list_case_close_definitions_error_handling(mock_chronicle_client): + """Test that API errors are caught and returned formatted.""" + with patch("secops_mcp.tools.case_close_definitions.get_chronicle_client", return_value=mock_chronicle_client), \ + patch("secops_mcp.tools.case_close_definitions.chronicle_paginated_request", side_effect=Exception("API failure")): + result = await list_case_close_definitions() + + assert isinstance(result, dict) + assert "error" in result + assert "API failure" in result["error"]