add optional batching for SMTP ban notifications

Queue ban notifications on a background worker and make tickets eligible after a configurable age or prolongation.
Send eligible tickets together when the batch count or delivery interval is reached, and flush pending tickets
before the stop notification.

Keep immediate delivery as the default and evaluate ticket values when sending. Document that age alone does not
guarantee observer completion, and cover batching, timing, prolongation, shutdown flushing, restored bans, and
delivery failures.
This commit is contained in:
CrazyMax 2026-09-16 13:40:18 +02:00
parent b5714f7747
commit 9523755d70
No known key found for this signature in database
GPG key ID: ADE44D8C9D44FBE4
4 changed files with 240 additions and 1 deletions

View file

@ -20,6 +20,7 @@ ver. 1.1.2-dev-1 (20??/??/??) - development nightly edition
`paths-arch.conf`, `paths-fedora.conf`, `paths-opensuse.conf` - nftables, `paths-freebsd.conf`, `paths-osx.conf ` - pf;
### New Features and Enhancements
* `action.d/smtp.py` - add optional SMTP batching by ticket age or prolongation, batch count and delivery interval
* `filter.d/cowrie.conf` - new filter and jail for Cowrie SSH/Telnet honeypot JSON log output (gh-4216)
* `filter.d/gitlab.conf` - `logtype` option `json` adding support for the `application_json.log` format that
superseded the deprecated `application.log`; default type `file` keeps the previous behaviour (gh-3566, gh-4230)

View file

