mteehan
75b86fd60d
acme:
- acme.sh chmods its tree to owner-only (700/600) every run, which
broke the two-user model: a tree left owner-only by one user made
every acme.sh call of the other exit 2
- normalize_acme_home() reopens group access (sudo chmod g+rwX,
files only — setgid dirs trip RestrictSUIDSGID); _run_acme_preflight
is the choke point before every daemon acme.sh call + startup
- acme service now runs as the daemon user; --log persists the raw CA
transcript; SYS_LOG=6 journals manual issue/renew runs
- timer daily-only: two runs/day landed inside ZeroSSL's 24h
validation backoff (Retry-After: 86400) — a permanent renewal lockout
- _collect_acme no longer raises on cert-list failure; reports
status.error (AcmeState.status) so the certs page can surface it
firewall: re-stamp the applied baseline on live zone mutations
(interfaces/services/rich-rules/masquerade/forward-ports) so cancel-all
reverts to post-mutation state, not a stale install-era snapshot;
set_masquerade syncs the declarative config for existing zones;
add_forward_port records toaddr only with toport
status: apply-all accepts {"force": true} (forwarded to the firewall
apply only); ApplyConfirm force checkbox; applyResultToasts() — the
errors map wins over the 200; ActionButton checks errors before the
success toast; dashboard uses ApplyConfirm
system_import: drift re-imports carry the existing apply-meta; first
import stamps the adopted content as applied (it is the running state)
— no phantom pending changes
nginx: get_config only re-saves when migration actually changed the
config (no more owner/mtime churn on every read)
install: repair mis-owned top-level system dirs (tmpfiles
unsafe-path-transition), warn with a full-repair command for deeper
mis-ownership
daemon/server: loop.get_exception_handler() (aiohttp API fix)
tests: 888 pytest + 24 node passing; ruff clean
438 lines
16 KiB
Python
438 lines
16 KiB
Python
"""Tests for daemon/handlers/status.py — aggregate pending + apply-all."""
|
|
|
|
from typing import Any, ClassVar
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from daemon.handlers import status
|
|
from lib.state import State
|
|
|
|
|
|
def _make_state(**kwargs):
|
|
"""Create a minimal in-memory state for pending checks."""
|
|
st = State()
|
|
for name, data in kwargs.items():
|
|
st.set(name, data)
|
|
return st
|
|
|
|
|
|
def _mock_state_store(state_dict):
|
|
"""Return a mock that looks like state_store.get()."""
|
|
mock = MagicMock()
|
|
mock.get.side_effect = lambda name: state_dict.get(name)
|
|
return mock
|
|
|
|
|
|
class TestFwChangeSummary:
|
|
"""Test fw_change_summary helper from status module."""
|
|
|
|
def test_interfaces_summary(self):
|
|
s = status.fw_change_summary(
|
|
"internal", "interfaces", {"config": ["eth1"], "live": []}
|
|
)
|
|
assert "Zone internal: interfaces changed" in s
|
|
assert "eth1" in s
|
|
|
|
def test_services_summary(self):
|
|
s = status.fw_change_summary(
|
|
"dmz", "services", {"config": ["ssh", "dns"], "live": ["ssh"]}
|
|
)
|
|
assert "Zone dmz: services changed" in s
|
|
|
|
def test_rich_rules_summary(self):
|
|
s = status.fw_change_summary(
|
|
"public", "rich_rules", {"config_count": 2, "live_count": 1}
|
|
)
|
|
assert "config: 2" in s
|
|
assert "live: 1" in s
|
|
|
|
def test_forward_ports_summary(self):
|
|
s = status.fw_change_summary(
|
|
"wan", "forward_ports", {"config_count": 3, "live_count": 0}
|
|
)
|
|
assert "Zone wan: port forwards differ" in s
|
|
|
|
def test_masquerade_summary(self):
|
|
s = status.fw_change_summary(
|
|
"lan", "masquerade", {"config": True, "live": False}
|
|
)
|
|
assert "Zone lan: masquerade changed" in s
|
|
|
|
def test_target_summary(self):
|
|
s = status.fw_change_summary(
|
|
"vpn", "target", {"config": "ACCEPT", "live": "default"}
|
|
)
|
|
assert "Zone vpn: target changed" in s
|
|
|
|
def test_unknown_type_summary(self):
|
|
s = status.fw_change_summary("public", "weird", {})
|
|
assert "Zone public: weird changed" in s
|
|
|
|
|
|
class TestHashSubsystem:
|
|
"""Test _hash_subsystem helper from status module."""
|
|
|
|
def test_no_state(self):
|
|
result = status._hash_subsystem("nginx", None)
|
|
assert result["pending_changes"] is False
|
|
assert result["summary"] == "Up to date"
|
|
|
|
def test_pending_true(self):
|
|
st = {"status": {"pending_changes": True}}
|
|
result = status._hash_subsystem("wireguard", st)
|
|
assert result["pending_changes"] is True
|
|
assert "unapplied changes" in result["summary"]
|
|
assert len(result["changes"]) == 1
|
|
|
|
def test_pending_false(self):
|
|
st = {"status": {"pending_changes": False}}
|
|
result = status._hash_subsystem("networkd", st)
|
|
assert result["pending_changes"] is False
|
|
|
|
def test_empty_status(self):
|
|
st = {}
|
|
result = status._hash_subsystem("dnsmasq", st)
|
|
assert result["pending_changes"] is False
|
|
|
|
|
|
class TestStatusPending:
|
|
"""Test the aggregate pending endpoint."""
|
|
|
|
@patch("daemon.handlers.status.state_store")
|
|
def test_all_synced(self, mock_store):
|
|
mock_store.get.return_value = {
|
|
"firewall": {"pending": {"needs_apply": False, "pending": []}},
|
|
"dnsmasq": {"status": {"pending_changes": False}},
|
|
"nginx": {"status": {"pending_changes": False}},
|
|
"wireguard": {"status": {"pending_changes": False}},
|
|
"networkd": {"status": {"pending_changes": False}},
|
|
}
|
|
result = status.status_pending(None, None)
|
|
assert result["total_changes"] == 0
|
|
assert not result["firewall"]["needs_apply"]
|
|
assert not result["dnsmasq"]["pending_changes"]
|
|
|
|
def _patch_store(self, data):
|
|
mock = MagicMock()
|
|
mock.get.side_effect = lambda name: data.get(name)
|
|
return patch("daemon.handlers.status.state_store", mock)
|
|
|
|
def test_firewall_pending_only(self):
|
|
with self._patch_store(
|
|
{
|
|
"firewall": {
|
|
"pending": {
|
|
"needs_apply": True,
|
|
"pending": [
|
|
{
|
|
"zone": "internal",
|
|
"type": "interfaces",
|
|
"config": ["eth1"],
|
|
"live": [],
|
|
}
|
|
],
|
|
}
|
|
},
|
|
"dnsmasq": {"status": {"pending_changes": False}},
|
|
"nginx": {"status": {"pending_changes": False}},
|
|
"wireguard": {"status": {"pending_changes": False}},
|
|
"networkd": {"status": {"pending_changes": False}},
|
|
}
|
|
):
|
|
result = status.status_pending(None, None)
|
|
assert result["total_changes"] == 1
|
|
assert result["firewall"]["change_count"] == 1
|
|
|
|
def test_multiple_subsystems_pending(self):
|
|
with self._patch_store(
|
|
{
|
|
"firewall": {
|
|
"pending": {
|
|
"needs_apply": True,
|
|
"pending": [
|
|
{
|
|
"zone": "lan",
|
|
"type": "services",
|
|
"config": ["ssh"],
|
|
"live": [],
|
|
},
|
|
{
|
|
"zone": "wan",
|
|
"type": "masquerade",
|
|
"config": True,
|
|
"live": False,
|
|
},
|
|
],
|
|
}
|
|
},
|
|
"dnsmasq": {"status": {"pending_changes": True}},
|
|
"nginx": {"status": {"pending_changes": False}},
|
|
"wireguard": {"status": {"pending_changes": True}},
|
|
"networkd": {"status": {"pending_changes": False}},
|
|
}
|
|
):
|
|
result = status.status_pending(None, None)
|
|
assert result["total_changes"] == 4 # 2 FW + 1 DHCP + 1 WG
|
|
assert result["firewall"]["change_count"] == 2
|
|
assert result["firewall"]["needs_apply"] is True
|
|
assert result["dnsmasq"]["pending_changes"] is True
|
|
assert result["wireguard"]["pending_changes"] is True
|
|
|
|
def test_empty_state(self):
|
|
with self._patch_store({}):
|
|
result = status.status_pending(None, None)
|
|
assert result["total_changes"] == 0
|
|
assert not result["firewall"]["needs_apply"]
|
|
|
|
def test_firewall_uncovered_advisory_not_counted(self):
|
|
with self._patch_store(
|
|
{
|
|
"firewall": {
|
|
"pending": {"needs_apply": False, "pending": []},
|
|
"uncovered_interfaces": ["eth1"],
|
|
},
|
|
"dnsmasq": {"status": {"pending_changes": False}},
|
|
"nginx": {"status": {"pending_changes": False}},
|
|
"wireguard": {"status": {"pending_changes": False}},
|
|
"networkd": {"status": {"pending_changes": False}},
|
|
}
|
|
):
|
|
result = status.status_pending(None, None)
|
|
assert result["firewall"]["uncovered_interfaces"] == ["eth1"]
|
|
assert result["firewall"]["coverage_warnings"]
|
|
assert "eth1" in result["firewall"]["coverage_warnings"][0]
|
|
# Advisory: must not flip needs_apply or count as a change.
|
|
assert not result["firewall"]["needs_apply"]
|
|
assert result["firewall"]["change_count"] == 0
|
|
assert result["total_changes"] == 0
|
|
|
|
def test_firewall_no_uncovered_no_warnings(self):
|
|
with self._patch_store(
|
|
{
|
|
"firewall": {
|
|
"pending": {"needs_apply": False, "pending": []},
|
|
"uncovered_interfaces": [],
|
|
},
|
|
"dnsmasq": {"status": {"pending_changes": False}},
|
|
"nginx": {"status": {"pending_changes": False}},
|
|
"wireguard": {"status": {"pending_changes": False}},
|
|
"networkd": {"status": {"pending_changes": False}},
|
|
}
|
|
):
|
|
result = status.status_pending(None, None)
|
|
assert result["firewall"]["uncovered_interfaces"] == []
|
|
assert result["firewall"]["coverage_warnings"] == []
|
|
assert result["total_changes"] == 0
|
|
|
|
def test_firewall_missing_uncovered_key_defaults_empty(self):
|
|
with self._patch_store(
|
|
{
|
|
"firewall": {
|
|
"pending": {"needs_apply": False, "pending": []},
|
|
},
|
|
"dnsmasq": None,
|
|
"nginx": None,
|
|
"wireguard": None,
|
|
"networkd": None,
|
|
}
|
|
):
|
|
result = status.status_pending(None, None)
|
|
assert result["firewall"]["uncovered_interfaces"] == []
|
|
assert result["firewall"]["coverage_warnings"] == []
|
|
|
|
def test_firewall_no_pending_key(self):
|
|
with self._patch_store(
|
|
{
|
|
"firewall": {},
|
|
"dnsmasq": {"status": {"pending_changes": False}},
|
|
"nginx": None,
|
|
"wireguard": None,
|
|
"networkd": None,
|
|
}
|
|
):
|
|
result = status.status_pending(None, None)
|
|
assert result["total_changes"] == 0
|
|
assert not result["firewall"]["needs_apply"]
|
|
|
|
|
|
class TestStatusApplyAll:
|
|
"""Test the apply-all endpoint.
|
|
|
|
Patches SYS_APPLY dict entries directly since they hold function
|
|
references at import time.
|
|
"""
|
|
|
|
_fake_pending_all: ClassVar[dict[str, Any]] = {
|
|
"firewall": {"needs_apply": False, "change_count": 0, "changes": []},
|
|
"dnsmasq": {"pending_changes": False, "summary": "Up to date", "changes": []},
|
|
"nginx": {"pending_changes": False, "summary": "Up to date", "changes": []},
|
|
"wireguard": {"pending_changes": False, "summary": "Up to date", "changes": []},
|
|
"networkd": {"pending_changes": False, "summary": "Up to date", "changes": []},
|
|
}
|
|
|
|
@patch("daemon.handlers.status.status_pending")
|
|
@patch("daemon.handlers.status.refresh_state")
|
|
def test_nothing_to_apply(self, mock_refresh, mock_pending):
|
|
mock_pending.return_value = self._fake_pending_all
|
|
result = status.status_apply_all(None, None)
|
|
assert result["applied"] == []
|
|
assert result["errors"] == {}
|
|
mock_refresh.assert_called_once()
|
|
|
|
def test_applies_pending_subsystems(self):
|
|
mock_net = MagicMock()
|
|
mock_fw = MagicMock()
|
|
|
|
pending_data = {**self._fake_pending_all}
|
|
pending_data["firewall"]["needs_apply"] = True
|
|
pending_data["firewall"]["change_count"] = 1
|
|
pending_data["networkd"]["pending_changes"] = True
|
|
|
|
with (
|
|
patch("daemon.handlers.status.status_pending", return_value=pending_data),
|
|
patch("daemon.handlers.status.refresh_state"),
|
|
patch.dict(
|
|
"daemon.handlers.status.SYS_APPLY",
|
|
{
|
|
"networkd": mock_net,
|
|
"firewall": mock_fw,
|
|
},
|
|
),
|
|
):
|
|
result = status.status_apply_all(None, None)
|
|
assert "networkd" in result["applied"]
|
|
assert "firewall" in result["applied"]
|
|
mock_net.assert_called_once()
|
|
mock_fw.assert_called_once()
|
|
|
|
def test_error_in_subsystem(self):
|
|
mock_fw = MagicMock(side_effect=RuntimeError("firewalld not running"))
|
|
|
|
pending_data = {**self._fake_pending_all}
|
|
pending_data["firewall"]["needs_apply"] = True
|
|
pending_data["firewall"]["change_count"] = 1
|
|
|
|
with (
|
|
patch("daemon.handlers.status.status_pending", return_value=pending_data),
|
|
patch("daemon.handlers.status.refresh_state"),
|
|
patch.dict("daemon.handlers.status.SYS_APPLY", {"firewall": mock_fw}),
|
|
):
|
|
result = status.status_apply_all(None, None)
|
|
assert "firewall" not in result["applied"]
|
|
assert "Firewall" in result["errors"]
|
|
assert "firewalld not running" in result["errors"]["Firewall"]
|
|
|
|
def test_order_is_respected(self):
|
|
call_order = []
|
|
|
|
def track(name):
|
|
def wrapper(*args):
|
|
call_order.append(name)
|
|
|
|
return wrapper
|
|
|
|
mock_net = MagicMock(side_effect=track("networkd"))
|
|
mock_wg = MagicMock(side_effect=track("wireguard"))
|
|
|
|
pending_data = {**self._fake_pending_all}
|
|
pending_data["wireguard"]["pending_changes"] = True
|
|
pending_data["networkd"]["pending_changes"] = True
|
|
|
|
with (
|
|
patch("daemon.handlers.status.status_pending", return_value=pending_data),
|
|
patch("daemon.handlers.status.refresh_state"),
|
|
patch.dict(
|
|
"daemon.handlers.status.SYS_APPLY",
|
|
{
|
|
"networkd": mock_net,
|
|
"wireguard": mock_wg,
|
|
},
|
|
),
|
|
):
|
|
status.status_apply_all(None, None)
|
|
assert call_order == ["networkd", "wireguard"]
|
|
|
|
def test_partial_failure_still_applies_others(self):
|
|
mock_fw = MagicMock(side_effect=RuntimeError("fail"))
|
|
mock_nginx = MagicMock()
|
|
|
|
pending_data = {**self._fake_pending_all}
|
|
pending_data["firewall"]["needs_apply"] = True
|
|
pending_data["firewall"]["change_count"] = 1
|
|
pending_data["nginx"]["pending_changes"] = True
|
|
|
|
with (
|
|
patch("daemon.handlers.status.status_pending", return_value=pending_data),
|
|
patch("daemon.handlers.status.refresh_state"),
|
|
patch.dict(
|
|
"daemon.handlers.status.SYS_APPLY",
|
|
{
|
|
"firewall": mock_fw,
|
|
"nginx": mock_nginx,
|
|
},
|
|
),
|
|
):
|
|
result = status.status_apply_all(None, None)
|
|
assert "firewall" not in result["applied"]
|
|
assert "nginx" in result["applied"]
|
|
assert "Firewall" in result["errors"]
|
|
mock_nginx.assert_called_once()
|
|
|
|
def test_force_body_forwarded_to_firewall_only(self):
|
|
mock_fw = MagicMock()
|
|
mock_nginx = MagicMock()
|
|
|
|
pending_data = {**self._fake_pending_all}
|
|
pending_data["firewall"]["needs_apply"] = True
|
|
pending_data["firewall"]["change_count"] = 1
|
|
pending_data["nginx"]["pending_changes"] = True
|
|
|
|
with (
|
|
patch("daemon.handlers.status.status_pending", return_value=pending_data),
|
|
patch("daemon.handlers.status.refresh_state"),
|
|
patch.dict(
|
|
"daemon.handlers.status.SYS_APPLY",
|
|
{
|
|
"firewall": mock_fw,
|
|
"nginx": mock_nginx,
|
|
},
|
|
),
|
|
):
|
|
status.status_apply_all(None, {"force": True})
|
|
mock_fw.assert_called_once_with(None, {"force": True})
|
|
mock_nginx.assert_called_once_with(None, None)
|
|
|
|
def test_no_body_passed_without_force(self):
|
|
mock_fw = MagicMock()
|
|
|
|
pending_data = {**self._fake_pending_all}
|
|
pending_data["firewall"]["needs_apply"] = True
|
|
pending_data["firewall"]["change_count"] = 1
|
|
|
|
with (
|
|
patch("daemon.handlers.status.status_pending", return_value=pending_data),
|
|
patch("daemon.handlers.status.refresh_state"),
|
|
patch.dict("daemon.handlers.status.SYS_APPLY", {"firewall": mock_fw}),
|
|
):
|
|
status.status_apply_all(None, None)
|
|
mock_fw.assert_called_once_with(None, None)
|
|
|
|
|
|
class TestSysOrder:
|
|
"""Verify SYS_ORDER and SYS_LABELS constants."""
|
|
|
|
def test_order_network_first(self):
|
|
assert status.SYS_ORDER[0] == "networkd"
|
|
|
|
def test_all_subsystems_present(self):
|
|
expected = {"networkd", "firewall", "wireguard", "dnsmasq", "nginx"}
|
|
assert set(status.SYS_ORDER) == expected
|
|
|
|
def test_labels_match(self):
|
|
for name in status.SYS_ORDER:
|
|
assert name in status.SYS_LABELS
|
|
assert name in status.SYS_APPLY
|
|
|
|
def test_apply_functions_callable(self):
|
|
for name in status.SYS_ORDER:
|
|
assert callable(status.SYS_APPLY[name])
|