| 1 | """ |
| 2 | Thin wrapper around Wazuh Manager's ``PUT /logtest`` endpoint. |
| 3 | |
| 4 | What this is for: |
| 5 | The Detections Catalog "Test a log" feature lets analysts paste a raw log |
| 6 | line and asks Wazuh "which rule(s) would have matched this?". Rather than |
| 7 | re-implementing Wazuh's decoder + rule engine in Python (months of work, |
| 8 | guaranteed drift), we use Wazuh's own logtest API — same engine that runs |
| 9 | in production, no semantic gap. |
| 10 | |
| 11 | Stateless mode only: |
| 12 | Wazuh's logtest supports stateful sessions (``token`` field) so multi-line |
| 13 | correlated rules can fire across calls. We don't need that for the |
| 14 | catalog's "test one log line" use case, and stateless calls don't leave |
| 15 | session state hanging on the Manager. If we ever want multi-line tests |
| 16 | we can extend with a ``token`` param + a ``DELETE /logtest/sessions/{token}`` |
| 17 | cleanup call. |
| 18 | |
| 19 | No DB writes, no schema changes — pure HTTP wrapper. |
| 20 | """ |
| 21 | |
| 22 | import json |
| 23 | from typing import Any |
| 24 | from typing import Dict |
| 25 | from typing import Optional |
| 26 | |
| 27 | from fastapi import HTTPException |
| 28 | from loguru import logger |
| 29 | |
| 30 | from app.connectors.wazuh_manager.utils.universal import send_put_request |
| 31 | |
| 32 | |
| 33 | async def run_logtest( |
| 34 | event: str, |
| 35 | log_format: str = "syslog", |
| 36 | location: str = "logtest", |
| 37 | ) -> Dict[str, Any]: |
| 38 | """ |
| 39 | Run a stateless logtest against the Wazuh Manager. |
| 40 | |
| 41 | Args: |
| 42 | event: The raw log line to evaluate (single line, no JSON envelope). |
| 43 | log_format: Wazuh log format. Common values: ``syslog`` (default), |
| 44 | ``json``, ``snort-full``, ``squid``, ``apache``, ``iis``, etc. |
| 45 | ``syslog`` is the most permissive and works for most operator- |
| 46 | captured log lines. |
| 47 | location: A pseudo-source label Wazuh records on the test. We pass |
| 48 | ``"logtest"`` by default — Wazuh uses this to scope ``if_*`` |
| 49 | location-based rule conditions, and a generic value avoids |
| 50 | accidentally matching location-specific rules. |
| 51 | |
| 52 | Returns: |
| 53 | A dict with keys: |
| 54 | - ``matched`` (bool): whether any rule matched |
| 55 | - ``rule`` (dict|None): the matched rule's summary (id, level, |
| 56 | description, groups, mitre, …) when matched, else None |
| 57 | - ``alert`` (dict|None): the full Wazuh alert envelope (decoder, |
| 58 | predecoder, data fields, full_log, …) — kept for the UI's |
| 59 | "what did Wazuh actually parse?" panel |
| 60 | - ``raw`` (dict): the unmodified Wazuh response payload, kept for |
| 61 | debugging when ``matched`` is False but the analyst expects a hit |
| 62 | |
| 63 | Raises: |
| 64 | HTTPException(400): event is empty / invalid input |
| 65 | HTTPException(503): Wazuh Manager unreachable / refused the request |
| 66 | """ |
| 67 | if not event or not event.strip(): |
| 68 | raise HTTPException(status_code=400, detail="event must be a non-empty log line") |
| 69 | |
| 70 | payload = { |
| 71 | "event": event, |
| 72 | "log_format": log_format, |
| 73 | "location": location, |
| 74 | } |
| 75 | |
| 76 | logger.debug(f"Running Wazuh logtest with format={log_format} location={location}") |
| 77 | |
| 78 | # WAZUH-PUT GOTCHA: send_put_request uses ``requests.put(data=...)`` which |
| 79 | # form-encodes dicts (key=value&key=value), but the Content-Type header is |
| 80 | # set to application/json. Wazuh's logtest endpoint then tries to parse |
| 81 | # the form-encoded body as JSON and 400s with |
| 82 | # ``Expecting value: line 1 column 1 (char 0)``. Pre-serializing to a JSON |
| 83 | # string sidesteps it — ``requests`` sends strings as the raw body without |
| 84 | # form encoding, so Wazuh sees actual JSON. (Cleaner fix would be a |
| 85 | # ``json_data=True`` flag on send_put_request, but that touches shared |
| 86 | # connector code used by every other Wazuh integration.) |
| 87 | response = await send_put_request(endpoint="/logtest", data=json.dumps(payload)) |
| 88 | |
| 89 | if not response or not response.get("success"): |
| 90 | # send_put_request returns a dict with success=False on transport |
| 91 | # failures; bubble its message up so the UI can show why. |
| 92 | raise HTTPException( |
| 93 | status_code=503, |
| 94 | detail=response.get("message", "Wazuh Manager logtest failed") if response else "Wazuh Manager not reachable", |
| 95 | ) |
| 96 | |
| 97 | # Wazuh logtest wraps everything two layers deep: |
| 98 | # response["data"]["data"]["output"] holds the actual logtest result. |
| 99 | # Be defensive — version drift has changed this shape before. |
| 100 | raw_payload = response.get("data") or {} |
| 101 | inner = raw_payload.get("data") or {} |
| 102 | output = inner.get("output") or {} |
| 103 | |
| 104 | rule_summary = _extract_rule_summary(output) |
| 105 | matched = rule_summary is not None |
| 106 | |
| 107 | return { |
| 108 | "matched": matched, |
| 109 | "rule": rule_summary, |
| 110 | "alert": output if output else None, |
| 111 | "raw": raw_payload, |
| 112 | } |
| 113 | |
| 114 | |
| 115 | def _extract_rule_summary(output: Dict[str, Any]) -> Optional[Dict[str, Any]]: |
| 116 | """ |
| 117 | Pull the matched-rule summary out of a Wazuh logtest output, normalizing |
| 118 | field names so the frontend can render a stable shape. |
| 119 | |
| 120 | Wazuh's logtest output puts the matched rule under ``output.rule`` — |
| 121 | same structure as a regular alert envelope. If no rule matched the |
| 122 | ``rule`` key is missing or the rule's id is 0, both of which mean |
| 123 | "no match" for our purposes. |
| 124 | """ |
| 125 | rule = output.get("rule") |
| 126 | if not isinstance(rule, dict): |
| 127 | return None |
| 128 | |
| 129 | rid = rule.get("id") |
| 130 | # Wazuh sometimes returns rule.id as a string; normalize to int when possible. |
| 131 | try: |
| 132 | rid_int = int(rid) if rid is not None else None |
| 133 | except (TypeError, ValueError): |
| 134 | rid_int = None |
| 135 | |
| 136 | # rule.id == 0 / None == no real match (Wazuh's "rule" entry can be a |
| 137 | # synthetic envelope even when no analyst-facing rule fired). |
| 138 | if not rid_int: |
| 139 | return None |
| 140 | |
| 141 | return { |
| 142 | "id": rid_int, |
| 143 | "level": rule.get("level"), |
| 144 | "description": rule.get("description") or "", |
| 145 | "groups": rule.get("groups") or [], |
| 146 | "mitre": (rule.get("mitre") or {}).get("id") or [], |
| 147 | "pci_dss": rule.get("pci_dss") or [], |
| 148 | "gdpr": rule.get("gdpr") or [], |
| 149 | "hipaa": rule.get("hipaa") or [], |
| 150 | "nist_800_53": rule.get("nist_800_53") or [], |
| 151 | "firedtimes": rule.get("firedtimes"), |
| 152 | } |