mteehan/vacuum-wall
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
master
vacuum-wall/tests/test_firewall.py
T
mteehan faa076370d refactor: daemon collectors, thin webui proxies, pure config reads 
- move state collectors from lib/state.py to daemon/collectors/ (7
  modules, registration side-effect; daemon/server.py imports the
  package before the first populate())
- webui/api: new daemon_route() decorator factory in common.py
  collapses the try/except daemon-proxy boilerplate in all 8
  blueprints (rules/params/body/transform keep responses identical)
- firewall: interface-coverage invariant — config is the source of
  truth for zone interfaces (absent key = empty, no hands-off
  zones); pure validate_coverage() enforced at save (400) and apply
  (409, force: true overrides), top-level `unmanaged` exemption
- lib: get_config() reads are now pure (no dir creation or writes);
  new lib/bootstrap.py creates runtime dirs and persists the
  one-shot nginx legacy migration at daemon start, after
  system_import (lib.nginx.migrate_config_file)
- lib/common: compute_pending() apply-bookkeeping helper
- daemon: emit_and_refresh() handler helper; refresh_state(bump=) so
  /status/refresh no longer bumps versions (poll/mutation only)
- acme: move --log last so acme.sh never treats a real arg as the
  log-file argument
- docs: AGENTS.md, config.md, state-model.md, api.md updated;
  HARDEN.md dropped (plan implemented); apply-confirm force wording

Tests: 917 passed; ruff check + format clean.
2026-09-03 00:40:56 +00:00

