mteehan/vacuum-wall
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix htmx refactor route mismatches and remaining TODO items

Browse Source 
- wireguard: POST /peers with JSON encoding (was /add-peer)
- rules: delete by rule_id in URL path (was JSON body); pass rule objects with id from server; add hx-disable to initial render
- nat: port forward delete uses URL path params to match blueprint
- nat: masquerade toggle uses native hx-post/hx-vals (was inline fetch)
- app.js renderers updated to use URL path deletes for rules and forwards
- remove TODO.md
This commit is contained in:
Mike Teehan
2026-05-17 01:15:52 +00:00
parent 0e7090a2cb
commit 37039351be
26 changed files with 1737 additions and 848 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/firewall.py
+248 -119
View File
@@ -9,12 +9,16 @@ Flask UI can inspect or restore previous configurations.
"""
import json
import logging
import os
import subprocess
from contextlib import suppress
from datetime import UTC
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from uuid import uuid4
logger = logging.getLogger(__name__)
PROJECT_DIR = Path(__file__).resolve().parent.parent
DATA_DIR: str = str(PROJECT_DIR / "data" / "firewall")
@@ -45,7 +49,12 @@ def _run(cmd: list[str], check: bool = True) -> str:
def _reload() -> None:
"""Reload firewalld so permanent changes take effect immediately."""
_run(["sudo", "firewall-cmd", "--reload"])
try:
_run(["sudo", "firewall-cmd", "--reload"])
logger.info("firewalld reloaded")
except RuntimeError as exc:
logger.error("firewalld reload failed: %s", exc)
raise
def _ensure_data_dir() -> None:
@@ -54,6 +63,11 @@ def _ensure_data_dir() -> None:
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
def _gen_id() -> str:
"""Generate a short unique identifier (8 hex characters)."""
return uuid4().hex[:8]
# ---------------------------------------------------------------------------
# Read-only queries
# ---------------------------------------------------------------------------
@@ -66,15 +80,7 @@ def get_available_zones() -> list[str]:
def get_active_zones() -> dict[str, list[str]]:
"""Return a dict mapping active zone names to their assigned interfaces.
Example return value::
{
"public": ["eth0"],
"internal": ["eth1"],
}
"""
"""Return a dict mapping active zone names to their assigned interfaces."""
output = _run(["sudo", "firewall-cmd", "--get-active-zones"])
zones: dict[str, list[str]] = {}
current_zone: str | None = None
@@ -82,7 +88,6 @@ def get_active_zones() -> dict[str, list[str]]:
stripped = raw_line.strip()
if not stripped:
continue
# Indented lines belong to the current zone section.
if raw_line.startswith(" "):
current_ifaces = (
zones[current_zone]
@@ -99,13 +104,7 @@ def get_active_zones() -> dict[str, list[str]]:
def get_zone_info(zone: str) -> dict[str, Any]:
"""Return detailed information for *zone*.
Keys in the returned dict include:
``name``, ``target``, ``interfaces``, ``sources``, ``services``,
``ports``, ``protocols``, ``forward-ports``, ``masquerade``,
``rich-rules``, ``ics``, ``icmp-blocks``, ``module``.
"""
"""Return detailed information for *zone*."""
output = _run(["sudo", "firewall-cmd", f"--zone={zone}", "--list-all"])
info: dict[str, Any] = {"name": zone}
for line in output.splitlines():
@@ -117,7 +116,6 @@ def get_zone_info(zone: str) -> dict[str, Any]:
value = value.strip()
if not value:
# Lines like "interfaces: " or "masquerade: " when disabled
if key in ("masquerade", "ics"):
info[key] = False
else:
@@ -138,12 +136,10 @@ def get_zone_info(zone: str) -> dict[str, Any]:
elif key in ("masquerade", "ics"):
info[key] = value.lower() == "yes"
elif key == "rich-rules":
# rich-rules can span multiple lines; we'll parse below.
info[key] = [value] if value else []
else:
info[key] = value
# rich-rules may already have been set; if not, default to empty.
info.setdefault("rich-rules", [])
info.setdefault("interfaces", [])
info.setdefault("sources", [])
@@ -177,7 +173,6 @@ def get_interfaces() -> list[str]:
ifaces: list[str] = []
for line in output.splitlines():
if line:
# Format: "NUM: NAME: <FLAGS> ..."
parts = line.split()
if len(parts) >= 2:
name = parts[1].rstrip(":")
@@ -212,14 +207,7 @@ def get_rich_rules(zone: str) -> list[str]:
def create_zone(zone: str, target: str = "default") -> None:
"""Create a new permanent zone in firewalld.
Args:
zone: Name of the zone to create.
Raises:
RuntimeError: If the zone already exists or creation fails.
"""
"""Create a new permanent zone in firewalld."""
_run(
[
"sudo",
@@ -230,16 +218,14 @@ def create_zone(zone: str, target: str = "default") -> None:
]
)
_reload()
logger.info("Firewall zone '%s' created (target=%s)", zone, target)
def delete_zone(zone: str) -> None:
"""Delete an existing zone.
Raises:
RuntimeError: If the zone does not exist or the deletion fails.
"""
"""Delete an existing zone."""
_run(["sudo", "firewall-cmd", f"--zone={zone}", "--delete", "--permanent"])
_reload()
logger.info("Firewall zone '%s' deleted", zone)
# ---------------------------------------------------------------------------
@@ -248,12 +234,7 @@ def delete_zone(zone: str) -> None:
def set_zone_interfaces(zone: str, interfaces: list[str]) -> None:
"""Assign *interfaces* to *zone*, replacing any existing assignments.
Existing interfaces on the zone are removed first so only the
provided list remains.
"""
# Remove current permanent interfaces for this zone.
"""Assign *interfaces* to *zone*, replacing any existing assignments."""
try:
current = get_zone_info(zone).get("interfaces", [])
except Exception:
@@ -270,7 +251,6 @@ def set_zone_interfaces(zone: str, interfaces: list[str]) -> None:
check=False,
)
# Add the desired set.
for iface in interfaces:
_run(
[
@@ -282,6 +262,7 @@ def set_zone_interfaces(zone: str, interfaces: list[str]) -> None:
]
)
_reload()
logger.info("Zone '%s' interfaces set to %s", zone, interfaces)
def add_zone_interface(zone: str, iface: str) -> None:
@@ -296,6 +277,7 @@ def add_zone_interface(zone: str, iface: str) -> None:
]
)
_reload()
logger.info("Interface '%s' added to zone '%s'", iface, zone)
def remove_zone_interface(zone: str, iface: str) -> None:
@@ -310,6 +292,7 @@ def remove_zone_interface(zone: str, iface: str) -> None:
]
)
_reload()
logger.info("Interface '%s' removed from zone '%s'", iface, zone)
# ---------------------------------------------------------------------------
@@ -319,7 +302,6 @@ def remove_zone_interface(zone: str, iface: str) -> None:
def set_zone_services(zone: str, services: list[str]) -> None:
"""Set services for *zone*, replacing any previously allowed services."""
# Remove all current services.
current = get_zone_info(zone).get("services", [])
for svc in current:
_run(
@@ -344,6 +326,7 @@ def set_zone_services(zone: str, services: list[str]) -> None:
]
)
_reload()
logger.info("Zone '%s' services set to %s", zone, services)
def add_zone_service(zone: str, service: str) -> None:
@@ -358,6 +341,7 @@ def add_zone_service(zone: str, service: str) -> None:
]
)
_reload()
logger.info("Service '%s' added to zone '%s'", service, zone)
def remove_zone_service(zone: str, service: str) -> None:
@@ -372,6 +356,7 @@ def remove_zone_service(zone: str, service: str) -> None:
]
)
_reload()
logger.info("Service '%s' removed from zone '%s'", service, zone)
# ---------------------------------------------------------------------------
@@ -379,12 +364,8 @@ def remove_zone_service(zone: str, service: str) -> None:
# ---------------------------------------------------------------------------
def add_rich_rule(zone: str, rule: str) -> None:
"""Add a rich rule to *zone*.
The *rule* argument should be a fully-formed rich-rule expression,
e.g. ``rule family="ipv4" port protocol="tcp" port="443" accept``.
"""
def add_rich_rule(zone: str, rule: str) -> dict[str, Any]:
"""Add a rich rule to *zone* and persist to declarative config."""
_run(
[
"sudo",
@@ -395,13 +376,14 @@ def add_rich_rule(zone: str, rule: str) -> None:
]
)
_reload()
_persist_rich_rule(zone, rule)
rule_entry = _get_rich_rule_entry(zone, rule)
logger.info("Rich rule added to zone '%s': %s", zone, rule[:80])
return rule_entry
def remove_rich_rule(zone: str, rule: str) -> None:
"""Remove a rich rule from *zone*.
The rule string must match exactly what was added.
"""
"""Remove a rich rule from *zone*."""
_run(
[
"sudo",
@@ -412,6 +394,55 @@ def remove_rich_rule(zone: str, rule: str) -> None:
]
)
_reload()
_unpersist_rich_rule(zone, rule)
logger.info("Rich rule removed from zone '%s': %s", zone, rule[:80])
def _persist_rich_rule(zone: str, rule: str) -> dict[str, Any]:
"""Add a rich rule to the declarative config with a generated id."""
cfg = config_get()
cfg.setdefault("zones", {})
cfg["zones"].setdefault(zone, {})
cfg["zones"][zone].setdefault("rich_rules", [])
existing_rules = cfg["zones"][zone]["rich_rules"]
rule_id = _gen_id()
entry = {"id": rule_id, "rule": rule}
existing_rules.append(entry)
config_set(cfg)
return entry
def _unpersist_rich_rule(zone: str, rule: str) -> None:
"""Remove a rich rule from the declarative config by rule string."""
cfg = config_get()
zone_cfg = cfg.get("zones", {}).get(zone, {})
rules = zone_cfg.get("rich_rules", [])
zone_cfg["rich_rules"] = [r for r in rules if r.get("rule") != rule]
config_set(cfg)
def _get_rich_rule_entry(zone: str, rule: str) -> dict[str, Any]:
"""Look up a rich rule entry in the declarative config."""
cfg = config_get()
for r in cfg.get("zones", {}).get(zone, {}).get("rich_rules", []):
if r.get("rule") == rule:
return r
return {"rule": rule}
def remove_rich_rule_by_id(zone: str, rule_id: str) -> None:
"""Remove a rich rule from *zone* by its config id."""
cfg = config_get()
zone_cfg = cfg.get("zones", {}).get(zone, {})
entry = None
for r in zone_cfg.get("rich_rules", []):
if r.get("id") == rule_id:
entry = r
break
if entry is None:
raise ValueError(f"Rich rule '{rule_id}' not found in zone '{zone}'")
rule = entry["rule"]
remove_rich_rule(zone, rule)
# ---------------------------------------------------------------------------
@@ -424,6 +455,7 @@ def set_masquerade(zone: str, enable: bool) -> None:
action = "--add-masquerade" if enable else "--remove-masquerade"
_run(["sudo", "firewall-cmd", f"--zone={zone}", action, "--permanent"])
_reload()
logger.info("Masquerade %s on zone '%s'", "enabled" if enable else "disabled", zone)
# ---------------------------------------------------------------------------
@@ -437,12 +469,8 @@ def add_forward_port(
protocol: str,
toaddr: str | None = None,
toport: int | None = None,
) -> None:
"""Add a port forwarding rule to *zone*.
Forward traffic arriving on ``port/protocol`` to
``toaddr:toport`` (or just ``toport`` when *toaddr* is omitted).
"""
) -> dict[str, Any]:
"""Add a port forwarding rule to *zone* and persist to declarative config."""
fwd = f"port={port}/proto={protocol}"
if toaddr and toport:
fwd += f"/toaddr={toaddr}/toport={toport}"
@@ -461,6 +489,10 @@ def add_forward_port(
]
)
_reload()
_persist_forward_port(zone, port, protocol, toaddr, toport)
fp_entry = _get_forward_port_entry(zone, port, protocol)
logger.info("Port forward added to zone '%s': %s", zone, fwd)
return fp_entry
def remove_forward_port(
@@ -470,10 +502,7 @@ def remove_forward_port(
toaddr: str | None = None,
toport: int | None = None,
) -> None:
"""Remove a previously added port-forwarding rule from *zone*.
All parameters must match the original rule exactly.
"""
"""Remove a previously added port-forwarding rule from *zone*."""
fwd = f"port={port}/proto={protocol}"
if toaddr and toport:
fwd += f"/toaddr={toaddr}/toport={toport}"
@@ -492,6 +521,80 @@ def remove_forward_port(
]
)
_reload()
_unpersist_forward_port(zone, port, protocol)
logger.info("Port forward removed from zone '%s': %s", zone, fwd)
def _persist_forward_port(
zone: str,
port: int,
protocol: str,
toaddr: str | None = None,
toport: int | None = None,
) -> dict[str, Any]:
"""Add a forward port to the declarative config with a generated id."""
cfg = config_get()
cfg.setdefault("zones", {})
cfg["zones"].setdefault(zone, {})
cfg["zones"][zone].setdefault("forward_ports", [])
fp_id = _gen_id()
entry: dict[str, Any] = {
"id": fp_id,
"port": port,
"proto": protocol,
}
if toaddr:
entry["toaddr"] = toaddr
if toport:
entry["toport"] = toport
cfg["zones"][zone]["forward_ports"].append(entry)
config_set(cfg)
return entry
def _unpersist_forward_port(zone: str, port: int, protocol: str) -> None:
"""Remove a forward port from the declarative config by port+proto."""
cfg = config_get()
fps = cfg.get("zones", {}).get(zone, {}).get("forward_ports", [])
cfg.setdefault("zones", {}).setdefault(zone, {})
cfg["zones"][zone]["forward_ports"] = [
fp for fp in fps if not (fp.get("port") == port and fp.get("proto") == protocol)
]
config_set(cfg)
def _get_forward_port_entry(
zone: str, port: int, protocol: str
) -> dict[str, Any]:
"""Look up a forward port entry in the declarative config."""
cfg = config_get()
for fp in cfg.get("zones", {}).get(zone, {}).get("forward_ports", []):
if fp.get("port") == port and fp.get("proto") == protocol:
return fp
entry: dict[str, Any] = {"port": port, "proto": protocol}
return entry
def remove_forward_port_by_id(zone: str, port: int, protocol: str) -> None:
"""Remove a forward port from *zone* by port+proto (id used by API layer)."""
cfg = config_get()
zone_cfg = cfg.get("zones", {}).get(zone, {})
entry = None
for fp in zone_cfg.get("forward_ports", []):
if fp.get("port") == port and fp.get("proto") == protocol:
entry = fp
break
if entry is None:
raise ValueError(
f"Forward port {port}/{protocol} not found in zone '{zone}'"
)
remove_forward_port(
zone,
port,
protocol,
toaddr=entry.get("toaddr"),
toport=entry.get("toport"),
)
# ---------------------------------------------------------------------------
@@ -500,10 +603,7 @@ def remove_forward_port(
def _parse_forward_port(raw: str) -> dict[str, Any]:
"""Parse a single forward-port specifier into a structured dict.
Input: ``port=443/proto=tcp/toaddr=192.168.1.5/toport=8080``
"""
"""Parse a single forward-port specifier into a structured dict."""
result: dict[str, Any] = {}
for piece in raw.split("/"):
if "=" not in piece:
@@ -533,12 +633,7 @@ def _parse_forward_ports(value: str) -> list[dict[str, Any]]:
def get_state() -> dict[str, Any]:
"""Return the complete current state of firewalld as a Python dict.
The dict contains all zones with their per-zone configuration, all
rich rules, masquerade settings, forward-port rules, and the set of
active interfaces.
"""
"""Return the complete current state of firewalld as a Python dict."""
zones: dict[str, dict[str, Any]] = {}
for name in get_available_zones():
try:
@@ -558,70 +653,43 @@ def get_state() -> dict[str, Any]:
def _now_iso() -> str:
"""Return the current UTC time as an ISO-8601 string."""
from datetime import datetime
return datetime.now(UTC).isoformat()
def save_backup() -> str:
"""Capture the full state and write it to RULES_FILE on disk.
Returns:
Absolute path to the written file.
"""
"""Capture the full state and write it to RULES_FILE on disk."""
_ensure_data_dir()
state = get_state()
with open(RULES_FILE, "w") as fh:
json.dump(state, fh, indent=2, default=str)
logger.info("Firewall state backup saved to %s", RULES_FILE)
return RULES_FILE
def load_backup() -> dict[str, Any]:
"""Read the JSON backup file and return the state dict.
Use :func:`restore_backup` to actually apply the loaded state.
Raises:
FileNotFoundError: When no backup file exists at RULES_FILE.
json.JSONDecodeError: When the file is not valid JSON.
Returns:
The loaded state dict.
"""
"""Read the JSON backup file and return the state dict."""
with open(RULES_FILE) as fh:
state: dict[str, Any] = json.load(fh)
return state
def restore_backup(state: dict[str, Any]) -> None:
"""Apply the zone configuration described in *state*.
Walks every zone in *state*["zones"] and re-creates services,
interfaces, forward ports, masquerade, and rich rules.
This is a *merge*: zones not present in the snapshot are **not**
touched.
"""
"""Apply the zone configuration described in *state*."""
zones_cfg = state.get("zones", {})
for zone_name, zinfo in zones_cfg.items():
# Ensure the zone exists.
if zone_name not in get_available_zones():
target = zinfo.get("target", "default")
create_zone(zone_name, target)
# Services
services = zinfo.get("services", [])
set_zone_services(zone_name, services)
# Interfaces
interfaces = zinfo.get("interfaces", [])
set_zone_interfaces(zone_name, interfaces)
# Masquerade
if zinfo.get("masquerade"):
set_masquerade(zone_name, True)
# Forward ports (stored as dicts, or raw strings from old backups)
for fp in zinfo.get("forward-ports", []):
if isinstance(fp, str):
fp_str = fp
@@ -643,7 +711,6 @@ def restore_backup(state: dict[str, Any]) -> None:
check=False,
)
# Rich rules
for rule in zinfo.get("rich-rules", []):
_run(
[
@@ -657,6 +724,7 @@ def restore_backup(state: dict[str, Any]) -> None:
)
_reload()
logger.info("Firewall backup restored, %d zones processed", len(zones_cfg))
# ---------------------------------------------------------------------------
@@ -688,6 +756,7 @@ def config_set(cfg: dict[str, Any]) -> None:
json.dump(cfg, fh, indent=2)
fh.write("\n")
os.replace(tmp, CONFIG_FILE)
logger.info("Firewall declarative config saved")
def _normalize_target(target: str) -> str:
@@ -713,11 +782,7 @@ def _live_target_to_config(target: str) -> str:
def config_pending() -> dict[str, Any]:
"""Compare declarative config against live firewalld state, return diff.
Returns a dict with ``pending`` (list of change dicts), ``needs_apply``
(bool), and ``live_zones`` (dict of zones not yet in config).
"""
"""Compare declarative config against live firewalld state, return diff."""
cfg = config_get()
live_state = get_state()
cfg_zones = cfg.get("zones", {})
@@ -779,6 +844,38 @@ def config_pending() -> dict[str, Any]:
}
)
cfg_rules = {
r.get("rule") for r in zone_cfg.get("rich_rules", [])
}
live_rules = set(live_zone.get("rich-rules", []))
if cfg_rules != live_rules:
changes.append(
{
"zone": zone_name,
"type": "rich_rules",
"config_count": len(cfg_rules),
"live_count": len(live_rules),
}
)
cfg_fps = {
(fp.get("port"), fp.get("proto"))
for fp in zone_cfg.get("forward_ports", [])
}
live_fps = {
(fp.get("port"), fp.get("proto"))
for fp in live_zone.get("forward-ports", [])
}
if cfg_fps != live_fps:
changes.append(
{
"zone": zone_name,
"type": "forward_ports",
"config_count": len(cfg_fps),
"live_count": len(live_fps),
}
)
for zone_name in live_zones:
if zone_name not in cfg_zones:
unknown_live[zone_name] = {
@@ -793,14 +890,7 @@ def config_pending() -> dict[str, Any]:
def config_apply() -> dict[str, Any]:
"""Apply the declarative config to live firewalld.
Takes a snapshot via ``save_backup()`` first, then reconciles each zone
in the config (create/update, interfaces, services, masquerade), reloads,
and takes another snapshot.
Returns a dict with ``applied_zones`` and a ``backup`` path.
"""
"""Apply the declarative config to live firewalld."""
cfg = config_get()
cfg_zones = cfg.get("zones", {})
@@ -836,11 +926,48 @@ def config_apply() -> dict[str, Any]:
if mq is not None:
set_masquerade(zone_name, mq)
for rule_entry in zone_cfg.get("rich_rules", []):
rule_str = rule_entry.get("rule", "") if isinstance(rule_entry, dict) else str(rule_entry)
if rule_str:
_run(
[
"sudo",
"firewall-cmd",
f"--zone={zone_name}",
f"--add-rich-rule={rule_str}",
"--permanent",
],
check=False,
)
for fp_entry in zone_cfg.get("forward_ports", []):
if isinstance(fp_entry, str):
fp_str = fp_entry
else:
parts = [f"port={fp_entry['port']}", f"proto={fp_entry['proto']}"]
if "toaddr" in fp_entry:
parts.append(f"toaddr={fp_entry['toaddr']}")
if "toport" in fp_entry:
parts.append(f"toport={fp_entry['toport']}")
fp_str = "/".join(parts)
_run(
[
"sudo",
"firewall-cmd",
f"--zone={zone_name}",
f"--add-forward-port={fp_str}",
"--permanent",
],
check=False,
)
applied.append(zone_name)
_reload()
backup_path = save_backup()
logger.info("Firewall config applied to %d zones", len(applied))
return {
"applied_zones": applied,
"backup": backup_path,
@@ -875,7 +1002,9 @@ __all__ = [
"get_zone_info",
"load_backup",
"remove_forward_port",
"remove_forward_port_by_id",
"remove_rich_rule",
"remove_rich_rule_by_id",
"remove_zone_interface",
"remove_zone_service",
"restore_backup",