"""Tests for lib/firewall.py (pure logic) and daemon/handlers/firewall.py (privilege boundary).""" from copy import deepcopy from unittest.mock import MagicMock, call, patch import pytest from daemon.handlers import firewall as daemonfirewall from daemon.server import ConflictError, NotFoundError from lib import firewall from lib.common import ( _APPLY_HASH_KEY, _LAST_APPLIED_CONFIG_KEY, config_hash, strip_apply_meta, ) # --------------------------------------------------------------------------- # lib/firewall.py — pure parsing (no sudo) # --------------------------------------------------------------------------- class TestParseForwardPorts: def test_lib_single_entry(self): result = firewall._parse_forward_ports("port=443/proto=tcp") assert len(result) == 1 assert result[0]["port"] == 443 assert result[0]["proto"] == "tcp" def test_lib_multiple_entries(self): result = firewall._parse_forward_ports( "port=443/proto=tcp port=80/proto=tcp/toaddr=10.0.0.1/toport=8080" ) assert len(result) == 2 assert result[0]["port"] == 443 assert result[1]["port"] == 80 assert result[1]["toaddr"] == "10.0.0.1" assert result[1]["toport"] == 8080 def test_lib_empty_string(self): assert firewall._parse_forward_ports("") == [] def test_daemon_no_redundant_import(self): assert not hasattr(daemonfirewall, "_parse_forward_ports") class TestParseActiveZones: def test_lib_parses_zones(self): result = firewall._parse_active_zones( "public\n eth0\ninternal\n eth1\n eth2" ) assert result == { "public": ["eth0"], "internal": ["eth1", "eth2"], } def test_lib_empty_output(self): assert firewall._parse_active_zones("") == {} def test_lib_zone_no_interfaces(self): assert firewall._parse_active_zones("dmz") == {"dmz": []} class TestParseZoneOutput: def test_lib_parses_zone(self): result = firewall._parse_zone_output( "public", ( "target: default\n" "interfaces: eth0\n" "services: ssh dhcp\n" "masquerade: yes\n" ), ) assert result["name"] == "public" assert result["services"] == ["ssh", "dhcp"] assert result["masquerade"] is True def test_lib_rich_rule_continuation_lines(self): """firewalld emits each rich rule on its own tab-indented line.""" rule = 'rule family="ipv4" port port="51820" protocol="udp" accept' result = firewall._parse_zone_output( "vpn-full", ( "target: default\n" "interfaces: \n" "rich rules: \n" "\t" + rule + "\n" "masquerade: yes\n" ), ) assert result["rich-rules"] == [rule] assert result["masquerade"] is True def test_lib_multiple_rich_rule_continuation_lines(self): rule_a = 'rule family="ipv4" port port="51820" protocol="udp" accept' rule_b = 'rule family="ipv4" source address="10.0.0.0/8" drop' result = firewall._parse_zone_output( "vpn-full", ("target: default\nrich rules: \n" + rule_a + "\n" + rule_b + "\n"), ) assert result["rich-rules"] == [rule_a, rule_b] def test_lib_no_rich_rules_when_no_continuation(self): result = firewall._parse_zone_output( "public", "target: default\nrich rules: \nmasquerade: no\n", ) assert result["rich-rules"] == [] class TestParseInterfaces: def test_lib_parses_interfaces(self): result = firewall._parse_interfaces( "1: lo: mtu 65536\n" "2: eth0: mtu 1500\n" ) assert result == ["lo", "eth0"] 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" # --------------------------------------------------------------------------- # lib/firewall.py — config helpers (no sudo) # --------------------------------------------------------------------------- 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.get_config() 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.save_config({"zones": {"test": {"interfaces": ["eth0"]}}}) import json as _json content = _json.loads(cfg_file.read_text()) assert content["zones"]["test"]["interfaces"] == ["eth0"] class TestConfigPending: @patch("lib.firewall.get_config") def test_detects_interface_drift(self, mock_cfg): mock_cfg.return_value = { "zones": { "public": { "interfaces": ["eth0"], "services": ["http"], "masquerade": False, }, }, } state = { "zones": { "public": { "interfaces": ["eth1"], "services": ["http"], "masquerade": False, }, }, } result = firewall.config_pending(state) assert result["needs_apply"] is True assert any(c["type"] == "interfaces" for c in result["pending"]) @patch("lib.firewall.get_config") def test_in_sync(self, mock_cfg): mock_cfg.return_value = { "zones": { "public": { "interfaces": ["eth0"], "services": ["http"], "masquerade": False, }, }, } state = { "zones": { "public": { "interfaces": ["eth0"], "services": ["http"], "masquerade": False, }, }, } result = firewall.config_pending(state) assert result["needs_apply"] is False @patch("lib.firewall.get_config") def test_detects_services_drift(self, mock_cfg): mock_cfg.return_value = { "zones": { "public": { "interfaces": ["eth0"], "services": ["http", "ssh"], "masquerade": False, }, }, } state = { "zones": { "public": { "interfaces": ["eth0"], "services": ["http"], "masquerade": False, }, }, } result = firewall.config_pending(state) assert any(c["type"] == "services" for c in result["pending"]) @patch("lib.firewall.get_config") def test_detects_unmanaged_zones(self, mock_cfg): # A custom live zone not in config is flagged as unmanaged. mock_cfg.return_value = {"zones": {}} state = { "zones": { "guest": { "interfaces": ["eth5"], "services": [], "masquerade": False, }, }, } result = firewall.config_pending(state) assert "guest" in result["unmanaged_zones"] @patch("lib.firewall.get_config") def test_built_in_zones_not_unmanaged(self, mock_cfg): # firewalld built-in zones are always present and must not be # reported as unmanaged, so they never surface as noise. mock_cfg.return_value = {"zones": {}} state = { "zones": { "public": {"interfaces": ["eth0"], "services": [], "masquerade": True}, "trusted": {"interfaces": ["lo"], "services": [], "masquerade": False}, "dmz": {"interfaces": ["eth7"], "services": [], "masquerade": False}, }, } result = firewall.config_pending(state) assert result["unmanaged_zones"] == {} @patch("lib.firewall.get_config") def test_public_masquerade_not_pending(self, mock_cfg): # public zone masquerade is driven by apply's propagation step, so a # config-vs-live masquerade mismatch on public is not a pending change. mock_cfg.return_value = { "zones": { "public": { "interfaces": ["eth0"], "services": ["http"], "masquerade": False, }, }, } state = { "zones": { "public": { "interfaces": ["eth0"], "services": ["http"], "masquerade": True, }, }, } result = firewall.config_pending(state) assert not any(c["type"] == "masquerade" for c in result["pending"]) assert result["needs_apply"] is False @patch("lib.firewall.get_config") def test_non_public_masquerade_is_pending(self, mock_cfg): # A non-public zone with a masquerade mismatch IS a pending change. mock_cfg.return_value = { "zones": { "internal": { "interfaces": ["eth1"], "services": [], "masquerade": False, }, }, } state = { "zones": { "internal": { "interfaces": ["eth1"], "services": [], "masquerade": True, }, }, } result = firewall.config_pending(state) assert any( c["type"] == "masquerade" and c["zone"] == "internal" for c in result["pending"] ) _PENDING_LIVE_PUBLIC = { "interfaces": ["eth0"], "services": ["http"], "masquerade": False, "target": "default", "rich-rules": [], "forward-ports": [], } class TestComputePendingChangesAbsentInterfaces: """Zones whose config lacks the 'interfaces' key are hands-off on apply, so their interfaces diff must not be reported; other field drift is.""" def test_services_drift_reported_without_interfaces_key(self): cfg = {"zones": {"public": {"services": ["http", "ssh"]}}} result = firewall._compute_pending_changes( cfg, {"public": _PENDING_LIVE_PUBLIC} ) types = {c["type"] for c in result["pending"]} assert "services" in types assert "interfaces" not in types def test_no_spurious_interfaces_entry_for_absent_key_zone(self): # Config in sync on everything except a missing interfaces key: the # zone's live interfaces are intentionally left alone by apply. cfg = {"zones": {"public": {"services": ["http"]}}} result = firewall._compute_pending_changes( cfg, {"public": _PENDING_LIVE_PUBLIC} ) assert result["pending"] == [] assert result["needs_apply"] is False def test_explicit_empty_interfaces_key_still_diffs(self): cfg = {"zones": {"public": {"interfaces": [], "services": ["http"]}}} result = firewall._compute_pending_changes( cfg, {"public": _PENDING_LIVE_PUBLIC} ) entries = [c for c in result["pending"] if c["type"] == "interfaces"] assert len(entries) == 1 assert entries[0]["config"] == [] assert entries[0]["live"] == ["eth0"] class TestTargetDriftSemantics: """Target is unmanaged when the config key is absent or normalizes to 'default' (WI-2, Option A); explicit ACCEPT/DROP/REJECT is fully managed.""" def test_absent_target_key_not_diffed(self): cfg = {"zones": {"public": {"interfaces": ["eth0"], "services": ["http"]}}} live = {"public": {**_PENDING_LIVE_PUBLIC, "target": "ACCEPT"}} result = firewall._compute_pending_changes(cfg, live) assert not any(c["type"] == "target" for c in result["pending"]) def test_explicit_default_target_not_diffed(self): cfg = { "zones": { "public": { "interfaces": ["eth0"], "services": ["http"], "target": "DEFAULT", } } } live = {"public": {**_PENDING_LIVE_PUBLIC, "target": "ACCEPT"}} result = firewall._compute_pending_changes(cfg, live) assert not any(c["type"] == "target" for c in result["pending"]) def test_explicit_accept_target_diffed(self): cfg = { "zones": { "public": { "interfaces": ["eth0"], "services": ["http"], "target": "ACCEPT", } } } live = {"public": _PENDING_LIVE_PUBLIC} # live target is 'default' result = firewall._compute_pending_changes(cfg, live) target_entries = [c for c in result["pending"] if c["type"] == "target"] assert len(target_entries) == 1 assert target_entries[0]["config"] == "ACCEPT" assert target_entries[0]["live"] == "default" # --------------------------------------------------------------------------- # lib/firewall.py — parse zone output (used by both lib and daemon) # --------------------------------------------------------------------------- class TestGetZoneInfo: def test_parses_zone_info(self): result = firewall._parse_zone_output( "public", ( "target: default\n" "interfaces: eth0\n" "sources: \n" "services: ssh dhcp\n" "ports: 8080/tcp\n" "protocols: \n" "forward-ports: \n" "masquerade: yes\n" "ics: no\n" "rich-rules: \n" "icmp-blocks: \n" "module: \n" ), ) assert result["name"] == "public" assert result["services"] == ["ssh", "dhcp"] assert result["ports"] == ["8080/tcp"] assert result["masquerade"] is True assert result["interfaces"] == ["eth0"] assert result["sources"] == [] assert result["rich-rules"] == [] # --------------------------------------------------------------------------- # lib/firewall.py — no sudo functions # --------------------------------------------------------------------------- class TestLibNoSudo: def test_no_run_import(self): import inspect source = inspect.getsource(firewall) assert "sudo=True" not in source, "lib/firewall.py must not call sudo" # --------------------------------------------------------------------------- # daemon/handlers/firewall.py — privileged operations (reads from state) # --------------------------------------------------------------------------- _FakeState = { "firewall": { "active_zones": {"public": ["eth0"], "internal": ["eth1"]}, "interfaces": [ { "name": "eth0", "mac": "aa:bb:cc:dd:ee:00", "state": "UP", "mtu": 1500, "ips": ["192.168.1.1/24"], "ipv6": [], "zone": "public", }, { "name": "eth1", "mac": "aa:bb:cc:dd:ee:01", "state": "UP", "mtu": 1500, "ips": ["10.0.0.1/24"], "ipv6": [], "zone": "internal", }, ], "available_services": ["ssh", "http", "dns"], "zones": { "public": { "name": "public", "interfaces": ["eth0"], "services": ["ssh"], "rich-rules": [], }, "internal": { "name": "internal", "interfaces": [], "services": [], "rich-rules": [], }, }, "rich_rules": { "public": [], "internal": [], }, "config": {"zones": {}}, "pending": {}, "timestamp": "2026-01-01T00:00:00+00:00", } } def _mock_state(): return _FakeState["firewall"] # GET endpoints read from state — mock lib.state.state.get() class TestDaemonGetInterfaces: @patch("lib.state.state") def test_parses_interfaces(self, mock_st): mock_st.get.return_value = _mock_state() result = daemonfirewall.get_interfaces(None, None) assert [i["name"] for i in result] == ["eth0", "eth1"] class TestDaemonGetZones: @patch("lib.state.state") def test_returns_zones(self, mock_st): mock_st.get.return_value = _mock_state() result = daemonfirewall.get_zones(None, None) assert "public" in result["active"] assert "internal" in result["active"] assert "public" in result["available"] class TestDaemonGetServices: @patch("lib.state.state") def test_returns_services(self, mock_st): mock_st.get.return_value = _mock_state() result = daemonfirewall.get_services(None, None) assert "ssh" in result assert "http" in result class TestDaemonGetRichRules: @patch("lib.state.state") def test_empty_rules(self, mock_st): mock_st.get.return_value = _mock_state() result = daemonfirewall.list_rich_rules(None, {"zone": "public"}) assert result == [] @patch("lib.state.state") def test_rules_with_ids(self, mock_st): mock_st.get.return_value = { **_mock_state(), "rich_rules": { "public": ['rule family="ipv4" port protocol="tcp" port="443" accept;'], }, } with patch.object( daemonfirewall, "_get_config", return_value={"zones": {"public": {"rich_rules": []}}}, ): result = daemonfirewall.list_rich_rules(None, {"zone": "public"}) assert len(result) == 1 class TestDaemonGetState: @patch("lib.state.state") def test_returns_full_state(self, mock_st): mock_st.get.return_value = _mock_state() result = daemonfirewall.get_state(None, None) assert "zones" in result assert "active_zones" in result assert "timestamp" in result assert "interfaces" in result assert len(result["interfaces"]) == 2 assert "public" in result["zones"] # --------------------------------------------------------------------------- # Mutation endpoints — still call subprocess (run) # --------------------------------------------------------------------------- class TestDaemonConfigApply: @patch( "lib.firewall.get_config", return_value={ "zones": { "public": { "target": "DEFAULT", "interfaces": ["eth0"], "services": ["http"], "masquerade": True, }, }, }, create=True, ) @patch( "daemon.handlers.firewall.run", return_value="public\ninternal\ntarget: default\ninterfaces: \nsources: \nservices: \nports: \nprotocols: \nforward-ports: \nmasquerade: no\nics: no\nrich-rules: \nicmp-blocks: \nmodule: \n", ) def test_applies_existing_zone(self, mock_run, mock_cfg): with ( patch("lib.network.get_config", return_value={"interfaces": {}}), patch( "daemon.handlers.firewall._save_backup", return_value="/tmp/rules.json" ), patch( "daemon.handlers.firewall._get_state", return_value={"zones": {"public": {}}}, ), patch("daemon.handlers.firewall.refresh_state"), patch( "daemon.handlers.firewall._get_config", return_value={"zones": {"public": {}}}, ), patch("daemon.handlers.firewall._save_config"), ): result = daemonfirewall._config_apply() assert result["applied_zones"] == ["public"] # --------------------------------------------------------------------------- # Management-lockout guard: default zone must keep https or ssh # --------------------------------------------------------------------------- class TestDaemonMgmtLockoutGuard: ZONES_OUT = "public\ninternal" @patch("daemon.handlers.firewall._default_zone", return_value="public") def test_set_zone_services_blocks_default_zone(self, mock_dz): with ( patch( "daemon.handlers.firewall.run", return_value=self.ZONES_OUT ) as mock_run, pytest.raises(ConflictError) as exc, ): daemonfirewall.set_zone_services( None, {"zone": "public", "services": ["http"]} ) assert "https and ssh" in str(exc.value) # Guard fires before any mutation: only the zone-existence check ran. assert mock_run.call_args_list == [ call(["firewall-cmd", "--get-zones"], sudo=True), ] @patch("daemon.handlers.firewall._default_zone", return_value="public") def test_set_zone_services_force_bypasses_guard(self, mock_dz): with ( patch("daemon.handlers.firewall.run", return_value=self.ZONES_OUT), patch.object( daemonfirewall, "_parse_zone_output", return_value={"services": []} ), patch.object(daemonfirewall, "_reload"), patch.object(daemonfirewall, "_get_config", return_value={"zones": {}}), patch.object(daemonfirewall, "_save_config") as mock_save, patch.object(daemonfirewall, "bus") as mock_bus, patch("daemon.handlers.firewall.refresh_state"), ): mock_bus.emit.return_value = MagicMock(affected_subsystems=[]) result = daemonfirewall.set_zone_services( None, {"zone": "public", "services": ["http"], "force": True} ) assert result == {"zone": "public", "services": ["http"]} cfg = mock_save.call_args[0][0] assert cfg["zones"]["public"]["services"] == ["http"] @patch("daemon.handlers.firewall._default_zone", return_value="internal") def test_set_zone_services_non_default_zone_allowed(self, mock_dz): with ( patch("daemon.handlers.firewall.run", return_value=self.ZONES_OUT), patch.object( daemonfirewall, "_parse_zone_output", return_value={"services": []} ), patch.object(daemonfirewall, "_reload"), patch.object(daemonfirewall, "_get_config", return_value={"zones": {}}), patch.object(daemonfirewall, "_save_config"), patch.object(daemonfirewall, "bus") as mock_bus, patch("daemon.handlers.firewall.refresh_state"), ): mock_bus.emit.return_value = MagicMock(affected_subsystems=[]) result = daemonfirewall.set_zone_services( None, {"zone": "public", "services": []} ) assert result == {"zone": "public", "services": []} def test_would_remove_mgmt_keeps_https(self): assert daemonfirewall._would_remove_mgmt("public", ["http", "https"]) is False assert daemonfirewall._would_remove_mgmt("public", ["ssh"]) is False def test_would_remove_mgmt_fails_closed_on_error(self): with patch( "daemon.handlers.firewall._default_zone", side_effect=RuntimeError("boom") ): assert daemonfirewall._would_remove_mgmt("public", ["http"]) is True @patch("daemon.handlers.firewall._default_zone", return_value="default-zone") def test_would_remove_mgmt_other_zone(self, mock_dz): assert daemonfirewall._would_remove_mgmt("public", ["http"]) is False @patch( "lib.firewall.get_config", return_value={ "zones": {"public": {"services": ["http"], "interfaces": ["eth0"]}} }, create=True, ) @patch("daemon.handlers.firewall._default_zone", return_value="public") def test_config_apply_blocks_lockout_before_backup(self, mock_dz, mock_cfg): with ( patch("daemon.handlers.firewall._save_backup") as mock_backup, pytest.raises(ConflictError) as exc, ): daemonfirewall._config_apply() assert "https and ssh" in str(exc.value) mock_backup.assert_not_called() @patch( "lib.firewall.get_config", return_value={ "zones": { "public": { "target": "DEFAULT", "interfaces": ["eth0"], "services": ["http"], "masquerade": False, }, }, }, create=True, ) @patch( "daemon.handlers.firewall.run", return_value="public\ninternal\ntarget: default\ninterfaces: \nsources: \nservices: \nports: \nprotocols: \nforward-ports: \nmasquerade: no\nics: no\nrich-rules: \nicmp-blocks: \nmodule: \n", ) @patch("daemon.handlers.firewall._default_zone", return_value="public") def test_config_apply_force_bypasses_guard(self, mock_dz, mock_run, mock_cfg): with ( patch( "daemon.handlers.firewall._save_backup", return_value="/tmp/rules.json" ), patch( "daemon.handlers.firewall._get_state", return_value={"zones": {"public": {}}}, ), patch("daemon.handlers.firewall.refresh_state"), patch( "daemon.handlers.firewall._get_config", return_value={"zones": {"public": {}}}, ), patch("daemon.handlers.firewall._save_config"), ): result = daemonfirewall._config_apply(force=True) assert result["applied_zones"] == ["public"] # --------------------------------------------------------------------------- # Interface-coverage guard: apply must not leave a network-managed interface # in no zone (clients lose connectivity/DHCP) unless forced. # --------------------------------------------------------------------------- def _make_run( active_out: str, zones_out: str = "public\ninternal", zone_out: str = ( "target: default\n" "interfaces: eth0\n" "services: http\n" "masquerade: no\n" "rich-rules: \n" "forward-ports: \n" ), ): def _side_effect(cmd, **kwargs): if cmd == ["firewall-cmd", "--get-active-zones"]: return active_out if cmd == ["firewall-cmd", "--get-zones"]: return zones_out if cmd and cmd[-1] == "--list-all": return zone_out return "" return _side_effect def _apply_with( cfg: dict, network_ifaces: dict, active_out: str, force: bool = False, backup: str = "/tmp/rules.json", ): """Run _config_apply with the standard mock set; return (result, mock_run).""" with ( patch("lib.network.get_config", return_value={"interfaces": network_ifaces}), patch("lib.firewall.get_config", return_value=cfg, create=True), patch( "daemon.handlers.firewall.run", side_effect=_make_run(active_out), ) as mock_run, patch("daemon.handlers.firewall._default_zone", return_value="internal"), patch("daemon.handlers.firewall._save_backup", return_value=backup), patch("daemon.handlers.firewall.refresh_state"), patch( "daemon.handlers.firewall._get_config", return_value=deepcopy(cfg), ), patch("daemon.handlers.firewall._save_config"), ): result = daemonfirewall._config_apply(force=force) return result, mock_run class TestDaemonInterfaceCoverageGuard: def test_absent_key_zone_keeps_live_interfaces_on_apply(self): cfg = {"zones": {"public": {"services": ["http"], "masquerade": False}}} result, mock_run = _apply_with(cfg, {"eth0": {}}, "public\n eth0\n") assert result["applied_zones"] == ["public"] # The guard reads live zones once, up front. assert mock_run.call_args_list[0].args[0] == [ "firewall-cmd", "--get-active-zones", ] # Hands off: no interface mutation commands for the absent-key zone. for c in mock_run.call_args_list: for arg in c.args[0]: assert not arg.startswith("--remove-interface=") assert not arg.startswith("--add-interface=") def test_explicit_empty_list_unassigns(self): cfg = {"zones": {"public": {"interfaces": [], "services": []}}} result, mock_run = _apply_with(cfg, {}, "public\n eth0\n") assert result["applied_zones"] == ["public"] cmds = [c.args[0] for c in mock_run.call_args_list] assert [ "firewall-cmd", "--zone=public", "--remove-interface=eth0", "--permanent", ] in cmds assert not any( any(a.startswith("--add-interface=") for a in cmd) for cmd in cmds ) def test_conflict_when_network_iface_goes_uncovered(self): cfg = {"zones": {"public": {"interfaces": ["eth1"], "services": []}}} with ( patch("lib.network.get_config", return_value={"interfaces": {"eth0": {}}}), patch("lib.firewall.get_config", return_value=cfg, create=True), patch( "daemon.handlers.firewall.run", side_effect=_make_run("public\n eth0\n"), ), patch("daemon.handlers.firewall._default_zone", return_value="internal"), pytest.raises(ConflictError) as exc, ): daemonfirewall._config_apply() assert "eth0" in str(exc.value) assert "force" in str(exc.value) def test_force_bypasses_coverage_guard(self): cfg = {"zones": {"public": {"interfaces": ["eth1"], "services": []}}} result, _ = _apply_with(cfg, {"eth0": {}}, "public\n eth0\n", force=True) assert result["applied_zones"] == ["public"] def test_guard_ignores_lo_and_wg(self): # lo/wg* are never guarded even though the network config carries them. cfg = {"zones": {"public": {"interfaces": [], "services": []}}} result, _ = _apply_with(cfg, {"lo": {}, "wg0": {}}, "public\n eth0\n") assert result["applied_zones"] == ["public"] def test_live_only_zone_interfaces_count_as_covered(self): # eth1 is held by 'guest', which is live but absent from the config — # apply never touches it, so eth1 counts as covered. cfg = {"zones": {"public": {"interfaces": [], "services": []}}} result, _ = _apply_with(cfg, {"eth1": {}}, "public\n eth0\nguest\n eth1\n") assert result["applied_zones"] == ["public"] def test_coverage_guard_conflict_writes_no_backup(self): # Guard conflict must be side-effect free, like the lockout conflict. cfg = {"zones": {"public": {"interfaces": ["eth1"], "services": []}}} with ( patch("lib.network.get_config", return_value={"interfaces": {"eth0": {}}}), patch("lib.firewall.get_config", return_value=cfg, create=True), patch( "daemon.handlers.firewall.run", side_effect=_make_run("public\n eth0\n"), ) as mock_run, patch("daemon.handlers.firewall._default_zone", return_value="internal"), patch("daemon.handlers.firewall._save_backup") as mock_backup, pytest.raises(ConflictError), ): daemonfirewall._config_apply() mock_backup.assert_not_called() # Only the guard's live-zone read ran — no mutation commands at all. assert [c.args[0] for c in mock_run.call_args_list] == [ ["firewall-cmd", "--get-active-zones"] ] # --------------------------------------------------------------------------- # create_zone — must run --new-zone first, then set only non-default targets # --------------------------------------------------------------------------- class TestDaemonCreateZone: def _run(self, body, run_return="public internal"): with ( patch("daemon.handlers.firewall.run", return_value=run_return) as mock_run, patch.object(daemonfirewall, "_reload"), patch.object(daemonfirewall, "bus") as mock_bus, patch("daemon.handlers.firewall.refresh_state"), ): mock_bus.emit.return_value = MagicMock(affected_subsystems=[]) result = daemonfirewall.create_zone(None, body) return result, mock_run def test_new_zone_always_created(self): result, mock_run = self._run({"name": "guest"}) assert result == {"zone": "guest"} cmds = [c.args[0] for c in mock_run.call_args_list] assert ["firewall-cmd", "--new-zone=guest", "--permanent"] in cmds def test_default_target_not_set(self): result, mock_run = self._run({"name": "guest", "target": "default"}) assert result == {"zone": "guest"} cmds = [c.args[0] for c in mock_run.call_args_list] assert not any("--set-target=" in " ".join(c) for c in cmds) def test_accept_target_is_set(self): result, mock_run = self._run({"name": "guest", "target": "ACCEPT"}) assert result == {"zone": "guest"} cmds = [c.args[0] for c in mock_run.call_args_list] assert [ "firewall-cmd", "--zone=guest", "--set-target=ACCEPT", "--permanent", ] in cmds def test_existing_zone_rejected(self): with ( patch("daemon.handlers.firewall.run", return_value="public guest"), pytest.raises(ValueError), ): daemonfirewall.create_zone(None, {"name": "guest"}) # --------------------------------------------------------------------------- # Pre-apply recovery snapshot (single backup write with a real payload) # --------------------------------------------------------------------------- class TestDaemonConfigApplyBackup: def test_pre_apply_snapshot_shape(self): # public carries https+ssh so the default-zone lockout guard does not fire. cfg = { "zones": { "public": { "interfaces": ["eth0"], "services": ["https", "ssh"], } } } with ( patch("lib.network.get_config", return_value={"interfaces": {}}), patch("lib.firewall.get_config", return_value=cfg, create=True), patch( "daemon.handlers.firewall.run", side_effect=_make_run("public\n eth0\n"), ), patch("daemon.handlers.firewall._default_zone", return_value="public"), patch( "daemon.handlers.firewall._save_backup", return_value="/tmp/rules.json", ) as mock_backup, patch("daemon.handlers.firewall.refresh_state"), patch("daemon.handlers.firewall._get_config", return_value=deepcopy(cfg)), patch("daemon.handlers.firewall._save_config"), ): result = daemonfirewall._config_apply() assert result["backup"] == "/tmp/rules.json" # A single pre-apply snapshot (no post-apply skeleton write). assert mock_backup.call_count == 1 snapshot = mock_backup.call_args[0][0] assert {"timestamp", "default_zone", "zones", "config"} <= set(snapshot) assert snapshot["default_zone"] == "public" assert snapshot["config"] == cfg # --------------------------------------------------------------------------- # Applied-baseline stamping # --------------------------------------------------------------------------- _STAMP_TEST_CFG = { "zones": { "public": { "target": "DEFAULT", "interfaces": ["eth0"], "services": ["http"], "masquerade": False, }, }, } class TestDaemonConfigApplyStamp: """Verify _config_apply records the applied baseline in the config file.""" ZONE_LIST_ALL_OUT = ( "target: default\ninterfaces: \nsources: \nservices: \nports: \n" "protocols: \nforward-ports: \nmasquerade: no\nics: no\nrich-rules: \n" "icmp-blocks: \nmodule: \n" ) @patch( "lib.firewall.get_config", return_value=_STAMP_TEST_CFG, create=True, ) @patch( "daemon.handlers.firewall.run", return_value=ZONE_LIST_ALL_OUT, ) def test_stamps_applied_baseline(self, mock_run, mock_cfg): with ( patch("lib.network.get_config", return_value={"interfaces": {"eth0": {}}}), patch( "daemon.handlers.firewall._save_backup", return_value="/tmp/rules.json", ), patch( "daemon.handlers.firewall._get_state", return_value={"zones": {"public": {}}}, ), patch("daemon.handlers.firewall.refresh_state"), patch( "daemon.handlers.firewall._get_config", return_value=deepcopy(_STAMP_TEST_CFG), ), patch("daemon.handlers.firewall._save_config") as mock_save, ): result = daemonfirewall._config_apply() assert result["applied_zones"] == ["public"] saved = mock_save.call_args[0][0] assert saved[_LAST_APPLIED_CONFIG_KEY] == strip_apply_meta(saved) assert saved[_APPLY_HASH_KEY] == config_hash(saved) # The snapshot is the applied (meta-stripped) config. assert saved[_LAST_APPLIED_CONFIG_KEY] == _STAMP_TEST_CFG class TestDaemonGetConfigEndpoint: def test_strips_apply_meta(self): with patch.object( daemonfirewall, "_get_config", return_value={ "zones": {}, _APPLY_HASH_KEY: "h", _LAST_APPLIED_CONFIG_KEY: {"zones": {}}, }, ): result = daemonfirewall.get_config(None, None) assert result == {"zones": {}} @patch( "daemon.handlers.firewall._config_apply", return_value={"applied_zones": ["public"], "backup": "/tmp/rules.json"}, ) @patch("daemon.handlers.firewall.bus") def test_config_apply_handler_force_propagation(self, mock_bus, mock_apply): mock_bus.emit.return_value = MagicMock(affected_subsystems=[]) with patch("daemon.handlers.firewall.refresh_state"): daemonfirewall.config_apply(None, None) mock_apply.assert_called_once_with(force=False) mock_apply.reset_mock() daemonfirewall.config_apply(None, {"force": True}) mock_apply.assert_called_once_with(force=True) class TestDaemonConfigPending: @patch("lib.state.state") def test_returns_pending(self, mock_st): mock_st.get.return_value = { **_mock_state(), "pending": {"needs_apply": True, "pending": [{"type": "services"}]}, } result = daemonfirewall.config_pending_handler(None, None) assert result["needs_apply"] is True @patch("lib.state.state") def test_no_state_mutation(self, mock_st): pending = { "needs_apply": True, "pending": [{"zone": "public", "type": "services"}], } mock_st.get.return_value = {**_mock_state(), "pending": pending} original_keys = set(pending.keys()) result = daemonfirewall.config_pending_handler(None, None) assert "pending_summary" in result assert set(pending.keys()) == original_keys, ( "config_pending_handler must not mutate state store pending dict" ) @patch("lib.state.state") def test_detail_text_interfaces(self, mock_st): mock_st.get.return_value = { **_mock_state(), "pending": { "needs_apply": True, "pending": [ { "zone": "internal", "type": "interfaces", "config": ["eth1", "eth2"], "live": ["eth1"], } ], }, } result = daemonfirewall.config_pending_handler(None, None) assert len(result["pending_summary"]) == 1 assert "Zone internal: interfaces changed" in result["pending_summary"][0] assert "eth2" in result["pending_summary"][0] @patch("lib.state.state") def test_detail_text_services(self, mock_st): mock_st.get.return_value = { **_mock_state(), "pending": { "needs_apply": True, "pending": [ { "zone": "dmz", "type": "services", "config": ["ssh", "dns"], "live": ["ssh"], } ], }, } result = daemonfirewall.config_pending_handler(None, None) assert len(result["pending_summary"]) == 1 assert "Zone dmz: services changed" in result["pending_summary"][0] @patch("lib.state.state") def test_detail_text_rich_rules(self, mock_st): mock_st.get.return_value = { **_mock_state(), "pending": { "needs_apply": True, "pending": [ { "zone": "public", "type": "rich_rules", "config_count": 3, "live_count": 1, } ], }, } result = daemonfirewall.config_pending_handler(None, None) assert len(result["pending_summary"]) == 1 assert "Zone public: rich rules differ" in result["pending_summary"][0] assert "config: 3" in result["pending_summary"][0] assert "live: 1" in result["pending_summary"][0] @patch("lib.state.state") def test_detail_text_masquerade(self, mock_st): mock_st.get.return_value = { **_mock_state(), "pending": { "needs_apply": True, "pending": [ { "zone": "wan", "type": "masquerade", "config": True, "live": False, } ], }, } result = daemonfirewall.config_pending_handler(None, None) assert len(result["pending_summary"]) == 1 assert "Zone wan: masquerade changed" in result["pending_summary"][0] assert "config: True" in result["pending_summary"][0] @patch("lib.state.state") def test_detail_text_target(self, mock_st): mock_st.get.return_value = { **_mock_state(), "pending": { "needs_apply": True, "pending": [ { "zone": "trusted", "type": "target", "config": "ACCEPT", "live": "default", } ], }, } result = daemonfirewall.config_pending_handler(None, None) assert len(result["pending_summary"]) == 1 assert "Zone trusted: target changed" in result["pending_summary"][0] assert "config: ACCEPT" in result["pending_summary"][0] @patch("lib.state.state") def test_detail_text_unknown_type(self, mock_st): mock_st.get.return_value = { **_mock_state(), "pending": { "needs_apply": True, "pending": [{"zone": "public", "type": "foobarLayout"}], }, } result = daemonfirewall.config_pending_handler(None, None) assert len(result["pending_summary"]) == 1 assert "Zone public: foobarLayout changed" in result["pending_summary"][0] @patch("lib.state.state") def test_detail_text_mixed_types(self, mock_st): mock_st.get.return_value = { **_mock_state(), "pending": { "needs_apply": True, "pending": [ { "zone": "internal", "type": "interfaces", "config": ["eth1"], "live": [], }, { "zone": "dmz", "type": "services", "config": ["ssh", "dns"], "live": ["ssh"], }, { "zone": "public", "type": "rich_rules", "config_count": 2, "live_count": 1, }, ], }, } result = daemonfirewall.config_pending_handler(None, None) assert len(result["pending_summary"]) == 3 assert "Zone internal: interfaces changed" in result["pending_summary"][0] assert "Zone dmz: services changed" in result["pending_summary"][1] assert "Zone public: rich rules differ" in result["pending_summary"][2] # --------------------------------------------------------------------------- # Zone validation in add_rich_rule, remove_rich_rule, remove_forward_port # --------------------------------------------------------------------------- class TestDaemonZoneValidation: @patch("daemon.handlers.firewall.run") def test_add_rich_rule_invalid_zone(self, mock_run): mock_run.return_value = "public\ninternal" with ( patch.object(daemonfirewall, "_get_config", return_value={"zones": {}}), patch.object(daemonfirewall, "_save_config"), pytest.raises(NotFoundError), ): daemonfirewall.add_rich_rule( None, { "zone": "nonexistent", "rule": "rule accept", }, ) @patch("daemon.handlers.firewall.run") def test_remove_rich_rule_invalid_zone(self, mock_run): mock_run.return_value = "public\ninternal" with ( patch.object(daemonfirewall, "_get_config", return_value={"zones": {}}), pytest.raises(NotFoundError), ): daemonfirewall.remove_rich_rule( None, {"zone": "nonexistent", "id": "abc123"} ) @patch("daemon.handlers.firewall.run") def test_remove_forward_port_invalid_zone(self, mock_run): mock_run.return_value = "public\ninternal" with ( patch.object(daemonfirewall, "_get_config", return_value={"zones": {}}), pytest.raises(NotFoundError), ): daemonfirewall.remove_forward_port( None, { "zone": "nonexistent", "port": 443, "proto": "tcp", }, ) # --------------------------------------------------------------------------- # lib/firewall parsing is reused by state module # --------------------------------------------------------------------------- class TestLibParseForwardPorts: def test_single_entry(self): result = firewall._parse_forward_ports("port=443/proto=tcp") assert len(result) == 1 assert result[0]["port"] == 443 assert result[0]["proto"] == "tcp" def test_empty_string(self): assert firewall._parse_forward_ports("") == [] # --------------------------------------------------------------------------- # lib/firewall.py — parse all zones output (--list-all-zones) # --------------------------------------------------------------------------- class TestParseAllZonesOutput: def test_parses_single_zone(self): result = firewall._parse_all_zones_output( "public\n" " target: default\n" " interfaces: eth0\n" " services: ssh http\n" " masquerade: yes\n" " rich rules: \n" ) assert "public" in result assert result["public"]["name"] == "public" assert result["public"]["interfaces"] == ["eth0"] assert result["public"]["services"] == ["ssh", "http"] assert result["public"]["masquerade"] is True assert result["public"]["rich-rules"] == [] def test_parses_multiple_zones(self): result = firewall._parse_all_zones_output( "public (default, active)\n" " target: default\n" " interfaces: eth0\n" " services: ssh\n" " masquerade: no\n" " rich rules: \n" "internal (active)\n" " target: ACCEPT\n" " interfaces: eth1\n" " services: dhcp\n" " masquerade: no\n" " rich rules: \n" "trusted\n" " target: ACCEPT\n" " interfaces: \n" " services: \n" " masquerade: no\n" " rich rules: \n" ) assert set(result.keys()) == {"public", "internal", "trusted"} assert result["public"]["interfaces"] == ["eth0"] assert result["internal"]["target"] == "ACCEPT" assert result["trusted"]["services"] == [] def test_parses_rich_rule_continuation_lines(self): rule = 'rule family="ipv4" port port="51820" protocol="udp" accept' result = firewall._parse_all_zones_output( "vpn-full\n" " target: default\n" " interfaces: \n" " rich rules: \n" "\t" + rule + "\n" " masquerade: yes\n" "dmz\n" " target: default\n" " rich rules: \n" ) assert result["vpn-full"]["rich-rules"] == [rule] assert result["vpn-full"]["masquerade"] is True assert result["dmz"]["rich-rules"] == [] def test_empty_output(self): assert firewall._parse_all_zones_output("") == {} assert firewall._parse_all_zones_output("\n \n") == {} def test_handles_blank_lines_between_zones(self): result = firewall._parse_all_zones_output( "public\n" " target: default\n" " interfaces: eth0\n" " rich rules: \n" "\n" "internal\n" " target: ACCEPT\n" " interfaces: eth1\n" " rich rules: \n" ) assert "public" in result assert "internal" in result assert result["public"]["interfaces"] == ["eth0"] assert result["internal"]["interfaces"] == ["eth1"] def test_all_default_fields_present(self): result = firewall._parse_all_zones_output( "dmz\n target: default\n interfaces: \n services: \n rich rules: \n" ) zone = result["dmz"] for field in ( "interfaces", "sources", "services", "ports", "protocols", "forward-ports", "masquerade", "ics", "icmp-blocks", "module", "target", "rich-rules", ): assert field in zone, f"Missing field: {field}" # --------------------------------------------------------------------------- # lib/firewall.py — service catalog descriptions # --------------------------------------------------------------------------- def _write_service(dir, name, body): (dir / f"{name}.xml").write_text(body) class TestGetServiceDescriptions: def test_parses_short_preferred(self, tmp_path): d1 = tmp_path / "builtin" d2 = tmp_path / "etc" d1.mkdir() d2.mkdir() _write_service( d1, "ssh", "OpenSSHRemote login", ) _write_service( d1, "http", "WWWWeb server", ) _write_service( d2, "custom", "Only a long description", ) result = firewall.get_service_descriptions([d1, d2]) assert result == { "ssh": "OpenSSH", "http": "WWW", "custom": "Only a long description", } def test_description_fallback_when_no_short(self, tmp_path): d = tmp_path / "svc" d.mkdir() _write_service( d, "ntp", "Time synchronization", ) assert firewall.get_service_descriptions([d]) == {"ntp": "Time synchronization"} def test_etc_overrides_builtin(self, tmp_path): builtin = tmp_path / "builtin" etc = tmp_path / "etc" builtin.mkdir() etc.mkdir() _write_service(builtin, "ssh", "Built-in SSH") _write_service(etc, "ssh", "Custom SSH") # builtin listed first, etc second (same order as the default dirs) assert firewall.get_service_descriptions([builtin, etc]) == { "ssh": "Custom SSH" } def test_missing_dirs_return_empty(self, tmp_path): assert ( firewall.get_service_descriptions([tmp_path / "nope1", tmp_path / "nope2"]) == {} ) def test_malformed_xml_skipped(self, tmp_path): d = tmp_path / "svc" d.mkdir() _write_service(d, "broken", "Unclosed") _write_service(d, "good", "Works") result = firewall.get_service_descriptions([d]) assert result == {"good": "Works"} def test_empty_text_not_recorded(self, tmp_path): d = tmp_path / "svc" d.mkdir() _write_service(d, "blank", "") assert firewall.get_service_descriptions([d]) == {} def test_explicit_dirs_not_cached(self, tmp_path): d1 = tmp_path / "first" d2 = tmp_path / "second" d1.mkdir() d2.mkdir() _write_service(d1, "a", "A1") result1 = firewall.get_service_descriptions([d1]) _write_service(d2, "a", "A2") result2 = firewall.get_service_descriptions([d1, d2]) assert result1 == {"a": "A1"} assert result2 == {"a": "A2"}