| 1 | """ |
| 2 | In-memory cache of per-rule firing counts from the Wazuh indexer. |
| 3 | |
| 4 | Why this exists: |
| 5 | The catalog wants to surface "how often has this rule fired in the last |
| 6 | 7d/30d?" for every Wazuh rule. The naive way (one ES query per rule, per |
| 7 | request, per user) would crush the indexer. Instead we do **one** terms |
| 8 | aggregation over the alerts indices, cache the result, and serve every |
| 9 | per-rule lookup from the dict. |
| 10 | |
| 11 | Cost: one ``_cat/indices`` call + one ``search`` request to the Wazuh |
| 12 | indexer every ``CACHE_TTL_MINUTES`` minutes (15 by default). The |
| 13 | aggregation is a date-filtered ``terms`` over rule.id with a |
| 14 | sub-aggregation for the 7d sub-window. Even across hundreds of millions of |
| 15 | docs, this returns in well under a second. |
| 16 | |
| 17 | Index auto-discovery (why we don't hardcode ``wazuh-alerts-*``): |
| 18 | In a default Wazuh install, all alerts live in ``wazuh-alerts-*``. In |
| 19 | real SOCFortress deployments they don't — different integrations write to |
| 20 | different indices (``office365-<customer>``, ``crowdstrike-<customer>``, |
| 21 | ``carbonblack-<customer>``, ``huntress_<customer>``, ad-hoc ``newest-*`` |
| 22 | test indices, …). Hardcoding ``wazuh-alerts-*`` misses all of them. |
| 23 | Maintaining a comma-separated env list is fragile — every new customer |
| 24 | integration would need a config bump. |
| 25 | |
| 26 | Instead we list every index in the cluster on each refresh, filter out |
| 27 | indices that are definitely *not* alert indices (Wazuh's internal |
| 28 | ``monitoring/statistics/states`` plus Kibana/OpenSearch system indices), |
| 29 | and aggregate across whatever's left. The aggregation's |
| 30 | ``exists`` filter on the rule-ID field naturally drops indices that don't |
| 31 | carry one, and ``_coerce_rule_id`` drops bucket keys that aren't |
| 32 | integer-parseable — so vendor indices with their own non-Wazuh rule IDs |
| 33 | (e.g. ``"Office365_FailedLogin"``) don't pollute the cache. |
| 34 | |
| 35 | Field-name fallback: |
| 36 | Wazuh's indexer schema varies across versions and shipping configurations. |
| 37 | The rule ID may live under ``rule.id`` (object mapping), ``rule_id`` (flat |
| 38 | mapping), or just ``id`` (older schemas). We try them in that order, taking |
| 39 | the first one that returns non-empty buckets — same fallback pattern used |
| 40 | in ``app/connectors/wazuh_manager/services/mitre.py``. |
| 41 | |
| 42 | No DB writes, no schema changes, no Alembic migrations, no env var — |
| 43 | pure read-only aggregation against the live indices plus an in-memory |
| 44 | dict on the CoPilot side. |
| 45 | """ |
| 46 | |
| 47 | import asyncio |
| 48 | import re |
| 49 | from datetime import datetime |
| 50 | from datetime import timedelta |
| 51 | from typing import Any |
| 52 | from typing import Dict |
| 53 | from typing import List |
| 54 | from typing import Optional |
| 55 | |
| 56 | from elasticsearch7.exceptions import RequestError |
| 57 | from loguru import logger |
| 58 | |
| 59 | # 15-min TTL strikes a balance: fresh enough that the dashboard reflects |
| 60 | # real-time tuning, cheap enough that the indexer barely notices. |
| 61 | CACHE_TTL_MINUTES = 15 |
| 62 | |
| 63 | # ES-native index pattern with exclusions. We pass this as a single short |
| 64 | # string (~150 bytes) to ``client.search(index=...)``, which lets ES resolve |
| 65 | # it server-side. The previous approach of enumerating indices via |
| 66 | # ``_cat/indices`` and passing the list client-side blew up at scale — |
| 67 | # 1582 indices in a single URL crosses the 4096-byte HTTP line limit |
| 68 | # (``too_long_http_line_exception``). |
| 69 | # |
| 70 | # Exclusion rationale: |
| 71 | # - ``.*`` / ``_*``: Kibana / OpenSearch / Security plugin internals. |
| 72 | # - ``wazuh-monitoring-*`` / ``wazuh-statistics-*`` / ``wazuh-states-*`` / |
| 73 | # ``wazuh-vulnerabilities-*``: Wazuh's own internal indices. Carry rule.id |
| 74 | # for system/control events with parent-template rule IDs (2 = firewall, |
| 75 | # 3 = ids, 4 = web-log) that never fire on real alerts but produce massive |
| 76 | # fake hit counts. ``wazuh-vulnerabilities-*`` is the agent vulnerability |
| 77 | # state index, also not alert data. |
| 78 | # - ``security-auditlog-*``: OpenSearch Security plugin's audit log. |
| 79 | # |
| 80 | # The integer-coerce filter in ``_coerce_rule_id`` is the second line of |
| 81 | # defense — even if a vendor index sneaks in, non-Wazuh rule IDs (string |
| 82 | # identifiers like ``"Office365_FailedLogin"``) get silently dropped. |
| 83 | ALERT_INDEX_PATTERN = ( |
| 84 | "*" ",-.*" ",-_*" ",-wazuh-monitoring-*" ",-wazuh-statistics-*" ",-wazuh-states-*" ",-wazuh-vulnerabilities-*" ",-security-auditlog-*" |
| 85 | ) |
| 86 | |
| 87 | # Equivalent regex list — used only by the introspection helper |
| 88 | # ``_discover_alert_indices`` to populate ``resolved_indices`` for |
| 89 | # logging/debugging. The search itself never uses these; ES handles the |
| 90 | # pattern above natively. |
| 91 | EXCLUDE_PATTERNS = [ |
| 92 | re.compile(r"^\."), |
| 93 | re.compile(r"^_"), |
| 94 | re.compile(r"^wazuh-monitoring"), |
| 95 | re.compile(r"^wazuh-statistics"), |
| 96 | re.compile(r"^wazuh-states"), |
| 97 | re.compile(r"^wazuh-vulnerabilities"), |
| 98 | re.compile(r"security-auditlog"), |
| 99 | ] |
| 100 | |
| 101 | # Try these field paths in order. First one that yields non-empty buckets |
| 102 | # wins and is cached for subsequent refreshes (avoids retrying the misses |
| 103 | # on every refresh). |
| 104 | RULE_ID_FIELD_OPTIONS = ["rule.id", "rule_id", "id"] |
| 105 | |
| 106 | |
| 107 | class WazuhFiringStatsCache: |
| 108 | """ |
| 109 | Per-rule firing-count cache, keyed by integer rule ID. |
| 110 | |
| 111 | Returns a small dict per rule: ``{"hits_30d": int, "hits_7d": int}``. |
| 112 | Rules with zero hits over the 30d window are deliberately omitted from |
| 113 | the dict — callers should treat "missing" as "0 hits" rather than |
| 114 | "data unavailable". The cache's ``is_available`` flag distinguishes |
| 115 | those two states explicitly. |
| 116 | """ |
| 117 | |
| 118 | def __init__(self) -> None: |
| 119 | # rule_id (int) -> {"hits_30d": int, "hits_7d": int} |
| 120 | self._stats: Dict[int, Dict[str, int]] = {} |
| 121 | self._last_refresh: Optional[datetime] = None |
| 122 | self._unavailable_reason: Optional[str] = None |
| 123 | # Field path that worked last time; tried first on next refresh. |
| 124 | self._winning_field: Optional[str] = None |
| 125 | # Indices discovered on last refresh — kept for introspection so |
| 126 | # operators can see "which indices did we actually aggregate over?" |
| 127 | # without having to spelunk through logs. |
| 128 | self._resolved_indices: List[str] = [] |
| 129 | self._lock = asyncio.Lock() |
| 130 | |
| 131 | # ---- introspection ---------------------------------------------------- |
| 132 | |
| 133 | @property |
| 134 | def is_stale(self) -> bool: |
| 135 | if self._last_refresh is None: |
| 136 | return True |
| 137 | return (datetime.utcnow() - self._last_refresh) > timedelta(minutes=CACHE_TTL_MINUTES) |
| 138 | |
| 139 | @property |
| 140 | def last_refresh(self) -> Optional[datetime]: |
| 141 | return self._last_refresh |
| 142 | |
| 143 | @property |
| 144 | def is_available(self) -> bool: |
| 145 | return self._unavailable_reason is None |
| 146 | |
| 147 | @property |
| 148 | def unavailable_reason(self) -> Optional[str]: |
| 149 | return self._unavailable_reason |
| 150 | |
| 151 | @property |
| 152 | def resolved_indices(self) -> List[str]: |
| 153 | """Indices that the last refresh actually aggregated across.""" |
| 154 | return list(self._resolved_indices) |
| 155 | |
| 156 | # ---- loading ---------------------------------------------------------- |
| 157 | |
| 158 | async def ensure_loaded(self) -> None: |
| 159 | """Lazy-load on first access; refresh on TTL expiry. Never raises.""" |
| 160 | if self.is_stale: |
| 161 | await self.refresh() |
| 162 | |
| 163 | async def refresh(self) -> int: |
| 164 | """ |
| 165 | Fire one terms-aggregation against the alerts index and rebuild the |
| 166 | per-rule stats dict. Returns the number of rules with at least one |
| 167 | hit in the 30d window. |
| 168 | |
| 169 | Never raises — failures are captured in ``unavailable_reason`` and |
| 170 | the existing cached snapshot is left in place. A transient indexer |
| 171 | outage shouldn't blank the firing-count column. |
| 172 | """ |
| 173 | # Local import to avoid pulling the indexer client at module-import |
| 174 | # time on deployments where the indexer connector isn't usable. |
| 175 | from app.connectors.wazuh_indexer.utils.universal import ( |
| 176 | create_wazuh_indexer_client_async, |
| 177 | ) |
| 178 | |
| 179 | async with self._lock: |
| 180 | logger.info("Refreshing Wazuh firing-stats cache from the indexer…") |
| 181 | |
| 182 | try: |
| 183 | client = await create_wazuh_indexer_client_async() |
| 184 | except Exception as exc: # noqa: BLE001 |
| 185 | self._unavailable_reason = _short_reason(exc) |
| 186 | logger.warning(f"Wazuh indexer client unavailable: {self._unavailable_reason}") |
| 187 | self._last_refresh = datetime.utcnow() |
| 188 | return len(self._stats) |
| 189 | |
| 190 | # Always search via the single ES wildcard+exclusion pattern — |
| 191 | # ES resolves it server-side, sub-millisecond, no HTTP line |
| 192 | # limits to worry about. The cat-indices discovery call below is |
| 193 | # only for the introspection / debug log so operators can see |
| 194 | # WHICH concrete indices the pattern expanded to without having |
| 195 | # to run the cat call themselves. Discovery failure is non-fatal |
| 196 | # — the search still runs on the pattern. |
| 197 | self._resolved_indices = await _discover_alert_indices(client) |
| 198 | logger.debug( |
| 199 | f"Firing-stats aggregating across {len(self._resolved_indices)} concrete index target(s); " |
| 200 | f"first few: {self._resolved_indices[:5]}", |
| 201 | ) |
| 202 | search_target = ALERT_INDEX_PATTERN |
| 203 | |
| 204 | # Try the winning field first (if we have one from a prior load), |
| 205 | # then fall back through the canonical list. First field that |
| 206 | # returns non-empty buckets wins. |
| 207 | fields_to_try: List[str] = [] |
| 208 | if self._winning_field: |
| 209 | fields_to_try.append(self._winning_field) |
| 210 | for f in RULE_ID_FIELD_OPTIONS: |
| 211 | if f not in fields_to_try: |
| 212 | fields_to_try.append(f) |
| 213 | |
| 214 | new_stats: Dict[int, Dict[str, int]] = {} |
| 215 | winner: Optional[str] = None |
| 216 | last_error: Optional[str] = None |
| 217 | |
| 218 | for field in fields_to_try: |
| 219 | try: |
| 220 | body = _build_firing_stats_query(field) |
| 221 | # ``ignore_unavailable`` swallows per-index permission / |
| 222 | # missing-index errors so one bad index doesn't fail the |
| 223 | # whole aggregation. ``allow_no_indices`` likewise stops |
| 224 | # the request from 4xx-ing when the pattern matches |
| 225 | # zero indices (fresh deployment, etc.). |
| 226 | response = await client.search( |
| 227 | index=search_target, |
| 228 | body=body, |
| 229 | ignore_unavailable=True, |
| 230 | allow_no_indices=True, |
| 231 | ) |
| 232 | buckets = response.get("aggregations", {}).get("by_rule", {}).get("buckets", []) |
| 233 | if not buckets: |
| 234 | # Empty buckets = field exists but no docs, OR field |
| 235 | # doesn't exist. Either way, try the next one. |
| 236 | continue |
| 237 | |
| 238 | for bucket in buckets: |
| 239 | rule_id = _coerce_rule_id(bucket.get("key")) |
| 240 | if rule_id is None: |
| 241 | continue |
| 242 | hits_30d = int(bucket.get("doc_count") or 0) |
| 243 | hits_7d = int( |
| 244 | bucket.get("last_7d", {}).get("doc_count") or 0, |
| 245 | ) |
| 246 | # ``max(timestamp)`` bucket comes back as |
| 247 | # ``{"value": 1700000000000.0, "value_as_string": "..."}`` |
| 248 | # (epoch millis + ISO string). Prefer the ISO string |
| 249 | # for direct UI rendering; fall back to None if the |
| 250 | # bucket is missing (shouldn't happen but defensive). |
| 251 | last_seen_bucket = bucket.get("last_seen") or {} |
| 252 | last_seen = last_seen_bucket.get("value_as_string") |
| 253 | new_stats[rule_id] = { |
| 254 | "hits_30d": hits_30d, |
| 255 | "hits_7d": hits_7d, |
| 256 | "last_seen": last_seen, |
| 257 | } |
| 258 | winner = field |
| 259 | break |
| 260 | except RequestError as re: |
| 261 | # Most common: field doesn't exist on the index — Wazuh |
| 262 | # versions differ. The .info payload has the human |
| 263 | # message (see CLAUDE.md "Things that bite"). |
| 264 | last_error = f"{field}: {str(re.info)[:120]}" |
| 265 | logger.debug(f"Firing-stats field '{field}' rejected: {last_error}") |
| 266 | continue |
| 267 | except Exception as exc: # noqa: BLE001 |
| 268 | last_error = f"{field}: {str(exc)[:120]}" |
| 269 | logger.debug(f"Firing-stats field '{field}' errored: {last_error}") |
| 270 | continue |
| 271 | finally: |
| 272 | pass |
| 273 | |
| 274 | try: |
| 275 | await client.close() |
| 276 | except Exception: |
| 277 | pass |
| 278 | |
| 279 | if winner is None: |
| 280 | # No field worked, or all returned empty. Treat as |
| 281 | # "unavailable" so the UI doesn't render a useless "0 hits" |
| 282 | # column for every rule when really we just can't tell. |
| 283 | self._unavailable_reason = last_error or "No rule-ID field returned hits in the alerts index." |
| 284 | self._last_refresh = datetime.utcnow() |
| 285 | logger.warning( |
| 286 | f"Could not resolve a rule-ID field for firing stats. {self._unavailable_reason}", |
| 287 | ) |
| 288 | return len(self._stats) |
| 289 | |
| 290 | self._stats = new_stats |
| 291 | self._winning_field = winner |
| 292 | self._unavailable_reason = None |
| 293 | self._last_refresh = datetime.utcnow() |
| 294 | logger.info( |
| 295 | f"Loaded firing stats for {len(self._stats)} rule(s) using field '{winner}'", |
| 296 | ) |
| 297 | return len(self._stats) |
| 298 | |
| 299 | # ---- accessors -------------------------------------------------------- |
| 300 | |
| 301 | def get(self, rule_id: int) -> Dict[str, Any]: |
| 302 | """ |
| 303 | Return ``{"hits_30d": int, "hits_7d": int, "last_seen": str | None}`` |
| 304 | for ``rule_id``. |
| 305 | |
| 306 | Missing rule IDs return zero counts + None last_seen — the rule |
| 307 | exists in the cache but simply hasn't fired in the 30d window. |
| 308 | Callers that need to distinguish "0 hits" from "stats unavailable" |
| 309 | should check ``is_available`` separately. |
| 310 | """ |
| 311 | return self._stats.get( |
| 312 | rule_id, |
| 313 | {"hits_30d": 0, "hits_7d": 0, "last_seen": None}, |
| 314 | ) |
| 315 | |
| 316 | |
| 317 | # --------------------------------------------------------------------------- |
| 318 | # Per-customer firing stats — on-demand, not cached |
| 319 | # --------------------------------------------------------------------------- |
| 320 | |
| 321 | |
| 322 | async def fetch_firing_stats_for_customer(customer_code: str) -> Dict[int, Dict[str, Any]]: |
| 323 | """ |
| 324 | Run the firing-stats aggregation filtered to a single customer code. |
| 325 | |
| 326 | Why this isn't cached: |
| 327 | The catalog has potentially dozens of customer codes; pre-caching all of |
| 328 | them would balloon memory and waste indexer cycles on customers nobody |
| 329 | is looking at. Instead we run the query on demand when an analyst picks |
| 330 | a customer from the dropdown. The query is the same shape as the global |
| 331 | refresh (one terms agg + sub-aggs), just with an extra term filter on |
| 332 | ``agent_labels_customer`` — sub-second on any realistic deployment. |
| 333 | |
| 334 | Returns ``{rule_id: {hits_30d, hits_7d, last_seen}}`` — same shape as |
| 335 | ``WazuhRulesFiringStatsCache.get()`` so callers can use it |
| 336 | interchangeably. Empty dict when: |
| 337 | - The customer code doesn't match any alerts in the 30d window. |
| 338 | - The winning_field couldn't be determined (indexer error / no field). |
| 339 | - The indexer is unreachable. |
| 340 | |
| 341 | All failures are silent / non-raising — the catalog falls back to "0 |
| 342 | hits for this customer" rather than erroring. |
| 343 | |
| 344 | Field naming note: the customer code is conventionally indexed under |
| 345 | ``agent_labels_customer`` (Graylog convention, flat) in SOCFortress |
| 346 | deployments — confirmed in the Suricata alert sample we tested. Wazuh |
| 347 | vanilla uses ``agent.labels.customer`` (nested). We try both. |
| 348 | """ |
| 349 | from app.connectors.wazuh_indexer.utils.universal import ( |
| 350 | create_wazuh_indexer_client_async, |
| 351 | ) |
| 352 | |
| 353 | if not customer_code or not customer_code.strip(): |
| 354 | return {} |
| 355 | |
| 356 | # Use whatever rule-ID field the global cache discovered — saves us from |
| 357 | # re-running the field-fallback dance per customer query. |
| 358 | field = wazuh_firing_stats_cache._winning_field or RULE_ID_FIELD_OPTIONS[0] |
| 359 | |
| 360 | try: |
| 361 | client = await create_wazuh_indexer_client_async() |
| 362 | except Exception as exc: # noqa: BLE001 |
| 363 | logger.warning(f"Per-customer firing stats: indexer unavailable ({_short_reason(exc)})") |
| 364 | return {} |
| 365 | |
| 366 | try: |
| 367 | # Try each customer-code field shape until one returns buckets. |
| 368 | # Same fallback philosophy as the rule-ID field discovery. |
| 369 | for customer_field in ("agent_labels_customer", "agent.labels.customer"): |
| 370 | try: |
| 371 | body = _build_firing_stats_query(field) |
| 372 | # Splice the customer filter into the existing query — keeps |
| 373 | # the agg shape identical so the bucket-unpacking code below |
| 374 | # matches the cached path exactly. |
| 375 | body["query"]["bool"]["filter"].append( |
| 376 | {"term": {customer_field: customer_code}}, |
| 377 | ) |
| 378 | response = await client.search( |
| 379 | index=ALERT_INDEX_PATTERN, |
| 380 | body=body, |
| 381 | ignore_unavailable=True, |
| 382 | allow_no_indices=True, |
| 383 | ) |
| 384 | buckets = response.get("aggregations", {}).get("by_rule", {}).get("buckets", []) |
| 385 | if not buckets: |
| 386 | continue |
| 387 | |
| 388 | out: Dict[int, Dict[str, Any]] = {} |
| 389 | for bucket in buckets: |
| 390 | rule_id = _coerce_rule_id(bucket.get("key")) |
| 391 | if rule_id is None: |
| 392 | continue |
| 393 | out[rule_id] = { |
| 394 | "hits_30d": int(bucket.get("doc_count") or 0), |
| 395 | "hits_7d": int(bucket.get("last_7d", {}).get("doc_count") or 0), |
| 396 | "last_seen": (bucket.get("last_seen") or {}).get("value_as_string"), |
| 397 | } |
| 398 | logger.debug( |
| 399 | f"Per-customer firing stats: {len(out)} rules for customer " f"{customer_code!r} via field {customer_field!r}", |
| 400 | ) |
| 401 | return out |
| 402 | except Exception as exc: # noqa: BLE001 |
| 403 | logger.debug( |
| 404 | f"Per-customer firing stats: customer field {customer_field!r} failed: {exc}", |
| 405 | ) |
| 406 | continue |
| 407 | return {} |
| 408 | finally: |
| 409 | try: |
| 410 | await client.close() |
| 411 | except Exception: |
| 412 | pass |
| 413 | |
| 414 | |
| 415 | def _build_firing_stats_query(field: str) -> Dict[str, Any]: |
| 416 | """ |
| 417 | One query gets us 30d totals + 7d sub-totals in a single round-trip. |
| 418 | |
| 419 | The outer ``terms`` agg buckets the last 30 days by rule ID; the nested |
| 420 | ``last_7d`` filter inside each bucket counts how many of those hits |
| 421 | fell in the most recent week. Painless-free, fast, and the result is |
| 422 | just a list of buckets — no docs returned (``size: 0``). |
| 423 | """ |
| 424 | return { |
| 425 | "size": 0, |
| 426 | "query": { |
| 427 | "bool": { |
| 428 | "filter": [ |
| 429 | {"range": {"timestamp": {"gte": "now-30d"}}}, |
| 430 | {"exists": {"field": field}}, |
| 431 | ], |
| 432 | }, |
| 433 | }, |
| 434 | "aggs": { |
| 435 | "by_rule": { |
| 436 | # 50000 is the ES "terms" agg's recommended ceiling for a |
| 437 | # single query; Wazuh ships ~3-5k rules so we'll never come |
| 438 | # close. We just don't want a silent truncation. |
| 439 | "terms": {"field": field, "size": 50000, "order": {"_count": "desc"}}, |
| 440 | "aggs": { |
| 441 | "last_7d": { |
| 442 | "filter": {"range": {"timestamp": {"gte": "now-7d"}}}, |
| 443 | }, |
| 444 | # max(timestamp) gives us "when did this rule last fire?" |
| 445 | # for free — same query, same scan, ES returns it as both |
| 446 | # epoch-millis and an ISO string. We thread the ISO form |
| 447 | # through to the UI so it can render relative time |
| 448 | # ("2 minutes ago", "26 days ago") without parsing. |
| 449 | "last_seen": { |
| 450 | "max": {"field": "timestamp"}, |
| 451 | }, |
| 452 | }, |
| 453 | }, |
| 454 | }, |
| 455 | } |
| 456 | |
| 457 | |
| 458 | async def _discover_alert_indices(client: Any) -> List[str]: |
| 459 | """ |
| 460 | List every index in the cluster and return the ones that look like |
| 461 | alert indices (i.e. not a system / Wazuh-internal / audit-log index). |
| 462 | |
| 463 | Returns a list of concrete index names — pass directly to |
| 464 | ``client.search(index=<list>)``. Returns ``[]`` on any failure or if |
| 465 | the cluster is empty; callers should fall back to a sensible default |
| 466 | pattern in that case. |
| 467 | |
| 468 | Why this is safe even with vendor indices that aren't Wazuh-alert |
| 469 | shaped: |
| 470 | |
| 471 | - The aggregation's ``exists`` filter on the rule-ID field drops |
| 472 | documents that don't carry one. |
| 473 | - ``_coerce_rule_id`` drops bucket keys that aren't integer-parseable — |
| 474 | vendor-native rule identifiers like ``"Office365_FailedLogin"`` |
| 475 | don't pollute the Wazuh rule cache. |
| 476 | |
| 477 | So the worst case of including a "wrong" index is zero contribution, |
| 478 | not garbage data. Erring on the side of inclusion is correct. |
| 479 | """ |
| 480 | try: |
| 481 | # ``cat.indices`` returns a list of dicts with at least an ``index`` |
| 482 | # field when ``format="json"``. We only need the name; ``h="index"`` |
| 483 | # restricts the columns to keep the response small. |
| 484 | cat_response = await client.cat.indices(format="json", h="index") |
| 485 | except Exception as exc: # noqa: BLE001 |
| 486 | # Permission errors, network blips, etc. Caller falls back. |
| 487 | logger.warning(f"Firing-stats index discovery failed: {exc}") |
| 488 | return [] |
| 489 | |
| 490 | if not isinstance(cat_response, list): |
| 491 | # Some clients return the raw HTTP response object instead of the |
| 492 | # parsed body — defend against shape drift. |
| 493 | logger.warning( |
| 494 | f"Unexpected cat.indices response shape: {type(cat_response).__name__}", |
| 495 | ) |
| 496 | return [] |
| 497 | |
| 498 | kept: List[str] = [] |
| 499 | for item in cat_response: |
| 500 | name = item.get("index") if isinstance(item, dict) else None |
| 501 | if not isinstance(name, str) or not name: |
| 502 | continue |
| 503 | if any(pat.search(name) for pat in EXCLUDE_PATTERNS): |
| 504 | continue |
| 505 | kept.append(name) |
| 506 | |
| 507 | return kept |
| 508 | |
| 509 | |
| 510 | def _coerce_rule_id(key: Any) -> Optional[int]: |
| 511 | """ |
| 512 | Bucket keys come back as strings or ints depending on the field mapping |
| 513 | (keyword vs long). Normalize to int; drop anything that doesn't parse. |
| 514 | """ |
| 515 | if key is None: |
| 516 | return None |
| 517 | if isinstance(key, int): |
| 518 | return key |
| 519 | if isinstance(key, str): |
| 520 | try: |
| 521 | return int(key) |
| 522 | except ValueError: |
| 523 | return None |
| 524 | return None |
| 525 | |
| 526 | |
| 527 | def _short_reason(exc: BaseException) -> str: |
| 528 | """Mirror of the helper in wazuh_rules_cache — keep it local to avoid coupling.""" |
| 529 | raw = str(exc).strip() or exc.__class__.__name__ |
| 530 | lower = raw.lower() |
| 531 | if "connection" in lower or "timed out" in lower or "timeout" in lower: |
| 532 | return "Wazuh indexer is not reachable." |
| 533 | if "401" in raw or "403" in raw or "unauthor" in lower: |
| 534 | return "Wazuh indexer rejected the credentials." |
| 535 | if "not configured" in lower or "not found" in lower: |
| 536 | return "Wazuh indexer connector is not configured." |
| 537 | return raw[:160] + ("…" if len(raw) > 160 else "") |
| 538 | |
| 539 | |
| 540 | # Module-level singleton — same pattern as the rules caches. |
| 541 | wazuh_firing_stats_cache = WazuhFiringStatsCache() |