ENH+TST: hybrid XARF v4 generator with optional xarf library

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

fail2ban/server/xarfreport.py

@@ -36,6 +36,13 @@ from ..helpers import getLogger
 
 logSys = getLogger(__name__)
 
+try:
+	import xarf as _xarf
+	_HAVE_XARF = True
+except ImportError:  # pragma: no cover - depends on optional install
+	_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"
@@ -82,3 +89,47 @@ def _build_login_attack_stdlib(data):
 		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 ValueError("xarf validation failed: %r" % (result.errors,))
+	return result.report.model_dump(by_alias=True, exclude_none=True)
+
+
+def build_login_attack(data):
+	"""Return a XARF v4 login_attack report dict from fail2ban ban data.
+
+	Uses the official ``xarf`` library when available; on any error falls
+	back to the stdlib builder so a report is still produced.
+	"""
+	if _HAVE_XARF:
+		try:
+			return _build_login_attack_lib(data)
+		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)

fail2ban/tests/utils.py

@@ -452,6 +452,7 @@ def gatherTests(regexps=None, opts=None):
 	# XARF v4 reporting
 	tests.addTest(loadTests(xarfreporttestcase.XarfEvidenceTest))
 	tests.addTest(loadTests(xarfreporttestcase.XarfLoginAttackStdlibTest))
+	tests.addTest(loadTests(xarfreporttestcase.XarfBuildLoginAttackTest))
 	# Database
 	tests.addTest(loadTests(databasetestcase.DatabaseTest))
 	# Observer

fail2ban/tests/xarfreporttestcase.py

@@ -100,3 +100,41 @@ class XarfLoginAttackStdlibTest(LogCaptureTestCase):
 		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 = xarfreport.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(xarfreport._HAVE_XARF, "xarf library not installed")
+	def testLibPathValidatesAgainstSchema(self):
+		# When the lib is present, output must pass xarf's own validation.
+		import xarf
+		r = xarfreport.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 = xarfreport._HAVE_XARF
+		orig_lib = xarfreport._build_login_attack_lib
+		try:
+			xarfreport._HAVE_XARF = True
+			def boom(data):
+				raise RuntimeError("simulated lib failure")
+			xarfreport._build_login_attack_lib = boom
+			r = xarfreport.build_login_attack(self._data())
+			self.assertEqual(r['type'], "login_attack")
+			self.assertLogged("falling back to stdlib")
+		finally:
+			xarfreport._HAVE_XARF = orig_flag
+			xarfreport._build_login_attack_lib = orig_lib