From 3a7e450b559b3d8edc2f5c3799074b3e14a559af Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 4 Aug 2026 20:06:57 +0200 Subject: [PATCH] feat(fail2ban): manage trusted IP addresses and networks --- AppImage/components/security.tsx | 255 ++++++++++++++++-- AppImage/messages/en/common.json | 33 +++ AppImage/messages/sk/common.json | 33 +++ AppImage/scripts/flask_security_routes.py | 76 ++++++ AppImage/scripts/security_manager.py | 192 +++++++++++++ .../tests/test_fail2ban_trusted_networks.py | 172 ++++++++++++ 6 files changed, 738 insertions(+), 23 deletions(-) create mode 100644 AppImage/scripts/tests/test_fail2ban_trusted_networks.py diff --git a/AppImage/components/security.tsx b/AppImage/components/security.tsx index 9d395b16..b318ef6c 100644 --- a/AppImage/components/security.tsx +++ b/AppImage/components/security.tsx @@ -251,6 +251,15 @@ export function Security() { }) const [f2bSavingConfig, setF2bSavingConfig] = useState(false) const [f2bApplyingJails, setF2bApplyingJails] = useState(false) + const [f2bTrustedNetworks, setF2bTrustedNetworks] = useState>([]) + const [f2bDetectedIp, setF2bDetectedIp] = useState("") + const [f2bTrustedInput, setF2bTrustedInput] = useState("") + const [f2bSavingTrusted, setF2bSavingTrusted] = useState(false) + const [f2bRemovingTrusted, setF2bRemovingTrusted] = useState(null) + const [f2bShowTrustedForm, setF2bShowTrustedForm] = useState(false) + const [f2bEditingTrusted, setF2bEditingTrusted] = useState(null) + const [f2bTrustedEditInput, setF2bTrustedEditInput] = useState("") + const [f2bTrustedNotice, setF2bTrustedNotice] = useState<{type: "success" | "error"; text: string} | null>(null) // SSL/HTTPS state const [sslEnabled, setSslEnabled] = useState(false) @@ -380,9 +389,10 @@ export function Security() { const loadFail2banDetails = async () => { try { setF2bDetailsLoading(true) - const [detailsRes, activityRes] = await Promise.all([ + const [detailsRes, activityRes, trustedRes] = await Promise.all([ fetchApi("/api/security/fail2ban/details"), fetchApi("/api/security/fail2ban/activity"), + fetchApi("/api/security/fail2ban/trusted-networks"), ]) if (detailsRes.success) { setF2bDetails({ @@ -395,6 +405,10 @@ export function Security() { if (activityRes.success) { setF2bActivity(activityRes.events || []) } + if (trustedRes.success) { + setF2bTrustedNetworks(trustedRes.entries || []) + setF2bDetectedIp(trustedRes.detected_ip || "") + } } catch { // Silently fail } finally { @@ -402,6 +416,86 @@ export function Security() { } } + const trustedNetworkError = (message?: string) => { + if (message?.includes("already trusted")) return st("errors.trustedNetworkExists") + if (message?.includes("Invalid IP") || message?.includes("Enter one IP")) return st("errors.invalidTrustedNetwork") + return st("errors.updateTrustedNetworksFailed") + } + + const handleAddTrustedNetwork = async (value?: string) => { + const candidate = (value || f2bTrustedInput).trim() + if (!candidate) { + setF2bTrustedNotice({ type: "error", text: st("errors.invalidTrustedNetwork") }) + return + } + setF2bSavingTrusted(true) + setF2bTrustedNotice(null) + try { + const data = await fetchApi("/api/security/fail2ban/trusted-networks", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ value: candidate }), + }) + if (data.success) { + setF2bTrustedNotice({ type: "success", text: st("messages.trustedNetworkAdded", { value: data.value || candidate }) }) + setF2bTrustedInput("") + setF2bShowTrustedForm(false) + await loadFail2banDetails() + } else { + setF2bTrustedNotice({ type: "error", text: trustedNetworkError(data.message) }) + } + } catch (err) { + setF2bTrustedNotice({ type: "error", text: trustedNetworkError(err instanceof Error ? err.message : undefined) }) + } finally { + setF2bSavingTrusted(false) + } + } + + const handleRemoveTrustedNetwork = async (value: string) => { + setF2bRemovingTrusted(value) + setF2bTrustedNotice(null) + try { + const data = await fetchApi("/api/security/fail2ban/trusted-networks", { + method: "DELETE", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ value }), + }) + if (data.success) { + setF2bTrustedNotice({ type: "success", text: st("messages.trustedNetworkRemoved", { value }) }) + await loadFail2banDetails() + } else { + setF2bTrustedNotice({ type: "error", text: st("errors.updateTrustedNetworksFailed") }) + } + } catch { + setF2bTrustedNotice({ type: "error", text: st("errors.updateTrustedNetworksFailed") }) + } finally { + setF2bRemovingTrusted(null) + } + } + + const handleUpdateTrustedNetwork = async () => { + if (!f2bEditingTrusted || !f2bTrustedEditInput.trim()) return + setF2bSavingTrusted(true) + setF2bTrustedNotice(null) + try { + const data = await fetchApi("/api/security/fail2ban/trusted-networks", { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ old_value: f2bEditingTrusted, new_value: f2bTrustedEditInput.trim() }), + }) + if (data.success) { + setF2bTrustedNotice({ type: "success", text: st("messages.trustedNetworkUpdated", { value: data.value }) }) + setF2bEditingTrusted(null) + setF2bTrustedEditInput("") + await loadFail2banDetails() + } + } catch (err) { + setF2bTrustedNotice({ type: "error", text: trustedNetworkError(err instanceof Error ? err.message : undefined) }) + } finally { + setF2bSavingTrusted(false) + } + } + const handleUnbanIp = async (jail: string, ip: string) => { const key = `${jail}:${ip}` setF2bUnbanning(key) @@ -3350,32 +3444,147 @@ ${(report.sections && report.sections.length > 0) ? ` {fail2banInfo.active && f2bDetails && ( <> - {/* Summary stats - inline */} -
-
- {st("fail2ban.jails")}: - {f2bDetails.jails.length} + {/* Global Fail2Ban allowlist */} +
+
+
+
+ +
+
+