@ -20,6 +20,9 @@
import socket
import smtplib
import email.policy
import math
import threading
import time
from email.message import EmailMessage
from email.utils import formatdate, formataddr
@ -77,7 +80,8 @@ class SMTPAction(ActionBase):
def __init__(
self, jail, name, host="localhost", ssl=False, user=None, password=None,
sendername="Fail2Ban", sender="fail2ban", dest="root", matches=None):
sendername="Fail2Ban", sender="fail2ban", dest="root", matches=None,
ban_delay=0, ban_age=2, ban_count=20):
"""Initialise action.
Parameters
@ -107,6 +111,15 @@ class SMTPAction(ActionBase):
Type of matches to be included from ban in email. Can be one
of "matches", "ipmatches" or "ipjailmatches". Default None
(see man jail.conf.5).
ban_delay : float, optional
Maximum interval in seconds between batches of eligible bans.
Default 0 sends each ban immediately without batching.
ban_age : float, optional
Minimum age in seconds before a queued ban is eligible. Default 2.
Prolongation makes a queued ban eligible immediately.
ban_count : int, optional
Send when this many eligible bans accumulate. Default 20.
Pending bans are flushed when the action stops.
"""
super(SMTPAction, self).__init__(jail, name)
@ -122,6 +135,19 @@ class SMTPAction(ActionBase):
self.toaddr = dest
self.matches = matches
self.ban_delay = float(ban_delay)
if not math.isfinite(self.ban_delay) or self.ban_delay < 0:
raise ValueError('ban_delay must be a finite, non-negative number')
self.ban_age = float(ban_age)
if not math.isfinite(self.ban_age) or self.ban_age < 0:
raise ValueError('ban_age must be a finite, non-negative number')
self.ban_count = int(ban_count)
if self.ban_count < 1 or self.ban_count != float(ban_count):
raise ValueError('ban_count must be a positive integer')
self._mail_pending = []
self._mail_condition = threading.Condition()
self._mail_stop = False
self._mail_thread = None
self.message_values = CallingMap(
jailname = self._jail.name,
@ -211,6 +237,13 @@ class SMTPAction(ActionBase):
def stop(self):
"""Sends email to recipients informing that the jail has stopped.
"""
if self._mail_thread is not None:
with self._mail_condition:
self._mail_stop = True
self._mail_condition.notify()
self._mail_thread.join()
self._mail_thread = None
self._mail_stop = False
self._sendMessage(
"[Fail2Ban] %(jailname)s: stopped on %(hostname)s" %
self.message_values,
@ -227,6 +260,69 @@ class SMTPAction(ActionBase):
"""
if aInfo.get('restored'):
return
if self.ban_delay:
with self._mail_condition:
# Keep the live ticket and discard values cached by earlier actions.
self._mail_pending.append((time.monotonic() + self.ban_age, aInfo.copy()))
if self._mail_thread is None:
self._mail_thread = threading.Thread(target=self._sendPendingBans)
self._mail_thread.daemon = True
self._mail_thread.start()
self._mail_condition.notify()
return
self._sendBan(aInfo)
@property
def _prolongable(self):
return bool(self.ban_delay)
def prolong(self, aInfo):
"""Make pending notifications for the prolonged ban eligible."""
with self._mail_condition:
self._mail_pending = [
(0, aInfo.copy()) if info['ip'] == aInfo['ip'] else (age, info)
for age, info in self._mail_pending]
self._mail_condition.notify()
def _sendPendingBans(self):
last = time.monotonic()
while True:
with self._mail_condition:
while True:
now = time.monotonic()
eligible = [info for age, info in self._mail_pending
if self._mail_stop or age <= now]
if eligible and (self._mail_stop or len(eligible) >= self.ban_count
or now >= last + self.ban_delay):
self._mail_pending = [(age, info) for age, info in self._mail_pending
if not self._mail_stop and age > now]
break
if self._mail_stop:
return
deadlines = [age for age, info in self._mail_pending if age > now]
if eligible:
deadlines.append(last + self.ban_delay)
self._mail_condition.wait(min(deadlines) - now if deadlines else None)
try:
self._sendBans(eligible)
except Exception:
self._logSys.exception('Failed to send batched ban notification')
last = time.monotonic()
def _sendBans(self, infos):
if len(infos) == 1:
self._sendBan(infos[0])
return
parts = []
for info in infos:
info.update(self.message_values)
parts.append((messages['ban']['head'] + messages['ban'].get(self.matches, '')) % info)
self._sendMessage(
'[Fail2Ban] %s: banned %i IPs from %s' %
(self.message_values['jailname'], len(infos), self.message_values['hostname']),
'\n'.join(parts) + messages['ban']['tail'])
def _sendBan(self, aInfo):
aInfo.update(self.message_values)
message = "".join([
messages['ban']['head'],

View file

@ -19,11 +19,13 @@
import os
import threading
import time
import unittest
import re
import sys
import types
import importlib
from unittest.mock import patch
from ..dummyjail import DummyJail
from ..utils import CONFIG_DIR, asyncserver, Utils, uni_decode
@ -114,6 +116,136 @@ class _SMTPActionTestCase():
def testNOPByRestored(self):
self._testBan(restored=True)
def testDelayedBan(self):
self.action.ban_delay = 0.01
self.action.ban_age = 0.01
ticket = BanTicket('127.0.0.2')
ticket.setBanTime(3600)
try:
self._exec_and_wait(lambda: self.action.ban(Actions.ActionInfo(ticket, self.jail)))
self.assertIn('banned for 3600 seconds', self.smtpd.data)
finally:
self.action.stop()
def testDelayedBanFlush(self):
self.action.ban_delay = 60
with patch.object(self.action, '_sendMessage') as send:
try:
for bantime in (3600, -1):
ticket = BanTicket('127.0.0.2')
aInfo = Actions.ActionInfo(ticket, self.jail)
# Simulate an earlier action caching the initial duration.
self.assertEqual(aInfo['bantime'], self.jail.actions.getBanTime())
self.action.ban(aInfo)
ticket.setBanTime(bantime)
# The original info may be reset/reused by the action executor.
aInfo.reset()
send.assert_not_called()
worker = self.action._mail_thread
finally:
self.action.stop()
self.assertFalse(worker.is_alive())
self.assertEqual(send.call_count, 2)
self.assertIn('banned for 3600 seconds', send.call_args_list[0][0][1])
self.assertIn('banned for -1 seconds', send.call_args_list[0][0][1])
self.assertIn('stopped', send.call_args_list[1][0][0])
def testDelayedBanFailure(self):
self.action.ban_delay = 60
self.action.ban_age = 0
self.action.ban_count = 1
with patch.object(self.action, '_sendBans', side_effect=[RuntimeError('SMTP unavailable'), None]) as send:
with patch.object(self.action._logSys, 'exception') as log:
with patch.object(self.action, '_sendMessage'):
try:
self.action.ban({'ip': '127.0.0.2'})
Utils.wait_for(lambda: log.call_count == 1, 3)
self.action.ban({'ip': '127.0.0.3'})
Utils.wait_for(lambda: send.call_count == 2, 3)
finally:
self.action.stop()
self.assertEqual(send.call_count, 2)
log.assert_called_once()
def testBatchProlongation(self):
self.action.ban_delay = 60
self.action.ban_age = 60
self.action.ban_count = 2
self.action.matches = 'matches'
with patch.object(self.action, '_sendMessage') as send:
try:
infos = [Actions.ActionInfo(BanTicket(ip), self.jail)
for ip in ('127.0.0.2', '127.0.0.3', '127.0.0.4')]
for info in infos:
self.action.ban(info)
send.assert_not_called()
for info in infos[:2]:
self.action.prolong(info)
Utils.wait_for(lambda: send.call_count == 1, 3)
self.assertEqual(send.call_count, 1)
self.assertIn('banned 2 IPs', send.call_args[0][0])
self.assertIn('127.0.0.2', send.call_args[0][1])
self.assertIn('127.0.0.3', send.call_args[0][1])
self.assertNotIn('127.0.0.4', send.call_args[0][1])
finally:
self.action.stop()
self.assertEqual(send.call_count, 3)
self.assertIn('127.0.0.4', send.call_args_list[1][0][1])
def testBatchAge(self):
self.action.ban_delay = 0.01
self.action.ban_age = 0.1
self.action.ban_count = 1
sent = threading.Event()
times = []
def record(*args):
times.append(time.monotonic())
sent.set()
with patch.object(self.action, '_sendMessage', side_effect=record):
try:
started = time.monotonic()
self.action.ban(Actions.ActionInfo(BanTicket('127.0.0.2'), self.jail))
self.assertTrue(sent.wait(3))
self.assertGreaterEqual(times[0] - started, self.action.ban_age)
finally:
self.action.stop()
def testBatchLatency(self):
self.action.ban_delay = 0.1
self.action.ban_age = 0
sent = threading.Event()
times = []
def record(*args):
times.append(time.monotonic())
sent.set()
with patch.object(self.action, '_sendMessage', side_effect=record):
try:
started = time.monotonic()
self.action.ban(Actions.ActionInfo(BanTicket('127.0.0.2'), self.jail))
self.assertTrue(sent.wait(3))
self.assertGreaterEqual(times[0] - started, self.action.ban_delay)
finally:
self.action.stop()
def testDelayedRestoredBan(self):
self.action.ban_delay = 60
self._testBan(restored=True)
self.assertIsNone(self.action._mail_thread)
self.assertTrue(not self.action._mail_pending)
def testBanDelayOption(self):
for delay in ('-1', 'nan', 'inf', 'invalid'):
with self.subTest(delay=delay):
with self.assertRaises(ValueError):
type(self.action)(self.jail, 'test', ban_delay=delay)
self.assertEqual(type(self.action)(self.jail, 'test', ban_delay='10').ban_delay, 10)
for option, values in (('ban_age', ('-1', 'nan', 'inf', 'invalid')),
('ban_count', ('0', '-1', '1.5', 'invalid'))):
for value in values:
with self.subTest(option=option, value=value):
with self.assertRaises(ValueError):
type(self.action)(self.jail, 'test', **{option: value})
def testOptions(self):
self._exec_and_wait(self.action.start)
self.assertEqual(self.smtpd.mailfrom, "fail2ban")

View file

@ -371,6 +371,16 @@ action = smtp.py[dest=chris@example.com, actname=smtp-chris]
smtp.py[dest=sally@example.com, actname=smtp-sally]
.fi
The Python SMTP action accepts \fBban_delay\fR to enable batching. The default is zero, which sends each ban immediately. For example:
.PP
.nf
action = smtp.py[host=localhost:25, dest=admin@example.com, ban_delay=10, ban_age=2, ban_count=20]
.fi
.PP
With batching enabled, a ban becomes eligible after \fBban_age\fR seconds (default 2), or when its prolongation is reported. A background worker sends eligible bans together when \fBban_count\fR (default 20) is reached or \fBban_delay\fR seconds have elapsed since the previous batch. The first interval starts when the worker starts. Young tickets remain queued even when the interval expires. Age and delay must be finite, non-negative seconds; count must be a positive integer.
.PP
Ticket values are evaluated when sending. Age alone does not guarantee that the observer has finished increasing the ban duration. Pending bans are flushed together before the stop notification. Delivery failures are logged and are not retried.
.SH "TIME ABBREVIATION FORMAT"
The time entries in fail2ban configuration (like \fBfindtime\fR or \fBbantime\fR) can be provided as integer in seconds or as string using special abbreviation format (e. g. \fB600\fR is the same as \fB10m\fR).