1799 lines
66 KiB
Python
Raw Permalink 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 common as daemoncommon
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: <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"]
)
_PENDING_LIVE_PUBLIC = {
"interfaces": ["eth0"],
"services": ["http"],
"masquerade": False,
"target": "default",
"rich-rules": [],
"forward-ports": [],
}
class TestComputePendingChangesAbsentInterfaces:
"""The config is the source of truth for zone interfaces: an absent
'interfaces' key counts as an empty list, so every config zone is
diffed on interfaces (no hands-off zones)."""
def test_services_and_interfaces_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
# Absent key counts as an empty list: live eth0 is a pending removal.
assert "interfaces" in types
def test_absent_key_zone_in_sync_live_reports_nothing(self):
# Config lacks the interfaces key and the live zone has no
# interfaces either — absent key equals the empty live set.
live = {**_PENDING_LIVE_PUBLIC, "interfaces": []}
cfg = {"zones": {"public": {"services": ["http"]}}}
result = firewall._compute_pending_changes(cfg, {"public": live})
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 TestValidateCoverage:
def test_all_covered(self):
fw = {"zones": {"public": {"interfaces": ["eth0"]}}}
net = {"interfaces": {"eth0": {}}}
assert firewall.validate_coverage(fw, net) == []
def test_uncovered_reported_sorted(self):
fw = {"zones": {"public": {"interfaces": ["eth1"]}}}
net = {"interfaces": {"eth5": {}, "eth0": {}}}
assert firewall.validate_coverage(fw, net) == ["eth0", "eth5"]
def test_unmanaged_exempts(self):
fw = {
"zones": {"public": {"interfaces": ["eth1"]}},
"unmanaged": ["eth0"],
}
net = {"interfaces": {"eth0": {}, "eth1": {}}}
assert firewall.validate_coverage(fw, net) == []
def test_lo_and_wg_exempt(self):
fw = {"zones": {}}
net = {"interfaces": {"lo": {}, "wg0": {}, "wg-full": {}}}
assert firewall.validate_coverage(fw, net) == []
def test_absent_key_counts_as_empty(self):
# A zone without an 'interfaces' key covers nothing.
fw = {"zones": {"public": {"services": ["http"]}}}
net = {"interfaces": {"eth0": {}}}
assert firewall.validate_coverage(fw, net) == ["eth0"]
def test_empty_network_config(self):
assert firewall.validate_coverage({"zones": {}}, {"interfaces": {}}) == []
assert firewall.validate_coverage({"zones": {}}, {}) == []
def test_non_dict_zone_and_non_list_unmanaged_ignored(self):
fw = {"zones": {"public": "oops"}, "unmanaged": "eth0"}
net = {"interfaces": {"eth0": {}}}
assert firewall.validate_coverage(fw, net) == ["eth0"]
def test_non_string_entries_ignored(self):
fw = {"zones": {"public": {"interfaces": [None, 7]}}, "unmanaged": [None]}
net = {"interfaces": {"eth0": {}}}
assert firewall.validate_coverage(fw, net) == ["eth0"]
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.common.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(daemoncommon, "bus") as mock_bus,
patch("daemon.handlers.common.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(daemoncommon, "bus") as mock_bus,
patch("daemon.handlers.common.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.common.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"]
# ---------------------------------------------------------------------------
# Coverage invariant: the config must cover every network-managed
# interface (or declare it unmanaged). Pure config check — the config is
# the source of truth for zone interfaces (absent key = empty), so there
# is no live-state comparison and no hands-off zones.
# ---------------------------------------------------------------------------
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.common.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 TestDaemonCoverageInvariant:
def test_conflict_when_network_iface_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") as mock_run,
patch("daemon.handlers.firewall._default_zone", return_value="internal"),
patch("daemon.handlers.firewall._save_backup") as mock_backup,
pytest.raises(ConflictError) as exc,
):
daemonfirewall._config_apply()
msg = str(exc.value)
assert "eth0" in msg
assert "unmanaged" in msg
assert "force" in msg
mock_backup.assert_not_called()
# Pure config check: the guard performs no live-state reads at all.
mock_run.assert_not_called()
def test_unmanaged_exempts_iface(self):
cfg = {
"zones": {"public": {"interfaces": ["eth1"], "services": []}},
"unmanaged": ["eth0"],
}
result, mock_run = _apply_with(cfg, {"eth0": {}}, "public\n eth0\n")
assert result["applied_zones"] == ["public"]
cmds = [c.args[0] for c in mock_run.call_args_list]
# Live eth0 is removed, config eth1 added exactly once (permanent).
assert [
"firewall-cmd",
"--zone=public",
"--remove-interface=eth0",
"--permanent",
] in cmds
assert (
cmds.count(
["firewall-cmd", "--zone=public", "--add-interface=eth1", "--permanent"]
)
== 1
)
def test_absent_key_zone_counts_as_empty(self):
# No hands-off zones: a zone without an 'interfaces' key covers
# nothing, so a managed interface left out of every zone blocks.
cfg = {"zones": {"public": {"services": ["http"], "masquerade": False}}}
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"),
patch("daemon.handlers.firewall._default_zone", return_value="internal"),
pytest.raises(ConflictError) as exc,
):
daemonfirewall._config_apply()
assert "eth0" in str(exc.value)
def test_live_only_zone_does_not_count_as_covered(self):
# eth1 is held by 'guest' live but the config (the source of
# truth) does not cover it — apply is blocked regardless of live.
cfg = {"zones": {"public": {"interfaces": [], "services": []}}}
with (
patch("lib.network.get_config", return_value={"interfaces": {"eth1": {}}}),
patch("lib.firewall.get_config", return_value=cfg, create=True),
patch(
"daemon.handlers.firewall.run",
side_effect=_make_run("public\n eth0\nguest\n eth1\n"),
),
patch("daemon.handlers.firewall._default_zone", return_value="internal"),
pytest.raises(ConflictError),
):
daemonfirewall._config_apply()
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_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
)
class TestDaemonSaveCoverageValidation:
"""The coverage invariant is enforced at save time too (POST/PATCH
/firewall/config), so bad configs are rejected before they are written."""
def test_save_blocks_uncovered(self):
body = {"zones": {"public": {"interfaces": ["eth1"], "services": []}}}
with (
patch("lib.network.get_config", return_value={"interfaces": {"eth0": {}}}),
patch("daemon.handlers.firewall._save_config") as mock_save,
pytest.raises(ValueError) as exc,
):
daemonfirewall.save_config_handler(None, body)
assert "eth0" in str(exc.value)
assert "unmanaged" in str(exc.value)
mock_save.assert_not_called()
def test_save_allows_unmanaged(self):
body = {
"zones": {"public": {"interfaces": ["eth1"], "services": []}},
"unmanaged": ["eth0"],
}
with (
patch("lib.network.get_config", return_value={"interfaces": {"eth0": {}}}),
patch("daemon.handlers.firewall._save_config") as mock_save,
patch.object(daemoncommon, "bus") as mock_bus,
patch("daemon.handlers.common.refresh_state"),
):
mock_bus.emit.return_value = MagicMock(affected_subsystems=[])
result = daemonfirewall.save_config_handler(None, body)
assert result == {"config_saved": True}
mock_save.assert_called_once()
def test_save_rejects_non_list_unmanaged(self):
body = {"zones": {"public": {"interfaces": ["eth0"]}}, "unmanaged": "eth0"}
with (
patch("lib.network.get_config", return_value={"interfaces": {"eth0": {}}}),
pytest.raises(ValueError) as exc,
):
daemonfirewall.save_config_handler(None, body)
assert "unmanaged" in str(exc.value)
def test_patch_blocks_merge_that_uncovers(self):
current = {"zones": {"public": {"interfaces": ["eth0"], "services": []}}}
body = {"zones": {"public": {"interfaces": ["eth1"]}}}
with (
patch("lib.network.get_config", return_value={"interfaces": {"eth0": {}}}),
patch(
"daemon.handlers.firewall._get_config", return_value=deepcopy(current)
),
patch("daemon.handlers.firewall._save_config") as mock_save,
pytest.raises(ValueError) as exc,
):
daemonfirewall.patch_config(None, body)
assert "eth0" in str(exc.value)
mock_save.assert_not_called()
def test_patch_allows_merge_that_covers(self):
current = {"zones": {"public": {"interfaces": ["eth0"], "services": []}}}
body = {"zones": {"internal": {"interfaces": ["eth1"], "services": []}}}
with (
patch("lib.network.get_config", return_value={"interfaces": {"eth0": {}}}),
patch(
"daemon.handlers.firewall._get_config", return_value=deepcopy(current)
),
patch("daemon.handlers.firewall._save_config") as mock_save,
patch.object(daemoncommon, "bus") as mock_bus,
patch("daemon.handlers.common.refresh_state"),
):
mock_bus.emit.return_value = MagicMock(affected_subsystems=[])
result = daemonfirewall.patch_config(None, body)
assert result == {"config_saved": True}
saved = mock_save.call_args[0][0]
assert set(saved["zones"]) == {"public", "internal"}
# ---------------------------------------------------------------------------
# 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(daemoncommon, "bus") as mock_bus,
patch("daemon.handlers.common.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.common.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.common.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 TestDaemonMutatorBaselineStamp:
"""Per-zone mutations apply to live firewalld immediately and must
re-stamp the applied baseline, so cancel-all reverts to the post-mutation
state instead of an older snapshot (regression: stale install-era
snapshot resurrected a phantom 'remove interface' pending change).
"""
ZONES_OUT = "public\ninternal"
@patch("daemon.handlers.firewall._reload")
@patch("daemon.handlers.firewall._get_config", return_value={"zones": {}})
@patch("daemon.handlers.firewall.run", return_value=ZONES_OUT)
def test_set_zone_interfaces_stamps_baseline(self, mock_run, mock_cfg, mock_reload):
with (
patch.object(daemonfirewall, "_save_config") as mock_save,
patch.object(daemoncommon, "bus") as mock_bus,
patch("daemon.handlers.common.refresh_state"),
):
mock_bus.emit.return_value = MagicMock(affected_subsystems=[])
daemonfirewall.set_zone_interfaces(
None, {"zone": "internal", "interfaces": ["eth1"]}
)
saved = mock_save.call_args[0][0]
assert saved["zones"]["internal"]["interfaces"] == ["eth1"]
assert saved[_LAST_APPLIED_CONFIG_KEY] == strip_apply_meta(saved)
assert saved[_APPLY_HASH_KEY] == config_hash(saved)
assert saved[_LAST_APPLIED_CONFIG_KEY]["zones"]["internal"]["interfaces"] == [
"eth1"
]
@patch("daemon.handlers.firewall._reload")
@patch("daemon.handlers.firewall._get_config", return_value={"zones": {}})
@patch("daemon.handlers.firewall.run", return_value=ZONES_OUT)
def test_set_zone_services_stamps_baseline(self, mock_run, mock_cfg, mock_reload):
with (
patch.object(
daemonfirewall, "_parse_zone_output", return_value={"services": []}
),
patch.object(daemonfirewall, "_save_config") as mock_save,
patch.object(daemoncommon, "bus") as mock_bus,
patch("daemon.handlers.common.refresh_state"),
):
mock_bus.emit.return_value = MagicMock(affected_subsystems=[])
daemonfirewall.set_zone_services(
None, {"zone": "internal", "services": ["ssh"]}
)
saved = mock_save.call_args[0][0]
assert saved["zones"]["internal"]["services"] == ["ssh"]
assert saved[_LAST_APPLIED_CONFIG_KEY]["zones"]["internal"]["services"] == [
"ssh"
]
@patch("daemon.handlers.firewall._reload")
@patch(
"daemon.handlers.firewall._get_config",
return_value={"zones": {"internal": {}}},
)
@patch("daemon.handlers.firewall.run")
def test_set_masquerade_syncs_config_and_stamps(
self, mock_run, mock_cfg, mock_reload
):
with (
patch.object(daemonfirewall, "_save_config") as mock_save,
patch.object(daemoncommon, "bus") as mock_bus,
patch("daemon.handlers.common.refresh_state"),
):
mock_bus.emit.return_value = MagicMock(affected_subsystems=[])
daemonfirewall.set_masquerade(None, {"zone": "internal", "enable": True})
saved = mock_save.call_args[0][0]
assert saved["zones"]["internal"]["masquerade"] is True
assert saved[_LAST_APPLIED_CONFIG_KEY] == strip_apply_meta(saved)
assert saved[_APPLY_HASH_KEY] == config_hash(saved)
@patch("daemon.handlers.firewall._reload")
@patch("daemon.handlers.firewall._get_config", return_value={"zones": {}})
@patch("daemon.handlers.firewall.run")
def test_set_masquerade_no_config_entry_skips_write(
self, mock_run, mock_cfg, mock_reload
):
"""A zone absent from the config must not gain a bare entry — that
would manufacture spurious service diffs on the next poll."""
with (
patch.object(daemonfirewall, "_save_config") as mock_save,
patch.object(daemoncommon, "bus") as mock_bus,
patch("daemon.handlers.common.refresh_state"),
):
mock_bus.emit.return_value = MagicMock(affected_subsystems=[])
daemonfirewall.set_masquerade(None, {"zone": "public", "enable": False})
mock_save.assert_not_called()
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.common.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.common.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",
"<service><short>OpenSSH</short><description>Remote login</description></service>",
)
_write_service(
d1,
"http",
"<service><short>WWW</short><description>Web server</description></service>",
)
_write_service(
d2,
"custom",
"<service><description>Only a long description</description></service>",
)
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",
"<service><description>Time synchronization</description></service>",
)
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", "<service><short>Built-in SSH</short></service>")
_write_service(etc, "ssh", "<service><short>Custom SSH</short></service>")
# 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", "<service><short>Unclosed")
_write_service(d, "good", "<service><short>Works</short></service>")
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", "<service></service>")
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", "<service><short>A1</short></service>")
result1 = firewall.get_service_descriptions([d1])
_write_service(d2, "a", "<service><short>A2</short></service>")
result2 = firewall.get_service_descriptions([d1, d2])
assert result1 == {"a": "A1"}
assert result2 == {"a": "A2"}
Reference in New Issue View Git Blame Copy Permalink