From 5b4f73d4fdfae1e35e81ed106e7f73dc0f917d02 Mon Sep 17 00:00:00 2001 From: leonfullxr Date: Sat, 18 Jul 2026 18:33:34 +0200 Subject: [PATCH 1/5] Add vulnerability email de-duplication integration The Vulnerability Detection module emits one alert per (package, CVE) pair. A single package often carries dozens of CVEs, so one scan produces a burst of near-identical alerts for the same package on the same agent, which the stock email integration turns into hundreds of near-identical messages. This integration collapses the burst: the first alert for a given (agent, package, status) sends one email, and every subsequent CVE for that same key inside the suppression window is counted but not mailed. Non-vulnerability alerts pass through unchanged. State is kept in a SQLite store in WAL mode, with the check-and-write done in a single BEGIN IMMEDIATE transaction so the ~1000 concurrent integratord processes of a burst serialize correctly without lost updates or lock errors. Suppressed alerts exit on one indexed lookup and never open an SMTP connection. The suppression window is anchored at the first notification rather than the last alert seen, so a steady drip of CVEs cannot postpone the window indefinitely and silence a package. Cleanup is an explicit prune mode driven by a command wodle, keeping maintenance off the alert hot path. --- .../vulnerability_email_dedup/README.md | 384 ++++++++++++++++++ .../vulnerability_email_dedup/custom-email | 38 ++ .../vulnerability_email_dedup/custom-email.py | 286 +++++++++++++ 3 files changed, 708 insertions(+) create mode 100644 integrations/vulnerability_email_dedup/README.md create mode 100644 integrations/vulnerability_email_dedup/custom-email create mode 100644 integrations/vulnerability_email_dedup/custom-email.py diff --git a/integrations/vulnerability_email_dedup/README.md b/integrations/vulnerability_email_dedup/README.md new file mode 100644 index 00000000..4b278008 --- /dev/null +++ b/integrations/vulnerability_email_dedup/README.md @@ -0,0 +1,384 @@ +# 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. +- Every subsequent CVE for that same agent and package inside the suppression + window is recorded and counted, but no email is sent. +- Non-vulnerability alerts are passed through and emailed unchanged, so the + script can also serve as the general email path. +- The window expires (24 hours by default), so a package that is still + vulnerable the next day notifies again and recurring problems stay visible. + +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 + +`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. Two properties make that safe and fast: + +- **Cheap lookup.** Suppressed alerts hit one indexed SQLite lookup and exit in + under a millisecond, without ever opening an SMTP connection. Only the winning + alert pays the cost of 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, so a burst produces no `database is locked` errors, no lost + updates, and no duplicate emails. + +The de-duplication key is `(agent_id, package_name, status)`. The package +**version** is stored for the email body but 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. + +The suppression window is anchored at the **first** notification rather than at +the last alert seen. This matters: anchoring on the last alert would let a +steady drip of CVEs slide the window forward indefinitely and silence a package +forever. + +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{Key seen inside window?} + E -- No --> F[Insert or reset row, COMMIT] + F --> G[Send one email] + E -- Yes --> H[Increment cve_count, append CVE, 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, which is how the test harness +redirects mail and storage into a sandbox without editing the 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) | Suppression window per agent, package and status. | +| `RETENTION_SECONDS` | `VULN_RETENTION` | `604800` (7d) | Age at which `prune` deletes expired rows. | + +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 expired more than +`RETENTION_SECONDS` ago and runs a `VACUUM` to reclaim file space. + +Keep this in perspective: the table is bounded by the number of **distinct** +`(agent, package, status)` combinations, not by alert volume. One thousand CVEs +for a single package is still one row. This is housekeeping, not a growth +problem, and 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 FROM dedup;" +``` + +Expected: + +``` +001|openssl|Active|2 +``` + +#### 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: Maintenance mode + +```bash +sudo -u wazuh /var/ossec/integrations/custom-email prune +tail -n 1 /var/ossec/logs/custom-email_integration.log +``` + +Expected: + +``` +2026-07-18T10:05:00 INFO Prune complete: removed 0 expired row(s). +``` + +--- + +### 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. Lower `DEDUP_TTL_SECONDS` if you want more frequent reminders. | +| 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 against a synthetic + 2,000-alert burst at 64-way concurrency: 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 00000000..fc3023be --- /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 00000000..09226310 --- /dev/null +++ b/integrations/vulnerability_email_dedup/custom-email.py @@ -0,0 +1,286 @@ +#!/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: the FIRST + alert seen for a given (agent, package, status) within a time window sends + one email; every other CVE for that same package and agent inside the window + is counted and suppressed. + +EXECUTION MODEL + wazuh-integratord runs this script ONCE PER ALERT, as a separate OS process. + A "batch of 1000 alerts" is therefore 1000 short-lived processes, not one + process handling 1000 items, so asyncio would not help here. What keeps it + fast and correct under a burst is: + - a cheap, indexed lookup so duplicates exit in under a millisecond + without ever opening an SMTP connection, and + - a SQLite store in WAL mode with a busy timeout, so many processes can + check-and-write concurrently without "database is locked" or lost + updates. + +DE-DUPLICATION KEY + (agent_id, package_name, status). Package VERSION is stored for the email + body but deliberately left 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') + +# How long one (agent, package, status) stays "already notified". The window is +# anchored at the FIRST notification, not at the last alert seen, so a package +# that keeps re-detecting cannot slide the window forward indefinitely and +# silence itself. After the window closes, the next detection emails again. +DEDUP_TTL_SECONDS = int(os.environ.get('VULN_DEDUP_TTL', 24 * 60 * 60)) + +# Rows older than this are removed by the scheduled maintenance run ("prune" +# mode, invoked by a command wodle). Kept generously long so a package that +# keeps re-detecting stays tracked across scans. +RETENTION_SECONDS = int(os.environ.get('VULN_RETENTION', 7 * 24 * 60 * 60)) + +BUSY_TIMEOUT_MS = 10000 # wait, do not error, on lock +MAX_CVES_STORED = 50 # cap the CVE list per row + +logging.basicConfig(filename=LOG_PATH, filemode='a', + format='%(asctime)s %(levelname)s %(message)s', + datefmt='%Y-%m-%dT%H:%M:%S', + level=logging.INFO) # DEBUG would spam under bursts + + +# --------------------------------- data store -------------------------------- +def open_db(): + conn = sqlite3.connect(DB_PATH, timeout=BUSY_TIMEOUT_MS / 1000.0) + 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, + cves TEXT NOT NULL, + PRIMARY KEY (agent_id, package, status) + ) + """) + return conn + + +def claim_or_suppress(conn, agent_id, package, status, cve): + """Atomically decide whether THIS process should send the email. + + Runs inside a single IMMEDIATE transaction, so concurrent processes for the + same package serialize on the write lock and exactly one of them wins. + Returns (should_email, cve_count, cves_list). No SMTP happens under the lock. + """ + now = time.time() + conn.execute("BEGIN IMMEDIATE") + try: + row = conn.execute( + "SELECT first_seen, last_seen, cve_count, cves " + "FROM dedup WHERE agent_id=? AND package=? AND status=?", + (agent_id, package, status)).fetchone() + + # row[0] is first_seen: the window is measured from when we notified, + # so a steady drip of CVEs cannot keep the window open forever. + if row is None or (now - row[0]) > DEDUP_TTL_SECONDS: + # New key, or the previous window expired. Notify and (re)open it. + cves = [cve] if cve else [] + conn.execute( + "INSERT INTO dedup (agent_id, package, status, first_seen, " + "last_seen, cve_count, cves) VALUES (?,?,?,?,?,?,?) " + "ON CONFLICT(agent_id, package, status) DO UPDATE SET " + "first_seen=excluded.first_seen, last_seen=excluded.last_seen, " + "cve_count=1, cves=excluded.cves", + (agent_id, package, status, now, now, 1, json.dumps(cves))) + conn.execute("COMMIT") + return True, 1, cves + + # Duplicate inside the window. Count it, remember the CVE, no email. + cves = json.loads(row[3]) + if cve and cve not in cves and len(cves) < MAX_CVES_STORED: + cves.append(cve) + conn.execute( + "UPDATE dedup SET last_seen=?, cve_count=cve_count+1, cves=? " + "WHERE agent_id=? AND package=? AND status=?", + (now, json.dumps(cves), agent_id, package, status)) + conn.execute("COMMIT") + return False, row[2] + 1, cves + except Exception: + conn.execute("ROLLBACK") + raise + + +def prune_old_rows(conn, vacuum=False): + """Delete rows whose window expired long ago. Called only from the scheduled + maintenance entry point, never on the alert hot path.""" + conn.execute("BEGIN IMMEDIATE") + cur = conn.execute("DELETE FROM dedup WHERE last_seen < ?", + (time.time() - RETENTION_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 + + msg = MIMEMultipart() + msg['From'] = SENDER_EMAIL + msg['To'] = receiver + msg['Subject'] = 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. Prunes expired 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 expired row(s).", deleted) + except Exception as e: + logging.error("Prune failed: %s", e) + sys.exit(0) + + 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(0) + + # Vulnerability alert: de-duplicate on (agent, package, status). + should_email, count, cves = True, 1, ([cve] if cve else []) + 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, cves = claim_or_suppress( + conn, agent_id, package, status, cve) + 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) + + 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: additional CVEs for this package on this agent will be " + f"suppressed for the next {DEDUP_TTL_SECONDS // 3600}h to reduce " + f"noise.\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(0) + + +if __name__ == '__main__': + main() From 2b3d390075ac6ce37a417a88f0afd9a0655d324c Mon Sep 17 00:00:00 2001 From: leonfullxr Date: Tue, 21 Jul 2026 10:35:12 +0200 Subject: [PATCH 2/5] Address Copilot review: logging fail-open + failure exit codes - Fall back to stderr logging when LOG_PATH is not writable, so a logging problem cannot stop the integration from sending mail. - Exit non-zero on prune failure and on email send failure, so failures are visible to the scheduler / integrator instead of appearing successful. - Clarify (docstring + README) that the package version is read from the triggering alert for the email body and is not persisted. --- .../vulnerability_email_dedup/README.md | 7 +++--- .../vulnerability_email_dedup/custom-email.py | 25 +++++++++++++------ 2 files changed, 21 insertions(+), 11 deletions(-) diff --git a/integrations/vulnerability_email_dedup/README.md b/integrations/vulnerability_email_dedup/README.md index 4b278008..c844705a 100644 --- a/integrations/vulnerability_email_dedup/README.md +++ b/integrations/vulnerability_email_dedup/README.md @@ -78,9 +78,10 @@ not one process handling 1,000 items. Two properties make that safe and fast: updates, and no duplicate emails. The de-duplication key is `(agent_id, package_name, status)`. The package -**version** is stored for the email body but 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. +**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. The suppression window is anchored at the **first** notification rather than at the last alert seen. This matters: anchoring on the last alert would let a diff --git a/integrations/vulnerability_email_dedup/custom-email.py b/integrations/vulnerability_email_dedup/custom-email.py index 09226310..1b8f53f1 100644 --- a/integrations/vulnerability_email_dedup/custom-email.py +++ b/integrations/vulnerability_email_dedup/custom-email.py @@ -24,10 +24,10 @@ updates. DE-DUPLICATION KEY - (agent_id, package_name, status). Package VERSION is stored for the email - body but deliberately left 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. + (agent_id, package_name, status). Package VERSION is read from the + triggering alert for the email body and is deliberately NOT persisted or + part 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: @@ -70,10 +70,16 @@ BUSY_TIMEOUT_MS = 10000 # wait, do not error, on lock MAX_CVES_STORED = 50 # cap the CVE list per row -logging.basicConfig(filename=LOG_PATH, filemode='a', - format='%(asctime)s %(levelname)s %(message)s', - datefmt='%Y-%m-%dT%H:%M:%S', - level=logging.INFO) # DEBUG would spam under bursts +_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 -------------------------------- @@ -194,6 +200,7 @@ def main(): logging.info("Prune complete: removed %d expired 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) try: @@ -235,6 +242,7 @@ def main(): 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). @@ -279,6 +287,7 @@ def main(): 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) From 4c31ace28a02bf682edb99a68374aa874c0a7777 Mon Sep 17 00:00:00 2001 From: leonfullxr Date: Tue, 21 Jul 2026 11:17:26 +0200 Subject: [PATCH 3/5] Address Copilot review: header sanitization, usage guard, DB cleanup - Strip CR/LF from email header values (the subject is built from alert-derived package/agent names) to prevent header injection and send-time crashes. - Close the SQLite connection if a PRAGMA/DDL fails during open_db, so a failed init does not leak the handle. - Add an explicit no-args usage check for a clearer log message (the IndexError was already caught by the existing handler; this only improves diagnostics). --- .../vulnerability_email_dedup/custom-email.py | 51 ++++++++++++------- 1 file changed, 32 insertions(+), 19 deletions(-) diff --git a/integrations/vulnerability_email_dedup/custom-email.py b/integrations/vulnerability_email_dedup/custom-email.py index 1b8f53f1..d27f4542 100644 --- a/integrations/vulnerability_email_dedup/custom-email.py +++ b/integrations/vulnerability_email_dedup/custom-email.py @@ -85,22 +85,26 @@ # --------------------------------- data store -------------------------------- def open_db(): conn = sqlite3.connect(DB_PATH, timeout=BUSY_TIMEOUT_MS / 1000.0) - 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, - cves TEXT NOT NULL, - PRIMARY KEY (agent_id, package, status) - ) - """) + 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, + cves TEXT NOT NULL, + PRIMARY KEY (agent_id, package, status) + ) + """) + except Exception: + conn.close() # don't leak the fd if init fails + raise return conn @@ -169,10 +173,15 @@ def send_email(receiver, subject, body): 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'] = SENDER_EMAIL - msg['To'] = receiver - msg['Subject'] = subject + 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) @@ -203,6 +212,10 @@ def main(): 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()) From 1cb29eabded85001d1c1bf25137c09daa806a2e8 Mon Sep 17 00:00:00 2001 From: leonfullxr Date: Tue, 28 Jul 2026 10:52:12 +0200 Subject: [PATCH 4/5] Anchor dedup window at first alert and make prune non-critical The suppression window is now decided entirely from first_seen at read time: no row or a fully elapsed row notifies and (re-)anchors the window at now, while an alert inside the window is counted without touching the anchor. A flood of alerts can no longer slide the window forward and silence a package. Because the read path already re-notifies past an elapsed window, prune is pure housekeeping. It drops rows on the same window and needs no particular schedule, so RETENTION_SECONDS goes away and DEDUP_TTL_SECONDS is the only knob left. Also drop the per-row CVE list. It cost a JSON parse and dump on the suppression hot path with no consumer, and the CVEs are already in alerts.json and the indexer. Add test_dedup.py, covering one email per key under a 1000-alert burst, no window slide while alerts keep arriving, a fresh email plus re-anchor once the window elapses, and prune taking only elapsed rows. --- .../vulnerability_email_dedup/README.md | 154 ++++++++++++------ .../vulnerability_email_dedup/custom-email.py | 119 +++++++------- .../vulnerability_email_dedup/test_dedup.py | 79 +++++++++ 3 files changed, 241 insertions(+), 111 deletions(-) create mode 100644 integrations/vulnerability_email_dedup/test_dedup.py diff --git a/integrations/vulnerability_email_dedup/README.md b/integrations/vulnerability_email_dedup/README.md index c844705a..a2c07a4f 100644 --- a/integrations/vulnerability_email_dedup/README.md +++ b/integrations/vulnerability_email_dedup/README.md @@ -29,13 +29,17 @@ 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. -- Every subsequent CVE for that same agent and package inside the suppression - window is recorded and counted, but no email is sent. +- 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. -- The window expires (24 hours by default), so a package that is still - vulnerable the next day notifies again and recurring problems stay visible. 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 @@ -64,29 +68,41 @@ concurrent writers. ### 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. Two properties make that safe and fast: +not one process handling 1,000 items. Three properties make that safe and cheap: -- **Cheap lookup.** Suppressed alerts hit one indexed SQLite lookup and exit in - under a millisecond, without ever opening an SMTP connection. Only the winning - alert pays the cost of sending mail. +- **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, so a burst produces no `database is locked` errors, no lost - updates, and no duplicate emails. - -The de-duplication key is `(agent_id, package_name, status)`. 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. - -The suppression window is anchored at the **first** notification rather than at -the last alert seen. This matters: anchoring on the last alert would let a -steady drip of CVEs slide the window forward indefinitely and silence a package -forever. + 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 @@ -97,10 +113,10 @@ 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{Key seen inside window?} - E -- No --> F[Insert or reset row, COMMIT] + 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, append CVE, COMMIT] + E -- Yes --> H[Increment cve_count,
first_seen unchanged, COMMIT] H --> I[Exit, no email] ``` @@ -110,14 +126,15 @@ graph TD #### Using the Integration Files -This integration ships two files: +This integration ships three files: | File | Purpose | | --- | --- | | `custom-email.py` | The integration logic. | | `custom-email` | Standard Wazuh shell wrapper that invokes the script with the embedded Python interpreter. | +| `test_dedup.py` | Self-check for the window logic. Not deployed to the manager. | -Copy both to the manager: +Copy the two runtime files to the manager: ```bash cp custom-email custom-email.py /var/ossec/integrations/ @@ -146,8 +163,11 @@ redirects mail and storage into a sandbox without editing the file. | `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) | Suppression window per agent, package and status. | -| `RETENTION_SECONDS` | `VULN_RETENTION` | `604800` (7d) | Age at which `prune` deletes expired rows. | +| `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 @@ -186,13 +206,21 @@ Notes on this block: #### Scheduled Maintenance Cleanup is an explicit maintenance mode rather than opportunistic work on the -alert hot path. `custom-email prune` deletes rows whose window expired more than -`RETENTION_SECONDS` ago and runs a `VACUUM` to reclaim file space. - -Keep this in perspective: the table is bounded by the number of **distinct** -`(agent, package, status)` combinations, not by alert volume. One thousand CVEs -for a single package is still one row. This is housekeeping, not a growth -problem, and once a day is ample. +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`: @@ -263,7 +291,29 @@ seen package or records the CVE and exits silently. ### Integration Testing -#### Test 1: First alert for a package sends an email +#### Test 1: Window logic self-check + +Run the bundled self-check from the integration directory. It exercises the +window rules against a temporary database, sends no mail, and needs nothing +outside the standard library: + +```bash +python3 test_dedup.py +``` + +Expected: + +``` +OK: all de-duplication window checks passed +``` + +It asserts the behaviours that would otherwise take a day of wall-clock time to +observe: that a burst of 1,000 alerts for one package produces exactly one +email with no lost counts, that a flood arriving inside the window neither +notifies nor moves the anchor, that the next alert after the window elapses +notifies and re-anchors, and that prune removes only fully elapsed rows. + +#### Test 2: First alert for a package sends an email Write a sample alert and invoke the script directly: @@ -293,7 +343,7 @@ 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 +#### Test 3: 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 @@ -305,16 +355,19 @@ 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 FROM dedup;" + "SELECT agent_id, package, status, cve_count, + CAST((strftime('%s','now') - first_seen) / 3600 AS INT) AS window_age_h + FROM dedup;" ``` -Expected: +Expected, with the anchor still showing zero hours of age because the second +alert did not move it: ``` -001|openssl|Active|2 +001|openssl|Active|2|0 ``` -#### Test 3: Burst behaviour and concurrency +#### Test 4: 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 @@ -330,17 +383,17 @@ 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: Maintenance mode +#### 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: +Expected, since the rows just created are still inside their window: ``` -2026-07-18T10:05:00 INFO Prune complete: removed 0 expired row(s). +2026-07-18T10:05:00 INFO Prune complete: removed 0 elapsed row(s). ``` --- @@ -353,7 +406,8 @@ Expected: | `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. Lower `DEDUP_TTL_SECONDS` if you want more frequent reminders. | +| 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`. | --- @@ -366,10 +420,10 @@ Expected: - **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 against a synthetic - 2,000-alert burst at 64-way concurrency: 300 distinct keys produced 300 - emails, `SUM(cve_count)` matched the 2,000 alerts fed in, and no lock errors - occurred. + third-party dependencies. Logic validated on Python 3.11 by the bundled + `test_dedup.py`, plus a synthetic 2,000-alert burst at 64-way concurrency: 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. diff --git a/integrations/vulnerability_email_dedup/custom-email.py b/integrations/vulnerability_email_dedup/custom-email.py index d27f4542..5d1833d7 100644 --- a/integrations/vulnerability_email_dedup/custom-email.py +++ b/integrations/vulnerability_email_dedup/custom-email.py @@ -7,27 +7,32 @@ 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: the FIRST - alert seen for a given (agent, package, status) within a time window sends - one email; every other CVE for that same package and agent inside the window - is counted and suppressed. + 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 "batch of 1000 alerts" is therefore 1000 short-lived processes, not one - process handling 1000 items, so asyncio would not help here. What keeps it - fast and correct under a burst is: - - a cheap, indexed lookup so duplicates exit in under a millisecond - without ever opening an SMTP connection, and - - a SQLite store in WAL mode with a busy timeout, so many processes can - check-and-write concurrently without "database is locked" or lost - updates. + 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 is deliberately NOT persisted or - part 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. + (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: @@ -56,19 +61,12 @@ LOG_PATH = os.environ.get('VULN_DEDUP_LOG', '/var/ossec/logs/custom-email_integration.log') -# How long one (agent, package, status) stays "already notified". The window is -# anchored at the FIRST notification, not at the last alert seen, so a package -# that keeps re-detecting cannot slide the window forward indefinitely and -# silence itself. After the window closes, the next detection emails again. +# 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)) -# Rows older than this are removed by the scheduled maintenance run ("prune" -# mode, invoked by a command wodle). Kept generously long so a package that -# keeps re-detecting stays tracked across scans. -RETENTION_SECONDS = int(os.environ.get('VULN_RETENTION', 7 * 24 * 60 * 60)) - BUSY_TIMEOUT_MS = 10000 # wait, do not error, on lock -MAX_CVES_STORED = 50 # cap the CVE list per row _LOG_FORMAT = '%(asctime)s %(levelname)s %(message)s' _LOG_DATEFMT = '%Y-%m-%dT%H:%M:%S' @@ -98,7 +96,6 @@ def open_db(): first_seen REAL NOT NULL, last_seen REAL NOT NULL, cve_count INTEGER NOT NULL, - cves TEXT NOT NULL, PRIMARY KEY (agent_id, package, status) ) """) @@ -108,57 +105,56 @@ def open_db(): return conn -def claim_or_suppress(conn, agent_id, package, status, cve): +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 package serialize on the write lock and exactly one of them wins. - Returns (should_email, cve_count, cves_list). No SMTP happens under the lock. + 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, last_seen, cve_count, cves " - "FROM dedup WHERE agent_id=? AND package=? AND status=?", + "SELECT first_seen, cve_count FROM dedup " + "WHERE agent_id=? AND package=? AND status=?", (agent_id, package, status)).fetchone() - # row[0] is first_seen: the window is measured from when we notified, - # so a steady drip of CVEs cannot keep the window open forever. - if row is None or (now - row[0]) > DEDUP_TTL_SECONDS: - # New key, or the previous window expired. Notify and (re)open it. - cves = [cve] if cve else [] + 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, cves) VALUES (?,?,?,?,?,?,?) " + "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, cves=excluded.cves", - (agent_id, package, status, now, now, 1, json.dumps(cves))) + "cve_count=1", + (agent_id, package, status, now, now)) conn.execute("COMMIT") - return True, 1, cves + return True, 1 - # Duplicate inside the window. Count it, remember the CVE, no email. - cves = json.loads(row[3]) - if cve and cve not in cves and len(cves) < MAX_CVES_STORED: - cves.append(cve) + # 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, cves=? " + "UPDATE dedup SET last_seen=?, cve_count=cve_count+1 " "WHERE agent_id=? AND package=? AND status=?", - (now, json.dumps(cves), agent_id, package, status)) + (now, agent_id, package, status)) conn.execute("COMMIT") - return False, row[2] + 1, cves + return False, row[1] + 1 except Exception: conn.execute("ROLLBACK") raise def prune_old_rows(conn, vacuum=False): - """Delete rows whose window expired long ago. Called only from the scheduled - maintenance entry point, never on the alert hot path.""" + """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 last_seen < ?", - (time.time() - RETENTION_SECONDS,)) + cur = conn.execute("DELETE FROM dedup WHERE first_seen < ?", + (time.time() - DEDUP_TTL_SECONDS,)) deleted = cur.rowcount conn.execute("COMMIT") if vacuum: @@ -200,13 +196,13 @@ def dig(d, *path, default=None): # ----------------------------------- main ------------------------------------ def main(): # Maintenance mode, invoked on a schedule by a command wodle rather than by - # integratord. Prunes expired rows, compacts the database, and exits. + # 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 expired row(s).", deleted) + 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 @@ -259,7 +255,7 @@ def main(): sys.exit(0) # Vulnerability alert: de-duplicate on (agent, package, status). - should_email, count, cves = True, 1, ([cve] if cve else []) + should_email, count = True, 1 try: conn = open_db() except Exception as e: @@ -269,8 +265,8 @@ def main(): if conn is not None: try: - should_email, count, cves = claim_or_suppress( - conn, agent_id, package, status, cve) + 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 @@ -282,6 +278,7 @@ def main(): 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" @@ -289,9 +286,9 @@ def main(): f"Status: {status}\n" f"Triggering CVE: {cve}\n" f"Rule: {rule_id} (level {rule_level}) -> {description}\n\n" - f"NOTE: additional CVEs for this package on this agent will be " - f"suppressed for the next {DEDUP_TTL_SECONDS // 3600}h to reduce " - f"noise.\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}', diff --git a/integrations/vulnerability_email_dedup/test_dedup.py b/integrations/vulnerability_email_dedup/test_dedup.py new file mode 100644 index 00000000..e49bec3e --- /dev/null +++ b/integrations/vulnerability_email_dedup/test_dedup.py @@ -0,0 +1,79 @@ +#!/usr/bin/env python3 +"""Self-check for the de-duplication window. Run: python3 test_dedup.py + +Covers the three behaviours the integration exists for: one email per key, +a window that never slides while alerts keep arriving, and a fresh email once +the window has elapsed since the FIRST alert. Uses a temporary database, sends +no mail, and needs nothing but the standard library. +""" + +import os +import sys +import time +import tempfile +import importlib.util + +TMP = tempfile.mkdtemp(prefix='vuln_dedup_test_') +os.environ['VULN_DEDUP_DB'] = os.path.join(TMP, 'dedup.db') +os.environ['VULN_DEDUP_LOG'] = os.path.join(TMP, 'dedup.log') + +# The integration file is named with a hyphen, so it cannot be imported by name. +_spec = importlib.util.spec_from_file_location( + 'dedup', os.path.join(os.path.dirname(os.path.abspath(__file__)), + 'custom-email.py')) +dedup = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(dedup) + +TTL = dedup.DEDUP_TTL_SECONDS +conn = dedup.open_db() + + +def claim(package, agent='001'): + return dedup.claim_or_suppress(conn, agent, package, 'Active') + + +def anchor_of(package, agent='001'): + return conn.execute("SELECT first_seen FROM dedup WHERE agent_id=? AND " + "package=? AND status=?", + (agent, package, 'Active')).fetchone()[0] + + +def backdate(package, seconds, agent='001'): + """Age a row as if its first alert had arrived `seconds` ago.""" + conn.execute("UPDATE dedup SET first_seen=? WHERE agent_id=? AND " + "package=? AND status=?", + (time.time() - seconds, agent, package, 'Active')) + + +# First alert notifies; the burst behind it does not. +assert claim('openssl') == (True, 1) +emails = sum(1 for _ in range(999) if claim('openssl')[0]) +assert emails == 0, f"burst sent {emails} extra emails" +assert claim('openssl')[1] == 1001, "lost an update under the burst" + +# A different package on the same agent is a different key. +assert claim('nginx') == (True, 1) + +# The window must not slide. A package first seen just under the TTL ago stays +# suppressed no matter how many more alerts arrive, and its anchor holds. +backdate('openssl', TTL - 3600) +before = anchor_of('openssl') +for _ in range(50): + assert claim('openssl')[0] is False, "flood slid the window forward" +assert anchor_of('openssl') == before, "anchor moved inside the window" + +# Once the TTL has elapsed since the FIRST alert, the next one notifies and +# re-anchors, and the alert right behind it is suppressed again. +backdate('openssl', TTL + 1800) +assert claim('openssl') == (True, 1), "no fresh email after the window elapsed" +assert time.time() - anchor_of('openssl') < 5, "window did not re-anchor" +assert claim('openssl')[0] is False, "re-anchored window is not suppressing" + +# Prune is housekeeping: it removes only rows whose window has fully elapsed. +backdate('nginx', TTL + 30 * 3600) +assert dedup.prune_old_rows(conn) == 1, "prune took the wrong number of rows" +assert conn.execute("SELECT COUNT(*) FROM dedup").fetchone()[0] == 1 + +conn.close() +print("OK: all de-duplication window checks passed") +sys.exit(0) From d654e9ec0a55fc0d8f6712df9227149f81576e49 Mon Sep 17 00:00:00 2001 From: leonfullxr Date: Tue, 28 Jul 2026 11:09:39 +0200 Subject: [PATCH 5/5] Ship only the integration and its README Drop test_dedup.py from the integration folder: the deliverable is the script and its documentation. The window rules it asserted are now a documented manual procedure instead, ageing first_seen with sqlite3 so both the no-slide and the rollover behaviour can be confirmed in seconds rather than over a day. Also round the window_age_h helper query, which truncated a 23 hour old anchor to 22. --- .../vulnerability_email_dedup/README.md | 94 ++++++++++++------- .../vulnerability_email_dedup/test_dedup.py | 79 ---------------- 2 files changed, 59 insertions(+), 114 deletions(-) delete mode 100644 integrations/vulnerability_email_dedup/test_dedup.py diff --git a/integrations/vulnerability_email_dedup/README.md b/integrations/vulnerability_email_dedup/README.md index a2c07a4f..4dc4f0cd 100644 --- a/integrations/vulnerability_email_dedup/README.md +++ b/integrations/vulnerability_email_dedup/README.md @@ -126,15 +126,14 @@ graph TD #### Using the Integration Files -This integration ships three 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. | -| `test_dedup.py` | Self-check for the window logic. Not deployed to the manager. | -Copy the two runtime files to the manager: +Copy both to the manager: ```bash cp custom-email custom-email.py /var/ossec/integrations/ @@ -152,8 +151,8 @@ 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, which is how the test harness -redirects mail and storage into a sandbox without editing the file. +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 | | --- | --- | --- | --- | @@ -291,29 +290,7 @@ seen package or records the CVE and exits silently. ### Integration Testing -#### Test 1: Window logic self-check - -Run the bundled self-check from the integration directory. It exercises the -window rules against a temporary database, sends no mail, and needs nothing -outside the standard library: - -```bash -python3 test_dedup.py -``` - -Expected: - -``` -OK: all de-duplication window checks passed -``` - -It asserts the behaviours that would otherwise take a day of wall-clock time to -observe: that a burst of 1,000 alerts for one package produces exactly one -email with no lost counts, that a flood arriving inside the window neither -notifies nor moves the anchor, that the next alert after the window elapses -notifies and re-anchors, and that prune removes only fully elapsed rows. - -#### Test 2: First alert for a package sends an email +#### Test 1: First alert for a package sends an email Write a sample alert and invoke the script directly: @@ -343,7 +320,7 @@ Expected: 2026-07-18T10:00:01 INFO Sent vuln alert package=openssl agent=001 cve=CVE-2026-0001 ``` -#### Test 3: A second CVE for the same package is suppressed +#### 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 @@ -356,7 +333,7 @@ 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((strftime('%s','now') - first_seen) / 3600 AS INT) AS window_age_h + CAST(ROUND((strftime('%s','now') - first_seen) / 3600.0) AS INT) AS window_age_h FROM dedup;" ``` @@ -367,7 +344,7 @@ alert did not move it: 001|openssl|Active|2|0 ``` -#### Test 4: Burst behaviour and concurrency +#### 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 @@ -383,6 +360,48 @@ 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 @@ -390,12 +409,17 @@ sudo -u wazuh /var/ossec/integrations/custom-email prune tail -n 1 /var/ossec/logs/custom-email_integration.log ``` -Expected, since the rows just created are still inside their window: +Expected, since the row was just re-anchored and is inside its window: ``` -2026-07-18T10:05:00 INFO Prune complete: removed 0 elapsed row(s). +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 @@ -420,8 +444,8 @@ Expected, since the rows just created are still inside their window: - **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 by the bundled - `test_dedup.py`, plus a synthetic 2,000-alert burst at 64-way concurrency: 300 + 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. diff --git a/integrations/vulnerability_email_dedup/test_dedup.py b/integrations/vulnerability_email_dedup/test_dedup.py deleted file mode 100644 index e49bec3e..00000000 --- a/integrations/vulnerability_email_dedup/test_dedup.py +++ /dev/null @@ -1,79 +0,0 @@ -#!/usr/bin/env python3 -"""Self-check for the de-duplication window. Run: python3 test_dedup.py - -Covers the three behaviours the integration exists for: one email per key, -a window that never slides while alerts keep arriving, and a fresh email once -the window has elapsed since the FIRST alert. Uses a temporary database, sends -no mail, and needs nothing but the standard library. -""" - -import os -import sys -import time -import tempfile -import importlib.util - -TMP = tempfile.mkdtemp(prefix='vuln_dedup_test_') -os.environ['VULN_DEDUP_DB'] = os.path.join(TMP, 'dedup.db') -os.environ['VULN_DEDUP_LOG'] = os.path.join(TMP, 'dedup.log') - -# The integration file is named with a hyphen, so it cannot be imported by name. -_spec = importlib.util.spec_from_file_location( - 'dedup', os.path.join(os.path.dirname(os.path.abspath(__file__)), - 'custom-email.py')) -dedup = importlib.util.module_from_spec(_spec) -_spec.loader.exec_module(dedup) - -TTL = dedup.DEDUP_TTL_SECONDS -conn = dedup.open_db() - - -def claim(package, agent='001'): - return dedup.claim_or_suppress(conn, agent, package, 'Active') - - -def anchor_of(package, agent='001'): - return conn.execute("SELECT first_seen FROM dedup WHERE agent_id=? AND " - "package=? AND status=?", - (agent, package, 'Active')).fetchone()[0] - - -def backdate(package, seconds, agent='001'): - """Age a row as if its first alert had arrived `seconds` ago.""" - conn.execute("UPDATE dedup SET first_seen=? WHERE agent_id=? AND " - "package=? AND status=?", - (time.time() - seconds, agent, package, 'Active')) - - -# First alert notifies; the burst behind it does not. -assert claim('openssl') == (True, 1) -emails = sum(1 for _ in range(999) if claim('openssl')[0]) -assert emails == 0, f"burst sent {emails} extra emails" -assert claim('openssl')[1] == 1001, "lost an update under the burst" - -# A different package on the same agent is a different key. -assert claim('nginx') == (True, 1) - -# The window must not slide. A package first seen just under the TTL ago stays -# suppressed no matter how many more alerts arrive, and its anchor holds. -backdate('openssl', TTL - 3600) -before = anchor_of('openssl') -for _ in range(50): - assert claim('openssl')[0] is False, "flood slid the window forward" -assert anchor_of('openssl') == before, "anchor moved inside the window" - -# Once the TTL has elapsed since the FIRST alert, the next one notifies and -# re-anchors, and the alert right behind it is suppressed again. -backdate('openssl', TTL + 1800) -assert claim('openssl') == (True, 1), "no fresh email after the window elapsed" -assert time.time() - anchor_of('openssl') < 5, "window did not re-anchor" -assert claim('openssl')[0] is False, "re-anchored window is not suppressing" - -# Prune is housekeeping: it removes only rows whose window has fully elapsed. -backdate('nginx', TTL + 30 * 3600) -assert dedup.prune_old_rows(conn) == 1, "prune took the wrong number of rows" -assert conn.execute("SELECT COUNT(*) FROM dedup").fetchone()[0] == 1 - -conn.close() -print("OK: all de-duplication window checks passed") -sys.exit(0)