diff --git a/lib/firewall.py b/lib/firewall.py index 2816c71..24f4ac6 100644 --- a/lib/firewall.py +++ b/lib/firewall.py @@ -11,11 +11,18 @@ Flask UI can inspect or restore previous configurations. import json import os import subprocess +from contextlib import suppress from datetime import UTC +from pathlib import Path from typing import Any -DATA_DIR: str = "/home/wall/vacuum-wall/data/firewall" +PROJECT_DIR = Path(__file__).resolve().parent.parent +DATA_DIR: str = str(PROJECT_DIR / "data" / "firewall") RULES_FILE: str = os.path.join(DATA_DIR, "rules.json") +CONFIG_DIR = PROJECT_DIR / "config" / "firewall" +CONFIG_FILE = CONFIG_DIR / "config.json" + +DEFAULT_CONFIG: dict[str, Any] = {"zones": {}} # --------------------------------------------------------------------------- @@ -44,6 +51,7 @@ def _reload() -> None: def _ensure_data_dir() -> None: """Create the data directory tree if it does not exist.""" os.makedirs(DATA_DIR, exist_ok=True) + CONFIG_DIR.mkdir(parents=True, exist_ok=True) # --------------------------------------------------------------------------- @@ -651,8 +659,199 @@ def restore_backup(state: dict[str, Any]) -> None: _reload() +# --------------------------------------------------------------------------- +# Declarative config management (config/firewall/config.json) +# --------------------------------------------------------------------------- + + +def _ensure_config_file() -> None: + """Create config directory and file if they do not exist.""" + CONFIG_DIR.mkdir(parents=True, exist_ok=True) + if not CONFIG_FILE.exists(): + with open(CONFIG_FILE, "w") as fh: + json.dump(DEFAULT_CONFIG, fh, indent=2) + fh.write("\n") + + +def config_get() -> dict[str, Any]: + """Return the declarative config from ``config/firewall/config.json``.""" + _ensure_config_file() + with open(CONFIG_FILE) as fh: + return json.load(fh) + + +def config_set(cfg: dict[str, Any]) -> None: + """Write *cfg* to ``config/firewall/config.json`` (atomic replace).""" + _ensure_config_file() + tmp = CONFIG_FILE.with_name(CONFIG_FILE.name + ".tmp") + with open(tmp, "w") as fh: + json.dump(cfg, fh, indent=2) + fh.write("\n") + os.replace(tmp, CONFIG_FILE) + + +def _normalize_target(target: str) -> str: + """Map between config JSON target names and firewalld target values.""" + if target == "ACCEPT": + return "ACCEPT" + if target == "DROP": + return "DROP" + if target == "REJECT": + return "REJECT" + return "default" + + +def _live_target_to_config(target: str) -> str: + """Map firewalld target value back to config JSON canonical form.""" + if target == "ACCEPT": + return "ACCEPT" + if target == "DROP": + return "DROP" + if target == "REJECT": + return "REJECT" + return "DEFAULT" + + +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). + """ + cfg = config_get() + live_state = get_state() + cfg_zones = cfg.get("zones", {}) + live_zones = live_state.get("zones", {}) + + changes: list[dict[str, Any]] = [] + unknown_live: dict[str, Any] = {} + + for zone_name, zone_cfg in cfg_zones.items(): + live_zone = live_zones.get(zone_name, {}) + if not zone_cfg.get("interfaces"): + continue + + cfg_ifaces = set(zone_cfg.get("interfaces", [])) + live_ifaces = set(live_zone.get("interfaces", [])) + if cfg_ifaces != live_ifaces: + changes.append( + { + "zone": zone_name, + "type": "interfaces", + "config": sorted(cfg_ifaces), + "live": sorted(live_ifaces), + } + ) + + cfg_services = set(zone_cfg.get("services", [])) + live_services = set(live_zone.get("services", [])) + if cfg_services != live_services: + changes.append( + { + "zone": zone_name, + "type": "services", + "config": sorted(cfg_services), + "live": sorted(live_services), + } + ) + + cfg_target = _normalize_target(zone_cfg.get("target", "DEFAULT")) + live_target = live_zone.get("target", "default") + if cfg_target != live_target: + changes.append( + { + "zone": zone_name, + "type": "target", + "config": cfg_target, + "live": live_target, + } + ) + + cfg_mq = zone_cfg.get("masquerade", False) + live_mq = live_zone.get("masquerade", False) + if cfg_mq != live_mq: + changes.append( + { + "zone": zone_name, + "type": "masquerade", + "config": cfg_mq, + "live": live_mq, + } + ) + + for zone_name in live_zones: + if zone_name not in cfg_zones: + unknown_live[zone_name] = { + "interfaces": live_zones[zone_name].get("interfaces", []), + } + + return { + "pending": changes, + "needs_apply": len(changes) > 0, + "unmanaged_zones": unknown_live, + } + + +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. + """ + cfg = config_get() + cfg_zones = cfg.get("zones", {}) + + save_backup() + + available = get_available_zones() + applied: list[str] = [] + for zone_name, zone_cfg in cfg_zones.items(): + need_create = zone_name not in available + + if need_create: + target = _normalize_target(zone_cfg.get("target", "DEFAULT")) + create_zone(zone_name, target) + else: + desired_target = _normalize_target(zone_cfg.get("target", "DEFAULT")) + if desired_target != "default": + with suppress(RuntimeError): + _run( + [ + "sudo", + "firewall-cmd", + f"--zone={zone_name}", + f"--set-target={desired_target}", + "--permanent", + ], + check=False, + ) + + set_zone_services(zone_name, zone_cfg.get("services", [])) + set_zone_interfaces(zone_name, zone_cfg.get("interfaces", [])) + + mq = zone_cfg.get("masquerade", False) + if mq is not None: + set_masquerade(zone_name, mq) + + applied.append(zone_name) + + _reload() + backup_path = save_backup() + + return { + "applied_zones": applied, + "backup": backup_path, + } + + __all__ = [ + "CONFIG_DIR", + "CONFIG_FILE", "DATA_DIR", + "DEFAULT_CONFIG", "RULES_FILE", "_reload", "_run", @@ -660,6 +859,10 @@ __all__ = [ "add_rich_rule", "add_zone_interface", "add_zone_service", + "config_apply", + "config_get", + "config_pending", + "config_set", "create_zone", "delete_zone", "get_active_zones", diff --git a/tests/test_firewall.py b/tests/test_firewall.py index 93aac62..62b3296 100644 --- a/tests/test_firewall.py +++ b/tests/test_firewall.py @@ -159,3 +159,286 @@ class TestGetState: assert "zones" in result assert "active_zones" in result assert "timestamp" in result + + +class TestNormalizeTarget: + def test_accept(self): + assert firewall._normalize_target("ACCEPT") == "ACCEPT" + + def test_drop(self): + assert firewall._normalize_target("DROP") == "DROP" + + def test_reject(self): + assert firewall._normalize_target("REJECT") == "REJECT" + + def test_default(self): + assert firewall._normalize_target("DEFAULT") == "default" + assert firewall._normalize_target("default") == "default" + assert firewall._normalize_target("UNKNOWN") == "default" + + +class TestLiveTargetToConfig: + def test_accept(self): + assert firewall._live_target_to_config("ACCEPT") == "ACCEPT" + + def test_drop(self): + assert firewall._live_target_to_config("DROP") == "DROP" + + def test_reject(self): + assert firewall._live_target_to_config("REJECT") == "REJECT" + + def test_default(self): + assert firewall._live_target_to_config("default") == "DEFAULT" + assert firewall._live_target_to_config("") == "DEFAULT" + + +class TestEnsureConfigFile: + def test_creates_file_if_missing(self, tmp_path): + cfg_dir = tmp_path / "config" / "firewall" + cfg_file = cfg_dir / "config.json" + with ( + patch.object(firewall, "CONFIG_DIR", cfg_dir), + patch.object(firewall, "CONFIG_FILE", cfg_file), + ): + firewall._ensure_config_file() + assert cfg_file.exists() + import json as _json + + content = _json.loads(cfg_file.read_text()) + assert content == {"zones": {}} + + def test_skips_existing_file(self, tmp_path): + cfg_dir = tmp_path / "config" / "firewall" + cfg_file = cfg_dir / "config.json" + cfg_dir.mkdir(parents=True) + cfg_file.write_text('{"zones": {"public": {}}}') + with ( + patch.object(firewall, "CONFIG_DIR", cfg_dir), + patch.object(firewall, "CONFIG_FILE", cfg_file), + ): + firewall._ensure_config_file() + content = cfg_file.read_text() + assert '{"zones": {"public": {}}}' in content + + +class TestConfigGet: + @patch("lib.firewall._ensure_config_file") + def test_returns_config(self, mock_ensure, tmp_path): + cfg_file = tmp_path / "config.json" + cfg_file.write_text( + '{"zones": {"public": {"interfaces": ["eth0"], "services": ["http"], "masquerade": true, "target": "DEFAULT"}}}' + ) + with patch.object(firewall, "CONFIG_FILE", cfg_file): + result = firewall.config_get() + assert result["zones"]["public"]["interfaces"] == ["eth0"] + assert result["zones"]["public"]["services"] == ["http"] + + +class TestConfigSet: + def test_writes_config_atomic(self, tmp_path): + cfg_file = tmp_path / "config.json" + with ( + patch.object(firewall, "CONFIG_FILE", cfg_file), + patch.object(firewall, "CONFIG_DIR", tmp_path), + ): + firewall.config_set({"zones": {"test": {"interfaces": ["eth0"]}}}) + import json as _json + + content = _json.loads(cfg_file.read_text()) + assert content["zones"]["test"]["interfaces"] == ["eth0"] + + +class TestConfigApply: + @patch("lib.firewall.config_get") + @patch("lib.firewall.save_backup") + @patch("lib.firewall.get_available_zones") + @patch("lib.firewall.create_zone") + @patch("lib.firewall.set_zone_services") + @patch("lib.firewall.set_zone_interfaces") + @patch("lib.firewall.set_masquerade") + @patch("lib.firewall._reload") + def test_applies_existing_zone( + self, + mock_reload, + mock_set_mq, + mock_set_ifaces, + mock_set_svcs, + mock_create, + mock_available, + mock_backup, + mock_cfg, + ): + mock_cfg.return_value = { + "zones": { + "public": { + "target": "DEFAULT", + "interfaces": ["eth0"], + "services": ["http", "https"], + "masquerade": True, + }, + }, + } + mock_available.return_value = ["public", "internal"] + mock_backup.return_value = "/tmp/rules.json" + result = firewall.config_apply() + assert result["applied_zones"] == ["public"] + assert result["backup"] == "/tmp/rules.json" + mock_set_ifaces.assert_called_once_with("public", ["eth0"]) + mock_set_svcs.assert_called_once_with("public", ["http", "https"]) + mock_set_mq.assert_called_once_with("public", True) + + @patch("lib.firewall.config_get") + @patch("lib.firewall.save_backup") + @patch("lib.firewall.get_available_zones") + @patch("lib.firewall.create_zone") + @patch("lib.firewall.set_zone_services") + @patch("lib.firewall.set_zone_interfaces") + @patch("lib.firewall.set_masquerade") + @patch("lib.firewall._reload") + def test_creates_new_zone( + self, + mock_reload, + mock_set_mq, + mock_set_ifaces, + mock_set_svcs, + mock_create, + mock_available, + mock_backup, + mock_cfg, + ): + mock_cfg.return_value = { + "zones": { + "custom": { + "target": "ACCEPT", + "interfaces": ["eth2"], + "services": [], + "masquerade": False, + }, + }, + } + mock_available.return_value = ["public", "internal"] + mock_backup.return_value = "/tmp/rules.json" + result = firewall.config_apply() + assert result["applied_zones"] == ["custom"] + mock_create.assert_called_once_with("custom", "ACCEPT") + mock_set_ifaces.assert_called_once_with("custom", ["eth2"]) + + +class TestConfigPending: + @patch("lib.firewall.config_get") + @patch("lib.firewall.get_state") + def test_detects_interface_drift(self, mock_state, mock_cfg): + mock_cfg.return_value = { + "zones": { + "public": { + "interfaces": ["eth0"], + "services": ["http"], + "masquerade": False, + }, + }, + } + mock_state.return_value = { + "zones": { + "public": { + "interfaces": ["eth1"], + "services": ["http"], + "masquerade": False, + }, + }, + } + result = firewall.config_pending() + assert result["needs_apply"] is True + assert any(c["type"] == "interfaces" for c in result["pending"]) + + @patch("lib.firewall.config_get") + @patch("lib.firewall.get_state") + def test_in_sync(self, mock_state, mock_cfg): + mock_cfg.return_value = { + "zones": { + "public": { + "interfaces": ["eth0"], + "services": ["http"], + "masquerade": False, + }, + }, + } + mock_state.return_value = { + "zones": { + "public": { + "interfaces": ["eth0"], + "services": ["http"], + "masquerade": False, + }, + }, + } + result = firewall.config_pending() + assert result["needs_apply"] is False + + @patch("lib.firewall.config_get") + @patch("lib.firewall.get_state") + def test_detects_services_drift(self, mock_state, mock_cfg): + mock_cfg.return_value = { + "zones": { + "public": { + "interfaces": ["eth0"], + "services": ["http", "ssh"], + "masquerade": False, + }, + }, + } + mock_state.return_value = { + "zones": { + "public": { + "interfaces": ["eth0"], + "services": ["http"], + "masquerade": False, + }, + }, + } + result = firewall.config_pending() + assert any(c["type"] == "services" for c in result["pending"]) + + @patch("lib.firewall.config_get") + @patch("lib.firewall.get_state") + def test_detects_unmanaged_zones(self, mock_state, mock_cfg): + mock_cfg.return_value = {"zones": {}} + mock_state.return_value = { + "zones": { + "public": { + "interfaces": ["eth0"], + "services": [], + "masquerade": False, + }, + }, + } + result = firewall.config_pending() + assert "public" in result["unmanaged_zones"] + + +class TestConfigEmptyZones: + @patch("lib.firewall.config_get") + @patch("lib.firewall.save_backup") + @patch("lib.firewall.get_available_zones") + @patch("lib.firewall.create_zone") + @patch("lib.firewall.set_zone_services") + @patch("lib.firewall.set_zone_interfaces") + @patch("lib.firewall.set_masquerade") + @patch("lib.firewall._reload") + def test_empty_config_no_ops( + self, + mock_reload, + mock_set_mq, + mock_set_ifaces, + mock_set_svcs, + mock_create, + mock_available, + mock_backup, + mock_cfg, + ): + mock_cfg.return_value = {"zones": {}} + mock_available.return_value = [] + mock_backup.return_value = "/tmp/rules.json" + result = firewall.config_apply() + assert result["applied_zones"] == [] + mock_create.assert_not_called() + mock_set_ifaces.assert_not_called() diff --git a/webui/api/firewall.py b/webui/api/firewall.py index 870de39..f9de628 100644 --- a/webui/api/firewall.py +++ b/webui/api/firewall.py @@ -9,6 +9,9 @@ from flask import Blueprint, jsonify, request from lib.firewall import ( add_forward_port, add_rich_rule, + config_get, + config_pending, + config_set, create_zone, delete_zone, get_active_zones, @@ -40,6 +43,60 @@ def _ok(data=None): return jsonify({"ok": True, "data": data}) +# --------------------------------------------------------------------------- +# Declarative config (two-step: save -> apply) +# --------------------------------------------------------------------------- + + +@bp.route("/config", methods=["GET"]) +def config_get_bp(): + try: + return _ok(config_get()) + except Exception as exc: + return _error(str(exc), 500) + + +@bp.route("/config", methods=["POST"]) +def config_set_bp(): + body = request.get_json(silent=True) or {} + if "zones" not in body: + return _error("'zones' key is required", 400) + if not isinstance(body["zones"], dict): + return _error("'zones' must be a dict", 400) + try: + config_set(body) + pending_info = config_pending() + return _ok( + { + "config_saved": True, + "pending": pending_info["pending"], + "needs_apply": pending_info["needs_apply"], + "unmanaged_zones": pending_info.get("unmanaged_zones", {}), + } + ) + except Exception as exc: + return _error(str(exc), 500) + + +@bp.route("/config/apply", methods=["POST"]) +def config_apply_bp(): + try: + from lib.firewall import config_apply as _config_apply + + result = _config_apply() + return _ok(result) + except Exception as exc: + return _error(str(exc), 500) + + +@bp.route("/config/pending", methods=["GET"]) +def config_pending_bp(): + try: + return _ok(config_pending()) + except Exception as exc: + return _error(str(exc), 500) + + # --------------------------------------------------------------------------- # Zones # --------------------------------------------------------------------------- diff --git a/webui/server.py b/webui/server.py index dc3a302..e1d92f9 100644 --- a/webui/server.py +++ b/webui/server.py @@ -15,7 +15,13 @@ from lib.acme import get_email, list_certs from lib.dnsmasq import get_config as dnsmasq_config from lib.dnsmasq import get_lease_table from lib.dnsmasq import get_status as dnsmasq_status -from lib.firewall import get_active_zones, get_interfaces, get_zone_info +from lib.firewall import ( + config_get, + config_pending, + get_active_zones, + get_interfaces, + get_zone_info, +) from lib.nginx import get_config as nginx_config from lib.nginx import get_domains from lib.wireguard import get_config as wg_config @@ -155,26 +161,34 @@ def dashboard(): certs=certs, wg_status=wg, services=_get_service_status(dnsmasq, wg), + firewall_config=_safely(config_get, {}), + firewall_pending=_safely(config_pending, {}), ) @app.route("/interfaces") def interfaces_page(): + firewall_config = _safely(config_get, {}) + firewall_pending = _safely(config_pending, {}) return render_template( "interfaces.html", interfaces=_safely(get_interfaces, []), active_zones=_safely(get_active_zones, {}), + firewall_config=firewall_config, + firewall_pending=firewall_pending, ) @app.route("/zones") def zones_page(): - zones = {} + firewall_config = _safely(config_get, {}) + firewall_pending = _safely(config_pending, {}) + zones_data = {} for name in _safely(get_active_zones, {}): - zones[name] = _safely(lambda n=name: get_zone_info(n), {}) + zones_data[name] = _safely(lambda n=name: get_zone_info(n), {}) return render_template( "zones.html", - zones=zones, + zones=zones_data, interfaces=_safely(get_interfaces, []), services=_safely( lambda: __import__( @@ -182,6 +196,8 @@ def zones_page(): ).get_services(), [], ), + firewall_config=firewall_config, + firewall_pending=firewall_pending, )