feat(fail2ban): manage trusted IP addresses and networks

AppImage/scripts/flask_security_routes.py

@@ -5,6 +5,8 @@ ProxMenux Security Routes
 Flask blueprint for firewall management and security tool detection.
 """
 
+import ipaddress
+
 from flask import Blueprint, jsonify, request
 from jwt_middleware import require_auth
 
@@ -234,6 +236,80 @@ def fail2ban_jail_config():
         return jsonify({"success": False, "message": str(e)}), 500
 
 
+@security_bp.route('/api/security/fail2ban/trusted-networks', methods=['GET'])
+@require_auth
+def fail2ban_trusted_networks():
+    """List global IP/CIDR addresses excluded from all Fail2Ban jails."""
+    if not security_manager:
+        return jsonify({"success": False, "message": "Security manager not available"}), 500
+    try:
+        detected_ip = request.remote_addr
+        try:
+            parsed_ip = ipaddress.ip_address(detected_ip) if detected_ip else None
+            if isinstance(parsed_ip, ipaddress.IPv6Address) and parsed_ip.ipv4_mapped:
+                parsed_ip = parsed_ip.ipv4_mapped
+            if not parsed_ip or parsed_ip.is_loopback:
+                detected_ip = None
+            else:
+                detected_ip = str(parsed_ip)
+        except ValueError:
+            detected_ip = None
+        return jsonify({
+            "success": True,
+            "entries": security_manager.get_fail2ban_trusted_networks(),
+            "detected_ip": detected_ip,
+        })
+    except Exception as e:
+        return jsonify({"success": False, "message": str(e)}), 500
+
+
+@security_bp.route('/api/security/fail2ban/trusted-networks', methods=['POST'])
+@require_auth
+def fail2ban_add_trusted_network():
+    """Add one global Fail2Ban IP/CIDR exclusion."""
+    if not security_manager:
+        return jsonify({"success": False, "message": "Security manager not available"}), 500
+    try:
+        data = request.json or {}
+        success, message, value = security_manager.add_fail2ban_trusted_network(data.get("value", ""))
+        status = 200 if success else 400
+        return jsonify({"success": success, "message": message, "value": value}), status
+    except Exception as e:
+        return jsonify({"success": False, "message": str(e)}), 500
+
+
+@security_bp.route('/api/security/fail2ban/trusted-networks', methods=['DELETE'])
+@require_auth
+def fail2ban_remove_trusted_network():
+    """Remove one user-managed global Fail2Ban IP/CIDR exclusion."""
+    if not security_manager:
+        return jsonify({"success": False, "message": "Security manager not available"}), 500
+    try:
+        data = request.json or {}
+        success, message = security_manager.remove_fail2ban_trusted_network(data.get("value", ""))
+        status = 200 if success else 400
+        return jsonify({"success": success, "message": message}), status
+    except Exception as e:
+        return jsonify({"success": False, "message": str(e)}), 500
+
+
+@security_bp.route('/api/security/fail2ban/trusted-networks', methods=['PUT'])
+@require_auth
+def fail2ban_update_trusted_network():
+    """Replace one user-managed global Fail2Ban IP/CIDR exclusion."""
+    if not security_manager:
+        return jsonify({"success": False, "message": "Security manager not available"}), 500
+    try:
+        data = request.json or {}
+        success, message, value = security_manager.update_fail2ban_trusted_network(
+            data.get("old_value", ""), data.get("new_value", "")
+        )
+        status = 200 if success else 400
+        return jsonify({"success": success, "message": message, "value": value}), status
+    except Exception as e:
+        return jsonify({"success": False, "message": str(e)}), 500
+
+
 @security_bp.route('/api/security/fail2ban/apply-jails', methods=['POST'])
 @require_auth
 def fail2ban_apply_jails():

AppImage/scripts/security_manager.py

@@ -11,6 +11,8 @@ import subprocess
 import re
 import fcntl
 import threading
+import ipaddress
+import tempfile
 from contextlib import contextmanager
 
 # =================================================================
@@ -80,6 +82,10 @@ def _is_pve_rule_line(stripped):
 # and quote/escape tricks. See audit Tier 1 #12b.
 _JAIL_NAME_RE = re.compile(r'^[A-Za-z0-9_][A-Za-z0-9_-]{0,63}$')
 
+FAIL2BAN_TRUSTED_NETWORKS_FILE = "/etc/fail2ban/jail.d/99-proxmenux-ignore.local"
+FAIL2BAN_LEGACY_GLOBAL_FILE = "/etc/fail2ban/jail.local"
+_FAIL2BAN_PROTECTED_NETWORKS = ("127.0.0.0/8", "::1")
+
 # Whitelist for the `level` argument to firewall functions. The audit flagged
 # that an unconstrained value here could one day be extended to `vm` and become
 # a path traversal sink. See audit Tier 1 #12d.
@@ -924,6 +930,192 @@ def classify_ip(ip_address):
     return "external"
 
 
+def _normalise_ip_or_network(value):
+    """Return a canonical IP/CIDR string, or raise ValueError."""
+    if not isinstance(value, str):
+        raise ValueError("Enter an IP address or CIDR network")
+    candidate = value.strip()
+    if not candidate or len(candidate) > 128 or any(ch.isspace() for ch in candidate):
+        raise ValueError("Enter one IP address or CIDR network at a time")
+    try:
+        if "/" in candidate:
+            parsed = ipaddress.ip_network(candidate, strict=False)
+            if parsed.prefixlen == 0 or parsed.is_multicast or parsed.is_unspecified:
+                raise ValueError
+            return parsed.with_prefixlen
+        parsed = ipaddress.ip_address(candidate)
+        if parsed.is_multicast or parsed.is_unspecified:
+            raise ValueError
+        return str(parsed)
+    except ValueError:
+        raise ValueError("Invalid IP address or CIDR network")
+
+
+def _parse_default_ignoreip(path):
+    """Read ignoreip values from the [DEFAULT] section of one config file."""
+    if not os.path.isfile(path):
+        return []
+    values = []
+    in_default = False
+    try:
+        with open(path, "r") as config_file:
+            for raw_line in config_file:
+                stripped = raw_line.strip()
+                if stripped.startswith("[") and stripped.endswith("]"):
+                    in_default = stripped.upper() == "[DEFAULT]"
+                    continue
+                if not in_default or not stripped or stripped.startswith(("#", ";")):
+                    continue
+                match = re.match(r"^ignoreip\s*=\s*(.*)$", stripped, re.IGNORECASE)
+                if match:
+                    raw_values = re.split(r"[\s,]+", match.group(1).strip())
+                    values.extend(value for value in raw_values if value)
+    except OSError:
+        return []
+    return values
+
+
+def _trusted_network_entries():
+    source = (FAIL2BAN_TRUSTED_NETWORKS_FILE
+              if os.path.isfile(FAIL2BAN_TRUSTED_NETWORKS_FILE)
+              else FAIL2BAN_LEGACY_GLOBAL_FILE)
+    entries = []
+    for value in (*_FAIL2BAN_PROTECTED_NETWORKS, *_parse_default_ignoreip(source)):
+        try:
+            normalised = _normalise_ip_or_network(value)
+        except ValueError:
+            continue
+        if normalised not in entries:
+            entries.append(normalised)
+    return entries
+
+
+def get_fail2ban_trusted_networks():
+    """Return the global Fail2Ban IP/CIDR allowlist managed by the Monitor."""
+    protected = {_normalise_ip_or_network(value) for value in _FAIL2BAN_PROTECTED_NETWORKS}
+    return [
+        {"value": value, "protected": value in protected}
+        for value in _trusted_network_entries()
+    ]
+
+
+def _write_trusted_networks(entries):
+    target = FAIL2BAN_TRUSTED_NETWORKS_FILE
+    directory = os.path.dirname(target)
+    os.makedirs(directory, exist_ok=True)
+    content = (
+        "# Managed by ProxMenux Monitor. Use the Security page to edit.\n"
+        "[DEFAULT]\n"
+        f"ignoreip = {' '.join(entries)}\n"
+        "ignoreself = true\n"
+    )
+    temp_path = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w", dir=directory, prefix=".proxmenux-ignore-", delete=False
+        ) as temp_file:
+            temp_file.write(content)
+            temp_path = temp_file.name
+        os.chmod(temp_path, 0o640)
+        os.replace(temp_path, target)
+    finally:
+        if temp_path and os.path.exists(temp_path):
+            os.unlink(temp_path)
+
+
+def _save_trusted_networks(entries):
+    """Persist and reload atomically; restore the previous file on failure."""
+    target = FAIL2BAN_TRUSTED_NETWORKS_FILE
+    lock_path = target + ".lock"
+    with _exclusive_file_lock(lock_path):
+        previous = None
+        existed = os.path.isfile(target)
+        if existed:
+            with open(target, "rb") as current_file:
+                previous = current_file.read()
+
+        _write_trusted_networks(entries)
+        rc, _, err = _run_cmd(["fail2ban-client", "reload"])
+        if rc == 0:
+            return True, "Fail2Ban trusted networks updated"
+
+        try:
+            if existed:
+                with open(target, "wb") as restore_file:
+                    restore_file.write(previous or b"")
+            elif os.path.exists(target):
+                os.unlink(target)
+            _run_cmd(["fail2ban-client", "reload"])
+        except OSError:
+            pass
+        return False, f"Fail2Ban rejected the configuration: {err or 'reload failed'}"
+
+
+def add_fail2ban_trusted_network(value):
+    try:
+        normalised = _normalise_ip_or_network(value)
+    except ValueError as exc:
+        return False, str(exc), None
+
+    entries = _trusted_network_entries()
+    candidate_network = ipaddress.ip_network(normalised, strict=False)
+    if any(
+        candidate_network.version == ipaddress.ip_network(entry, strict=False).version
+        and candidate_network.subnet_of(ipaddress.ip_network(entry, strict=False))
+        for entry in entries
+    ):
+        return False, "This IP address or network is already trusted", normalised
+    entries.append(normalised)
+    success, message = _save_trusted_networks(entries)
+    return success, message, normalised
+
+
+def remove_fail2ban_trusted_network(value):
+    try:
+        normalised = _normalise_ip_or_network(value)
+    except ValueError as exc:
+        return False, str(exc)
+
+    protected = {_normalise_ip_or_network(item) for item in _FAIL2BAN_PROTECTED_NETWORKS}
+    if normalised in protected:
+        return False, "Required local addresses cannot be removed"
+
+    entries = _trusted_network_entries()
+    if normalised not in entries:
+        return False, "Trusted IP address or network was not found"
+    entries.remove(normalised)
+    return _save_trusted_networks(entries)
+
+
+def update_fail2ban_trusted_network(old_value, new_value):
+    try:
+        old_normalised = _normalise_ip_or_network(old_value)
+        new_normalised = _normalise_ip_or_network(new_value)
+    except ValueError as exc:
+        return False, str(exc), None
+
+    protected = {_normalise_ip_or_network(item) for item in _FAIL2BAN_PROTECTED_NETWORKS}
+    if old_normalised in protected:
+        return False, "Required local addresses cannot be changed", None
+
+    entries = _trusted_network_entries()
+    if old_normalised not in entries:
+        return False, "Trusted IP address or network was not found", None
+
+    other_entries = [entry for entry in entries if entry != old_normalised]
+    candidate_network = ipaddress.ip_network(new_normalised, strict=False)
+    if any(
+        candidate_network.version == ipaddress.ip_network(entry, strict=False).version
+        and candidate_network.subnet_of(ipaddress.ip_network(entry, strict=False))
+        for entry in other_entries
+    ):
+        return False, "This IP address or network is already trusted", new_normalised
+
+    entries[entries.index(old_normalised)] = new_normalised
+    success, message = _save_trusted_networks(entries)
+    return success, message, new_normalised
+
+
 def update_jail_config(jail_name, maxretry=None, bantime=None, findtime=None):
     """
     Update Fail2Ban jail configuration (maxretry, bantime, findtime).

