mteehan/vacuum-wall
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
575cf06a4b1c386a81e17988378660965489ce0e
vacuum-wall/tests/test_firewall.py
T
mteehan 9088f34345 sync: add cross-subsystem event bus for config consistency 
Add EventBus with loop guards to keep firewall, dnsmasq, wireguard,
and network configs consistent. Handlers emit SyncEvent after mutations;
subscribers compute diffs and write JSON without manual cascade loops.
2026-06-30 01:18:44 +00:00

636 lines
21 KiB
Python
Raw Blame History

"""Tests for lib/firewall.py (pure logic) and daemon/handlers/firewall.py (privilege boundary)."""
from unittest.mock import 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": []}
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):
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 (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"),
):
result = daemonfirewall._config_apply()
assert result["applied_zones"] == ["public"]
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
# ---------------------------------------------------------------------------
# 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