mteehan/vacuum-wall
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
55309cfd868c4f61331513e4d23e0ee19983e2fa
vacuum-wall/tests/test_firewall.py
T
mteehan 55309cfd86 status: cancel-all reverts pending changes to last applied config 
- lib.common.revert_to_applied(): restore a config file from its
  _last_applied_config snapshot (stamped hash); no baseline -> skip with
  reason, file untouched
- firewall config_apply now stamps the applied baseline like the other
  subsystems; GET /firewall/config and the state collector strip the
  internal _last_applied_* keys
- POST /status/cancel-all + /api/status/cancel-all: revert pending
  subsystems, {cancelled, skipped, errors}, partial-failure safe
- dashboard: "Cancel All Changes" button with confirm modal
  (CancelConfirm, reuses the pending-changes modal rows); the pending
  changes card is hidden entirely when nothing is pending
- tests: revert_to_applied, status_cancel_all, firewall stamping/meta
  stripping, /api/status/cancel-all route, node tests for CancelConfirm;
  firewall _config_apply tests no longer write the real repo config
- docs: api.md, state-model.md, config.md, hoover.md
2026-08-21 02:10:05 +00:00

1104 lines
38 KiB
Python
Raw Blame History

"""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
class TestParseInterfaces:
def test_lib_parses_interfaces(self):
result = firewall._parse_interfaces(
"1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536\n"
"2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> 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"]
)
# ---------------------------------------------------------------------------
# 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(
"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"]
_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(
"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_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}"
Reference in New Issue View Git Blame Copy Permalink