| 1 | """ |
| 2 | Service layer for the Detection Catalog — a discovery surface over the same |
| 3 | rules already loaded by the CoPilot Searches feature. |
| 4 | |
| 5 | Architecture notes: |
| 6 | - This is a **pure view layer**. It does NOT fetch anything from GitHub and |
| 7 | does NOT maintain its own cache. It walks the in-memory ``rules_cache`` |
| 8 | populated by ``app.integrations.copilot_searches.services.copilot_searches`` |
| 9 | and aggregates differently per dimension (analytic_story, product, |
| 10 | data_source, etc.). |
| 11 | - MITRE tactic names are resolved via the in-memory ``mitre_matrix`` populated |
| 12 | by ``app.integrations.copilot_searches.services.mitre_coverage``. Same |
| 13 | reasoning: no second fetch, single source of truth. |
| 14 | - Both backing caches load lazily on first access (``ensure_loaded``), and |
| 15 | refresh of the CoPilot Searches cache is the single user-facing refresh |
| 16 | path — there is no separate "refresh the catalog" action. |
| 17 | - All aggregation happens fresh on each request. The corpus is small enough |
| 18 | (~500 rules) that walking it is sub-millisecond; a separate cache layer |
| 19 | here would just create another invalidation surface. |
| 20 | """ |
| 21 | |
| 22 | from collections import defaultdict |
| 23 | from typing import Any |
| 24 | from typing import Dict |
| 25 | from typing import List |
| 26 | from typing import Optional |
| 27 | |
| 28 | from loguru import logger |
| 29 | |
| 30 | from app.integrations.copilot_searches.services.copilot_searches import rules_cache |
| 31 | from app.integrations.copilot_searches.services.mitre_coverage import mitre_matrix |
| 32 | from app.integrations.copilot_searches.services.wazuh_firing_stats_cache import ( |
| 33 | fetch_firing_stats_for_customer, |
| 34 | ) |
| 35 | from app.integrations.copilot_searches.services.wazuh_firing_stats_cache import ( |
| 36 | wazuh_firing_stats_cache, |
| 37 | ) |
| 38 | from app.integrations.copilot_searches.services.wazuh_rules_cache import ( |
| 39 | wazuh_rules_cache, |
| 40 | ) |
| 41 | |
| 42 | # --------------------------------------------------------------------------- |
| 43 | # Helpers |
| 44 | # --------------------------------------------------------------------------- |
| 45 | |
| 46 | |
| 47 | def _ensure_list_of_str(value: Any) -> List[str]: |
| 48 | """Return ``value`` as a list of strings, tolerating None / scalars / mixed lists.""" |
| 49 | if value is None: |
| 50 | return [] |
| 51 | if isinstance(value, str): |
| 52 | return [value] |
| 53 | if isinstance(value, list): |
| 54 | return [v for v in value if isinstance(v, str)] |
| 55 | return [] |
| 56 | |
| 57 | |
| 58 | def _resolve_tactics_for_mitre_ids(mitre_ids: List[str]) -> List[str]: |
| 59 | """ |
| 60 | Map a list of MITRE technique IDs (e.g. ``["T1059.007", "T1218"]``) to the |
| 61 | unique set of tactic display names (e.g. ``["Execution", "Defense Evasion"]``). |
| 62 | |
| 63 | Uses the in-memory MITRE matrix; returns an empty list if the matrix isn't |
| 64 | loaded yet (caller is expected to have invoked ``mitre_matrix.ensure_loaded``). |
| 65 | """ |
| 66 | techniques = mitre_matrix.techniques |
| 67 | tactics_by_short = {t["short_name"]: t["name"] for t in mitre_matrix.tactics} |
| 68 | |
| 69 | tactic_names: List[str] = [] |
| 70 | seen: set[str] = set() |
| 71 | for raw in mitre_ids: |
| 72 | if not isinstance(raw, str): |
| 73 | continue |
| 74 | # Match against the base technique (e.g. T1059.007 -> T1059); sub-techniques |
| 75 | # carry their parent's tactic mapping in the STIX bundle. |
| 76 | base = raw.split(".")[0].strip().upper() |
| 77 | meta = techniques.get(base) |
| 78 | if meta is None: |
| 79 | continue |
| 80 | for short_name in meta.get("tactic_short_names", []): |
| 81 | display = tactics_by_short.get(short_name) |
| 82 | if display and display not in seen: |
| 83 | seen.add(display) |
| 84 | tactic_names.append(display) |
| 85 | return tactic_names |
| 86 | |
| 87 | |
| 88 | def _rule_mitre_ids(rule: Dict[str, Any]) -> List[str]: |
| 89 | return _ensure_list_of_str(rule.get("tags", {}).get("mitre_attack_id")) |
| 90 | |
| 91 | |
| 92 | def _rule_stories(rule: Dict[str, Any]) -> List[str]: |
| 93 | return _ensure_list_of_str(rule.get("tags", {}).get("analytic_story")) |
| 94 | |
| 95 | |
| 96 | def _rule_products(rule: Dict[str, Any]) -> List[str]: |
| 97 | return _ensure_list_of_str(rule.get("tags", {}).get("product")) |
| 98 | |
| 99 | |
| 100 | def _rule_data_sources(rule: Dict[str, Any]) -> List[str]: |
| 101 | return _ensure_list_of_str(rule.get("data_source")) |
| 102 | |
| 103 | |
| 104 | def _rule_max_date(rule: Dict[str, Any]) -> Optional[str]: |
| 105 | date = rule.get("date") |
| 106 | return date if isinstance(date, str) else None |
| 107 | |
| 108 | |
| 109 | def _newest(a: Optional[str], b: Optional[str]) -> Optional[str]: |
| 110 | """Pick the later of two ISO date strings; None-tolerant.""" |
| 111 | if a is None: |
| 112 | return b |
| 113 | if b is None: |
| 114 | return a |
| 115 | return a if a >= b else b |
| 116 | |
| 117 | |
| 118 | # --------------------------------------------------------------------------- |
| 119 | # Stories index — table-of-stories landing |
| 120 | # --------------------------------------------------------------------------- |
| 121 | |
| 122 | |
| 123 | async def list_stories() -> List[Dict[str, Any]]: |
| 124 | """ |
| 125 | Aggregate all rules into the set of unique analytic stories, with the |
| 126 | per-story summary fields needed to render the index table: |
| 127 | |
| 128 | name | data_sources | tactics | products | date | detection_count |
| 129 | |
| 130 | Rules carrying multiple ``analytic_story`` tags contribute to every story |
| 131 | they're tagged with (intentional — a single detection genuinely belongs |
| 132 | to several stories in the published taxonomy). |
| 133 | |
| 134 | Rules with no ``analytic_story`` tag are silently excluded from the |
| 135 | Stories surface. (They still appear in the Rules grid; the catalog is |
| 136 | just a story-centric view.) |
| 137 | """ |
| 138 | await rules_cache.ensure_loaded() |
| 139 | await mitre_matrix.ensure_loaded() |
| 140 | |
| 141 | agg: Dict[str, Dict[str, Any]] = defaultdict( |
| 142 | lambda: { |
| 143 | "data_sources": set(), |
| 144 | "products": set(), |
| 145 | "tactic_names": [], # ordered, dedup'd |
| 146 | "detection_count": 0, |
| 147 | "date": None, |
| 148 | }, |
| 149 | ) |
| 150 | |
| 151 | for rule in rules_cache.get_all_rules(): |
| 152 | story_names = _rule_stories(rule) |
| 153 | if not story_names: |
| 154 | continue |
| 155 | |
| 156 | rule_tactics = _resolve_tactics_for_mitre_ids(_rule_mitre_ids(rule)) |
| 157 | rule_date = _rule_max_date(rule) |
| 158 | |
| 159 | for story_name in story_names: |
| 160 | row = agg[story_name] |
| 161 | row["data_sources"].update(_rule_data_sources(rule)) |
| 162 | row["products"].update(_rule_products(rule)) |
| 163 | for t in rule_tactics: |
| 164 | if t not in row["tactic_names"]: |
| 165 | row["tactic_names"].append(t) |
| 166 | row["detection_count"] += 1 |
| 167 | row["date"] = _newest(row["date"], rule_date) |
| 168 | |
| 169 | return [ |
| 170 | { |
| 171 | "name": name, |
| 172 | "data_sources": sorted(row["data_sources"]), |
| 173 | "tactics": row["tactic_names"], |
| 174 | "products": sorted(row["products"]), |
| 175 | "date": row["date"], |
| 176 | "detection_count": row["detection_count"], |
| 177 | } |
| 178 | for name, row in sorted(agg.items(), key=lambda kv: kv[0].lower()) |
| 179 | ] |
| 180 | |
| 181 | |
| 182 | # --------------------------------------------------------------------------- |
| 183 | # Single story detail |
| 184 | # --------------------------------------------------------------------------- |
| 185 | |
| 186 | |
| 187 | async def get_story_detail(story_name: str) -> Optional[Dict[str, Any]]: |
| 188 | """ |
| 189 | Build the detail payload for one analytic story: |
| 190 | |
| 191 | - header metadata (id slug, author, latest version, latest date) |
| 192 | - aggregated description (auto-generated from member detections) |
| 193 | - detections table (one row per rule in the story) |
| 194 | - data sources (deduplicated) |
| 195 | - references (deduplicated) |
| 196 | |
| 197 | Returns ``None`` if no rule in the cache carries this story tag. |
| 198 | """ |
| 199 | await rules_cache.ensure_loaded() |
| 200 | await mitre_matrix.ensure_loaded() |
| 201 | |
| 202 | target = story_name.strip() |
| 203 | members: List[Dict[str, Any]] = [rule for rule in rules_cache.get_all_rules() if target in _rule_stories(rule)] |
| 204 | if not members: |
| 205 | return None |
| 206 | |
| 207 | detections: List[Dict[str, Any]] = [] |
| 208 | data_sources: set[str] = set() |
| 209 | references: List[str] = [] |
| 210 | references_seen: set[str] = set() |
| 211 | products: set[str] = set() |
| 212 | authors: set[str] = set() |
| 213 | tactic_names: List[str] = [] |
| 214 | tactic_seen: set[str] = set() |
| 215 | type_counts: Dict[str, int] = defaultdict(int) |
| 216 | latest_date: Optional[str] = None |
| 217 | latest_version: Optional[int] = None |
| 218 | |
| 219 | for rule in members: |
| 220 | rule_tactics = _resolve_tactics_for_mitre_ids(_rule_mitre_ids(rule)) |
| 221 | for t in rule_tactics: |
| 222 | if t not in tactic_seen: |
| 223 | tactic_seen.add(t) |
| 224 | tactic_names.append(t) |
| 225 | |
| 226 | rule_type = rule.get("type") or "Unknown" |
| 227 | type_counts[rule_type] += 1 |
| 228 | |
| 229 | detections.append( |
| 230 | { |
| 231 | "id": rule.get("id"), |
| 232 | "name": rule.get("name", rule.get("id", "(unnamed)")), |
| 233 | "type": rule_type, |
| 234 | "severity": rule.get("response", {}).get("severity"), |
| 235 | "mitre_attack_id": _rule_mitre_ids(rule), |
| 236 | "tactics": rule_tactics, |
| 237 | "description": rule.get("description"), |
| 238 | }, |
| 239 | ) |
| 240 | |
| 241 | data_sources.update(_rule_data_sources(rule)) |
| 242 | products.update(_rule_products(rule)) |
| 243 | author = rule.get("author") |
| 244 | if isinstance(author, str) and author.strip(): |
| 245 | authors.add(author.strip()) |
| 246 | |
| 247 | for ref in _ensure_list_of_str(rule.get("references")): |
| 248 | if ref not in references_seen: |
| 249 | references_seen.add(ref) |
| 250 | references.append(ref) |
| 251 | |
| 252 | rule_date = _rule_max_date(rule) |
| 253 | latest_date = _newest(latest_date, rule_date) |
| 254 | |
| 255 | version = rule.get("version") |
| 256 | if isinstance(version, int): |
| 257 | if latest_version is None or version > latest_version: |
| 258 | latest_version = version |
| 259 | |
| 260 | detections.sort(key=lambda d: (d["name"] or "").lower()) |
| 261 | |
| 262 | # Auto-generated "Why it matters" narrative — used until/unless a |
| 263 | # curated story_metadata file is added in a sibling repo folder. |
| 264 | type_summary = ", ".join(f"{count} {kind}" for kind, count in sorted(type_counts.items())) |
| 265 | why_it_matters = ( |
| 266 | f"This story contains {len(detections)} detection(s) " |
| 267 | f"covering {len(tactic_names)} MITRE ATT&CK tactic(s)" |
| 268 | + (f" ({', '.join(tactic_names)})" if tactic_names else "") |
| 269 | + ". " |
| 270 | + (f"Detection types: {type_summary}. " if type_summary else "") |
| 271 | + "Story metadata is auto-generated from member detections; curated narratives can be added later via a story-metadata file in the rule repo." |
| 272 | ) |
| 273 | |
| 274 | description = ( |
| 275 | f"{len(detections)} detection(s) tagged with the analytic story " |
| 276 | f"'{target}'. Member detections span {len(tactic_names)} MITRE tactic(s)." |
| 277 | ) |
| 278 | |
| 279 | return { |
| 280 | "name": target, |
| 281 | "id": _story_slug(target), |
| 282 | "description": description, |
| 283 | "why_it_matters": why_it_matters, |
| 284 | "detections": detections, |
| 285 | "data_sources": sorted(data_sources), |
| 286 | "tactics": tactic_names, |
| 287 | "products": sorted(products), |
| 288 | "authors": sorted(authors), |
| 289 | "references": references, |
| 290 | "date": latest_date, |
| 291 | "version": latest_version, |
| 292 | "detection_count": len(detections), |
| 293 | } |
| 294 | |
| 295 | |
| 296 | def _story_slug(name: str) -> str: |
| 297 | """A URL/display-safe slug for a story name. Stable per name.""" |
| 298 | safe = "".join(ch if ch.isalnum() or ch in ("-", "_") else "_" for ch in name.strip().lower()) |
| 299 | while "__" in safe: |
| 300 | safe = safe.replace("__", "_") |
| 301 | return safe.strip("_") or "story" |
| 302 | |
| 303 | |
| 304 | # --------------------------------------------------------------------------- |
| 305 | # Catalog stats (for an Overview pane / facet counts) |
| 306 | # --------------------------------------------------------------------------- |
| 307 | |
| 308 | |
| 309 | async def get_catalog_stats() -> Dict[str, Any]: |
| 310 | """ |
| 311 | Lightweight aggregation for the Catalog overview header — how many |
| 312 | detections, stories, products, data sources we have indexed right now. |
| 313 | Same in-memory walk pattern as ``list_stories``; cheap. |
| 314 | |
| 315 | Includes Wazuh-side counts so the header reflects the union of both |
| 316 | corpora. Wazuh fields are returned as 0 / ``wazuh_available=false`` when |
| 317 | the Wazuh Manager is unreachable rather than failing the whole call — |
| 318 | the Stories tab must keep working even on Wazuh outages. |
| 319 | """ |
| 320 | await rules_cache.ensure_loaded() |
| 321 | await mitre_matrix.ensure_loaded() |
| 322 | await wazuh_rules_cache.ensure_loaded() |
| 323 | |
| 324 | stories: set[str] = set() |
| 325 | products: set[str] = set() |
| 326 | data_sources: set[str] = set() |
| 327 | tactic_names: set[str] = set() |
| 328 | detection_count = 0 |
| 329 | |
| 330 | for rule in rules_cache.get_all_rules(): |
| 331 | detection_count += 1 |
| 332 | stories.update(_rule_stories(rule)) |
| 333 | products.update(_rule_products(rule)) |
| 334 | data_sources.update(_rule_data_sources(rule)) |
| 335 | tactic_names.update(_resolve_tactics_for_mitre_ids(_rule_mitre_ids(rule))) |
| 336 | |
| 337 | return { |
| 338 | "detection_count": detection_count, |
| 339 | "story_count": len(stories), |
| 340 | "product_count": len(products), |
| 341 | "data_source_count": len(data_sources), |
| 342 | "tactic_count": len(tactic_names), |
| 343 | "last_refresh": rules_cache.last_refresh, |
| 344 | # Wazuh-side counts — read-only mirror of the wazuh_rules_cache state. |
| 345 | "wazuh_rule_count": wazuh_rules_cache.rules_count, |
| 346 | "wazuh_last_refresh": wazuh_rules_cache.last_refresh, |
| 347 | "wazuh_available": wazuh_rules_cache.is_available, |
| 348 | "wazuh_unavailable_reason": wazuh_rules_cache.unavailable_reason, |
| 349 | } |
| 350 | |
| 351 | |
| 352 | # --------------------------------------------------------------------------- |
| 353 | # Compliance pivot — group Wazuh rules by compliance framework control ID |
| 354 | # --------------------------------------------------------------------------- |
| 355 | |
| 356 | |
| 357 | # Frameworks we expose in the compliance pivot. Maps the API-friendly key |
| 358 | # (also the URL query value) to the human-readable label + the Wazuh |
| 359 | # rule-dict field name. Wazuh stores these as lists of control identifiers |
| 360 | # per rule. New frameworks: add a row here, no other code change needed. |
| 361 | COMPLIANCE_FRAMEWORKS: Dict[str, Dict[str, str]] = { |
| 362 | "pci_dss": {"label": "PCI DSS", "field": "pci_dss"}, |
| 363 | "gdpr": {"label": "GDPR", "field": "gdpr"}, |
| 364 | "hipaa": {"label": "HIPAA", "field": "hipaa"}, |
| 365 | "nist_800_53": {"label": "NIST 800-53", "field": "nist_800_53"}, |
| 366 | "tsc": {"label": "TSC", "field": "tsc"}, |
| 367 | "gpg13": {"label": "GPG13", "field": "gpg13"}, |
| 368 | } |
| 369 | |
| 370 | |
| 371 | def _rule_compliance_values(rule: Dict[str, Any], field: str) -> List[str]: |
| 372 | """Pull and normalize a compliance-array field off a cached Wazuh rule.""" |
| 373 | raw = rule.get(field) |
| 374 | if isinstance(raw, list): |
| 375 | return [v for v in raw if isinstance(v, str) and v.strip()] |
| 376 | if isinstance(raw, str): |
| 377 | return [v.strip() for v in raw.split(",") if v.strip()] |
| 378 | return [] |
| 379 | |
| 380 | |
| 381 | async def list_compliance_pivot(framework: str) -> Dict[str, Any]: |
| 382 | """ |
| 383 | Group every Wazuh rule by its values for the given compliance framework. |
| 384 | |
| 385 | Output shape: |
| 386 | { |
| 387 | "framework": "pci_dss", |
| 388 | "framework_label": "PCI DSS", |
| 389 | "groups": [ |
| 390 | { |
| 391 | "control": "10.2.4", |
| 392 | "rule_count": 23, |
| 393 | "total_hits_30d": 487, |
| 394 | "rule_ids": [5710, 5711, ...] |
| 395 | }, |
| 396 | ... |
| 397 | ], |
| 398 | ... |
| 399 | } |
| 400 | |
| 401 | Rules that don't carry ANY value for the framework field are excluded — |
| 402 | "no compliance tag" isn't a group, it's just absence. The frontend can |
| 403 | surface that count separately if useful, but for the pivot table itself |
| 404 | we only list real controls. |
| 405 | |
| 406 | Groups are sorted by ``total_hits_30d`` descending, then by ``control`` |
| 407 | alphabetically — the "what's actively firing for PCI 10.2.4?" question |
| 408 | is the most common analyst entry point. |
| 409 | """ |
| 410 | framework_meta = COMPLIANCE_FRAMEWORKS.get(framework) |
| 411 | if not framework_meta: |
| 412 | raise ValueError( |
| 413 | f"Unknown compliance framework {framework!r}. Valid: {list(COMPLIANCE_FRAMEWORKS)}", |
| 414 | ) |
| 415 | |
| 416 | await wazuh_rules_cache.ensure_loaded() |
| 417 | await wazuh_firing_stats_cache.ensure_loaded() |
| 418 | |
| 419 | field = framework_meta["field"] |
| 420 | |
| 421 | # control_id -> {"rule_ids": [], "total_hits_30d": int} |
| 422 | grouped: Dict[str, Dict[str, Any]] = defaultdict( |
| 423 | lambda: {"rule_ids": [], "total_hits_30d": 0, "total_hits_7d": 0}, |
| 424 | ) |
| 425 | rules_with_compliance = 0 |
| 426 | |
| 427 | for raw_rule in wazuh_rules_cache.get_all_rules(): |
| 428 | controls = _rule_compliance_values(raw_rule, field) |
| 429 | if not controls: |
| 430 | continue |
| 431 | rules_with_compliance += 1 |
| 432 | |
| 433 | rid = raw_rule.get("id") |
| 434 | stats = wazuh_firing_stats_cache.get(rid) if isinstance(rid, int) else {"hits_30d": 0, "hits_7d": 0} |
| 435 | |
| 436 | for control in controls: |
| 437 | bucket = grouped[control] |
| 438 | if isinstance(rid, int): |
| 439 | bucket["rule_ids"].append(rid) |
| 440 | bucket["total_hits_30d"] += stats["hits_30d"] |
| 441 | bucket["total_hits_7d"] += stats["hits_7d"] |
| 442 | |
| 443 | groups = [ |
| 444 | { |
| 445 | "control": control, |
| 446 | "rule_count": len(bucket["rule_ids"]), |
| 447 | "rule_ids": sorted(bucket["rule_ids"]), |
| 448 | "total_hits_30d": bucket["total_hits_30d"], |
| 449 | "total_hits_7d": bucket["total_hits_7d"], |
| 450 | } |
| 451 | for control, bucket in grouped.items() |
| 452 | ] |
| 453 | # Most-noisy-controls first; ties broken by control ID alphabetical so |
| 454 | # the order is stable across calls. |
| 455 | groups.sort(key=lambda g: (-g["total_hits_30d"], g["control"])) |
| 456 | |
| 457 | return { |
| 458 | "framework": framework, |
| 459 | "framework_label": framework_meta["label"], |
| 460 | "groups": groups, |
| 461 | "control_count": len(groups), |
| 462 | "rules_with_compliance": rules_with_compliance, |
| 463 | "total_rules": len(wazuh_rules_cache.get_all_rules()), |
| 464 | "firing_stats_available": wazuh_firing_stats_cache.is_available, |
| 465 | } |
| 466 | |
| 467 | |
| 468 | def list_compliance_frameworks() -> List[Dict[str, str]]: |
| 469 | """List the frameworks the compliance pivot supports. Drives the UI selector.""" |
| 470 | return [{"key": key, "label": meta["label"]} for key, meta in COMPLIANCE_FRAMEWORKS.items()] |
| 471 | |
| 472 | |
| 473 | # --------------------------------------------------------------------------- |
| 474 | # Logtest — "which rule would match this log line?" |
| 475 | # --------------------------------------------------------------------------- |
| 476 | |
| 477 | |
| 478 | async def run_log_test( |
| 479 | event: str, |
| 480 | log_format: str = "syslog", |
| 481 | location: str = "logtest", |
| 482 | ) -> Dict[str, Any]: |
| 483 | """ |
| 484 | Wrapper that runs Wazuh's logtest and decorates the result with details |
| 485 | the catalog already has cached (so the UI gets one consistent rule |
| 486 | shape regardless of whether it's reading from the index table, the |
| 487 | detail modal, or a logtest match). |
| 488 | |
| 489 | Returns a dict the route can serialize directly: |
| 490 | - ``matched``: bool |
| 491 | - ``rule_id``, ``description``, ``level``, ``groups``, ``mitre``, |
| 492 | ``tactics``: from the logtest output, enriched with the catalog's |
| 493 | mitre_matrix-resolved tactic display names when we have them |
| 494 | - ``alert``: the full Wazuh alert envelope (decoder/predecoder/data) |
| 495 | - ``unavailable_reason``: filled when the logtest call itself failed |
| 496 | so the UI can show an inline error |
| 497 | """ |
| 498 | from app.connectors.wazuh_manager.services.logtest import run_logtest |
| 499 | |
| 500 | await mitre_matrix.ensure_loaded() |
| 501 | |
| 502 | try: |
| 503 | result = await run_logtest(event=event, log_format=log_format, location=location) |
| 504 | except Exception as exc: # noqa: BLE001 |
| 505 | # Service exceptions are converted to a structured "unavailable" |
| 506 | # response rather than re-raised — the catalog should feel like a |
| 507 | # stable surface; the route wraps real HTTPException for input- |
| 508 | # validation errors but transport hiccups are reported inline. |
| 509 | logger.warning(f"Logtest failed: {exc}") |
| 510 | return { |
| 511 | "matched": False, |
| 512 | "rule": None, |
| 513 | "alert": None, |
| 514 | "tactics": [], |
| 515 | "unavailable_reason": str(exc), |
| 516 | } |
| 517 | |
| 518 | tactics: List[str] = [] |
| 519 | rule = result.get("rule") |
| 520 | if rule: |
| 521 | # Enrich with resolved tactic display names using the catalog's |
| 522 | # mitre_matrix — Wazuh only emits T-IDs; we want analyst-readable |
| 523 | # tactic names too so the result panel matches the rest of the catalog. |
| 524 | tactics = _resolve_tactics_for_mitre_ids(rule.get("mitre") or []) |
| 525 | |
| 526 | return { |
| 527 | "matched": result["matched"], |
| 528 | "rule": rule, |
| 529 | "alert": result.get("alert"), |
| 530 | "tactics": tactics, |
| 531 | "unavailable_reason": None, |
| 532 | } |
| 533 | |
| 534 | |
| 535 | # --------------------------------------------------------------------------- |
| 536 | # Coverage Gaps — MITRE techniques unaddressed by either rule corpus |
| 537 | # --------------------------------------------------------------------------- |
| 538 | |
| 539 | |
| 540 | async def list_coverage_gaps() -> Dict[str, Any]: |
| 541 | """ |
| 542 | Return every MITRE ATT&CK technique that is NOT covered by any rule in |
| 543 | the CoPilot Searches corpus OR the Wazuh ruleset. |
| 544 | |
| 545 | A technique is "covered" if at least one rule (from either source) |
| 546 | declares its T-ID or any sub-technique thereof. We normalize to base |
| 547 | technique IDs (``T1059.007 → T1059``) so a sub-technique counts as |
| 548 | coverage for its parent — same convention used elsewhere in this file |
| 549 | when resolving tactics for a rule. |
| 550 | |
| 551 | Output is sorted by technique ID so the UI list is stable across loads. |
| 552 | Sub-techniques are deliberately collapsed into their parents — analysts |
| 553 | care about "do we have anything for PowerShell (T1059)?" not "do we |
| 554 | have anything for T1059.001 specifically?". Listing every sub-technique |
| 555 | individually would flood the gap report with thousands of rows. |
| 556 | """ |
| 557 | await rules_cache.ensure_loaded() |
| 558 | await mitre_matrix.ensure_loaded() |
| 559 | await wazuh_rules_cache.ensure_loaded() |
| 560 | |
| 561 | # Collect the set of covered base-technique IDs across both corpora. |
| 562 | covered_ids: set[str] = set() |
| 563 | for rule in rules_cache.get_all_rules(): |
| 564 | for tid in _rule_mitre_ids(rule): |
| 565 | base = tid.split(".")[0].strip().upper() |
| 566 | if base: |
| 567 | covered_ids.add(base) |
| 568 | for rule in wazuh_rules_cache.get_all_rules(): |
| 569 | for tid in _wazuh_mitre_ids(rule): |
| 570 | base = tid.split(".")[0].strip().upper() |
| 571 | if base: |
| 572 | covered_ids.add(base) |
| 573 | |
| 574 | techniques = mitre_matrix.techniques |
| 575 | tactics_by_short = {t["short_name"]: t["name"] for t in mitre_matrix.tactics} |
| 576 | |
| 577 | gaps: List[Dict[str, Any]] = [] |
| 578 | base_technique_count = 0 |
| 579 | for tid, meta in techniques.items(): |
| 580 | # Skip sub-techniques in the gap list — see docstring rationale. |
| 581 | if meta.get("is_subtechnique"): |
| 582 | continue |
| 583 | base_technique_count += 1 |
| 584 | if tid in covered_ids: |
| 585 | continue |
| 586 | |
| 587 | tactic_names: List[str] = [] |
| 588 | for short in meta.get("tactic_short_names", []): |
| 589 | display = tactics_by_short.get(short) |
| 590 | if display and display not in tactic_names: |
| 591 | tactic_names.append(display) |
| 592 | |
| 593 | gaps.append( |
| 594 | { |
| 595 | "technique_id": tid, |
| 596 | "technique_name": meta.get("name", tid), |
| 597 | "tactics": tactic_names, |
| 598 | "url": meta.get("url"), |
| 599 | }, |
| 600 | ) |
| 601 | |
| 602 | gaps.sort(key=lambda g: g["technique_id"]) |
| 603 | |
| 604 | covered_count = base_technique_count - len(gaps) |
| 605 | coverage_pct = (covered_count / base_technique_count * 100) if base_technique_count else 0.0 |
| 606 | |
| 607 | return { |
| 608 | "gaps": gaps, |
| 609 | "gap_count": len(gaps), |
| 610 | "covered_count": covered_count, |
| 611 | "total_techniques": base_technique_count, |
| 612 | "coverage_pct": round(coverage_pct, 1), |
| 613 | } |
| 614 | |
| 615 | |
| 616 | # --------------------------------------------------------------------------- |
| 617 | # Wazuh Rules tab — list + single-rule detail |
| 618 | # --------------------------------------------------------------------------- |
| 619 | |
| 620 | |
| 621 | def _wazuh_groups(rule: Dict[str, Any]) -> List[str]: |
| 622 | """ |
| 623 | Wazuh's ``groups`` field is already a list in the schema but Wazuh |
| 624 | occasionally returns it as a comma-joined string. Tolerate both. |
| 625 | """ |
| 626 | raw = rule.get("groups") |
| 627 | if isinstance(raw, list): |
| 628 | return [g for g in raw if isinstance(g, str) and g.strip()] |
| 629 | if isinstance(raw, str): |
| 630 | return [g.strip() for g in raw.split(",") if g.strip()] |
| 631 | return [] |
| 632 | |
| 633 | |
| 634 | def _wazuh_mitre_ids(rule: Dict[str, Any]) -> List[str]: |
| 635 | """ |
| 636 | Same tolerance as ``_wazuh_groups`` — Wazuh's ``mitre`` field is sometimes |
| 637 | a list of T-IDs, sometimes a single string. Normalize. |
| 638 | """ |
| 639 | raw = rule.get("mitre") |
| 640 | if isinstance(raw, list): |
| 641 | return [m for m in raw if isinstance(m, str) and m.strip()] |
| 642 | if isinstance(raw, str): |
| 643 | return [m.strip() for m in raw.split(",") if m.strip()] |
| 644 | return [] |
| 645 | |
| 646 | |
| 647 | def _wazuh_compliance(rule: Dict[str, Any]) -> Dict[str, List[str]]: |
| 648 | """ |
| 649 | Group all compliance-framework arrays into one nested dict so the UI can |
| 650 | render them as labelled chip groups without hard-coding the framework |
| 651 | list in the frontend. |
| 652 | """ |
| 653 | return { |
| 654 | "pci_dss": _ensure_list_of_str(rule.get("pci_dss")), |
| 655 | "gdpr": _ensure_list_of_str(rule.get("gdpr")), |
| 656 | "hipaa": _ensure_list_of_str(rule.get("hipaa")), |
| 657 | "nist_800_53": _ensure_list_of_str(rule.get("nist_800_53")), |
| 658 | "tsc": _ensure_list_of_str(rule.get("tsc")), |
| 659 | "gpg13": _ensure_list_of_str(rule.get("gpg13")), |
| 660 | } |
| 661 | |
| 662 | |
| 663 | def _wazuh_row(rule: Dict[str, Any]) -> Dict[str, Any]: |
| 664 | """ |
| 665 | Project a cached Wazuh rule down to the columns the index table renders. |
| 666 | |
| 667 | Firing counts come from ``wazuh_firing_stats_cache`` if available — a |
| 668 | missing entry means "0 hits in the 30d window", not "no data". The cache's |
| 669 | own ``is_available`` flag (mirrored on the list envelope as |
| 670 | ``firing_stats_available``) is what tells the UI whether to render the |
| 671 | Hits column at all vs. hiding it because we can't tell. |
| 672 | """ |
| 673 | rid = rule.get("id") |
| 674 | stats = wazuh_firing_stats_cache.get(rid) if isinstance(rid, int) else {"hits_30d": 0, "hits_7d": 0, "last_seen": None} |
| 675 | return { |
| 676 | "id": rid, |
| 677 | "level": rule.get("level"), |
| 678 | "status": rule.get("status"), |
| 679 | "description": rule.get("description") or "", |
| 680 | "filename": rule.get("filename") or "", |
| 681 | "relative_dirname": rule.get("relative_dirname") or "", |
| 682 | "groups": _wazuh_groups(rule), |
| 683 | "mitre": _wazuh_mitre_ids(rule), |
| 684 | "hits_7d": stats["hits_7d"], |
| 685 | "hits_30d": stats["hits_30d"], |
| 686 | "last_seen": stats.get("last_seen"), |
| 687 | } |
| 688 | |
| 689 | |
| 690 | async def list_wazuh_rules(customer_code: Optional[str] = None) -> Dict[str, Any]: |
| 691 | """ |
| 692 | Return the full cached Wazuh ruleset projected to the index-table shape. |
| 693 | |
| 694 | Returns the whole corpus in one shot (typical Wazuh install ships ~3–5k |
| 695 | rules, ~3–5 MB JSON). Filtering and pagination happen client-side in the |
| 696 | same pattern as the Stories index — keeps the API simple and the UX |
| 697 | responsive (no round-trip per filter keystroke). |
| 698 | |
| 699 | When ``customer_code`` is set, firing counts (hits_7d / hits_30d / |
| 700 | last_seen) are scoped to that customer's alerts via a fresh ES query |
| 701 | (not cached — see ``fetch_firing_stats_for_customer`` for rationale). |
| 702 | The rule list itself is the same — every rule is shown — but the hit |
| 703 | columns reflect "what's been firing for THIS customer." Rules without |
| 704 | any hits for the customer get zeros. |
| 705 | |
| 706 | Always returns a populated envelope. When Wazuh is unavailable, ``rules`` |
| 707 | is empty and ``available=False`` + ``unavailable_reason`` carries the |
| 708 | explanation so the frontend can render an inline empty state. |
| 709 | """ |
| 710 | await wazuh_rules_cache.ensure_loaded() |
| 711 | # Load firing stats too — they're cached separately with their own TTL, |
| 712 | # so this is cheap. _wazuh_row reads from the firing-stats cache below. |
| 713 | await wazuh_firing_stats_cache.ensure_loaded() |
| 714 | |
| 715 | # Per-customer override: fetch a fresh per-customer aggregation and |
| 716 | # splice it into each row in place of the global stats. We don't mutate |
| 717 | # the row builder — instead build rows with global stats, then patch the |
| 718 | # firing fields. Keeps _wazuh_row's semantics clean and isolated. |
| 719 | customer_stats: Dict[int, Dict[str, Any]] = {} |
| 720 | if customer_code: |
| 721 | customer_stats = await fetch_firing_stats_for_customer(customer_code) |
| 722 | |
| 723 | rows: List[Dict[str, Any]] = [] |
| 724 | for raw in wazuh_rules_cache.get_all_rules(): |
| 725 | row = _wazuh_row(raw) |
| 726 | if customer_code: |
| 727 | # Override stats with the per-customer numbers. Missing rule_id |
| 728 | # in customer_stats means "this rule hasn't fired for this |
| 729 | # customer" — zero everything out rather than leaving the global |
| 730 | # numbers in place, which would be misleading. |
| 731 | rid = row.get("id") |
| 732 | cstats = customer_stats.get(rid) if isinstance(rid, int) else None |
| 733 | if cstats: |
| 734 | row["hits_7d"] = cstats["hits_7d"] |
| 735 | row["hits_30d"] = cstats["hits_30d"] |
| 736 | row["last_seen"] = cstats.get("last_seen") |
| 737 | else: |
| 738 | row["hits_7d"] = 0 |
| 739 | row["hits_30d"] = 0 |
| 740 | row["last_seen"] = None |
| 741 | rows.append(row) |
| 742 | |
| 743 | # Sort by integer ID for a stable, intuitive default order — operators |
| 744 | # think of Wazuh rules numerically. |
| 745 | rows.sort(key=lambda r: r["id"] if isinstance(r["id"], int) else 0) |
| 746 | |
| 747 | return { |
| 748 | "rules": rows, |
| 749 | "total": len(rows), |
| 750 | "available": wazuh_rules_cache.is_available, |
| 751 | "unavailable_reason": wazuh_rules_cache.unavailable_reason, |
| 752 | "last_refresh": wazuh_rules_cache.last_refresh, |
| 753 | # Firing-stats availability mirrored on the envelope so the UI knows |
| 754 | # whether to render the Hits column (no point showing "0" for every |
| 755 | # row when the indexer isn't reachable — that'd be misleading). |
| 756 | "firing_stats_available": wazuh_firing_stats_cache.is_available, |
| 757 | "firing_stats_unavailable_reason": wazuh_firing_stats_cache.unavailable_reason, |
| 758 | "firing_stats_last_refresh": wazuh_firing_stats_cache.last_refresh, |
| 759 | # Echoes the request so the UI can confirm the scope it's showing. |
| 760 | # Empty string when the global view is being served. |
| 761 | "customer_code": customer_code or "", |
| 762 | } |
| 763 | |
| 764 | |
| 765 | async def get_wazuh_rule_detail(rule_id: int) -> Optional[Dict[str, Any]]: |
| 766 | """ |
| 767 | Build the full meta payload for a single Wazuh rule — fed into the detail |
| 768 | modal. Returns ``None`` when the cache doesn't know that ID (caller maps |
| 769 | that to 404). |
| 770 | |
| 771 | No second Wazuh API call: everything the UI needs is already on the cached |
| 772 | list response. The ``details`` field carries the if-then logic (if_sid, |
| 773 | match, regex, decoded_as, …) which the modal pretty-prints as chips, and |
| 774 | ``source_xml`` is a reconstructed ``<rule>`` block the modal renders as a |
| 775 | code snippet — analysts asked to "see how the rule is written" without |
| 776 | the auth and parsing overhead of fetching the original ``.xml`` file. |
| 777 | """ |
| 778 | await wazuh_rules_cache.ensure_loaded() |
| 779 | await mitre_matrix.ensure_loaded() |
| 780 | await wazuh_firing_stats_cache.ensure_loaded() |
| 781 | |
| 782 | rule = wazuh_rules_cache.get_rule(rule_id) |
| 783 | if rule is None: |
| 784 | return None |
| 785 | |
| 786 | mitre_ids = _wazuh_mitre_ids(rule) |
| 787 | tactic_names = _resolve_tactics_for_mitre_ids(mitre_ids) |
| 788 | stats = wazuh_firing_stats_cache.get(rule_id) |
| 789 | |
| 790 | return { |
| 791 | "id": rule.get("id"), |
| 792 | "level": rule.get("level"), |
| 793 | "status": rule.get("status"), |
| 794 | "description": rule.get("description") or "", |
| 795 | "filename": rule.get("filename") or "", |
| 796 | "relative_dirname": rule.get("relative_dirname") or "", |
| 797 | "groups": _wazuh_groups(rule), |
| 798 | "mitre": mitre_ids, |
| 799 | "tactics": tactic_names, # resolved via mitre_matrix for richer display |
| 800 | "compliance": _wazuh_compliance(rule), |
| 801 | # The if/then logic dict, pretty-printed by the modal as labelled |
| 802 | # rows. Passed through as-is so the UI can iterate any keys Wazuh |
| 803 | # decides to emit (we don't want to hard-code the field list). |
| 804 | "details": rule.get("details") or {}, |
| 805 | # Reconstructed XML for the "Rule Source" section. See |
| 806 | # _synthesize_rule_xml below for the why-not-fetch-the-file rationale. |
| 807 | "source_xml": _synthesize_rule_xml(rule), |
| 808 | # Firing counts from the indexer. ``firing_stats_available`` mirrors |
| 809 | # the cache state — the modal hides the hits panel entirely when |
| 810 | # we can't pull stats (don't show "0 hits" when we really mean |
| 811 | # "indexer unreachable, no idea"). |
| 812 | "hits_7d": stats["hits_7d"], |
| 813 | "hits_30d": stats["hits_30d"], |
| 814 | "last_seen": stats.get("last_seen"), |
| 815 | "firing_stats_available": wazuh_firing_stats_cache.is_available, |
| 816 | "firing_stats_unavailable_reason": wazuh_firing_stats_cache.unavailable_reason, |
| 817 | } |
| 818 | |
| 819 | |
| 820 | # --------------------------------------------------------------------------- |
| 821 | # Rule source XML synthesis |
| 822 | # --------------------------------------------------------------------------- |
| 823 | |
| 824 | |
| 825 | def _synthesize_rule_xml(rule: Dict[str, Any]) -> str: |
| 826 | """ |
| 827 | Reconstruct the ``<rule>`` XML block for a Wazuh rule from its cached |
| 828 | metadata. |
| 829 | |
| 830 | Why synthesize instead of fetching the original file via Wazuh's |
| 831 | ``GET /rules/files/{filename}``: |
| 832 | |
| 833 | - That endpoint returns the **whole file** — a single .xml file routinely |
| 834 | contains dozens of rules wrapped in ``<group>`` blocks. We'd have to |
| 835 | parse it and extract just the rule with the matching ``id``, which |
| 836 | adds an XML-parsing dependency and a second network round-trip. |
| 837 | - The original file endpoint is also admin-scoped. Proxying it through |
| 838 | the catalog (which is admin|analyst|customer_user) means an extra |
| 839 | route, extra schema, and extra service plumbing for what amounts to |
| 840 | formatting. |
| 841 | - Every field we'd need to fill the XML is already on the cached rule — |
| 842 | ``id``, ``level``, ``description``, ``groups``, ``mitre``, compliance |
| 843 | arrays, and the free-form ``details`` dict carrying the if-then logic. |
| 844 | |
| 845 | The result is functionally identical to what an analyst would see if they |
| 846 | cracked open the source ``.xml`` file: a clean ``<rule id="…" level="…">`` |
| 847 | block with the same children. It may not be byte-identical (comments, |
| 848 | original whitespace, attribute order) — if anyone needs the exact source |
| 849 | later we can layer a "View raw file" button on top of a future |
| 850 | ``/catalog/wazuh-rules/{id}/source`` endpoint. |
| 851 | """ |
| 852 | rid = rule.get("id") |
| 853 | level = rule.get("level") |
| 854 | # Header attributes — id + level only. Wazuh's other rule attributes |
| 855 | # (frequency, timeframe, maxsize, …) live under ``details`` and are |
| 856 | # rendered as child elements below for consistency. |
| 857 | header_parts = [f'id="{rid}"' if rid is not None else None, f'level="{level}"' if level is not None else None] |
| 858 | header = " ".join(p for p in header_parts if p) |
| 859 | |
| 860 | lines: List[str] = [f"<rule {header}>" if header else "<rule>"] |
| 861 | |
| 862 | # Children. Order tries to match what hand-written Wazuh rule files look |
| 863 | # like in practice: decoded_as → description → group → if-then logic → |
| 864 | # mitre → compliance. The exact order isn't load-bearing but it keeps |
| 865 | # the output readable. |
| 866 | details = rule.get("details") or {} |
| 867 | |
| 868 | # decoded_as is conventionally near the top. |
| 869 | if "decoded_as" in details: |
| 870 | lines.append(f" <decoded_as>{_xml_escape(_stringify_detail_value(details['decoded_as']))}</decoded_as>") |
| 871 | |
| 872 | description = rule.get("description") |
| 873 | if description: |
| 874 | lines.append(f" <description>{_xml_escape(description)}</description>") |
| 875 | |
| 876 | # Groups are stored as a list in the cache; Wazuh's file convention is a |
| 877 | # single ``<group>`` with comma-separated values (trailing comma included). |
| 878 | groups = _wazuh_groups(rule) |
| 879 | if groups: |
| 880 | joined = ",".join(groups) + "," |
| 881 | lines.append(f" <group>{_xml_escape(joined)}</group>") |
| 882 | |
| 883 | # Remaining details — anything we haven't already rendered above. We |
| 884 | # iterate whatever keys Wazuh emits so new fields appear automatically. |
| 885 | rendered_keys = {"decoded_as"} |
| 886 | for key, value in details.items(): |
| 887 | if key in rendered_keys: |
| 888 | continue |
| 889 | lines.append(_render_detail_element(key, value)) |
| 890 | |
| 891 | # MITRE — Wazuh's convention is a single <mitre> wrapper with <id> |
| 892 | # children per technique. |
| 893 | mitre_ids = _wazuh_mitre_ids(rule) |
| 894 | if mitre_ids: |
| 895 | lines.append(" <mitre>") |
| 896 | for mid in mitre_ids: |
| 897 | lines.append(f" <id>{_xml_escape(mid)}</id>") |
| 898 | lines.append(" </mitre>") |
| 899 | |
| 900 | # Compliance arrays — one element per value, matching Wazuh's source |
| 901 | # format (multiple <pci_dss>X.Y.Z</pci_dss> rather than a list). |
| 902 | compliance = _wazuh_compliance(rule) |
| 903 | for framework_key, values in compliance.items(): |
| 904 | for v in values: |
| 905 | # Use the schema key verbatim except for the NIST renaming, which |
| 906 | # uses the hyphenated form in source files. |
| 907 | tag = "nist-800-53" if framework_key == "nist_800_53" else framework_key |
| 908 | lines.append(f" <{tag}>{_xml_escape(v)}</{tag}>") |
| 909 | |
| 910 | lines.append("</rule>") |
| 911 | return "\n".join(lines) |
| 912 | |
| 913 | |
| 914 | def _render_detail_element(key: str, value: Any) -> str: |
| 915 | """ |
| 916 | Render one ``details`` entry as an XML child element. |
| 917 | |
| 918 | Wazuh's API returns mixed shapes for these. A few cases worth handling: |
| 919 | |
| 920 | - **String** → ``<key>value</key>``. The common case (if_sid, regex, …). |
| 921 | - **Dict with a "pattern" key** → ``<key>pattern_value</key>``. Wazuh's |
| 922 | API sometimes wraps match/regex values in ``{"pattern": "…"}``; the |
| 923 | analyst-facing XML just shows the text content. |
| 924 | - **Dict with other keys** → treat the dict as attributes + a possible |
| 925 | text body. Mirrors Wazuh's ``<match type="pcre2">…</match>`` shape. |
| 926 | - **List** → emit one element per item (e.g. multiple ``<match>`` lines). |
| 927 | - **Anything else** → JSON-encode as the text body. Last-resort fallback |
| 928 | so we don't silently drop unknown shapes. |
| 929 | """ |
| 930 | if isinstance(value, list): |
| 931 | # One element per list entry, recursively rendered. |
| 932 | return "\n".join(_render_detail_element(key, item) for item in value) |
| 933 | |
| 934 | if isinstance(value, dict): |
| 935 | # The "pattern" wrapper case — flatten to text content. |
| 936 | if set(value.keys()) == {"pattern"}: |
| 937 | return f" <{key}>{_xml_escape(_stringify_detail_value(value['pattern']))}</{key}>" |
| 938 | # Mixed dict — pull out any text-ish key as the body, treat the rest |
| 939 | # as attributes. This is a heuristic; Wazuh's exact shape varies. |
| 940 | text_keys = {"pattern", "value", "text", "#text"} |
| 941 | attrs: Dict[str, Any] = {} |
| 942 | body: Optional[str] = None |
| 943 | for k, v in value.items(): |
| 944 | if k in text_keys and body is None: |
| 945 | body = _stringify_detail_value(v) |
| 946 | else: |
| 947 | attrs[k] = v |
| 948 | attr_str = "".join(f' {ak}="{_xml_escape(str(av))}"' for ak, av in attrs.items()) |
| 949 | if body is not None: |
| 950 | return f" <{key}{attr_str}>{_xml_escape(body)}</{key}>" |
| 951 | return f" <{key}{attr_str} />" |
| 952 | |
| 953 | # Scalar — string, number, bool, None. |
| 954 | text = _stringify_detail_value(value) |
| 955 | if text == "": |
| 956 | return f" <{key} />" |
| 957 | return f" <{key}>{_xml_escape(text)}</{key}>" |
| 958 | |
| 959 | |
| 960 | def _stringify_detail_value(value: Any) -> str: |
| 961 | """Best-effort scalar-to-string. JSON-encodes complex values.""" |
| 962 | if value is None: |
| 963 | return "" |
| 964 | if isinstance(value, str): |
| 965 | return value |
| 966 | if isinstance(value, (int, float, bool)): |
| 967 | return str(value) |
| 968 | # Last resort — render unknown shapes as compact JSON rather than |
| 969 | # Python's repr (which would emit single quotes and 'True'/'None'). |
| 970 | try: |
| 971 | import json |
| 972 | |
| 973 | return json.dumps(value, separators=(",", ":")) |
| 974 | except Exception: |
| 975 | return str(value) |
| 976 | |
| 977 | |
| 978 | def _xml_escape(text: str) -> str: |
| 979 | """ |
| 980 | Minimal XML escaping for the five reserved characters. We don't use |
| 981 | ``xml.sax.saxutils.escape`` because we need to handle ``"`` and ``'`` as |
| 982 | well (they appear inside attribute values, and we use both quote styles |
| 983 | in synthesized output). |
| 984 | """ |
| 985 | return text.replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """).replace("'", "'") |