diff --git a/.gitignore b/.gitignore index 5f1b8924..688e3a5e 100644 --- a/.gitignore +++ b/.gitignore @@ -11,3 +11,4 @@ __pycache__ .vagrant/ .idea/ .venv/ +bin/fail2ban-python diff --git a/ChangeLog b/ChangeLog index 32f2dec0..921d5a13 100644 --- a/ChangeLog +++ b/ChangeLog @@ -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 diff --git a/config/action.d/xarf-login-attack.conf b/config/action.d/xarf-login-attack.conf index 00da7f79..9affd0b2 100644 --- a/config/action.d/xarf-login-attack.conf +++ b/config/action.d/xarf-login-attack.conf @@ -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 diff --git a/config/action.d/xarf.py b/config/action.d/xarf.py new file mode 100644 index 00000000..80a4f04b --- /dev/null +++ b/config/action.d/xarf.py @@ -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 " port " pair - e.g. sshd's verbose + "Connection from 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 banip ` 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 diff --git a/config/jail.conf b/config/jail.conf index 8f02e9af..39afe1f9 100644 --- a/config/jail.conf +++ b/config/jail.conf @@ -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'] diff --git a/fail2ban/tests/action_d/test_xarf.py b/fail2ban/tests/action_d/test_xarf.py new file mode 100644 index 00000000..215e0031 --- /dev/null +++ b/fail2ban/tests/action_d/test_xarf.py @@ -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:": + 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 banip `: 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")