Skip to content
Open
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
33 changes: 33 additions & 0 deletions docs/servers/secops_mcp.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
22 changes: 22 additions & 0 deletions docs/servers/secops_soar_mcp.md
Original file line number Diff line number Diff line change
Expand Up @@ -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:**
Expand Down
45 changes: 45 additions & 0 deletions server/secops-soar/secops_soar_mcp/case_management.py
Original file line number Diff line number Diff line change
Expand Up @@ -445,6 +445,51 @@ 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."}

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.")],
Expand Down
1 change: 1 addition & 0 deletions server/secops-soar/secops_soar_mcp/utils/consts.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand Down
87 changes: 87 additions & 0 deletions server/secops-soar/tests/unit/test_case_management.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
# 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
1 change: 1 addition & 0 deletions server/secops/secops_mcp/tools/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 *
Expand Down
94 changes: 94 additions & 0 deletions server/secops/secops_mcp/tools/case_close_definitions.py
Original file line number Diff line number Diff line change
@@ -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(f"Listing case close definitions (page_size={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: {str(e)}"
logger.error(error_msg)
return {"error": error_msg}
Loading