CERBERUS/fail2ban
1
0
Fork 0
mirror of https://github.com/fail2ban/fail2ban.git synced 2026-10-21 11:31:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge 96a2f29dc8 into 6c88de71d2

Browse source
This commit is contained in:
Tobias Knecht 2026-10-12 11:37:43 +02:00 • committed by GitHub
commit 8375f69ccf
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 966 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -11,3 +11,4 @@ __pycache__
.vagrant/
.idea/
.venv/
bin/fail2ban-python

11
ChangeLog
View file

@ -18,10 +18,19 @@ ver. 1.1.2-dev-1 (20??/??/??) - development nightly edition
* `paths-*.conf`: add default banactions for several includes depending on default net-filter for distribution (gh-4224, gh-4229):
`paths-arch.conf`, `paths-fedora.conf`, `paths-opensuse.conf` - nftables, `paths-freebsd.conf`, `paths-osx.conf ` - pf;
### Deprecations
* `action.d/xarf-login-attack.conf` - deprecated (emits obsolete x-arf 0.2);
superseded by the `xarf` action (XARF v4). Will be removed in a future release (gh-3820)
### New Features and Enhancements
* `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)
* `action.d/xarf.py` - new action reporting login attacks as XARF v4
(https://xarf.org) by e-mail to the IP's abuse contact (resolved via the
Abusix Contact DB). Uses the optional `xarf` python package for schema
validation when installed, otherwise a built-in stdlib generator. Enable
via `action_xarf_v4` in jail.conf (gh-3820)
ver. 1.1.1 (2026/08/15) - triple-one-win
@ -99,7 +108,7 @@ ver. 1.1.1 (2026/08/15) - triple-one-win
several log messages will be tagged with as originating from a process named "sshd-session" rather than "sshd" (gh-3782)
- `ddos` and `aggressive` modes: regex extended for timeout before authentication (optional connection from part, gh-3907)
* `filter.d/vsftpd.conf` - fixed regex (if failures generated by systemd-journal, gh-3954)
* `filter.d/froxlor-auth.conf` - updated the regex to the new logging situation for froxlor and changed logpath in jail.conf (gh-4075).
* `filter.d/froxlor-auth.conf` - updated the regex to the new logging situation for froxlor and changed logpath in jail.conf (gh-4075).
### New Features and Enhancements
* `fail2ban-server` supports `systemd` socket activation (`sd_listen_fds(3)`); optional `fail2ban.socket` unit provided

5
config/action.d/xarf-login-attack.conf
View file

@ -30,6 +30,11 @@
#
#
# DEPRECATED: this action emits the obsolete x-arf 0.2 format. It is superseded
# by the `xarf` action (XARF v4) - see action_xarf_v4 in jail.conf. This action
# remains functional for now but will be removed in a future release. Please
# migrate to action_xarf_v4.
[Definition]
# bypass ban/unban for restored tickets

464
config/action.d/xarf.py Normal file
View file

@ -0,0 +1,464 @@
# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: t -*-
# vi: set ft=python sts=4 ts=4 sw=4 noet :
# This file is part of Fail2Ban.
#
# Fail2Ban is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# Fail2Ban is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Fail2Ban; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
"""Fail2Ban action: report login attacks as XARF v4 over email.
Resolves the offending IP's abuse contact (Abusix Contact DB), builds a
XARF v4 login_attack report, and sends it via the XARF v4 email transport
format (RFC 5965 multipart/report). Requires the ``dig`` command
(bind-utils) for contact resolution.
IMPORTANT: Reporting an IP for abuse is a serious complaint. Only use this
action where you are confident the activity is genuinely abusive, e.g.:
* recidive (the IP has been banned repeatedly),
* a high maxretry, well beyond a normal user mistyping a password,
* filters with a low likelihood of false positives.
"""
import base64
import hashlib
import json
import re
import shlex
import socket
import subprocess
import time
import uuid
from email import encoders
from email.mime.base import MIMEBase
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.utils import formatdate, make_msgid
from fail2ban.helpers import getLogger
from fail2ban.server.actions import ActionBase
# Loaded standalone (see Utils.load_python_module), so __name__ is just
# "xarf" - name explicitly under the fail2ban.* hierarchy so it reaches
# the daemon's configured log handlers (and the test log capture).
logSys = getLogger("fail2ban.xarf")
# --- XARF v4 report generator -----------------------------------------
#
# Builds the XARF v4 login_attack document from fail2ban ban data. Kept
# local to this action (its only consumer today) rather than in
# fail2ban/server/: if a second XARF-based transport needs it later, it
# can be extracted then.
#
# Uses the official ``xarf`` library when importable (full schema
# validation and spec-tracking); otherwise falls back to a stdlib-only
# builder that produces the same document shape.
try:
import xarf as _xarf
# Only use the library when it exposes the expected XARF v4 generator API
# (guards against an unrelated/older package squatting the `xarf` name).
_HAVE_XARF = hasattr(_xarf, "create_report") and hasattr(_xarf, "create_evidence")
except Exception as e: # pragma: no cover - depends on optional install
# Broad on purpose: a broken/incompatible install (e.g. a pydantic
# version mismatch) can raise more than ImportError at import time, and
# that must not take down the whole action - fall back to the stdlib
# builder instead.
logSys.warning("xarf library unusable (%s: %s); using stdlib builder",
type(e).__name__, e)
_xarf = None
_HAVE_XARF = False
# Pinned fallback spec version used by the stdlib builder when the
# official ``xarf`` library is not installed.
XARF_VERSION_FALLBACK = "4.2.0"
CATEGORY = "connection"
TYPE = "login_attack"
class XarfValidationError(Exception):
"""Raised when the xarf library validated a report and rejected it.
Kept distinct from other failures (missing dependency, unexpected
library errors) so build_login_attack() can choose not to send a
document the library has already told us is schema-invalid, rather
than silently shipping the same data unvalidated via the stdlib
fallback.
"""
def _build_evidence_stdlib(text, description=None):
"""Return a single XARF v4 evidence item for a block of log text."""
raw = text.encode('utf-8')
ev = {
"content_type": "text/plain",
"payload": base64.b64encode(raw).decode('ascii'),
"hash": "sha256:" + hashlib.sha256(raw).hexdigest(),
"size": len(raw),
}
if description is not None:
ev["description"] = description
return ev
def _build_login_attack_stdlib(data):
"""Build a XARF v4 login_attack report dict using only the stdlib."""
report = {
"xarf_version": XARF_VERSION_FALLBACK,
"report_id": str(uuid.uuid4()),
"timestamp": data["timestamp"],
"reporter": dict(data["reporter"]),
"sender": dict(data["sender"]),
"source_identifier": data["source_identifier"],
"category": CATEGORY,
"type": TYPE,
"protocol": data["protocol"],
"first_seen": data["first_seen"],
}
# optional fields — include only when present and not None:
for key in ("evidence_source", "service", "destination_ip",
"destination_port", "source_port", "attempt_count"):
val = data.get(key)
if val is not None:
report[key] = val
evtext = data.get("evidence_text")
if evtext:
report["evidence"] = [
_build_evidence_stdlib(evtext, description="fail2ban log matches")]
return report
def _build_login_attack_lib(data):
"""Build + validate a login_attack report via the official xarf lib."""
kwargs = {
"protocol": data["protocol"],
"first_seen": data["first_seen"],
"timestamp": data["timestamp"],
}
for key in ("evidence_source", "service", "destination_ip",
"destination_port", "source_port", "attempt_count"):
val = data.get(key)
if val is not None:
kwargs[key] = val
evtext = data.get("evidence_text")
if evtext:
kwargs["evidence"] = [_xarf.create_evidence(
"text/plain", evtext, description="fail2ban log matches")]
result = _xarf.create_report(
category=CATEGORY,
type=TYPE,
source_identifier=data["source_identifier"],
reporter=dict(data["reporter"]),
sender=dict(data["sender"]),
**kwargs)
if result.errors or result.report is None:
raise XarfValidationError(
"xarf validation failed: %r" % (result.errors,))
report = result.report.model_dump(by_alias=True, exclude_none=True)
# create_report() stamps `timestamp` with its own value (now) rather than
# honoring the one we pass in, leaving the document internally
# inconsistent with `first_seen`; enforce the ban-time timestamp we were
# given.
report["timestamp"] = data["timestamp"]
return report
def build_login_attack(data):
"""Return a XARF v4 login_attack report dict from fail2ban ban data, or
None if the report is known to be schema-invalid.
Uses the official ``xarf`` library when available. If the library
itself is unavailable or fails unexpectedly, falls back to the stdlib
builder so a report is still produced (availability over strictness).
If the library successfully validates the data and rejects it as
non-conformant, no report is returned at all: sending a document the
library has already flagged as invalid would be worse than not
reporting.
"""
if _HAVE_XARF:
try:
return _build_login_attack_lib(data)
except XarfValidationError as e:
logSys.error(
"xarf library rejected report as schema-invalid (%s); "
"not sending an unvalidated fallback", e)
return None
except Exception as e: # noqa: broad - never block reporting
logSys.warning(
"xarf library report build failed (%s: %s); "
"falling back to stdlib builder", type(e).__name__, e)
return _build_login_attack_stdlib(data)
# --- XARF v4 email action -----------------------------------------
# Strict address validation for abuse contacts resolved from DNS (untrusted):
# rejects anything that is not an email and anything that could be an argv flag.
_EMAIL_CRE = re.compile(r'^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$')
class XarfV4Action(ActionBase):
def __init__(self, jail, name,
reporter_org=None, reporter_contact=None, reporter_domain=None,
sender_org=None, sender_contact=None, sender_domain=None,
service="unspecified", port="0", protocol="tcp",
sender=None, mailcmd="/usr/sbin/sendmail", mailargs=None,
resolver="abuse-contacts.abusix.org", matches="ipmatches"):
# reporter_*/sender_* are the XARF v4 report identities; the separate
# `sender` arg is the e-mail envelope From address (self.envelope_from).
super(XarfV4Action, self).__init__(jail, name)
from fail2ban.server.ipdns import DNSUtils
self.reporter = {"org": reporter_org, "contact": reporter_contact,
"domain": reporter_domain}
self.sender_id = {"org": sender_org, "contact": sender_contact,
"domain": sender_domain}
self.service = service
self.port = port
self.protocol = protocol
self.envelope_from = sender or ("fail2ban@" + DNSUtils.getHostname())
self.mailcmd = mailcmd
# default envelope-sender flag so SPF/DMARC can align with envelope_from;
# operators can override entirely via mailargs.
self.mailargs = mailargs if mailargs is not None else ("-f " + self.envelope_from)
self.resolver = resolver
self.matches = matches
# bypass ban/unban for restored tickets
self.norestored = 1
def _identity_ok(self):
return all(self.reporter.values()) and all(self.sender_id.values())
def start(self):
pass
def stop(self):
pass
def unban(self, aInfo):
pass
def _reverse_ip(self, ip):
"""Return the reversed-nibble/octet label for a TXT lookup."""
from fail2ban.server.ipdns import IPAddr
return IPAddr(ip).getPTR("")
def _dig_txt(self, fqdn):
"""Return raw `dig +short TXT` output for fqdn (overridable in tests)."""
try:
out = subprocess.check_output(
["dig", "+short", "-t", "txt", "-q", fqdn],
universal_newlines=True, timeout=30)
except Exception as e:
self._logSys.error("xarf action %s: dig failed for %s: %s",
self._name, fqdn, e)
return ''
return out
def _resolveAbuseContacts(self, ip):
"""Resolve abuse contact email(s) for ip via the Abusix Contact DB."""
fqdn = self._reverse_ip(ip) + self.resolver
self._logSys.debug("xarf action %s: try to resolve %s",
self._name, fqdn)
raw = self._dig_txt(fqdn)
addrs = []
for line in raw.splitlines():
line = line.strip().strip('"')
if not line or line.startswith(';;'):
continue
for part in line.split(','):
part = part.strip()
if not part or part.startswith('-'):
continue
if _EMAIL_CRE.match(part):
addrs.append(part)
else:
self._logSys.warning(
"xarf action %s: ignoring invalid abuse contact %r",
self._name, part)
return addrs
def _build_email(self, report):
"""Wrap a XARF v4 report dict in the XARF v4 email transport format.
The returned message has no recipient header; the caller is
responsible for setting the recipients when sending.
"""
ip = report.get("source_identifier", "")
msg = MIMEMultipart("report")
msg.set_param("report-type", "feedback-report")
msg["Subject"] = "Abuse report (login-attack) about %s" % ip
msg["From"] = self.envelope_from
msg["Auto-Submitted"] = "auto-generated"
msg["Date"] = formatdate()
msg["Message-ID"] = make_msgid()
human = ("An abuse login-attack report follows.\n\n"
"Report-Type: %s\nSource: %s\nTimestamp: %s\nReport-ID: %s\n\n"
"The machine-readable XARF v4 report is attached as xarf.json.\n"
% (report.get("type", "login_attack"), ip,
report.get("timestamp", ""), report.get("report_id", "")))
msg.attach(MIMEText(human, "plain", "us-ascii"))
fb = MIMEBase("message", "feedback-report")
fb.set_payload(
"Feedback-Type: xarf\nUser-Agent: Fail2Ban\nVersion: 1\n")
fb.add_header("Content-Disposition", "inline")
msg.attach(fb)
jsonpart = MIMEBase("application", "json", name="xarf.json")
jsonpart.set_payload(json.dumps(report).encode("utf-8"))
encoders.encode_base64(jsonpart)
jsonpart.add_header(
"Content-Disposition", "attachment", filename="xarf.json")
msg.attach(jsonpart)
return msg
def _iso(self, epoch):
return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(float(epoch)))
def _extract_source_port(self, text, ip):
"""Best-effort: extract the attacker source port from matched log text.
fail2ban generally knows only the attacked (destination) port; some
services (e.g. sshd) log the source port in the failure line. The
pattern is anchored to the banned IP itself, since a single log line
can name more than one "<addr> port <n>" pair - e.g. sshd's verbose
"Connection from <ip> port 51353 on 127.0.0.1 port 22" also names the
local listening port, which is not the attacker's port. When present,
the most recent (triggering) match is used. Returns an int in
1..65535, or None when unavailable.
"""
if not text:
return None
cre = re.compile(
r'\b' + re.escape(str(ip)) + r'\s+port\s+(\d{1,5})\b')
matches = cre.findall(text)
if not matches:
return None
try:
p = int(matches[-1])
except (TypeError, ValueError):
return None
return p if 1 <= p <= 65535 else None
def _resolve_port(self, port):
"""Best-effort parse of fail2ban's `port` option into a single XARF
v4 `destination_port`.
`port` may be a service name (``ssh``), a comma-separated list
(``http,https``) or a colon/dash range (``0:65535``); XARF v4 wants
one numeric port in 1..65535, so anything else is treated as unknown
(``None``) rather than guessed at.
"""
if not port:
return None
first = str(port).split(',', 1)[0].strip()
if not first or ':' in first or '-' in first:
return None
try:
p = int(first)
except (TypeError, ValueError):
try:
p = socket.getservbyname(first)
except OSError:
return None
return p if 1 <= p <= 65535 else None
def _aInfoToData(self, aInfo):
dport = self._resolve_port(self.port)
ts = self._iso(aInfo.get('time') or time.time())
evidence = aInfo.get(self.matches) or aInfo.get('matches') or ''
failures = aInfo.get('failures')
data = {
"source_identifier": str(aInfo['ip']),
"timestamp": ts,
"first_seen": ts,
"protocol": self.protocol,
"service": self.service,
"attempt_count": int(failures) if failures is not None else None,
"evidence_text": evidence,
"evidence_source": "log",
"reporter": self.reporter,
"sender": self.sender_id,
}
if dport is not None:
data["destination_port"] = dport
sport = self._extract_source_port(evidence, aInfo['ip'])
if sport is not None:
data["source_port"] = sport
return data
def _sendmail(self, recipients, msg):
cmd = (shlex.split(self.mailcmd) + shlex.split(self.mailargs)
+ ['--'] + list(recipients))
p = subprocess.Popen(cmd, stdin=subprocess.PIPE)
try:
p.communicate(msg.as_bytes(), timeout=30)
except subprocess.TimeoutExpired:
p.kill()
p.communicate()
self._logSys.error(
"xarf action %s: mail command timed out", self._name)
return
if p.returncode:
self._logSys.error("xarf action %s: mail command failed (rc=%s)",
self._name, p.returncode)
def ban(self, aInfo):
if aInfo.get('restored'):
return
if not self._identity_ok():
self._logSys.error(
"xarf action %s: reporter/sender identity is incomplete; "
"set reporter_org/contact/domain and sender_org/contact/"
"domain - skipping report", self._name)
return
if not (aInfo.get(self.matches) or aInfo.get('matches')):
# e.g. a manual `fail2ban-client set <jail> banip <ip>` ticket:
# no log evidence to report and nothing to substantiate a claim.
self._logSys.info(
"xarf action %s: no evidence for %s (manual ban?); skipping "
"report", self._name, aInfo.get('ip'))
return
contacts = self._resolveAbuseContacts(str(aInfo['ip']))
if not contacts:
self._logSys.info(
"xarf action %s: no abuse contact for %s; skipping",
self._name, aInfo['ip'])
return
try:
data = self._aInfoToData(aInfo)
report = build_login_attack(data)
if report is None:
# the xarf library validated the report and rejected it as
# schema-invalid; sending it unvalidated would be worse than
# not reporting at all.
self._logSys.warning(
"xarf action %s: report for %s failed schema "
"validation; not sending", self._name, aInfo.get('ip'))
return
msg = self._build_email(report)
msg["To"] = ", ".join(contacts)
self._sendmail(contacts, msg)
except Exception as e: # never block the ban path
self._logSys.error(
"xarf action %s: failed to build/send report for %s: %s",
self._name, aInfo.get('ip'), e)
return
self._logSys.debug("xarf action %s: report sent for %s to %s",
self._name, aInfo['ip'], contacts)
Action = XarfV4Action

25
config/jail.conf
View file

@ -198,6 +198,15 @@ port = 0:65535
# Format of user-agent https://tools.ietf.org/html/rfc7231#section-5.5.3
fail2ban_agent = Fail2Ban/%(fail2ban_version)s
# XARF v4 reporter/sender identity (set these in jail.local to use action_xarf_v4)
# (xarf_sender_* defaults to xarf_reporter_* if not overridden)
xarf_reporter_org =
xarf_reporter_contact =
xarf_reporter_domain =
xarf_sender_org = %(xarf_reporter_org)s
xarf_sender_contact = %(xarf_reporter_contact)s
xarf_sender_domain = %(xarf_reporter_domain)s
#
# Action shortcuts. To be used to define action parameter
@ -220,6 +229,7 @@ action_mw = %(action_)s
action_mwl = %(action_)s
%(mta)s-whois-lines[sender="%(sender)s", dest="%(destemail)s", logpath="%(logpath)s", chain="%(chain)s"]
# DEPRECATED (x-arf 0.2): use action_xarf_v4 (XARF v4) instead; this will be removed in a future release.
# See the IMPORTANT note in action.d/xarf-login-attack for when to use this action
#
# ban & send a xarf e-mail to abuse contact of IP address and include relevant log lines
@ -227,6 +237,21 @@ action_mwl = %(action_)s
action_xarf = %(action_)s
xarf-login-attack[service=%(__name__)s, sender="%(sender)s", logpath="%(logpath)s", port="%(port)s"]
# See the IMPORTANT note in action.d/xarf.py for when to use this action
#
# ban & send a XARF v4 login-attack report by e-mail to the IP's abuse
# contact (resolved via the Abusix Contact DB). Requires reporter/sender
# identity (xarf_*_org/contact/domain) set in jail.local. Install the official
# XARF v4 `xarf` python package (pip install xarf) for full schema validation;
# a built-in generator is used otherwise.
action_xarf_v4 = %(action_)s
xarf.py[service="%(__name__)s", port="%(port)s", protocol="%(protocol)s",
sender="%(sender)s",
reporter_org="%(xarf_reporter_org)s", reporter_contact="%(xarf_reporter_contact)s",
reporter_domain="%(xarf_reporter_domain)s",
sender_org="%(xarf_sender_org)s", sender_contact="%(xarf_sender_contact)s",
sender_domain="%(xarf_sender_domain)s"]
# ban & send a notification to one or more of the 120+ services supported by Apprise.
# action = %(action_)s
# apprise[config="/alternate/path/to/apprise.yaml", args='--tag fail2ban']

461
fail2ban/tests/action_d/test_xarf.py Normal file
View file

@ -0,0 +1,461 @@
# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: t -*-
# vi: set ft=python sts=4 ts=4 sw=4 noet :
# This file is part of Fail2Ban.
#
# Fail2Ban is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# Fail2Ban is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Fail2Ban; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
__author__ = "Fail2Ban Developers"
__copyright__ = "Copyright (c) 2025 Fail2Ban Developers"
__license__ = "GPL"
import base64
import hashlib
import os
import subprocess
import sys
import unittest
from ..dummyjail import DummyJail
from ..utils import CONFIG_DIR, Utils, LogCaptureTestCase
_ACTFILE = os.path.join(CONFIG_DIR, "action.d", "xarf.py")
try:
import xarf as _xarf_lib
_HAVE_XARF_LIB = (hasattr(_xarf_lib, "create_report")
and hasattr(_xarf_lib, "create_evidence"))
except Exception:
_HAVE_XARF_LIB = False
class XarfEvidenceTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.mod = Utils.load_python_module(_ACTFILE)
def testBuildEvidenceStdlib(self):
text = "Dec 31 11:59:59 sshd: auth failure from 87.142.124.10"
ev = self.mod._build_evidence_stdlib(text, description="logs")
self.assertEqual(ev['content_type'], "text/plain")
self.assertEqual(ev['description'], "logs")
# payload is base64 of the utf-8 text:
self.assertEqual(
base64.b64decode(ev['payload']).decode('utf-8'), text)
# hash is "sha256:<hexdigest>":
digest = hashlib.sha256(text.encode('utf-8')).hexdigest()
self.assertEqual(ev['hash'], "sha256:" + digest)
self.assertEqual(ev['size'], len(text.encode('utf-8')))
def testBuildEvidenceStdlibNoDescription(self):
ev = self.mod._build_evidence_stdlib("a log line")
self.assertNotIn('description', ev)
self.assertEqual(ev['content_type'], "text/plain")
class XarfLoginAttackStdlibTest(LogCaptureTestCase):
@classmethod
def setUpClass(cls):
cls.mod = Utils.load_python_module(_ACTFILE)
def _data(self, **over):
data = {
"source_identifier": "87.142.124.10",
"timestamp": "2025-01-11T12:17:20Z",
"first_seen": "2025-01-11T06:17:20Z",
"protocol": "tcp",
"service": "sshd",
"destination_port": 22,
"source_port": 45621,
"attempt_count": 5,
"evidence_text": "auth failure from 87.142.124.10",
"evidence_source": "log",
"reporter": {"org": "Acme", "contact": "abuse@acme.example",
"domain": "acme.example"},
"sender": {"org": "Acme", "contact": "abuse@acme.example",
"domain": "acme.example"},
}
data.update(over)
return data
def testBuildLoginAttackStdlib(self):
r = self.mod._build_login_attack_stdlib(self._data())
self.assertEqual(r['xarf_version'], self.mod.XARF_VERSION_FALLBACK)
self.assertEqual(r['category'], "connection")
self.assertEqual(r['type'], "login_attack")
self.assertEqual(r['source_identifier'], "87.142.124.10")
self.assertEqual(r['destination_port'], 22)
self.assertEqual(r['source_port'], 45621)
self.assertEqual(r['protocol'], "tcp")
self.assertEqual(r['service'], "sshd")
self.assertEqual(r['attempt_count'], 5)
self.assertEqual(r['timestamp'], "2025-01-11T12:17:20Z")
self.assertEqual(r['first_seen'], "2025-01-11T06:17:20Z")
self.assertEqual(r['reporter']['org'], "Acme")
self.assertEqual(r['sender']['domain'], "acme.example")
self.assertEqual(len(r['evidence']), 1)
self.assertEqual(r['evidence'][0]['content_type'], "text/plain")
import uuid as _uuid
_uuid.UUID(r['report_id'])
def testBuildLoginAttackStdlibOmitsMissingOptionals(self):
data = self._data()
del data['destination_port']
del data['attempt_count']
r = self.mod._build_login_attack_stdlib(data)
self.assertNotIn('destination_port', r)
self.assertNotIn('attempt_count', r)
self.assertIn('source_identifier', r)
self.assertIn('protocol', r)
self.assertIn('first_seen', r)
class XarfBuildLoginAttackTest(XarfLoginAttackStdlibTest):
def testPublicEntryPointReturnsValidShape(self):
# Regardless of whether the lib is present, the public function
# returns a dict with the required v4 fields.
r = self.mod.build_login_attack(self._data())
for field in ("xarf_version", "report_id", "timestamp", "reporter",
"sender", "source_identifier", "category", "type",
"protocol", "first_seen"):
self.assertIn(field, r)
self.assertEqual(r['category'], "connection")
self.assertEqual(r['type'], "login_attack")
@unittest.skipUnless(_HAVE_XARF_LIB, "xarf library not installed")
def testLibPathValidatesAgainstSchema(self):
# When the lib is present, output must pass xarf's own validation.
import xarf
r = self.mod.build_login_attack(self._data())
result = xarf.parse(r)
self.assertEqual(result.errors, [])
def testFallbackOnLibError(self):
# Force the lib path to raise; expect a clean fallback + warning.
orig_flag = self.mod._HAVE_XARF
orig_lib = self.mod._build_login_attack_lib
try:
self.mod._HAVE_XARF = True
def boom(data):
raise RuntimeError("simulated lib failure")
self.mod._build_login_attack_lib = boom
r = self.mod.build_login_attack(self._data())
self.assertEqual(r['type'], "login_attack")
self.assertLogged("falling back to stdlib")
finally:
self.mod._HAVE_XARF = orig_flag
self.mod._build_login_attack_lib = orig_lib
def testDropsReportOnLibValidationRejection(self):
# When the lib validates and rejects the data (e.g. a genuinely
# missing required field), no report should be sent at all - not
# even the unvalidated stdlib version of the same broken data.
orig_flag = self.mod._HAVE_XARF
orig_lib = self.mod._build_login_attack_lib
try:
self.mod._HAVE_XARF = True
def reject(data):
raise self.mod.XarfValidationError(
"xarf validation failed: "
"[\"'source_port' is a required property\"]")
self.mod._build_login_attack_lib = reject
r = self.mod.build_login_attack(self._data())
self.assertIsNone(r)
self.assertLogged("rejected report as schema-invalid")
finally:
self.mod._HAVE_XARF = orig_flag
self.mod._build_login_attack_lib = orig_lib
class XarfImportGuardTest(LogCaptureTestCase):
def testHaveXarfFalseOnBrokenImport(self):
# A broken/incompatible `xarf` install (e.g. a pydantic version
# mismatch) can raise more than ImportError while `xarf` itself
# imports fine but blows up on attribute access; that must not
# crash the whole module import.
class _BrokenXarf(object):
def __getattr__(self, name):
raise RuntimeError("simulated broken xarf install")
orig_mod = sys.modules.get('xarf')
sys.modules['xarf'] = _BrokenXarf()
try:
mod = Utils.load_python_module(_ACTFILE)
self.assertFalse(mod._HAVE_XARF)
self.assertLogged("xarf library unusable")
finally:
if orig_mod is not None:
sys.modules['xarf'] = orig_mod
else:
sys.modules.pop('xarf', None)
class XarfV4ActionTest(LogCaptureTestCase):
def setUp(self):
super(XarfV4ActionTest, self).setUp()
self.__jail = DummyJail()
self.mod = Utils.load_python_module(_ACTFILE)
self.Action = self.mod.Action
def _mk(self, **over):
opts = dict(
reporter_org="Acme", reporter_contact="abuse@acme.example",
reporter_domain="acme.example",
sender_org="Acme", sender_contact="abuse@acme.example",
sender_domain="acme.example",
service="sshd", port="22")
opts.update(over)
return self.Action(self.__jail, "xarf", **opts)
def testMissingIdentitySkips(self):
act = self._mk(reporter_org=None)
sent = []
act._sendmail = lambda recipients, msg: sent.append((recipients, msg))
act._resolveAbuseContacts = lambda ip: ["abuse@isp.example"]
act.ban({'ip': '87.142.124.10', 'failures': 3,
'time': 1736597840, 'ipmatches': 'log line'})
self.assertEqual(sent, [])
self.assertLogged("reporter/sender identity is incomplete")
def testMissingSenderIdentitySkips(self):
act = self._mk(sender_domain=None)
sent = []
act._sendmail = lambda recipients, msg: sent.append((recipients, msg))
act._resolveAbuseContacts = lambda ip: ["abuse@isp.example"]
act.ban({'ip': '87.142.124.10', 'failures': 3,
'time': 1736597840, 'ipmatches': 'log line'})
self.assertEqual(sent, [])
self.assertLogged("reporter/sender identity is incomplete")
def testResolveParsesDigTxt(self):
act = self._mk()
# stub the dig invocation: return a TXT answer like dig +short
act._dig_txt = lambda fqdn: '"abuse-1@isp.example, abuse-2@isp.example"'
addrs = act._resolveAbuseContacts('87.142.124.10')
self.assertEqual(addrs, ["abuse-1@isp.example", "abuse-2@isp.example"])
def testResolveBuildsAbusixFqdn(self):
act = self._mk()
seen = []
act._dig_txt = lambda fqdn: seen.append(fqdn) or '"abuse@isp.example"'
act._resolveAbuseContacts('87.142.124.10')
self.assertEqual(seen, ["10.124.142.87.abuse-contacts.abusix.org"])
def testResolveBuildsAbusixFqdnIPv6(self):
act = self._mk()
seen = []
act._dig_txt = lambda fqdn: seen.append(fqdn) or '"abuse@isp.example"'
act._resolveAbuseContacts('2001:db8::1')
self.assertEqual(seen, ["1.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.8.b.d.0.1.0.0.2.abuse-contacts.abusix.org"])
def testResolveEmptyOnNoAnswer(self):
act = self._mk()
act._dig_txt = lambda fqdn: ''
self.assertEqual(act._resolveAbuseContacts('87.142.124.10'), [])
def testResolveRejectsFlagAndJunkTokens(self):
# DNS is untrusted: drop argv-flag-like and non-email tokens.
act = self._mk()
act._dig_txt = lambda fqdn: '"-X/tmp/evil, not-an-email, good@isp.example"'
self.assertEqual(
act._resolveAbuseContacts('87.142.124.10'), ["good@isp.example"])
self.assertLogged("ignoring invalid abuse contact")
def testBuildEmailStructure(self):
import base64 as _b64, json as _json
act = self._mk()
report = {"xarf_version": "4.2.0", "type": "login_attack",
"source_identifier": "87.142.124.10",
"timestamp": "2025-01-11T12:17:20Z", "report_id": "abc-123"}
msg = act._build_email(report)
self.assertEqual(msg.get_content_type(), "multipart/report")
self.assertEqual(msg.get_param("report-type"), "feedback-report")
parts = msg.get_payload()
self.assertEqual(len(parts), 3)
self.assertEqual(parts[0].get_content_type(), "text/plain")
self.assertEqual(parts[1].get_content_type(), "message/feedback-report")
fbtext = parts[1].get_payload()
self.assertIn("Feedback-Type: xarf", fbtext)
self.assertIn("User-Agent", fbtext)
self.assertIn("Version: 1", fbtext)
self.assertEqual(parts[2].get_content_type(), "application/json")
self.assertEqual(parts[2].get_filename(), "xarf.json")
decoded = _b64.b64decode(parts[2].get_payload())
self.assertEqual(_json.loads(decoded)['source_identifier'],
"87.142.124.10")
# subject references the source IP:
self.assertIn("87.142.124.10", msg['Subject'])
def testBanSendsReport(self):
import base64 as _b64, json as _json
act = self._mk(service="sshd", port="22")
sent = []
act._resolveAbuseContacts = lambda ip: ["abuse@isp.example"]
act._sendmail = lambda recipients, msg: sent.append((recipients, msg))
act.ban({'ip': '87.142.124.10', 'failures': 7,
'time': 1736597840, 'ipmatches': 'Jan 11 sshd auth failure'})
self.assertEqual(len(sent), 1)
recipients, msg = sent[0]
self.assertEqual(recipients, ["abuse@isp.example"])
self.assertEqual(msg['To'], "abuse@isp.example")
jsonpart = msg.get_payload()[2]
report = _json.loads(_b64.b64decode(jsonpart.get_payload()))
self.assertEqual(report['source_identifier'], '87.142.124.10')
self.assertEqual(report['type'], 'login_attack')
self.assertEqual(report['destination_port'], 22)
self.assertEqual(report['attempt_count'], 7)
self.assertEqual(len(report['evidence']), 1)
def testAInfoToDataDropsPortRange(self):
act = self._mk(port="0:65535")
d = act._aInfoToData({'ip': '1.2.3.4', 'time': 1736597840,
'failures': 2, 'ipmatches': 'x'})
self.assertNotIn('destination_port', d)
self.assertEqual(d['source_identifier'], '1.2.3.4')
def testAInfoToDataResolvesServiceName(self):
act = self._mk(port="ssh")
d = act._aInfoToData({'ip': '1.2.3.4', 'time': 1736597840,
'failures': 2, 'ipmatches': 'x'})
self.assertEqual(d['destination_port'], 22)
def testAInfoToDataCommaListUsesFirstEntry(self):
act = self._mk(port="http,https")
d = act._aInfoToData({'ip': '1.2.3.4', 'time': 1736597840,
'failures': 2, 'ipmatches': 'x'})
self.assertEqual(d['destination_port'], 80)
def testAInfoToDataOmitsZeroPort(self):
# port="0" "parses" but 0 violates the schema's minimum of 1.
act = self._mk(port="0")
d = act._aInfoToData({'ip': '1.2.3.4', 'time': 1736597840,
'failures': 2, 'ipmatches': 'x'})
self.assertNotIn('destination_port', d)
def testAInfoExtractsSourcePortFromSshdLine(self):
act = self._mk(port="22")
d = act._aInfoToData({'ip': '203.0.113.5', 'time': 1736597840,
'failures': 4,
'ipmatches': 'Failed password for root from 203.0.113.5 port 52431 ssh2'})
self.assertEqual(d['source_port'], 52431)
self.assertEqual(d['destination_port'], 22)
def testAInfoNoSourcePortWhenAbsent(self):
act = self._mk(port="22")
d = act._aInfoToData({'ip': '203.0.113.5', 'time': 1736597840,
'failures': 4,
'ipmatches': 'authentication failure for kevin from 203.0.113.5'})
self.assertNotIn('source_port', d)
def testAInfoSourcePortLastMatchWins(self):
act = self._mk(port="22")
d = act._aInfoToData({'ip': '203.0.113.5', 'time': 1736597840,
'failures': 2,
'ipmatches': 'from 203.0.113.5 port 1111 ssh2\nfrom 203.0.113.5 port 2222 ssh2'})
self.assertEqual(d['source_port'], 2222)
def testAInfoSourcePortIgnoresServicePortInVerboseLine(self):
# LogLevel VERBOSE sshd logs both the attacker's port and the local
# listening port on the same line; only the one next to the
# attacker's own IP is the attacker's source port.
act = self._mk(port="22")
d = act._aInfoToData({'ip': '115.249.163.77', 'time': 1736597840,
'failures': 1,
'ipmatches':
'Connection from 115.249.163.77 port 51353 on 127.0.0.1 port 22'})
self.assertEqual(d['source_port'], 51353)
def testBanSkipsWhenNoEvidence(self):
# e.g. `fail2ban-client set <jail> banip <ip>`: no log evidence at
# all, nothing to substantiate a complaint.
act = self._mk()
sent = []
act._resolveAbuseContacts = lambda ip: ["abuse@isp.example"]
act._sendmail = lambda recipients, msg: sent.append((recipients, msg))
act.ban({'ip': '87.142.124.10', 'failures': 0, 'time': 1736597840,
'ipmatches': []})
self.assertEqual(sent, [])
self.assertLogged("no evidence")
def testBanSkipsWhenReportFailsValidation(self):
act = self._mk()
sent = []
act._resolveAbuseContacts = lambda ip: ["abuse@isp.example"]
act._sendmail = lambda recipients, msg: sent.append((recipients, msg))
orig = self.mod.build_login_attack
self.mod.build_login_attack = lambda data: None
try:
act.ban({'ip': '87.142.124.10', 'failures': 3,
'time': 1736597840, 'ipmatches': 'log line'})
finally:
self.mod.build_login_attack = orig
self.assertEqual(sent, [])
self.assertLogged("failed schema validation; not sending")
def testSendmailPassesEnvelopeFromFlagByDefault(self):
act = self._mk(sender="fail2ban@example.com")
captured = {}
class FakeProc(object):
def __init__(self, cmd, **kw):
captured['cmd'] = cmd
self.returncode = 0
def communicate(self, data=None, timeout=None):
return (b'', b'')
orig_popen = subprocess.Popen
subprocess.Popen = FakeProc
try:
act._sendmail(["abuse@isp.example"],
act._build_email({"source_identifier": "1.2.3.4"}))
finally:
subprocess.Popen = orig_popen
self.assertIn('-f', captured['cmd'])
self.assertEqual(
captured['cmd'][captured['cmd'].index('-f') + 1],
"fail2ban@example.com")
def testSendmailRespectsMailargsOverride(self):
act = self._mk(mailargs="-oi -f custom@example.com")
captured = {}
class FakeProc(object):
def __init__(self, cmd, **kw):
captured['cmd'] = cmd
self.returncode = 0
def communicate(self, data=None, timeout=None):
return (b'', b'')
orig_popen = subprocess.Popen
subprocess.Popen = FakeProc
try:
act._sendmail(["abuse@isp.example"],
act._build_email({"source_identifier": "1.2.3.4"}))
finally:
subprocess.Popen = orig_popen
self.assertEqual(
captured['cmd'][captured['cmd'].index('-f') + 1],
"custom@example.com")
self.assertIn('-oi', captured['cmd'])
def testBanNoContactNoSend(self):
act = self._mk()
sent = []
act._resolveAbuseContacts = lambda ip: []
act._sendmail = lambda recipients, msg: sent.append((recipients, msg))
act.ban({'ip': '87.142.124.10', 'failures': 1,
'time': 1736597840, 'ipmatches': 'x'})
self.assertEqual(sent, [])
self.assertLogged("no abuse contact")