| 1 | import json |
| 2 | from pathlib import Path |
| 3 | from typing import Any |
| 4 | from typing import Dict |
| 5 | from typing import List |
| 6 | |
| 7 | from elasticsearch7.exceptions import RequestError |
| 8 | from fastapi import HTTPException |
| 9 | from loguru import logger |
| 10 | from sqlalchemy import select |
| 11 | from sqlalchemy.ext.asyncio import AsyncSession |
| 12 | |
| 13 | from app.connectors.wazuh_indexer.utils.universal import AlertsQueryBuilder |
| 14 | from app.connectors.wazuh_indexer.utils.universal import ( |
| 15 | create_wazuh_indexer_client_async, |
| 16 | ) |
| 17 | from app.db.universal_models import EnabledDashboards |
| 18 | from app.db.universal_models import EventSources |
| 19 | from app.siem.schema.dashboards import DashboardCategory |
| 20 | from app.siem.schema.dashboards import DashboardCategoryWithTemplates |
| 21 | from app.siem.schema.dashboards import DashboardTemplate |
| 22 | from app.siem.schema.dashboards import EnableDashboardRequest |
| 23 | from app.siem.schema.dashboards import PanelResult |
| 24 | |
| 25 | TEMPLATES_DIR = Path(__file__).resolve().parent.parent / "dashboard_templates" |
| 26 | |
| 27 | |
| 28 | # ── Template browsing (filesystem) ────────────────────────────── |
| 29 | |
| 30 | |
| 31 | def list_categories() -> List[DashboardCategory]: |
| 32 | """Return every category that has a valid _card.json.""" |
| 33 | categories: List[DashboardCategory] = [] |
| 34 | if not TEMPLATES_DIR.is_dir(): |
| 35 | return categories |
| 36 | for child in sorted(TEMPLATES_DIR.iterdir()): |
| 37 | card_path = child / "_card.json" |
| 38 | if child.is_dir() and card_path.is_file(): |
| 39 | data = json.loads(card_path.read_text()) |
| 40 | categories.append(DashboardCategory(**data)) |
| 41 | return categories |
| 42 | |
| 43 | |
| 44 | def get_category_detail(category_id: str) -> DashboardCategoryWithTemplates: |
| 45 | """Load one category card + all its dashboard templates.""" |
| 46 | category_dir = TEMPLATES_DIR / category_id |
| 47 | card_path = category_dir / "_card.json" |
| 48 | if not category_dir.is_dir() or not card_path.is_file(): |
| 49 | raise HTTPException(status_code=404, detail=f"Dashboard category '{category_id}' not found") |
| 50 | |
| 51 | card = json.loads(card_path.read_text()) |
| 52 | templates: List[DashboardTemplate] = [] |
| 53 | for tpl_file in sorted(category_dir.glob("*.json")): |
| 54 | if tpl_file.name.startswith("_"): |
| 55 | continue |
| 56 | tpl_data = json.loads(tpl_file.read_text()) |
| 57 | templates.append(DashboardTemplate(**tpl_data)) |
| 58 | |
| 59 | return DashboardCategoryWithTemplates(**card, templates=templates) |
| 60 | |
| 61 | |
| 62 | # ── Enabled dashboards (database) ─────────────────────────────── |
| 63 | |
| 64 | |
| 65 | async def get_enabled_dashboards( |
| 66 | customer_code: str, |
| 67 | db: AsyncSession, |
| 68 | ) -> List[EnabledDashboards]: |
| 69 | logger.info(f"Fetching enabled dashboards for customer {customer_code}") |
| 70 | result = await db.execute( |
| 71 | select(EnabledDashboards).where(EnabledDashboards.customer_code == customer_code), |
| 72 | ) |
| 73 | return result.scalars().all() |
| 74 | |
| 75 | |
| 76 | async def get_enabled_dashboards_for_customers( |
| 77 | customer_codes: List[str], |
| 78 | db: AsyncSession, |
| 79 | ) -> List[EnabledDashboards]: |
| 80 | """Fetch enabled dashboards across a set of customers. |
| 81 | |
| 82 | ``customer_codes`` is the caller's already-resolved effective set: ``["*"]`` |
| 83 | means all customers (admin/analyst), an empty list means none. |
| 84 | """ |
| 85 | logger.info(f"Fetching enabled dashboards for customers {customer_codes}") |
| 86 | query = select(EnabledDashboards) |
| 87 | if "*" not in customer_codes: |
| 88 | query = query.where(EnabledDashboards.customer_code.in_(customer_codes)) |
| 89 | result = await db.execute(query) |
| 90 | return result.scalars().all() |
| 91 | |
| 92 | |
| 93 | async def enable_dashboard( |
| 94 | request: EnableDashboardRequest, |
| 95 | db: AsyncSession, |
| 96 | ) -> EnabledDashboards: |
| 97 | logger.info( |
| 98 | f"Enabling dashboard {request.library_card}/{request.template_id} " f"for customer {request.customer_code}", |
| 99 | ) |
| 100 | |
| 101 | # Verify the category and template exist on disk |
| 102 | category_dir = TEMPLATES_DIR / request.library_card |
| 103 | card_path = category_dir / "_card.json" |
| 104 | template_path = category_dir / f"{request.template_id}.json" |
| 105 | if not card_path.is_file(): |
| 106 | raise HTTPException(status_code=404, detail=f"Dashboard category '{request.library_card}' not found") |
| 107 | if not template_path.is_file(): |
| 108 | raise HTTPException( |
| 109 | status_code=404, |
| 110 | detail=f"Dashboard template '{request.template_id}' not found in category '{request.library_card}'", |
| 111 | ) |
| 112 | |
| 113 | # Verify event source exists |
| 114 | es_result = await db.execute( |
| 115 | select(EventSources).where(EventSources.id == request.event_source_id), |
| 116 | ) |
| 117 | if not es_result.scalars().first(): |
| 118 | raise HTTPException(status_code=404, detail=f"Event source {request.event_source_id} not found") |
| 119 | |
| 120 | # Check for duplicate |
| 121 | result = await db.execute( |
| 122 | select(EnabledDashboards).where( |
| 123 | EnabledDashboards.customer_code == request.customer_code, |
| 124 | EnabledDashboards.event_source_id == request.event_source_id, |
| 125 | EnabledDashboards.library_card == request.library_card, |
| 126 | EnabledDashboards.template_id == request.template_id, |
| 127 | ), |
| 128 | ) |
| 129 | if result.scalars().first(): |
| 130 | raise HTTPException( |
| 131 | status_code=400, |
| 132 | detail="This dashboard is already enabled for this customer and event source", |
| 133 | ) |
| 134 | |
| 135 | row = EnabledDashboards( |
| 136 | customer_code=request.customer_code, |
| 137 | event_source_id=request.event_source_id, |
| 138 | library_card=request.library_card, |
| 139 | template_id=request.template_id, |
| 140 | display_name=request.display_name, |
| 141 | ) |
| 142 | db.add(row) |
| 143 | await db.flush() |
| 144 | await db.refresh(row) |
| 145 | await db.commit() |
| 146 | return row |
| 147 | |
| 148 | |
| 149 | async def disable_dashboard( |
| 150 | dashboard_id: int, |
| 151 | db: AsyncSession, |
| 152 | ) -> None: |
| 153 | result = await db.execute( |
| 154 | select(EnabledDashboards).where(EnabledDashboards.id == dashboard_id), |
| 155 | ) |
| 156 | row = result.scalars().first() |
| 157 | if not row: |
| 158 | raise HTTPException(status_code=404, detail="Enabled dashboard not found") |
| 159 | await db.delete(row) |
| 160 | await db.commit() |
| 161 | logger.info(f"Disabled dashboard {dashboard_id}") |
| 162 | |
| 163 | |
| 164 | # ── Panel data (execute queries for dashboard rendering) ───────── |
| 165 | |
| 166 | |
| 167 | def _is_text_field_agg_error(exc: RequestError) -> bool: |
| 168 | """True iff the Elasticsearch error is the one raised when a terms agg |
| 169 | targets a `text`-typed field (which lacks per-document field data by |
| 170 | default). The error string is stable across ES 7.x and is the cue we |
| 171 | use to retry the agg against `<field>.keyword`. |
| 172 | |
| 173 | Sample message body (in `exc.info`): |
| 174 | "Text fields are not optimised for operations that require |
| 175 | per-document field data like aggregations and sorting..." |
| 176 | |
| 177 | Note: `str(exc)` for elasticsearch7 RequestError only includes the |
| 178 | error code ("search_phase_execution_exception"), NOT the message — |
| 179 | the human-readable text lives on `exc.info`. We match against both |
| 180 | so the type code and the specific cause both have to line up. |
| 181 | """ |
| 182 | info = getattr(exc, "info", "") or "" |
| 183 | return getattr(exc, "error", "") == "search_phase_execution_exception" and "Text fields are not optimised" in str(info) |
| 184 | |
| 185 | |
| 186 | def _compute_histogram_interval(timerange: str) -> str: |
| 187 | """Pick a reasonable date_histogram interval for the given timerange.""" |
| 188 | mapping = { |
| 189 | "1h": "1m", |
| 190 | "6h": "10m", |
| 191 | "24h": "30m", |
| 192 | "3d": "3h", |
| 193 | "7d": "6h", |
| 194 | "14d": "12h", |
| 195 | "30d": "1d", |
| 196 | } |
| 197 | return mapping.get(timerange, "1h") |
| 198 | |
| 199 | |
| 200 | async def get_panel_data( |
| 201 | dashboard_id: int, |
| 202 | timerange: str, |
| 203 | db: AsyncSession, |
| 204 | ) -> Dict[str, Any]: |
| 205 | """Execute each panel's query and return aggregated data for ECharts.""" |
| 206 | |
| 207 | # Load the enabled dashboard row |
| 208 | result = await db.execute( |
| 209 | select(EnabledDashboards).where(EnabledDashboards.id == dashboard_id), |
| 210 | ) |
| 211 | dashboard = result.scalars().first() |
| 212 | if not dashboard: |
| 213 | raise HTTPException(status_code=404, detail="Enabled dashboard not found") |
| 214 | |
| 215 | # Load the event source to get index_pattern + time_field |
| 216 | es_result = await db.execute( |
| 217 | select(EventSources).where(EventSources.id == dashboard.event_source_id), |
| 218 | ) |
| 219 | event_source = es_result.scalars().first() |
| 220 | if not event_source: |
| 221 | raise HTTPException(status_code=404, detail="Event source not found") |
| 222 | if not event_source.enabled: |
| 223 | raise HTTPException(status_code=400, detail="Event source is disabled") |
| 224 | |
| 225 | # Load template JSON from disk |
| 226 | template_path = TEMPLATES_DIR / dashboard.library_card / f"{dashboard.template_id}.json" |
| 227 | if not template_path.is_file(): |
| 228 | raise HTTPException(status_code=404, detail="Dashboard template file not found") |
| 229 | |
| 230 | template = json.loads(template_path.read_text()) |
| 231 | panels = template.get("panels", []) |
| 232 | |
| 233 | # Load category card for accent color |
| 234 | card_path = TEMPLATES_DIR / dashboard.library_card / "_card.json" |
| 235 | accent_color = "#38bdf8" |
| 236 | if card_path.is_file(): |
| 237 | card = json.loads(card_path.read_text()) |
| 238 | accent_color = card.get("color", accent_color) |
| 239 | |
| 240 | # Build time range filter |
| 241 | query_builder = AlertsQueryBuilder() |
| 242 | query_builder.add_time_range(timerange=timerange, timestamp_field=event_source.time_field) |
| 243 | time_filter = query_builder.query["query"]["bool"]["must"] |
| 244 | |
| 245 | es_client = await create_wazuh_indexer_client_async("Wazuh-Indexer") |
| 246 | results: Dict[str, PanelResult] = {} |
| 247 | |
| 248 | try: |
| 249 | for panel in panels: |
| 250 | pid = panel["id"] |
| 251 | ptype = panel["type"] |
| 252 | lucene = panel.get("lucene", "*") |
| 253 | field = panel.get("field") |
| 254 | size = panel.get("size", 10) |
| 255 | |
| 256 | try: |
| 257 | # Build base query with time filter + Lucene |
| 258 | body: dict = { |
| 259 | "query": { |
| 260 | "bool": { |
| 261 | "must": [ |
| 262 | *time_filter, |
| 263 | {"query_string": {"query": lucene, "default_operator": "AND"}}, |
| 264 | ], |
| 265 | }, |
| 266 | }, |
| 267 | "size": 0, |
| 268 | } |
| 269 | |
| 270 | if ptype == "stat": |
| 271 | # Just need the total count |
| 272 | resp = await es_client.search(index=event_source.index_pattern, body=body) |
| 273 | total = resp["hits"]["total"] |
| 274 | count = total["value"] if isinstance(total, dict) else total |
| 275 | results[pid] = PanelResult(type="stat", value=count) |
| 276 | |
| 277 | elif ptype == "histogram": |
| 278 | interval = _compute_histogram_interval(timerange) |
| 279 | body["aggs"] = { |
| 280 | "over_time": { |
| 281 | "date_histogram": { |
| 282 | "field": event_source.time_field, |
| 283 | "fixed_interval": interval, |
| 284 | "min_doc_count": 0, |
| 285 | }, |
| 286 | }, |
| 287 | } |
| 288 | resp = await es_client.search(index=event_source.index_pattern, body=body) |
| 289 | buckets = resp["aggregations"]["over_time"]["buckets"] |
| 290 | labels = [b["key_as_string"] for b in buckets] |
| 291 | data = [b["doc_count"] for b in buckets] |
| 292 | results[pid] = PanelResult(type="histogram", labels=labels, data=data) |
| 293 | |
| 294 | elif ptype in ("pie", "bar_h"): |
| 295 | if not field: |
| 296 | results[pid] = PanelResult(type=ptype, error="No field specified for aggregation") |
| 297 | continue |
| 298 | |
| 299 | def _build_terms_agg(agg_field: str) -> dict: |
| 300 | return { |
| 301 | "top_values": { |
| 302 | "terms": {"field": agg_field, "size": size}, |
| 303 | }, |
| 304 | } |
| 305 | |
| 306 | body["aggs"] = _build_terms_agg(field) |
| 307 | try: |
| 308 | resp = await es_client.search(index=event_source.index_pattern, body=body) |
| 309 | except RequestError as exc: |
| 310 | # Elasticsearch refuses terms aggs on `text` fields by default. |
| 311 | # If the index maps `field` as text, retry with the conventional |
| 312 | # `field.keyword` subfield. Cheap one-shot fallback that handles |
| 313 | # the common case where customer index templates differ. |
| 314 | if _is_text_field_agg_error(exc) and not field.endswith(".keyword"): |
| 315 | logger.info( |
| 316 | f"Panel {pid}: '{field}' is text-typed in {event_source.index_pattern}; " |
| 317 | f"retrying with '{field}.keyword'", |
| 318 | ) |
| 319 | body["aggs"] = _build_terms_agg(f"{field}.keyword") |
| 320 | resp = await es_client.search(index=event_source.index_pattern, body=body) |
| 321 | else: |
| 322 | raise |
| 323 | buckets = resp["aggregations"]["top_values"]["buckets"] |
| 324 | labels = [str(b["key"]) for b in buckets] |
| 325 | data = [b["doc_count"] for b in buckets] |
| 326 | results[pid] = PanelResult(type=ptype, labels=labels, data=data) |
| 327 | |
| 328 | else: |
| 329 | results[pid] = PanelResult(type=ptype, error=f"Unknown panel type: {ptype}") |
| 330 | |
| 331 | except Exception as e: |
| 332 | logger.error(f"Error querying panel {pid}: {e}") |
| 333 | results[pid] = PanelResult(type=ptype, error=str(e)) |
| 334 | finally: |
| 335 | await es_client.close() |
| 336 | |
| 337 | return { |
| 338 | "results": results, |
| 339 | "template": template, |
| 340 | "accent_color": accent_color, |
| 341 | "customer_code": dashboard.customer_code, |
| 342 | "source_name": event_source.name, |
| 343 | } |