feat(secops): add comprehensive SOAR parity tools for Chronicle 1P (Git Stack 2/2)

server/secops/secops_mcp/tools/__init__.py

@@ -15,14 +15,18 @@
 
 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 *

server/secops/secops_mcp/tools/case_alert_management.py

@@ -92,7 +92,7 @@ async def list_case_alerts(
             }
         return response.json()
     except Exception as e:
-        logger.error(f"Error listing case alerts for case {case_id}: {e}")
+        logger.error("Error listing case alerts for case %s: %s", case_id, e)
         return {"error": f"Failed to list case alerts: {str(e)}"}
 
 
@@ -132,7 +132,7 @@ async def get_case_alert(
             }
         return response.json()
     except Exception as e:
-        logger.error(f"Error getting case alert {alert_id}: {e}")
+        logger.error("Error getting case alert %s: %s", alert_id, e)
         return {"error": f"Failed to get case alert: {str(e)}"}
 
 
@@ -212,7 +212,7 @@ async def update_case_alert(
             }
         return response.json()
     except Exception as e:
-        logger.error(f"Error updating case alert {alert_id}: {e}")
+        logger.error("Error updating case alert %s: %s", alert_id, e)
         return {"error": f"Failed to update case alert: {str(e)}"}
 
 
@@ -256,7 +256,7 @@ async def change_alert_priority(
             region=region,
         )
     except Exception as e:
-        logger.error(f"Error changing alert priority: {e}")
+        logger.error("Error changing alert priority: %s", e)
         return {"error": f"Failed to change alert priority: {str(e)}"}
 
 
@@ -304,7 +304,7 @@ async def set_alert_custom_fields(
             region=region,
         )
     except Exception as e:
-        logger.error(f"Error setting alert custom fields: {e}")
+        logger.error("Error setting alert custom fields: %s", e)
         return {"error": f"Failed to set alert custom fields: {str(e)}"}
 
 
@@ -349,7 +349,7 @@ async def move_case_alert(
             }
         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(f"Error moving case alert: {e}")
+        logger.error("Error moving case alert: %s", e)
         return {"error": f"Failed to move case alert: {str(e)}"}
 
 
@@ -391,7 +391,7 @@ async def add_alert_tag(
             }
         return response.json() if response.text else {"status": "SUCCESS", "message": f"Added tag '{tag}' to alert"}
     except Exception as e:
-        logger.error(f"Error adding tag to alert: {e}")
+        logger.error("Error adding tag to alert: %s", e)
         return {"error": f"Failed to add tag to alert: {str(e)}"}
 
 
@@ -433,5 +433,138 @@ async def remove_alert_tag(
             }
         return response.json() if response.text else {"status": "SUCCESS", "message": f"Removed tag '{tag}' from alert"}
     except Exception as e:
-        logger.error(f"Error removing tag from alert: {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,
+    )
+