ENH+TST: add XARF v4 email action skeleton with identity validation

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

config/action.d/xarf.py

@@ -0,0 +1,82 @@
+# 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 official XARF email
+transport (RFC 5965 multipart/report). Requires the ``dig`` command
+(bind-utils) for contact resolution.
+"""
+
+from fail2ban.server.actions import ActionBase
+from fail2ban.server import xarfreport
+
+
+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",
+			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)
+		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"
+		self.mailcmd = mailcmd
+		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 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
+		# remaining wiring added in later tasks
+		self._logSys.debug("xarf action %s: ban() reached", self._name)
+
+
+Action = XarfV4Action

fail2ban/tests/servertestcase.py

@@ -2411,3 +2411,43 @@ class ServerConfigReaderTests(LogCaptureTestCase):
 					ticket = _actions.Actions.ActionInfo(ticket, dmyjail)
 					action.ban(ticket)
 					self.assertLogged(*tests[test], all=True)
+
+
+class XarfV4ActionTest(LogCaptureTestCase):
+
+	def setUp(self):
+		super(XarfV4ActionTest, self).setUp()
+		self.__jail = DummyJail()
+		actfile = os.path.join(CONFIG_DIR, "action.d", "xarf.py")
+		mod = Utils.load_python_module(actfile)
+		self.Action = 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")

fail2ban/tests/utils.py

@@ -417,6 +417,7 @@ def gatherTests(regexps=None, opts=None):
 	tests.addTest(loadTests(servertestcase.RegexTests))
 	tests.addTest(loadTests(servertestcase.LoggingTests))
 	tests.addTest(loadTests(servertestcase.ServerConfigReaderTests))
+	tests.addTest(loadTests(servertestcase.XarfV4ActionTest))
 	tests.addTest(loadTests(actiontestcase.CommandActionTest))
 	tests.addTest(loadTests(actionstestcase.ExecuteActions))
 	# Ticket, BanTicket, FailTicket