"""Tests for lib/firewall.py (pure logic) and daemon/handlers/firewall.py (privilege boundary).""" from unittest.mock import MagicMock, patch import pytest from daemon.handlers import firewall as daemonfirewall from daemon.server import NotFoundError from lib import firewall # --------------------------------------------------------------------------- # 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": []} def test_daemon_import_same(self): assert daemonfirewall._parse_active_zones is firewall._parse_active_zones 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_daemon_import_same(self): assert daemonfirewall._parse_zone_output is firewall._parse_zone_output 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): mock_cfg.return_value = {"zones": {}} state = { "zones": { "public": { "interfaces": ["eth0"], "services": [], "masquerade": False, }, }, } result = firewall.config_pending(state) assert "public" in result["unmanaged_zones"] # --------------------------------------------------------------------------- # 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 # --------------------------------------------------------------------------- def _mock_run_factory(*outputs): """Create a mock run() that cycles through outputs on successive calls.""" idx = [0] def side_effect(*args, **kwargs): result = outputs[idx[0] % len(outputs)] idx[0] += 1 if result is RuntimeError: raise RuntimeError("command failed") return result return side_effect class TestDaemonParseForwardPorts: def test_handler_uses_get_forward_ports(self): assert callable(daemonfirewall._get_forward_ports) class TestDaemonParseActiveZones: def test_parses_active_zones(self): result = daemonfirewall._parse_active_zones( "public\n eth0\ninternal\n eth1\n eth2" ) assert result == { "public": ["eth0"], "internal": ["eth1", "eth2"], } def test_empty_output(self): assert daemonfirewall._parse_active_zones("") == {} def test_zone_with_no_interfaces(self): assert daemonfirewall._parse_active_zones("dmz") == {"dmz": []} class TestDaemonParseZoneOutput: def test_parses_zone_info(self): result = daemonfirewall._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"] class TestDaemonGetInterfaces: @patch("daemon.handlers.firewall.run") def test_parses_interfaces(self, mock_run): link_out = ( "1: lo: mtu 65536\n" "2: eth0: mtu 1500\n" "3: eth1: mtu 1500\n" ) mock_run.return_value = link_out result = daemonfirewall.get_interfaces(None, None) assert [i["name"] for i in result] == ["lo", "eth0", "eth1"] class TestDaemonGetRichRules: @patch("daemon.handlers.firewall.run") def test_single_rule(self, mock_run): mock_run.return_value = ( 'rule family="ipv4" port protocol="tcp" port="443" accept;' ) cfg_mock = MagicMock(return_value={"zones": {"public": {"rich_rules": []}}}) with patch.object(daemonfirewall, "_get_config", cfg_mock): result = daemonfirewall.list_rich_rules(None, {"zone": "public"}) assert len(result) == 1 @patch("daemon.handlers.firewall.run") def test_empty_rules(self, mock_run): mock_run.return_value = "" cfg_mock = MagicMock(return_value={"zones": {"public": {"rich_rules": []}}}) with patch.object(daemonfirewall, "_get_config", cfg_mock): result = daemonfirewall.list_rich_rules(None, {"zone": "public"}) assert result == [] @patch("daemon.handlers.firewall.run") def test_multiline_rule(self, mock_run): mock_run.return_value = ( 'rule family="ipv4"\n source address="10.0.0.0/24"\n reject;' ) cfg_mock = MagicMock(return_value={"zones": {"public": {"rich_rules": []}}}) with patch.object(daemonfirewall, "_get_config", cfg_mock): result = daemonfirewall.list_rich_rules(None, {"zone": "public"}) assert len(result) == 1 assert "10.0.0.0/24" in result[0]["rule"] class TestDaemonGetState: @patch("daemon.handlers.firewall.run") def test_returns_full_state(self, mock_run): def run_side_effect(args, **kwargs): if "--get-zones" in args: return "public\ninternal" if "--get-active-zones" in args: return "public\n eth0\ninternal\n eth1" if "--get-services" in args: return "ssh http dns" if "ip" in args[0]: if "link" in args: return "1: lo: mtu 65536\n2: eth0: mtu 1500 link/ether aa:bb:cc\n" if "addr" in args: return "2: eth0 inet 192.168.1.1/24 brd 192.168.1.255 scope global eth0\n" if "--list-all" in args: return ( "target: default\n" "interfaces: eth0\n" "sources: \n" "services: \n" "ports: \n" "protocols: \n" "forward-ports: \n" "masquerade: no\n" "ics: no\n" "rich-rules: \n" "icmp-blocks: \n" "module: \n" ) return "" mock_run.side_effect = run_side_effect result = daemonfirewall._get_state() 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"] class TestDaemonConfigApply: @patch("daemon.handlers.firewall._save_backup") @patch("daemon.handlers.firewall._get_state") @patch("daemon.handlers.firewall._get_lib_config") @patch("daemon.handlers.firewall.run") def test_applies_existing_zone(self, mock_run, mock_cfg, mock_state, mock_backup): mock_cfg.return_value = { "zones": { "public": { "target": "DEFAULT", "interfaces": ["eth0"], "services": ["http", "https"], "masquerade": True, }, }, } mock_run.return_value = ( "public\ninternal\ntarget: default\n" "interfaces: \n" "sources: \n" "services: \n" "ports: \n" "protocols: \n" "forward-ports: \n" "masquerade: no\n" "ics: no\n" "rich-rules: \n" "icmp-blocks: \n" "module: \n" ) mock_state.return_value = {"zones": {"public": {}}} mock_backup.return_value = "/tmp/rules.json" result = daemonfirewall._config_apply() assert result["applied_zones"] == ["public"] assert result["backup"] == "/tmp/rules.json" calls = [str(c) for c in mock_run.call_args_list] assert any("--add-service=" in c for c in calls) assert any("--add-interface=" in c for c in calls) @patch("daemon.handlers.firewall._save_backup") @patch("daemon.handlers.firewall._get_state") @patch("daemon.handlers.firewall._get_lib_config") @patch("daemon.handlers.firewall.run") def test_creates_new_zone(self, mock_run, mock_cfg, mock_state, mock_backup): mock_cfg.return_value = { "zones": { "custom": { "target": "ACCEPT", "interfaces": ["eth2"], "services": [], "masquerade": False, }, }, } mock_run.return_value = ( "public\ninternal\ntarget: default\n" "interfaces: \n" "sources: \n" "services: \n" "ports: \n" "protocols: \n" "forward-ports: \n" "masquerade: no\n" "ics: no\n" "rich-rules: \n" "icmp-blocks: \n" "module: \n" ) mock_state.return_value = {"zones": {"custom": {}}} mock_backup.return_value = "/tmp/rules.json" result = daemonfirewall._config_apply() assert result["applied_zones"] == ["custom"] @patch("daemon.handlers.firewall._save_backup") @patch("daemon.handlers.firewall._get_state") @patch("daemon.handlers.firewall._get_lib_config") @patch("daemon.handlers.firewall.run") def test_empty_config_no_ops(self, mock_run, mock_cfg, mock_state, mock_backup): mock_cfg.return_value = {"zones": {}} mock_run.return_value = "" mock_state.return_value = {"zones": {}} mock_backup.return_value = "/tmp/rules.json" result = daemonfirewall._config_apply() assert result["applied_zones"] == [] class TestDaemonConfigPending: @patch("daemon.handlers.firewall._get_state") @patch("lib.firewall.get_config") def test_detects_interface_drift(self, mock_cfg, mock_state): 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 = daemonfirewall.config_pending(None, None) assert result["needs_apply"] is True @patch("daemon.handlers.firewall._get_state") @patch("lib.firewall.get_config") def test_in_sync(self, mock_cfg, mock_state): 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 = daemonfirewall.config_pending(None, None) assert result["needs_apply"] is False # --------------------------------------------------------------------------- # 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", }, ) # --------------------------------------------------------------------------- # Forward port removal during config_apply # --------------------------------------------------------------------------- class TestDaemonConfigApplyForwardPorts: @patch("daemon.handlers.firewall._save_backup") @patch("daemon.handlers.firewall._get_state") @patch("daemon.handlers.firewall._get_lib_config") @patch("daemon.handlers.firewall.run") def test_removes_stale_forward_ports( self, mock_run, mock_cfg, mock_state, mock_backup ): mock_cfg.return_value = { "zones": { "public": { "interfaces": [], "services": [], "masquerade": False, "forward_ports": [ {"id": "fp_new", "port": 8443, "proto": "tcp"}, ], }, }, } mock_run.return_value = ( "public\ntarget: default\n" "interfaces: \n" "sources: \n" "services: \n" "ports: \n" "protocols: \n" "forward-ports: port=443/proto=tcp\n" "masquerade: no\n" "ics: no\n" "rich-rules: \n" "icmp-blocks: \n" "module: \n" ) mock_state.return_value = {"zones": {"public": {}}} mock_backup.return_value = "/tmp/rules.json" result = daemonfirewall._config_apply() assert result["applied_zones"] == ["public"] calls = [str(c) for c in mock_run.call_args_list] assert any("--remove-forward-port=" in c for c in calls) assert any("--add-forward-port=" in c for c in calls)