| 1 | """ |
| 2 | Service functions for threshold alert event resolution and timeline retrieval. |
| 3 | |
| 4 | When a Graylog threshold alert fires, there is no individual event _index/_id available. |
| 5 | This module uses the replay_info (Lucene query + timerange) and group_by_fields from the |
| 6 | Graylog webhook to find the first matching event in OpenSearch, and provides timeline |
| 7 | retrieval for threshold alerts using stored metadata. |
| 8 | """ |
| 9 | |
| 10 | from datetime import datetime |
| 11 | from typing import Any |
| 12 | from typing import Dict |
| 13 | from typing import List |
| 14 | from typing import Optional |
| 15 | from typing import Tuple |
| 16 | |
| 17 | from loguru import logger |
| 18 | from sqlalchemy.ext.asyncio import AsyncSession |
| 19 | from sqlalchemy.future import select |
| 20 | |
| 21 | from app.connectors.wazuh_indexer.utils.universal import ( |
| 22 | create_wazuh_indexer_client_async, |
| 23 | ) |
| 24 | from app.incidents.config.threshold_index_mapping import get_index_config_for_source |
| 25 | from app.incidents.models import AssetFieldName |
| 26 | from app.incidents.models import ThresholdAlertMetadata |
| 27 | |
| 28 | |
| 29 | def _format_datetime(dt: datetime) -> str: |
| 30 | """Format a datetime for OpenSearch range queries, matching the index time format.""" |
| 31 | return dt.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] |
| 32 | |
| 33 | |
| 34 | async def resolve_threshold_event( |
| 35 | replay_query: str, |
| 36 | timerange_start: datetime, |
| 37 | timerange_end: datetime, |
| 38 | group_by_fields: Dict[str, str], |
| 39 | source: str, |
| 40 | ) -> Tuple[str, str]: |
| 41 | """ |
| 42 | Query OpenSearch to find the first event matching a threshold alert's conditions. |
| 43 | |
| 44 | Uses the replay_info from the Graylog webhook to reconstruct the search and find |
| 45 | the actual underlying event that contributed to the threshold count. |
| 46 | |
| 47 | Args: |
| 48 | replay_query: Lucene query string from replay_info.query (e.g. "rule_id:5503"). |
| 49 | timerange_start: Start of the threshold evaluation window. |
| 50 | timerange_end: End of the threshold evaluation window. |
| 51 | group_by_fields: Group-by field values from the threshold event (e.g. {"data_dstuser": "taylor"}). |
| 52 | source: The SOURCE field from the Graylog alert (e.g. "wazuh"). |
| 53 | |
| 54 | Returns: |
| 55 | Tuple of (index_name, index_id) from the first matching hit, |
| 56 | or ("not_applicable", "not_applicable") if no event is found or source is unmapped. |
| 57 | """ |
| 58 | try: |
| 59 | index_pattern, time_field = get_index_config_for_source(source) |
| 60 | except ValueError: |
| 61 | return ("not_applicable", "not_applicable") |
| 62 | |
| 63 | es_client = await create_wazuh_indexer_client_async("Wazuh-Indexer") |
| 64 | try: |
| 65 | must_clauses: List[Dict[str, Any]] = [ |
| 66 | {"query_string": {"query": replay_query, "default_operator": "AND"}}, |
| 67 | { |
| 68 | "range": { |
| 69 | time_field: { |
| 70 | "gte": _format_datetime(timerange_start), |
| 71 | "lte": _format_datetime(timerange_end), |
| 72 | }, |
| 73 | }, |
| 74 | }, |
| 75 | ] |
| 76 | |
| 77 | # Add match clauses for each group_by_fields entry |
| 78 | for field_name, field_value in group_by_fields.items(): |
| 79 | if field_value: |
| 80 | must_clauses.append({"match": {field_name: field_value}}) |
| 81 | |
| 82 | query = { |
| 83 | "query": {"bool": {"must": must_clauses}}, |
| 84 | "sort": [{time_field: {"order": "asc"}}], |
| 85 | } |
| 86 | |
| 87 | logger.info( |
| 88 | f"Resolving threshold event for source '{source}' in index '{index_pattern}' " |
| 89 | f"with query: {replay_query}, group_by: {group_by_fields}, " |
| 90 | f"timerange: {timerange_start} -> {timerange_end}", |
| 91 | ) |
| 92 | |
| 93 | response = await es_client.search(index=index_pattern, body=query, size=1) |
| 94 | hits = response["hits"]["hits"] |
| 95 | |
| 96 | if hits: |
| 97 | hit = hits[0] |
| 98 | resolved_index = hit["_index"] |
| 99 | resolved_id = hit["_id"] |
| 100 | logger.info(f"Resolved threshold event: index={resolved_index}, id={resolved_id}") |
| 101 | return (resolved_index, resolved_id) |
| 102 | |
| 103 | logger.warning( |
| 104 | f"No matching event found for threshold alert in '{index_pattern}' " |
| 105 | f"with query '{replay_query}' and group_by {group_by_fields}", |
| 106 | ) |
| 107 | return ("not_applicable", "not_applicable") |
| 108 | |
| 109 | except Exception as e: |
| 110 | logger.error(f"Error resolving threshold event: {e}") |
| 111 | return ("not_applicable", "not_applicable") |
| 112 | finally: |
| 113 | await es_client.close() |
| 114 | |
| 115 | |
| 116 | async def resolve_threshold_asset( |
| 117 | index_name: str, |
| 118 | index_id: str, |
| 119 | source: str, |
| 120 | session: AsyncSession, |
| 121 | ) -> str: |
| 122 | """ |
| 123 | Fetch the resolved event document from OpenSearch and resolve the asset name |
| 124 | using the AssetFieldName configuration for the given source. |
| 125 | |
| 126 | Args: |
| 127 | index_name: The OpenSearch index name of the resolved event. |
| 128 | index_id: The OpenSearch document ID of the resolved event. |
| 129 | source: The source type (e.g. "wazuh"). |
| 130 | session: Database session for looking up asset field names. |
| 131 | |
| 132 | Returns: |
| 133 | The resolved asset name, or "No asset found" if resolution fails. |
| 134 | """ |
| 135 | if index_name == "not_applicable" or index_id == "not_applicable": |
| 136 | return "No asset found" |
| 137 | |
| 138 | # Look up asset field names for this source from the database |
| 139 | result = await session.execute( |
| 140 | select(AssetFieldName.field_name).where(AssetFieldName.source == source).distinct(), |
| 141 | ) |
| 142 | asset_field_name = result.scalars().first() |
| 143 | if not asset_field_name: |
| 144 | logger.warning(f"No asset field name configured for source '{source}'") |
| 145 | return "No asset found" |
| 146 | |
| 147 | possible_fields = [f.strip() for f in asset_field_name.split(",")] |
| 148 | logger.info(f"Asset field candidates for source '{source}': {possible_fields}") |
| 149 | |
| 150 | # Fetch the event document from OpenSearch |
| 151 | es_client = await create_wazuh_indexer_client_async("Wazuh-Indexer") |
| 152 | try: |
| 153 | doc = await es_client.get(index=index_name, id=index_id) |
| 154 | event_source = doc.get("_source", {}) |
| 155 | |
| 156 | for field in possible_fields: |
| 157 | if field in event_source and event_source[field]: |
| 158 | logger.info(f"Resolved threshold asset name '{event_source[field]}' from field '{field}'") |
| 159 | return event_source[field] |
| 160 | |
| 161 | logger.warning(f"No asset name found in event for fields: {possible_fields}") |
| 162 | return "No asset found" |
| 163 | except Exception as e: |
| 164 | logger.error(f"Error fetching event for asset resolution: {e}") |
| 165 | return "No asset found" |
| 166 | finally: |
| 167 | await es_client.close() |
| 168 | |
| 169 | |
| 170 | async def save_threshold_metadata( |
| 171 | alert_id: int, |
| 172 | event_definition_id: str, |
| 173 | replay_query: str, |
| 174 | timerange_start: datetime, |
| 175 | timerange_end: datetime, |
| 176 | group_by_fields: Dict[str, str], |
| 177 | source_streams: List[str], |
| 178 | source: str, |
| 179 | resolved_index_name: str, |
| 180 | resolved_index_id: str, |
| 181 | session: AsyncSession, |
| 182 | ) -> ThresholdAlertMetadata: |
| 183 | """ |
| 184 | Persist threshold alert metadata to the database for later timeline retrieval. |
| 185 | |
| 186 | Args: |
| 187 | alert_id: The CoPilot alert ID returned by create_alert_full. |
| 188 | event_definition_id: The Graylog event definition ID. |
| 189 | replay_query: Lucene query from replay_info.query. |
| 190 | timerange_start: Start of the threshold evaluation window. |
| 191 | timerange_end: End of the threshold evaluation window. |
| 192 | group_by_fields: Group-by field key/value pairs. |
| 193 | source_streams: Graylog source stream IDs. |
| 194 | source: The SOURCE field value (e.g. "wazuh"). |
| 195 | resolved_index_name: OpenSearch index name of the resolved event. |
| 196 | resolved_index_id: OpenSearch document ID of the resolved event. |
| 197 | session: Database session. |
| 198 | |
| 199 | Returns: |
| 200 | The created ThresholdAlertMetadata record. |
| 201 | """ |
| 202 | metadata = ThresholdAlertMetadata( |
| 203 | alert_id=alert_id, |
| 204 | event_definition_id=event_definition_id, |
| 205 | replay_query=replay_query, |
| 206 | timerange_start=timerange_start, |
| 207 | timerange_end=timerange_end, |
| 208 | group_by_fields=group_by_fields, |
| 209 | source_streams=source_streams, |
| 210 | source=source, |
| 211 | resolved_index_name=resolved_index_name, |
| 212 | resolved_index_id=resolved_index_id, |
| 213 | ) |
| 214 | session.add(metadata) |
| 215 | await session.commit() |
| 216 | logger.info(f"Saved threshold alert metadata for alert ID {alert_id}") |
| 217 | return metadata |
| 218 | |
| 219 | |
| 220 | async def retrieve_threshold_alert_timeline( |
| 221 | alert_id: Optional[int], |
| 222 | index_name: str, |
| 223 | index_id: str, |
| 224 | session: AsyncSession, |
| 225 | ) -> Optional[List[Dict[str, Any]]]: |
| 226 | """ |
| 227 | Retrieve the timeline for a threshold alert using stored metadata. |
| 228 | |
| 229 | Looks up ThresholdAlertMetadata by alert_id first, then falls back to |
| 230 | resolved_index_name + resolved_index_id (which the frontend always provides). |
| 231 | |
| 232 | Args: |
| 233 | alert_id: The CoPilot alert ID (may be None if not provided by the caller). |
| 234 | index_name: The OpenSearch index name from the asset. |
| 235 | index_id: The OpenSearch document ID from the asset. |
| 236 | session: Database session. |
| 237 | |
| 238 | Returns: |
| 239 | List of OpenSearch hit dicts if this is a threshold alert, or None if no |
| 240 | threshold metadata exists (meaning it's not a threshold alert). |
| 241 | """ |
| 242 | metadata = None |
| 243 | |
| 244 | if alert_id is not None: |
| 245 | result = await session.execute( |
| 246 | select(ThresholdAlertMetadata).where(ThresholdAlertMetadata.alert_id == alert_id), |
| 247 | ) |
| 248 | metadata = result.scalars().first() |
| 249 | |
| 250 | if metadata is None: |
| 251 | result = await session.execute( |
| 252 | select(ThresholdAlertMetadata).where( |
| 253 | ThresholdAlertMetadata.resolved_index_name == index_name, |
| 254 | ThresholdAlertMetadata.resolved_index_id == index_id, |
| 255 | ), |
| 256 | ) |
| 257 | metadata = result.scalars().first() |
| 258 | |
| 259 | if metadata is None: |
| 260 | return None |
| 261 | |
| 262 | try: |
| 263 | index_pattern, time_field = get_index_config_for_source(metadata.source) |
| 264 | except ValueError: |
| 265 | logger.warning(f"Cannot retrieve threshold timeline: source '{metadata.source}' is not mapped") |
| 266 | return [] |
| 267 | |
| 268 | es_client = await create_wazuh_indexer_client_async("Wazuh-Indexer") |
| 269 | try: |
| 270 | must_clauses: List[Dict[str, Any]] = [ |
| 271 | {"query_string": {"query": metadata.replay_query, "default_operator": "AND"}}, |
| 272 | { |
| 273 | "range": { |
| 274 | time_field: { |
| 275 | "gte": _format_datetime(metadata.timerange_start), |
| 276 | "lte": _format_datetime(metadata.timerange_end), |
| 277 | }, |
| 278 | }, |
| 279 | }, |
| 280 | ] |
| 281 | |
| 282 | group_by = metadata.group_by_fields or {} |
| 283 | for field_name, field_value in group_by.items(): |
| 284 | if field_value: |
| 285 | must_clauses.append({"match": {field_name: field_value}}) |
| 286 | |
| 287 | query = { |
| 288 | "query": {"bool": {"must": must_clauses}}, |
| 289 | "sort": [{time_field: {"order": "asc"}}], |
| 290 | } |
| 291 | |
| 292 | logger.info( |
| 293 | f"Fetching threshold alert timeline for alert ID {alert_id} " f"in index '{index_pattern}' with query: {metadata.replay_query}", |
| 294 | ) |
| 295 | |
| 296 | response = await es_client.search(index=index_pattern, body=query, size=50) |
| 297 | return response["hits"]["hits"] |
| 298 | |
| 299 | except Exception as e: |
| 300 | logger.error(f"Error fetching threshold alert timeline for alert ID {alert_id}: {e}") |
| 301 | return [] |
| 302 | finally: |
| 303 | await es_client.close() |