diff --git a/fail2ban/server/xarfreport.py b/fail2ban/server/xarfreport.py index edc418e0..f0cbc2ce 100644 --- a/fail2ban/server/xarfreport.py +++ b/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) diff --git a/fail2ban/tests/utils.py b/fail2ban/tests/utils.py index 6456c47a..9bd10553 100644 --- a/fail2ban/tests/utils.py +++ b/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 diff --git a/fail2ban/tests/xarfreporttestcase.py b/fail2ban/tests/xarfreporttestcase.py index 91d2fd1c..38f3f4f5 100644 --- a/fail2ban/tests/xarfreporttestcase.py +++ b/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