diff --git a/integrations/vulnerability_email_dedup/README.md b/integrations/vulnerability_email_dedup/README.md new file mode 100644 index 0000000..4dc4f0c --- /dev/null +++ b/integrations/vulnerability_email_dedup/README.md @@ -0,0 +1,463 @@ +# Vulnerability Email De-duplication - Wazuh Integration + +## Table of Contents + +* [Introduction](#introduction) +* [Prerequisites](#prerequisites) +* [How It Works](#how-it-works) +* [Installation and Configuration](#installation-and-configuration) + * [Using the Integration Files](#using-the-integration-files) + * [Script Configuration](#script-configuration) + * [Integrator Configuration](#integrator-configuration) + * [Scheduled Maintenance](#scheduled-maintenance) +* [Integration Steps](#integration-steps) +* [Integration Testing](#integration-testing) +* [Troubleshooting](#troubleshooting) +* [Provenance and Maintenance](#provenance-and-maintenance) +* [Sources](#sources) + +--- + +### Introduction + +The Wazuh Vulnerability Detection module emits one alert per (package, CVE) pair. +A single package frequently carries dozens of CVEs, so one scan produces a burst +of near-identical alerts for the same package on the same agent, often hundreds +or thousands at once. Routed to email through the stock integration, that burst +becomes hundreds of near-identical messages. + +This integration is a drop-in replacement for a `custom-email` script that +collapses the burst: + +- The first alert for a given agent, package and status sends one email and + opens a 24 hour window. +- Every subsequent CVE for that same agent and package inside the window is + counted, but no email is sent. +- The window is measured from that first alert and never slides. More alerts + arriving inside it do not extend it and do not start a new one. +- Once 24 hours have passed since the first alert, the next detection emails + again and opens a fresh window, so a package that stays vulnerable stays + visible without becoming noise. +- Non-vulnerability alerts are passed through and emailed unchanged, so the + script can also serve as the general email path. + +How much noise this removes depends on how densely CVEs cluster per package. In +a synthetic load test, 2,000 vulnerability alerts spread across 300 distinct +`(agent, package, status)` combinations, fed through the script at 64-way +concurrency, produced exactly 300 emails: an 85% reduction, one notification per +affected package rather than one per CVE. The sum of the recorded per-row CVE +counts equalled the 2,000 alerts fed in, confirming no updates were lost to the +concurrent writers. + +--- + +### Prerequisites + +- Wazuh manager with the Vulnerability Detection module enabled. +- Vulnerability alerts using the Wazuh 4.8+ schema, that is + `data.vulnerability.cve`, `data.vulnerability.package.name`, + `data.vulnerability.package.version` and `data.vulnerability.status`. +- Python 3.6 or later. The script uses only the standard library (`sqlite3`, + `smtplib`, `json`), so no `pip install` step is required. The Wazuh embedded + interpreter at `/var/ossec/framework/python/bin/python3` satisfies this. +- A reachable SMTP relay. The defaults assume a local MTA on `127.0.0.1:25`. +- Write access to the directory holding the de-duplication database. The default + is `/var/ossec/logs/`, which is already writable by the `wazuh` user. + +--- + +### How It Works + +Each `(agent_id, package_name, status)` gets one row, anchored at `first_seen`: +the moment its first alert arrived. Every alert resolves to one of three cases, +decided entirely at read time: + +| State of the row | Decision | +| --- | --- | +| No row exists | Send one email, create the row anchored at now. | +| `now - first_seen >= 24h` | The window has elapsed. Send one email, re-anchor `first_seen` at now. | +| Inside the window | Suppress. Increment the counter and **leave `first_seen` untouched**. | + +That last line is the whole de-duplication rule. Because a suppressed alert +never moves the anchor, no volume of alerts can push the window forward, which +is the failure mode of a last-seen window: a package detected every few hours +would slide its own window indefinitely and never notify again. + +The package **version** is shown in the email body (read from the triggering +alert, not persisted) but is deliberately kept out of the key, so a version bump +that is still vulnerable does not re-notify. Move it into the key if you want +per-version notifications. + +`wazuh-integratord` runs an integration script **once per alert, as a separate +OS process**. A burst of 1,000 alerts is therefore 1,000 short-lived processes, +not one process handling 1,000 items. Three properties make that safe and cheap: + +- **Cheap lookup.** Per alert the script does one indexed SQLite lookup and one + small write. Suppressed alerts exit in under a millisecond, without ever + opening an SMTP connection. Only the notifying alert pays for sending mail. +- **Safe concurrency.** The store runs in WAL mode with a busy timeout, and the + check-and-write happens inside a single `BEGIN IMMEDIATE` transaction. + Concurrent processes for the same package serialize on the write lock and + exactly one wins at each window boundary, so a burst produces no + `database is locked` errors, no lost updates, and no duplicate emails. +- **Nothing runs between alerts.** No daemon, no polling, no in-memory state. + The table holds one row per currently active package and agent, so it stays in + the kilobytes. + +If the database cannot be opened or the de-duplication check raises, the script +**fails open** and sends the email. Losing a notification is worse than sending +a duplicate. + +```mermaid +graph TD + A[Vulnerability alert] --> B{Has package name?} + B -- No --> C[Send email, stock behaviour] + B -- Yes --> D[BEGIN IMMEDIATE on SQLite] + D --> E{Row exists and
within 24h of first_seen?} + E -- No --> F[Insert or re-anchor
first_seen = now, COMMIT] + F --> G[Send one email] + E -- Yes --> H[Increment cve_count,
first_seen unchanged, COMMIT] + H --> I[Exit, no email] +``` + +--- + +### Installation and Configuration + +#### Using the Integration Files + +This integration ships two files: + +| File | Purpose | +| --- | --- | +| `custom-email.py` | The integration logic. | +| `custom-email` | Standard Wazuh shell wrapper that invokes the script with the embedded Python interpreter. | + +Copy both to the manager: + +```bash +cp custom-email custom-email.py /var/ossec/integrations/ +``` + +Apply the ownership and permissions Wazuh requires. These are not optional: +`wazuh-integratord` refuses to execute an integration file that is +group- or world-writable, and logs `has write permissions` instead of running. + +```bash +chmod 750 /var/ossec/integrations/custom-email* +chown root:wazuh /var/ossec/integrations/custom-email* +``` + +#### Script Configuration + +Edit the configuration block at the top of `custom-email.py`. Every value also +accepts an environment variable override, so mail and storage can be redirected +into a sandbox for testing without editing the deployed file. + +| Setting | Environment variable | Default | Description | +| --- | --- | --- | --- | +| `SMTP_SERVER` | `VULN_SMTP_HOST` | `127.0.0.1` | SMTP relay host. | +| `SMTP_PORT` | `VULN_SMTP_PORT` | `25` | SMTP relay port. | +| `SENDER_EMAIL` | `VULN_SENDER` | `wazuh@example.com` | From address. | +| `RECEIVER_EMAIL` | `VULN_RECEIVER` | `soc@example.com` | Destination address. | +| `DB_PATH` | `VULN_DEDUP_DB` | `/var/ossec/logs/vuln_dedup.db` | De-duplication store. | +| `LOG_PATH` | `VULN_DEDUP_LOG` | `/var/ossec/logs/custom-email_integration.log` | Script log file. | +| `DEDUP_TTL_SECONDS` | `VULN_DEDUP_TTL` | `86400` (24h) | Window per agent, package and status, measured from the first alert. | + +`DEDUP_TTL_SECONDS` is the only behavioural knob. Set it to `43200` for a 12 +hour window or `172800` for 48 hours; nothing else in the script needs changing, +and the prune job picks up the new value automatically. + +The script does not perform SMTP authentication or STARTTLS. If your relay +requires either, extend `send_email()` with `server.starttls()` and +`server.login()`. + +#### Integrator Configuration + +Add the integration block to `/var/ossec/etc/ossec.conf` on the manager, inside +``: + +```xml + + ... + + custom-email + vulnerability-detector + json + + ... + +``` + +Notes on this block: + +- `json` is required. It is what makes the alert + arrive as the JSON file the script parses. +- `vulnerability-detector` scopes the integration to + vulnerability alerts, which is the burst this script exists to handle. +- Filters within a single `` block are ANDed. To also use this as + the general email path for high-severity alerts, add a **second** + `` block pointing at the same script with `10` + instead of the group filter. Non-vulnerability alerts are emailed unchanged. +- `` must match the filename in `/var/ossec/integrations/`, and the name + must begin with `custom-`. + +#### Scheduled Maintenance + +Cleanup is an explicit maintenance mode rather than opportunistic work on the +alert hot path. `custom-email prune` deletes rows whose window has fully elapsed +and runs a `VACUUM` to reclaim file space. + +**Prune is not load-bearing.** Every de-duplication decision is made from +`first_seen` when the alert arrives, and the read path already treats an elapsed +row as "notify and re-anchor". So if prune runs late, or skips a day, or never +runs at all, the only consequence is some rows for packages that went quiet +sitting around until the next run. No missed emails, no duplicate emails. That +also means the schedule needs no particular timing: no midnight pin, no +alignment with scan windows. + +Keep the growth in perspective: the table is bounded by the number of +**distinct** `(agent, package, status)` combinations seen in the last window, not +by alert volume. One thousand CVEs for a single package is still one row. Prune +is one `DELETE` on an indexed column, so once a day is ample. + +Since the script lives on the manager, schedule it with the manager-native +`command` wodle rather than system cron. Add to `/var/ossec/etc/ossec.conf`: + +```xml + + no + vuln-dedup-prune + /var/ossec/integrations/custom-email prune + 1d + no + yes + 60 + +``` + +Notes on this block: + +- `yes` matters. Without it the command's stdout + is forwarded to the analysis engine as an event, which is not what you want + from a maintenance job. +- Remote commands do **not** need to be enabled for this. The + `wazuh_command.remote_commands=1` setting in `local_internal_options.conf` is + only required when the command is set through a shared or centralized + configuration. This wodle lives in the manager's own local `ossec.conf`. Note + that pushing configuration through the Wazuh API or dashboard applies its own + `remote_commands` allow-list from `api.yaml`; editing the file on disk and + restarting avoids that. +- `60` is used rather than `0`. A timeout of `0` means + infinite, and there is a long-standing quirk where it can still raise a + `Timeout overtaken` error. Prune finishes in milliseconds regardless. +- `wazuh-modulesd` runs all wodles serially in one loop. Prune is fast enough + that this is a non-issue here, but it is worth remembering before giving any + other wodle an infinite timeout. + +Both `wazuh-integratord` (alert path) and `wazuh-modulesd` (prune wodle) run as +the `wazuh` user, so they share ownership of the database cleanly. The `.db`, +`.db-wal` and `.db-shm` files appear on the first alert. + +--- + +### Integration Steps + +1. Copy `custom-email` and `custom-email.py` into `/var/ossec/integrations/` and + set ownership and permissions as above. +2. Set `SMTP_SERVER`, `SENDER_EMAIL` and `RECEIVER_EMAIL` in `custom-email.py`. +3. Add the `` block and the prune `` block to + `/var/ossec/etc/ossec.conf`. +4. Restart the manager to apply the configuration: + +```bash +systemctl restart wazuh-manager +# or: /var/ossec/bin/wazuh-control restart +``` + +5. Confirm the integration loaded: + +```bash +grep -i "Enabling integration for: 'custom-email'" /var/ossec/logs/ossec.log +``` + +Data then flows as follows: the Vulnerability Detection module raises alerts, +`wazuh-integratord` matches them against the `` filter and invokes +the script once per alert, and the script either sends one email for a newly +seen package or records the CVE and exits silently. + +--- + +### Integration Testing + +#### Test 1: First alert for a package sends an email + +Write a sample alert and invoke the script directly: + +```bash +cat > /tmp/alert1.json <<'EOF' +{ + "timestamp": "2026-07-18T10:00:00.000+0000", + "location": "vulnerability-detector", + "rule": {"id": "23505", "level": 10, "description": "CVE-2026-0001 affects openssl"}, + "agent": {"id": "001", "name": "web-server-01"}, + "data": {"vulnerability": { + "cve": "CVE-2026-0001", + "severity": "High", + "status": "Active", + "package": {"name": "openssl", "version": "3.0.2-0ubuntu1.10"} + }} +} +EOF + +sudo -u wazuh /var/ossec/integrations/custom-email /tmp/alert1.json +tail -n 1 /var/ossec/logs/custom-email_integration.log +``` + +Expected: + +``` +2026-07-18T10:00:01 INFO Sent vuln alert package=openssl agent=001 cve=CVE-2026-0001 +``` + +#### Test 2: A second CVE for the same package is suppressed + +```bash +sed 's/CVE-2026-0001/CVE-2026-0002/g' /tmp/alert1.json > /tmp/alert2.json +sudo -u wazuh /var/ossec/integrations/custom-email /tmp/alert2.json +``` + +No new `Sent vuln alert` line appears. Suppressed alerts are silent at the +default `INFO` level. The row instead shows the incremented count: + +```bash +sqlite3 /var/ossec/logs/vuln_dedup.db \ + "SELECT agent_id, package, status, cve_count, + CAST(ROUND((strftime('%s','now') - first_seen) / 3600.0) AS INT) AS window_age_h + FROM dedup;" +``` + +Expected, with the anchor still showing zero hours of age because the second +alert did not move it: + +``` +001|openssl|Active|2|0 +``` + +#### Test 3: Burst behaviour and concurrency + +Generate a burst of alerts for a handful of packages and confirm that no update +is lost. The invariant to check is that the sum of `cve_count` across all rows +equals the number of alerts fed in. If that holds under concurrency, the atomic +claim is working and no process silently overwrote another. + +```bash +sqlite3 /var/ossec/logs/vuln_dedup.db \ + "SELECT SUM(cve_count) FROM dedup;" + +# highest-volume packages, that is the noisiest ones this integration collapsed +sqlite3 /var/ossec/logs/vuln_dedup.db \ + "SELECT agent_id, package, status, cve_count FROM dedup ORDER BY cve_count DESC LIMIT 10;" +``` + +#### Test 4: Window rollover, without waiting 24 hours + +The two rules worth confirming are that the window does not slide, and that it +reopens 24 hours after the **first** alert. Both can be checked immediately by +ageing the row's anchor by hand instead of waiting. + +First, prove the window does not slide. Age the anchor to 23 hours old, replay +an alert, and re-read the anchor: + +```bash +sqlite3 /var/ossec/logs/vuln_dedup.db \ + "UPDATE dedup SET first_seen = first_seen - 23*3600 WHERE package='openssl';" + +sudo -u wazuh /var/ossec/integrations/custom-email /tmp/alert2.json + +sqlite3 /var/ossec/logs/vuln_dedup.db \ + "SELECT cve_count, + CAST(ROUND((strftime('%s','now') - first_seen) / 3600.0) AS INT) AS window_age_h + FROM dedup WHERE package='openssl';" +``` + +No email is sent, and `window_age_h` is still `23`. The count rose but the +anchor did not move, so replaying more alerts can never push the window +forward. + +Now age it past the window and replay again: + +```bash +sqlite3 /var/ossec/logs/vuln_dedup.db \ + "UPDATE dedup SET first_seen = first_seen - 2*3600 WHERE package='openssl';" + +sudo -u wazuh /var/ossec/integrations/custom-email /tmp/alert2.json +tail -n 1 /var/ossec/logs/custom-email_integration.log +``` + +A fresh email is sent and the row re-anchors, so `cve_count` returns to `1` and +`window_age_h` to `0`: + +``` +2026-07-18T10:10:00 INFO Sent vuln alert package=openssl agent=001 cve=CVE-2026-0002 +``` + +#### Test 5: Maintenance mode + +```bash +sudo -u wazuh /var/ossec/integrations/custom-email prune +tail -n 1 /var/ossec/logs/custom-email_integration.log +``` + +Expected, since the row was just re-anchored and is inside its window: + +``` +2026-07-18T10:15:00 INFO Prune complete: removed 0 elapsed row(s). +``` + +Age it past the window again and prune removes it, which is the housekeeping +path. Removing an elapsed row changes no behaviour: the next alert for that +package simply creates a new row and sends one email, exactly as an elapsed row +would have. + +--- + +### Troubleshooting + +| Symptom | Likely cause | +| --- | --- | +| No emails at all, nothing in the script log | The integration did not load. Check `grep -i "Enabling integration" /var/ossec/logs/ossec.log` and confirm `` matches the filename. | +| `has write permissions` in `ossec.log` | Permissions are too broad. Re-apply `chmod 750` and `chown root:wazuh`. | +| One email per CVE, no suppression | Alerts are not matching the expected schema. Confirm `data.vulnerability.package.name` is present in the raw alert. | +| `DB open failed ... sending without dedup` | The database path is not writable by the `wazuh` user. Check `DB_PATH` and its parent directory. | +| Emails stop for a package that is still vulnerable | Expected inside the window. Check `first_seen` for that row: one fresh email is sent on the first detection after it plus `DEDUP_TTL_SECONDS`. Lower the TTL if you want more frequent reminders. | +| The dedup table is larger than expected | Prune has not run recently. It is housekeeping only, so this affects nothing but disk. Run `custom-email prune` and confirm the wodle is enabled. | +| SMTP errors in the script log | Verify relay reachability with `nc -zv 25`. | + +--- + +### Provenance and Maintenance + +- **Original source**: The Wazuh custom email integration example from the + official documentation, extended with de-duplication, a persistent store and a + maintenance mode. +- **Adapted by**: Leon Fuller. +- **Tested versions**: Wazuh manager 4.x with Vulnerability Detection enabled + and the 4.8+ vulnerability alert schema. Python 3 standard library only, no + third-party dependencies. Logic validated on Python 3.11: the window rules + above, plus a synthetic 2,000-alert burst at 64-way concurrency, where 300 + distinct keys produced 300 emails, `SUM(cve_count)` matched the 2,000 alerts + fed in, and no lock errors occurred. +- **Maintainer**: Leon Fuller. +- **Support boundary**: Community-maintained and provided as is. Not covered by + Wazuh commercial support. + +--- + +### Sources + +- +- +- +- +- diff --git a/integrations/vulnerability_email_dedup/custom-email b/integrations/vulnerability_email_dedup/custom-email new file mode 100644 index 0000000..fc3023b --- /dev/null +++ b/integrations/vulnerability_email_dedup/custom-email @@ -0,0 +1,38 @@ +#!/bin/sh +# Copyright (C) 2015, Wazuh Inc. +# Created by Wazuh, Inc. . +# This program is free software; you can redistribute it and/or modify it under the terms of GPLv2 + +WPYTHON_BIN="framework/python/bin/python3" + +SCRIPT_PATH_NAME="$0" + +DIR_NAME="$(cd $(dirname ${SCRIPT_PATH_NAME}); pwd -P)" +SCRIPT_NAME="$(basename ${SCRIPT_PATH_NAME})" + +case ${DIR_NAME} in + */active-response/bin | */wodles*) + if [ -z "${WAZUH_PATH}" ]; then + WAZUH_PATH="$(cd ${DIR_NAME}/../..; pwd)" + fi + + PYTHON_SCRIPT="${DIR_NAME}/${SCRIPT_NAME}.py" + ;; + */bin) + if [ -z "${WAZUH_PATH}" ]; then + WAZUH_PATH="$(cd ${DIR_NAME}/..; pwd)" + fi + + PYTHON_SCRIPT="${WAZUH_PATH}/framework/scripts/$(echo ${SCRIPT_NAME} | sed 's/\-/_/g').py" + ;; + */integrations) + if [ -z "${WAZUH_PATH}" ]; then + WAZUH_PATH="$(cd ${DIR_NAME}/..; pwd)" + fi + + PYTHON_SCRIPT="${DIR_NAME}/${SCRIPT_NAME}.py" + ;; +esac + + +${WAZUH_PATH}/${WPYTHON_BIN} ${PYTHON_SCRIPT} "$@" diff --git a/integrations/vulnerability_email_dedup/custom-email.py b/integrations/vulnerability_email_dedup/custom-email.py new file mode 100644 index 0000000..5d1833d --- /dev/null +++ b/integrations/vulnerability_email_dedup/custom-email.py @@ -0,0 +1,305 @@ +#!/usr/bin/env python3 +# Wazuh custom-email integration with vulnerability-alert de-duplication. +# Adapted from the Wazuh custom email integration example. +# This program is free software; you can redistribute it and/or modify it +# under the terms of GPLv2. +""" +WHAT IT SOLVES + The Vulnerability Detection module emits one alert per (package, CVE). A + single package with many CVEs produces a burst of near-identical alerts for + the same agent, often ~1000 at once. This script collapses them into one + email per (agent, package, status) per window. + +DE-DUPLICATION MODEL + Each (agent, package, status) row is anchored at first_seen, the moment its + first alert arrived. Three cases, decided at read time: + - No row: notify, anchor the window at now. + - Row whose window has fully elapsed: notify, re-anchor at now. + - Row inside the window: suppress, count it, and leave first_seen + untouched so the window never slides forward. + Because the decision is made entirely from first_seen, correctness does not + depend on the prune job running on time, or at all. + +EXECUTION MODEL + wazuh-integratord runs this script once per alert, as a separate OS process. + A burst of 1000 alerts is 1000 short-lived processes, not one process + handling 1000 items. What keeps it fast and correct is one indexed lookup + plus one small write per alert, against SQLite in WAL mode with a busy + timeout, so concurrent processes serialize on the write lock instead of + failing with "database is locked". + +DE-DUPLICATION KEY + (agent_id, package_name, status). Package version is read from the + triggering alert for the email body and deliberately kept out of the key, so + a version bump that is still vulnerable does not re-notify. Move it into the + key if you want per-version notifications. + +FIELD PATHS + Follows Wazuh 4.8+ vulnerability alerts: + data.vulnerability.cve / .package.name / .package.version / .status + + Non-vulnerability alerts are passed straight through and emailed unchanged, + so this stays a drop-in replacement for the stock integration. +""" + +import os +import sys +import json +import time +import sqlite3 +import logging + +# ------------------------------- configuration ------------------------------- +# Every value can be overridden by an environment variable so a test harness can +# redirect mail and storage into a sandbox without editing this file. +SMTP_SERVER = os.environ.get('VULN_SMTP_HOST', '127.0.0.1') +SMTP_PORT = int(os.environ.get('VULN_SMTP_PORT', '25')) +SENDER_EMAIL = os.environ.get('VULN_SENDER', 'wazuh@example.com') +RECEIVER_EMAIL = os.environ.get('VULN_RECEIVER', 'soc@example.com') + +DB_PATH = os.environ.get('VULN_DEDUP_DB', '/var/ossec/logs/vuln_dedup.db') +LOG_PATH = os.environ.get('VULN_DEDUP_LOG', + '/var/ossec/logs/custom-email_integration.log') + +# The single knob: how long one (agent, package, status) stays "already +# notified", measured from its first alert. Change this alone to move to 12h or +# 48h; nothing else in the script needs to know. +DEDUP_TTL_SECONDS = int(os.environ.get('VULN_DEDUP_TTL', 24 * 60 * 60)) + +BUSY_TIMEOUT_MS = 10000 # wait, do not error, on lock + +_LOG_FORMAT = '%(asctime)s %(levelname)s %(message)s' +_LOG_DATEFMT = '%Y-%m-%dT%H:%M:%S' +try: + logging.basicConfig(filename=LOG_PATH, filemode='a', format=_LOG_FORMAT, + datefmt=_LOG_DATEFMT, level=logging.INFO) # DEBUG would spam under bursts +except OSError: + # LOG_PATH not writable (missing directory or permissions): fall back to + # stderr so a logging problem never stops the integration from sending mail. + logging.basicConfig(stream=sys.stderr, format=_LOG_FORMAT, + datefmt=_LOG_DATEFMT, level=logging.INFO) + + +# --------------------------------- data store -------------------------------- +def open_db(): + conn = sqlite3.connect(DB_PATH, timeout=BUSY_TIMEOUT_MS / 1000.0) + try: + conn.isolation_level = None # we manage BEGIN/COMMIT ourselves + conn.execute("PRAGMA journal_mode=WAL") # concurrent readers, one writer + conn.execute("PRAGMA synchronous=NORMAL") # fast, still crash-safe in WAL + conn.execute("PRAGMA busy_timeout=%d" % BUSY_TIMEOUT_MS) + conn.execute(""" + CREATE TABLE IF NOT EXISTS dedup ( + agent_id TEXT NOT NULL, + package TEXT NOT NULL, + status TEXT NOT NULL, + first_seen REAL NOT NULL, + last_seen REAL NOT NULL, + cve_count INTEGER NOT NULL, + PRIMARY KEY (agent_id, package, status) + ) + """) + except Exception: + conn.close() # don't leak the fd if init fails + raise + return conn + + +def claim_or_suppress(conn, agent_id, package, status): + """Atomically decide whether THIS process should send the email. + + Runs inside a single IMMEDIATE transaction, so concurrent processes for the + same key serialize on the write lock and exactly one wins at each window + boundary. Returns (should_email, cve_count). No SMTP happens under the lock. + """ + now = time.time() + conn.execute("BEGIN IMMEDIATE") + try: + row = conn.execute( + "SELECT first_seen, cve_count FROM dedup " + "WHERE agent_id=? AND package=? AND status=?", + (agent_id, package, status)).fetchone() + + if row is None or now - row[0] >= DEDUP_TTL_SECONDS: + # New key, or its window has fully elapsed. Notify and anchor a + # fresh window at now. + conn.execute( + "INSERT INTO dedup (agent_id, package, status, first_seen, " + "last_seen, cve_count) VALUES (?,?,?,?,?,1) " + "ON CONFLICT(agent_id, package, status) DO UPDATE SET " + "first_seen=excluded.first_seen, last_seen=excluded.last_seen, " + "cve_count=1", + (agent_id, package, status, now, now)) + conn.execute("COMMIT") + return True, 1 + + # Inside the window: count it, no email, and deliberately do not touch + # first_seen. This is what stops a flood of alerts from pushing the + # window forward and silencing the package indefinitely. + conn.execute( + "UPDATE dedup SET last_seen=?, cve_count=cve_count+1 " + "WHERE agent_id=? AND package=? AND status=?", + (now, agent_id, package, status)) + conn.execute("COMMIT") + return False, row[1] + 1 + except Exception: + conn.execute("ROLLBACK") + raise + + +def prune_old_rows(conn, vacuum=False): + """Housekeeping only: drop rows whose window has fully elapsed, so packages + that went quiet stop occupying space. Correctness does not depend on this, + since the read path already re-notifies past an expired window, so this is + safe to run on any schedule or to miss entirely.""" + conn.execute("BEGIN IMMEDIATE") + cur = conn.execute("DELETE FROM dedup WHERE first_seen < ?", + (time.time() - DEDUP_TTL_SECONDS,)) + deleted = cur.rowcount + conn.execute("COMMIT") + if vacuum: + conn.execute("VACUUM") # reclaim file space, brief exclusive lock + return deleted + + +# ----------------------------------- email ----------------------------------- +def send_email(receiver, subject, body): + # Imported lazily so the suppression path never pays this import cost. + import smtplib + from email.mime.multipart import MIMEMultipart + from email.mime.text import MIMEText + + # Strip CR/LF from header values: alert-derived fields (package, agent name) + # in the subject could otherwise raise on send or allow header injection. + def _hdr(value): + return str(value).replace('\r', ' ').replace('\n', ' ') + + msg = MIMEMultipart() + msg['From'] = _hdr(SENDER_EMAIL) + msg['To'] = _hdr(receiver) + msg['Subject'] = _hdr(subject) + msg.attach(MIMEText(body, 'plain')) + with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server: + server.send_message(msg) + + +# ---------------------------------- helpers ---------------------------------- +def dig(d, *path, default=None): + """Safe nested dict lookup: dig(alert, 'data', 'vulnerability', 'cve').""" + for k in path: + if not isinstance(d, dict) or k not in d: + return default + d = d[k] + return d + + +# ----------------------------------- main ------------------------------------ +def main(): + # Maintenance mode, invoked on a schedule by a command wodle rather than by + # integratord. Drops elapsed rows, compacts the database, and exits. + if len(sys.argv) > 1 and sys.argv[1] == 'prune': + try: + conn = open_db() + deleted = prune_old_rows(conn, vacuum=True) + conn.close() + logging.info("Prune complete: removed %d elapsed row(s).", deleted) + except Exception as e: + logging.error("Prune failed: %s", e) + sys.exit(1) # surface maintenance failures to the scheduler + sys.exit(0) + + if len(sys.argv) < 2: + logging.error("Usage: %s | prune", sys.argv[0]) + sys.exit(1) + + try: + with open(sys.argv[1]) as f: + alert = json.loads(f.read()) + except Exception as e: + logging.error("Failed to read/parse alert file: %s", e) + sys.exit(1) + + timestamp = alert.get('timestamp', '') + location = alert.get('location', '') + rule = alert.get('rule', {}) + rule_id = rule.get('id', '') + rule_level = rule.get('level', '') + description = rule.get('description', '') + agent = alert.get('agent', {}) + agent_id = agent.get('id', '') + agent_name = agent.get('name', '') + + receiver = RECEIVER_EMAIL + + vuln = dig(alert, 'data', 'vulnerability', default={}) + package = dig(vuln, 'package', 'name') + version = dig(vuln, 'package', 'version', default='') + cve = vuln.get('cve') or '' + status = vuln.get('status', 'Active') + severity = vuln.get('severity', '') + + # Non-vulnerability alert, or a vulnerability alert with no package: behave + # like the stock script and just send it. + if not package: + body = (f"Wazuh Notification.\n{timestamp}\n" + f"Received From: {location}\n" + f"Rule: {rule_id} (level {rule_level}) -> {description}\n" + f"Agent: {agent_name} ({agent_id})\n" + f"END OF NOTIFICATION") + try: + send_email(receiver, 'Alert Notification', body) + logging.info("Sent non-vuln alert rule=%s agent=%s", rule_id, agent_id) + except Exception as e: + logging.error("Failed to send email: %s", e) + sys.exit(1) # make delivery failures observable upstream + sys.exit(0) + + # Vulnerability alert: de-duplicate on (agent, package, status). + should_email, count = True, 1 + try: + conn = open_db() + except Exception as e: + # Store unavailable: fail OPEN (send) rather than silently drop alerts. + logging.error("DB open failed (%s); sending without dedup.", e) + conn = None + + if conn is not None: + try: + should_email, count = claim_or_suppress( + conn, agent_id, package, status) + except Exception as e: + logging.error("Dedup check failed (%s); sending to be safe.", e) + should_email = True + finally: + conn.close() + + if not should_email: + logging.debug("Suppressed %s for %s on %s (count=%d)", + cve, package, agent_id, count) + sys.exit(0) + + hours = DEDUP_TTL_SECONDS // 3600 + body = (f"Wazuh Vulnerability Notification.\n{timestamp}\n" + f"Agent: {agent_name} ({agent_id})\n" + f"Package: {package} {version}\n" + f"Severity: {severity}\n" + f"Status: {status}\n" + f"Triggering CVE: {cve}\n" + f"Rule: {rule_id} (level {rule_level}) -> {description}\n\n" + f"NOTE: further CVEs for this package on this agent are grouped into " + f"this notification for {hours}h from this first alert. If the " + f"package is still vulnerable after that, you get one fresh email.\n" + f"END OF NOTIFICATION") + try: + send_email(receiver, f'Vulnerability Alert: {package} on {agent_name}', + body) + logging.info("Sent vuln alert package=%s agent=%s cve=%s", + package, agent_id, cve) + except Exception as e: + logging.error("Failed to send email: %s", e) + sys.exit(1) # make delivery failures observable upstream + sys.exit(0) + + +if __name__ == '__main__': + main()