{st("fail2ban.trustedNetworks.title")}

+

{st("fail2ban.trustedNetworks.description")}

+
+
+
-
-
- {st("fail2ban.bannedIps")}: - a + j.currently_banned, 0) > 0 ? "text-red-500" : "text-green-500"}`}> + + {f2bTrustedNotice && ( +
+ {f2bTrustedNotice.text} +
+ )} + + {f2bShowTrustedForm && ( +
+
+ + setF2bTrustedInput(event.target.value)} + onKeyDown={(event) => { if (event.key === "Enter") { event.preventDefault(); handleAddTrustedNetwork() } }} + placeholder={st("fail2ban.trustedNetworks.placeholder")} + className="h-8 font-mono text-xs mt-1" + /> +

{st("fail2ban.trustedNetworks.example")}

+
+
+ {f2bDetectedIp && !f2bTrustedNetworks.some((entry) => entry.value === f2bDetectedIp) ? ( + + ) : } +
+ + +
+
+
+ )} + +
+ {st("fail2ban.trustedNetworks.columns.status")} + {st("fail2ban.trustedNetworks.columns.type")} + {st("fail2ban.trustedNetworks.columns.address")} + {st("fail2ban.trustedNetworks.columns.actions")} +
+ +
+ {f2bTrustedNetworks.map((entry) => ( +
+ {f2bEditingTrusted === entry.value ? ( +
+
+ + setF2bTrustedEditInput(event.target.value)} className="h-8 font-mono text-xs mt-1" /> +
+ + +
+ ) : ( +
+
+ + {entry.protected ? st("fail2ban.trustedNetworks.system") : st("fail2ban.trustedNetworks.trusted")} + +
+
+ + {entry.value.includes("/") ? "CIDR" : "IP"} + +
+ {entry.value} +
+ {!entry.protected ? ( + <> + + + + ) : ( + {st("fail2ban.trustedNetworks.required")} + )} +
+
+ )} +
+ ))} +
+ +
+ +

{st("fail2ban.trustedNetworks.warning")}

+
+
+ + {/* Summary stats */} +
+
+

{f2bDetails.jails.length}

+

{st("fail2ban.jails")}

+
+
a + j.currently_banned, 0) > 0 ? "bg-red-500/5 border-red-500/20" : "bg-muted/50 border-border"}`}> +