AppImage/scripts/tests/test_fail2ban_trusted_networks.py

@@ -0,0 +1,172 @@
+import importlib.util
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+
+MODULE_PATH = Path(__file__).resolve().parents[1] / "security_manager.py"
+SPEC = importlib.util.spec_from_file_location("security_manager_under_test", MODULE_PATH)
+security_manager = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(security_manager)
+
+
+class Fail2BanTrustedNetworksTests(unittest.TestCase):
+    def setUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp_dir.cleanup)
+        root = Path(self.temp_dir.name)
+        self.managed_file = root / "jail.d" / "99-proxmenux-ignore.local"
+        self.legacy_file = root / "jail.local"
+        self.paths_patch = mock.patch.multiple(
+            security_manager,
+            FAIL2BAN_TRUSTED_NETWORKS_FILE=str(self.managed_file),
+            FAIL2BAN_LEGACY_GLOBAL_FILE=str(self.legacy_file),
+        )
+        self.paths_patch.start()
+        self.addCleanup(self.paths_patch.stop)
+        self.command_patch = mock.patch.object(
+            security_manager, "_run_cmd", return_value=(0, "OK", "")
+        )
+        self.run_command = self.command_patch.start()
+        self.addCleanup(self.command_patch.stop)
+
+    def write_legacy(self, content):
+        self.legacy_file.write_text(content)
+
+    def test_reads_and_normalises_existing_global_entries(self):
+        self.write_legacy(
+            "[DEFAULT]\nignoreip = 127.0.0.1/8, ::1 192.168.10.15 10.1.2.9/24\n"
+            "\n[sshd]\nenabled = true\n"
+        )
+
+        entries = security_manager.get_fail2ban_trusted_networks()
+
+        self.assertEqual(
+            [entry["value"] for entry in entries],
+            ["127.0.0.0/8", "::1", "192.168.10.15", "10.1.2.0/24"],
+        )
+        self.assertTrue(entries[0]["protected"])
+        self.assertTrue(entries[1]["protected"])
+        self.assertFalse(entries[2]["protected"])
+
+    def test_adds_ipv4_network_and_reloads_fail2ban(self):
+        success, message, value = security_manager.add_fail2ban_trusted_network(
+            "192.168.50.123/24"
+        )
+
+        self.assertTrue(success, message)
+        self.assertEqual(value, "192.168.50.0/24")
+        content = self.managed_file.read_text()
+        self.assertIn("ignoreip = 127.0.0.0/8 ::1 192.168.50.0/24", content)
+        self.run_command.assert_called_once_with(["fail2ban-client", "reload"])
+
+    def test_adds_ipv6_network(self):
+        success, message, value = security_manager.add_fail2ban_trusted_network(
+            "fd12:3456:789a::42/64"
+        )
+
+        self.assertTrue(success, message)
+        self.assertEqual(value, "fd12:3456:789a::/64")
+
+    def test_rejects_invalid_or_multiple_values_without_writing(self):
+        for value in ("not-an-ip", "10.0.0.1 10.0.0.2", "10.0.0.0/99"):
+            with self.subTest(value=value):
+                success, _, normalised = security_manager.add_fail2ban_trusted_network(value)
+                self.assertFalse(success)
+                self.assertIsNone(normalised)
+
+        self.assertFalse(self.managed_file.exists())
+        self.run_command.assert_not_called()
+
+    def test_duplicate_is_rejected(self):
+        self.write_legacy("[DEFAULT]\nignoreip = 10.0.0.0/24\n")
+
+        success, message, value = security_manager.add_fail2ban_trusted_network("10.0.0.9/24")
+
+        self.assertFalse(success)
+        self.assertIn("already trusted", message)
+        self.assertEqual(value, "10.0.0.0/24")
+        self.run_command.assert_not_called()
+
+    def test_address_already_covered_by_network_is_rejected(self):
+        self.write_legacy("[DEFAULT]\nignoreip = 10.0.0.0/24\n")
+
+        success, message, value = security_manager.add_fail2ban_trusted_network("10.0.0.42")
+
+        self.assertFalse(success)
+        self.assertIn("already trusted", message)
+        self.assertEqual(value, "10.0.0.42")
+
+    def test_rejects_network_that_would_disable_all_bans(self):
+        for value in ("0.0.0.0/0", "::/0"):
+            with self.subTest(value=value):
+                success, _, normalised = security_manager.add_fail2ban_trusted_network(value)
+                self.assertFalse(success)
+                self.assertIsNone(normalised)
+
+    def test_protected_network_cannot_be_removed(self):
+        success, message = security_manager.remove_fail2ban_trusted_network("127.0.0.1/8")
+
+        self.assertFalse(success)
+        self.assertIn("cannot be removed", message)
+        self.run_command.assert_not_called()
+
+    def test_removes_user_network_and_keeps_other_entries(self):
+        self.managed_file.parent.mkdir(parents=True)
+        self.managed_file.write_text(
+            "[DEFAULT]\nignoreip = 127.0.0.0/8 ::1 10.0.0.0/24 192.168.1.5\n"
+        )
+
+        success, message = security_manager.remove_fail2ban_trusted_network("10.0.0.0/24")
+
+        self.assertTrue(success, message)
+        content = self.managed_file.read_text()
+        self.assertNotIn("10.0.0.0/24", content)
+        self.assertIn("192.168.1.5", content)
+
+    def test_updates_user_network_in_place(self):
+        self.managed_file.parent.mkdir(parents=True)
+        self.managed_file.write_text(
+            "[DEFAULT]\nignoreip = 127.0.0.0/8 ::1 10.0.0.0/24 192.168.1.5\n"
+        )
+
+        success, message, value = security_manager.update_fail2ban_trusted_network(
+            "10.0.0.0/24", "10.20.30.99/24"
+        )
+
+        self.assertTrue(success, message)
+        self.assertEqual(value, "10.20.30.0/24")
+        content = self.managed_file.read_text()
+        self.assertNotIn("10.0.0.0/24", content)
+        self.assertIn("10.20.30.0/24", content)
+        self.assertIn("192.168.1.5", content)
+
+    def test_protected_network_cannot_be_updated(self):
+        success, message, value = security_manager.update_fail2ban_trusted_network(
+            "::1", "::2"
+        )
+
+        self.assertFalse(success)
+        self.assertIn("cannot be changed", message)
+        self.assertIsNone(value)
+
+    def test_reload_failure_restores_previous_file(self):
+        self.managed_file.parent.mkdir(parents=True)
+        original = "[DEFAULT]\nignoreip = 127.0.0.0/8 ::1 10.0.0.0/24\n"
+        self.managed_file.write_text(original)
+        self.run_command.side_effect = [
+            (1, "", "configuration error"),
+            (0, "OK", ""),
+        ]
+
+        success, message, _ = security_manager.add_fail2ban_trusted_network("192.168.1.0/24")
+
+        self.assertFalse(success)
+        self.assertIn("configuration error", message)
+        self.assertEqual(self.managed_file.read_text(), original)
+        self.assertEqual(self.run_command.call_count, 2)
+
+
+if __name__ == "__main__":
+    unittest.main()