Add declarative firewall config with save-then-apply workflow
New two-step config flow: POST /config saves desired state to config/firewall/config.json, GET /config/pending diffs against live firewalld state, POST /config/apply synchronizes live state. Adds target normalization helpers and full test coverage for config CRUD and pending diff logic.
This commit is contained in:
@@ -159,3 +159,286 @@ class TestGetState:
|
||||
assert "zones" in result
|
||||
assert "active_zones" in result
|
||||
assert "timestamp" in result
|
||||
|
||||
|
||||
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"
|
||||
|
||||
|
||||
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.config_get()
|
||||
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.config_set({"zones": {"test": {"interfaces": ["eth0"]}}})
|
||||
import json as _json
|
||||
|
||||
content = _json.loads(cfg_file.read_text())
|
||||
assert content["zones"]["test"]["interfaces"] == ["eth0"]
|
||||
|
||||
|
||||
class TestConfigApply:
|
||||
@patch("lib.firewall.config_get")
|
||||
@patch("lib.firewall.save_backup")
|
||||
@patch("lib.firewall.get_available_zones")
|
||||
@patch("lib.firewall.create_zone")
|
||||
@patch("lib.firewall.set_zone_services")
|
||||
@patch("lib.firewall.set_zone_interfaces")
|
||||
@patch("lib.firewall.set_masquerade")
|
||||
@patch("lib.firewall._reload")
|
||||
def test_applies_existing_zone(
|
||||
self,
|
||||
mock_reload,
|
||||
mock_set_mq,
|
||||
mock_set_ifaces,
|
||||
mock_set_svcs,
|
||||
mock_create,
|
||||
mock_available,
|
||||
mock_backup,
|
||||
mock_cfg,
|
||||
):
|
||||
mock_cfg.return_value = {
|
||||
"zones": {
|
||||
"public": {
|
||||
"target": "DEFAULT",
|
||||
"interfaces": ["eth0"],
|
||||
"services": ["http", "https"],
|
||||
"masquerade": True,
|
||||
},
|
||||
},
|
||||
}
|
||||
mock_available.return_value = ["public", "internal"]
|
||||
mock_backup.return_value = "/tmp/rules.json"
|
||||
result = firewall.config_apply()
|
||||
assert result["applied_zones"] == ["public"]
|
||||
assert result["backup"] == "/tmp/rules.json"
|
||||
mock_set_ifaces.assert_called_once_with("public", ["eth0"])
|
||||
mock_set_svcs.assert_called_once_with("public", ["http", "https"])
|
||||
mock_set_mq.assert_called_once_with("public", True)
|
||||
|
||||
@patch("lib.firewall.config_get")
|
||||
@patch("lib.firewall.save_backup")
|
||||
@patch("lib.firewall.get_available_zones")
|
||||
@patch("lib.firewall.create_zone")
|
||||
@patch("lib.firewall.set_zone_services")
|
||||
@patch("lib.firewall.set_zone_interfaces")
|
||||
@patch("lib.firewall.set_masquerade")
|
||||
@patch("lib.firewall._reload")
|
||||
def test_creates_new_zone(
|
||||
self,
|
||||
mock_reload,
|
||||
mock_set_mq,
|
||||
mock_set_ifaces,
|
||||
mock_set_svcs,
|
||||
mock_create,
|
||||
mock_available,
|
||||
mock_backup,
|
||||
mock_cfg,
|
||||
):
|
||||
mock_cfg.return_value = {
|
||||
"zones": {
|
||||
"custom": {
|
||||
"target": "ACCEPT",
|
||||
"interfaces": ["eth2"],
|
||||
"services": [],
|
||||
"masquerade": False,
|
||||
},
|
||||
},
|
||||
}
|
||||
mock_available.return_value = ["public", "internal"]
|
||||
mock_backup.return_value = "/tmp/rules.json"
|
||||
result = firewall.config_apply()
|
||||
assert result["applied_zones"] == ["custom"]
|
||||
mock_create.assert_called_once_with("custom", "ACCEPT")
|
||||
mock_set_ifaces.assert_called_once_with("custom", ["eth2"])
|
||||
|
||||
|
||||
class TestConfigPending:
|
||||
@patch("lib.firewall.config_get")
|
||||
@patch("lib.firewall.get_state")
|
||||
def test_detects_interface_drift(self, mock_state, mock_cfg):
|
||||
mock_cfg.return_value = {
|
||||
"zones": {
|
||||
"public": {
|
||||
"interfaces": ["eth0"],
|
||||
"services": ["http"],
|
||||
"masquerade": False,
|
||||
},
|
||||
},
|
||||
}
|
||||
mock_state.return_value = {
|
||||
"zones": {
|
||||
"public": {
|
||||
"interfaces": ["eth1"],
|
||||
"services": ["http"],
|
||||
"masquerade": False,
|
||||
},
|
||||
},
|
||||
}
|
||||
result = firewall.config_pending()
|
||||
assert result["needs_apply"] is True
|
||||
assert any(c["type"] == "interfaces" for c in result["pending"])
|
||||
|
||||
@patch("lib.firewall.config_get")
|
||||
@patch("lib.firewall.get_state")
|
||||
def test_in_sync(self, mock_state, mock_cfg):
|
||||
mock_cfg.return_value = {
|
||||
"zones": {
|
||||
"public": {
|
||||
"interfaces": ["eth0"],
|
||||
"services": ["http"],
|
||||
"masquerade": False,
|
||||
},
|
||||
},
|
||||
}
|
||||
mock_state.return_value = {
|
||||
"zones": {
|
||||
"public": {
|
||||
"interfaces": ["eth0"],
|
||||
"services": ["http"],
|
||||
"masquerade": False,
|
||||
},
|
||||
},
|
||||
}
|
||||
result = firewall.config_pending()
|
||||
assert result["needs_apply"] is False
|
||||
|
||||
@patch("lib.firewall.config_get")
|
||||
@patch("lib.firewall.get_state")
|
||||
def test_detects_services_drift(self, mock_state, mock_cfg):
|
||||
mock_cfg.return_value = {
|
||||
"zones": {
|
||||
"public": {
|
||||
"interfaces": ["eth0"],
|
||||
"services": ["http", "ssh"],
|
||||
"masquerade": False,
|
||||
},
|
||||
},
|
||||
}
|
||||
mock_state.return_value = {
|
||||
"zones": {
|
||||
"public": {
|
||||
"interfaces": ["eth0"],
|
||||
"services": ["http"],
|
||||
"masquerade": False,
|
||||
},
|
||||
},
|
||||
}
|
||||
result = firewall.config_pending()
|
||||
assert any(c["type"] == "services" for c in result["pending"])
|
||||
|
||||
@patch("lib.firewall.config_get")
|
||||
@patch("lib.firewall.get_state")
|
||||
def test_detects_unmanaged_zones(self, mock_state, mock_cfg):
|
||||
mock_cfg.return_value = {"zones": {}}
|
||||
mock_state.return_value = {
|
||||
"zones": {
|
||||
"public": {
|
||||
"interfaces": ["eth0"],
|
||||
"services": [],
|
||||
"masquerade": False,
|
||||
},
|
||||
},
|
||||
}
|
||||
result = firewall.config_pending()
|
||||
assert "public" in result["unmanaged_zones"]
|
||||
|
||||
|
||||
class TestConfigEmptyZones:
|
||||
@patch("lib.firewall.config_get")
|
||||
@patch("lib.firewall.save_backup")
|
||||
@patch("lib.firewall.get_available_zones")
|
||||
@patch("lib.firewall.create_zone")
|
||||
@patch("lib.firewall.set_zone_services")
|
||||
@patch("lib.firewall.set_zone_interfaces")
|
||||
@patch("lib.firewall.set_masquerade")
|
||||
@patch("lib.firewall._reload")
|
||||
def test_empty_config_no_ops(
|
||||
self,
|
||||
mock_reload,
|
||||
mock_set_mq,
|
||||
mock_set_ifaces,
|
||||
mock_set_svcs,
|
||||
mock_create,
|
||||
mock_available,
|
||||
mock_backup,
|
||||
mock_cfg,
|
||||
):
|
||||
mock_cfg.return_value = {"zones": {}}
|
||||
mock_available.return_value = []
|
||||
mock_backup.return_value = "/tmp/rules.json"
|
||||
result = firewall.config_apply()
|
||||
assert result["applied_zones"] == []
|
||||
mock_create.assert_not_called()
|
||||
mock_set_ifaces.assert_not_called()
|
||||
|
||||
Reference in New Issue
Block a user