#!/usr/bin/env python3 """ SysDeck - Firewall Bridge Helper Author: Jeremy Anderson (https://dcos.net) v0.0.44 PUBLIC-SERVER VARIANTS + SERVICE/PORT EDITOR. Adds three new public-server firewall templates: - remote-admin.sh SSH (22) + Cockpit (9090). For VPS/cloud hosts where the operator needs remote shell + web admin. - public-webserver.sh Caddy (80/443) + Varnish (8080) + MariaDB (3306, loopback-only, defense-in-depth drop). For public web servers with a reverse-proxy/cache stack and a database. - ai-llm.sh Ollama (11434) + OpenWebUI (3000) + Hermes (8000) + Odysseus (8001) + SSH (22). For self-hosted AI LLM stacks on a personal/team workstation. Also adds the service/port editor (4 new bridge subcommands): services detect running listening ports + cross-ref the SERVICES_REGISTRY (ss -tlnp + /proc/net/tcp fallback). Returns the full inventory. service-info show one service's full registry entry + the detected port from its config file. set-service-port edit the port in the service's config file (atomic write: tmpfile + fsync + rename), then systemctl restart the service. Validates the service id against the registry (no arbitrary file edits), validates the port (1..65535), and resolves the config path with os.path.realpath + base-dir allowlist check. restart-service just restart the service (no port change). Useful for "I edited the config by hand" flows. FULL MANAGER, NOT A MONITOR: the bridge can start, stop, restart, apply a template, ban an IP, unban an IP, show the ban list, and show service detection — every operation the panel offers is wired to a subcommand below. v0.0.31 turned the firewall panel into a full manager (apply / stop / restart / ban / unban / clear-bans / detect / check). v0.0.37 UNIFIED BACKEND + EXPANDED CVE HARDENING. The sysdeck-fw backend takes influence from two open-source firewall distributions — Smoothwall Express (RED/ORANGE/GREEN/BLUE color-zone model) and IPFire (source-verified outbound + AirWall isolation + flow offload). We do not ship a template called "smoothwall" or "ipfire" — those are other projects' names. The unified sysdeck-fw template preserves the feature sets we took influence from under our own identifier. v0.0.37 also expands the CVE research to cover the COMMERCIAL web admin UI panels the user meant by "webmin" (cPanel, Plesk, DirectAdmin, CloudPanel, aaPanel, Froxlor, InterWorx, BrainyCP, CyberPanel, HestiaCP, VestaCP, FastPanel, CWP). The full CVE table is in docs/SECURITY-HARDENING.md. New validators added in v0.0.37: - _validate_domain RFC 1035 strict domain regex (CVE-2025-66431 Plesk) - _validate_email parseaddr + charset regex + metachar reject (CVE-2026-26279 Froxlor) - _validate_cron 5-field cron syntax (CVE-2023-53945 BrainyCP) - _validate_mysql_id MySQL identifier (CVE-2026-58048 cPanel) - _sanitize_for_file strips \r\n\0 (CVE-2026-41940 cPanel) - _decode_then_validate decode -> canonicalize -> validate (CVE-2026-29205 cPanel cpdavd) - safe_tar_create tar argument-injection defense using --null -T - (CVE-2025-48702 aaPanel, IWX-CVE-2022-8384 InterWorx) Backends shipped in v0.0.37: custom default; uses the vps-webserver.sh / no-services.sh templates (basic nftables rulesets that work on any Arch/Debian host). Backwards-compatible with v0.0.35. cilium Cilium eBPF datapath. Replaces nftables as the datapath — packets are filtered in BPF programs at XDP and tc ingress/egress before they reach the kernel networking stack. Identity-based policy (not IP-based). L7 policy via Envoy. Requires cilium + cilium-agent. sysdeck-fw SysDeck FW — unified nftables zone firewall. Takes influence from Smoothwall Express (zone matrix) and IPFire (source-verified outbound + AirWall + flow offload) under our own identifier. RED/ORANGE/GREEN/BLUE zone matrix + source-verified outbound + AirWall + flow offload + DMZ forwards. Per user directive v0.0.36: "iptables is old now" — the suite ships only nftables-native and eBPF-native backends. UFW (nftables frontend, no eBPF) is skipped. fwbuilder (GUI rule generator, too complex for the average user) is skipped. Legacy iptables-only firewalls without eBPF integration points are skipped. Subcommands added in v0.0.36: backends list available firewall backends with detected availability (cilium binary present? kernel BPF features? etc.) backend-info show one backend's details + install hint active-backend return the currently selected backend switch-backend switch the active backend (writes /var/lib/sysdeck/firewall/backend and stops the previous backend cleanly) install-backend install the backend's binary deps via the packages module (pacman / apt / dnf) cilium-status cilium status --brief (JSON) cilium-endpoints cilium endpoint list (JSON) cilium-policy cilium policy get (JSON) cilium-policy-apply cilium policy apply cilium-policy-validate cilium policy validate security-hardening return the CVE-derived hardening checklist applied to this bridge (for display in the panel's Security Card) Subcommands: ruleset / chains / templates / template-info / detect / apply / stop / restart / status / ban / unban / banned / clear-bans / check Cockpit way (per user directive): mutating operations run via the bridge's subprocess call to nft / cilium / systemctl / bash — and the JS panel passes { superuser: 'try' } to cockpit.spawn so the cockpit bridge prompts the operator for auth via polkit. No `sudo` shell-out from JS; the polkit action org.sysdeck.firewall.modify (shipped since v0.0.17) authorizes /usr/bin/nft, /usr/sbin/nft, /usr/bin/cilium, /usr/sbin/cilium, /usr/bin/cilium-agent, and /usr/bin/helm. SECURITY HARDENING (v0.0.36) — derived from real CVE disclosures for Webmin, Cockpit, Ajenti, ISPConfig, and Virtualmin. See `docs/SECURITY-HARDENING.md` for the full CVE table. Key changes: - Strict allowlist regex per input type (template names, IP addresses, interface names, backend names, filenames). CVE-2024-2947 lesson. - "--" separator before any user-supplied positional in argv. CVE-2026-4631 lesson. - Environment scrubbing on every privileged subprocess (LD_PRELOAD, LD_LIBRARY_PATH, PYTHONPATH, BASH_ENV, ENV, PERL5OPT all dropped). CVE-2024-6126 lesson. - Path resolution with os.path.realpath + startswith(base_dir) check. CVE-2022-30708 lesson. - No eval / pickle / yaml.unsafe_load. CVE-2019-15642 lesson. - Error responses truncated to 4 KiB and stripped of non-printable bytes. CVE-2022-36446 lesson. Templates are executable shell scripts in /usr/share/sysdeck/firewall/templates/.sh. They implement the following subcommands (the bridge invokes them as `bash [args]`): start build ruleset from detected services, validate, load stop delete the firewall table restart stop + start detect print service detection summary (no rule changes) status print firewall running state + ban lists check validate the ruleset Templates must be POSIX-compliant bash and work on both Arch and Debian. The templates shipped in this release are: vps-webserver.sh (public VPS web tier) no-services.sh (SSH-only hardened host) cilium.sh (Cilium eBPF backend) sysdeck-fw.sh (unified nftables zone firewall) remote-admin.sh (v0.0.44, SSH + Cockpit public-server variant) public-webserver.sh (v0.0.44, Caddy + Varnish + MariaDB variant) ai-llm.sh (v0.0.44, Ollama + OpenWebUI + Hermes + Odysseus variant) Usage: python3 /usr/lib/sysdeck/bridge/firewall.py ruleset python3 /usr/lib/sysdeck/bridge/firewall.py templates python3 /usr/lib/sysdeck/bridge/firewall.py apply vps-webserver python3 /usr/lib/sysdeck/bridge/firewall.py ban 1.2.3.4 python3 /usr/lib/sysdeck/bridge/firewall.py backends python3 /usr/lib/sysdeck/bridge/firewall.py switch-backend cilium python3 /usr/lib/sysdeck/bridge/firewall.py cilium-status """ import ipaddress import json import os import re import shlex import shutil import subprocess import sys from pathlib import Path from typing import Any # ── Constants ──────────────────────────────────────────────────────── TEMPLATES_DIR = Path("/usr/share/sysdeck/firewall/templates") POLICIES_DIR = Path("/usr/share/sysdeck/firewall/policies") STATE_DIR = Path("/var/lib/sysdeck/firewall") ACTIVE_FILE = STATE_DIR / "active" BACKEND_FILE = STATE_DIR / "backend" TABLE_NAME = "firewall" # name of the inet table the templates manage BAN_SETS = ("ssh_abuse", "port_scanners", "connlimit_abuse") # v0.0.36: strict allowlist regexes per input type. # Derived from CVE-2024-2947 (Cockpit sosreport command injection via # crafted filename) and CVE-2026-4631 (Cockpit SSH argv injection). # Reject on first mismatch — do NOT attempt to "sanitize" by stripping # bad chars (CVE-2020-35606 showed that approach is bypassable). TEMPLATE_NAME_RE = re.compile(r"^[a-zA-Z0-9_-]{1,64}$") BACKEND_NAME_RE = re.compile(r"^[a-zA-Z0-9_-]{1,32}$") INTERFACE_NAME_RE = re.compile(r"^[a-zA-Z0-9._-]{1,15}$") # Linux IFNAMSIZ FILENAME_RE = re.compile(r"^[A-Za-z0-9._-]{1,64}$") # v0.0.36: scrubbed environment for privileged subprocesses. # Drops LD_PRELOAD, LD_LIBRARY_PATH, PYTHONPATH, BASH_ENV, ENV, PERL5OPT # (the CVE-2024-6126 lesson — env-var injection via pam_env user_readenv). SCRUBBED_ENV = { "PATH": "/usr/sbin:/usr/bin:/sbin:/bin", "LANG": "C", "LC_ALL": "C", } # v0.0.36: firewall backend registry. Each entry is the static metadata # for the backend. Availability is probed at runtime (in cmd_backends) # so the panel can show "installed" vs "installable". # Per user directive v0.0.36: "iptables is old now" — only nftables-native # and eBPF-native backends ship. UFW, fwbuilder, and legacy iptables-only # firewalls are explicitly excluded (see EXCLUDED_BACKENDS below). FIREWALL_BACKENDS: list[dict[str, Any]] = [ { "id": "custom", "name": "Custom (basic nftables templates)", "description": ( "Default. Uses the vps-webserver.sh and no-services.sh " "templates shipped with sysdeck. Modern nftables syntax " "with sets, verdict maps, synproxy, and eBPF integration " "points. Works on any Arch/Debian host." ), "technology": "nftables", "ebpf": False, "template": None, # uses whatever the operator selects "install_hint": None, "install_packages": [], }, { "id": "cilium", "name": "Cilium (eBPF datapath)", "description": ( "Cilium replaces nftables as the datapath. Packets are " "filtered in BPF programs at XDP and tc ingress/egress " "before they reach the kernel networking stack. Identity-" "based policy (not IP-based). L7 policy via Envoy. " "Requires kernel 5.10+ and the cilium + cilium-agent binaries." ), "technology": "ebpf", "ebpf": True, "template": "cilium", "install_hint": ( "Install on Arch: sudo pacman -S cilium-cli\n" "Install on Debian: sudo apt install cilium-cli\n" "Install via Helm: helm repo add cilium https://helm.cilium.io/\n" " helm install cilium cilium/cilium -n kube-system" ), "install_packages": ["cilium-cli"], }, { "id": "sysdeck-fw", "name": "SysDeck FW (unified nftables zones)", "description": ( "Unified nftables zone firewall. Takes influence from " "Smoothwall Express (RED/ORANGE/GREEN/BLUE color-zone " "model) and IPFire (source-verified outbound + AirWall " "isolation for BLUE/WiFi + flow offload) under our own " "identifier. Optional flow offload for hardware " "acceleration. DMZ port-forwarding. Modern nftables syntax " "with sets, verdict maps, synproxy, and eBPF integration " "points." ), "technology": "nftables", "ebpf": False, "template": "sysdeck-fw", "install_hint": None, "install_packages": [], }, ] # Per user directive v0.0.36: backends explicitly excluded from the # dropdown, with the reason. The panel renders this as a muted info # block beneath the backend selector so the operator understands why # these are missing. EXCLUDED_BACKENDS: list[dict[str, str]] = [ { "id": "ufw", "reason": ( "UFW is a frontend for nftables/iptables — it does not " "use eBPF and adds no value over the custom backend. " "Skipped per user directive v0.0.36." ), }, { "id": "fwbuilder", "reason": ( "fwbuilder is a GUI rule generator — too complex for the " "average user. Skipped per user directive v0.0.36." ), }, { "id": "iptables-legacy", "reason": ( "iptables-legacy is the pre-nftables firewall. iptables " "is old now — the eBPF era has moved past it. Skipped per " "user directive v0.0.36." ), }, { "id": "iptables-nft", "reason": ( "iptables-nft is a compatibility wrapper around nftables. " "The custom backend uses nftables natively — the wrapper " "adds no value. Skipped per user directive v0.0.36." ), }, { "id": "shorewall", "reason": ( "Shorewall is iptables-based and has no eBPF integration " "points. The sysdeck-fw backend covers the zone-firewall " "use case with modern nftables syntax. " "Skipped per user directive v0.0.36." ), }, { "id": "smoothwall", "reason": ( "Smoothwall Express is a separate Linux distribution with " "its own trademark. We took influence from its " "RED/ORANGE/GREEN/BLUE zone model for the sysdeck-fw " "backend — we do not ship a template called \"smoothwall\" " "because we cannot call our rewrite by another project's " "name." ), }, { "id": "ipfire", "reason": ( "IPFire is a separate Linux distribution with its own " "trademark. We took influence from its source-verified " "outbound + AirWall isolation + flow offload for the " "sysdeck-fw backend — we do not ship a template called " "\"ipfire\" because we cannot call our rewrite by another " "project's name." ), }, ] # nft line patterns. Compiled once at import. CHAIN_RE = re.compile(r"^\s*chain\s+(?P\S+)\s*\{") RULE_RE = re.compile(r"^\s*(?P.+?)\s+#\s*handle\s+(?P\d+)") # ── nft invocation ────────────────────────────────────────────────── # # run_nft() captures stdout and stderr. check=False everywhere — the # bridge is invoked by cockpit.spawn which has its own error handling, # and we want to surface nft's stderr in the JSON response rather than # raise an exception the JS panel would have to display as an alert. def _nft(args: list[str], check: bool = False) -> tuple[int, str, str]: """Run `nft ` and return (rc, stdout, stderr). Never raises. v0.0.36 hardening: - env scrubbed (SCRUBBED_ENV) — defeats LD_PRELOAD / PYTHONPATH / etc. (CVE-2024-6126 lesson). - subprocess.run with shell=False and array argv — no shell interpolation (CVE-2019-15107 / CVE-2024-2947 lesson). """ nft_bin = shutil.which("nft") or "/usr/sbin/nft" try: r = subprocess.run( [nft_bin, *args], capture_output=True, text=True, check=False, timeout=30, env=SCRUBBED_ENV, ) return r.returncode, _sanitize_output(r.stdout), _sanitize_output(r.stderr) except (FileNotFoundError, OSError, subprocess.TimeoutExpired) as exc: return 127, "", str(exc) def _nft_ok(args: list[str]) -> bool: """Return True if `nft ` succeeds.""" rc, _, _ = _nft(args) return rc == 0 # ── v0.0.36: input validation helpers ────────────────────────────── # # Strict allowlist validation per input type. Each function returns # True if the input matches the allowlist regex, False otherwise. # The bridge calls these BEFORE the input enters any argv element. # # CVE-2024-2947 lesson: never trust a filename. Validate first, then # pass to subprocess as a separate argv element (never string-interpolate). # CVE-2026-4631 lesson: even with shell=False, an attacker can inject # option flags if user input is in the argv. Insert "--" before any # user-supplied positional. # CVE-2020-35606 lesson: do NOT attempt to "sanitize" by stripping bad # chars — reject on first mismatch. The original Webmin fix stripped # newlines but %0A / %0C still bypassed it. def _validate_template_name(name: str) -> bool: """Return True if name is a valid template identifier.""" if not name or len(name) > 64: return False return bool(TEMPLATE_NAME_RE.match(name)) def _validate_backend_name(name: str) -> bool: """Return True if name is a valid backend identifier.""" if not name or len(name) > 32: return False return bool(BACKEND_NAME_RE.match(name)) def _validate_interface(name: str) -> bool: """Return True if name is a valid Linux interface name (IFNAMSIZ).""" if not name or len(name) > 15: return False return bool(INTERFACE_NAME_RE.match(name)) def _validate_filename(name: str) -> bool: """Return True if name is a safe filename (no path separators). Used for any user-supplied filename that will enter argv. Rejects `..`, `/`, NUL, shell metachars, and anything outside printable ASCII. CVE-2024-2947 lesson. """ if not name or len(name) > 64: return False return bool(FILENAME_RE.match(name)) def _validate_ip(ip: str) -> bool: """Return True if ip is a valid IPv4 or IPv6 address. Uses ipaddress.ip_address for strict validation — rejects hostnames, truncated octets, leading zeros, etc. CVE-2024-2947 lesson: even IPs should be strictly validated before entering argv. """ if not ip or len(ip) > 45: # IPv6 max is 45 chars return False try: ipaddress.ip_address(ip) return True except ValueError: return False def _sanitize_output(text: str, max_len: int = 4096) -> str: """Truncate and strip non-printable bytes from command output. CVE-2022-36446 lesson: command output that includes attacker-controlled bytes (e.g. an nft error message quoting the offending rule) must not be passed verbatim to the JS panel. We truncate to 4 KiB and replace non-printable bytes with a space. """ if not text: return "" if len(text) > max_len: text = text[:max_len] + " ... (truncated)" # Replace any byte outside printable ASCII + tab/newline/cr with space. return "".join(c if (32 <= ord(c) < 127 or c in "\t\n\r") else " " for c in text) def _resolve_path_under_base(path_str: str, base_dir: Path) -> Path | None: """Resolve path_str and verify it lives under base_dir. Uses os.path.realpath to defeat symlink chains, then verifies the real path starts with base_dir. CVE-2022-30708 lesson. Returns the resolved Path on success, None on rejection. """ if not path_str or ".." in Path(path_str).parts: return None try: real = Path(os.path.realpath(path_str)) except (OSError, ValueError): return None try: real.relative_to(base_dir) except ValueError: return None return real # ── v0.0.37 validators (from commercial-web-panel CVE research) ──── # # These validators cover the new threat models exposed by cPanel, Plesk, # CyberPanel, aaPanel, CloudPanel, HestiaCP, VestaCP, Froxlor, InterWorx, # BrainyCP, DirectAdmin, and CWP. See docs/SECURITY-HARDENING.md §2.6+ # for the full checklist. # RFC 1035 domain name regex. Rejects domains containing shell # metacharacters, path separators, or any byte outside the LDH # (letters/digits/hyphens) set + dots. DOMAIN_RE = re.compile( r"^(?=.{1,253}$)" r"([a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)" r"(\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)*$" ) # Email regex (after parseaddr). Permits letters, digits, and _%+-. in # the local part; letters, digits, and -. in the domain. EMAIL_RE = re.compile(r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$") # 5-field cron schedule regex (minute hour day month weekday). # Each field allows: digits, *, /, -, comma. Rejects everything else. # Per CVE-2023-53945 (BrainyCP) — the cron *command* must NEVER be # user-supplied, but the schedule must also be validated. CRON_FIELD_RE = re.compile(r"^[0-9*/,-]+$") CRON_SCHEDULE_RE = re.compile(r"^[0-9*/,-]+ +[0-9*/,-]+ +[0-9*/,-]+ +[0-9*/,-]+ +[0-9*/,-]+$") # MySQL identifier regex. Per CVE-2026-58048 (cPanel) — identifiers # must be backtick-quoted AND reject embedded backticks. The regex # permits letters, digits, underscore, and $ (MySQL allows $); max 64 # chars (MySQL hard limit); must not start with a digit. MYSQL_ID_RE = re.compile(r"^[A-Za-z_$][A-Za-z0-9_$]{0,63}$") # MySQL reserved words that must never be used as identifiers. MYSQL_RESERVED = frozenset({ "mysql", "information_schema", "performance_schema", "sys", "root", "admin", "test", "tmp", "temp", "database", "table", "column", "index", "key", "primary", "foreign", "references", "constraint", "default", "null", "not", "true", "false", "select", "insert", "update", "delete", "create", "drop", "alter", "rename", "grant", "revoke", "user", "password", "host", "db", }) def _validate_domain(domain: str) -> bool: """Return True if domain is a valid RFC 1035 domain name. CVE-2025-66431 (Plesk domain-creation RCE-as-root) lesson: domain names flow into root-run scripts (log symlink rotation, vhost config, nginx/apache conf). Validate strictly BEFORE any root-run helper sees them. Rejects: - Shell metacharacters (; | & $ ` ( ) < > \\ ' ") - Path separators (/ \\) - Whitespace - .. or leading/trailing hyphen - Domains > 253 chars or labels > 63 chars - IDN (must be punycode-encoded by the caller first) - Wildcard domains (operator must opt-in separately) """ if not domain or len(domain) > 253: return False if any(c in domain for c in "/\\;|&$`()<>\"' \t\n\r\0"): return False if ".." in domain: return False return bool(DOMAIN_RE.match(domain)) def _validate_email(addr_str: str) -> bool: """Return True if addr_str is a valid email address. CVE-2026-26279 (Froxlor) lesson: Froxlor's email-input validation had a logic bug that disabled format checking for fields declared as email type, allowing shell metacharacters through. We use parseaddr FIRST (catches most malformed addresses), then a strict charset regex, then SEPARATELY reject shell metacharacters — defense in depth on top of the regex. Rejects: - Anything parseaddr rejects - Shell metacharacters even if the regex would accept them - Length > 254 chars (RFC 5321) """ if not addr_str or len(addr_str) > 254: return False # parseaddr returns (realname, email_address); we want the address. # NOTE: the parameter is named addr_str (not email) to avoid # shadowing the email.utils import. from email.utils import parseaddr _, addr = parseaddr(addr_str) if not addr or "@" not in addr: return False if not EMAIL_RE.match(addr): return False # Defense in depth: reject shell metacharacters even if regex passes. if any(c in addr for c in ";|&$`()<>!{}\n\r\0"): return False return True def _validate_cron_schedule(schedule: str) -> bool: """Return True if schedule is a valid 5-field cron schedule. CVE-2023-53945 (BrainyCP) lesson: BrainyCP let users inject arbitrary commands through the crontab interface. The cron *schedule* must be validated as 5-field syntax only; the cron *command* must NEVER be user-supplied. This validator checks the SCHEDULE only. The command is a separate concern — the bridge only accepts a command from a pre-defined allowlist, never from operator input. """ if not schedule or len(schedule) > 200: return False return bool(CRON_SCHEDULE_RE.match(schedule)) def _validate_mysql_identifier(identifier: str) -> bool: """Return True if identifier is a valid MySQL identifier. CVE-2026-58048 (cPanel) lesson: cPanel's DB rename dropped SQL mode restrictions, allowing the user to run SQL in root context. Identifiers must be strictly validated, backtick-quoted, AND reject MySQL reserved words. """ if not identifier or len(identifier) > 64: return False if not MYSQL_ID_RE.match(identifier): return False if identifier.lower() in MYSQL_RESERVED: return False # Reject embedded backticks (defeats backtick-quote escape attacks). if "`" in identifier: return False return True def _sanitize_for_file(value: str) -> str: """Strip \\r, \\n, \\0 from a value before writing to a line-oriented file. CVE-2026-41940 (cPanel session-file CRLF injection) lesson: any value written to a file that is later parsed line-by-line (session files, polkit action files, sudoers fragments, cron files, /etc/hosts, DNS zone files, nginx/apache conf) must have CR/LF/NUL STRIPPED, not just rejected. An attacker who can inject "\\r\\nuser=root\\r\\n" into a session file gains root. """ if not value: return "" return value.translate({0x0d: None, 0x0a: None, 0x00: None}) def _decode_then_validate(encoded: str, validator_fn, *, allow_pct: bool = False) -> bool: """URL-decode + canonicalize + validate a value. CVE-2026-29205 (cPanel cpdavd) lesson: cPanel's regex validated the ENCODED URI form (where %2F satisfies [^/]+), then decoded it into a real / — enabling path traversal. The fix is to decode FIRST, then canonicalize, then validate. This helper: 1. URL-decodes the input (urllib.parse.unquote). 2. If the decoded form differs from the encoded form in a security-relevant way (contains %2e, %2f, %5c, %00), rejects. 3. Calls validator_fn on the decoded form. The `allow_pct` flag is for the rare case where a literal % is expected in the value (e.g. a SQL LIKE pattern); it disables the security-relevant-difference check. """ if not encoded: return False import urllib.parse decoded = urllib.parse.unquote(encoded) if not allow_pct: # Reject if the encoded form contained %-encoded path traversal # or NUL bytes — defense in depth on top of the decoded-form # validator. lower = encoded.lower() for seq in ("%2e", "%2f", "%5c", "%00", "%0a", "%0d"): if seq in lower: return False return validator_fn(decoded) def safe_tar_create(archive_path: Path, files: list[Path], cwd: Path) -> tuple[int, str, str]: """Create a tar.gz archive safely, defeating argument injection. CVE-2025-48702 (aaPanel) + IWX-CVE-2022-8384 (InterWorx) lesson: subprocess.run with shell=False + a "--" separator is NECESSARY but NOT SUFFICIENT for tar/zip/find/rsync. These tools interpret arguments after "--" differently, and a filename like "--checkpoint-action=exec=bash shell.sh" can still execute code. The defense is to keep filenames OUT of argv entirely by passing them via stdin using tar's --null -T - mode (read NUL-delimited filenames from stdin). Returns (rc, stdout, stderr). Never raises. """ if not archive_path or not files or not cwd: return 1, "", "invalid arguments" # Validate archive_path and cwd with the v0.0.36 path rules. if not _validate_filename(archive_path.name): return 1, "", f"invalid archive name: {archive_path.name!r}" # Validate each file: reject names starting with - or /, containing # newlines, or outside the cwd. safe_files: list[Path] = [] for f in files: name = f.name if name.startswith("-") or name.startswith("/"): return 1, "", f"unsafe filename (starts with - or /): {name!r}" if any(c in name for c in "\n\r\0"): return 1, "", f"unsafe filename (contains control chars): {name!r}" # Resolve and verify under cwd. try: real = Path(os.path.realpath(f)) real.relative_to(Path(os.path.realpath(cwd))) except (ValueError, OSError): return 1, "", f"file escapes cwd: {f!r}" safe_files.append(real) # Build the NUL-delimited manifest. manifest = b"\0".join(str(f.relative_to(Path(os.path.realpath(cwd)))).encode() for f in safe_files) + b"\0" tar_bin = shutil.which("tar") or "/usr/bin/tar" try: r = subprocess.run( [tar_bin, "--null", "-czf", str(archive_path), "-T", "-"], input=manifest, capture_output=True, check=False, timeout=300, env=SCRUBBED_ENV, cwd=str(cwd), ) return r.returncode, _sanitize_output(r.stdout or ""), _sanitize_output(r.stderr or "") except (FileNotFoundError, OSError, subprocess.TimeoutExpired) as exc: return 127, "", str(exc) # ── Ruleset parser ──────────────────────────────────────────────── def parse_ruleset(output: str) -> list[dict[str, Any]]: """Parse `nft list ruleset` output into structured rules.""" rules: list[dict[str, Any]] = [] current_chain: str | None = None for line in output.splitlines(): chain_match = CHAIN_RE.match(line) if chain_match: current_chain = chain_match.group("name") continue if line.strip() == "}": current_chain = None continue rule_match = RULE_RE.match(line) if rule_match and current_chain: rules.append({ "chain": current_chain, "spec": rule_match.group("spec").strip(), "handle": int(rule_match.group("handle")), }) return rules def list_chains(output: str) -> list[str]: """Extract chain names from the ruleset output.""" return [ m.group("name") for line in output.splitlines() if (m := CHAIN_RE.match(line)) ] # ── Template enumeration ─────────────────────────────────────────── # # A template is a *.sh file in TEMPLATES_DIR. We discover them by glob # and extract metadata from a comment block at the top of the file: # # # Name: vps-webserver # # Description: Service-aware firewall for VPS web servers # # Distro: arch,debian # # Services: ssh,caddy,varnish,forgejo # # If the comment block is absent, we fall back to filename-stem and # a generic description. This makes it trivial for operators to drop # a new template into TEMPLATES_DIR and have it appear in the panel. _TEMPLATE_FIELD_RE = re.compile( r"^\s*#\s*(?PName|Description|Distro|Services)\s*:\s*(?P.+)$", re.IGNORECASE, ) def _parse_template_metadata(path: Path) -> dict[str, Any]: """Extract metadata from the comment header of a template file. Looks for `# Key: value` lines in the first 60 lines of the file. Returns a dict with name, description, distros[], services[]. """ meta: dict[str, Any] = { "name": path.stem, "path": str(path), "description": "", "distros": [], "services": [], } try: with path.open(encoding="utf-8", errors="replace") as fh: for i, line in enumerate(fh): if i >= 60: break m = _TEMPLATE_FIELD_RE.match(line) if not m: continue key = m.group("key").lower() val = m.group("val").strip() if key == "name": meta["name"] = val elif key == "description": meta["description"] = val elif key == "distro": meta["distros"] = [v.strip() for v in val.split(",") if v.strip()] elif key == "services": meta["services"] = [v.strip() for v in val.split(",") if v.strip()] except (OSError, PermissionError): pass # Auto-derive services from filename if not declared in header. if not meta["services"]: if "vps" in path.stem or "webserver" in path.stem: meta["services"] = ["ssh", "caddy", "varnish", "forgejo"] elif "no-services" in path.stem: meta["services"] = ["ssh"] if not meta["description"]: meta["description"] = f"Firewall template: {meta['name']}" return meta def list_templates() -> list[dict[str, Any]]: """List all *.sh templates in TEMPLATES_DIR. Returns a list of {name, path, description, distros, services} dicts sorted by name. Returns [] if TEMPLATES_DIR doesn't exist (e.g. the package wasn't installed correctly). """ if not TEMPLATES_DIR.is_dir(): return [] templates: list[dict[str, Any]] = [] for p in sorted(TEMPLATES_DIR.glob("*.sh")): if not p.is_file(): continue if not os.access(p, os.X_OK | os.R_OK): continue meta = _parse_template_metadata(p) templates.append(meta) return templates def template_info(name: str) -> dict[str, Any] | None: """Return metadata for a single template by name (stem or filename). Accepts 'vps-webserver' or 'vps-webserver.sh'. Returns None if not found. v0.0.36 hardening: validates name against TEMPLATE_NAME_RE before resolving the path. CVE-2024-2947 lesson — never trust a filename. """ if not name: return None stem = name.removesuffix(".sh") if not _validate_template_name(stem): return None candidate = TEMPLATES_DIR / f"{stem}.sh" # Resolve and verify it's under TEMPLATES_DIR (symlink defense). resolved = _resolve_path_under_base(str(candidate), TEMPLATES_DIR) if resolved is None or not resolved.is_file(): return None return _parse_template_metadata(resolved) # ── Active template tracking ─────────────────────────────────────── # # When the operator applies a template, we write its name to # /var/lib/sysdeck/firewall/active. detect / status / restart use # this to know which template to invoke. If the file is absent (e.g. # the operator applied a ruleset by hand outside the panel), detect # and status fall back to the first installed template. def _read_active_template() -> str | None: """Return the name of the currently active template, or None.""" try: return ACTIVE_FILE.read_text(encoding="utf-8").strip() or None except (FileNotFoundError, PermissionError, OSError): return None def _write_active_template(name: str | None) -> None: """Record the active template name, or clear it if name is None.""" try: STATE_DIR.mkdir(parents=True, exist_ok=True) if name is None: ACTIVE_FILE.unlink(missing_ok=True) else: ACTIVE_FILE.write_text(name, encoding="utf-8") except (PermissionError, OSError): # Not fatal — the bridge runs as the cockpit user, and STATE_DIR # may need root. The cockpit superuser channel handles this for # apply/stop; for read-only detect/status we just don't track. pass def _resolve_template_path(name: str | None) -> Path | None: """Resolve a template name to a file path. If name is given, look it up. If name is None, use the active template; if no active template, use the first installed template. v0.0.36 hardening: validates name against TEMPLATE_NAME_RE and resolves under TEMPLATES_DIR (symlink defense). CVE-2024-2947 + CVE-2022-30708 lessons. """ if name: stem = name.removesuffix(".sh") if not _validate_template_name(stem): return None candidate = TEMPLATES_DIR / f"{stem}.sh" resolved = _resolve_path_under_base(str(candidate), TEMPLATES_DIR) return resolved if (resolved and resolved.is_file()) else None active = _read_active_template() if active: stem = active.removesuffix(".sh") if _validate_template_name(stem): candidate = TEMPLATES_DIR / f"{stem}.sh" resolved = _resolve_path_under_base(str(candidate), TEMPLATES_DIR) if resolved and resolved.is_file(): return resolved templates = list_templates() if templates: return Path(templates[0]["path"]) return None # ── Template invocation ──────────────────────────────────────────── # # Templates are bash scripts. We invoke them with `bash ` # rather than executing them directly — this avoids the executable-bit # requirement at runtime (though we still install them 0755 so they # can be run directly for debugging). def _run_template(name: str | None, action: str, extra_args: list[str] | None = None) -> tuple[int, str, str]: """Run `