a + j.currently_banned, 0) > 0 ? "text-red-500" : "text-foreground"}`}> {f2bDetails.jails.reduce((a, j) => a + j.currently_banned, 0)} - +

+

a + j.currently_banned, 0) > 0 ? "text-red-500/70" : "text-muted-foreground"}`}>{st("fail2ban.bannedIps")}

-
-
- {st("fail2ban.totalBans")}: - - {f2bDetails.jails.reduce((a, j) => a + j.total_banned, 0)} - +
+

{f2bDetails.jails.reduce((a, j) => a + j.total_banned, 0)}

+

{st("fail2ban.totalBans")}

-
-
- {st("fail2ban.failedAttempts")}: - - {f2bDetails.jails.reduce((a, j) => a + j.total_failed, 0)} - +
+

{f2bDetails.jails.reduce((a, j) => a + j.total_failed, 0)}

+

{st("fail2ban.failedAttempts")}

diff --git a/AppImage/messages/en/common.json b/AppImage/messages/en/common.json index 8392a658..f38bab81 100644 --- a/AppImage/messages/en/common.json +++ b/AppImage/messages/en/common.json @@ -1502,6 +1502,9 @@ "auditFailed": "Audit failed", "startAuditFailed": "Failed to start audit", "updateJailConfigFailed": "Failed to save protection settings", + "invalidTrustedNetwork": "Enter one valid IP address or CIDR network.", + "trustedNetworkExists": "This IP address or network is already trusted.", + "updateTrustedNetworksFailed": "Failed to update trusted IP addresses and networks.", "ruleNeedsPortOrSource": "Add a destination port or source address first.", "addRuleFailed": "Failed to add rule", "deleteRuleFailed": "Failed to delete rule", @@ -1537,6 +1540,9 @@ "missingJailsApplied": "Missing protections were added.", "auditCompleted": "Lynis audit completed.", "jailConfigUpdated": "Protection settings were saved.", + "trustedNetworkAdded": "{value} is now trusted by Fail2Ban.", + "trustedNetworkRemoved": "{value} was removed from trusted networks.", + "trustedNetworkUpdated": "Trusted entry was changed to {value}.", "ruleAdded": "Firewall rule was added.", "ruleDeleted": "Firewall rule was deleted.", "ruleUpdated": "Firewall rule was updated.", @@ -1825,6 +1831,33 @@ "maxRetries": "5", "banTime": "3600", "findTime": "600" + }, + "trustedNetworks": { + "title": "Trusted IP addresses and networks", + "description": "Fail2Ban will never block these addresses.", + "required": "Required", + "remove": "Remove from trusted networks", + "placeholder": "192.168.1.25 or 192.168.1.0/24", + "add": "Add", + "addCurrent": "Add my IP ({ip})", + "addRule": "Add address / network", + "addressOrNetwork": "IP address or CIDR network", + "example": "One address: 192.168.1.25 · Entire network: 192.168.1.0/24", + "useCurrent": "Use my current IP: {ip}", + "singleIp": "IP address", + "network": "Network", + "edit": "Edit", + "delete": "Delete", + "save": "Save changes", + "trusted": "Trusted", + "system": "System", + "columns": { + "status": "Status", + "type": "Type", + "address": "Address / network", + "actions": "Actions" + }, + "warning": "Trust only addresses you control. If you trust a whole network, Fail2Ban will also ignore a compromised device in that network." } }, "lynis": { diff --git a/AppImage/messages/sk/common.json b/AppImage/messages/sk/common.json index f2e98787..8eae81cb 100644 --- a/AppImage/messages/sk/common.json +++ b/AppImage/messages/sk/common.json @@ -1502,6 +1502,9 @@ "auditFailed": "Audit zlyhal", "startAuditFailed": "Nepodarilo sa spustiť audit", "updateJailConfigFailed": "Nepodarilo sa uložiť nastavenie ochrany", + "invalidTrustedNetwork": "Zadajte jednu platnú IP adresu alebo sieť v tvare CIDR.", + "trustedNetworkExists": "Táto IP adresa alebo sieť už je medzi dôveryhodnými.", + "updateTrustedNetworksFailed": "Nepodarilo sa upraviť dôveryhodné IP adresy a siete.", "ruleNeedsPortOrSource": "Najprv zadajte cieľový port alebo zdrojovú adresu.", "addRuleFailed": "Nepodarilo sa pridať pravidlo", "deleteRuleFailed": "Nepodarilo sa vymazať pravidlo", @@ -1537,6 +1540,9 @@ "missingJailsApplied": "Chýbajúce ochrany boli doplnené.", "auditCompleted": "Lynis audit je hotový.", "jailConfigUpdated": "Nastavenie ochrany bolo uložené.", + "trustedNetworkAdded": "Fail2Ban už nebude blokovať {value}.", + "trustedNetworkRemoved": "{value} bola odstránená z dôveryhodných sietí.", + "trustedNetworkUpdated": "Dôveryhodná položka bola zmenená na {value}.", "ruleAdded": "Firewall pravidlo bolo pridané.", "ruleDeleted": "Firewall pravidlo bolo vymazané.", "ruleUpdated": "Firewall pravidlo bolo upravené.", @@ -1825,6 +1831,33 @@ "maxRetries": "5", "banTime": "3600", "findTime": "600" + }, + "trustedNetworks": { + "title": "Dôveryhodné IP adresy a siete", + "description": "Tieto adresy Fail2Ban nikdy nezablokuje.", + "required": "Povinné", + "remove": "Odstrániť z dôveryhodných sietí", + "placeholder": "192.168.1.25 alebo 192.168.1.0/24", + "add": "Pridať", + "addCurrent": "Pridať moju IP ({ip})", + "addRule": "Pridať adresu / sieť", + "addressOrNetwork": "IP adresa alebo sieť v tvare CIDR", + "example": "Jedna adresa: 192.168.1.25 · Celá sieť: 192.168.1.0/24", + "useCurrent": "Použiť moju aktuálnu IP: {ip}", + "singleIp": "IP adresa", + "network": "Sieť", + "edit": "Upraviť", + "delete": "Vymazať", + "save": "Uložiť zmeny", + "trusted": "Dôveryhodná", + "system": "Systémová", + "columns": { + "status": "Stav", + "type": "Typ", + "address": "Adresa / sieť", + "actions": "Akcie" + }, + "warning": "Dôverujte iba adresám, ktoré máte pod kontrolou. Pri celej sieti bude Fail2Ban ignorovať aj napadnuté zariadenie v tejto sieti." } }, "lynis": { diff --git a/AppImage/scripts/flask_security_routes.py b/AppImage/scripts/flask_security_routes.py index 821e1e00..4dcbc18c 100644 --- a/AppImage/scripts/flask_security_routes.py +++ b/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(): diff --git a/AppImage/scripts/security_manager.py b/AppImage/scripts/security_manager.py index 94fc1f6d..8fe8a74f 100644 --- a/AppImage/scripts/security_manager.py +++ b/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). diff --git a/AppImage/scripts/tests/test_fail2ban_trusted_networks.py b/AppImage/scripts/tests/test_fail2ban_trusted_networks.py new file mode 100644 index 00000000..5ff7884b --- /dev/null +++ b/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()