diff --git a/docs/servers/secops_mcp.md b/docs/servers/secops_mcp.md index df99f877..fd368ba8 100644 --- a/docs/servers/secops_mcp.md +++ b/docs/servers/secops_mcp.md @@ -827,7 +827,180 @@ The service account or user credentials need the following Chronicle roles: - `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 investigation associations grouped by detection ID, with verdict and confidence information. +### Case Management Tools + +- **`list_cases(page_size=50, page_token=None, filter_query=None, order_by=None, project_id=None, customer_id=None, region=None)`** + - **Description:** List cases in Chronicle Security Operations with optional filtering (e.g. stage, priority, assignee, status, tags). + - **Parameters:** + - `page_size` (optional): Maximum number of cases to return (default: 50). + - `page_token` (optional): Pagination token for subsequent pages. + - `filter_query` (optional): Filter expression (e.g., `'priority = "HIGH" AND status = "OPEN"'`). + - `order_by` (optional): Sort order for results (e.g., `'create_time desc'`). + - `project_id` (optional): Google Cloud project ID. + - `customer_id` (optional): Chronicle customer ID. + - `region` (optional): Chronicle region. + - **Returns:** Dictionary containing list of cases and pagination metadata. + +- **`get_case(case_id, project_id=None, customer_id=None, region=None)`** + - **Description:** Retrieve a single case by ID with complete metadata including priority, stage, assignee, tags, insights, and SLA status. + - **Parameters:** + - `case_id` (required): The case ID or full resource name. + - **Returns:** Complete case resource object. + +- **`get_case_full_details(case_id, project_id=None, customer_id=None, region=None)`** + - **Description:** Perform parallel multi-resource triage by concurrently fetching the case details, all associated alerts, and analyst comments in a single unified payload. + - **Parameters:** + - `case_id` (required): The case ID. + - **Returns:** Dictionary containing `case_details`, `case_alerts`, and `case_comments`. + +- **`update_case(case_id, update_fields, update_mask=None, project_id=None, customer_id=None, region=None)`** + - **Description:** Update case properties and metadata with automatic FieldMask generation. + - **Parameters:** + - `case_id` (required): The case ID to update. + - `update_fields` (required): Dictionary of fields to update (e.g., `{'displayName': 'Updated Case Title'}`). + - `update_mask` (optional): Comma-separated list of field paths (e.g., `'display_name,priority'`). + +- **`change_case_priority(case_id, priority, project_id=None, customer_id=None, region=None)`** + - **Description:** Update case priority, firing the Chronicle SOAR *Case Priority Changed* automated reaction trigger. + - **Parameters:** + - `case_id` (required): The case ID. + - `priority` (required): New priority level (`'LOW'`, `'MEDIUM'`, `'HIGH'`, `'CRITICAL'`). + +- **`change_case_stage(case_id, stage, project_id=None, customer_id=None, region=None)`** + - **Description:** Update case investigation stage, firing the Chronicle SOAR *Case Stage Changed* reaction trigger. + - **Parameters:** + - `case_id` (required): The case ID. + - `stage` (required): New stage name (e.g., `'Triage'`, `'Investigation'`, `'Containment'`, `'Remediation'`). + +- **`assign_case(case_id, assignee, project_id=None, customer_id=None, region=None)`** + - **Description:** Assign or reassign case ownership, firing the Chronicle SOAR *Case Assignee Changed* reaction trigger. + - **Parameters:** + - `case_id` (required): The case ID. + - `assignee` (required): Email or identifier of the assignee. + +- **`set_custom_case_fields(case_id, custom_fields, project_id=None, customer_id=None, region=None)`** + - **Description:** Set custom case fields, firing the Chronicle SOAR *Custom Case Field Changed* reaction trigger. + - **Parameters:** + - `case_id` (required): The case ID. + - `custom_fields` (required): Key-value dictionary of custom fields. + +- **`add_case_tag(case_id, tag, project_id=None, customer_id=None, region=None)`** / **`remove_case_tag(case_id, tag, ...)`** + - **Description:** Add or remove a categorization tag on a case. + +- **`add_case_insight(case_id, insight_text, project_id=None, customer_id=None, region=None)`** + - **Description:** Record an analytical finding, evidence summary, or key decision into the case evidence journal. + +- **`pause_case_sla(case_id, reason=None, ...)`** / **`resume_case_sla(case_id, ...)`** + - **Description:** Pause or resume the active SLA countdown timer for a case. + +- **`close_case(case_id, close_reason=None, root_cause=None, ...)`** / **`reopen_case(case_id, ...)`** + - **Description:** Transition a case to closed or reopened status with resolution reasoning. + +- **`list_case_comments(case_id, page_size=50, page_token=None, ...)`** + - **Description:** List all analyst notes, automated system comments, and discussion threads attached to a case. + +- **`create_case_comment(case_id, comment, ...)`** (Alias: **`post_case_comment`**) + - **Description:** Post a new analyst comment, triage note, or communication entry to the case wall. + +--- + +### Case Alert Management Tools + +- **`list_case_alerts(case_id, page_size=50, page_token=None, filter_query=None, ...)`** + - **Description:** List all security alerts grouped under a specific case. + +- **`get_case_alert(case_id, alert_id, ...)`** + - **Description:** Retrieve full details for a specific alert within a case. + +- **`update_case_alert(case_id, alert_id, update_fields, update_mask=None, ...)`** + - **Description:** Update alert properties with FieldMask support. + +- **`change_alert_priority(case_id, alert_id, priority, ...)`** + - **Description:** Update alert priority, firing the Chronicle SOAR *Alert Priority Changed* reaction trigger. + +- **`set_alert_custom_fields(case_id, alert_id, custom_fields, ...)`** + - **Description:** Set custom fields on an alert, firing the Chronicle SOAR *Alert Custom Field Changed* reaction trigger. + +- **`move_case_alert(source_case_id, alert_id, target_case_id, ...)`** + - **Description:** Move an alert from one case to another to consolidate or split incident investigations. + +- **`add_alert_tag(case_id, alert_id, tag, ...)`** / **`remove_alert_tag(case_id, alert_id, tag, ...)`** + - **Description:** Add or remove categorization tags on an alert. + +- **`list_alert_group_identifiers_by_case(case_id, ...)`** + - **Description:** List alert group identifiers associated with a case for correlated multi-alert investigations. + +- **`list_events_by_alert(case_id, alert_id, page_size=50, ...)`** (Alias: **`list_involved_events`**) + - **Description:** Query and stream the raw forensic events and security telemetry that triggered a specific alert. + +--- + +### Entity Investigation Tools + +- **`get_involved_entity(case_id, alert_id, entity_id, ...)`** + - **Description:** Retrieve details for a specific entity involved in an alert in SOAR context. + +- **`list_involved_entities(case_id, alert_id=None, page_size=50, ...)`** + - **Description:** List all entities (hosts, users, IPs, domains, hashes) involved in a case or alert. + +- **`get_entities_by_alert_group_identifiers(case_id, alert_group_identifiers, ...)`** + - **Description:** Fetch entities associated with specific alert group IDs. + +- **`get_entity_details(identifier, ...)`** + - **Description:** Retrieve rich enrichment data, threat intelligence score, and asset context for an entity. + +- **`search_entity(term=None, entity_types=None, is_suspicious=None, page_size=50, ...)`** + - **Description:** Search across all known entities in the Chronicle SOAR environment matching specific attributes. + +--- + +### Integration Management & Action Execution Tools + +- **`list_integrations(page_size=50, page_token=None, ...)`** + - **Description:** Discover all third-party and custom integrations configured on the tenant (e.g. VirusTotal, SentinelOne, Slack, Jira). + +- **`list_integration_actions(integration_id, page_size=50, ...)`** + - **Description:** List all executable manual and automated actions supported by an integration (e.g. Enrich IP, Isolate Host). + +- **`list_integration_instances(integration_id, page_size=50, ...)`** + - **Description:** List configured environment instances and instance GUIDs for an integration. + +- **`execute_integration_instance_test(integration_id, instance_id, ...)`** + - **Description:** Execute a live connectivity test ('ping') to verify that SecOps can communicate with the third-party service using configured credentials. + +- **`execute_manual_action(case_id, action_name, action_provider="Scripts", properties=None, target_entities=None, ...)`** + - **Description:** Execute a response action (e.g. enrich IOC, block IP, isolate endpoint) on a case, alert, or target entities. + +- **`get_action_result_by_id(action_result_id, ...)`** + - **Description:** Retrieve the execution status, logs, and structured output of an asynchronous action. + +--- + +### Playbook Lifecycle & Execution Tools + +- **`list_playbooks(playbook_types=None, filter_query=None, page_size=50, ...)`** + - **Description:** List automated playbook workflows configured in Chronicle SOAR (e.g. Regular and Nested playbooks). + +- **`get_playbook(playbook_id, ...)`** + - **Description:** Retrieve workflow definition, trigger conditions, and step graph for a playbook. + +- **`list_playbook_instances(playbook_id=None, case_id=None, page_size=50, ...)`** + - **Description:** List historical playbook execution runs and instance tracking data. + +- **`execute_playbook(playbook_id, case_id, alert_id=None, ...)`** (Alias: **`trigger_playbook`**) + - **Description:** Manually trigger a SOAR playbook workflow on a case or alert. + +--- + +### Connector Event Tools + +- **`list_connector_events(connector_id=None, page_size=50, ...)`** + - **Description:** List raw events ingested via security data connectors. + +- **`get_connector_event(connector_event_id, ...)`** + - **Description:** Retrieve complete raw payload and processing status for a connector event. + +--- ## Usage Examples @@ -1123,9 +1296,75 @@ Entities previously on this watchlist will now use their base risk scores without the 4.0x multiplier. ``` +### Example 9: Parallel Multi-Resource Case Triage & Trigger Workflow + +``` +User: Can you pull up case 1042, give me a full breakdown of the incident, and raise its priority if it involves compromised credentials? + +AI uses: get_case_full_details(case_id="1042") + +AI: Here is the unified triage summary for Case #1042: +- Title: Suspicious Kerberoasting Activity Against Service Account +- Priority: MEDIUM +- Stage: Triage +- Assignee: None (Unassigned) +- Associated Alerts (2): + 1. "Potential Kerberoasting SPN Ticket Request" (High) + 2. "Excessive Kerberos TGS Requests from Workstation" (Medium) +- Involved Entities: + - User: `svc_backup` + - Host: `ws-fin-08.corp.internal` +- Recent Comments: + - "Automated ingest: Correlated 2 detections into single incident." + +Because this involves active credential abuse against a privileged service account, I recommend escalating to CRITICAL. + +User: Yes, escalate the priority to CRITICAL, assign it to me (secops-lead@company.com), and post a triage note. + +AI uses: change_case_priority(case_id="1042", priority="CRITICAL") +AI uses: assign_case(case_id="1042", assignee="secops-lead@company.com") +AI uses: create_case_comment( + case_id="1042", + comment="Priority escalated to CRITICAL due to confirmed service account credential exposure. Assigned to secops-lead for containment." +) + +AI: Case #1042 updated: +- Priority changed to CRITICAL (Chronicle SOAR Case Priority Changed trigger dispatched). +- Assigned to secops-lead@company.com (Case Assignee Changed trigger dispatched). +- Triage comment recorded in the case wall. +``` + +### Example 10: Integration Connectivity & Manual Action Execution + +``` +User: Test our VirusTotal integration to see if it's healthy, then enrich the suspicious IP 198.51.100.44 on Case 1042. + +AI uses: list_integration_instances(integration_id="VirusTotalV3") +AI extracts instance ID: "3e9496eb-09cd-4b3c-a4ce-4c788d6663a7" + +AI uses: execute_integration_instance_test( + integration_id="VirusTotalV3", + instance_id="3e9496eb-09cd-4b3c-a4ce-4c788d6663a7" +) + +AI: VirusTotal V3 connectivity test succeeded: +"Successfully connected to the VirusTotal server with the provided connection parameters!" + +Now executing the manual enrichment action on Case #1042... + +AI uses: execute_manual_action( + case_id="1042", + action_name="VirusTotalV3_Enrich IP", + target_entities=[{"entity": "198.51.100.44", "entityType": "IP Address"}] +) + +AI: Manual enrichment action dispatched. VirusTotal reputation and ASN data attached to the case evidence graph. +``` + ## Additional Resources - [Chronicle API Documentation](https://cloud.google.com/chronicle/docs/reference/rest) - [UDM Schema Reference](https://cloud.google.com/chronicle/docs/reference/udm-field-list) - [Chronicle Security Operations Overview](https://cloud.google.com/chronicle/docs/overview) - [Chronicle Query Language Syntax](https://cloud.google.com/chronicle/docs/reference/yara-l-2-0-syntax) + diff --git a/server/secops/README.md b/server/secops/README.md index 8b1d7600..c7a39ec8 100644 --- a/server/secops/README.md +++ b/server/secops/README.md @@ -271,6 +271,65 @@ These tools help you leverage Google-curated detection content: - **Investigation**: Use `search_curated_detections` to analyze threats detected by curated rules - **Configuration**: Manage alerting settings and precision tuning for optimal detection coverage +### Case Management & SOAR Trigger Tools +These tools provide complete 1P Case lifecycle operations, automated reaction triggers, and multi-resource aggregation: +- **`list_cases`**: List cases with filtering by stage, priority, assignee, status, or tags +- **`get_case`**: Retrieve complete case resource metadata +- **`get_case_full_details`**: Perform high-performance parallel triage by concurrently retrieving case details, attached alerts, and analyst comments in a unified payload +- **`update_case`**: Update case metadata with automatic FieldMask generation +- **`change_case_priority`**: Update priority, firing the Chronicle SOAR *Case Priority Changed* automated reaction trigger +- **`change_case_stage`**: Update stage, firing the Chronicle SOAR *Case Stage Changed* reaction trigger +- **`assign_case`**: Assign case owner, firing the Chronicle SOAR *Case Assignee Changed* reaction trigger +- **`set_custom_case_fields`**: Set custom fields, firing the Chronicle SOAR *Custom Case Field Changed* reaction trigger +- **`add_case_tag` / `remove_case_tag`**: Add or remove categorization tags +- **`add_case_insight`**: Record analyst insights and evidence notes in the case journal +- **`pause_case_sla` / `resume_case_sla`**: Pause or resume the active SLA countdown timer +- **`close_case` / `reopen_case`**: Close or reopen case investigations with resolution reasoning +- **`list_case_comments`**: Retrieve analyst notes and discussion threads +- **`create_case_comment` (alias: `post_case_comment`)**: Post comments to the case wall + +### Case Alert Management & Forensics Tools +These tools manage security alerts within cases, grouping, and raw telemetry inspection: +- **`list_case_alerts`**: List alerts attached to a specific case +- **`get_case_alert`**: Get full alert details and metadata +- **`update_case_alert`**: Update alert fields with FieldMask support +- **`change_alert_priority`**: Update alert priority, firing the Chronicle SOAR *Alert Priority Changed* reaction trigger +- **`set_alert_custom_fields`**: Set custom fields on an alert, firing the Chronicle SOAR *Alert Custom Field Changed* reaction trigger +- **`move_case_alert`**: Move an alert between cases to consolidate or split incident investigations +- **`add_alert_tag` / `remove_alert_tag`**: Add or remove categorization tags on an alert +- **`list_alert_group_identifiers_by_case`**: List alert group identifiers associated with a case +- **`list_events_by_alert` (alias: `list_involved_events`)**: Query and stream raw security telemetry events and forensic logs that triggered the alert + +### Entity Investigation Tools +These tools provide SOAR-context entity analysis and search: +- **`get_involved_entity`**: Retrieve specific entity details in SOAR alert context +- **`list_involved_entities`**: List all entities (hosts, users, IPs, domains, hashes) involved in a case or alert +- **`get_entities_by_alert_group_identifiers`**: Fetch entities associated with specific alert group IDs +- **`get_entity_details`**: Retrieve rich threat intelligence and asset context for an entity +- **`search_entity`**: Search entities across the Chronicle SOAR platform matching specific attributes and flags + +### Integration Management & Action Execution Tools +These tools manage third-party and custom SOAR integrations and action execution: +- **`list_integrations`**: Discover all configured integrations (e.g. VirusTotal, SentinelOne, Slack, Jira) +- **`list_integration_actions`**: List executable manual and automated actions supported by an integration +- **`list_integration_instances`**: List configured environment instances and instance GUIDs +- **`execute_integration_instance_test`**: Execute a live connectivity test ('ping') to verify credentials and connectivity +- **`execute_manual_action`**: Execute manual actions (e.g. enrich IOC, block IP, isolate host) on a case or alert +- **`get_action_result_by_id`**: Retrieve execution status, logs, and output data for an action run + +### Playbook Lifecycle & Execution Tools +These tools manage automated SOAR workflow playbooks: +- **`list_playbooks`**: List automated playbooks (regular and nested workflows) +- **`get_playbook`**: Retrieve workflow definitions, step graphs, and trigger configurations +- **`list_playbook_instances`**: List historical playbook execution runs and instance tracking data +- **`execute_playbook` (alias: `trigger_playbook`)**: Manually trigger a playbook workflow on a case or alert + +### Connector Event Tools +These tools inspect raw connector ingestion events: +- **`list_connector_events`**: List raw events ingested via security data connectors +- **`get_connector_event`**: Retrieve raw payload and processing status for a connector event + + ## Configuration ### MCP Server Configuration diff --git a/server/secops/secops_mcp/tools/__init__.py b/server/secops/secops_mcp/tools/__init__.py index 1b16e316..ce045770 100644 --- a/server/secops/secops_mcp/tools/__init__.py +++ b/server/secops/secops_mcp/tools/__init__.py @@ -13,14 +13,20 @@ # limitations under the License. """Security Operations MCP tools package.""" +from .case_alert_management import * +from .case_management import * +from .connector_event_management import * from .curated_rules_management import * from .data_table_management import * +from .entity_investigation import * from .entity_lookup import * from .feed_management import * +from .integration_management import * from .investigation_management import * from .ioc_matches import * from .log_ingestion import * from .parser_management import * +from .playbook_management import * from .reference_list_management import * from .rule_exclusions import * from .search import * diff --git a/server/secops/secops_mcp/tools/case_alert_management.py b/server/secops/secops_mcp/tools/case_alert_management.py new file mode 100644 index 00000000..2c96ebd7 --- /dev/null +++ b/server/secops/secops_mcp/tools/case_alert_management.py @@ -0,0 +1,570 @@ +# Copyright 2025 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 Chronicle 1P Case Alert Management.""" + +import logging +from typing import Any, Dict, List, Optional + +from secops_mcp.server import get_chronicle_client, server + +logger = logging.getLogger("secops-mcp") + + +def _format_case_alert_name(instance_id: str, case_id: str, alert_id: str) -> str: + """Format full case alert resource name.""" + if alert_id.startswith("projects/"): + return alert_id + if case_id.startswith("projects/"): + return f"{case_id}/caseAlerts/{alert_id}" + return f"{instance_id}/cases/{case_id}/caseAlerts/{alert_id}" + + +def _get_base_endpoint(chronicle: Any, version: str = "v1") -> str: + """Get the base REST endpoint for Chronicle 1P resources.""" + if version == "v1" and hasattr(chronicle, "base_v1_url") and chronicle.base_v1_url: + return f"{chronicle.base_v1_url}/{chronicle.instance_id}" + base_url = chronicle.base_url + if version == "v1" and "/v1alpha" in base_url: + base_url = base_url.replace("/v1alpha", "/v1") + return f"{base_url}/{chronicle.instance_id}" + + +@server.tool() +async def list_case_alerts( + case_id: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, + filter_query: Optional[str] = None, + page_size: int = 50, + page_token: Optional[str] = None, +) -> Dict[str, Any]: + """List CaseAlerts within a specific Case using the 1P CaseAlertService REST API. + + Retrieves all alerts associated with a given case, including their status, + priority, detection rule details, and SLA information. + + **Workflow Integration:** + - Examine all alerts grouped into a security incident case. + - Filter for unresolved alerts or critical severity detections. + + Args: + case_id (str): The ID or name of the case whose alerts to list. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + filter_query (Optional[str]): Filter expression (e.g. 'priority = "HIGH"'). + page_size (int): Number of alerts to return per page. Defaults to 50. + page_token (Optional[str]): Token for retrieving the next page. + + Returns: + Dict[str, Any]: List of CaseAlert objects and pagination token. + """ + try: + if not case_id: + return {"error": "case_id parameter is required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + short_case_id = case_id.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_case_id}/caseAlerts" + + params: Dict[str, Any] = {"pageSize": page_size} + if filter_query: + params["filter"] = filter_query + if page_token: + params["pageToken"] = page_token + + response = chronicle.session.get(url, params=params) + if response.status_code != 200: + return { + "error": f"Failed to list case alerts: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error listing case alerts for case %s: %s", case_id, e) + return {"error": f"Failed to list case alerts: {str(e)}"} + + +@server.tool() +async def get_case_alert( + case_id: str, + alert_id: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Get details of a specific CaseAlert using the 1P CaseAlertService REST API. + + Args: + case_id (str): The Case ID or full resource name. + alert_id (str): The Alert ID or full resource name. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Detailed CaseAlert object. + """ + try: + if not case_id or not alert_id: + return {"error": "Both case_id and alert_id parameters are required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + short_case_id = case_id.split("/")[-1] + short_alert_id = alert_id.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_case_id}/caseAlerts/{short_alert_id}" + + response = chronicle.session.get(url) + if response.status_code != 200: + return { + "error": f"Failed to get case alert {alert_id}: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error getting case alert %s: %s", alert_id, e) + return {"error": f"Failed to get case alert: {str(e)}"} + + +@server.tool() +async def update_case_alert( + case_id: str, + alert_id: str, + update_mask: str, + priority: Optional[str] = None, + status: Optional[str] = None, + custom_field_values: Optional[List[Dict[str, Any]]] = None, + closure_reason: Optional[str] = None, + closure_comment: Optional[str] = None, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Update a CaseAlert in Chronicle using the 1P CaseAlertService.UpdateCaseAlert REST API. + + Modifies alert properties such as priority, status, custom field values, or closure feedback. + + **Reaction Triggers Activated:** + - Setting `priority` triggers **Alert Priority Changed**. + - Setting `custom_field_values` triggers **Alert Custom Field Changed**. + + Args: + case_id (str): Case ID or full resource name. + alert_id (str): Alert ID or full resource name. + update_mask (str): Comma-separated list of fields being updated + (e.g., "priority", "status", "custom_field_values", "feedback_summary"). + priority (Optional[str]): Updated priority ("PRIORITY_LOW", "PRIORITY_MEDIUM", + "PRIORITY_HIGH", "PRIORITY_CRITICAL", or short names "LOW", "MEDIUM", "HIGH", "CRITICAL"). + status (Optional[str]): Updated alert status (e.g., "OPEN", "CLOSED"). + custom_field_values (Optional[List[Dict[str, Any]]]): Custom fields to attach to the alert. + closure_reason (Optional[str]): Closure justification if closing the alert. + closure_comment (Optional[str]): Notes explaining the closure. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: The updated CaseAlert object. + """ + try: + if not case_id or not alert_id: + return {"error": "Both case_id and alert_id parameters are required"} + if not update_mask: + return {"error": "update_mask parameter is required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + short_case_id = case_id.split("/")[-1] + short_alert_id = alert_id.split("/")[-1] + alert_name = _format_case_alert_name(chronicle.instance_id, short_case_id, short_alert_id) + url = f"{_get_base_endpoint(chronicle)}/cases/{short_case_id}/caseAlerts/{short_alert_id}" + + body: Dict[str, Any] = {"name": alert_name} + if priority is not None: + p_upper = priority.upper() + if not p_upper.startswith("PRIORITY_") and p_upper in ["LOW", "MEDIUM", "HIGH", "CRITICAL", "INFO"]: + p_upper = f"PRIORITY_{p_upper}" + body["priority"] = p_upper + if status is not None: + body["status"] = status + if custom_field_values is not None: + body["customFieldValues"] = custom_field_values + if closure_reason is not None or closure_comment is not None: + body["feedbackSummary"] = { + "closureReason": closure_reason, + "comment": closure_comment, + } + + params = {"updateMask": update_mask} + response = chronicle.session.patch(url, params=params, json=body) + if response.status_code != 200: + return { + "error": f"Failed to update case alert: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error updating case alert %s: %s", alert_id, e) + return {"error": f"Failed to update case alert: {str(e)}"} + + +@server.tool() +async def change_alert_priority( + case_id: str, + alert_id: str, + priority: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Change the priority level of a CaseAlert using Chronicle 1P REST API. + + Triggers the **Alert Priority Changed** reaction trigger in SOAR playbooks. + + Args: + case_id (str): The Case ID containing the alert. + alert_id (str): The Alert ID to update. + priority (str): New priority ("PRIORITY_LOW", "PRIORITY_MEDIUM", + "PRIORITY_HIGH", "PRIORITY_CRITICAL", or short names "LOW", "MEDIUM", "HIGH", "CRITICAL"). + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Updated CaseAlert object or status confirmation. + """ + try: + p_upper = priority.upper() + if not p_upper.startswith("PRIORITY_") and p_upper in ["LOW", "MEDIUM", "HIGH", "CRITICAL", "INFO"]: + p_upper = f"PRIORITY_{p_upper}" + + return await update_case_alert( + case_id=case_id, + alert_id=alert_id, + update_mask="priority", + priority=p_upper, + project_id=project_id, + customer_id=customer_id, + region=region, + ) + except Exception as e: + logger.error("Error changing alert priority: %s", e) + return {"error": f"Failed to change alert priority: {str(e)}"} + + +@server.tool() +async def set_alert_custom_fields( + case_id: str, + alert_id: str, + custom_fields: Dict[str, Any], + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Set custom field values on a CaseAlert using Chronicle 1P REST API. + + Triggers the **Alert Custom Field Changed** reaction trigger in SOAR playbooks. + + Args: + case_id (str): The Case ID containing the alert. + alert_id (str): The Alert ID to update. + custom_fields (Dict[str, Any]): Dictionary mapping custom field names to values. + Example: {"MalwareFamily": "Emotet", "Confidence": "High"} + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Updated CaseAlert object. + """ + try: + if not custom_fields: + return {"error": "custom_fields dictionary cannot be empty"} + + custom_field_values = [ + {"fieldName": k, "value": str(v) if not isinstance(v, (dict, list)) else str(v)} + for k, v in custom_fields.items() + ] + + return await update_case_alert( + case_id=case_id, + alert_id=alert_id, + update_mask="custom_field_values", + custom_field_values=custom_field_values, + project_id=project_id, + customer_id=customer_id, + region=region, + ) + except Exception as e: + logger.error("Error setting alert custom fields: %s", e) + return {"error": f"Failed to set alert custom fields: {str(e)}"} + + +@server.tool() +async def move_case_alert( + source_case_id: str, + alert_id: str, + destination_case_id: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Move a CaseAlert from one Case to another using Chronicle 1P REST API. + + Args: + source_case_id (str): Case ID currently containing the alert. + alert_id (str): Alert ID to move. + destination_case_id (str): Destination Case ID or resource name. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: MoveAlertResponse confirmation. + """ + try: + if not source_case_id or not alert_id or not destination_case_id: + return {"error": "source_case_id, alert_id, and destination_case_id parameters are all required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + short_src_case_id = source_case_id.split("/")[-1] + short_alert_id = alert_id.split("/")[-1] + short_dest_case_id = destination_case_id.split("/")[-1] + + url = f"{_get_base_endpoint(chronicle)}/cases/{short_src_case_id}/caseAlerts/{short_alert_id}:move" + + body = {"destinationCaseId": short_dest_case_id} + response = chronicle.session.post(url, json=body) + if response.status_code != 200: + return { + "error": f"Failed to move alert: {response.status_code} - {response.text}" + } + return response.json() if response.text else {"status": "SUCCESS", "message": f"Moved alert {short_alert_id} to case {short_dest_case_id}"} + except Exception as e: + logger.error("Error moving case alert: %s", e) + return {"error": f"Failed to move case alert: {str(e)}"} + + +@server.tool() +async def add_alert_tag( + case_id: str, + alert_id: str, + tag: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Add a categorization tag to a CaseAlert using Chronicle 1P REST API. + + Args: + case_id (str): Case ID containing the alert. + alert_id (str): Alert ID. + tag (str): Tag string to add to the alert. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Response confirmation. + """ + try: + if not case_id or not alert_id or not tag: + return {"error": "case_id, alert_id, and tag parameters are all required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + short_case_id = case_id.split("/")[-1] + short_alert_id = alert_id.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_case_id}/caseAlerts/{short_alert_id}:addTag" + + response = chronicle.session.post(url, json={"tag": tag}) + if response.status_code != 200: + return { + "error": f"Failed to add tag to alert: {response.status_code} - {response.text}" + } + return response.json() if response.text else {"status": "SUCCESS", "message": f"Added tag '{tag}' to alert"} + except Exception as e: + logger.error("Error adding tag to alert: %s", e) + return {"error": f"Failed to add tag to alert: {str(e)}"} + + +@server.tool() +async def remove_alert_tag( + case_id: str, + alert_id: str, + tag: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Remove a categorization tag from a CaseAlert using Chronicle 1P REST API. + + Args: + case_id (str): Case ID containing the alert. + alert_id (str): Alert ID. + tag (str): Tag string to remove. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Response confirmation. + """ + try: + if not case_id or not alert_id or not tag: + return {"error": "case_id, alert_id, and tag parameters are all required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + short_case_id = case_id.split("/")[-1] + short_alert_id = alert_id.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_case_id}/caseAlerts/{short_alert_id}:removeTag" + + response = chronicle.session.post(url, json={"tag": tag}) + if response.status_code != 200: + return { + "error": f"Failed to remove tag from alert: {response.status_code} - {response.text}" + } + return response.json() if response.text else {"status": "SUCCESS", "message": f"Removed tag '{tag}' from alert"} + except Exception as e: + logger.error("Error removing tag from alert: %s", e) + return {"error": f"Failed to remove tag from alert: {str(e)}"} + + +@server.tool() +async def list_alert_group_identifiers_by_case( + case_id: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, + page_size: int = 50, + page_token: Optional[str] = None, +) -> Dict[str, Any]: + """List alert group identifiers associated with a specific Case in Chronicle SOAR. + + Retrieves grouping keys used for correlation, playbook execution stages, or analyst assignment. + + Args: + case_id (str): The Case ID or full resource name. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + page_size (int): Max number of results. Defaults to 50. + page_token (Optional[str]): Pagination token. + + Returns: + Dict[str, Any]: List of alert group identifier strings and pagination info. + """ + try: + if not case_id: + return {"error": "case_id parameter is required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + short_case_id = case_id.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_case_id}:listAlertGroupIdentifiers" + + params: Dict[str, Any] = {"pageSize": page_size} + if page_token: + params["pageToken"] = page_token + + response = chronicle.session.get(url, params=params) + if response.status_code != 200: + # Fallback to extracting from list_case_alerts if dedicated endpoint is not active + alerts_res = await list_case_alerts(case_id=short_case_id, project_id=project_id, customer_id=customer_id, region=region) + if "caseAlerts" in alerts_res: + group_ids = list({ + gid for a in alerts_res["caseAlerts"] + for gid in a.get("alertGroupIdentifiers", []) + }) + return {"alertGroupIdentifiers": group_ids, "caseId": short_case_id} + return { + "error": f"Failed to list alert group identifiers: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error listing alert group identifiers for case %s: %s", case_id, e) + return {"error": f"Failed to list alert group identifiers: {str(e)}"} + + +@server.tool() +async def list_events_by_alert( + case_id: str, + alert_id: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, + page_size: int = 50, + page_token: Optional[str] = None, +) -> Dict[str, Any]: + """List the underlying security events (UDM events) associated with a specific alert. + + Retrieves the raw ground truth telemetry events that triggered the alert, + vital for verifying alerts, inspecting command lines, network connections, and forensic analysis. + + Args: + case_id (str): The Case ID containing the alert. + alert_id (str): The Alert ID or full resource name. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + page_size (int): Max number of events to return. Defaults to 50. + page_token (Optional[str]): Pagination token. + + Returns: + Dict[str, Any]: List of raw/UDM event objects linked to the alert. + """ + try: + if not case_id or not alert_id: + return {"error": "Both case_id and alert_id parameters are required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + short_case_id = case_id.split("/")[-1] + short_alert_id = alert_id.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_case_id}/caseAlerts/{short_alert_id}:listEvents" + + params: Dict[str, Any] = {"pageSize": page_size} + if page_token: + params["pageToken"] = page_token + + response = chronicle.session.get(url, params=params) + if response.status_code != 200: + # Fallback to get_case_alert with expand=events + alert_detail = await get_case_alert(case_id=short_case_id, alert_id=short_alert_id, project_id=project_id, customer_id=customer_id, region=region) + if "events" in alert_detail: + return {"events": alert_detail["events"], "alertId": short_alert_id} + return { + "error": f"Failed to list events for alert: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error listing events for alert %s: %s", alert_id, e) + return {"error": f"Failed to list events for alert: {str(e)}"} + + +@server.tool() +async def list_involved_events( + case_id: str, + alert_id: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, + page_size: int = 50, + page_token: Optional[str] = None, +) -> Dict[str, Any]: + """Alias for list_events_by_alert. Retrieves security telemetry events for a case alert.""" + return await list_events_by_alert( + case_id=case_id, + alert_id=alert_id, + project_id=project_id, + customer_id=customer_id, + region=region, + page_size=page_size, + page_token=page_token, + ) + diff --git a/server/secops/secops_mcp/tools/case_management.py b/server/secops/secops_mcp/tools/case_management.py new file mode 100644 index 00000000..fb94aa3c --- /dev/null +++ b/server/secops/secops_mcp/tools/case_management.py @@ -0,0 +1,918 @@ +# Copyright 2025 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 Chronicle 1P Case Management.""" + +import logging +from typing import Any, Dict, List, Optional, Union + +from secops_mcp.server import get_chronicle_client, server + +logger = logging.getLogger("secops-mcp") + + +def _format_case_name(instance_id: str, case_id: str) -> str: + """Format full case resource name if only ID is provided.""" + if case_id.startswith("projects/"): + return case_id + return f"{instance_id}/cases/{case_id}" + + +def _get_base_endpoint(chronicle: Any, version: str = "v1") -> str: + """Get the base REST endpoint for Chronicle 1P resources.""" + if version == "v1" and hasattr(chronicle, "base_v1_url") and chronicle.base_v1_url: + return f"{chronicle.base_v1_url}/{chronicle.instance_id}" + base_url = chronicle.base_url + if version == "v1" and "/v1alpha" in base_url: + base_url = base_url.replace("/v1alpha", "/v1") + return f"{base_url}/{chronicle.instance_id}" + + +@server.tool() +async def list_cases( + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, + filter_query: Optional[str] = None, + order_by: Optional[str] = None, + page_size: int = 50, + page_token: Optional[str] = None, +) -> Dict[str, Any]: + """List Cases in Chronicle using the 1P CaseService REST API. + + Retrieves a paginated list of cases from the Chronicle instance. Supports + filtering by priority, stage, status, assignee, and tags. + + **Workflow Integration:** + - Discover open or assigned cases requiring investigation. + - Monitor case queue health and workload distribution across SOC analysts. + - Query cases matching specific detection rules or severity levels. + + **Use Cases:** + - "List all open cases assigned to me" + - "Show high priority cases in the Triage stage" + - "Retrieve recent cases created in the last 24 hours" + + Args: + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region (e.g., "us", "europe"). + filter_query (Optional[str]): Filter expression (e.g., + 'priority = "HIGH" AND status = "OPEN"'). + order_by (Optional[str]): Sort order for results (e.g., "create_time desc"). + page_size (int): Number of cases to return per page (max 100). Defaults to 50. + page_token (Optional[str]): Pagination token for retrieving next page. + + Returns: + Dict[str, Any]: Dictionary containing list of cases and nextPageToken. + """ + try: + chronicle = get_chronicle_client(project_id, customer_id, region) + url = f"{_get_base_endpoint(chronicle)}/cases" + + params: Dict[str, Any] = {"pageSize": page_size} + if filter_query: + params["filter"] = filter_query + if order_by: + params["orderBy"] = order_by + if page_token: + params["pageToken"] = page_token + + response = chronicle.session.get(url, params=params) + if response.status_code != 200: + return { + "error": f"Failed to list cases: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error listing cases: %s", e) + return {"error": f"Failed to list cases: {str(e)}"} + + +@server.tool() +async def get_case( + case_id: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Get full details of a specific Case using the 1P CaseService REST API. + + Retrieves comprehensive metadata for a single case, including its priority, + stage, assignee, display name, tags, SLA status, and custom fields. + + **Workflow Integration:** + - Inspect case state before executing triage or remediation actions. + - Verify updated priorities, stages, or custom field values. + + Args: + case_id (str): Case ID (e.g., "case_12345") or full resource name. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Case object with detailed properties. + """ + try: + if not case_id: + return {"error": "case_id parameter is required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + case_name = _format_case_name(chronicle.instance_id, case_id) + url = f"{_get_base_endpoint(chronicle)}/cases/{case_name.split('/')[-1]}" + + response = chronicle.session.get(url) + if response.status_code != 200: + return { + "error": f"Failed to get case {case_id}: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error getting case %s: %s", case_id, e) + return {"error": f"Failed to get case: {str(e)}"} + + +@server.tool() +async def update_case( + case_id: str, + update_mask: str, + display_name: Optional[str] = None, + priority: Optional[str] = None, + stage: Optional[str] = None, + assignee: Optional[str] = None, + custom_field_values: Optional[List[Dict[str, Any]]] = None, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Update a Case in Chronicle using the 1P CaseService.UpdateCase REST API. + + Updates selected fields on a case, such as priority, stage, assignee, + display name, or custom fields. + + **Reaction Triggers Activated:** + - Setting `priority` triggers **Case Priority Changed**. + - Setting `stage` triggers **Case Stage Changed**. + - Setting `assignee` triggers **Case Assignee Changed**. + - Setting `custom_field_values` triggers **Custom Case Field Changed**. + + Args: + case_id (str): Case ID or full resource name. + update_mask (str): Comma-separated list of fields being updated + (e.g., "priority,stage", "assignee", "custom_field_values"). + display_name (Optional[str]): Updated case title. + priority (Optional[str]): Updated priority level (e.g., "PRIORITY_LOW", + "PRIORITY_MEDIUM", "PRIORITY_HIGH", "PRIORITY_CRITICAL", "LOW", "HIGH"). + stage (Optional[str]): Target stage (e.g., "Triage", "Containment", "Remediation"). + assignee (Optional[str]): Analyst email or username. + custom_field_values (Optional[List[Dict[str, Any]]]): List of custom field + objects with `fieldName` and `value`. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: The updated Case object. + """ + try: + if not case_id: + return {"error": "case_id parameter is required"} + if not update_mask: + return {"error": "update_mask parameter is required (e.g., 'priority', 'stage')"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + case_name = _format_case_name(chronicle.instance_id, case_id) + short_id = case_name.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_id}" + + body: Dict[str, Any] = {"name": case_name} + if display_name is not None: + body["displayName"] = display_name + if priority is not None: + # Normalize priority if needed + p_upper = priority.upper() + if not p_upper.startswith("PRIORITY_") and p_upper in ["LOW", "MEDIUM", "HIGH", "CRITICAL", "INFO"]: + p_upper = f"PRIORITY_{p_upper}" + body["priority"] = p_upper + if stage is not None: + body["stage"] = stage + if assignee is not None: + body["assignee"] = assignee + if custom_field_values is not None: + body["customFieldValues"] = custom_field_values + + params = {"updateMask": update_mask} + response = chronicle.session.patch(url, params=params, json=body) + if response.status_code != 200: + return { + "error": f"Failed to update case {case_id}: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error updating case %s: %s", case_id, e) + return {"error": f"Failed to update case: {str(e)}"} + + +@server.tool() +async def change_case_priority( + case_ids: Union[str, List[str]], + priority: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Change the priority level for one or more cases using Chronicle 1P REST API. + + Triggers the **Case Priority Changed** reaction trigger in SOAR playbooks. + + Args: + case_ids (Union[str, List[str]]): Single case ID or list of case IDs. + priority (str): New priority level ("PRIORITY_LOW", "PRIORITY_MEDIUM", + "PRIORITY_HIGH", "PRIORITY_CRITICAL", or short names "LOW", "MEDIUM", "HIGH", "CRITICAL"). + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Success status or execution confirmation. + """ + try: + chronicle = get_chronicle_client(project_id, customer_id, region) + ids = [case_ids] if isinstance(case_ids, str) else case_ids + if not ids: + return {"error": "case_ids cannot be empty"} + + p_upper = priority.upper() + if not p_upper.startswith("PRIORITY_") and p_upper in ["LOW", "MEDIUM", "HIGH", "CRITICAL", "INFO"]: + p_upper = f"PRIORITY_{p_upper}" + + full_names = [_format_case_name(chronicle.instance_id, cid) for cid in ids] + url = f"{_get_base_endpoint(chronicle)}/cases:executeBulkChangePriority" + + body = {"names": full_names, "priority": p_upper} + response = chronicle.session.post(url, json=body) + if response.status_code != 200: + return { + "error": f"Failed to change case priority: {response.status_code} - {response.text}" + } + return {"status": "SUCCESS", "message": f"Updated priority to {p_upper} for {len(full_names)} case(s)"} + except Exception as e: + logger.error("Error changing case priority: %s", e) + return {"error": f"Failed to change case priority: {str(e)}"} + + +@server.tool() +async def change_case_stage( + case_ids: Union[str, List[str]], + stage: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Change the investigation stage for one or more cases using Chronicle 1P REST API. + + Triggers the **Case Stage Changed** reaction trigger in SOAR playbooks. + + Args: + case_ids (Union[str, List[str]]): Single case ID or list of case IDs. + stage (str): Target investigation lifecycle stage (e.g. "Triage", "Containment", "Remediation", "Closure"). + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Success status or execution confirmation. + """ + try: + chronicle = get_chronicle_client(project_id, customer_id, region) + ids = [case_ids] if isinstance(case_ids, str) else case_ids + if not ids: + return {"error": "case_ids cannot be empty"} + if not stage: + return {"error": "stage parameter is required"} + + full_names = [_format_case_name(chronicle.instance_id, cid) for cid in ids] + url = f"{_get_base_endpoint(chronicle)}/cases:executeBulkChangeStage" + + body = {"names": full_names, "stage": stage} + response = chronicle.session.post(url, json=body) + if response.status_code != 200: + return { + "error": f"Failed to change case stage: {response.status_code} - {response.text}" + } + return {"status": "SUCCESS", "message": f"Updated stage to '{stage}' for {len(full_names)} case(s)"} + except Exception as e: + logger.error("Error changing case stage: %s", e) + return {"error": f"Failed to change case stage: {str(e)}"} + + +@server.tool() +async def assign_case( + case_ids: Union[str, List[str]], + assignee: Optional[str] = None, + soc_role: Optional[str] = None, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Assign one or more cases to a user or SOC role using Chronicle 1P REST API. + + Triggers the **Case Assignee Changed** reaction trigger in SOAR playbooks. + + Args: + case_ids (Union[str, List[str]]): Single case ID or list of case IDs. + assignee (Optional[str]): User email or username to assign the case(s) to. + soc_role (Optional[str]): SOC role identifier to assign the case(s) to. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Success status or execution confirmation. + """ + try: + chronicle = get_chronicle_client(project_id, customer_id, region) + ids = [case_ids] if isinstance(case_ids, str) else case_ids + if not ids: + return {"error": "case_ids cannot be empty"} + if not assignee and not soc_role: + return {"error": "Either assignee or soc_role must be provided"} + + full_names = [_format_case_name(chronicle.instance_id, cid) for cid in ids] + url = f"{_get_base_endpoint(chronicle)}/cases:executeBulkAssign" + + body: Dict[str, Any] = {"names": full_names} + if assignee: + body["assignee"] = assignee + if soc_role: + body["socRole"] = soc_role + + response = chronicle.session.post(url, json=body) + if response.status_code != 200: + return { + "error": f"Failed to assign case(s): {response.status_code} - {response.text}" + } + assigned_target = assignee or soc_role + return {"status": "SUCCESS", "message": f"Assigned {len(full_names)} case(s) to {assigned_target}"} + except Exception as e: + logger.error("Error assigning case: %s", e) + return {"error": f"Failed to assign case: {str(e)}"} + + +@server.tool() +async def set_custom_case_fields( + case_id: str, + custom_fields: Dict[str, Any], + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Set custom field values on a Case using Chronicle 1P REST API. + + Triggers the **Custom Case Field Changed** reaction trigger in SOAR playbooks. + + Args: + case_id (str): Case ID or full resource name. + custom_fields (Dict[str, Any]): Dictionary mapping custom field names to values. + Example: {"ImpactLevel": "High", "Department": "Finance"} + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Updated Case object. + """ + try: + if not case_id: + return {"error": "case_id parameter is required"} + if not custom_fields: + return {"error": "custom_fields dictionary cannot be empty"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + case_name = _format_case_name(chronicle.instance_id, case_id) + short_id = case_name.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_id}" + + custom_field_values = [ + {"fieldName": k, "value": str(v) if not isinstance(v, (dict, list)) else str(v)} + for k, v in custom_fields.items() + ] + + body = { + "name": case_name, + "customFieldValues": custom_field_values, + } + params = {"updateMask": "custom_field_values"} + response = chronicle.session.patch(url, params=params, json=body) + if response.status_code != 200: + return { + "error": f"Failed to set custom case fields: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error setting custom case fields: %s", e) + return {"error": f"Failed to set custom case fields: {str(e)}"} + + +@server.tool() +async def add_case_tag( + case_id: str, + tag: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Add a categorization tag to a Case using Chronicle 1P REST API. + + Args: + case_id (str): Case ID or full resource name. + tag (str): Tag string to add to the case. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Response confirmation. + """ + try: + if not case_id or not tag: + return {"error": "Both case_id and tag parameters are required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + case_name = _format_case_name(chronicle.instance_id, case_id) + short_id = case_name.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_id}:addTag" + + response = chronicle.session.post(url, json={"tag": tag}) + if response.status_code != 200: + return { + "error": f"Failed to add tag to case: {response.status_code} - {response.text}" + } + return response.json() if response.text else {"status": "SUCCESS", "message": f"Added tag '{tag}'"} + except Exception as e: + logger.error("Error adding tag to case: %s", e) + return {"error": f"Failed to add tag: {str(e)}"} + + +@server.tool() +async def remove_case_tag( + case_id: str, + tag: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Remove a categorization tag from a Case using Chronicle 1P REST API. + + Args: + case_id (str): Case ID or full resource name. + tag (str): Tag string to remove. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Response confirmation. + """ + try: + if not case_id or not tag: + return {"error": "Both case_id and tag parameters are required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + case_name = _format_case_name(chronicle.instance_id, case_id) + short_id = case_name.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_id}:removeTag" + + response = chronicle.session.post(url, json={"tag": tag}) + if response.status_code != 200: + return { + "error": f"Failed to remove tag from case: {response.status_code} - {response.text}" + } + return response.json() if response.text else {"status": "SUCCESS", "message": f"Removed tag '{tag}'"} + except Exception as e: + logger.error("Error removing tag from case: %s", e) + return {"error": f"Failed to remove tag: {str(e)}"} + + +@server.tool() +async def add_case_insight( + case_id: str, + content: str, + insight_type: Optional[str] = None, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Add an analyst note or automation insight to a Case using Chronicle 1P REST API. + + Args: + case_id (str): Case ID or full resource name. + content (str): Text content of the insight or observation. + insight_type (Optional[str]): Type/classification of the insight. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Created insight details. + """ + try: + if not case_id or not content: + return {"error": "Both case_id and content parameters are required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + case_name = _format_case_name(chronicle.instance_id, case_id) + short_id = case_name.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_id}:createInsight" + + body: Dict[str, Any] = {"content": content} + if insight_type: + body["type"] = insight_type + + response = chronicle.session.post(url, json=body) + if response.status_code != 200: + return { + "error": f"Failed to add insight to case: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error adding insight to case: %s", e) + return {"error": f"Failed to add insight: {str(e)}"} + + +@server.tool() +async def pause_case_sla( + case_id: str, + reason: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Pause the SLA timer for a Case using Chronicle 1P REST API. + + Args: + case_id (str): Case ID or full resource name. + reason (str): Justification for pausing the SLA. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Status confirmation. + """ + try: + if not case_id or not reason: + return {"error": "Both case_id and reason parameters are required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + case_name = _format_case_name(chronicle.instance_id, case_id) + short_id = case_name.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_id}:pauseSla" + + response = chronicle.session.post(url, json={"reason": reason}) + if response.status_code != 200: + return { + "error": f"Failed to pause SLA: {response.status_code} - {response.text}" + } + return {"status": "SUCCESS", "message": f"Paused SLA on case {case_id}"} + except Exception as e: + logger.error("Error pausing SLA on case: %s", e) + return {"error": f"Failed to pause SLA: {str(e)}"} + + +@server.tool() +async def resume_case_sla( + case_id: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Resume a paused SLA timer for a Case using Chronicle 1P REST API. + + Args: + case_id (str): Case ID or full resource name. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Status confirmation. + """ + try: + if not case_id: + return {"error": "case_id parameter is required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + case_name = _format_case_name(chronicle.instance_id, case_id) + short_id = case_name.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_id}:resumeSla" + + response = chronicle.session.post(url, json={}) + if response.status_code != 200: + return { + "error": f"Failed to resume SLA: {response.status_code} - {response.text}" + } + return {"status": "SUCCESS", "message": f"Resumed SLA on case {case_id}"} + except Exception as e: + logger.error("Error resuming SLA on case: %s", e) + return {"error": f"Failed to resume SLA: {str(e)}"} + + +@server.tool() +async def close_case( + case_ids: Union[str, List[str]], + closure_reason: str, + root_cause: Optional[str] = None, + comment: Optional[str] = None, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Close one or more cases with reason and root cause using Chronicle 1P REST API. + + Args: + case_ids (Union[str, List[str]]): Single case ID or list of case IDs. + closure_reason (str): Reason for closing the case (e.g. "False Positive", "Resolved", "Maintenance"). + root_cause (Optional[str]): Detailed root cause description. + comment (Optional[str]): Closure commentary or summary note. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Status confirmation. + """ + try: + chronicle = get_chronicle_client(project_id, customer_id, region) + ids = [case_ids] if isinstance(case_ids, str) else case_ids + if not ids: + return {"error": "case_ids cannot be empty"} + if not closure_reason: + return {"error": "closure_reason parameter is required"} + + full_names = [_format_case_name(chronicle.instance_id, cid) for cid in ids] + url = f"{_get_base_endpoint(chronicle)}/cases:executeBulkClose" + + body: Dict[str, Any] = { + "names": full_names, + "closureReason": closure_reason, + } + if root_cause: + body["rootCause"] = root_cause + if comment: + body["comment"] = comment + + response = chronicle.session.post(url, json=body) + if response.status_code != 200: + return { + "error": f"Failed to close case(s): {response.status_code} - {response.text}" + } + return {"status": "SUCCESS", "message": f"Closed {len(full_names)} case(s)"} + except Exception as e: + logger.error("Error closing case(s): %s", e) + return {"error": f"Failed to close case: {str(e)}"} + + +@server.tool() +async def reopen_case( + case_ids: Union[str, List[str]], + comment: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Reopen one or more closed cases with an explanatory comment using Chronicle 1P REST API. + + Args: + case_ids (Union[str, List[str]]): Single case ID or list of case IDs. + comment (str): Reason explaining why the case is being reopened. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Status confirmation. + """ + try: + chronicle = get_chronicle_client(project_id, customer_id, region) + ids = [case_ids] if isinstance(case_ids, str) else case_ids + if not ids: + return {"error": "case_ids cannot be empty"} + if not comment: + return {"error": "comment parameter is required explaining why the case is being reopened"} + + full_names = [_format_case_name(chronicle.instance_id, cid) for cid in ids] + url = f"{_get_base_endpoint(chronicle)}/cases:executeBulkReopen" + + body = {"names": full_names, "comment": comment} + response = chronicle.session.post(url, json=body) + if response.status_code != 200: + return { + "error": f"Failed to reopen case(s): {response.status_code} - {response.text}" + } + return {"status": "SUCCESS", "message": f"Reopened {len(full_names)} case(s)"} + except Exception as e: + logger.error("Error reopening case(s): %s", e) + return {"error": f"Failed to reopen case: {str(e)}"} + + +@server.tool() +async def list_case_comments( + case_id: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, + filter_query: Optional[str] = None, + order_by: Optional[str] = None, + page_size: int = 50, + page_token: Optional[str] = None, +) -> Dict[str, Any]: + """List all case comments for a given Case in Chronicle using 1P CaseCommentService. + + Retrieves a paginated list of comments associated with a specific SOAR case, + essential for understanding the timeline of investigation and reviewing analyst notes. + + Args: + case_id (str): The Case ID or full resource name. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + filter_query (Optional[str]): Filter expression (e.g., 'user = "analyst@example.com"'). + order_by (Optional[str]): Sort order (e.g., "create_time desc"). + page_size (int): Max number of comments to return. Defaults to 50. + page_token (Optional[str]): Pagination token for next page. + + Returns: + Dict[str, Any]: List of CaseComment objects and pagination token. + """ + try: + if not case_id: + return {"error": "case_id parameter is required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + case_name = _format_case_name(chronicle.instance_id, case_id) + short_id = case_name.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_id}/caseComments" + + params: Dict[str, Any] = {"pageSize": page_size} + if filter_query: + params["filter"] = filter_query + if order_by: + params["orderBy"] = order_by + if page_token: + params["pageToken"] = page_token + + response = chronicle.session.get(url, params=params) + if response.status_code != 200: + return { + "error": f"Failed to list case comments: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error listing comments for case %s: %s", case_id, e) + return {"error": f"Failed to list case comments: {str(e)}"} + + +@server.tool() +async def create_case_comment( + case_id: str, + comment: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Create a new comment on a Case in Chronicle using 1P CaseCommentService. + + Adds a new structured comment to an existing SOAR case for documenting findings, + decisions, and analyst notes. + + Args: + case_id (str): The Case ID or full resource name. + comment (str): The text content of the comment. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: The created CaseComment object. + """ + try: + if not case_id or not comment: + return {"error": "Both case_id and comment parameters are required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + case_name = _format_case_name(chronicle.instance_id, case_id) + short_id = case_name.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_id}/caseComments" + + body = {"comment": comment} + response = chronicle.session.post(url, json=body) + if response.status_code != 200: + return { + "error": f"Failed to create case comment: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error creating comment for case %s: %s", case_id, e) + return {"error": f"Failed to create case comment: {str(e)}"} + + +@server.tool() +async def post_case_comment( + case_id: str, + comment: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Alias for create_case_comment. Posts a comment to a specific case within SOAR.""" + return await create_case_comment( + case_id=case_id, + comment=comment, + project_id=project_id, + customer_id=customer_id, + region=region, + ) + + +@server.tool() +async def get_case_full_details( + case_id: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Retrieve consolidated, full details for a case by aggregating metadata, alerts, and comments. + + Fetches the core case object, associated security alerts, and comment timeline in parallel, + providing a comprehensive 360-degree incident investigation overview in a single call. + + Args: + case_id (str): The Case ID or full resource name. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Aggregated dictionary containing `case_details`, `case_alerts`, + `case_comments`, and `alert_count`. + """ + import asyncio + try: + if not case_id: + return {"error": "case_id parameter is required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + case_name = _format_case_name(chronicle.instance_id, case_id) + short_id = case_name.split("/")[-1] + base = _get_base_endpoint(chronicle) + + case_url = f"{base}/cases/{short_id}" + alerts_url = f"{base}/cases/{short_id}/caseAlerts" + comments_url = f"{base}/cases/{short_id}/caseComments" + + # Execute requests in parallel using thread executor / session calls + loop = asyncio.get_event_loop() + case_fut = loop.run_in_executor(None, lambda: chronicle.session.get(case_url)) + alerts_fut = loop.run_in_executor(None, lambda: chronicle.session.get(alerts_url)) + comments_fut = loop.run_in_executor(None, lambda: chronicle.session.get(comments_url)) + + case_res, alerts_res, comments_res = await asyncio.gather( + case_fut, alerts_fut, comments_fut, return_exceptions=True + ) + + def _safe_json(res): + if isinstance(res, Exception): + return {"error": str(res)} + if hasattr(res, "status_code") and res.status_code == 200: + return res.json() + return {"status_code": getattr(res, "status_code", None), "text": getattr(res, "text", str(res))} + + case_data = _safe_json(case_res) + alerts_data = _safe_json(alerts_res) + comments_data = _safe_json(comments_res) + + alerts_list = alerts_data.get("caseAlerts", []) if isinstance(alerts_data, dict) else [] + + return { + "case_id": short_id, + "case_details": case_data, + "case_alerts": alerts_data, + "alert_count": len(alerts_list), + "case_comments": comments_data, + } + except Exception as e: + logger.error("Error retrieving full details for case %s: %s", case_id, e) + return {"error": f"Failed to get full case details: {str(e)}"} + diff --git a/server/secops/secops_mcp/tools/connector_event_management.py b/server/secops/secops_mcp/tools/connector_event_management.py new file mode 100644 index 00000000..e5236694 --- /dev/null +++ b/server/secops/secops_mcp/tools/connector_event_management.py @@ -0,0 +1,119 @@ +# Copyright 2025 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 SOAR Connector Events.""" + +import logging +from typing import Any, Dict, Optional + +from secops_mcp.server import get_chronicle_client, server + +logger = logging.getLogger("secops-mcp") + + +def _get_base_endpoint(chronicle: Any, version: str = "v1") -> str: + """Get the base REST endpoint for Chronicle 1P resources.""" + if version == "v1" and hasattr(chronicle, "base_v1_url") and chronicle.base_v1_url: + return f"{chronicle.base_v1_url}/{chronicle.instance_id}" + base_url = chronicle.base_url + if version == "v1" and "/v1alpha" in base_url: + base_url = base_url.replace("/v1alpha", "/v1") + return f"{base_url}/{chronicle.instance_id}" + + +@server.tool() +async def list_connector_events( + connector_id: Optional[str] = None, + filter_query: Optional[str] = None, + order_by: Optional[str] = None, + page_size: int = 50, + page_token: Optional[str] = None, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """List connector events ingested through Chronicle SOAR connectors. + + Args: + connector_id (Optional[str]): Optional filter by connector identifier. + filter_query (Optional[str]): Filter expression. + order_by (Optional[str]): Sort order (e.g. "create_time desc"). + page_size (int): Max events to return. Defaults to 50. + page_token (Optional[str]): Pagination token. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: List of connector event objects and pagination metadata. + """ + try: + chronicle = get_chronicle_client(project_id, customer_id, region) + url = f"{_get_base_endpoint(chronicle)}/connectorEvents" + + params: Dict[str, Any] = {"pageSize": page_size} + if connector_id: + params["connectorId"] = connector_id + if filter_query: + params["filter"] = filter_query + if order_by: + params["orderBy"] = order_by + if page_token: + params["pageToken"] = page_token + + response = chronicle.session.get(url, params=params) + if response.status_code != 200: + return { + "error": f"Failed to list connector events: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error listing connector events: %s", e) + return {"error": f"Failed to list connector events: {str(e)}"} + + +@server.tool() +async def get_connector_event( + connector_event_id: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Retrieve full details of a specific connector event. + + Args: + connector_event_id (str): The unique connector event ID or resource name. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Detailed connector event object. + """ + try: + if not connector_event_id: + return {"error": "connector_event_id parameter is required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + short_id = connector_event_id.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/connectorEvents/{short_id}" + + response = chronicle.session.get(url) + if response.status_code != 200: + return { + "error": f"Failed to get connector event: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error getting connector event %s: %s", connector_event_id, e) + return {"error": f"Failed to get connector event: {str(e)}"} diff --git a/server/secops/secops_mcp/tools/entity_investigation.py b/server/secops/secops_mcp/tools/entity_investigation.py new file mode 100644 index 00000000..af4463cc --- /dev/null +++ b/server/secops/secops_mcp/tools/entity_investigation.py @@ -0,0 +1,287 @@ +# Copyright 2025 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 SOAR Entity Investigation and Discovery.""" + +import logging +from typing import Any, Dict, List, Optional + +from secops_mcp.server import get_chronicle_client, server + +logger = logging.getLogger("secops-mcp") + + +def _get_base_endpoint(chronicle: Any, version: str = "v1") -> str: + """Get the base REST endpoint for Chronicle 1P resources.""" + if version == "v1" and hasattr(chronicle, "base_v1_url") and chronicle.base_v1_url: + return f"{chronicle.base_v1_url}/{chronicle.instance_id}" + base_url = chronicle.base_url + if version == "v1" and "/v1alpha" in base_url: + base_url = base_url.replace("/v1alpha", "/v1") + return f"{base_url}/{chronicle.instance_id}" + + +@server.tool() +async def get_involved_entity( + case_id: str, + alert_id: str, + involved_entity_id: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Retrieve detailed properties of an involved entity associated with a case alert. + + Args: + case_id (str): The Case ID or full resource name. + alert_id (str): The Alert ID or full resource name. + involved_entity_id (str): The unique involved entity identifier. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Detailed InvolvedEntity object including type, identifier, + suspicious flag, environment, and enrichments. + """ + try: + if not case_id or not alert_id or not involved_entity_id: + return {"error": "case_id, alert_id, and involved_entity_id parameters are all required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + short_case_id = case_id.split("/")[-1] + short_alert_id = alert_id.split("/")[-1] + short_entity_id = involved_entity_id.split("/")[-1] + + url = ( + f"{_get_base_endpoint(chronicle)}/cases/{short_case_id}/caseAlerts/" + f"{short_alert_id}/involvedEntities/{short_entity_id}" + ) + + response = chronicle.session.get(url) + if response.status_code != 200: + return { + "error": f"Failed to get involved entity: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error getting involved entity %s: %s", involved_entity_id, e) + return {"error": f"Failed to get involved entity: {str(e)}"} + + +@server.tool() +async def list_involved_entities( + case_id: str, + alert_id: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, + filter_query: Optional[str] = None, + page_size: int = 50, + page_token: Optional[str] = None, +) -> Dict[str, Any]: + """List all entities (IPs, domains, hashes, users, hostnames) involved in a case alert. + + Args: + case_id (str): The Case ID or full resource name. + alert_id (str): The Alert ID or full resource name. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + filter_query (Optional[str]): Filter expression. + page_size (int): Max entities to return. Defaults to 50. + page_token (Optional[str]): Pagination token. + + Returns: + Dict[str, Any]: List of InvolvedEntity objects and pagination token. + """ + try: + if not case_id or not alert_id: + return {"error": "Both case_id and alert_id parameters are required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + short_case_id = case_id.split("/")[-1] + short_alert_id = alert_id.split("/")[-1] + + url = f"{_get_base_endpoint(chronicle)}/cases/{short_case_id}/caseAlerts/{short_alert_id}/involvedEntities" + params: Dict[str, Any] = {"pageSize": page_size} + if filter_query: + params["filter"] = filter_query + if page_token: + params["pageToken"] = page_token + + response = chronicle.session.get(url, params=params) + if response.status_code != 200: + return { + "error": f"Failed to list involved entities: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error listing involved entities for alert %s: %s", alert_id, e) + return {"error": f"Failed to list involved entities: {str(e)}"} + + +@server.tool() +async def get_entities_by_alert_group_identifiers( + case_id: str, + alert_group_identifiers: List[str], + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Retrieve entities involved in specific alert groups within a Case. + + Args: + case_id (str): The Case ID or full resource name. + alert_group_identifiers (List[str]): List of alert group identifiers. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Entities linked to the specified alert group identifiers. + """ + try: + if not case_id or not alert_group_identifiers: + return {"error": "Both case_id and alert_group_identifiers parameters are required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + short_case_id = case_id.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_case_id}:getEntitiesByAlertGroupIdentifiers" + + body = {"caseId": short_case_id, "alertGroupIdentifiers": alert_group_identifiers} + response = chronicle.session.post(url, json=body) + if response.status_code != 200: + return { + "error": f"Failed to get entities by alert group identifiers: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error getting entities by alert groups for case %s: %s", case_id, e) + return {"error": f"Failed to get entities by alert group identifiers: {str(e)}"} + + +@server.tool() +async def get_entity_details( + entity_identifier: str, + entity_type: str, + entity_environment: Optional[str] = None, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Fetch detailed information and enrichments for a specific entity known to the SOAR platform. + + Args: + entity_identifier (str): The unique identifier of the entity (e.g., "192.168.1.100", "user@corp.com"). + entity_type (str): The type of the entity (e.g., "IP Address", "Hostname", "User", "Hash"). + entity_environment (Optional[str]): The environment context (e.g., "Production", "Corporate"). + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Detailed entity attributes, risk scores, and enrichment metadata. + """ + try: + if not entity_identifier or not entity_type: + return {"error": "Both entity_identifier and entity_type parameters are required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + url = f"{_get_base_endpoint(chronicle)}:fetchFullEntityDetails" + + body = { + "entityIdentifier": entity_identifier, + "entityType": entity_type, + "entityEnvironment": entity_environment or "Default", + } + response = chronicle.session.post(url, json=body) + if response.status_code != 200: + return { + "error": f"Failed to get entity details: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error getting entity details for %s: %s", entity_identifier, e) + return {"error": f"Failed to get entity details: {str(e)}"} + + +@server.tool() +async def search_entity( + term: Optional[str] = None, + entity_types: Optional[List[str]] = None, + is_suspicious: Optional[bool] = None, + is_internal_asset: Optional[bool] = None, + is_enriched: Optional[bool] = None, + network_name: Optional[List[str]] = None, + environment_name: Optional[List[str]] = None, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, + page_size: int = 50, +) -> Dict[str, Any]: + """Search for entities across the SOAR platform matching specific attributes and flags. + + Args: + term (Optional[str]): Partial string to match against entity names or identifiers. + entity_types (Optional[List[str]]): List of types to filter by (e.g. ['IP Address', 'Hostname']). + is_suspicious (Optional[bool]): Filter for entities flagged as suspicious. + is_internal_asset (Optional[bool]): Filter for internal assets. + is_enriched (Optional[bool]): Filter for entities with threat intelligence enrichment. + network_name (Optional[List[str]]): Filter by network identifiers. + environment_name (Optional[List[str]]): Filter by environment names. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + page_size (int): Max entities to return. Defaults to 50. + + Returns: + Dict[str, Any]: List of matching entity objects. + """ + try: + chronicle = get_chronicle_client(project_id, customer_id, region) + url = f"{_get_base_endpoint(chronicle, version='v1alpha')}:searchEntities" + + # 1. Try standard GET with indicator query parameter + if term: + get_params = {"indicator": term, "pageSize": page_size} + get_resp = chronicle.session.get(url, params=get_params) + if get_resp.status_code == 200: + return get_resp.json() + + # 2. Fall back to structured POST request + body: Dict[str, Any] = {"pageSize": page_size} + if term: + body["term"] = term + if entity_types: + body["type"] = entity_types + if is_suspicious is not None: + body["isSuspicious"] = is_suspicious + if is_internal_asset is not None: + body["isInternalAsset"] = is_internal_asset + if is_enriched is not None: + body["isEnriched"] = is_enriched + if network_name: + body["networkName"] = network_name + if environment_name: + body["environmentName"] = environment_name + + response = chronicle.session.post(url, json=body) + if response.status_code != 200: + return { + "error": f"Failed to search entities: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error searching entities: %s", e) + return {"error": f"Failed to search entities: {str(e)}"} diff --git a/server/secops/secops_mcp/tools/integration_management.py b/server/secops/secops_mcp/tools/integration_management.py new file mode 100644 index 00000000..1f9bc05c --- /dev/null +++ b/server/secops/secops_mcp/tools/integration_management.py @@ -0,0 +1,331 @@ +# Copyright 2025 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 SOAR Integrations and Manual Action Execution.""" + +import logging +from typing import Any, Dict, List, Optional + +from secops_mcp.server import get_chronicle_client, server + +logger = logging.getLogger("secops-mcp") + + +def _get_base_endpoint(chronicle: Any, version: str = "v1") -> str: + """Get the base REST endpoint for Chronicle 1P resources.""" + if version == "v1" and hasattr(chronicle, "base_v1_url") and chronicle.base_v1_url: + return f"{chronicle.base_v1_url}/{chronicle.instance_id}" + base_url = chronicle.base_url + if version == "v1" and "/v1alpha" in base_url: + base_url = base_url.replace("/v1alpha", "/v1") + return f"{base_url}/{chronicle.instance_id}" + + +@server.tool() +async def list_integrations( + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, + filter_query: Optional[str] = None, + order_by: Optional[str] = None, + page_size: int = 50, + page_token: Optional[str] = None, +) -> Dict[str, Any]: + """List all SOAR Integrations configured for a Chronicle instance. + + Retrieves a paginated list of third-party tool connections (EDR, Firewall, SIEM, TI, Ticketing). + + Args: + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + filter_query (Optional[str]): Filter expression (e.g. 'Identifier = "SiemplifyUtilities"'). + order_by (Optional[str]): Sort order (e.g. "DisplayName asc"). + page_size (int): Max results per page. Defaults to 50. + page_token (Optional[str]): Pagination token. + + Returns: + Dict[str, Any]: List of integration objects and pagination metadata. + """ + try: + chronicle = get_chronicle_client(project_id, customer_id, region) + url = f"{_get_base_endpoint(chronicle)}/integrations" + + params: Dict[str, Any] = {"pageSize": page_size} + if filter_query: + params["filter"] = filter_query + if order_by: + params["orderBy"] = order_by + if page_token: + params["pageToken"] = page_token + + response = chronicle.session.get(url, params=params) + if response.status_code != 200: + return { + "error": f"Failed to list integrations: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error listing integrations: %s", e) + return {"error": f"Failed to list integrations: {str(e)}"} + + +@server.tool() +async def list_integration_actions( + integration_id: str = "-", + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, + filter_query: Optional[str] = None, + order_by: Optional[str] = None, + page_size: int = 50, + page_token: Optional[str] = None, +) -> Dict[str, Any]: + """List all actions provided by a SOAR Integration (or across all integrations if integration_id='-'). + + Args: + integration_id (str): Integration ID or '-' for all integrations. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + filter_query (Optional[str]): Filter expression. + order_by (Optional[str]): Sort order. + page_size (int): Max results per page. Defaults to 50. + page_token (Optional[str]): Pagination token. + + Returns: + Dict[str, Any]: List of integration actions (e.g. 'block_ip', 'get_user_details'). + """ + try: + chronicle = get_chronicle_client(project_id, customer_id, region) + short_id = integration_id.split("/")[-1] if integration_id != "-" else "-" + url = f"{_get_base_endpoint(chronicle)}/integrations/{short_id}/actions" + + params: Dict[str, Any] = {"pageSize": page_size} + if filter_query: + params["filter"] = filter_query + if order_by: + params["orderBy"] = order_by + if page_token: + params["pageToken"] = page_token + + response = chronicle.session.get(url, params=params) + if response.status_code != 200: + return { + "error": f"Failed to list integration actions: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error listing integration actions: %s", e) + return {"error": f"Failed to list integration actions: {str(e)}"} + + +@server.tool() +async def list_integration_instances( + integration_id: str = "-", + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, + filter_query: Optional[str] = None, + order_by: Optional[str] = None, + page_size: int = 50, + page_token: Optional[str] = None, +) -> Dict[str, Any]: + """List configured instances of an integration to retrieve instance GUIDs for action execution. + + Args: + integration_id (str): Integration ID or '-' for all integrations. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + filter_query (Optional[str]): Filter expression. + order_by (Optional[str]): Sort order. + page_size (int): Max instances to return. Defaults to 50. + page_token (Optional[str]): Pagination token. + + Returns: + Dict[str, Any]: List of integration instances with instance GUIDs. + """ + try: + chronicle = get_chronicle_client(project_id, customer_id, region) + short_id = integration_id.split("/")[-1] if integration_id != "-" else "-" + url = f"{_get_base_endpoint(chronicle)}/integrations/{short_id}/integrationInstances" + + params: Dict[str, Any] = {"pageSize": page_size} + if filter_query: + params["filter"] = filter_query + if order_by: + params["orderBy"] = order_by + if page_token: + params["pageToken"] = page_token + + response = chronicle.session.get(url, params=params) + if response.status_code != 200: + # Fallback to alternative instances subpath + alt_url = f"{_get_base_endpoint(chronicle)}/integrations/{short_id}/instances" + alt_resp = chronicle.session.get(alt_url, params=params) + if alt_resp.status_code == 200: + return alt_resp.json() + return { + "error": f"Failed to list integration instances: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error listing integration instances: %s", e) + return {"error": f"Failed to list integration instances: {str(e)}"} + + +@server.tool() +async def execute_integration_instance_test( + integration_id: str, + instance_id: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Execute a live connectivity test ('ping') for a specific integration instance. + + Verifies that Google SecOps can successfully communicate with the third-party + service (e.g. VirusTotal, Slack, SIEM/EDR) using the configured credentials. + + Args: + integration_id (str): Integration identifier (e.g. 'VirusTotalV3', 'Tools'). + instance_id (str): Integration instance identifier or GUID. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Test result containing success status and message. + """ + try: + chronicle = get_chronicle_client(project_id, customer_id, region) + short_integ = integration_id.split("/")[-1] + short_inst = instance_id.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/integrations/{short_integ}/integrationInstances/{short_inst}:executeTest" + + response = chronicle.session.post(url, json={}) + if response.status_code != 200: + return { + "error": f"Failed to execute integration test: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error executing integration test on %s: %s", integration_id, e) + return {"error": f"Failed to execute integration test: {str(e)}"} + + +@server.tool() +async def execute_manual_action( + case_id: str, + action_name: str, + action_provider: str = "Scripts", + properties: Optional[Dict[str, Any]] = None, + target_entities: Optional[List[Dict[str, Any]]] = None, + alert_group_identifiers: Optional[List[str]] = None, + scope: Optional[str] = None, + is_predefined_scope: bool = True, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Execute a manual SOAR action or script on a case, alert, or target entities. + + Executes response actions such as isolating hosts, blocking IPs, disabling accounts, + or executing threat intelligence queries. + + Args: + case_id (str): The Case ID. + action_name (str): Action name (e.g. 'SiemplifyUtilities_Ping', 'VirusTotal_Enrich IP'). + action_provider (str): Provider name, typically 'Scripts'. Defaults to 'Scripts'. + properties (Optional[Dict[str, Any]]): Dictionary containing `ScriptName`, `IntegrationInstance` GUID, + and `ScriptParametersEntityFields` JSON string. + target_entities (Optional[List[Dict[str, Any]]]): Entities targeted by this action. + alert_group_identifiers (Optional[List[str]]): Alert group identifiers associated with action. + scope (Optional[str]): Action scope (e.g. 'All entities'). + is_predefined_scope (bool): Whether predefined scope is used. Defaults to True. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Action execution result or tracking ID. + """ + try: + if not case_id or not action_name: + return {"error": "Both case_id and action_name parameters are required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + short_case_id = case_id.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_case_id}:executeManualAction" + + body: Dict[str, Any] = { + "caseId": short_case_id, + "actionName": action_name, + "actionProvider": action_provider, + "isPredefinedScope": is_predefined_scope, + } + if properties: + body["properties"] = properties + if target_entities: + body["targetEntities"] = target_entities + if alert_group_identifiers: + body["alertGroupIdentifiers"] = alert_group_identifiers + if scope: + body["scope"] = scope + + response = chronicle.session.post(url, json=body) + if response.status_code != 200: + return { + "error": f"Failed to execute manual action: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error executing manual action %s: %s", action_name, e) + return {"error": f"Failed to execute manual action: {str(e)}"} + + +@server.tool() +async def get_action_result_by_id( + action_result_id: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Retrieve the result and execution logs of an asynchronous SOAR action execution by its result ID. + + Args: + action_result_id (str): The action result execution ID. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Execution status, output data, logs, and affected entities. + """ + try: + if not action_result_id: + return {"error": "action_result_id parameter is required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + url = f"{_get_base_endpoint(chronicle)}/actionResults/{action_result_id}" + + response = chronicle.session.get(url) + if response.status_code != 200: + return { + "error": f"Failed to get action result: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error getting action result %s: %s", action_result_id, e) + return {"error": f"Failed to get action result: {str(e)}"} diff --git a/server/secops/secops_mcp/tools/playbook_management.py b/server/secops/secops_mcp/tools/playbook_management.py new file mode 100644 index 00000000..21214284 --- /dev/null +++ b/server/secops/secops_mcp/tools/playbook_management.py @@ -0,0 +1,245 @@ +# Copyright 2025 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 SOAR Playbook Management and Execution.""" + +import logging +from typing import Any, Dict, List, Optional + +from secops_mcp.server import get_chronicle_client, server + +logger = logging.getLogger("secops-mcp") + + +def _get_base_endpoint(chronicle: Any, version: str = "v1") -> str: + """Get the base REST endpoint for Chronicle 1P resources.""" + if version == "v1" and hasattr(chronicle, "base_v1_url") and chronicle.base_v1_url: + return f"{chronicle.base_v1_url}/{chronicle.instance_id}" + base_url = chronicle.base_url + if version == "v1" and "/v1alpha" in base_url: + base_url = base_url.replace("/v1alpha", "/v1") + return f"{base_url}/{chronicle.instance_id}" + + +@server.tool() +async def list_playbooks( + playbook_types: Optional[List[str]] = None, + filter_query: Optional[str] = None, + page_size: int = 50, + page_token: Optional[str] = None, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """List all available playbooks (automated workflows) configured in the Chronicle SOAR instance. + + Args: + playbook_types (Optional[List[str]]): Filter by playbook types (e.g. ['REGULAR', 'NESTED']). + filter_query (Optional[str]): Additional filter expression. + page_size (int): Max playbooks to return. Defaults to 50. + page_token (Optional[str]): Pagination token. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: List of playbook objects and metadata. + """ + try: + chronicle = get_chronicle_client(project_id, customer_id, region) + # 1. Try Google OnePlatform v1alpha legacyPlaybooks RPC first + v1alpha_base = f"{_get_base_endpoint(chronicle, version='v1alpha')}" + legacy_url = f"{v1alpha_base}/legacyPlaybooks:legacyGetWorkflowMenuCardsWithEnvFilter" + legacy_body = {"legacyPayload": playbook_types or ["REGULAR", "NESTED"]} + + legacy_resp = chronicle.session.post(legacy_url, json=legacy_body) + if legacy_resp.status_code == 200: + return legacy_resp.json() + + # 2. Fall back to standard v1 playbooks REST endpoint + url = f"{_get_base_endpoint(chronicle)}/playbooks" + params: Dict[str, Any] = {"pageSize": page_size} + if playbook_types: + params["playbookTypes"] = playbook_types + if filter_query: + params["filter"] = filter_query + if page_token: + params["pageToken"] = page_token + + response = chronicle.session.get(url, params=params) + if response.status_code != 200: + return { + "error": f"Failed to list playbooks: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error listing playbooks: %s", e) + return {"error": f"Failed to list playbooks: {str(e)}"} + + +@server.tool() +async def get_playbook( + playbook_id: str, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Get full details, configuration, trigger conditions, and step definitions for a specific playbook. + + Args: + playbook_id (str): The unique Playbook ID or resource name. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Detailed Playbook object. + """ + try: + if not playbook_id: + return {"error": "playbook_id parameter is required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + short_id = playbook_id.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/playbooks/{short_id}" + + response = chronicle.session.get(url) + if response.status_code != 200: + return { + "error": f"Failed to get playbook: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error getting playbook %s: %s", playbook_id, e) + return {"error": f"Failed to get playbook: {str(e)}"} + + +@server.tool() +async def list_playbook_instances( + case_id: str, + alert_group_identifier: Optional[str] = None, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """List historical and active playbook execution instances for a given Case and/or Alert Group. + + Args: + case_id (str): The Case ID or full resource name. + alert_group_identifier (Optional[str]): Optional alert group identifier to filter executions. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: List of playbook execution cards and step outcomes. + """ + try: + if not case_id: + return {"error": "case_id parameter is required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + short_case_id = case_id.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_case_id}:listPlaybookInstances" + + params: Dict[str, Any] = {} + if alert_group_identifier: + params["alertGroupIdentifier"] = alert_group_identifier + + response = chronicle.session.get(url, params=params) + if response.status_code != 200: + return { + "error": f"Failed to list playbook instances: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error listing playbook instances for case %s: %s", case_id, e) + return {"error": f"Failed to list playbook instances: {str(e)}"} + + +@server.tool() +async def execute_playbook( + case_id: str, + playbook_id: str, + alert_group_identifier: Optional[str] = None, + scope: Optional[str] = None, + target_entities: Optional[List[Dict[str, Any]]] = None, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Execute/trigger a specific SOAR playbook manually on a Case or Alert. + + Args: + case_id (str): The Case ID to run the playbook on. + playbook_id (str): The ID of the playbook to execute. + alert_group_identifier (Optional[str]): Optional target alert group identifier. + scope (Optional[str]): Execution scope (e.g. 'All entities', 'Specific entities'). + target_entities (Optional[List[Dict[str, Any]]]): Entities to pass to the playbook execution. + project_id (Optional[str]): Google Cloud project ID. + customer_id (Optional[str]): Chronicle customer/instance ID. + region (Optional[str]): Chronicle region. + + Returns: + Dict[str, Any]: Playbook execution tracking status and instance ID. + """ + try: + if not case_id or not playbook_id: + return {"error": "Both case_id and playbook_id parameters are required"} + + chronicle = get_chronicle_client(project_id, customer_id, region) + short_case_id = case_id.split("/")[-1] + short_playbook_id = playbook_id.split("/")[-1] + url = f"{_get_base_endpoint(chronicle)}/cases/{short_case_id}/playbooks/{short_playbook_id}:execute" + + body: Dict[str, Any] = {"playbookId": short_playbook_id} + if alert_group_identifier: + body["alertGroupIdentifier"] = alert_group_identifier + if scope: + body["scope"] = scope + if target_entities: + body["targetEntities"] = target_entities + + response = chronicle.session.post(url, json=body) + if response.status_code != 200: + return { + "error": f"Failed to execute playbook: {response.status_code} - {response.text}" + } + return response.json() + except Exception as e: + logger.error("Error executing playbook %s on case %s: %s", playbook_id, case_id, e) + return {"error": f"Failed to execute playbook: {str(e)}"} + + +@server.tool() +async def trigger_playbook( + case_id: str, + playbook_id: str, + alert_group_identifier: Optional[str] = None, + scope: Optional[str] = None, + target_entities: Optional[List[Dict[str, Any]]] = None, + project_id: Optional[str] = None, + customer_id: Optional[str] = None, + region: Optional[str] = None, +) -> Dict[str, Any]: + """Alias for execute_playbook. Manually triggers a SOAR playbook workflow on a case or alert.""" + return await execute_playbook( + case_id=case_id, + playbook_id=playbook_id, + alert_group_identifier=alert_group_identifier, + scope=scope, + target_entities=target_entities, + project_id=project_id, + customer_id=customer_id, + region=region, + ) diff --git a/server/secops/tests/test_secops_case_management.py b/server/secops/tests/test_secops_case_management.py new file mode 100644 index 00000000..46488374 --- /dev/null +++ b/server/secops/tests/test_secops_case_management.py @@ -0,0 +1,321 @@ +# Copyright 2025 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 1P Case and Case Alert Management MCP tools.""" + +import os +import sys +import unittest +from unittest.mock import MagicMock, patch + +# Ensure server/secops is in sys.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 secops if not installed +try: + import secops +except ImportError: + mock_secops = MagicMock() + sys.modules["secops"] = mock_secops + sys.modules["secops.chronicle"] = MagicMock() + sys.modules["secops.exceptions"] = MagicMock() + +# Mock mcp if not installed +try: + import mcp + import mcp.server.fastmcp +except ImportError: + 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 + +# Mock pytest if not installed +try: + import pytest +except ImportError: + mock_pytest = MagicMock() + mock_pytest.mark.asyncio = lambda f: f + mock_pytest.fixture = lambda f: f + sys.modules["pytest"] = mock_pytest + pytest = mock_pytest + +from secops_mcp.tools.case_management import ( + list_cases, + get_case, + update_case, + change_case_priority, + change_case_stage, + assign_case, + set_custom_case_fields, + add_case_tag, + remove_case_tag, + add_case_insight, + pause_case_sla, + resume_case_sla, + close_case, + reopen_case, +) +from secops_mcp.tools.case_alert_management import ( + list_case_alerts, + get_case_alert, + update_case_alert, + change_alert_priority, + set_alert_custom_fields, + move_case_alert, + add_alert_tag, + remove_alert_tag, +) + + +@pytest.fixture +def mock_chronicle(): + client = MagicMock() + client.instance_id = "projects/test-proj/locations/us/instances/test-cust" + client.base_url = "https://chronicle.googleapis.com/v1alpha" + client.base_v1_url = "https://chronicle.googleapis.com/v1" + + # Mock response + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.text = '{"name": "test"}' + mock_resp.json.return_value = {"name": "test", "cases": []} + client.session.get.return_value = mock_resp + client.session.post.return_value = mock_resp + client.session.patch.return_value = mock_resp + + return client + + +# --- Case Management Tool Tests --- + + +@pytest.mark.asyncio +async def test_list_cases(mock_chronicle): + with patch("secops_mcp.tools.case_management.get_chronicle_client", return_value=mock_chronicle): + result = await list_cases( + project_id="test-proj", + customer_id="test-cust", + region="us", + filter_query='priority = "HIGH"', + page_size=25, + ) + assert "error" not in result + mock_chronicle.session.get.assert_called_once() + args, kwargs = mock_chronicle.session.get.call_args + assert "/cases" in args[0] + assert kwargs["params"]["pageSize"] == 25 + assert kwargs["params"]["filter"] == 'priority = "HIGH"' + + +@pytest.mark.asyncio +async def test_get_case(mock_chronicle): + with patch("secops_mcp.tools.case_management.get_chronicle_client", return_value=mock_chronicle): + result = await get_case(case_id="case_123") + assert "error" not in result + mock_chronicle.session.get.assert_called_once() + args, _ = mock_chronicle.session.get.call_args + assert "/cases/case_123" in args[0] + + +@pytest.mark.asyncio +async def test_update_case(mock_chronicle): + with patch("secops_mcp.tools.case_management.get_chronicle_client", return_value=mock_chronicle): + result = await update_case( + case_id="case_123", + update_mask="priority,stage", + priority="HIGH", + stage="Containment", + ) + assert "error" not in result + mock_chronicle.session.patch.assert_called_once() + args, kwargs = mock_chronicle.session.patch.call_args + assert "/cases/case_123" in args[0] + assert kwargs["params"]["updateMask"] == "priority,stage" + assert kwargs["json"]["priority"] == "PRIORITY_HIGH" + assert kwargs["json"]["stage"] == "Containment" + + +@pytest.mark.asyncio +async def test_change_case_priority_trigger(mock_chronicle): + with patch("secops_mcp.tools.case_management.get_chronicle_client", return_value=mock_chronicle): + result = await change_case_priority( + case_ids=["case_1", "case_2"], + priority="CRITICAL", + ) + assert result["status"] == "SUCCESS" + mock_chronicle.session.post.assert_called_once() + args, kwargs = mock_chronicle.session.post.call_args + assert ":executeBulkChangePriority" in args[0] + assert kwargs["json"]["priority"] == "PRIORITY_CRITICAL" + assert len(kwargs["json"]["names"]) == 2 + + +@pytest.mark.asyncio +async def test_change_case_stage_trigger(mock_chronicle): + with patch("secops_mcp.tools.case_management.get_chronicle_client", return_value=mock_chronicle): + result = await change_case_stage( + case_ids="case_1", + stage="Triage", + ) + assert result["status"] == "SUCCESS" + mock_chronicle.session.post.assert_called_once() + args, kwargs = mock_chronicle.session.post.call_args + assert ":executeBulkChangeStage" in args[0] + assert kwargs["json"]["stage"] == "Triage" + + +@pytest.mark.asyncio +async def test_assign_case_trigger(mock_chronicle): + with patch("secops_mcp.tools.case_management.get_chronicle_client", return_value=mock_chronicle): + result = await assign_case( + case_ids="case_1", + assignee="analyst@example.com", + ) + assert result["status"] == "SUCCESS" + mock_chronicle.session.post.assert_called_once() + args, kwargs = mock_chronicle.session.post.call_args + assert ":executeBulkAssign" in args[0] + assert kwargs["json"]["assignee"] == "analyst@example.com" + + +@pytest.mark.asyncio +async def test_set_custom_case_fields_trigger(mock_chronicle): + with patch("secops_mcp.tools.case_management.get_chronicle_client", return_value=mock_chronicle): + result = await set_custom_case_fields( + case_id="case_1", + custom_fields={"Department": "SecOps", "Impact": "High"}, + ) + assert "error" not in result + mock_chronicle.session.patch.assert_called_once() + args, kwargs = mock_chronicle.session.patch.call_args + assert "/cases/case_1" in args[0] + assert kwargs["params"]["updateMask"] == "custom_field_values" + assert len(kwargs["json"]["customFieldValues"]) == 2 + + +@pytest.mark.asyncio +async def test_add_and_remove_case_tag(mock_chronicle): + with patch("secops_mcp.tools.case_management.get_chronicle_client", return_value=mock_chronicle): + add_res = await add_case_tag(case_id="case_1", tag="Phishing") + assert "error" not in add_res + rem_res = await remove_case_tag(case_id="case_1", tag="Phishing") + assert "error" not in rem_res + + +@pytest.mark.asyncio +async def test_sla_and_case_closure(mock_chronicle): + with patch("secops_mcp.tools.case_management.get_chronicle_client", return_value=mock_chronicle): + pause_res = await pause_case_sla(case_id="case_1", reason="Waiting for customer response") + assert pause_res["status"] == "SUCCESS" + resume_res = await resume_case_sla(case_id="case_1") + assert resume_res["status"] == "SUCCESS" + + close_res = await close_case(case_ids="case_1", closure_reason="Resolved", comment="Fixed") + assert close_res["status"] == "SUCCESS" + reopen_res = await reopen_case(case_ids="case_1", comment="New activity detected") + assert reopen_res["status"] == "SUCCESS" + + +# --- Case Alert Management Tool Tests --- + + +@pytest.mark.asyncio +async def test_list_case_alerts(mock_chronicle): + with patch("secops_mcp.tools.case_alert_management.get_chronicle_client", return_value=mock_chronicle): + result = await list_case_alerts(case_id="case_1") + assert "error" not in result + mock_chronicle.session.get.assert_called_once() + args, _ = mock_chronicle.session.get.call_args + assert "/cases/case_1/caseAlerts" in args[0] + + +@pytest.mark.asyncio +async def test_get_case_alert(mock_chronicle): + with patch("secops_mcp.tools.case_alert_management.get_chronicle_client", return_value=mock_chronicle): + result = await get_case_alert(case_id="case_1", alert_id="alert_99") + assert "error" not in result + mock_chronicle.session.get.assert_called_once() + args, _ = mock_chronicle.session.get.call_args + assert "/cases/case_1/caseAlerts/alert_99" in args[0] + + +@pytest.mark.asyncio +async def test_change_alert_priority_trigger(mock_chronicle): + with patch("secops_mcp.tools.case_alert_management.get_chronicle_client", return_value=mock_chronicle): + result = await change_alert_priority( + case_id="case_1", + alert_id="alert_99", + priority="CRITICAL", + ) + assert "error" not in result + mock_chronicle.session.patch.assert_called_once() + args, kwargs = mock_chronicle.session.patch.call_args + assert "/cases/case_1/caseAlerts/alert_99" in args[0] + assert kwargs["params"]["updateMask"] == "priority" + assert kwargs["json"]["priority"] == "PRIORITY_CRITICAL" + + +@pytest.mark.asyncio +async def test_set_alert_custom_fields_trigger(mock_chronicle): + with patch("secops_mcp.tools.case_alert_management.get_chronicle_client", return_value=mock_chronicle): + result = await set_alert_custom_fields( + case_id="case_1", + alert_id="alert_99", + custom_fields={"MalwareFamily": "Emotet"}, + ) + assert "error" not in result + mock_chronicle.session.patch.assert_called_once() + args, kwargs = mock_chronicle.session.patch.call_args + assert "/cases/case_1/caseAlerts/alert_99" in args[0] + assert kwargs["params"]["updateMask"] == "custom_field_values" + + +@pytest.mark.asyncio +async def test_move_case_alert(mock_chronicle): + with patch("secops_mcp.tools.case_alert_management.get_chronicle_client", return_value=mock_chronicle): + result = await move_case_alert( + source_case_id="case_1", + alert_id="alert_99", + destination_case_id="case_2", + ) + assert "error" not in result + mock_chronicle.session.post.assert_called_once() + args, kwargs = mock_chronicle.session.post.call_args + assert "/cases/case_1/caseAlerts/alert_99:move" in args[0] + assert kwargs["json"]["destinationCaseId"] == "case_2" + + +@pytest.mark.asyncio +async def test_alert_tags(mock_chronicle): + with patch("secops_mcp.tools.case_alert_management.get_chronicle_client", return_value=mock_chronicle): + add_res = await add_alert_tag(case_id="case_1", alert_id="alert_99", tag="CriticalAsset") + assert "error" not in add_res + rem_res = await remove_alert_tag(case_id="case_1", alert_id="alert_99", tag="CriticalAsset") + assert "error" not in rem_res + + +# All tests are defined above as pytest async test functions + diff --git a/server/secops/tests/test_secops_soar_parity.py b/server/secops/tests/test_secops_soar_parity.py new file mode 100644 index 00000000..380d7414 --- /dev/null +++ b/server/secops/tests/test_secops_soar_parity.py @@ -0,0 +1,256 @@ +# Copyright 2025 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 SOAR Parity MCP tools in secops.""" + +import asyncio +import os +import sys +import unittest +from unittest.mock import MagicMock, patch + +# Ensure server/secops is in sys.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 secops if not installed +try: + import secops +except ImportError: + mock_secops = MagicMock() + sys.modules["secops"] = mock_secops + sys.modules["secops.chronicle"] = MagicMock() + sys.modules["secops.exceptions"] = MagicMock() + +# Mock mcp if not installed +try: + import mcp + import mcp.server.fastmcp +except ImportError: + 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 + +# Mock pytest if not installed +try: + import pytest +except ImportError: + mock_pytest = MagicMock() + mock_pytest.mark.asyncio = lambda f: f + mock_pytest.fixture = lambda f: f + sys.modules["pytest"] = mock_pytest + pytest = mock_pytest + +from secops_mcp.tools.case_management import ( + list_case_comments, + create_case_comment, + post_case_comment, + get_case_full_details, +) +from secops_mcp.tools.case_alert_management import ( + list_alert_group_identifiers_by_case, + list_events_by_alert, + list_involved_events, +) +from secops_mcp.tools.entity_investigation import ( + get_involved_entity, + list_involved_entities, + get_entities_by_alert_group_identifiers, + get_entity_details, + search_entity, +) +from secops_mcp.tools.integration_management import ( + list_integrations, + list_integration_actions, + list_integration_instances, + execute_integration_instance_test, + execute_manual_action, + get_action_result_by_id, +) +from secops_mcp.tools.playbook_management import ( + list_playbooks, + get_playbook, + list_playbook_instances, + execute_playbook, + trigger_playbook, +) +from secops_mcp.tools.connector_event_management import ( + list_connector_events, + get_connector_event, +) + + +@pytest.fixture +def mock_chronicle(): + client = MagicMock() + client.instance_id = "projects/test-proj/locations/us/instances/test-cust" + client.base_url = "https://chronicle.googleapis.com/v1alpha" + client.base_v1_url = "https://chronicle.googleapis.com/v1" + + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.text = '{"name": "test"}' + mock_resp.json.return_value = {"name": "test", "items": []} + client.session.get.return_value = mock_resp + client.session.post.return_value = mock_resp + client.session.patch.return_value = mock_resp + + return client + + +# --- Case Comments and Full Details Tests --- + + +@pytest.mark.asyncio +async def test_case_comments(mock_chronicle): + with patch("secops_mcp.tools.case_management.get_chronicle_client", return_value=mock_chronicle): + list_res = await list_case_comments(case_id="case_123") + assert "error" not in list_res + mock_chronicle.session.get.assert_called() + + create_res = await create_case_comment(case_id="case_123", comment="Test comment") + assert "error" not in create_res + mock_chronicle.session.post.assert_called() + + post_res = await post_case_comment(case_id="case_123", comment="Alias comment") + assert "error" not in post_res + + +@pytest.mark.asyncio +async def test_get_case_full_details(mock_chronicle): + with patch("secops_mcp.tools.case_management.get_chronicle_client", return_value=mock_chronicle): + res = await get_case_full_details(case_id="case_123") + assert "error" not in res + assert "case_details" in res + assert "case_alerts" in res + assert "case_comments" in res + + +# --- Alert Group Identifiers and Events Tests --- + + +@pytest.mark.asyncio +async def test_alert_group_and_events(mock_chronicle): + with patch("secops_mcp.tools.case_alert_management.get_chronicle_client", return_value=mock_chronicle): + groups_res = await list_alert_group_identifiers_by_case(case_id="case_123") + assert "error" not in groups_res + + events_res = await list_events_by_alert(case_id="case_123", alert_id="alert_456") + assert "error" not in events_res + + inv_events_res = await list_involved_events(case_id="case_123", alert_id="alert_456") + assert "error" not in inv_events_res + + +# --- Entity Investigation Tests --- + + +@pytest.mark.asyncio +async def test_entity_investigation(mock_chronicle): + with patch("secops_mcp.tools.entity_investigation.get_chronicle_client", return_value=mock_chronicle): + inv_ent = await get_involved_entity(case_id="c1", alert_id="a1", involved_entity_id="e1") + assert "error" not in inv_ent + + list_inv = await list_involved_entities(case_id="c1", alert_id="a1") + assert "error" not in list_inv + + by_groups = await get_entities_by_alert_group_identifiers(case_id="c1", alert_group_identifiers=["g1"]) + assert "error" not in by_groups + + ent_details = await get_entity_details(entity_identifier="192.168.1.1", entity_type="IP Address") + assert "error" not in ent_details + + search_res = await search_entity(term="corp", is_suspicious=True) + assert "error" not in search_res + + +# --- Integration Management Tests --- + + +@pytest.mark.asyncio +async def test_integration_management(mock_chronicle): + with patch("secops_mcp.tools.integration_management.get_chronicle_client", return_value=mock_chronicle): + integrations = await list_integrations() + assert "error" not in integrations + + actions = await list_integration_actions(integration_id="SiemplifyUtilities") + assert "error" not in actions + + instances = await list_integration_instances(integration_id="SiemplifyUtilities") + assert "error" not in instances + + inst_test = await execute_integration_instance_test( + integration_id="VirusTotalV3", + instance_id="3e9496eb-09cd-4b3c-a4ce-4c788d6663a7", + ) + assert "error" not in inst_test + + exec_action = await execute_manual_action( + case_id="c1", + action_name="SiemplifyUtilities_Ping", + properties={"ScriptName": "SiemplifyUtilities_Ping"}, + ) + assert "error" not in exec_action + + action_res = await get_action_result_by_id(action_result_id="res_123") + assert "error" not in action_res + + +# --- Playbook Management Tests --- + + +@pytest.mark.asyncio +async def test_playbook_management(mock_chronicle): + with patch("secops_mcp.tools.playbook_management.get_chronicle_client", return_value=mock_chronicle): + playbooks = await list_playbooks(playbook_types=["REGULAR"]) + assert "error" not in playbooks + + playbook = await get_playbook(playbook_id="pb_123") + assert "error" not in playbook + + instances = await list_playbook_instances(case_id="c1") + assert "error" not in instances + + exec_pb = await execute_playbook(case_id="c1", playbook_id="pb_123") + assert "error" not in exec_pb + + trig_pb = await trigger_playbook(case_id="c1", playbook_id="pb_123") + assert "error" not in trig_pb + + +# --- Connector Events Tests --- + + +@pytest.mark.asyncio +async def test_connector_events(mock_chronicle): + with patch("secops_mcp.tools.connector_event_management.get_chronicle_client", return_value=mock_chronicle): + events = await list_connector_events(connector_id="conn_1") + assert "error" not in events + + event = await get_connector_event(connector_event_id="ev_123") + assert "error" not in event + + +# All tests are defined above as pytest async test functions